hold/pds: fix two scanner-disconnect panics in the scan broadcaster

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) <noreply@anthropic.com>

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)

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))
+	}
+}