fix(scanner): Pass 2 dropped by Pass 1 slot race; clock-independent staging deadline test (#1648)

Fixes the two E2E failures on main after #1628 (#1602) that block
v0.71.0.

## Root causes
1. **macOS: `TestServiceStartScanDeepOnRunsSourceResolutionAndPass2`**
(run 39002404847). Not caused by #1628 (Pass 2 staging/in-process path
unchanged; #1628's own main run was green). Real production race:
`OnScanCompleted` spawns `startPass2` inside `executeScan`, before the
deferred `clearActiveJob` frees the server's engine slot. When Pass 2
reaches `StartScan` first it is rejected with "scan already in
progress", logged at Warn and dropped: no Pass 2 job is ever saved (test
times out after 10s) and the resolved source dir leaked. Slow runners
widen the window.
2. **Windows: `TestStagedSource_ScanDeadlineCoversStaging`**. A 1ns
`context.WithTimeout` is not observably expired on the coarse Windows
clock, so staging of a tiny dir finished before the ctx check.

## Fix
- `ErrScanInProgress` sentinel; `startPass2` retries (bounded 10s) while
the blocking job is a Pass-1 job, releases the resolved source and saves
a failed Pass 2 job if it still cannot start.
- Staging test now uses a 20ms timeout, plus a non-skipped source file
and a stall in the `stagingAfterReadBatch` seam past the deadline, so
the following ctx check deterministically fails. Still proves staging
runs under the scan deadline.
- New regression test `TestStartPass2WaitsForPass1SlotRelease` (fails
before: Pass 2 saved as failed with "scan already in progress").

## Verification
`go test -race ./internal/security/...` green, `-count=30` on the
touched tests, `GOOS=windows go vet`, both golangci-lint runs (bare and
`--build-tags server`) 0 issues. The macOS flake did not reproduce
locally under CPU load; diagnosis is from code plus the deterministic
regression test.

Author: Dumbris

internal/security/scanner/engine.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"io"
 	"os"
