scanner: bound a scan job, and lose the race to the hold on purpose

Nothing limited how long one job could take. The worker's context was the
process's, buildOCILayout took none, and blob downloads used a package-level
client whose five-minute timeout is per request with no context, so a 19-layer
image had a hundred-minute worst case on downloads alone and cancellation could
not touch it. At the default single worker, one wedged job stopped that scanner
entirely.

scanner.job_timeout, default 8m, against the hold's 10m scanning timeout. Both
clocks start at the same instant: the worker sends "started" on dequeue and
derives the job context on the next line, so the scanner loses by two minutes,
which is enough for its terminal message to cross the socket and be recorded.
If the hold wins instead it re-dispatches while this scanner is still working,
which is duplicate work recorded under a generic reason. A scanner cannot read
the hold's config, so the relation is a mirrored constant used only for a
boot-time warning, and the same warning fires if the deadline is disabled.

What is actually bounded, since a deadline the code cannot honour is worse than
none: presign, download, stereoscope's Provide, Syft's CreateSBOM, and Grype,
which does have FindMatchesContext even though FindMatches does not.
stereoscope's img.Read takes no context and is 81% of a scan, so it is checked
either side rather than interrupted. Abandoning it on a goroutine would trade a
bounded overrun for one writing gigabytes into a directory the caller has
already deleted. max_image_size remains the real bound on that stage.

A timeout reports error, not skipped. It describes this host at this moment, a
contended CPU or a slow bucket, not the image, and skips are never retried, so
one bad afternoon would retire an image permanently with nothing in the record
to say why. Retry cost is bounded on the other side by max_image_size and by
the stale-scan schedule. The classification asks the job context rather than
the error, because several stages replace the cause and the uninterruptible one
knows nothing about the deadline, and a job that finishes after an overrun
still reports its real result rather than throwing away completed work.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01U1Km3N3uUmeGaj7VbaM8PF

SCANNER_BUGS.md

@@ -46,6 +46,7 @@ directions, and four reached the hold-side panic.
 | **S** | Digest path traversal (CONFIRMED by execution: 50 bytes written outside the scan directory) and the total absence of content verification. New strict `scanner.ParseDigest`, validation at three boundaries, streaming hash-and-size verification, and `MaxImageSize` now enforced against real transferred bytes rather than the sizes a manifest claims. | `scanner/digest.go`, `scan/extractor.go`, `client/hold.go` |
 | **N** | Concurrency. The proactive capacity gate was depth-one hold-wide, defeating both extra workers and extra scanner processes. Dispatch depth is now the sum of worker counts scanners advertise on connect, the gate is proactive-scoped, and dispatch prefers the least-loaded scanner. Disconnects no longer hand off running work. Terminal handlers check `assigned_to`. Boot reconciliation added. | `scan_broadcaster.go`, `scanner/types.go`, `client/hold.go` |
 | **F4** | The hold's scanning deadline measured queueing, not scanning, because the scanner acks on receipt and `handleAck` never refreshed `assigned_at`. New `started` message sent by the worker that dequeues a job, with separate scanning (10m from `started_at`) and queueing (60m from `assigned_at`) budgets. | `scanner/types.go`, `client/hold.go`, `scan_broadcaster.go` |
