hold/scanner: stop one undispatchable job freezing all scanning

Vulnerability scanning produced nothing across the whole deployment for nine days, from 2026-08-25 01:20:48 until a scanner restart on 2026-09-03. The scanner was connected and idle, the hold's discovery pass kept reporting unscannedFound=15 every four hours, and no scan_jobs row was created in that entire window. hasActiveJobs counted pending, assigned and processing rows globally with no age bound, and waitForCapacity spins while it is true. dispatchLoop calls it before popping any candidate, so a single pending row that never reached a terminal state reported "busy" forever: discovery kept pushing candidates into unscannedQueue and nothing ever popped them. That is why the symptom was an empty queue rather than a growing one. Nothing papered over it because push-triggered enqueue only fires for owner or a tier with scan_on_push, which in production means pro alone. All 210 manifests pushed to this hold in that window came from free, supporter, or accounts with no crew row, so the frozen proactive loop was the only source of jobs. Nor could it recover on its own. Only Enqueue and drainPendingJobs dispatch a pending row, and drainPendingJobs runs only when a scanner newly connects; reDispatchTimedOut considered assigned rows only. The hold had been up since Aug 14 and the scanner since Aug 21 on the same websocket, so the drain path had not run since the row appeared. So bound the capacity gate to pending rows younger than pendingStaleAfter, give reDispatchTimedOut a pending reclaim, and check RowsAffected on the assign UPDATE now that two dispatchers can race for a row. waitForCapacity warns and names the blocking jobs after ten minutes without capacity, because the failure mode above was completely silent. Two adjacent fixes for the same outage. The scanner never called InitLogger, so log_level and log_shipper were dead config and an idle scanner was mute, which is what made nine days invisible. And skipReason now also skips a job whose layers contain nothing tar-shaped: the job that wedged this queue was an in-toto attestation whose config mediaType is an ordinary image config, so the existing config-type check missed it and buildOCILayout would have handed Syft an empty image. The regression tests were verified against the old logic first: three of them fail on it and pass on the fix. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01DPWkeCKcbtGoyXyyeMhSps
2026-09-10 04:06:06 +00:00 · 2026-09-02 21:12:13 -05:00
parent af7522b154
commit dfd604b106
7 changed files with 430 additions and 34 deletions
@@ -21,6 +21,25 @@ import (
 	"github.com/gorilla/websocket"
 )

+const (
+	// pendingReclaimAfter is how long a job may sit in 'pending' before the
+	// re-dispatch loop offers it to a scanner again. Enqueue and
+	// drainPendingJobs are both one-shot, so without a periodic re-offer a row
+	// whose only dispatch attempt failed waits for the next scanner connect.
+	pendingReclaimAfter = 1 * time.Minute
+
+	// pendingStaleAfter is how long a pending job counts as active for dispatch
+	// throttling. Assigned and processing jobs need no such bound — the ack and
+	// processing timeouts always resolve them — but 'pending' is otherwise
+	// unbounded, and one row stuck there froze proactive scanning for nine days.
+	pendingStaleAfter = 15 * time.Minute
+
+	// capacityStallWarnAfter is how long the dispatch loop waits with no
+	// capacity before it says so. A silent stall is what made this class of
+	// failure invisible until users noticed missing scans.
+	capacityStallWarnAfter = 10 * time.Minute
+)
+
 // ScanBroadcaster manages scanner WebSocket connections and dispatches scan jobs
 // using a competing-consumer pattern. Jobs are persisted in SQLite and dispatched
 // round-robin to connected scanners.