@@ -314,6 +315,10 @@ func (n *NoopCallback) OnScannerFailed(_ *ScanJob, _ string, _ error)          {
 func (n *NoopCallback) OnScanCompleted(_ *ScanJob, _ []*ScanReport)            {}
 func (n *NoopCallback) OnScanFailed(_ *ScanJob, _ error)                       {}
 
+// ErrScanInProgress is returned (wrapped) by StartScan when the server already
+// has a registered scan job.
+var ErrScanInProgress = errors.New("scan already in progress")
+
 // StartScan begins a scan of the specified server
 // Returns the scan job immediately; scanning runs in the background
 func (e *Engine) StartScan(ctx context.Context, req ScanRequest, callback ScanCallback) (*ScanJob, error) {
@@ -333,7 +338,7 @@ func (e *Engine) StartScan(ctx context.Context, req ScanRequest, callback ScanCa
 		// Snapshot: the running scan keeps writing to the live record.
 		snapshot := existing.clone()
 		e.mu.Unlock()
-		return snapshot, fmt.Errorf("scan already in progress for server %s (job %s)", req.ServerName, snapshot.ID)
+		return snapshot, fmt.Errorf("%w for server %s (job %s)", ErrScanInProgress, req.ServerName, snapshot.ID)
 	}
 	if check := e.registrationCheck; check != nil && !check(req.ServerName, req.Incarnation) {
 		e.mu.Unlock()

internal/security/scanner/export_ingested_contracts_test.go

@@ -342,7 +342,7 @@ func TestRT06_Pass2_LiveExportError_RunsOnlyMetadataIndependentScanners(t *testi
 	source := staleSourceDir(t)
 	info := &ServerInfo{Name: "toctou-p2", Protocol: "stdio", Command: "node", Args: []string{"server.js"}, WorkingDir: source}
 	svc, store := newR3Service(t, &liveFailProvider{info: info, liveErr: errors.New("server not found: toctou-p2")})
-	svc.startPass2("toctou-p2", info, 1)
+	svc.startPass2("toctou-p2", info, 1, "")
 
 	require.Eventually(t, func() bool {
 		jobs, err := store.ListScanJobs("toctou-p2")
@@ -436,7 +436,7 @@ func TestRT06_Pass2_LiveExportError_TrivyRunsMetadataScannerRefused(t *testing.T
 		return "", "", 0, nil
 	}
 
-	svc.startPass2("toctou-p2", info, 1)
+	svc.startPass2("toctou-p2", info, 1, "")
 	require.Eventually(t, func() bool {
 		jobs, err := store.ListScanJobs("toctou-p2")
 		return err == nil && len(jobs) == 1 && jobs[0].Status != ScanJobStatusRunning && jobs[0].Status != ScanJobStatusPending

internal/security/scanner/late_scan_r1_test.go

@@ -31,7 +31,7 @@ func TestPass2_R1_UsesOriginatingIncarnation(t *testing.T) {
 	svc.saveFailedPass2Job("acme", "source resolution failed", 1) // A's job was incarnation 1
 	assert.Empty(t, base.jobs, "A's Pass 2 placeholder is not stored under B")
 
-	svc.startPass2("acme", &ServerInfo{Name: "acme", Protocol: "stdio", Command: "true"}, 1)
+	svc.startPass2("acme", &ServerInfo{Name: "acme", Protocol: "stdio", Command: "true"}, 1, "")
 	assert.Nil(t, svc.engine.GetActiveJob("acme"), "no Pass 2 runs for a replaced incarnation")
 	assert.Empty(t, base.jobs)
 	assert.Empty(t, base.reports)

internal/security/scanner/service.go

@@ -1154,7 +1154,7 @@ func (a *scanCallbackAdapter) OnScanCompleted(job *ScanJob, reports []*ScanRepor
 			// The ORIGINATING job's incarnation (review round 1): a late
 			// Pass 1 of a removed server must not bind Pass 2 to a
 			// same-name replacement.
-			go a.service.startPass2(job.ServerName, a.serverInfo, job.Incarnation)
+			go a.service.startPass2(job.ServerName, a.serverInfo, job.Incarnation, job.ID)
 		}
 	}
 }
@@ -1453,7 +1453,9 @@ func (s *Service) StartScan(ctx context.Context, serverName string, dryRun bool,
 // inc is the incarnation of the Pass 1 job that scheduled it: Pass 2 and its
 // failed-source placeholder bind to it, and Pass 2 does not start at all once
 // that incarnation is no longer current (Spec 116 FR-014a, review round 1).
-func (s *Service) startPass2(serverName string, serverInfo *ServerInfo, inc int64) {
+// pass1JobID names the Pass-1 job that scheduled this audit ("" when none):
+// only that job's not-yet-cleared slot is waited on.
+func (s *Service) startPass2(serverName string, serverInfo *ServerInfo, inc int64, pass1JobID string) {
 	if !s.incarnationIsCurrent(serverName, inc) {
 		s.logger.Info("Pass 2 skipped: the server was removed or replaced since Pass 1 started",
 			zap.String("server", serverName), zap.Int64("incarnation", inc))
@@ -1554,14 +1556,47 @@ func (s *Service) startPass2(serverName string, serverInfo *ServerInfo, inc int6
 		scanPass: ScanPassSupplyChainAudit,
 	}
 
-	_, err := s.engine.StartScan(ctx, req, callback)
+	// Pass 1's OnScanCompleted (which spawns this goroutine) runs inside
+	// executeScan BEFORE its deferred clearActiveJob releases the server's
+	// slot, so a quick Pass 2 can find Pass 1's own job still registered and
+	// be rejected with ErrScanInProgress, silently dropping the audit. Wait
+	// (bounded) for that slot to free instead.
+	_, err := s.startScanAfterPass1Slot(ctx, req, callback, pass1JobID)
 	if err != nil {
 		s.logger.Warn("Failed to start Pass 2 scan",
 			zap.String("server", serverName),
 			zap.Error(err),
 		)
-		// If the engine rejected (e.g., scan already in progress for same server),
-		// just log and move on. Pass 2 is best-effort.
+		// The callback only owns the cleanup for an ACCEPTED scan.
+		if resolvedCleanup != nil {
+			resolvedCleanup()
+		}
+		s.saveFailedPass2Job(serverName, "supply chain audit could not start: "+err.Error(), inc)
+	}
+}
+
+// pass2SlotWait bounds how long Pass 2 waits for Pass 1's job to leave the
+// engine's active set (a var so tests can shrink it).
+var pass2SlotWait = 10 * time.Second
+
+// startScanAfterPass1Slot starts req, retrying while the only thing in the way
+// is the originating Pass-1 job (pass1JobID), which has finished its work but
+// not yet been cleared. Any other blocker (a user-triggered scan that started
+// in between, another Pass 2) is not waited on.
+func (s *Service) startScanAfterPass1Slot(ctx context.Context, req ScanRequest, cb ScanCallback, pass1JobID string) (*ScanJob, error) {
+	deadline := time.Now().Add(pass2SlotWait)
+	for {
+		job, err := s.engine.StartScan(ctx, req, cb)
+		if err == nil || !errors.Is(err, ErrScanInProgress) {
+			return job, err
+		}
+		if blocker := s.engine.GetActiveJob(req.ServerName); pass1JobID == "" || (blocker != nil && blocker.ID != pass1JobID) {
+			return job, err // not our Pass 1: do not wait on it
+		}
+		if !time.Now().Before(deadline) {
+			return job, err
+		}
+		time.Sleep(10 * time.Millisecond)
 	}
 }
 

internal/security/scanner/service_test.go

@@ -2963,3 +2963,65 @@ func TestGetScanReportByJobIDVerdictMatchesSummary(t *testing.T) {
 		t.Errorf("raw report Summary.Dangerous must keep the threat-level count, got %d", report.Summary.Dangerous)
 	}
 }
+
+// TestStartPass2WaitsForPass1SlotRelease: Pass 1's OnScanCompleted spawns Pass
+// 2 inside executeScan, before the deferred clearActiveJob frees the server's
+// engine slot. Pass 2 used to be rejected with "scan already in progress" and
+// silently dropped (no job ever saved) whenever it won that race. It must wait
+// for the slot instead.
+func TestStartPass2WaitsForPass1SlotRelease(t *testing.T) {
+	dir := t.TempDir()
+	logger := zap.NewNop()
+	store := newMockStorage()
+	svc := NewService(store, NewRegistry(dir, logger), NewDockerRunner(logger), dir, logger)
+	info := &ServerInfo{Name: "srv-slot", Protocol: "stdio", Command: "node", Args: []string{"server.js"}, WorkingDir: t.TempDir()}
+	svc.SetServerInfoProvider(&scanSpyProvider{info: info})
+	svc.SetDeepScan(true, nil)
+
+	// Simulate Pass 1 completed but not yet cleared.
+	pass1 := &ScanJob{ID: "pass1", ServerName: "srv-slot", Status: ScanJobStatusCompleted, ScanPass: ScanPassSecurityScan}
+	svc.engine.mu.Lock()
+	svc.engine.activeScans["srv-slot"] = pass1
+	svc.engine.mu.Unlock()
+	go func() {
+		time.Sleep(300 * time.Millisecond)
+		svc.engine.clearActiveJob("srv-slot", pass1)
+	}()
+
+	svc.startPass2("srv-slot", info, 0, "pass1")
+	waitForPass2Saved(t, store, "srv-slot")
+	jobs, _ := store.ListScanJobs("srv-slot")
+	for _, j := range jobs {
+		if j.ScanPass == ScanPassSupplyChainAudit && j.Status == ScanJobStatusFailed {
+			t.Fatalf("Pass 2 must run once the slot frees, got failed job: %s", j.Error)
+		}
+	}
+	waitForScanIdle(t, svc, "srv-slot")
+}
+
+// TestStartPass2DoesNotWaitOnUnrelatedPass1: when the slot is held by a
+// different Pass-1 job (e.g. a user-triggered rescan that started after the
+// originating Pass 1 was cleared), Pass 2 must not wait out pass2SlotWait on
+// it; it records a failed audit at once.
+func TestStartPass2DoesNotWaitOnUnrelatedPass1(t *testing.T) {
+	dir := t.TempDir()
+	logger := zap.NewNop()
+	store := newMockStorage()
+	svc := NewService(store, NewRegistry(dir, logger), NewDockerRunner(logger), dir, logger)
+	info := &ServerInfo{Name: "srv-other", Protocol: "stdio", Command: "node", Args: []string{"server.js"}, WorkingDir: t.TempDir()}
+	svc.SetServerInfoProvider(&scanSpyProvider{info: info})
+	svc.SetDeepScan(true, nil)
+
+	other := &ScanJob{ID: "user-rescan", ServerName: "srv-other", Status: ScanJobStatusRunning, ScanPass: ScanPassSecurityScan}
+	svc.engine.mu.Lock()
+	svc.engine.activeScans["srv-other"] = other
+	svc.engine.mu.Unlock()
+	t.Cleanup(func() { svc.engine.clearActiveJob("srv-other", other) })
+
+	start := time.Now()
+	svc.startPass2("srv-other", info, 0, "pass1")
+	if elapsed := time.Since(start); elapsed > 2*time.Second {
+		t.Fatalf("Pass 2 waited %s on an unrelated Pass-1 job", elapsed)
+	}
+	waitForPass2Saved(t, store, "srv-other")
+}

internal/security/scanner/source_staging_test.go

@@ -301,6 +301,7 @@ func TestStagedSource_CopyCancelledByDeadline(t *testing.T) {
 // when the caller passed context.Background().
 func TestStagedSource_ScanDeadlineCoversStaging(t *testing.T) {
 	source := writeToolsJSON(t, []map[string]interface{}{rt06Tool(rt06Poisoned)})
+	require.NoError(t, os.WriteFile(filepath.Join(source, "main.js"), []byte("x"), 0o644)) // a non-skipped entry for staging to check ctx on
 	snapshot, err := os.ReadFile(filepath.Join(source, "tools.json"))
 	require.NoError(t, err)
 	logger := zap.NewNop()
@@ -312,9 +313,16 @@ func TestStagedSource_ScanDeadlineCoversStaging(t *testing.T) {
 		return "", "", 0, nil
 	}
 	req := ScanRequest{ServerName: "dl", SourceDir: source, ToolDefinitions: snapshot}
-	plugin := &ScannerPlugin{ID: "mcp-scan", DockerImage: "example/mcp-scan:test", Command: []string{"scan"}, Timeout: "1ns"}
+	// A 1ns timeout is not reliably expired on coarse-clock platforms
+	// (Windows), so staging could finish before the deadline was observable.
+	// Use a short timeout and stall inside staging, after the first directory
+	// batch, until the deadline has certainly passed: the next ctx check in
+	// staging must then fail. This proves staging runs under the scan deadline.
+	t.Cleanup(func() { stagingAfterReadBatch = nil })
+	stagingAfterReadBatch = func(int) { time.Sleep(300 * time.Millisecond) }
+	plugin := &ScannerPlugin{ID: "mcp-scan", DockerImage: "example/mcp-scan:test", Command: []string{"scan"}, Timeout: "20ms"}
 	_, _, err = engine.runSingleScanner(context.Background(), plugin, req)
-	require.Error(t, err, "an already-expired scan deadline stops staging")
+	require.Error(t, err, "an expired scan deadline stops staging")
 	assert.ErrorIs(t, err, context.DeadlineExceeded)
 }