+| **D** | Nothing bounded one scan job. New `scanner.job_timeout` (default 8m, under the hold's 10m scanning budget) starts when a worker dequeues, at the same instant the `started` message goes out. A context now reaches every blob request (`http.NewRequestWithContext`), Syft's cataloging and Grype's matcher (`FindMatchesContext`). A timeout is reported as a retryable `error`, and the scan directory is removed. Stereoscope's `img.Read()` takes no context and remains uninterruptible; `vuln.max_image_size` is its bound. | `scan/worker.go`, `scan/extractor.go`, `scan/syft.go`, `scan/grype.go`, `client/hold.go`, `internal/config/config.go` |
 
 Both modules pass `go test -race`. Three columns were added to `scan_jobs`
 (`origin`, `started_at`, `disconnected_at`) with per-column migration for
@@ -63,7 +64,7 @@ existing holds, plus an `origin, status` index.
 | MEDIUM | **The cgroup memory ceiling needs rechecking.** See section 3. | section 3 |
 | LOW | Reports are ~14.5 KB per rendered row (1.8 MB for 125 matches), parsed on every render. The report descriptor hardcodes Grype `v0.107.1` against a `v0.118.0` dependency. The report's embedded `summary` is read by nothing. | section 6 |
 | LOW | `pkg/config/viper.go:34` discards `ReadInConfig()`'s error, so a malformed YAML file is silently ignored whole, in scanner, hold and appview alike. | section 9 |
-| LOW | Shutdown: blob downloads carry no request context, so `pool.Wait()` can block for minutes against a 30 second grace period; `queue.Close()` drains rather than cancels. `HoldClient.Close()` panics if called twice. | section 9 |
+| LOW | Shutdown: `queue.Close()` drains rather than cancels, and `HoldClient.Close()` panics if called twice. (The blob-download half is fixed: requests now carry the job context, so `pool.Wait()` returns on cancellation.) | section 9 |
 | LOW | A webhook for a summary-less scan reports all-zero counts, because `db.Scan` has no status column. Only affects deployments running `vuln.enabled: false`. | section 7 |
 | LOW | `assigned_at` is stored as local-offset RFC3339 while `created_at` normalises to `...Z`. Correct today, breaks across a DST transition on a non-UTC host. | section 12 |
 | — | Test infrastructure: `e2e.Start` writes the package-level `scan.JobCooldown`, so two harnesses alive in one test race. Bound each harness in a `t.Run` subtest, or make the cooldown per-pool. | section 6 |
@@ -143,12 +144,34 @@ scanner's queue fills, but the classification itself is unchanged.
 Fixed at the protocol level: the hold now measures scanning from when a worker
 reports starting, not from dispatch.
 
-Still absent inside the scanner: no per-job timeout exists. The worker's context
-is the process's, `buildOCILayout` takes no context, and neither `img.Read()`
-nor `FindMatches` accept one. Blob downloads use a package-level `httpClient`
-whose 5 minute timeout is *per request*, so a 19-layer image has a 100 minute
-worst case on download alone and worker-context cancellation cannot abort a
-download at all.
+Fixed inside the scanner too. `scanner.job_timeout` (default 8 minutes) is
+started by the worker that dequeues the job, at the same instant it sends
+`started`, so the scanner's budget and the hold's 10-minute one measure the same
+interval and the scanner's is the shorter. It reaches the blob requests
+(`http.NewRequestWithContext` on both the presign call and the body), Syft's
+cataloging, and Grype's matcher, which turned out to have a `FindMatchesContext`
+alongside the `FindMatches` that has no cancellation. A timeout is reported as
+an `error`, not a `skipped`: the cause is the host, not the image, and
+`max_image_size` already refuses the pathological cases before a byte moves.
+Threading the context also fixed shutdown, which used to wait out the HTTP
+client's five-minute per-request timeout.
+
+**One stage remains uninterruptible, and the bound is not honoured across it.**
+`img.Read()` is stereoscope's layer extraction, it takes no context, and it is
+81% of a scan. The deadline is checked immediately before and after it, so a job
+that has already spent its budget does not go on to extract, and an overrun is
+noticed as soon as extraction returns — but in between, a pathologically slow
+decompression runs to completion and can outlast both the scanner's deadline and
+the hold's. Running it on an abandoned goroutine was rejected: it trades a
+bounded overrun for a leaked goroutine writing gigabytes into a directory the
+caller has already cleaned up. `vuln.max_image_size` is the real bound on that
+stage, which is the argument for keeping it tight on a small host.
+
+Misconfiguration is one-sided and warned about at boot: if `job_timeout` is set
+at or above the hold's scanning timeout, the hold reclaims and re-dispatches a
+row this scanner is still working on, which is the duplicate-scan behaviour the
+deadline exists to prevent. The scanner cannot read the hold's configuration, so
+the check is against the value the hold shipped with and is advisory.
 
 There is still no ping/pong, read deadline or read limit on either side of the
 scanner WebSocket.
@@ -409,6 +432,8 @@ Agent reports in sections 7 onward cite some test names that no longer exist:
 | `TestScanUnsubscribe_UnassignsJobsOnce` | `TestScanUnsubscribe_MarksItsOwnJobsOnce` |
 | `TestScanUnsubscribe_ReoffersAJobTheScannerIsStillRunning` | `TestScanUnsubscribe_HoldsAJobForAReconnectingScanner` |
 | `TestScanHasActiveJobs_*` | `TestScanActiveProactiveJobs_*` |
+| `TestStalledDownloadHasNoShortDeadline` | `TestStalledDownloadIsWaitedOutWithinTheJobBudget` |
+| `TestShutdownDoesNotInterruptInFlightDownload` | `TestShutdownAbortsAnInFlightDownload` (inverted; now in `deadline_test.go`) |
 
 The blob edge cases that pinned the absence of digest validation moved from
 `blob_edge_test.go` to `blob_integrity_test.go`, where they assert the fix.

deploy/upcloud/configs/scanner.yaml.tmpl

@@ -25,6 +25,16 @@ scanner:
     # available on the host.
     workers: 1
     queue_size: 100
+    # Must stay below the hold's 10m scanning timeout, which it measures from
+    # the "started" message this scanner sends when a worker picks the job up.
+    # Whichever fires first decides: this one stops the scan and tells the hold
+    # why, the hold's marks the row failed and re-dispatches work that is still
+    # running. Two minutes of margin for the terminal message to land.
+    #
+    # Note what the deadline does not cover: stereoscope's layer extraction
+    # takes no context and cannot be interrupted, and it is ~81% of a scan. The
+    # bound on that stage is max_image_size below, not this.
+    job_timeout: 8m
 vuln:
     enabled: true
     db_path: "{{.BasePath}}/scanner/vulndb"

scanner/internal/client/hold.go

@@ -3,6 +3,7 @@
 package client
 
 import (
+	"context"
 	"crypto/rand"
 	"crypto/sha256"
 	"encoding/hex"
@@ -46,8 +47,14 @@ const (
 	holdMaxMessageSize = 16 << 20
 )
 
-// httpClient is used for blob downloads and presigned URL requests
-// with a timeout to prevent stalled connections from leaking memory.
+// httpClient is used for blob downloads and presigned URL requests.
+//
+// Its Timeout is a per-request backstop and nothing more: a job that fetches a
+// config plus nineteen layers gets twenty of these in series, so on its own it
+// bounds one request at five minutes and a job at a hundred. The bound that
+// matters is the caller's context, which every request below carries; this
+// only stops a single wedged request from outliving a job whose deadline the
+// operator has disabled.
 var httpClient = &http.Client{Timeout: 5 * time.Minute}
 
 // HoldClient manages the WebSocket connection to a hold service
@@ -450,13 +457,13 @@ func (c *HoldClient) Close() {
 // The digest is a scanner.Digest rather than a string so that only a validated
 // "sha256:<hex>" can ever be asked for. Callers parse once, at the boundary,
 // and the same value then names the blob on the wire and the file on disk.
-func GetBlobPresignedURL(holdEndpoint, holdDID string, digest scanner.Digest, secret string) (string, error) {
+func GetBlobPresignedURL(ctx context.Context, holdEndpoint, holdDID string, digest scanner.Digest, secret string) (string, error) {
 	reqURL := fmt.Sprintf("%s/xrpc/com.atproto.sync.getBlob?did=%s&cid=%s&method=GET",
 		holdEndpoint,
 		url.QueryEscape(holdDID),
 		url.QueryEscape(digest.String()))
 
-	req, err := http.NewRequest("GET", reqURL, nil)
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, reqURL, nil)
 	if err != nil {
 		return "", fmt.Errorf("failed to create request: %w", err)
 	}
@@ -526,8 +533,18 @@ type BlobExpectation struct {
 // and the byte count are both accumulated by the same io.Copy that writes the
 // blob, so a scan can never catalog bytes that were not checked, and a blob
 // over the ceiling stops costing bandwidth one byte past it.
-func DownloadBlob(presignedURL, destPath string, want BlobExpectation) (int64, error) {
-	resp, err := httpClient.Get(presignedURL)
+//
+// The context bounds the whole transfer, not just the round trip: the request
+// body is read under it, so a hold that answers promptly and then dribbles
+// bytes forever is cancelled by the job deadline rather than by the client's
+// five-minute per-request ceiling.
+func DownloadBlob(ctx context.Context, presignedURL, destPath string, want BlobExpectation) (int64, error) {
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, presignedURL, nil)
+	if err != nil {
+		return 0, fmt.Errorf("failed to create download request: %w", err)
+	}
+
+	resp, err := httpClient.Do(req)
 	if err != nil {
 		return 0, fmt.Errorf("failed to download blob: %w", err)
 	}

scanner/internal/config/config.go

@@ -3,6 +3,7 @@ package config
 
 import (
 	"fmt"
+	"time"
 
 	"github.com/spf13/viper"
 
@@ -42,6 +43,16 @@ type ScannerConfig struct {
 
 	// Maximum priority queue depth.
 	QueueSize int `yaml:"queue_size" comment:"Maximum priority queue depth."`
+
+	// Wall-clock budget for one scan job, measured from the moment a worker
+	// dequeues it — the same instant the "started" message goes to the hold,
+	// so this budget and the hold's scanning timeout measure the same
+	// interval.
+	//
+	// It must stay below the hold's, or the hold reclaims and re-dispatches a
+	// job this scanner is still working on. See the comment on
+	// scan.holdScanningTimeout.
+	JobTimeout time.Duration `yaml:"job_timeout" comment:"Wall-clock budget for one scan job, from the moment a worker picks it up. Must stay BELOW the hold's scanning timeout (10m), or the hold reclaims the job while this scanner is still running it and the image is scanned twice. 0 disables the deadline. Default: 8m."`
 }
 
 // VulnConfig defines vulnerability scanning settings.
@@ -74,6 +85,7 @@ func setScannerDefaults(v *viper.Viper) {
 	// Scanner defaults
 	v.SetDefault("scanner.workers", 1)
 	v.SetDefault("scanner.queue_size", 100)
+	v.SetDefault("scanner.job_timeout", "8m")
 
 	// Vuln defaults
 	v.SetDefault("vuln.enabled", true)

scanner/internal/e2e/bench_test.go

@@ -335,11 +335,11 @@ func assembleLayout(job *scanner.ScanJob, tmpDir string) (string, func(), error)
 		if err != nil {
 			return err
 		}
-		url, err := client.GetBlobPresignedURL(job.HoldEndpoint, job.HoldDID, parsed, "")
+		url, err := client.GetBlobPresignedURL(context.Background(), job.HoldEndpoint, job.HoldDID, parsed, "")
 		if err != nil {
 			return err
 		}
-		_, err = client.DownloadBlob(url, filepath.Join(blobsDir, parsed.Hex), client.BlobExpectation{
+		_, err = client.DownloadBlob(context.Background(), url, filepath.Join(blobsDir, parsed.Hex), client.BlobExpectation{
 			Digest:   parsed,
 			MaxBytes: -1,
 		})

scanner/internal/e2e/blob_edge_test.go

@@ -399,17 +399,15 @@ func TestTruncatedBodyIsDetected(t *testing.T) {
 	assertNoLeakedScanDirs(t, h)
 }
 
-// TestStalledDownloadHasNoShortDeadline holds a response body open and shows
-// the scanner simply waits. client.httpClient's only bound is a 5 minute
-// per-request Timeout, and DownloadBlob builds no request context, so
-// cancelling the worker context (shutdown, SIGTERM) does not abort a download
-// in flight either.
+// TestStalledDownloadIsWaitedOutWithinTheJobBudget holds a response body open
+// and shows that a stall shorter than the job's deadline is simply waited out.
+// There is no per-blob impatience: a slow hold is not a broken one, and the
+// only thing that ends a download early is the job budget running out.
 //
-// The stall here is deliberately short. The point is not to sit out the real
-// timeout but to show there is no shorter one: a 2 second stall costs the
-// worker 2 seconds, and a hold that accepts the connection and then says
-// nothing costs it five minutes per blob, times the number of layers.
-func TestStalledDownloadHasNoShortDeadline(t *testing.T) {
+// The stall here is deliberately short, and the harness runs the shipped
+// job_timeout, so nothing here fires. What happens when the budget does run out
+// is TestJobDeadlineFailsAStalledBlobBody in deadline_test.go.
+func TestStalledDownloadIsWaitedOutWithinTheJobBudget(t *testing.T) {
 	const stall = 2 * time.Second
 
 	cfgBytes := unparseableConfig()

scanner/internal/e2e/deadline_test.go

@@ -0,0 +1,401 @@
+package e2e
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"atcr.io/scanner/internal/client"
+	"atcr.io/scanner/internal/config"
+	"atcr.io/scanner/internal/mockhold"
+	"atcr.io/scanner/internal/queue"
+	"atcr.io/scanner/internal/scan"
+)
+
+// This file covers the per-job deadline: the bound on how long one scan may
+// occupy a worker, and what the scanner says when it gives up.
+//
+// The ordering that makes any of it worth having is with the hold. The hold
+// stamps started_at when a worker sends 'started' and fails the row
+// scanningTimeout after that (10 minutes, pkg/hold/pds/scan_broadcaster.go).
+// If the scanner's own deadline fires first it sends a terminal message and
+// the hold records the real reason; if the hold's fires first it reclaims a
+// row the scanner is still working on and the work is done twice. Every
+// default here exists to keep the scanner strictly first.
+
+// --- a stall that can be observed being aborted ----------------------------
+
+// abortableGate is newGate with one difference that is the whole point: its
+// handler watches the request context as well as the release channel, so a
+// test can tell an aborted request from one that merely finished.
+//
+// The plain gate cannot: it blocks on <-release unconditionally, so a
+// cancelled request looks identical to a served one from the server side.
+type abortableGate struct {
+	srv      *httptest.Server
+	entered  chan struct{}
+	release  chan struct{}
+	aborted  atomic.Int64
+	released atomic.Int64
+}
+
+func newAbortableGate(t *testing.T) *abortableGate {
+	t.Helper()
+	g := &abortableGate{entered: make(chan struct{}, 8), release: make(chan struct{})}
+	g.srv = httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case g.entered <- struct{}{}:
+		default:
+		}
+		select {
+		case <-r.Context().Done():
+			g.aborted.Add(1)
+			return
+		case <-g.release:
+			g.released.Add(1)
+			http.Error(w, "gate released", http.StatusNotFound)
+		}
+	}))
+	t.Cleanup(func() {
+		g.openOnce()
+		g.srv.Close()
+	})
+	return g
+}
+
+func (g *abortableGate) waitEntered(t *testing.T, timeout time.Duration) {
+	t.Helper()
+	select {
+	case <-g.entered:
+	case <-time.After(timeout):
+		t.Fatal("worker never reached the stalled blob fetch")
+	}
+}
+
+func (g *abortableGate) openOnce() {
+	defer func() { _ = recover() }()
+	close(g.release)
+}
+
+// waitAborted waits for the server to observe the client hanging up.
+func (g *abortableGate) waitAborted(t *testing.T, timeout time.Duration) bool {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if g.aborted.Load() > 0 {
+			return true
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return false
+}
+
+// --- shutdown ---------------------------------------------------------------
+
+// TestShutdownAbortsAnInFlightDownload is the cancellability precondition for
+// everything else in this file. A per-job deadline is a context deadline, and
+// a context is worth nothing on a code path that never looks at one.
+//
+// The blob fetches go through client.GetBlobPresignedURL and
+// client.DownloadBlob. Both used a package-level http.Client and built their
+// requests with http.NewRequest, so no context reached the transport at all
+// and the only bound was the client's own five-minute per-request Timeout.
+// Cancelling the pool context (what cmd/scanner does on SIGTERM, before
+// WorkerPool.Wait) therefore did nothing: the worker stayed inside the fetch,
+// Wait blocked for as long as the fetch did, and a 30 second termination grace
+// period turned into a SIGKILL with the scan directory left behind.
+func TestShutdownAbortsAnInFlightDownload(t *testing.T) {
+	hold := mockhold.New(mockhold.NewMemory(), mockhold.WithSecret(testSecret))
+	t.Cleanup(hold.Close)
+
+	cfg := config.DefaultConfig()
+	cfg.Hold.URL = hold.URL()
+	cfg.Hold.Secret = testSecret
+	cfg.Scanner.Workers = 1
+	cfg.Vuln.Enabled = false
+	cfg.Vuln.TmpDir = t.TempDir()
+
+	origTmpDir, hadTmpDir := os.LookupEnv("TMPDIR")
+	t.Cleanup(func() {
+		if hadTmpDir {
+			os.Setenv("TMPDIR", origTmpDir)
+			return
+		}
+		os.Unsetenv("TMPDIR")
+	})
+
+	restoreCooldown := scan.JobCooldown
+	scan.JobCooldown = testJobCooldown
+
+	q := queue.NewJobQueue(cfg.Scanner.QueueSize)
+	c := client.NewHoldClient(cfg.Hold.URL, cfg.Hold.Secret, q)
+	pool := scan.NewWorkerPool(cfg, q, c)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	pool.Start(ctx)
+	go c.Connect()
+
+	var closed bool
+	t.Cleanup(func() {
+		cancel()
+		if !closed {
+			c.Close()
+		}
+		q.Close()
+		pool.Wait()
+		scan.JobCooldown = restoreCooldown
+	})
+
+	if err := hold.WaitForScanner(10 * time.Second); err != nil {
+		t.Fatalf("scanner never connected: %v", err)
+	}
+
+	g := newAbortableGate(t)
+	job := synthJob("probe/shutdown")
+	job.HoldEndpoint = g.srv.URL
+	if _, err := hold.SendJob(job); err != nil {
+		t.Fatalf("send job: %v", err)
+	}
+	g.waitEntered(t, 15*time.Second)
+
+	// The shutdown sequence cmd/scanner runs on SIGTERM.
+	cancel()
+	c.Close()
+	closed = true
+	q.Close()
+
+	waited := make(chan struct{})
+	go func() { pool.Wait(); close(waited) }()
+
+	select {
+	case <-waited:
+	case <-time.After(10 * time.Second):
+		t.Fatal("worker did not exit on context cancellation: the download " +
+			"path still ignores ctx, so shutdown waits out the HTTP client's " +
+			"own timeout instead")
+	}
+
+	if !g.waitAborted(t, 5*time.Second) {
+		t.Error("the server never saw the request hang up; the fetch was not " +
+			"actually cancelled, the worker just stopped waiting for it")
+	}
+
+	entries, err := os.ReadDir(cfg.Vuln.TmpDir)
+	if err != nil {
+		t.Fatalf("read tmp dir: %v", err)
+	}
+	for _, e := range entries {
+		t.Errorf("scan directory left behind after a cancelled download: %s", e.Name())
+	}
+}
+
+// --- the per-job deadline ---------------------------------------------------
+
+// withJobTimeout shortens the per-job deadline. Production defaults to
+// something that must accommodate a genuinely large image; these scenarios
+// need it to fire in milliseconds.
+func withJobTimeout(d time.Duration) Option {
+	return func(c *config.Config) { c.Scanner.JobTimeout = d }
+}
+
+// TestJobDeadlineFailsAStalledPresignRequest is the headline behaviour: a job
+// that wedges in the download stage is given up on, on its own, without the
+// hold having to notice.
+//
+// The stall is the presigned-URL request, which is the first network call a
+// job makes and the one buildOCILayout used to make with no context at all.
+// The assertion that matters is the timing: the terminal message has to arrive
+// while the server is still holding the request open, or the scanner did not
+// give up, it just outlasted the fault.
+func TestJobDeadlineFailsAStalledPresignRequest(t *testing.T) {
+	const jobTimeout = 400 * time.Millisecond
+
+	hold := newHold(mockhold.NewMemory())
+	h := startScanner(t, hold, withJobTimeout(jobTimeout))
+
+	g := newAbortableGate(t)
+	job := synthJob("probe/deadline")
+	job.HoldEndpoint = g.srv.URL
+
+	started := time.Now()
+	seq, err := h.Hold.SendJob(job)
+	if err != nil {
+		t.Fatalf("send job: %v", err)
+	}
+	g.waitEntered(t, 15*time.Second)
+
+	msg := h.AwaitTerminal(t, seq, 10*time.Second)
+	elapsed := time.Since(started)
+
+	// A timeout is not a property of the image: the same bytes may well scan
+	// on a faster host, or on this one when it is not contending with a hold
+	// doing garbage collection. 'skipped' is terminal at the hold and would
+	// retire the image permanently on the evidence of one slow afternoon.
+	if msg.Type != "error" {
+		t.Fatalf("timeout reported as %q (%s%s), want \"error\" so the hold retries it",
+			msg.Type, msg.Error, msg.Reason)
+	}
+	if !strings.Contains(msg.Error, "timed out") {
+		t.Errorf("error does not say the job timed out, so the hold records a "+
+			"transport failure instead: %q", msg.Error)
+	}
+	if elapsed > 5*time.Second {
+		t.Errorf("gave up after %s, far past the %s deadline: the deadline is "+
+			"not what ended the job", elapsed, jobTimeout)
+	}
+	if g.released.Load() > 0 {
+		t.Error("the stalled request was served rather than abandoned; the " +
+			"scanner waited the fault out")
+	}
+	if !g.waitAborted(t, 5*time.Second) {
+		t.Error("the server never saw the request hang up: the deadline ended " +
+			"the job but left the fetch running")
+	}
+
+	// The scan directory is created before the first download and removed by
+	// buildOCILayout's own cleanup on failure. A deadline that fires mid-
+	// download must not be the one path that skips it.
+	assertNoLeakedScanDirs(t, h)
+}
+
+// TestJobDeadlineFailsAStalledBlobBody is the same deadline one call later.
+//
+// GetBlobPresignedURL and DownloadBlob are separate requests through the same
+// client, and only the second one streams bytes. A hold that answers the
+// presign promptly and then dribbles the body is the more realistic stall, and
+// it exercises the io.Copy rather than the round trip.
+func TestJobDeadlineFailsAStalledBlobBody(t *testing.T) {
+	const jobTimeout = 400 * time.Millisecond
+
+	stalled := make(chan struct{})
+	var aborted atomic.Int64
+	hold := newHold(mockhold.NewMemory(), mockhold.WithBlobResponseHook(
+		func(w http.ResponseWriter, r *http.Request, digest string) bool {
+			w.Header().Set("Content-Length", "4096")
+			w.WriteHeader(http.StatusOK)
+			_, _ = w.Write([]byte("partial"))
+			if f, ok := w.(http.Flusher); ok {
+				f.Flush()
+			}
+			select {
+			case <-r.Context().Done():
+				aborted.Add(1)
+			case <-stalled:
+			}
+			return true
+		}))
+	t.Cleanup(func() { close(stalled) })
+
+	h := startScanner(t, hold, withJobTimeout(jobTimeout))
+
+	cfgBytes := unparseableConfig()
+	started := time.Now()
+	seq, err := h.Hold.SendJob(jobFor(desc(digestOf(cfgBytes), int64(len(cfgBytes)), configType)))
+	if err != nil {
+		t.Fatalf("send job: %v", err)
+	}
+
+	msg := h.AwaitTerminal(t, seq, 10*time.Second)
+	elapsed := time.Since(started)
+
+	if msg.Type != "error" {
+		t.Fatalf("want error, got %s: %s%s", msg.Type, msg.Error, msg.Reason)
+	}
+	if !strings.Contains(msg.Error, "timed out") {
+		t.Errorf("error does not say the job timed out: %q", msg.Error)
+	}
+	if elapsed > 5*time.Second {
+		t.Errorf("gave up after %s against a %s deadline", elapsed, jobTimeout)
+	}
+	deadline := time.Now().Add(5 * time.Second)
+	for aborted.Load() == 0 && time.Now().Before(deadline) {
+		time.Sleep(10 * time.Millisecond)
+	}
+	if aborted.Load() == 0 {
+		t.Error("the body stream was never cancelled; the download outlived the job")
+	}
+	assertNoLeakedScanDirs(t, h)
+}
+
+// TestJobDeadlineFreesTheWorkerForTheNextJob is the reason the deadline exists.
+//
+// At the shipped default of one worker, a job with no bound stops that scanner
+// entirely: everything behind it sits acked in the priority queue while the
+// hold's dispatch budget stays spent. The deadline is only worth having if the
+// worker comes back.
+func TestJobDeadlineFreesTheWorkerForTheNextJob(t *testing.T) {
+	hold := newHold(mockhold.NewMemory())
+	h := startScanner(t, hold, withJobTimeout(400*time.Millisecond))
+
+	g := newAbortableGate(t)
+	wedged := synthJob("probe/wedged")
+	wedged.HoldEndpoint = g.srv.URL
+	wedgedSeq, err := h.Hold.SendJob(wedged)
+	if err != nil {
+		t.Fatalf("send wedged job: %v", err)
+	}
+	g.waitEntered(t, 15*time.Second)
+
+	// A helm chart is refused on its config media type alone, so it reaches a
+	// terminal message without touching the network: whether it is answered
+	// depends only on whether a worker ever dequeues it.
+	next := helmJob(t)
+	nextSeq, err := h.Hold.SendJob(next)
+	if err != nil {
+		t.Fatalf("send follow-up job: %v", err)
+	}
+
+	if msg := h.AwaitTerminal(t, wedgedSeq, 10*time.Second); msg.Type != "error" {
+		t.Fatalf("wedged job ended as %s, want error", msg.Type)
+	}
+	if msg := h.AwaitTerminal(t, nextSeq, 10*time.Second); msg.Type != "skipped" {
+		t.Fatalf("follow-up job ended as %s (%s%s), want skipped: the worker "+
+			"never got back to it", msg.Type, msg.Error, msg.Reason)
+	}
+}
+
+// TestJobDeadlineDefaultLeavesTheHoldRoomToHearAboutIt pins the ordering the
+// whole design rests on.
+//
+// The hold fails a row scanningTimeout after the worker's 'started' message
+// (10 minutes, pkg/hold/pds/scan_broadcaster.go) and hands it back to the
+// pending pool. The scanner sends 'started' at the same instant it starts its
+// own clock, so the two budgets measure the same interval and the scanner's
+// has to be the shorter one — with enough margin left for the terminal message
+// to cross the WebSocket and be recorded.
+//
+// Misconfigured the other way (job_timeout at or above the hold's budget) the
+// hold wins the race: it marks the row failed and re-dispatches while this
+// scanner is still working, which is the duplicate-scan behaviour the deadline
+// was added to prevent. Nothing here can enforce that across two processes;
+// this only pins the shipped default.
+func TestJobDeadlineDefaultLeavesTheHoldRoomToHearAboutIt(t *testing.T) {
+	// pkg/hold/pds/scan_broadcaster.go: scanningTimeout.
+	const holdScanningTimeout = 10 * time.Minute
+
+	got := config.DefaultConfig().Scanner.JobTimeout
+	if got <= 0 {
+		t.Fatalf("scanner.job_timeout defaults to %s: shipped with no deadline at all", got)
+	}
+	if got >= holdScanningTimeout {
+		t.Fatalf("scanner.job_timeout defaults to %s, at or past the hold's %s "+
+			"scanning budget: the hold reclaims the row first and the scan is "+
+			"run twice", got, holdScanningTimeout)
+	}
+	if margin := holdScanningTimeout - got; margin < time.Minute {
+		t.Errorf("only %s between the scanner's deadline and the hold's; too "+
+			"little room for the terminal message to land", margin)
+	}
+	// A node:22 scan measured 16s end to end on a fast workstation, 81% of it
+	// stereoscope extraction, on hardware far better than production. The
+	// default has to leave room for that on a much smaller host.
+	if got < 2*time.Minute {
+		t.Errorf("scanner.job_timeout defaults to %s, too short for a large "+
+			"image on a small host", got)
+	}
+}

scanner/internal/e2e/protocol_test.go

@@ -1,11 +1,9 @@
 package e2e
 
 import (
-	"context"
 	"encoding/json"
 	"net/http"
 	"net/http/httptest"
-	"os"
 	"strings"
 	"testing"
 	"time"
@@ -15,7 +13,6 @@ import (
 	"atcr.io/scanner/internal/config"
 	"atcr.io/scanner/internal/mockhold"
 	"atcr.io/scanner/internal/queue"
-	"atcr.io/scanner/internal/scan"
 )
 
 // --- helpers ---------------------------------------------------------------
@@ -613,94 +610,14 @@ func TestQueueCloseDrainsRatherThanCancels(t *testing.T) {
 	t.Logf("Close() left %d jobs to be dequeued and scanned after shutdown began", len(drained))
 }
 
-// TestShutdownDoesNotInterruptInFlightDownload proves processJob ignores
-// context cancellation everywhere that matters. Syft and Grype take ctx, but
-// the blob fetches go through client.GetBlobPresignedURL / DownloadBlob, which
-// use a package-level http.Client with no request context at all. Cancelling
-// the pool's context while a fetch is in flight changes nothing: the worker
-// stays inside the download until the client's own five-minute timeout, and
-// WorkerPool.Wait (which cmd/scanner calls on SIGTERM, after cancel) blocks
-// for just as long. Under a typical 30-second termination grace period that is
-// a SIGKILL, with the scan directory left behind because cleanup never runs.
-func TestShutdownDoesNotInterruptInFlightDownload(t *testing.T) {
-	hold := mockhold.New(mockhold.NewMemory(), mockhold.WithSecret(testSecret))
-	t.Cleanup(hold.Close)
-
-	cfg := config.DefaultConfig()
-	cfg.Hold.URL = hold.URL()
-	cfg.Hold.Secret = testSecret
-	cfg.Scanner.Workers = 1
-	cfg.Vuln.Enabled = false
-	cfg.Vuln.TmpDir = t.TempDir()
-
-	origTmpDir, hadTmpDir := os.LookupEnv("TMPDIR")
-	t.Cleanup(func() {
-		if hadTmpDir {
-			os.Setenv("TMPDIR", origTmpDir)
-			return
-		}
-		os.Unsetenv("TMPDIR")
-	})
-
-	restoreCooldown := scan.JobCooldown
-	scan.JobCooldown = 10 * time.Millisecond
-	t.Cleanup(func() { scan.JobCooldown = restoreCooldown })
-
-	q := queue.NewJobQueue(cfg.Scanner.QueueSize)
-	c := client.NewHoldClient(cfg.Hold.URL, cfg.Hold.Secret, q)
-	pool := scan.NewWorkerPool(cfg, q, c)
-
-	ctx, cancel := context.WithCancel(context.Background())
-	pool.Start(ctx)
-	go c.Connect()
-
-	// HoldClient.Close is not idempotent (see TestHoldClientCloseIsNotIdempotent),
-	// so the cleanup must not repeat the shutdown the test itself performs.
-	var closed bool
-	t.Cleanup(func() {
-		cancel()
-		if !closed {
-			c.Close()
-		}
-		q.Close()
-	})
-
-	if err := hold.WaitForScanner(10 * time.Second); err != nil {
-		t.Fatalf("scanner never connected: %v", err)
-	}
-
-	g := newGate(t)
-	job := synthJob("probe/shutdown")
-	job.HoldEndpoint = g.srv.URL
-	if _, err := hold.SendJob(job); err != nil {
-		t.Fatalf("send job: %v", err)
-	}
-	g.waitEntered(t, 15*time.Second)
-
-	// This is the shutdown sequence cmd/scanner runs on SIGTERM.
-	cancel()
-	c.Close()
-	closed = true
-	q.Close()
-
-	waited := make(chan struct{})
-	go func() { pool.Wait(); close(waited) }()
-
-	select {
-	case <-waited:
-		t.Fatal("worker exited on context cancellation; the download path now honours ctx")
-	case <-time.After(500 * time.Millisecond):
-		// Expected: the worker is still inside an uncancellable HTTP fetch.
-	}
-
-	g.open()
-	select {
-	case <-waited:
-	case <-time.After(30 * time.Second):
-		t.Fatal("worker never exited even after the download completed")
-	}
-	t.Log("context cancellation does not reach blob downloads; shutdown waits on the HTTP timeout")
-}
+// Shutdown and in-flight downloads used to be pinned here, by
+// TestShutdownDoesNotInterruptInFlightDownload: the blob fetches built their
+// requests with http.NewRequest and no context, so cancelling the pool context
+// changed nothing and WorkerPool.Wait blocked until the HTTP client's own
+// five-minute timeout expired. That is fixed, and the opposite is now asserted
+// by TestShutdownAbortsAnInFlightDownload in deadline_test.go, which also
+// checks the server sees the request hang up rather than the worker merely
+// walking away from it.
 
 // TestHoldClientCloseIsNotIdempotent pins a sharp edge rather than a live bug:
 // HoldClient.Close closes c.done unconditionally, so a second call panics the

scanner/internal/e2e/stuck_test.go

@@ -230,19 +230,20 @@ func stuckTerminalFor(h *Harness, seq int64) (mockhold.Message, bool) {
 // stops: one job that never finishes, and a worker pool with nothing to
 // interrupt it.
 //
-// processJob takes a context and honours it nowhere. buildOCILayout does not
-// take one at all, and the worker's context is the process's, cancelled only at
-// shutdown — there is no per-job deadline anywhere in the scanner. So a job
-// that hangs holds the only worker (scanner.workers defaults to 1) for as long
-// as it hangs, and every job behind it sits in the priority queue having
-// already been acked.
+// A job that hangs holds the only worker (scanner.workers defaults to 1), and
+// every job behind it sits in the priority queue having already been acked.
+// The blocking itself is not a bug — one worker runs one scan — and this test
+// pins its shape.
 //
-// Here the hang is a blob the hold never answers. That one is bounded, at
-// 5 minutes per HTTP request by client.httpClient — see
-// TestBlobDownloadsAreSerialSoTheirTimeoutsAdd for how far that bound stretches.
-// The unbounded version is the same shape with no timeout at all: Syft's
-// stereoscope extraction and Grype's matching run outside any deadline, and a
-// layer that decompresses forever holds the worker forever.
+// What has changed is that it now ends. scanner.job_timeout bounds the job
+// (8 minutes by default, under the hold's 10), so the head of the line is
+// eventually failed and the queue moves; deadline_test.go asserts that with a
+// deadline short enough to watch. Here the default is in force and the window
+// is seconds, so the blocking is what is visible.
+//
+// One stage is still outside the deadline: stereoscope's img.Read() takes no
+// context, so a layer that decompresses pathologically slowly overruns and the
+// job ends when extraction does. vuln.max_image_size is the bound on that one.
 func TestStuckJobBlocksEveryJobBehindIt(t *testing.T) {
 	src := newStuckSource().gated()
 	h := Start(t, src)
@@ -408,15 +409,17 @@ func TestAcksLandLongBeforeTheWorkDoes(t *testing.T) {
 	}
 }
 
-// TestBlobDownloadsAreSerialSoTheirTimeoutsAdd shows why the one timeout the
-// download path does have is not a bound on a job.
+// TestBlobDownloadsAreSerialSoTheirTimeoutsAdd shows why client.httpClient's
+// own timeout was never a bound on a job, and so why the job needs its own.
 //
 // buildOCILayout fetches the config and then every layer in sequence, each
 // through client.httpClient, whose Timeout is 5 minutes and applies per
-// request. A manifest with 19 layers therefore has a worst case of 20 × 5
-// minutes before the job fails — twice the hold's ten-minute processing
-// deadline, so the hold gives up on a job the scanner is still legitimately
-// working on, and does so without re-dispatching it.
+// request. A manifest with 19 layers therefore had a worst case of 20 x 5
+// minutes before the job failed, ten times the hold's ten-minute scanning
+// deadline. The serial shape asserted below has not changed; what bounds it
+// now is scanner.job_timeout, which every one of these requests carries as its
+// context, so the sum stops at the job budget rather than at 20 times the
+// per-request one.
 func TestBlobDownloadsAreSerialSoTheirTimeoutsAdd(t *testing.T) {
 	job := stuckJob("serial", 3)
 
@@ -453,9 +456,9 @@ func TestBlobDownloadsAreSerialSoTheirTimeoutsAdd(t *testing.T) {
 		t.Errorf("job took %s, less than the sum of its four downloads", elapsed)
 	}
 
-	t.Logf("4 serial downloads took %s; each is bounded only by the 5-minute "+
-		"client timeout, so this job's worst case is 20 minutes against a "+
-		"10-minute hold deadline", elapsed.Round(time.Millisecond))
+	t.Logf("4 serial downloads took %s; the per-request client timeout still "+
+		"adds up across them, and the job's context is what caps the total",
+		elapsed.Round(time.Millisecond))
 }
 
 // TestReDispatchAfterDisconnectScansTheSameImageTwice proves the duplicate-scan

scanner/internal/scan/deadline_test.go

@@ -0,0 +1,150 @@
+package scan
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	scanner "atcr.io/scanner"
+	"atcr.io/scanner/internal/config"
+)
+
+// These are the unit-level assertions behind the per-job deadline: how the
+// worker derives a job's context, how the shipped default relates to the
+// hold's own budget, and that the scan directory is removed when the deadline
+// fires mid-download. The end-to-end shape lives in internal/e2e/deadline_test.go.
+
+func poolWithJobTimeout(d time.Duration) *WorkerPool {
+	cfg := config.DefaultConfig()
+	cfg.Scanner.JobTimeout = d
+	return &WorkerPool{cfg: cfg}
+}
+
+// TestJobContextCarriesTheConfiguredDeadline pins the knob to the context the
+// job actually runs under. Everything downstream (the HTTP requests, Syft's
+// cataloging, Grype's matcher) is bounded by this context and by nothing else.
+func TestJobContextCarriesTheConfiguredDeadline(t *testing.T) {
+	wp := poolWithJobTimeout(5 * time.Second)
+
+	ctx, cancel := wp.jobContext(context.Background())
+	defer cancel()
+
+	deadline, ok := ctx.Deadline()
+	if !ok {
+		t.Fatal("job context carries no deadline, so nothing downstream can be bounded")
+	}
+	if left := time.Until(deadline); left <= 0 || left > 5*time.Second {
+		t.Errorf("job deadline is %s away, want (0, 5s]", left)
+	}
+}
+
+// TestZeroJobTimeoutDisablesTheDeadline keeps the escape hatch honest. An
+// operator scanning very large images on a hold with a longer budget can turn
+// the deadline off, and must still get a context that shutdown can cancel.
+func TestZeroJobTimeoutDisablesTheDeadline(t *testing.T) {
+	wp := poolWithJobTimeout(0)
+
+	parent, cancelParent := context.WithCancel(context.Background())
+	ctx, cancel := wp.jobContext(parent)
+	defer cancel()
+
+	if _, ok := ctx.Deadline(); ok {
+		t.Error("job_timeout=0 still produced a deadline")
+	}
+
+	cancelParent()
+	select {
+	case <-ctx.Done():
+	case <-time.After(time.Second):
+		t.Error("job context does not follow the pool context, so shutdown cannot interrupt a job")
+	}
+}
+
+// TestDefaultJobTimeoutFiresBeforeTheHoldGivesUp is the ordering constraint,
+// asserted against the constant the scanner keeps for the hold's budget.
+//
+// If the hold's deadline lands first it marks the row failed and re-dispatches
+// it while this worker is still scanning: two scanners on one image, and the
+// recorded reason is the hold's generic timeout rather than anything the
+// scanner observed. The scanner has to lose the race on purpose.
+func TestDefaultJobTimeoutFiresBeforeTheHoldGivesUp(t *testing.T) {
+	got := config.DefaultConfig().Scanner.JobTimeout
+	if got <= 0 {
+		t.Fatal("the shipped default disables the deadline")
+	}
+	if got >= holdScanningTimeout {
+		t.Fatalf("default job_timeout %s is not below the hold's %s scanning budget", got, holdScanningTimeout)
+	}
+	if margin := holdScanningTimeout - got; margin < time.Minute {
+		t.Errorf("only %s of margin for the terminal message to reach the hold", margin)
+	}
+}
+
+// TestBuildOCILayoutCleansUpWhenTheDeadlineFires is the temp-directory
+// assertion on the timeout path specifically.
+//
+// buildOCILayout creates scan-* before its first download and removes it on
+// every failure it returns. A deadline that fires mid-transfer is a new way to
+// leave that function, and the 3.8x extraction amplification measured for a
+// node:22 image is what a leak costs: a scanner that times out repeatedly
+// fills its volume.
+func TestBuildOCILayoutCleansUpWhenTheDeadlineFires(t *testing.T) {
+	stalled := make(chan struct{})
+	defer close(stalled)
+
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		select {
+		case <-r.Context().Done():
+		case <-stalled:
+		}
+	}))
+	defer srv.Close()
+
+	tmpDir := t.TempDir()
+	job := &scanner.ScanJob{
+		ManifestDigest: "sha256:" + strings.Repeat("a", 64),
+		Repository:     "deadline/cleanup",
+		HoldDID:        "did:web:hold.example",
+		HoldEndpoint:   srv.URL,
+		Config: scanner.BlobDescriptor{
+			Digest:    "sha256:" + strings.Repeat("c", 64),
+			Size:      100,
+			MediaType: "application/vnd.oci.image.config.v1+json",
+		},
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+	defer cancel()
+
+	started := time.Now()
+	_, _, err := buildOCILayout(ctx, job, tmpDir, "", 0)
+	if err == nil {
+		t.Fatal("buildOCILayout succeeded against a server that never answers")
+	}
+	if elapsed := time.Since(started); elapsed > 5*time.Second {
+		t.Errorf("buildOCILayout took %s to notice a 200ms deadline", elapsed)
+	}
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Errorf("error does not carry the deadline: %v", err)
+	}
+
+	// A timeout must not be reported as a skip. The hold never re-offers a
+	// skipped record, so a slow afternoon would retire the image for good.
+	var skip *SkipError
+	if errors.As(err, &skip) {
+		t.Errorf("a timed-out download was classified as a permanent skip: %v", err)
+	}
+
+	entries, err := os.ReadDir(tmpDir)
+	if err != nil {
+		t.Fatalf("read tmp dir: %v", err)
+	}
+	for _, e := range entries {
+		t.Errorf("scan directory left behind after the deadline fired: %s", e.Name())
+	}
+}

scanner/internal/scan/extractor.go

@@ -1,6 +1,7 @@
 package scan
 
 import (
+	"context"
 	"crypto/sha256"
 	"encoding/json"
 	"errors"
@@ -55,7 +56,12 @@ type ociIndex struct {
 // maxBytes is the ceiling on the total transferred for this job; zero or less
 // means unlimited. It is spent down blob by blob, so it bounds the real bytes
 // on disk rather than the numbers the manifest record claims.
-func buildOCILayout(job *scanner.ScanJob, tmpDir, secret string, maxBytes int64) (string, func(), error) {
+//
+// ctx bounds the whole download stage. Every request below carries it, so a
+// cancelled or expired context aborts the transfer in flight rather than
+// waiting out the HTTP client's own per-request timeout, and every exit from
+// here removes the scan directory — the deadline path included.
+func buildOCILayout(ctx context.Context, job *scanner.ScanJob, tmpDir, secret string, maxBytes int64) (string, func(), error) {
 	scanDir, err := os.MkdirTemp(tmpDir, "scan-*")
 	if err != nil {
 		return "", nil, fmt.Errorf("failed to create temp directory: %w", err)
@@ -94,6 +100,15 @@ func buildOCILayout(job *scanner.ScanJob, tmpDir, secret string, maxBytes int64)
 		Layers:        make([]ociDescriptor, 0, len(job.Layers)),
 	}
 	for _, ref := range referencedBlobs(job) {
+		// Checked per blob as well as inside each request: a job that has
+		// already spent its budget must not open the next connection, and the
+		// error here names the deadline rather than whatever the transport
+		// happens to report when it is torn down mid-handshake.
+		if err := ctx.Err(); err != nil {
+			cleanup()
+			return "", nil, fmt.Errorf("downloading %s: %w", ref.what(), err)
+		}
+
 		digest, err := scanner.ParseDigest(ref.Descriptor.Digest)
 		if err != nil {
 			cleanup()
@@ -102,7 +117,7 @@ func buildOCILayout(job *scanner.ScanJob, tmpDir, secret string, maxBytes int64)
 		slog.Info("Downloading blob", "blob", ref.what(), "digest", digest,
 			"declaredSize", ref.Descriptor.Size, "mediaType", ref.Descriptor.MediaType)
 
-		n, err := downloadBlob(job, digest, ref.Descriptor.Size, remaining, blobsDir, secret)
+		n, err := downloadBlob(ctx, job, digest, ref.Descriptor.Size, remaining, blobsDir, secret)
 		if err != nil {
 			cleanup()
 			return "", nil, blobFailure(ref.what(), err)
@@ -232,6 +247,8 @@ func referencedBlobs(job *scanner.ScanJob) []blobRef {
 // bytes already stored belongs on the skip side. A transport fault (a 5xx, a
 // dropped connection, an expired presigned URL) stays an error, because those
 // really can succeed next time.
+// A cancelled or expired context is emphatically on the error side: the
+// deadline is a property of this host on this afternoon, not of the image.
 func blobFailure(what string, err error) error {
 	if errors.Is(err, client.ErrBlobCorrupt) || errors.Is(err, client.ErrBlobTooLarge) {
 		return &SkipError{Reason: fmt.Sprintf("%s: %v", what, err)}
@@ -242,7 +259,7 @@ func blobFailure(what string, err error) error {
 // downloadBlob fetches one validated blob into the blobs directory and returns
 // how many bytes arrived. maxBytes is the remaining job budget, negative for
 // unbounded.
-func downloadBlob(job *scanner.ScanJob, digest scanner.Digest, declaredSize, maxBytes int64, blobsDir, secret string) (int64, error) {
+func downloadBlob(ctx context.Context, job *scanner.ScanJob, digest scanner.Digest, declaredSize, maxBytes int64, blobsDir, secret string) (int64, error) {
 	destPath := filepath.Join(blobsDir, digest.Hex)
 
 	// The invariant, asserted rather than assumed. ParseDigest has already
@@ -253,11 +270,11 @@ func downloadBlob(job *scanner.ScanJob, digest scanner.Digest, declaredSize, max
 		return 0, fmt.Errorf("refusing to write blob %s outside %s", digest, blobsDir)
 	}
 
-	presignedURL, err := client.GetBlobPresignedURL(job.HoldEndpoint, job.HoldDID, digest, secret)
+	presignedURL, err := client.GetBlobPresignedURL(ctx, job.HoldEndpoint, job.HoldDID, digest, secret)
 	if err != nil {
 		return 0, fmt.Errorf("failed to get presigned URL for %s: %w", digest, err)
 	}
-	return client.DownloadBlob(presignedURL, destPath, client.BlobExpectation{
+	return client.DownloadBlob(ctx, presignedURL, destPath, client.BlobExpectation{
 		Digest:       digest,
 		DeclaredSize: declaredSize,
 		MaxBytes:     maxBytes,

scanner/internal/scan/grype.go

@@ -132,7 +132,13 @@ func scanVulnerabilities(ctx context.Context, s *sbom.SBOM, vulnDBPath string) (
 		NormalizeByCVE:        true,
 	}
 
-	allMatches, _, err := vulnerabilityMatcher.FindMatches(grypePackages, pkgContext)
+	// FindMatchesContext, not FindMatches: the two differ only in whether the
+	// matcher loop checks for cancellation, and Grype's own doc comment on
+	// FindMatches says so. Matching is 0.35 to 0.5s on a small image, so this
+	// is rarely the stage a deadline lands in, but it is the stage after the
+	// uninterruptible one and the cheapest possible place to notice that the
+	// budget is gone.
+	allMatches, _, err := vulnerabilityMatcher.FindMatchesContext(ctx, grypePackages, pkgContext)
 	if err != nil {
 		return nil, "", scanner.VulnerabilitySummary{}, fmt.Errorf("failed to find vulnerabilities: %w", err)
 	}

scanner/internal/scan/syft.go

@@ -38,11 +38,37 @@ func generateSBOM(ctx context.Context, ociLayoutDir, sourceRef string) (*sbom.SB
 		return nil, nil, "", fmt.Errorf("failed to load OCI image: %w", err)
 	}
 
+	// The one stage of the pipeline no deadline reaches.
+	//
+	// img.Read() is stereoscope's layer extraction: it takes no context, and
+	// there is no supported way to interrupt it. It is also the most expensive
+	// thing the scanner does, measured at 81% of a node:22 scan. Checking
+	// before it starts is what can be done honestly — a job that has already
+	// spent its budget on downloads does not go on to spend another thirteen
+	// minutes extracting — and checking after is how an overrun is noticed at
+	// all. In between, the job runs to completion no matter what the deadline
+	// says.
+	//
+	// The alternative would be to run it on its own goroutine and return when
+	// the context fires, which trades a bounded overrun for a leaked goroutine
+	// still writing gigabytes into a directory the caller has cleaned up. That
+	// is worse. vuln.max_image_size is the real bound on this stage: it caps
+	// the input, and with it the extraction.
+	if err := ctx.Err(); err != nil {
+		_ = img.Cleanup()
+		return nil, nil, "", fmt.Errorf("before reading OCI image: %w", err)
+	}
+
 	if err := img.Read(); err != nil {
-		img.Cleanup()
+		_ = img.Cleanup()
 		return nil, nil, "", fmt.Errorf("failed to read OCI image: %w", err)
 	}
 
+	if err := ctx.Err(); err != nil {
+		_ = img.Cleanup()
+		return nil, nil, "", fmt.Errorf("after reading OCI image: %w", err)
+	}
+
 	// Wrap in Syft source — src.Close() calls img.Cleanup() internally,
 	// so we don't defer img.Cleanup() separately.
 	//

scanner/internal/scan/worker.go

@@ -71,12 +71,31 @@ func (wp *WorkerPool) Start(ctx context.Context) {
 		}()
 	}
 
+	// Say so at boot rather than leaving it to be inferred from duplicate
+	// scans months later. The scanner cannot read the hold's configuration, so
+	// this compares against the value the hold shipped with; a hold running a
+	// different budget makes this advisory, which is why it is a warning and
+	// not a refusal to start.
+	switch d := wp.jobTimeout(); {
+	case d <= 0:
+		slog.Warn("No per-job scan deadline configured; one wedged job will "+
+			"hold its worker until the process restarts",
+			"config", "scanner.job_timeout")
+	case d >= holdScanningTimeout:
+		slog.Warn("scanner.job_timeout is at or above the hold's scanning "+
+			"timeout, so the hold will give up on a job before this scanner "+
+			"does and re-dispatch work that is still running",
+			"job_timeout", d, "hold_scanning_timeout", holdScanningTimeout)
+	}
+
 	for i := 0; i < wp.cfg.Scanner.Workers; i++ {
 		wp.wg.Add(1)
 		go wp.worker(ctx, i)
 	}
 
-	slog.Info("Scanner worker pool started", "workers", wp.cfg.Scanner.Workers)
+	slog.Info("Scanner worker pool started",
+		"workers", wp.cfg.Scanner.Workers,
+		"job_timeout", wp.jobTimeout())
 }
 
 // Wait blocks until all workers finish
@@ -108,16 +127,56 @@ func (wp *WorkerPool) worker(ctx context.Context, id int) {
 		// tell queueing from scanning without this.
 		wp.client.SendStarted(job.Seq)
 
-		result, err := wp.processJob(ctx, job)
+		// The job's own clock starts here, at the same instant the hold's
+		// does. See holdScanningTimeout for why the two must not be equal.
+		jobCtx, cancelJob := wp.jobContext(ctx)
+		result, err := wp.processJob(jobCtx, job)
+
+		// Ask the context, not the error. Only the context knows whether the
+		// budget ran out; several stages wrap or replace the cause on the way
+		// back, and one of them (stereoscope's extraction) does not take a
+		// context at all, so the error it returns after an overrun says
+		// nothing about the deadline.
+		//
+		// Only when the job actually failed. A scan that finished inside an
+		// uninterruptible stage after the deadline passed still holds a real
+		// verdict, and the hold has not reclaimed the row: reporting a timeout
+		// instead would throw away work that is already done.
+		timedOut := err != nil && errors.Is(jobCtx.Err(), context.DeadlineExceeded)
+		cancelJob()
+
 		if err != nil {
 			var skipErr *SkipError
-			if errors.As(err, &skipErr) {
+			switch {
+			case timedOut:
+				// "error" and not "skipped", deliberately. A timeout is a
+				// statement about this host at this moment — a contended CPU,
+				// a slow S3, a co-tenant hold running garbage collection — not
+				// about the image, and the same bytes may well scan on the
+				// next pass. The hold re-offers errors and never re-offers
+				// skips, so a skip here would retire an image permanently on
+				// one bad afternoon.
+				//
+				// The cost of being wrong in this direction is a rescan at
+				// full price on the stale-scan schedule, and vuln.max_image_size
+				// already refuses the pathological images before a byte moves.
+				// The cost of being wrong in the other direction is an image
+				// that is never scanned again and no way to tell from the
+				// record why.
+				slog.Error("Scan job timed out",
+					"worker_id", id,
+					"repository", job.Repository,
+					"timeout", wp.jobTimeout(),
+					"error", err)
+				wp.client.SendError(job.Seq, fmt.Sprintf(
+					"scan timed out after %s: %v", wp.jobTimeout(), err))
+			case errors.As(err, &skipErr):
 				slog.Info("Scan job skipped",
 					"worker_id", id,
 					"repository", job.Repository,
 					"reason", skipErr.Reason)
 				wp.client.SendSkipped(job.Seq, skipErr.Reason)
-			} else {
+			default:
 				slog.Error("Scan job failed",
 					"worker_id", id,
 					"repository", job.Repository,
@@ -159,6 +218,39 @@ func (wp *WorkerPool) worker(ctx context.Context, id int) {
 	}
 }
 
+// holdScanningTimeout mirrors scanningTimeout in
+// pkg/hold/pds/scan_broadcaster.go. It is duplicated rather than imported
+// because the scanner is a separate module and a hold may be running a
+// different version than this scanner; it is a reference point for a startup
+// warning, never a bound this process enforces.
+//
+// The ordering it exists to protect: the hold stamps started_at when it
+// receives the "started" message a worker sends on dequeue, and fails the row
+// this long after. The scanner starts its own clock at the same instant, so
+// the two budgets measure the same interval and whichever is shorter decides
+// what happens. Shorter here means the scanner stops, sends a terminal message,
+// and the hold records the real reason. Shorter there means the hold marks the
+// row failed and hands it to the next scanner with a worker free, while this
+// one is still scanning: the same image scanned twice, and the verdict that
+// arrives late lands on a row somebody else now owns.
+const holdScanningTimeout = 10 * time.Minute
+
+// jobTimeout is the configured per-job budget, zero or less meaning none.
+func (wp *WorkerPool) jobTimeout() time.Duration {
+	return wp.cfg.Scanner.JobTimeout
+}
+
+// jobContext derives the context one job runs under.
+//
+// It always derives from the pool context, deadline or not, so shutdown
+// interrupts a job whether or not an operator has kept the deadline.
+func (wp *WorkerPool) jobContext(ctx context.Context) (context.Context, context.CancelFunc) {
+	if d := wp.jobTimeout(); d > 0 {
+		return context.WithTimeout(ctx, d)
+	}
+	return context.WithCancel(ctx)
+}
+
 // JobCooldown is the pause a worker takes after each job so Go's GC can
 // reclaim what Syft and Grype allocated before the next scan starts.
 //
@@ -258,7 +350,7 @@ func (wp *WorkerPool) processJob(ctx context.Context, job *scanner.ScanJob) (*sc
 
 	// Step 1: Build OCI image layout from hold via presigned URLs
 	slog.Info("Building OCI layout", "repository", job.Repository)
-	ociLayoutDir, cleanup, err := buildOCILayout(job, wp.cfg.Vuln.TmpDir, wp.cfg.Hold.Secret, wp.cfg.Vuln.MaxImageSize)
+	ociLayoutDir, cleanup, err := buildOCILayout(ctx, job, wp.cfg.Vuln.TmpDir, wp.cfg.Hold.Secret, wp.cfg.Vuln.MaxImageSize)
 	if err != nil {
 		return nil, fmt.Errorf("failed to build OCI layout: %w", err)
 	}