From 1853c0c1d3ee5d5406dc43ad477f84e05cf85d11 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 5 Sep 2026 15:41:27 -0500 Subject: [PATCH] hold/scanner: detect dead connections and get storage off the reader Three defects that all turn on whether the hold knows what a scanner is doing. Neither end had a keepalive, a read deadline, or a read limit, so a half-open connection was invisible until some other timeout fired. That got worse with capacity-aware dispatch: a dead-but-connected scanner holds its advertised worker count out of the budget and keeps winning jobs. Both ends now ping every 30s against a 90s read deadline, so three unanswered pings condemn a connection. Detection takes about 90 seconds, after which the existing reconnect grace reclaims the rows, against the 60-minute queueing timeout that was previously the only escape. Liveness decides when a scanner is gone; the grace window decides when its work is reassignable. Read limits are asymmetric and deliberately generous, because exceeding one closes the connection rather than truncating, which would turn a large but legitimate result into a permanent retry loop. Write deadlines were absent everywhere; the scanner in particular held a mutex across an unbounded write, so a wedged write silenced it without disconnecting it. handleResult did two S3 uploads and a CAR commit inline on the reader goroutine, so a slow S3 looked exactly like a dead scanner. Terminal messages now go to a per-subscriber storage goroutine while acks and starts stay on the reader. One goroutine, not a pool: every path ends in CreateScanRecord, which serialises on the repo lock anyway, and ordering is worth more than parallelism that cannot be used. Uploads and the record write get separate budgets, so a stalled upload cannot spend the time the record needs and the record write stays unconditional. checkPredecessor cached an inconclusive answer as a definitive negative in a map that is never reset, so one unreachable hold meant its manifests were never scanned again for the life of the process. gc.go already carried the corrected logic for the same problem; this follows it rather than inventing a second approach, and also stops treating an unparseable captain record as definitive. The test PDS had to move off ":memory:", which go-libsql scopes per connection: writing a scan record from any goroutine but the caller's got a connection with no tables. That was invisible while every record write happened inline. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01U1Km3N3uUmeGaj7VbaM8PF --- pkg/hold/pds/scan_broadcaster.go | 401 ++++++++++++++-- .../pds/scan_broadcaster_liveness_test.go | 453 ++++++++++++++++++ .../pds/scan_broadcaster_predecessor_test.go | 173 +++++++ pkg/hold/pds/scan_broadcaster_test.go | 39 +- scanner/internal/client/hold.go | 119 ++++- scanner/internal/client/hold_test.go | 156 ++++++ 6 files changed, 1302 insertions(+), 39 deletions(-) create mode 100644 pkg/hold/pds/scan_broadcaster_liveness_test.go create mode 100644 pkg/hold/pds/scan_broadcaster_predecessor_test.go create mode 100644 scanner/internal/client/hold_test.go diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index a5bc4e7..6f2aff5 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -6,9 +6,11 @@ import ( "database/sql" "encoding/hex" "encoding/json" + "errors" "fmt" "io" "log/slog" + "net" "net/http" "net/url" "strings" @@ -95,6 +97,46 @@ const ( // a dispatch budget, so it is clamped rather than trusted. maxScannerCapacity = 32 + // scannerPingInterval is how often the hold pings an idle scanner + // connection. + scannerPingInterval = 30 * time.Second + + // scannerPongWait is how long a connection may go without a pong (or any + // other frame) before the hold treats it as dead. Three missed pings. + scannerPongWait = 90 * time.Second + + // scannerWriteWait bounds a single write to a scanner. + scannerWriteWait = 30 * time.Second + + // scannerMaxMessageSize caps one frame from a scanner. gorilla's default is + // unlimited, so a broken or hostile peer can make the hold allocate + // whatever it likes. + // + // The ceiling has to be generous, because a result message carries the + // whole SBOM and the whole Grype report inline as JSON strings: exceeding + // the limit is not a truncation but a connection close, and a legitimate + // result that trips it becomes a permanent retry loop for that image. The + // largest report measured in this repo is 1.8 MB for 125 matches; 128 MiB + // is roughly seventy times the largest frame anyone has documented here, + // while still bounding a single allocation to something the host survives. + scannerMaxMessageSize = 128 << 20 + + // scannerResultTimeout bounds the blob uploads one result message costs. + scannerResultTimeout = 5 * time.Minute + + // scannerRecordTimeout bounds the PDS record write a terminal message + // costs. It is deliberately a separate budget from scannerResultTimeout: a + // stalled S3 endpoint must not be able to spend the time the record write + // needs. The record is the only thing that stops the appview reporting the + // manifest as never scanned, so it is the last thing that may be skipped. + scannerRecordTimeout = 60 * time.Second + + // scannerStorageQueueDepth is how many terminal messages one connection may + // have waiting on storage before the reader has to wait for room. It is + // small on purpose: each queued message holds an entire SBOM and Grype + // report in memory, and the hold shares a 1 GiB host with the scanner. + scannerStorageQueueDepth = 8 + // maxScannerInstanceID bounds the scanner-supplied identity that becomes // assigned_to. Same reasoning: it is client input that ends up in a // database column. @@ -128,13 +170,30 @@ type ScanBroadcaster struct { ownsDB bool // true when this broadcaster opened the connection itself // Proactive scan scheduling - rescanInterval time.Duration // Minimum interval between re-scans (0 = disabled) - stopCh chan struct{} // Signal to stop background goroutines - wg sync.WaitGroup // Wait for background goroutines to finish - predecessorCache map[string]bool // holdDID → "has this hold been migrated (has successor)?" - relayEndpoints []string // Relay URLs for listReposByCollection (failover order) - relayStartIdx int // Rotates per discovery pass so load is shared across endpoints - relayStartMu sync.Mutex + rescanInterval time.Duration // Minimum interval between re-scans (0 = disabled) + stopCh chan struct{} // Signal to stop background goroutines + wg sync.WaitGroup // Wait for background goroutines to finish + // predecessorCache answers "has this hold been migrated into us?" without + // re-dialling. Only definitive answers belong here. The cache is never + // reset, so a false recorded from an unreachable hold would outlive the + // outage and stop that hold's manifests being scanned for the life of the + // process. + predecessorCache map[string]bool + + // predecessorUnresolved holds the DIDs whose predecessor status could not + // be determined during the current discovery pass. It exists only so that + // one unreachable hold costs a single 5s timeout per pass rather than one + // per manifest, and it is cleared at the start of every pass so a hold that + // was down once is re-checked next time instead of being written off. + predecessorUnresolved map[string]bool + + // predecessorMu guards both maps. Only the discovery goroutine reaches them + // today; the lock is what makes that a property of the code rather than of + // the current call graph. + predecessorMu sync.Mutex + relayEndpoints []string // Relay URLs for listReposByCollection (failover order) + relayStartIdx int // Rotates per discovery pass so load is shared across endpoints + relayStartMu sync.Mutex // Work queues for proactive scanning (populated by discovery/stale goroutines) unscannedQueue chan *scanCandidate // Medium priority: manifests with no scan record @@ -148,6 +207,47 @@ type ScanBroadcaster struct { // activeJobsErrs counts consecutive hasActiveJobs query failures, so a // persistent database fault can fail open instead of freezing dispatch. activeJobsErrs atomic.Int64 + + // WebSocket liveness knobs. Zero means "use the package default"; tests + // shrink them so a half-open connection can be provoked in milliseconds + // rather than in the minute and a half production waits. + pingInterval time.Duration + pongWait time.Duration + writeWait time.Duration + + // resultTimeout bounds the off-reader work a terminal message triggers. + // Zero means the package default. + resultTimeout time.Duration +} + +// pingEvery, pongDeadline, writeDeadline and resultDeadline apply the package +// defaults to the per-broadcaster knobs above. +func (sb *ScanBroadcaster) pingEvery() time.Duration { + if sb.pingInterval > 0 { + return sb.pingInterval + } + return scannerPingInterval +} + +func (sb *ScanBroadcaster) pongDeadline() time.Duration { + if sb.pongWait > 0 { + return sb.pongWait + } + return scannerPongWait +} + +func (sb *ScanBroadcaster) writeDeadline() time.Duration { + if sb.writeWait > 0 { + return sb.writeWait + } + return scannerWriteWait +} + +func (sb *ScanBroadcaster) resultDeadline() time.Duration { + if sb.resultTimeout > 0 { + return sb.resultTimeout + } + return scannerResultTimeout } // ScanSubscriber represents a connected scanner WebSocket client @@ -161,6 +261,17 @@ type ScanSubscriber struct { // connect. It is the unit both the proactive dispatch depth and // per-scanner admission control are counted in. capacity int + + // storage carries the work a terminal message triggers off the reader + // goroutine, and readerDone says when nothing more will be submitted. + // + // One goroutine drains storage, so messages are handled in the order they + // arrived. That matters: two messages about one job must not be applied out + // of order, and the alternative — a pool that hashes seq to a worker — buys + // parallelism the PDS cannot use anyway, since CreateScanRecord serialises + // on the hold's single per-uid repo lock. + storage chan func() + readerDone chan struct{} } // effectiveCapacity is capacity with the pre-declaration default applied. @@ -533,10 +644,12 @@ func (sb *ScanBroadcaster) enqueue(job *ScanJobEvent, origin string) error { func (sb *ScanBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, instanceID string, capacity int) *ScanSubscriber { id := sb.subscriberID(instanceID) sub := &ScanSubscriber{ - conn: conn, - send: make(chan *ScanJobEvent, 20), - id: id, - done: make(chan struct{}), + conn: conn, + send: make(chan *ScanJobEvent, 20), + id: id, + done: make(chan struct{}), + storage: make(chan func(), scannerStorageQueueDepth), + readerDone: make(chan struct{}), } sub.capacity = capacity @@ -557,9 +670,13 @@ func (sb *ScanBroadcaster) Subscribe(conn *websocket.Conn, cursor int64, instanc "capacity", sub.effectiveCapacity(), "totalSubscribers", total) - // Start writer goroutine (sends jobs to scanner) + // Start writer goroutine (sends jobs to scanner, and pings it) go sb.handleWriter(sub) + // Start the storage goroutine before the reader, so the reader always has + // somewhere to hand a terminal message. + go sb.handleStorage(sub) + // Start reader goroutine (receives acks/results/errors from scanner) go sb.handleReader(sub) @@ -851,6 +968,12 @@ func (sb *ScanBroadcaster) subscriberLoads() (map[string]int, bool) { func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) { defer sub.conn.Close() + // The writer owns every write on this connection, which is what makes the + // keepalive possible: gorilla permits one writer at a time, so the ping has + // to come from here rather than from a timer of its own. + ping := time.NewTicker(sb.pingEvery()) + defer ping.Stop() + // 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. @@ -859,6 +982,20 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) { select { case <-sub.done: return + case <-ping.C: + // Every write now carries a deadline. Without one a write into a + // socket whose peer has stopped reading blocks until the kernel + // gives up, which for a half-open connection is on the order of + // hours, and this goroutine is what the whole connection's + // liveness rests on. + if err := sub.conn.WriteControl(websocket.PingMessage, nil, + time.Now().Add(sb.writeDeadline())); err != nil { + slog.Warn("Failed to ping scanner, dropping the connection", + "subscriberId", sub.id, "error", err) + sb.Unsubscribe(sub) + return + } + continue case job = <-sub.send: } @@ -868,6 +1005,12 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) { continue } + if err := sub.conn.SetWriteDeadline(time.Now().Add(sb.writeDeadline())); err != nil { + slog.Error("Failed to set write deadline for scan job", + "seq", job.Seq, "subscriberId", sub.id, "error", err) + sb.Unsubscribe(sub) + return + } if err := sub.conn.WriteMessage(websocket.TextMessage, data); err != nil { slog.Error("Failed to write scan job to WebSocket", "seq", job.Seq, @@ -879,20 +1022,63 @@ func (sb *ScanBroadcaster) handleWriter(sub *ScanSubscriber) { } } -// handleReader receives ack/result/error messages from a scanner +// handleReader receives ack/result/error messages from a scanner. +// +// Two rules hold here, and they are the same rule seen from two sides: this +// goroutine must never stop reading, and what it reads must prove the scanner +// is alive. +// +// It never stops reading because the storage a terminal message triggers — two +// S3 uploads and a CAR commit — is handed to handleStorage instead of being run +// inline. It used to run here, and while it ran every other job's ack, started, +// result and error on that connection sat unread in the socket buffer, with no +// deadline on any of it. +// +// Liveness comes from the read deadline, refreshed by a pong and by any other +// frame. Nothing used to ask: a connection that was open at this end and gone +// at the other stayed in sb.subscribers for the life of the process, holding +// its advertised worker count out of the dispatch budget and winning jobs that +// could only ever time out. The deadline is three ping intervals, so it takes +// three unanswered pings to condemn a connection. func (sb *ScanBroadcaster) handleReader(sub *ScanSubscriber) { - defer sb.Unsubscribe(sub) + defer func() { + // Before Unsubscribe, so handleStorage learns that nothing more will be + // submitted only once this goroutine has genuinely stopped submitting. + close(sub.readerDone) + sb.Unsubscribe(sub) + }() + + sub.conn.SetReadLimit(scannerMaxMessageSize) + refreshRead := func() { + if err := sub.conn.SetReadDeadline(time.Now().Add(sb.pongDeadline())); err != nil { + slog.Debug("Failed to set scanner read deadline", + "subscriberId", sub.id, "error", err) + } + } + refreshRead() + sub.conn.SetPongHandler(func(string) error { + refreshRead() + return nil + }) for { _, data, err := sub.conn.ReadMessage() if err != nil { - if websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure) { + switch { + case isReadTimeout(err): + slog.Warn("Scanner stopped answering, dropping the connection", + "subscriberId", sub.id, + "silentFor", sb.pongDeadline(), + "error", err) + case websocket.IsUnexpectedCloseError(err, websocket.CloseGoingAway, websocket.CloseNormalClosure): slog.Error("Scanner WebSocket read error", "subscriberId", sub.id, "error", err) } return } + // Any frame is proof of life, not just a pong. + refreshRead() var msg ScannerMessage if err := json.Unmarshal(data, &msg); err != nil { @@ -904,19 +1090,73 @@ func (sb *ScanBroadcaster) handleReader(sub *ScanSubscriber) { switch msg.Type { case "ack": + // One guarded UPDATE. Cheap enough to stay on the reader, and + // keeping it here means an ack is never delayed behind a scan + // record write. sb.handleAck(sub, msg.Seq) case "started": sb.handleStarted(sub, msg.Seq) case "result": - sb.handleResult(sub, msg) + sb.submitStorage(sub, func() { sb.handleResult(sub, msg) }) case "error": - sb.handleError(sub, msg) + sb.submitStorage(sub, func() { sb.handleError(sub, msg) }) case "skipped": - sb.handleSkipped(sub, msg) + sb.submitStorage(sub, func() { sb.handleSkipped(sub, msg) }) default: slog.Warn("Unknown scanner message type", "type", msg.Type, "subscriberId", sub.id) + continue + } + + // Handing work over can block when storage is backed up, and time + // spent waiting for our own storage is not evidence against the + // scanner. Restart the clock rather than counting that wait against it. + refreshRead() + } +} + +// isReadTimeout reports whether a read failed because the deadline expired +// rather than because the peer or the network did something. +func isReadTimeout(err error) bool { + var netErr net.Error + return errors.As(err, &netErr) && netErr.Timeout() +} + +// submitStorage hands a terminal message's work to handleStorage. +// +// The send is a plain blocking one, and safe: handleReader is the only sender, +// and handleStorage does not exit until handleReader has closed readerDone and +// the queue has drained. A verdict is never dropped — it is the only thing that +// stops the appview reporting the manifest as never scanned. +func (sb *ScanBroadcaster) submitStorage(sub *ScanSubscriber, fn func()) { + sub.storage <- fn +} + +// handleStorage runs the storage work terminal messages trigger, one at a time +// and in the order the messages arrived. +// +// Serial and ordered is the point, not a limitation. Two messages about one job +// applied out of order would let a stale verdict overwrite a fresh one, and the +// concurrency there was worth nothing anyway: every one of these writes ends in +// CreateScanRecord, which serialises on the hold's single per-uid repo lock. +func (sb *ScanBroadcaster) handleStorage(sub *ScanSubscriber) { + for { + select { + case fn := <-sub.storage: + fn() + case <-sub.readerDone: + // The reader has stopped. Finish what it already handed over: those + // are verdicts the scanner really produced, and dropping one leaves + // its manifest looking unscanned until the next stale pass. + for { + select { + case fn := <-sub.storage: + fn() + default: + return + } + } } } } @@ -1014,9 +1254,23 @@ func (sb *ScanBroadcaster) claimJobForTerminal(sub *ScanSubscriber, seq int64, k return true } -// handleResult processes a completed scan result: uploads SBOM blob + stores scan record in PDS +// handleResult processes a completed scan result: uploads SBOM blob + stores scan record in PDS. +// +// It runs on handleStorage's goroutine, not on the reader's, and on a context +// with a deadline. It used to be neither: an S3 endpoint that accepted the +// connection and then stalled held the entire scanner connection hostage +// indefinitely, and because the scanner was still nominally connected nothing +// requeued the work either. +// +// The two budgets are separate on purpose. The uploads get scannerResultTimeout +// between them; the record write gets its own fresh scannerRecordTimeout, so a +// stalled upload cannot spend the time the record needs. The record is written +// whichever way the uploads went — a blob already in S3 with no record pointing +// at it is orphaned, and a manifest with no record at all reads as never +// scanned. func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) { - ctx := context.Background() + ctx, cancel := context.WithTimeout(context.Background(), sb.resultDeadline()) + defer cancel() // Before the S3 uploads, not after: a scanner that does not hold this job // should not get its payload stored either. @@ -1102,7 +1356,10 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) "atcr-scanner-v1.0.0", ) - rpath, _, err := sb.pds.CreateScanRecord(ctx, scanRecord) + recordCtx, recordCancel := context.WithTimeout(context.Background(), scannerRecordTimeout) + defer recordCancel() + + rpath, _, err := sb.pds.CreateScanRecord(recordCtx, scanRecord) if err != nil { slog.Error("Failed to store scan record in PDS", "seq", msg.Seq, @@ -1157,7 +1414,11 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) // loop won't immediately retry. Failed records still get retried on the // rescan interval since failures may be transient (network, OOM, etc.). func (sb *ScanBroadcaster) handleError(sub *ScanSubscriber, msg ScannerMessage) { - ctx := context.Background() + // Bounded, and off the reader goroutine: this writes a CAR delta under the + // hold's per-uid repo lock, which every layer record, stats increment and + // Bluesky post also contends for. + ctx, cancel := context.WithTimeout(context.Background(), scannerRecordTimeout) + defer cancel() if !sb.claimJobForTerminal(sub, msg.Seq, "error") { return @@ -1206,7 +1467,8 @@ func (sb *ScanBroadcaster) handleError(sub *ScanSubscriber, msg ScannerMessage) // status="skipped". The stale-scan loop will leave these records alone — the // outcome won't change until the scanner gains support for the artifact type. func (sb *ScanBroadcaster) handleSkipped(sub *ScanSubscriber, msg ScannerMessage) { - ctx := context.Background() + ctx, cancel := context.WithTimeout(context.Background(), scannerRecordTimeout) + defer cancel() if !sb.claimJobForTerminal(sub, msg.Seq, "skipped") { return @@ -1706,6 +1968,12 @@ func (sb *ScanBroadcaster) runDiscoveryPass() { ctx, cancel := context.WithTimeout(context.Background(), 30*time.Minute) defer cancel() + // A hold that could not be reached last pass gets asked again this one. + // predecessorUnresolved is a within-pass memo, not a verdict. + sb.predecessorMu.Lock() + sb.predecessorUnresolved = nil + sb.predecessorMu.Unlock() + // Fetch DID list from relay userDIDs := sb.fetchManifestDIDs(ctx) if len(userDIDs) == 0 { @@ -2238,6 +2506,15 @@ func (sb *ScanBroadcaster) resolveManifestForCandidate(ctx context.Context, cand // isOurManifest checks if a manifest's holdDID matches this hold directly, // or if the manifest's hold has been migrated (has a successor label set). +// +// An answer we could not get is not an answer of "no". A predecessor is by +// definition a retired hold, so being briefly unreachable is its normal +// condition, and this used to cache the resulting false forever: one slow reply +// during the first discovery pass after boot meant every manifest naming that +// hold went unscanned for the life of the process, silently, at Debug level. +// pkg/hold/gc hit the same defect with a worse blast radius (deleted blobs +// rather than missed scans) and fixed it with the definitive/unresolved split +// this follows. func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bool { if holdDID == "" { return false @@ -2248,20 +2525,48 @@ func (sb *ScanBroadcaster) isOurManifest(ctx context.Context, holdDID string) bo return true } - // Check predecessor cache + // Held across the fetch. The lock is uncontended today, and holding it also + // means two callers can never dial the same unreachable hold at once. + sb.predecessorMu.Lock() + defer sb.predecessorMu.Unlock() + if isPredecessor, cached := sb.predecessorCache[holdDID]; cached { return isPredecessor } + // Already unreachable earlier in this pass. Answer the same way without + // paying another timeout; the next pass starts fresh and re-checks. + if sb.predecessorUnresolved[holdDID] { + return false + } + // Fetch captain record from the other hold's PDS to check successor - isPredecessor := sb.checkPredecessor(ctx, holdDID) + isPredecessor, definitive := sb.checkPredecessor(ctx, holdDID) + if !definitive { + if sb.predecessorUnresolved == nil { + sb.predecessorUnresolved = make(map[string]bool) + } + sb.predecessorUnresolved[holdDID] = true + slog.Warn("Proactive scan: predecessor status unresolved, not adopting this hold's manifests this pass", + "holdDID", holdDID) + return false + } + + if sb.predecessorCache == nil { + sb.predecessorCache = make(map[string]bool) + } sb.predecessorCache[holdDID] = isPredecessor return isPredecessor } // checkPredecessor fetches a hold's captain record to check if it has a successor label // (meaning the hold has been migrated/retired and its manifests should be scanned by us). -func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) bool { +// +// The second return value reports whether the answer is definitive. It is false +// whenever the hold could not be reached or its reply could not be understood: +// those are cases where the hold may well be a predecessor and we simply cannot +// tell. Callers must not read an inconclusive result as "not a predecessor". +func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) (isPredecessor, definitive bool) { fetchCtx, cancel := context.WithTimeout(ctx, 5*time.Second) defer cancel() @@ -2269,9 +2574,21 @@ func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) if err != nil { slog.Debug("Proactive scan: failed to resolve predecessor hold URL", "holdDID", holdDID, "error", err) - return false + return false, false } + return sb.checkPredecessorAt(fetchCtx, holdDID, holdURL) +} + +// checkPredecessorAt is checkPredecessor with the hold's base URL already +// resolved, split out so the fetch-and-parse half can be exercised against a +// local server. It carries the same contract: the second return value is false +// whenever the answer is inconclusive rather than negative. +// +// A non-200 is inconclusive rather than negative on the same reasoning as +// pkg/hold/gc: a reachable service that cannot produce its own captain record +// is malfunctioning, not answering. +func (sb *ScanBroadcaster) checkPredecessorAt(ctx context.Context, holdDID, holdURL string) (isPredecessor, definitive bool) { // Fetch captain record: com.atproto.repo.getRecord recordURL := fmt.Sprintf("%s/xrpc/com.atproto.repo.getRecord?repo=%s&collection=%s&rkey=self", holdURL, @@ -2279,53 +2596,63 @@ func (sb *ScanBroadcaster) checkPredecessor(ctx context.Context, holdDID string) url.QueryEscape(atproto.CaptainCollection), ) - req, err := http.NewRequestWithContext(fetchCtx, "GET", recordURL, nil) + req, err := http.NewRequestWithContext(ctx, "GET", recordURL, nil) if err != nil { - return false + return false, false } resp, err := http.DefaultClient.Do(req) if err != nil { slog.Debug("Proactive scan: failed to fetch predecessor captain record", "holdDID", holdDID, "error", err) - return false + return false, false } defer resp.Body.Close() if resp.StatusCode != http.StatusOK { - return false + slog.Debug("Proactive scan: predecessor captain record fetch returned non-200", + "holdDID", holdDID, "status", resp.StatusCode) + return false, false } body, err := io.ReadAll(io.LimitReader(resp.Body, 1<<20)) // 1MB limit if err != nil { - return false + slog.Debug("Proactive scan: failed to read predecessor captain record", + "holdDID", holdDID, "error", err) + return false, false } var envelope struct { Value json.RawMessage `json:"value"` } if err := json.Unmarshal(body, &envelope); err != nil { - return false + slog.Debug("Proactive scan: failed to parse predecessor captain envelope", + "holdDID", holdDID, "error", err) + return false, false } var captain atproto.CaptainRecord if err := json.Unmarshal(envelope.Value, &captain); err != nil { - return false + slog.Debug("Proactive scan: failed to parse predecessor captain record", + "holdDID", holdDID, "error", err) + return false, false } + // The hold answered and declares no successor. This is the one negative we + // are entitled to cache. if captain.Successor == "" { - return false + return false, true } if captain.Successor != sb.holdDID { slog.Debug("Proactive scan: hold has successor, but it is not us", "holdDID", holdDID, "successor", captain.Successor, "ourHoldDID", sb.holdDID) - return false + return false, true } slog.Info("Proactive scan: discovered migrated hold pointing at us as successor", "holdDID", holdDID, "successor", captain.Successor) - return true + return true, true } // hasConnectedScanners returns true if at least one scanner is connected. diff --git a/pkg/hold/pds/scan_broadcaster_liveness_test.go b/pkg/hold/pds/scan_broadcaster_liveness_test.go new file mode 100644 index 0000000..ee82f55 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_liveness_test.go @@ -0,0 +1,453 @@ +package pds + +import ( + "context" + "net/http" + "net/http/httptest" + "strings" + "sync" + "testing" + "time" + + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + "github.com/gorilla/websocket" + + "atcr.io/pkg/s3" +) + +// wsConnPair is wsPair with the client side handed back too, because these +// tests need to decide whether the client answers a ping and what it writes. +// A gorilla connection only replies to a ping from inside ReadMessage, so a +// client that never reads is exactly a scanner whose process is gone while its +// TCP connection survives: NAT expiry, a load balancer, a host that vanished. +func wsConnPair(t *testing.T) (server, client *websocket.Conn) { + t.Helper() + + upgrader := websocket.Upgrader{} + accepted := make(chan *websocket.Conn, 1) + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + t.Errorf("upgrade: %v", err) + return + } + accepted <- conn + })) + t.Cleanup(srv.Close) + + cl, _, err := websocket.DefaultDialer.Dial("ws"+strings.TrimPrefix(srv.URL, "http"), nil) + if err != nil { + t.Fatalf("dial: %v", err) + } + t.Cleanup(func() { _ = cl.Close() }) + + select { + case conn := <-accepted: + return conn, cl + case <-time.After(5 * time.Second): + t.Fatal("server never completed the upgrade") + return nil, nil + } +} + +// waitFor polls until cond is true, failing with msg if it never becomes true. +func waitFor(t *testing.T, within time.Duration, msg string, cond func() bool) { + t.Helper() + + deadline := time.Now().Add(within) + for time.Now().Before(deadline) { + if cond() { + return + } + time.Sleep(2 * time.Millisecond) + } + t.Fatal(msg) +} + +// TestScanBroadcaster_HalfOpenScannerIsDisconnected is the hold half of F7. +// +// Nothing in the hold ever asked a scanner connection whether it was still +// there: no ping, no read deadline, no pong handler. A connection that is open +// at this end and gone at the other therefore stays in sb.subscribers for as +// long as the process lives. That is not merely untidy now that a scanner +// declares a worker count on connect: the dead subscriber's advertised +// capacity is subtracted from the dispatch budget, it keeps winning jobs in +// selectSubscriberLocked, and every row it wins sits marked as its own until +// the queueing timeout fires an hour later. +// +// The client here never calls ReadMessage, so it never answers a ping — the +// half-open shape, reproduced without needing to break a network. +func TestScanBroadcaster_HalfOpenScannerIsDisconnected(t *testing.T) { + sb := newTestScanBroadcaster(t) + sb.pingInterval = 20 * time.Millisecond + sb.pongWait = 200 * time.Millisecond + sb.writeWait = time.Second + + server, _ := wsConnPair(t) + sb.Subscribe(server, 0, "silent-scanner", 4) + + waitFor(t, 5*time.Second, + "a scanner that answers no pings is still subscribed: its advertised "+ + "capacity stays out of the dispatch budget and it keeps winning jobs", + func() bool { return subscriberCount(sb) == 0 }) +} + +// The other side of the same rule: a scanner that is merely idle must not be +// disconnected. gorilla answers a ping from inside ReadMessage, so a client +// running a read loop pongs without the test doing anything, which is what the +// real scanner does. +func TestScanBroadcaster_IdleButResponsiveScannerStaysSubscribed(t *testing.T) { + sb := newTestScanBroadcaster(t) + sb.pingInterval = 20 * time.Millisecond + sb.pongWait = 200 * time.Millisecond + sb.writeWait = time.Second + + server, client := wsConnPair(t) + go func() { + for { + if _, _, err := client.ReadMessage(); err != nil { + return + } + } + }() + + sb.Subscribe(server, 0, "live-scanner", 4) + + // Comfortably more than several pongWait periods. A liveness check that + // fires on silence rather than on unanswered pings would have cut this + // connection many times over by now. + time.Sleep(time.Second) + if got := subscriberCount(sb); got != 1 { + t.Fatalf("subscriber count = %d, want 1: a responsive but idle scanner was disconnected", got) + } +} + +// A read limit must not be able to truncate a legitimate result. Results carry +// the whole SBOM and the whole Grype report inline as JSON strings, and the +// largest report measured in this repo is 1.8 MB for 125 matches, with a big +// image's SBOM larger still. This sends a multi-megabyte SBOM through a real +// socket and requires it to arrive whole and be stored. +func TestScanBroadcaster_LargeResultIsNotTruncated(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sb.pingInterval = 50 * time.Millisecond + sb.pongWait = 5 * time.Second + sb.writeWait = 5 * time.Second + + server, client := wsConnPair(t) + go func() { + for { + if _, _, err := client.ReadMessage(); err != nil { + return + } + } + }() + + const digest = "sha256:bigsbom" + seq := seedJobWithDigest(t, sb, digest) + + const instance = "big-result-scanner" + if _, err := sb.db.Exec( + `UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`, + instance, time.Now(), seq); err != nil { + t.Fatalf("assign: %v", err) + } + + sb.Subscribe(server, 0, instance, 1) + + // 12 MiB of SBOM: several times the largest report this repo has measured, + // and far past any frame size a naive limit would have picked. + big := `{"spdxVersion":"SPDX-2.3","filler":"` + strings.Repeat("a", 12<<20) + `"}` + if err := client.WriteJSON(map[string]any{ + "type": "result", "seq": seq, "sbom": big, + }); err != nil { + t.Fatalf("write large result: %v", err) + } + + waitFor(t, 30*time.Second, + "a large but legitimate result never completed: the read limit truncated it "+ + "or the connection was closed under it", + func() bool { return jobStatus(t, sb, seq) == "completed" }) + + _, record, err := sb.pds.GetScanRecord(context.Background(), digest) + if err != nil { + t.Fatalf("no scan record for the large result: %v", err) + } + if record.SbomBlob == nil { + t.Error("scan record carries no SBOM blob") + } +} + +// blockingS3 is a MockS3Client whose PutObject can be parked until the test +// lets it go, which is how a slow S3 endpoint is reproduced without a network. +// +// blockFirst parks only the first upload. That asymmetry is what makes +// out-of-order handling visible: if a later message is handled while an earlier +// one is still uploading, the later one's upload runs straight through and +// finishes first. +type blockingS3 struct { + *s3.MockS3Client + entered chan struct{} + release chan struct{} + completed chan int64 // ContentLength of each upload, as it finishes + blockFirst bool + + mu sync.Mutex + seen int +} + +func (b *blockingS3) PutObject(ctx context.Context, in *awss3.PutObjectInput, opts ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) { + b.mu.Lock() + b.seen++ + first := b.seen == 1 + b.mu.Unlock() + + if !b.blockFirst || first { + // Buffered and non-blocking: one result can trigger two uploads, and + // the test only ever waits for the first. + select { + case b.entered <- struct{}{}: + default: + } + select { + case <-b.release: + case <-ctx.Done(): + return nil, ctx.Err() + } + } + + out, err := b.MockS3Client.PutObject(ctx, in, opts...) + if in.ContentLength != nil { + select { + case b.completed <- *in.ContentLength: + default: + } + } + return out, err +} + +func newBlockingS3(t *testing.T, sb *ScanBroadcaster) *blockingS3 { + t.Helper() + + b := &blockingS3{ + MockS3Client: s3.NewMockS3Client(""), + entered: make(chan struct{}, 64), + release: make(chan struct{}), + completed: make(chan int64, 64), + } + sb.s3 = &s3.S3Service{Client: b, Bucket: "test-bucket"} + return b +} + +// TestScanBroadcaster_ReaderKeepsReadingDuringResultStorage is F6. +// +// handleResult ran inline on handleReader's goroutine: two S3 uploads and a CAR +// commit, on a context.Background() with no deadline. For the whole of that the +// subscriber's socket is not being read, so every other job's ack, started, +// result and error sits in the kernel buffer. It also interacts badly with the +// liveness work above — a scanner stalled behind a slow S3 upload looks exactly +// like a scanner that died, because the pong that would prove otherwise is only +// processed from inside ReadMessage. +// +// The interleaving is built, not raced: the S3 stand-in parks inside the upload +// on an unbuffered channel and stays there until the test releases it. +func TestScanBroadcaster_ReaderKeepsReadingDuringResultStorage(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + blocker := newBlockingS3(t, sb) + sb.pingInterval = 50 * time.Millisecond + sb.pongWait = 10 * time.Second + sb.writeWait = 5 * time.Second + + server, client := wsConnPair(t) + go func() { + for { + if _, _, err := client.ReadMessage(); err != nil { + return + } + } + }() + + const instance = "busy-scanner" + slow := seedJobWithDigest(t, sb, "sha256:slowupload") + other := seedJobWithDigest(t, sb, "sha256:otherjob") + for _, seq := range []int64{slow, other} { + if _, err := sb.db.Exec( + `UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`, + instance, time.Now(), seq); err != nil { + t.Fatalf("assign %d: %v", seq, err) + } + } + + sb.Subscribe(server, 0, instance, 2) + + if err := client.WriteJSON(map[string]any{ + "type": "result", "seq": slow, "sbom": testSBOM, + }); err != nil { + t.Fatalf("write result: %v", err) + } + + // The upload is now parked. Whatever happens next happens while S3 is slow. + select { + case <-blocker.entered: + case <-time.After(10 * time.Second): + t.Fatal("the SBOM upload never started") + } + + if err := client.WriteJSON(map[string]any{"type": "ack", "seq": other}); err != nil { + t.Fatalf("write ack: %v", err) + } + + waitFor(t, 5*time.Second, + "a second job's ack went unread while an S3 upload was in flight: the "+ + "reader goroutine is doing storage work", + func() bool { return jobStatus(t, sb, other) == "processing" }) + + close(blocker.release) + + waitFor(t, 10*time.Second, + "the released result never completed", + func() bool { return jobStatus(t, sb, slow) == "completed" }) +} + +// The bound on that off-reader work. An S3 endpoint that accepts the connection +// and then stalls used to hold a result forever, because the context carried no +// deadline at all. It must now give up and still leave the row in a terminal +// state rather than in limbo. +func TestScanBroadcaster_ResultStorageIsBounded(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + blocker := newBlockingS3(t, sb) + sb.resultTimeout = 300 * time.Millisecond + sb.pingInterval = 50 * time.Millisecond + sb.pongWait = 10 * time.Second + sb.writeWait = 5 * time.Second + + server, client := wsConnPair(t) + go func() { + for { + if _, _, err := client.ReadMessage(); err != nil { + return + } + } + }() + + const instance = "stalled-s3-scanner" + seq := seedJobWithDigest(t, sb, "sha256:stalleds3") + if _, err := sb.db.Exec( + `UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`, + instance, time.Now(), seq); err != nil { + t.Fatalf("assign: %v", err) + } + + sb.Subscribe(server, 0, instance, 1) + + if err := client.WriteJSON(map[string]any{ + "type": "result", "seq": seq, "sbom": testSBOM, + }); err != nil { + t.Fatalf("write result: %v", err) + } + + select { + case <-blocker.entered: + case <-time.After(10 * time.Second): + t.Fatal("the SBOM upload never started") + } + // blocker.release is never closed: S3 has stalled for good. + + waitFor(t, 10*time.Second, + "a stalled S3 upload never gave up: the result work carries no deadline", + func() bool { + s := jobStatus(t, sb, seq) + return s == "completed" || s == "failed" + }) + + // The record write is unconditional. It is the only thing that stops the + // appview showing this manifest as never scanned, and it must not be + // spent out of the same budget the stalled upload just exhausted. + if _, _, err := sb.pds.GetScanRecord(context.Background(), "sha256:stalleds3"); err != nil { + t.Errorf("no scan record written after a stalled upload: %v", err) + } + + // And the connection is still healthy: a slow backend is the hold's + // problem, not evidence against the scanner. + if got := subscriberCount(sb); got != 1 { + t.Errorf("subscriber count = %d, want 1: a slow S3 upload disconnected a healthy scanner", got) + } +} + +// Moving the work off the reader must not lose any of it, and must not let two +// messages about one job land out of order. Both results below name the same +// job; the second is the newer verdict and must be applied after the first. +// +// Only the first upload is parked, which is what makes the assertion decisive +// rather than lucky: a handler that starts the second result while the first is +// still uploading finishes it immediately, because nothing is holding it back. +func TestScanBroadcaster_ResultsForOneJobStayOrdered(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + blocker := newBlockingS3(t, sb) + blocker.blockFirst = true + sb.pingInterval = 50 * time.Millisecond + sb.pongWait = 10 * time.Second + sb.writeWait = 5 * time.Second + + server, client := wsConnPair(t) + go func() { + for { + if _, _, err := client.ReadMessage(); err != nil { + return + } + } + }() + + const instance = "ordered-scanner" + seq := seedJobWithDigest(t, sb, "sha256:ordered") + if _, err := sb.db.Exec( + `UPDATE scan_jobs SET status='assigned', assigned_to=?, assigned_at=? WHERE seq=?`, + instance, time.Now(), seq); err != nil { + t.Fatalf("assign: %v", err) + } + + sb.Subscribe(server, 0, instance, 1) + + // Two SBOMs of different, recognisable sizes, and no vulnerability report, + // so each result produces exactly one upload. + const firstSize, secondSize = 4096, 8192 + send := func(size int) { + t.Helper() + if err := client.WriteJSON(map[string]any{ + "type": "result", "seq": seq, "sbom": strings.Repeat("x", size), + }); err != nil { + t.Fatalf("write result: %v", err) + } + } + + send(firstSize) + // Park the first result inside its upload, then queue the second behind it. + select { + case <-blocker.entered: + case <-time.After(10 * time.Second): + t.Fatal("the first upload never started") + } + send(secondSize) + + // Nothing may finish while the first result is still in flight. + select { + case size := <-blocker.completed: + t.Fatalf("an upload of %d bytes finished while an earlier result for the "+ + "same job was still in flight: a stale verdict can overwrite a fresh one", size) + case <-time.After(time.Second): + } + + close(blocker.release) + + for i, want := range []int64{firstSize, secondSize} { + select { + case got := <-blocker.completed: + if got != want { + t.Fatalf("upload %d was %d bytes, want %d: the two results were "+ + "applied out of arrival order", i, got, want) + } + case <-time.After(15 * time.Second): + t.Fatalf("only %d of 2 results reached storage: a message was dropped", i) + } + } +} diff --git a/pkg/hold/pds/scan_broadcaster_predecessor_test.go b/pkg/hold/pds/scan_broadcaster_predecessor_test.go new file mode 100644 index 0000000..be6b0c3 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_predecessor_test.go @@ -0,0 +1,173 @@ +package pds + +import ( + "context" + "net/http" + "net/http/httptest" + "testing" +) + +// TestScanPredecessor_UnreachableHoldIsNotCachedAsNegative is F5. +// +// isOurManifest cached whatever checkPredecessor returned, and checkPredecessor +// returned a bare false for every failure mode: DID resolution failed, the HTTP +// request failed, a non-200 came back, the body would not read, the JSON would +// not parse. predecessorCache is never invalidated and never expires, so one +// five-second timeout against a predecessor hold means every manifest naming +// that hold is treated as not ours for the life of the process. A predecessor +// is by definition a retired hold, so "briefly unreachable" is its normal +// condition and the cache is filled on the first discovery pass after boot. +// +// pkg/hold/gc solved exactly this: a second return value saying whether the +// answer is definitive, definitive answers cached, inconclusive ones parked in +// a per-run map that is cleared each pass. This is the same rule for the scan +// broadcaster. +func TestScanPredecessor_UnreachableHoldIsNotCachedAsNegative(t *testing.T) { + sb := &ScanBroadcaster{ + holdDID: "did:web:us.example.com", + predecessorCache: make(map[string]bool), + } + + // .invalid is reserved by RFC 2606 and never resolves, so the check cannot + // reach a conclusion. + const unreachable = "did:web:unreachable.invalid" + + if sb.isOurManifest(context.Background(), unreachable) { + t.Fatal("an unreachable hold was adopted: ownership was asserted on no evidence") + } + + if _, cached := sb.predecessorCache[unreachable]; cached { + t.Error("an inconclusive check was cached as a negative in predecessorCache, " + + "which is never reset: one blip means this hold's manifests are never " + + "scanned again for the life of the process") + } + + if !sb.predecessorUnresolved[unreachable] { + t.Error("an inconclusive check was not recorded in predecessorUnresolved: " + + "the pass pays the five-second timeout again for every manifest naming this hold") + } + + // The second call inside one pass is answered from predecessorUnresolved + // rather than re-dialled, and must agree with the first. + if sb.isOurManifest(context.Background(), unreachable) { + t.Error("the second call disagreed with the first") + } +} + +// A pass boundary is what re-opens the question. predecessorUnresolved exists +// only to stop one unreachable hold costing a timeout per manifest within a +// single pass; a hold that was down during one pass must be asked again on the +// next one. +func TestScanPredecessor_UnresolvedIsClearedEachPass(t *testing.T) { + sb := &ScanBroadcaster{ + holdDID: "did:web:us.example.com", + predecessorCache: make(map[string]bool), + predecessorUnresolved: map[string]bool{"did:web:wasdown.invalid": true}, + stopCh: make(chan struct{}), + } + + // No relay endpoints, so the pass fetches no DIDs and returns immediately. + // Clearing has to happen before that, at the top of the pass. + sb.runDiscoveryPass() + + if sb.predecessorUnresolved["did:web:wasdown.invalid"] { + t.Error("predecessorUnresolved survived a discovery pass: a hold that was " + + "down once is written off instead of being re-checked") + } +} + +// The one negative worth caching: the hold answered, and said it has no +// successor. That is an answer, not a failure, and repeating the request would +// only cost time. +func TestScanPredecessor_DefinitiveAnswersAreCached(t *testing.T) { + sb := &ScanBroadcaster{ + holdDID: "did:web:us.example.com", + predecessorCache: make(map[string]bool), + } + + tests := []struct { + name string + body string + wantPredecesor bool + wantDefinitive bool + }{ + { + name: "no successor", + body: `{"uri":"at://x","value":{"successor":""}}`, + wantPredecesor: false, + wantDefinitive: true, + }, + { + name: "successor is us", + body: `{"uri":"at://x","value":{"successor":"did:web:us.example.com"}}`, + wantPredecesor: true, + wantDefinitive: true, + }, + { + name: "successor is a third hold", + body: `{"uri":"at://x","value":{"successor":"did:web:elsewhere.example.com"}}`, + wantPredecesor: false, + wantDefinitive: true, + }, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + srv := newCaptainServer(t, 200, tc.body) + got, definitive := sb.checkPredecessorAt(context.Background(), "did:web:other.example.com", srv) + if got != tc.wantPredecesor || definitive != tc.wantDefinitive { + t.Errorf("checkPredecessorAt = (%v, %v), want (%v, %v)", + got, definitive, tc.wantPredecesor, tc.wantDefinitive) + } + }) + } +} + +// A reachable hold that cannot produce its own captain record is malfunctioning, +// not answering. Reading a 500 as "not a predecessor" is the same permanent +// mistake as reading a timeout that way. +func TestScanPredecessor_NonDefinitiveReplies(t *testing.T) { + sb := &ScanBroadcaster{ + holdDID: "did:web:us.example.com", + predecessorCache: make(map[string]bool), + } + + tests := []struct { + name string + status int + body string + }{ + {"server error", 500, `{"error":"InternalServerError"}`}, + {"not found", 404, `{"error":"RecordNotFound"}`}, + {"unparseable envelope", 200, `not json at all`}, + {"unparseable record", 200, `{"uri":"at://x","value":"not an object"}`}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + srv := newCaptainServer(t, tc.status, tc.body) + isPredecessor, definitive := sb.checkPredecessorAt(context.Background(), "did:web:other.example.com", srv) + if definitive { + t.Errorf("reply treated as definitive; it would be cached forever "+ + "(isPredecessor=%v)", isPredecessor) + } + if isPredecessor { + t.Error("an inconclusive reply reported as a predecessor") + } + }) + } +} + +// newCaptainServer serves one canned reply at the getRecord endpoint and +// returns its base URL. +func newCaptainServer(t *testing.T, status int, body string) string { + t.Helper() + + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _, _ = w.Write([]byte(body)) + })) + t.Cleanup(srv.Close) + return srv.URL +} diff --git a/pkg/hold/pds/scan_broadcaster_test.go b/pkg/hold/pds/scan_broadcaster_test.go index 099e7b6..a9b9cfc 100644 --- a/pkg/hold/pds/scan_broadcaster_test.go +++ b/pkg/hold/pds/scan_broadcaster_test.go @@ -1,7 +1,10 @@ package pds import ( + "context" "database/sql" + "os" + "path/filepath" "testing" "time" @@ -44,7 +47,7 @@ func newRecordingScanBroadcaster(t *testing.T) *ScanBroadcaster { sb.inflight = make(map[string]struct{}) sb.ackTimeout = 5 * time.Minute - pds, _ := setupTestPDS(t) + pds := setupScanTestPDS(t) sb.pds = pds sb.holdDID = pds.did sb.s3 = &s3.S3Service{Client: s3.NewMockS3Client(""), Bucket: "test-bucket"} @@ -52,6 +55,40 @@ func newRecordingScanBroadcaster(t *testing.T) *ScanBroadcaster { return sb } +// setupScanTestPDS is setupTestPDS on a file-backed database rather than +// ":memory:". +// +// go-libsql's in-memory database is per *connection*, not per database, so the +// moment two goroutines write through one *sql.DB the second one gets a +// connection with no tables in it — "no such table: blocks" out of the +// carstore. That did not matter while every scan record was written on the +// caller's goroutine. It does now: terminal messages are stored on the +// subscriber's storage goroutine, so the record write and the test's readback +// are on different goroutines by construction. +func setupScanTestPDS(t *testing.T) *HoldPDS { + t.Helper() + + ctx := context.Background() + tmpDir := t.TempDir() + + keyPath := filepath.Join(tmpDir, "signing-key") + if err := os.WriteFile(keyPath, sharedTestKey, 0600); err != nil { + t.Fatalf("write signing key: %v", err) + } + + pds, err := NewHoldPDS(ctx, "did:web:hold.example.com", "https://hold.example.com", + "https://atcr.io", filepath.Join(tmpDir, "pds.sqlite3"), keyPath, false) + if err != nil { + t.Fatalf("create test PDS: %v", err) + } + if err := pds.repomgr.InitNewActor(ctx, pds.uid, "", pds.did, "", "", ""); err != nil { + t.Fatalf("initialize test repo: %v", err) + } + t.Cleanup(func() { pds.Close() }) + + return pds +} + // newTestScanSubscriber mirrors what Subscribe builds, registered with the // broadcaster so Unsubscribe finds it. func newTestScanSubscriber(t *testing.T, sb *ScanBroadcaster, bufSize int) *ScanSubscriber { diff --git a/scanner/internal/client/hold.go b/scanner/internal/client/hold.go index d03290e..2fc79e5 100644 --- a/scanner/internal/client/hold.go +++ b/scanner/internal/client/hold.go @@ -22,6 +22,30 @@ import ( "github.com/gorilla/websocket" ) +const ( + // holdPingInterval is how often the scanner pings an idle hold connection. + holdPingInterval = 30 * time.Second + + // holdPongWait is how long the connection may go without a pong (or any + // other frame) before the scanner treats the hold as gone. Three missed + // pings. + holdPongWait = 90 * time.Second + + // holdWriteWait bounds a single write to the hold. + holdWriteWait = 30 * time.Second + + // holdMaxMessageSize caps one frame from the hold. gorilla's default is + // unlimited, so without it a broken or hostile hold decides how much this + // process allocates. + // + // A job frame is small — a config descriptor and a layer list — so 16 MiB + // is several orders of magnitude of headroom over anything the real hold + // sends. Note the asymmetry with the hold's own limit: results travel the + // other way and carry whole SBOMs inline, so the hold's ceiling has to be + // far higher than this one. + holdMaxMessageSize = 16 << 20 +) + // httpClient is used for blob downloads and presigned URL requests // with a timeout to prevent stalled connections from leaking memory. var httpClient = &http.Client{Timeout: 5 * time.Minute} @@ -46,6 +70,33 @@ type HoldClient struct { // the dispatch budget for this connection; a hold that does not know the // parameter ignores it and assumes one. workers int + + // WebSocket liveness knobs. Zero means "use the package default"; tests + // shrink them so a half-open connection can be provoked in milliseconds. + pingInterval time.Duration + pongWait time.Duration + writeWait time.Duration +} + +func (c *HoldClient) pingEvery() time.Duration { + if c.pingInterval > 0 { + return c.pingInterval + } + return holdPingInterval +} + +func (c *HoldClient) pongDeadline() time.Duration { + if c.pongWait > 0 { + return c.pongWait + } + return holdPongWait +} + +func (c *HoldClient) writeDeadline() time.Duration { + if c.writeWait > 0 { + return c.writeWait + } + return holdWriteWait } // NewHoldClient creates a new hold client @@ -153,6 +204,27 @@ func (c *HoldClient) connectOnce(cursor int64) error { slog.Info("Connected to hold service") + // Liveness. This loop used to sit in ReadMessage with no deadline and send + // nothing, so a hold that went away without closing the TCP connection left + // this process permanently "connected": never reconnecting, so never + // re-draining the hold's pending rows, and writing every result it computed + // into a socket that went nowhere. + conn.SetReadLimit(holdMaxMessageSize) + refreshRead := func() { + if err := conn.SetReadDeadline(time.Now().Add(c.pongDeadline())); err != nil { + slog.Debug("Failed to set hold read deadline", "error", err) + } + } + refreshRead() + conn.SetPongHandler(func(string) error { + refreshRead() + return nil + }) + + stopPing := make(chan struct{}) + defer close(stopPing) + go c.pingLoop(conn, stopPing) + // Read messages from hold for { _, data, err := conn.ReadMessage() @@ -162,8 +234,37 @@ func (c *HoldClient) connectOnce(cursor int64) error { } return err } + // Any frame is proof of life, not just a pong. + refreshRead() c.handleFrame(data) + refreshRead() + } +} + +// pingLoop keeps the connection provably alive from this end. gorilla permits +// WriteControl concurrently with every other method, so this needs none of the +// write mutex and cannot be delayed behind a result being sent. +func (c *HoldClient) pingLoop(conn *websocket.Conn, stop <-chan struct{}) { + ticker := time.NewTicker(c.pingEvery()) + defer ticker.Stop() + + for { + select { + case <-stop: + return + case <-c.done: + return + case <-ticker.C: + if err := conn.WriteControl(websocket.PingMessage, nil, + time.Now().Add(c.writeDeadline())); err != nil { + // Closing is what unblocks the read loop, which then + // reconnects. Returning quietly would leave it parked. + slog.Warn("Failed to ping hold, dropping the connection", "error", err) + _ = conn.Close() + return + } + } } } @@ -303,6 +404,18 @@ func (c *HoldClient) SendSkipped(seq int64, reason string) { c.sendJSON(scanner.SkippedMessage{Type: "skipped", Seq: seq, Reason: reason}) } +// sendJSON writes one message to the hold under the write mutex. +// +// The deadline is what stops a wedged write from taking the process with it: +// a result is megabytes of SBOM, and a hold that has stopped reading fills the +// socket buffer and blocks the write. Without a deadline that write blocks +// until the kernel gives up on the connection, which is on the order of hours, +// with the mutex held the whole time — so every worker that finishes a scan +// queues up behind it and the scanner goes quiet without ever disconnecting. +// +// A failed write closes the connection rather than just logging. The read loop +// is then released, connectOnce returns, and Connect redials; leaving it open +// means the next write blocks the same way against the same dead socket. func (c *HoldClient) sendJSON(v any) { c.mu.Lock() defer c.mu.Unlock() @@ -312,8 +425,12 @@ func (c *HoldClient) sendJSON(v any) { return } + if err := c.conn.SetWriteDeadline(time.Now().Add(c.writeDeadline())); err != nil { + slog.Error("Failed to set WebSocket write deadline", "error", err) + } if err := c.conn.WriteJSON(v); err != nil { - slog.Error("Failed to send WebSocket message", "error", err) + slog.Error("Failed to send WebSocket message, dropping the connection", "error", err) + _ = c.conn.Close() } } diff --git a/scanner/internal/client/hold_test.go b/scanner/internal/client/hold_test.go new file mode 100644 index 0000000..e75e26d --- /dev/null +++ b/scanner/internal/client/hold_test.go @@ -0,0 +1,156 @@ +package client + +import ( + "net/http" + "net/http/httptest" + "strings" + "testing" + "time" + + "atcr.io/scanner/internal/queue" + "github.com/gorilla/websocket" +) + +// upgradeOnly accepts a WebSocket and hands the server side to onConn, which +// decides how lifelike the peer is. The returned URL is what a HoldClient dials. +func upgradeOnly(t *testing.T, onConn func(*websocket.Conn)) string { + t.Helper() + + upgrader := websocket.Upgrader{} + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + conn, err := upgrader.Upgrade(w, r, nil) + if err != nil { + return + } + onConn(conn) + })) + t.Cleanup(srv.Close) + return "ws" + strings.TrimPrefix(srv.URL, "http") +} + +func newTestHoldClient(t *testing.T, url string) *HoldClient { + t.Helper() + + c := NewHoldClient(url, "secret", queue.NewJobQueue(8)) + c.pingInterval = 20 * time.Millisecond + c.pongWait = 200 * time.Millisecond + c.writeWait = time.Second + t.Cleanup(func() { + defer func() { _ = recover() }() // Close panics if already closed + c.Close() + }) + return c +} + +// TestHoldClient_DetectsHalfOpenConnection is the scanner half of F7. +// +// connectOnce sat in ReadMessage with no deadline and sent no pings, so a hold +// that went away without closing the TCP connection left this scanner +// permanently "connected": it never reconnects, never re-drains the hold's +// pending rows, and every result it computes is written into a socket that goes +// nowhere. The peer here upgrades and then does nothing at all — it never reads, +// so it never pongs. +func TestHoldClient_DetectsHalfOpenConnection(t *testing.T) { + held := make(chan struct{}) + url := upgradeOnly(t, func(conn *websocket.Conn) { + <-held // hold the connection open, answering nothing + _ = conn.Close() + }) + t.Cleanup(func() { close(held) }) + + c := newTestHoldClient(t, url) + + errCh := make(chan error, 1) + go func() { errCh <- c.connectOnce(-1) }() + + select { + case err := <-errCh: + if err == nil { + t.Fatal("connectOnce returned nil for a peer that answered nothing") + } + case <-time.After(5 * time.Second): + t.Fatal("a hold that answers no pings never disconnected the scanner: " + + "the client waits in ReadMessage with no deadline and sends no pings") + } +} + +// The other side of the rule. An idle hold is the normal state — the dfd604b +// commit message records a hold and scanner that shared one WebSocket for a +// week — so silence alone must not drop the connection. gorilla answers a ping +// from inside ReadMessage, which is what the real hold's reader does. +func TestHoldClient_StaysConnectedWhileHoldAnswers(t *testing.T) { + url := upgradeOnly(t, func(conn *websocket.Conn) { + for { + if _, _, err := conn.ReadMessage(); err != nil { + _ = conn.Close() + return + } + } + }) + + c := newTestHoldClient(t, url) + + errCh := make(chan error, 1) + go func() { errCh <- c.connectOnce(-1) }() + + // Many pongWait periods with no traffic but pings. + select { + case err := <-errCh: + t.Fatalf("an idle but responsive hold was dropped: %v", err) + case <-time.After(time.Second): + } + + c.Close() + select { + case <-errCh: + case <-time.After(5 * time.Second): + t.Fatal("connectOnce did not return after Close") + } +} + +// A job frame is small — a config descriptor and a layer list — so the read +// limit that bounds a hostile or broken hold must still admit anything the real +// one sends with room to spare. +func TestHoldClient_AcceptsARealisticJobFrame(t *testing.T) { + delivered := make(chan struct{}) + url := upgradeOnly(t, func(conn *websocket.Conn) { + // A 128-layer image, which is well past the OCI-practical limit. + var layers []string + for i := 0; i < 128; i++ { + layers = append(layers, + `{"digest":"sha256:`+strings.Repeat("a", 64)+`","size":123456,"mediaType":"application/vnd.oci.image.layer.v1.tar+gzip"}`) + } + frame := `{"type":"job","seq":1,"manifestDigest":"sha256:` + strings.Repeat("b", 64) + `",` + + `"repository":"repo","tag":"latest","userDid":"did:plc:user","userHandle":"user.example.com",` + + `"holdDid":"did:web:hold.example.com","holdEndpoint":"https://hold.example.com","tier":"deckhand",` + + `"config":{"digest":"sha256:` + strings.Repeat("c", 64) + `","size":1024,"mediaType":"application/vnd.oci.image.config.v1+json"},` + + `"layers":[` + strings.Join(layers, ",") + `]}` + if err := conn.WriteMessage(websocket.TextMessage, []byte(frame)); err != nil { + return + } + for { + if _, _, err := conn.ReadMessage(); err != nil { + _ = conn.Close() + return + } + select { + case delivered <- struct{}{}: + default: + } + } + }) + + c := newTestHoldClient(t, url) + c.pongWait = 5 * time.Second + go func() { _ = c.connectOnce(-1) }() + + select { + case <-delivered: + case <-time.After(5 * time.Second): + t.Fatal("a realistic job frame was never acknowledged: the read limit rejected it") + } + + if c.queue.Len() != 1 { + t.Errorf("queue depth = %d, want 1: the job frame did not survive the read", c.queue.Len()) + } +}