mirror of
https://tangled.org/evan.jarrett.net/at-container-registry
synced 2026-08-29 04:06:58 +00:00
hold/pds: stop firehose backfill panicking on subscriber disconnect
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) <noreply@anthropic.com>
This commit is contained in:
Evan Jarrett
co-authored by
Claude Opus 5
Claude Opus 5
parent
e1cbcb7a97
commit
ca539b1f9d
+49
-10
@@ -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(),
|
||||
|
||||
@@ -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))
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user