From ca539b1f9dcf05c4318e2b1d12b801233edd27d7 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 8 Aug 2026 21:33:29 -0500 Subject: [PATCH] hold/pds: stop firehose backfill panicking on subscriber disconnect MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A relay that connected with a stale cursor and then dropped mid-backfill took the whole hold process down: panic: send on closed channel pds.sendBackfillMsg events.go:869 pds.backfillFromDatabase events.go:809 pds.backfillSubscriber events.go:716 Subscribe spawns backfillSubscriber in its own goroutine, and that goroutine writes to sub.send without holding b.mu. Unsubscribe closed sub.send under the lock, so a disconnect during backfill closed the channel out from under an in-flight send. select cannot guard that — a send on a closed channel panics unconditionally. sub.send is now never closed. Subscriber gains a done channel that Unsubscribe closes instead, and every sender that runs unlocked selects on it. The map check in Unsubscribe keeps the close single-shot, which matters because both readPump and handleSubscriber call it on the way out. handleSubscriber selects on done rather than ranging over send, since nothing closes send any more. Broadcast and BroadcastIdentity were already safe (they send under b.mu, which excludes Unsubscribe). backfillFromMemory was safe too via b.mu.RLock, but now routes through the shared helper so a disconnect aborts immediately instead of stalling up to 5s per event while holding the read lock and blocking every broadcast. The bug dates to 2025-10, so every build since is affected. It only fires when a backfill goroutine exists, which Subscribe skips when cursor == currentSeq — that is why caught-up relays never triggered it and a hold whose relays are all behind is exposed on every reconnect. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/hold/pds/events.go | 59 +++++++++++++--- pkg/hold/pds/events_test.go | 136 ++++++++++++++++++++++++++++++++++++ 2 files changed, 185 insertions(+), 10 deletions(-) diff --git a/pkg/hold/pds/events.go b/pkg/hold/pds/events.go index 1067d33..bf1e0f9 100644 --- a/pkg/hold/pds/events.go +++ b/pkg/hold/pds/events.go @@ -43,6 +43,14 @@ type Subscriber struct { conn *websocket.Conn send chan firehoseMsg cursor int64 // Last sequence number this subscriber has seen + + // done is closed by Unsubscribe to signal that this subscriber has gone + // away. It exists because backfill runs in its own goroutine without + // holding b.mu: closing send from Unsubscribe while a backfill was + // mid-flight panicked the whole process with "send on closed channel". + // send is therefore never closed — senders select on done instead, and + // the channel is collected once the subscriber is unreachable. + done chan struct{} } // HistoricalEvent stores past events for cursor-based backfill @@ -434,6 +442,7 @@ func (b *EventBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, userAge conn: conn, send: make(chan firehoseMsg, 10), // Buffer 10 events cursor: cursor, + done: make(chan struct{}), } b.mu.Lock() @@ -522,9 +531,15 @@ func (b *EventBroadcaster) Unsubscribe(sub *Subscriber) { b.mu.Lock() defer b.mu.Unlock() + // The map check keeps this single-shot: both readPump and handleSubscriber + // call Unsubscribe on their way out, and only the first finds sub present. + // + // Close done, never send. A backfill goroutine may still be writing to + // send without holding b.mu, and closing a channel out from under an + // in-flight send panics unrecoverably — select cannot guard it. if _, ok := b.subscribers[sub]; ok { delete(b.subscribers, sub) - close(sub.send) + close(sub.done) } } @@ -863,10 +878,25 @@ func (b *EventBroadcaster) loadIdentityEvents(cursor int64) ([]*identityMsg, err } // sendBackfillMsg pushes a message into the subscriber's send channel with a -// 5-second slow-subscriber timeout. Returns false if the timeout fired and the -// caller should abandon the backfill. +// 5-second slow-subscriber timeout. Returns false if the subscriber went away +// or the timeout fired, in which case the caller should abandon the backfill. +// +// The done case is what makes this safe to call from the backfill goroutine, +// which does not hold b.mu. A disconnect mid-backfill used to close send under +// this send and panic the process. func sendBackfillMsg(sub *Subscriber, msg firehoseMsg, seq int64) bool { + // Non-blocking pre-check so an already-disconnected subscriber always + // aborts. Without it select would pick at random whenever done is closed + // and send still has buffer space, making the abort nondeterministic. select { + case <-sub.done: + return false + default: + } + + select { + case <-sub.done: + return false case sub.send <- msg: return true case <-time.After(5 * time.Second): @@ -882,12 +912,11 @@ func (b *EventBroadcaster) backfillFromMemory(sub *Subscriber, cursor int64) { for _, he := range b.eventHistory { if he.Seq > cursor { - select { - case sub.send <- &commitMsg{ev: he.Event}: - // Sent - case <-time.After(5 * time.Second): - // Timeout, subscriber too slow - slog.Warn("Backfill timeout for subscriber", "seq", he.Seq) + // Holding b.mu.RLock excludes Unsubscribe's Lock, so send cannot + // be closed under us here. Route through the shared helper anyway + // so a disconnect aborts promptly instead of stalling 5s per event + // while the read lock blocks every broadcast. + if !sendBackfillMsg(sub, &commitMsg{ev: he.Event}, he.Seq) { return } } @@ -921,7 +950,17 @@ func (b *EventBroadcaster) handleSubscriber(sub *Subscriber) { sub.conn.Close() }() - for msg := range sub.send { + // Ranging over send would block forever now that nothing closes it, so + // select on done as the termination signal. done is closed by whichever of + // readPump/handleSubscriber unwinds first. + for { + var msg firehoseMsg + select { + case <-sub.done: + return + case msg = <-sub.send: + } + header := events.EventHeader{ Op: events.EvtKindMessage, MsgType: msg.msgType(), diff --git a/pkg/hold/pds/events_test.go b/pkg/hold/pds/events_test.go index 82919c0..5122dc1 100644 --- a/pkg/hold/pds/events_test.go +++ b/pkg/hold/pds/events_test.go @@ -822,3 +822,139 @@ func TestIdentityMsg_MarshalBody(t *testing.T) { t.Error("expected non-empty CBOR body") } } + +// newTestSubscriber builds a Subscriber the way Subscribe does, with done +// initialised. Struct literals with a nil done channel make Unsubscribe panic +// on close, so any test that unsubscribes must go through this. +func newTestSubscriber(bufSize int, cursor int64) *Subscriber { + return &Subscriber{ + conn: nil, + send: make(chan firehoseMsg, bufSize), + cursor: cursor, + done: make(chan struct{}), + } +} + +// TestSendBackfillMsg_AbortsOnDisconnect verifies that a subscriber which has +// already gone away aborts the backfill rather than buffering more work, and +// that it does so deterministically even when the send buffer has room. +func TestSendBackfillMsg_AbortsOnDisconnect(t *testing.T) { + sub := newTestSubscriber(10, 0) + close(sub.done) + + if sendBackfillMsg(sub, &commitMsg{ev: &RepoCommitEvent{Seq: 1}}, 1) { + t.Fatal("sendBackfillMsg returned true for a disconnected subscriber") + } + if len(sub.send) != 0 { + t.Errorf("expected nothing queued for a disconnected subscriber, got %d", len(sub.send)) + } + + // Live subscriber with buffer space still sends. + live := newTestSubscriber(1, 0) + if !sendBackfillMsg(live, &commitMsg{ev: &RepoCommitEvent{Seq: 2}}, 2) { + t.Error("sendBackfillMsg returned false for a live subscriber with buffer space") + } +} + +// TestBackfillFromDatabase_ConcurrentUnsubscribe is the regression test for a +// panic that killed the whole hold process: +// +// panic: send on closed channel +// pds.sendBackfillMsg events.go:869 +// pds.backfillFromDatabase events.go:809 +// +// backfillFromDatabase runs in its own goroutine and does not hold b.mu, so a +// subscriber disconnecting mid-backfill had Unsubscribe close send out from +// under an in-flight send. Any relay reconnecting with a stale cursor and then +// dropping would take the service down. Unsubscribe now closes done instead +// and send is never closed. +// +// The subscriber uses a 1-slot buffer with no reader so the backfill is +// reliably blocked in the send when Unsubscribe fires. +func TestBackfillFromDatabase_ConcurrentUnsubscribe(t *testing.T) { + for i := 0; i < 25; i++ { + func() { + dbPath := t.TempDir() + "/events.db" + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 500, dbPath) + defer func() { _ = broadcaster.Close() }() + + ctx := context.Background() + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + for j := 0; j < 50; j++ { + broadcaster.Broadcast(ctx, &RepoEvent{ + NewRoot: testCID, + Rev: "rev", + RepoSlice: []byte("car"), + Ops: []RepoOp{}, + }) + } + + sub := newTestSubscriber(1, 0) + broadcaster.mu.Lock() + broadcaster.subscribers[sub] = true + broadcaster.mu.Unlock() + + done := make(chan struct{}) + go func() { + defer close(done) + // Pre-fix this panicked instead of returning. + _ = broadcaster.backfillFromDatabase(sub, 0) + }() + + broadcaster.Unsubscribe(sub) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("backfillFromDatabase did not return after Unsubscribe") + } + }() + } +} + +// TestUnsubscribe_IsSingleShot verifies the map guard holds: readPump and +// handleSubscriber both call Unsubscribe on their way out, and a second close +// of done would panic. +func TestUnsubscribe_IsSingleShot(t *testing.T) { + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") + sub := newTestSubscriber(1, 0) + + broadcaster.mu.Lock() + broadcaster.subscribers[sub] = true + broadcaster.mu.Unlock() + + broadcaster.Unsubscribe(sub) + broadcaster.Unsubscribe(sub) // must be a no-op, not a double close + + select { + case <-sub.done: + default: + t.Error("done was not closed by Unsubscribe") + } +} + +// TestBroadcast_AfterUnsubscribe verifies a removed subscriber stops receiving +// events and that broadcasting does not touch its channel. +func TestBroadcast_AfterUnsubscribe(t *testing.T) { + broadcaster := NewEventBroadcaster("did:web:hold.example.com", 10, "") + ctx := context.Background() + sub := newTestSubscriber(4, 0) + + broadcaster.mu.Lock() + broadcaster.subscribers[sub] = true + broadcaster.mu.Unlock() + + broadcaster.Unsubscribe(sub) + + testCID, _ := cid.Decode("bafyreib2rxk3rkhh5ylyxj3x3gathxt3s32qvwj2lf3qg4kmzr6b7teqke") + broadcaster.Broadcast(ctx, &RepoEvent{ + NewRoot: testCID, + Rev: "rev", + RepoSlice: []byte("car"), + Ops: []RepoOp{}, + }) + + if len(sub.send) != 0 { + t.Errorf("unsubscribed subscriber received %d events", len(sub.send)) + } +}