@@ -396,7 +415,7 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) {
 	sb.nextIdx++

 	// Mark as assigned in database
-	_, err := sb.db.Exec(`
+	res, err := sb.db.Exec(`
 		UPDATE scan_jobs SET status = 'assigned', assigned_to = ?, assigned_at = ?
 		WHERE seq = ? AND status = 'pending'
 	`, sub.id, time.Now(), job.Seq)
@@ -404,6 +423,13 @@ func (sb *ScanBroadcaster) dispatchJob(job *ScanJobEvent) {
 		slog.Error("Failed to assign scan job", "seq", job.Seq, "error", err)
 		return
 	}
+	// Two paths hand out pending rows now (drainPendingJobs on connect, and the
+	// re-dispatch loop), so a row that is no longer pending was claimed by the
+	// other one. Sending it anyway would scan it twice.
+	if n, err := res.RowsAffected(); err == nil && n == 0 {
+		slog.Debug("Scan job no longer pending, skipping dispatch", "seq", job.Seq)
+		return
+	}

 	// Send to subscriber
 	select {
@@ -790,9 +816,16 @@ func (sb *ScanBroadcaster) reDispatchLoop() {
 }

 // reDispatchTimedOut finds jobs that were assigned but not acked/completed within timeout,
+// re-offers jobs that have been sitting in 'pending' with nobody to hand them to,
 // and also marks stuck processing jobs as failed.
 // Collects timed-out rows first, closes cursor, then resets and re-dispatches
 // to avoid holding a SELECT cursor open during UPDATEs (prevents SQLite BUSY).
+//
+// Pending rows matter as much as assigned ones: Enqueue and drainPendingJobs are
+// both one-shot dispatchers, so before this loop covered 'pending' a row whose
+// single dispatch attempt failed (or that a disconnect unassigned while no
+// scanner reconnected afterwards) stayed pending forever — and hasActiveJobs()
+// counted it, which stopped proactive scanning deployment-wide.
 func (sb *ScanBroadcaster) reDispatchTimedOut() {
 	timeout := time.Now().Add(-sb.ackTimeout)

@@ -809,25 +842,31 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() {
 	}

 	rows, err := sb.db.Query(`
-		SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json
+		SELECT seq, manifest_digest, repository, tag, user_did, user_handle, hold_did, hold_endpoint, tier, config_json, layers_json, status
 		FROM scan_jobs
-		WHERE status = 'assigned' AND assigned_at < ?
+		WHERE (status = 'assigned' AND assigned_at < ?)
+		   OR (status = 'pending' AND datetime(created_at) < datetime('now', ?))
 		ORDER BY seq ASC
-	`, timeout)
+	`, timeout, sqliteAgoModifier(pendingReclaimAfter))
 	if err != nil {
 		slog.Error("Failed to query timed-out scan jobs", "error", err)
 		return
 	}

-	var jobs []*ScanJobEvent
+	type reclaimable struct {
+		job    *ScanJobEvent
+		status string
+	}
+
+	var jobs []reclaimable
 	for rows.Next() {
 		job := &ScanJobEvent{Type: "job"}
-		var configJSON, layersJSON string
+		var configJSON, layersJSON, status string

 		err := rows.Scan(
 			&job.Seq, &job.ManifestDigest, &job.Repository, &job.Tag,
 			&job.UserDID, &job.UserHandle, &job.HoldDID, &job.HoldEndpoint,
-			&job.Tier, &configJSON, &layersJSON,
+			&job.Tier, &configJSON, &layersJSON, &status,
 		)
 		if err != nil {
 			continue
@@ -835,27 +874,41 @@ func (sb *ScanBroadcaster) reDispatchTimedOut() {

 		job.Config = json.RawMessage(configJSON)
 		job.Layers = json.RawMessage(layersJSON)
-		jobs = append(jobs, job)
+		jobs = append(jobs, reclaimable{job: job, status: status})
 	}
 	rows.Close()

-	for _, job := range jobs {
-		_, err = sb.db.Exec(`
-			UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL
-			WHERE seq = ?
-		`, job.Seq)
-		if err != nil {
-			continue
+	for _, r := range jobs {
+		job := r.job
+
+		// Already-pending rows need no reset — and skipping the UPDATE keeps a
+		// row another dispatcher claimed between the SELECT and here assigned,
+		// so dispatchJob's status guard can drop it instead of double-sending.
+		if r.status != "pending" {
+			_, err = sb.db.Exec(`
+				UPDATE scan_jobs SET status = 'pending', assigned_to = NULL, assigned_at = NULL
+				WHERE seq = ?
+			`, job.Seq)
+			if err != nil {
+				continue
+			}
 		}

-		slog.Info("Re-dispatching timed-out scan job",
+		slog.Info("Re-dispatching scan job",
 			"seq", job.Seq,
-			"repository", job.Repository)
+			"repository", job.Repository,
+			"previousStatus", r.status)

 		sb.dispatchJob(job)
 	}
 }

+// sqliteAgoModifier renders a duration as a SQLite datetime() modifier that
+// walks backwards from 'now', e.g. 15m becomes "-900 seconds".
+func sqliteAgoModifier(d time.Duration) string {
+	return fmt.Sprintf("-%d seconds", int64(d.Seconds()))
+}
+
 // Close stops background goroutines and closes the scan broadcaster's database connection
 func (sb *ScanBroadcaster) Close() error {
 	if sb.stopCh != nil {
@@ -1273,10 +1326,20 @@ func (sb *ScanBroadcaster) dispatchLoop() {
 // waitForCapacity blocks until there are no active proactive scan jobs.
 // Returns false if stopCh is closed.
 func (sb *ScanBroadcaster) waitForCapacity() bool {
+	blockedSince := time.Now()
+	var lastWarn time.Time
+
 	for {
 		if !sb.hasActiveJobs() {
 			return true
 		}
+
+		if blocked := time.Since(blockedSince); blocked >= capacityStallWarnAfter &&
+			time.Since(lastWarn) >= capacityStallWarnAfter {
+			sb.logStalledCapacity(blocked)
+			lastWarn = time.Now()
+		}
+
 		select {
 		case <-sb.stopCh:
 			return false
@@ -1488,13 +1551,23 @@ func (sb *ScanBroadcaster) hasConnectedScanners() bool {
 	return len(sb.subscribers) > 0
 }

-// hasActiveJobs returns true if there are any pending, assigned, or processing scan jobs.
+// hasActiveJobs returns true if there are any assigned or processing scan jobs,
+// or any recently-created pending ones.
+//
+// Pending rows older than pendingStaleAfter are deliberately not counted. A job
+// that has been pending that long is one no scanner can be given (dispatch
+// failed, or no scanner is connected), and counting it blocks the proactive
+// dispatch loop for as long as the row exists — which is how a single
+// undispatchable job stopped scanning deployment-wide. The re-dispatch loop
+// keeps re-offering such rows, so ignoring them here costs nothing when a
+// scanner is available.
 func (sb *ScanBroadcaster) hasActiveJobs() bool {
 	var count int
 	err := sb.db.QueryRow(`
 		SELECT COUNT(*) FROM scan_jobs
-		WHERE status IN ('pending', 'assigned', 'processing')
-	`).Scan(&count)
+		WHERE status IN ('assigned', 'processing')
+		   OR (status = 'pending' AND datetime(created_at) > datetime('now', ?))
+	`, sqliteAgoModifier(pendingStaleAfter)).Scan(&count)
 	if err != nil {
 		slog.Error("Failed to check active scan jobs", "error", err)
 		return true // Assume busy on error
@@ -1502,6 +1575,32 @@ func (sb *ScanBroadcaster) hasActiveJobs() bool {
 	return count > 0
 }

+// logStalledCapacity reports what is holding the dispatch loop back, so a stall
+// shows up in the hold's logs instead of only as an absence of scan results.
+func (sb *ScanBroadcaster) logStalledCapacity(blockedFor time.Duration) {
+	var (
+		count     int
+		oldestSeq sql.NullInt64
+		statuses  sql.NullString
+	)
+	err := sb.db.QueryRow(`
+		SELECT COUNT(*), MIN(seq), GROUP_CONCAT(DISTINCT status)
+		FROM scan_jobs
+		WHERE status IN ('pending', 'assigned', 'processing')
+	`).Scan(&count, &oldestSeq, &statuses)
+	if err != nil {
+		slog.Warn("Proactive scan dispatch stalled; could not inspect active jobs",
+			"blockedFor", blockedFor.Truncate(time.Minute), "error", err)
+		return
+	}
+
+	slog.Warn("Proactive scan dispatch stalled waiting on active jobs",
+		"blockedFor", blockedFor.Truncate(time.Minute),
+		"activeJobs", count,
+		"oldestSeq", oldestSeq.Int64,
+		"statuses", statuses.String)
+}
+
 func generateSubscriberID() string {
 	b := make([]byte, 8)
 	_, _ = rand.Read(b)
@@ -0,0 +1,167 @@
+package pds
+
+import (
+	"testing"
+	"time"
+)
+
+// backdateJob ages a job's created_at so the staleness windows in
+// hasActiveJobs and reDispatchTimedOut can be exercised without sleeping.
+func backdateJob(t *testing.T, sb *ScanBroadcaster, seq int64, minutes int) {
+	t.Helper()
+
+	_, err := sb.db.Exec(
+		`UPDATE scan_jobs SET created_at = datetime('now', ?) WHERE seq = ?`,
+		sqliteAgoModifier(time.Duration(minutes)*time.Minute), seq)
+	if err != nil {
+		t.Fatalf("backdate job %d: %v", seq, err)
+	}
+}
+
+func jobStatus(t *testing.T, sb *ScanBroadcaster, seq int64) string {
+	t.Helper()
+
+	var status string
+	if err := sb.db.QueryRow(`SELECT status FROM scan_jobs WHERE seq = ?`, seq).Scan(&status); err != nil {
+		t.Fatalf("query status for %d: %v", seq, err)
+	}
+	return status
+}
+
+// TestScanHasActiveJobs_IgnoresLongPendingJob is the regression test for the
+// nine-day deployment-wide scanning outage. A single job sat in 'pending' with
+// nothing left to dispatch it, hasActiveJobs() counted it forever, and
+// waitForCapacity() therefore never let the proactive dispatch loop enqueue
+// another job — so discovery kept finding unscanned images and creating none.
+func TestScanHasActiveJobs_IgnoresLongPendingJob(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	seedPendingJobs(t, sb, 1)
+
+	if !sb.hasActiveJobs() {
+		t.Fatal("a freshly enqueued pending job must count as active")
+	}
+
+	backdateJob(t, sb, 1, 60)
+
+	if sb.hasActiveJobs() {
+		t.Error("a job pending for an hour must not block dispatch capacity")
+	}
+}
+
+// TestScanHasActiveJobs_CountsAssignedAndProcessing guards the other half:
+// assigned and processing jobs have their own reclaim timeouts, so they must
+// still hold capacity no matter how old the row is.
+func TestScanHasActiveJobs_CountsAssignedAndProcessing(t *testing.T) {
+	for _, status := range []string{"assigned", "processing"} {
+		t.Run(status, func(t *testing.T) {
+			sb := newTestScanBroadcaster(t)
+			seedPendingJobs(t, sb, 1)
+			backdateJob(t, sb, 1, 60)
+
+			if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = ? WHERE seq = 1`, status); err != nil {
+				t.Fatalf("set status: %v", err)
+			}
+
+			if !sb.hasActiveJobs() {
+				t.Errorf("%s job must hold dispatch capacity", status)
+			}
+		})
+	}
+}
+
+// TestScanHasActiveJobs_IgnoresTerminalJobs keeps completed and failed rows out
+// of the capacity check — the table holds tens of thousands of them.
+func TestScanHasActiveJobs_IgnoresTerminalJobs(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	seedPendingJobs(t, sb, 2)
+
+	if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'completed' WHERE seq = 1`); err != nil {
+		t.Fatalf("complete: %v", err)
+	}
+	if _, err := sb.db.Exec(`UPDATE scan_jobs SET status = 'failed' WHERE seq = 2`); err != nil {
+		t.Fatalf("fail: %v", err)
+	}
+
+	if sb.hasActiveJobs() {
+		t.Error("terminal jobs must not hold dispatch capacity")
+	}
+}
+
+// TestScanReDispatch_ReoffersStalePendingJob covers the missing lease reclaim.
+// Enqueue and drainPendingJobs are both one-shot, so a pending row used to wait
+// for the next scanner connect — nine days, in production. The re-dispatch loop
+// now hands it to a connected scanner on its own.
+func TestScanReDispatch_ReoffersStalePendingJob(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	sub := newTestScanSubscriber(t, sb, 4)
+	seedPendingJobs(t, sb, 1)
+	backdateJob(t, sb, 1, 30)
+
+	sb.reDispatchTimedOut()
+
+	if got := jobStatus(t, sb, 1); got != "assigned" {
+		t.Errorf("stale pending job not re-dispatched, status=%q", got)
+	}
+	if len(sub.send) != 1 {
+		t.Errorf("scanner received %d jobs, want 1", len(sub.send))
+	}
+}
+
+// TestScanReDispatch_LeavesFreshPendingJobAlone stops the loop from racing the
+// dispatch Enqueue just did, which would scan the same manifest twice.
+func TestScanReDispatch_LeavesFreshPendingJobAlone(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	sub := newTestScanSubscriber(t, sb, 4)
+	seedPendingJobs(t, sb, 1)
+
+	sb.reDispatchTimedOut()
+
+	if got := jobStatus(t, sb, 1); got != "pending" {
+		t.Errorf("just-enqueued job was re-dispatched, status=%q", got)
+	}
+	if len(sub.send) != 0 {
+		t.Errorf("scanner received %d jobs, want 0", len(sub.send))
+	}
+}
+
+// TestScanReDispatch_PendingSurvivesWithNoScanner verifies the reclaim is a
+// no-op rather than a corruption when no scanner is connected: the row stays
+// pending and gets offered again on the next tick.
+func TestScanReDispatch_PendingSurvivesWithNoScanner(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	seedPendingJobs(t, sb, 1)
+	backdateJob(t, sb, 1, 30)
+
+	sb.reDispatchTimedOut()
+
+	if got := jobStatus(t, sb, 1); got != "pending" {
+		t.Errorf("job status changed with no scanner connected, status=%q", got)
+	}
+}
+
+// TestScanDispatchJob_SkipsClaimedJob covers the double-send window opened by
+// having two dispatchers for pending rows: whichever loses the UPDATE race must
+// not also push the job onto a scanner's queue.
+func TestScanDispatchJob_SkipsClaimedJob(t *testing.T) {
+	sb := newTestScanBroadcaster(t)
+	sub := newTestScanSubscriber(t, sb, 4)
+	seedPendingJobs(t, sb, 1)
+
+	// Another dispatcher got there first.
+	if _, err := sb.db.Exec(`UPDATE scan_jobs SET status='assigned', assigned_to='other' WHERE seq = 1`); err != nil {
+		t.Fatalf("claim: %v", err)
+	}
+
+	sb.dispatchJob(&ScanJobEvent{Type: "job", Seq: 1, Repository: "repo"})
+
+	if len(sub.send) != 0 {
+		t.Errorf("already-claimed job was dispatched again (%d sends)", len(sub.send))
+	}
+	var assignedTo string
+	if err := sb.db.QueryRow(`SELECT assigned_to FROM scan_jobs WHERE seq = 1`).Scan(&assignedTo); err != nil {
+		t.Fatalf("query assigned_to: %v", err)
+	}
+	if assignedTo != "other" {
+		t.Errorf("assignment stolen from the first dispatcher, assigned_to=%q", assignedTo)
+	}
+}