From 05b856bf4ef5c943b4bd435a3ba295414fbcf4c5 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 8 Aug 2026 21:37:40 -0500 Subject: [PATCH] hold/pds: fix two scanner-disconnect panics in the scan broadcaster MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Same class of bug as the firehose backfill, plus a second one found alongside it. Both take down the whole hold process. 1. send on closed channel. Subscribe spawns drainPendingJobs in its own goroutine, and it sends to sub.send without holding sb.mu. Unsubscribe closed sub.send under the lock, so a scanner disconnecting during the drain closed the channel out from under an in-flight send. The existing `case <-sub.done` guard did not help: done meant "writer goroutine exited" and was closed by handleWriter, which is a different event from unsubscribing. 2. close of closed channel. Unsubscribe closed sub.send unconditionally, but it is called from two places — handleWriter on write error, and handleReader in its defer. A scanner dropping mid-write hits both, and the slice-removal loop had no guard, so the second call fell straight through to the close. The unassign UPDATE ran twice for the same reason, which could return jobs a replacement scanner had already been handed. sub.send is now never closed. done is repurposed to mean "this subscriber is gone", closed only by Unsubscribe and guarded on whether the subscriber was actually still registered. That makes drainPendingJobs' existing done case correct, and handleWriter selects on done rather than ranging over send. dispatchJob was already safe — it sends under sb.mu, which excludes Unsubscribe. hold01 is unaffected in practice (scanner disabled, no shared secret), but seamark-hold runs the scanner continuously and is exposed on any scanner restart. Co-Authored-By: Claude Opus 5 (1M context) --- pkg/hold/pds/scan_broadcaster.go | 32 ++++- pkg/hold/pds/scan_broadcaster_test.go | 180 ++++++++++++++++++++++++++ 2 files changed, 206 insertions(+), 6 deletions(-) create mode 100644 pkg/hold/pds/scan_broadcaster_test.go diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index f49159e..52e94bb 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -342,13 +342,23 @@ func (sb *ScanBroadcaster) Unsubscribe(sub *ScanSubscriber) { sb.mu.Lock() defer sb.mu.Unlock() + found := false for i, s := range sb.subscribers { if s == sub { sb.subscribers = append(sb.subscribers[:i], sb.subscribers[i+1:]...) + found = true break } } + // A dropped scanner unwinds both handleWriter (on write error) and + // handleReader (in its defer), and each calls Unsubscribe. Everything below + // must happen exactly once: re-running the UPDATE would unassign jobs a + // replacement scanner had already picked up, and closing done twice panics. + if !found { + return + } + // Mark assigned/processing jobs as pending again so they can be re-dispatched. // Including 'processing' handles scanner crashes mid-scan. _, err := sb.db.Exec(` @@ -361,7 +371,10 @@ func (sb *ScanBroadcaster) Unsubscribe(sub *ScanSubscriber) { "error", err) } - close(sub.send) + // Close done, never send. drainPendingJobs runs in its own goroutine + // without holding sb.mu, so closing send here would race an in-flight + // send and panic the process unrecoverably. + close(sub.done) slog.Info("Scanner unsubscribed", "id", sub.id, @@ -411,12 +424,19 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) { // handleWriter sends jobs to a scanner over its WebSocket connection func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) { - defer func() { - sub.conn.Close() - close(sub.done) - }() + defer sub.conn.Close() + + // done is closed by Unsubscribe, not here — ranging over send would block + // forever now that nothing closes it. handleReader always unsubscribes on + // its way out, so this goroutine is guaranteed to be released. + for { + var job *ScanJobEvent + select { + case <-sub.done: + return + case job = <-sub.send: + } - for job := range sub.send { data, err := json.Marshal(job) if err != nil { slog.Error("Failed to marshal scan job", "seq", job.Seq, "error", err) diff --git a/pkg/hold/pds/scan_broadcaster_test.go b/pkg/hold/pds/scan_broadcaster_test.go new file mode 100644 index 0000000..a6aeae2 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_test.go @@ -0,0 +1,180 @@ +package pds + +import ( + "database/sql" + "testing" + "time" +) + +// newTestScanBroadcaster builds a ScanBroadcaster with just a database, no +// background goroutines. Subscriber lifecycle is all that is under test here, +// so the constructor's discovery/dispatch/stale loops (and their s3 and PDS +// dependencies) are deliberately skipped. +func newTestScanBroadcaster(t *testing.T) *ScanBroadcaster { + t.Helper() + + db, err := sql.Open("libsql", "file:"+t.TempDir()+"/scan.db") + if err != nil { + t.Fatalf("open scan db: %v", err) + } + t.Cleanup(func() { _ = db.Close() }) + + sb := &ScanBroadcaster{ + db: db, + holdDID: "did:web:hold.example.com", + holdEndpoint: "https://hold.example.com", + } + if err := sb.initSchema(); err != nil { + t.Fatalf("initSchema: %v", err) + } + return sb +} + +// newTestScanSubscriber mirrors what Subscribe builds, registered with the +// broadcaster so Unsubscribe finds it. +func newTestScanSubscriber(t *testing.T, sb *ScanBroadcaster, bufSize int) *ScanSubscriber { + t.Helper() + + sub := &ScanSubscriber{ + conn: nil, // no websocket needed; handleWriter is not exercised here + send: make(chan *ScanJobEvent, bufSize), + id: "test-subscriber", + done: make(chan struct{}), + } + + sb.mu.Lock() + sb.subscribers = append(sb.subscribers, sub) + sb.mu.Unlock() + + return sub +} + +func seedPendingJobs(t *testing.T, sb *ScanBroadcaster, n int) { + t.Helper() + + for i := 0; i < n; i++ { + _, err := sb.db.Exec(` + INSERT INTO scan_jobs + (manifest_digest, repository, tag, user_did, user_handle, + hold_did, hold_endpoint, tier, config_json, layers_json, status) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, 'pending') + `, "sha256:deadbeef", "repo", "latest", "did:plc:user", "user.example.com", + sb.holdDID, sb.holdEndpoint, "deckhand", "{}", "[]") + if err != nil { + t.Fatalf("seed job %d: %v", i, err) + } + } +} + +// TestScanUnsubscribe_IsIdempotent covers a panic reachable on any scanner that +// dropped mid-write. Unsubscribe used to close sub.send unconditionally, but it +// is called from two places — handleWriter on write error, and handleReader in +// its defer — so a write failure followed by the read side unwinding produced +// "panic: close of closed channel". The slice-removal loop had no guard, so the +// second call fell straight through to the close. +func TestScanUnsubscribe_IsIdempotent(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + sb.Unsubscribe(sub) + sb.Unsubscribe(sub) // must be a no-op, not a second close + + select { + case <-sub.done: + default: + t.Error("done was not closed by Unsubscribe") + } + + sb.mu.RLock() + n := len(sb.subscribers) + sb.mu.RUnlock() + if n != 0 { + t.Errorf("expected subscriber removed, got %d remaining", n) + } +} + +// TestScanUnsubscribe_UnassignsJobsOnce verifies the idempotency guard protects +// the job-reassignment UPDATE too. Re-running it would unassign jobs that a +// replacement scanner had already been given. +func TestScanUnsubscribe_UnassignsJobsOnce(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status='assigned', assigned_to=?`, sub.id); err != nil { + t.Fatalf("assign: %v", err) + } + + sb.Unsubscribe(sub) + + var status string + if err := sb.db.QueryRow(`SELECT status FROM scan_jobs LIMIT 1`).Scan(&status); err != nil { + t.Fatalf("query: %v", err) + } + if status != "pending" { + t.Errorf("expected job returned to pending, got %q", status) + } + + // Hand the job to a "replacement" scanner, then unsubscribe the dead one + // again. The guard must stop it from stealing the job back. + if _, err := sb.db.Exec(`UPDATE scan_jobs SET status='assigned', assigned_to=?`, "replacement"); err != nil { + t.Fatalf("reassign: %v", err) + } + sb.Unsubscribe(sub) + + if err := sb.db.QueryRow(`SELECT status FROM scan_jobs LIMIT 1`).Scan(&status); err != nil { + t.Fatalf("query: %v", err) + } + if status != "assigned" { + t.Errorf("second Unsubscribe stole the replacement's job, status=%q", status) + } +} + +// TestScanDrainPendingJobs_ConcurrentUnsubscribe is the regression test for +// "panic: send on closed channel" in the scanner path — the same defect as the +// firehose backfill. Subscribe spawns drainPendingJobs in its own goroutine and +// it sends to sub.send without holding sb.mu, so a scanner disconnecting during +// the drain had Unsubscribe close the channel under an in-flight send. +// +// The subscriber uses a 1-slot buffer with no reader so the drain is reliably +// blocked in the send when Unsubscribe fires. +func TestScanDrainPendingJobs_ConcurrentUnsubscribe(t *testing.T) { + for i := 0; i < 25; i++ { + func() { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 1) + seedPendingJobs(t, sb, 50) + + done := make(chan struct{}) + go func() { + defer close(done) + // Pre-fix this panicked instead of returning. + sb.drainPendingJobs(sub, 0) + }() + + sb.Unsubscribe(sub) + + select { + case <-done: + case <-time.After(10 * time.Second): + t.Fatal("drainPendingJobs did not return after Unsubscribe") + } + }() + } +} + +// TestScanDispatchJob_AfterUnsubscribe verifies a removed scanner stops being +// dispatched to and that dispatch does not touch its channel. +func TestScanDispatchJob_AfterUnsubscribe(t *testing.T) { + sb := newTestScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + seedPendingJobs(t, sb, 1) + + sb.Unsubscribe(sub) + + sb.dispatchJob(&ScanJobEvent{Type: "job", Seq: 1, Repository: "repo"}) + + if len(sub.send) != 0 { + t.Errorf("unsubscribed scanner received %d jobs", len(sub.send)) + } +}