From 61deadc1245f11d4b1d6cfb7222778ab398d2440 Mon Sep 17 00:00:00 2001 From: Evan Jarrett Date: Sat, 5 Sep 2026 15:53:57 -0500 Subject: [PATCH] hold: reuse stored scan blobs when a rescan finds nothing new The stale-scan loop rescans an image on a schedule and handleResult uploaded both artifacts every time, so each pass wrote a fresh SBOM blob and orphaned the one before it. Nothing reclaims those, because GC walks only the OCI blob prefix and scan artifacts live under /repos. Keep the stored blobs when the record already on file was written by this same scanner version and carries a vulnerability-report reference equal to the one this report would be stored under, then keep each blob individually and only if it is still in S3. The report reference alone is not quite enough evidence on its own: the report lists matched packages, not every package, so an SBOM could gain a package with no known CVEs and leave the report byte-identical. Content is pinned by the manifest digest, so that can only happen across a scanner or Syft change, which is what the version clause closes. It is inert until that constant moves, and one comparison is cheaper than remembering to add it at the moment it first matters. Every failure path declines to reuse, which costs an upload and never costs correctness, including the reuse check's own timeout, which is carved out of the upload budget rather than given its own so a slow read cannot eat the deadline it stands in for. A record whose earlier upload failed heals on the next rescan, keeping the report and rewriting only the missing SBOM. scannedAt still advances. runStalePass selects on it, so freezing it would make every deduplicated record permanently stale and rescanned forever, which costs far more than the bytes saved. A scanner running with vulnerability scanning off sends no report, so there is no stable digest to compare and its SBOM still churns. Closing that means either parsing SPDX in the hold to normalise a timestamp and a UUID, or a lexicon change; a test pins the gap rather than leaving it to be rediscovered. Note for anyone extending GC to /repos later: a live SBOM object used to be rewritten on every rescan and so was always young. It now keeps its original mtime for the life of the content, so an age-based rule there would delete referenced blobs. Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01U1Km3N3uUmeGaj7VbaM8PF --- pkg/hold/pds/profile.go | 38 +- pkg/hold/pds/scan_broadcaster.go | 151 ++++++- pkg/hold/pds/scan_broadcaster_dedupe_test.go | 389 +++++++++++++++++++ 3 files changed, 559 insertions(+), 19 deletions(-) create mode 100644 pkg/hold/pds/scan_broadcaster_dedupe_test.go diff --git a/pkg/hold/pds/profile.go b/pkg/hold/pds/profile.go index db69d66..5b8d806 100644 --- a/pkg/hold/pds/profile.go +++ b/pkg/hold/pds/profile.go @@ -67,15 +67,17 @@ func downloadImage(ctx context.Context, url string) ([]byte, string, error) { return data, contentType, nil } -// uploadBlobToStorage uploads a blob to the hold's S3 storage and returns a blob reference. -// This stores the blob at the ATProto path for the hold's DID. -func uploadBlobToStorage(ctx context.Context, s3svc *s3.S3Service, did string, data []byte, mimeType string) (*lexutil.LexBlob, error) { +// blobRefForBytes computes the blob reference these bytes would be stored +// under, without touching S3. Blob storage is content-addressed, so this is +// also the answer to "is this payload already uploaded" — see +// reusableScanBlobs, which compares a fresh scan result against the reference +// on the record already on file. It is split out of uploadBlobToStorage rather +// than reimplemented so the two can never disagree about the CID. +func blobRefForBytes(data []byte, mimeType string) (*lexutil.LexBlob, error) { if len(data) == 0 { return nil, fmt.Errorf("empty blob data") } - size := int64(len(data)) - // Compute SHA-256 hash hash := sha256.Sum256(data) @@ -89,21 +91,29 @@ func uploadBlobToStorage(ctx context.Context, s3svc *s3.S3Service, did string, d // ATProto uses CIDv1 with raw codec for blobs blobCID := cid.NewCidV1(0x55, mh) + // Create blob reference in the format expected by bsky.ActorProfile + return &lexutil.LexBlob{ + Ref: lexutil.LexLink(blobCID), + MimeType: mimeType, + Size: int64(len(data)), + }, nil +} + +// uploadBlobToStorage uploads a blob to the hold's S3 storage and returns a blob reference. +// This stores the blob at the ATProto path for the hold's DID. +func uploadBlobToStorage(ctx context.Context, s3svc *s3.S3Service, did string, data []byte, mimeType string) (*lexutil.LexBlob, error) { + blob, err := blobRefForBytes(data, mimeType) + if err != nil { + return nil, err + } + // Store blob via S3 at ATProto path - path := atprotoBlobPath(did, blobCID.String()) + path := atprotoBlobPath(did, blob.Ref.String()) if err := s3svc.PutBytes(ctx, path, data, mimeType); err != nil { return nil, fmt.Errorf("failed to put blob: %w", err) } - // Create blob reference in the format expected by bsky.ActorProfile - lexLink := lexutil.LexLink(blobCID) - blob := &lexutil.LexBlob{ - Ref: lexLink, - MimeType: mimeType, - Size: size, - } - return blob, nil } diff --git a/pkg/hold/pds/scan_broadcaster.go b/pkg/hold/pds/scan_broadcaster.go index 6f2aff5..cbd997b 100644 --- a/pkg/hold/pds/scan_broadcaster.go +++ b/pkg/hold/pds/scan_broadcaster.go @@ -131,6 +131,14 @@ const ( // manifest as never scanned, so it is the last thing that may be skipped. scannerRecordTimeout = 60 * time.Second + // scannerReuseTimeout bounds the "has this content already been stored" + // lookup one result message costs: one record read out of the CAR store and + // up to two S3 HEADs. It is carved out of scannerResultTimeout rather than + // given its own budget, because it is a substitute for the uploads that + // follow it. Kept short: every path out of it falls back to uploading, so + // giving up early costs bytes, not correctness. + scannerReuseTimeout = 15 * 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 @@ -1268,6 +1276,11 @@ func (sb *ScanBroadcaster) claimJobForTerminal(sub *ScanSubscriber, seq int64, k // 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. +// +// The uploads are skipped entirely when the rescan produced materially the same +// result as the record already on file; see reusableScanBlobs. The record is +// still rewritten in that case, pointing at the blobs already stored, so +// scannedAt advances and the stale loop moves on. func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) { ctx, cancel := context.WithTimeout(context.Background(), sb.resultDeadline()) defer cancel() @@ -1304,10 +1317,18 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) return } + // What, if anything, this result may keep from the record already on file. + // Empty unless the rescan produced materially the same thing. + reuse := sb.reusableScanBlobs(ctx, manifestDigest, scannerVersion, msg) + // Upload SBOM as a blob to the hold's PDS blob storage (like manifest blobs) var sbomBlob *lexutil.LexBlob - if msg.SBOM != "" { - blob, err := uploadBlobToStorage(ctx, sb.s3, sb.holdDID, []byte(msg.SBOM), "application/spdx+json") + switch { + case msg.SBOM == "": + case reuse.sbom != nil: + sbomBlob = reuse.sbom + default: + blob, err := uploadBlobToStorage(ctx, sb.s3, sb.holdDID, []byte(msg.SBOM), sbomMimeType) if err != nil { slog.Error("Failed to upload SBOM blob to PDS storage", "seq", msg.Seq, @@ -1319,8 +1340,12 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) // Upload vulnerability report as a blob (full Grype JSON with CVE details) var vulnReportBlob *lexutil.LexBlob - if msg.VulnReport != "" { - blob, err := uploadBlobToStorage(ctx, sb.s3, sb.holdDID, []byte(msg.VulnReport), "application/vnd.atcr.vulnerabilities+json") + switch { + case msg.VulnReport == "": + case reuse.vuln != nil: + vulnReportBlob = reuse.vuln + default: + blob, err := uploadBlobToStorage(ctx, sb.s3, sb.holdDID, []byte(msg.VulnReport), vulnReportMimeType) if err != nil { slog.Error("Failed to upload VulnReport blob to PDS storage", "seq", msg.Seq, @@ -1330,6 +1355,14 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) } } + if reuse.sbom != nil || reuse.vuln != nil { + slog.Info("Rescan produced the stored result; keeping the existing blobs", + "seq", msg.Seq, + "manifest", manifestDigest, + "sbomReused", reuse.sbom != nil, + "vulnReportReused", reuse.vuln != nil) + } + // Store scan result as a record in the hold's embedded PDS. // // A result with no summary is a completed scan from a scanner running with @@ -1353,7 +1386,7 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) manifestDigest, repository, userDID, sbomBlob, vulnReportBlob, critical, high, medium, low, total, - "atcr-scanner-v1.0.0", + scannerVersion, ) recordCtx, recordCancel := context.WithTimeout(context.Background(), scannerRecordTimeout) @@ -1410,6 +1443,114 @@ func (sb *ScanBroadcaster) handleResult(sub *ScanSubscriber, msg ScannerMessage) } } +// Blob media types for the two artifacts a scan produces. Named because the +// vulnerability report's type is also needed to compute the reference a fresh +// report would be stored under, and a mismatch there would silently defeat +// deduplication. +const ( + sbomMimeType = "application/spdx+json" + vulnReportMimeType = "application/vnd.atcr.vulnerabilities+json" + + // scannerVersion stamps every record this build writes, and gates blob + // reuse: see reusableScanBlobs. + scannerVersion = "atcr-scanner-v1.0.0" +) + +// scanBlobReuse is what a result may keep from the scan record already on file +// instead of uploading its own copy. A nil field means "upload it". +type scanBlobReuse struct { + sbom *lexutil.LexBlob + vuln *lexutil.LexBlob +} + +// reusableScanBlobs decides whether a rescan produced materially the same +// result as the record already on file, and therefore may keep its blobs. +// +// The rule: keep the stored blobs when the record on file was written by this +// same scanner version and carries a vulnerability report whose reference +// equals the one this report would be stored under. Each blob is then kept +// individually, and only if it is still in S3. +// +// Why the report and not the SBOM. The stale-scan loop rescans on a timer, and +// an unchanged image rescanned a week later yields a byte-identical Grype +// report — it carries no timestamp and no database build date, only matches, +// source, distro and the Grype version (scanner/internal/scan/grype.go). The +// SBOM does not: Syft stamps creationInfo.created from the wall clock and +// appends a random UUID to documentNamespace, so its bytes, and therefore its +// content-addressed reference, move on every single run. The SBOM can never +// recognise itself; the report has to do it for both. +// +// Why the scanner version is part of the rule. The report lists matched +// packages, not every package, so it is possible in principle for the SBOM to +// change while the report does not — a Syft upgrade that starts cataloguing an +// ecosystem with no known vulnerabilities would do it. It is only possible +// across a scanner change, because the content itself is pinned by the manifest +// digest, so refusing to reuse across versions closes the hole completely. The +// version is a build constant today, which makes this clause inert until it +// starts moving; it is cheap, and the alternative is remembering to add it at +// the moment it is first needed. +// +// A vulnerability database update is not a case this suppresses: new matches +// mean a different report, no reference match, and both blobs are written +// fresh. That is correct — the report genuinely changed. +// +// Every failure path here returns "reuse nothing", which costs an upload and +// never costs correctness. That includes the timeout: the lookup runs on a +// short budget carved out of the caller's, so a slow CAR read cannot eat the +// upload deadline it is standing in for. +func (sb *ScanBroadcaster) reusableScanBlobs(ctx context.Context, manifestDigest, version string, msg ScannerMessage) scanBlobReuse { + var none scanBlobReuse + + if sb.pds == nil || sb.s3 == nil || msg.VulnReport == "" { + return none + } + + fresh, err := blobRefForBytes([]byte(msg.VulnReport), vulnReportMimeType) + if err != nil { + return none + } + + ctx, cancel := context.WithTimeout(ctx, scannerReuseTimeout) + defer cancel() + + _, prev, err := sb.pds.GetScanRecord(ctx, manifestDigest) + if err != nil || prev == nil { + // No record yet: the overwhelmingly common first-scan path. + return none + } + if prev.ScannerVersion != version { + return none + } + if prev.VulnReportBlob == nil || prev.VulnReportBlob.Ref != fresh.Ref { + // Either the previous upload failed and left nothing to compare + // against, or the result really is different. + return none + } + + // Materially unchanged. Each blob is kept only if the object it names is + // still there: a record can outlive its blob (a lifecycle rule, a restore, + // a GC pass), and reusing a dangling reference would make that permanent, + // since every later rescan reaches this same decision. + var reuse scanBlobReuse + if sb.blobStillStored(ctx, prev.VulnReportBlob) { + reuse.vuln = prev.VulnReportBlob + } + if msg.SBOM != "" && sb.blobStillStored(ctx, prev.SbomBlob) { + reuse.sbom = prev.SbomBlob + } + return reuse +} + +// blobStillStored reports whether the object a blob reference names is still in +// the hold's bucket. Any error is treated as "gone", which re-uploads. +func (sb *ScanBroadcaster) blobStillStored(ctx context.Context, blob *lexutil.LexBlob) bool { + if blob == nil { + return false + } + _, err := sb.s3.Stat(ctx, atprotoBlobPath(sb.holdDID, blob.Ref.String())) + return err == nil +} + // handleError marks a job as failed and creates a scan record so the stale // loop won't immediately retry. Failed records still get retried on the // rescan interval since failures may be transient (network, OOM, etc.). diff --git a/pkg/hold/pds/scan_broadcaster_dedupe_test.go b/pkg/hold/pds/scan_broadcaster_dedupe_test.go new file mode 100644 index 0000000..c772047 --- /dev/null +++ b/pkg/hold/pds/scan_broadcaster_dedupe_test.go @@ -0,0 +1,389 @@ +package pds + +import ( + "context" + "errors" + "fmt" + "sync" + "testing" + "time" + + "atcr.io/pkg/atproto" + "atcr.io/pkg/s3" + awss3 "github.com/aws/aws-sdk-go-v2/service/s3" + lexutil "github.com/bluesky-social/indigo/lex/util" +) + +// The stale-scan loop rescans every image on a fixed interval, and every pass +// used to upload a fresh SBOM blob and a fresh vulnerability-report blob even +// when the content had not changed by a byte. The previous blobs were left in +// S3 with nothing referencing them, and nothing reclaims them. +// +// e57e31c made the vulnerability report byte-stable for unchanged content (the +// Syft source reference is the manifest digest now, not a per-job temp path). +// The SBOM is not stable and cannot be made stable at the artifact layer: Syft +// stamps creationInfo.created from time.Now() and appends a random UUID to +// documentNamespace. So the SBOM's own digest can never be the dedupe key — +// the report's digest has to stand in for it. +// +// These tests pin the storage-layer rule: on a rescan whose vulnerability +// report is byte-identical to the one already on file, the hold keeps the +// blobs it already has instead of uploading new ones. + +// sbomAt fakes the only part of a rescanned SBOM that legitimately moves: the +// creation timestamp Syft stamps from the wall clock. Same packages, different +// bytes, therefore a different digest on every pass. +func sbomAt(created string) string { + return fmt.Sprintf( + `{"spdxVersion":"SPDX-2.3","creationInfo":{"created":%q},"packages":[]}`, + created) +} + +// testVulnReport is what Grype emits for this content against one database +// snapshot. It carries no timestamp and no database build date (see +// scanner/internal/scan/grype.go), so it is byte-identical across rescans until +// either the matches or the Grype version actually change. +const testVulnReport = `{"matches":[],"source":"sha256:abc","descriptor":{"name":"grype","version":"v0.107.1"}}` + +// mockClient reaches the S3 stand-in behind a broadcaster built by +// newRecordingScanBroadcaster. +func mockClient(t *testing.T, sb *ScanBroadcaster) *s3.MockS3Client { + t.Helper() + + m, ok := sb.s3.Client.(*s3.MockS3Client) + if !ok { + t.Fatalf("s3 client is %T, want *s3.MockS3Client", sb.s3.Client) + } + return m +} + +// putCount is how many objects have been written to S3 so far. Uploads are the +// thing under test, so every assertion here is a delta on this number. +func putCount(t *testing.T, sb *ScanBroadcaster) int { + t.Helper() + + return len(mockClient(t, sb).PutObjectCalls) +} + +// blobStored reports whether the object a blob reference names is still in S3. +func blobStored(t *testing.T, sb *ScanBroadcaster, ref *lexutil.LexBlob) bool { + t.Helper() + + if ref == nil { + return false + } + _, err := sb.s3.Stat(context.Background(), atprotoBlobPath(sb.holdDID, ref.Ref.String())) + return err == nil +} + +// deliverScanResult runs one full scan cycle for a digest: seed the job, hand +// it to the scanner, and deliver the terminal result. Returns the scan record +// the hold wrote. +func deliverScanResult(t *testing.T, sb *ScanBroadcaster, sub *ScanSubscriber, digest, sbom, vulnReport string) *atproto.ScanRecord { + t.Helper() + + seq := seedJobWithDigest(t, sb, digest) + assignJob(t, sb, seq, sub, 0) + sb.handleAck(sub, seq) + sb.addInflight(digest) + + msg := ScannerMessage{Type: "result", Seq: seq, SBOM: sbom, VulnReport: vulnReport} + if vulnReport != "" { + msg.Summary = &VulnerabilitySummary{} + } + sb.handleResult(sub, msg) + + if got := jobStatus(t, sb, seq); got != "completed" { + t.Fatalf("job %d status = %q, want completed", seq, got) + } + + _, record, err := sb.pds.GetScanRecord(context.Background(), digest) + if err != nil { + t.Fatalf("get scan record for %s: %v", digest, err) + } + return record +} + +// ageScanRecord backdates the record's scannedAt, which is what the stale loop +// measures. Rescans in these tests are otherwise seconds apart, and RFC3339 has +// one-second resolution. +func ageScanRecord(t *testing.T, sb *ScanBroadcaster, digest string, age time.Duration) { + t.Helper() + + _, record, err := sb.pds.GetScanRecord(context.Background(), digest) + if err != nil { + t.Fatalf("get scan record for %s: %v", digest, err) + } + record.ScannedAt = time.Now().Add(-age).Format(time.RFC3339) + if _, _, err := sb.pds.CreateScanRecord(context.Background(), record); err != nil { + t.Fatalf("backdate scan record for %s: %v", digest, err) + } +} + +func scannedAt(t *testing.T, record *atproto.ScanRecord) time.Time { + t.Helper() + + at, err := time.Parse(time.RFC3339, record.ScannedAt) + if err != nil { + t.Fatalf("parse scannedAt %q: %v", record.ScannedAt, err) + } + return at +} + +// TestScanHandleResult_UnchangedRescanReusesBlobs is the whole point: a rescan +// of unchanged content produces a fresh SBOM (different bytes, different +// digest) and an identical vulnerability report. The report's stability is the +// evidence that nothing material changed, so both blobs already in S3 are kept +// and nothing is uploaded. +func TestScanHandleResult_UnchangedRescanReusesBlobs(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + const digest = "sha256:unchangedcontent" + + first := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), testVulnReport) + if first.SbomBlob == nil || first.VulnReportBlob == nil { + t.Fatalf("first scan stored blobs sbom:%v vuln:%v, want both", + first.SbomBlob != nil, first.VulnReportBlob != nil) + } + if got := putCount(t, sb); got != 2 { + t.Fatalf("first scan uploaded %d objects, want 2 (SBOM + report)", got) + } + + // The stale loop only rescans records older than the rescan interval. + ageScanRecord(t, sb, digest, 8*24*time.Hour) + before := putCount(t, sb) + + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), testVulnReport) + + if got := putCount(t, sb) - before; got != 0 { + t.Errorf("rescan of unchanged content uploaded %d new objects, want 0: "+ + "the previous blobs are still in S3 and nothing reclaims the ones "+ + "they are replaced by", got) + } + if second.SbomBlob == nil || second.VulnReportBlob == nil { + t.Fatalf("rescan stored blobs sbom:%v vuln:%v, want both", + second.SbomBlob != nil, second.VulnReportBlob != nil) + } + if second.SbomBlob.Ref != first.SbomBlob.Ref { + t.Errorf("SBOM ref moved from %s to %s; the first blob is now orphaned", + first.SbomBlob.Ref, second.SbomBlob.Ref) + } + if second.VulnReportBlob.Ref != first.VulnReportBlob.Ref { + t.Errorf("vulnerability report ref moved from %s to %s", + first.VulnReportBlob.Ref, second.VulnReportBlob.Ref) + } + + // scannedAt must still advance. The stale loop selects on it, so a record + // frozen at the old timestamp is re-selected on every pass forever. + if age := time.Since(scannedAt(t, second)); age > time.Minute { + t.Errorf("scannedAt is %s old after a deduplicated rescan; the stale "+ + "loop will re-select this record on every pass", age) + } +} + +// TestScanHandleResult_ChangedVulnReportUploadsFresh is the other half. A +// vulnerability database update legitimately changes the report, and a changed +// report means the result is not the one on file: both blobs are written fresh +// and the record points at them. +func TestScanHandleResult_ChangedVulnReportUploadsFresh(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + const digest = "sha256:dbupdated" + + first := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), testVulnReport) + before := putCount(t, sb) + + // Same image, newer vulnerability database: a CVE that did not exist last + // week now matches. + const newReport = `{"matches":[{"vulnerability":{"id":"CVE-2026-0001"}}],"source":"sha256:abc","descriptor":{"name":"grype","version":"v0.107.1"}}` + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), newReport) + + if got := putCount(t, sb) - before; got != 2 { + t.Errorf("rescan after a database update uploaded %d objects, want 2: "+ + "the report really changed and the record must point at it", got) + } + if second.VulnReportBlob == nil || second.VulnReportBlob.Ref == first.VulnReportBlob.Ref { + t.Errorf("vulnerability report ref did not move; the record still points " + + "at the pre-update report") + } + if second.SbomBlob == nil || second.SbomBlob.Ref == first.SbomBlob.Ref { + t.Errorf("SBOM ref did not move; a stale-result verdict must not reuse " + + "blobs") + } + if !blobStored(t, sb, second.VulnReportBlob) { + t.Error("the new vulnerability report is not in S3") + } + if !blobStored(t, sb, second.SbomBlob) { + t.Error("the new SBOM is not in S3") + } +} + +// TestScanHandleResult_HealsAFailedSBOMUpload covers what a failed upload +// leaves behind: an ok record with a nil SBOM ref and a real report ref. The +// report still says the result is unchanged, so the report blob is kept, but +// there is no SBOM to keep and the rescan must upload one. +func TestScanHandleResult_HealsAFailedSBOMUpload(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + // Fail the very first PutObject, which is the SBOM: handleResult uploads + // the SBOM before the report. + failing := &failFirstPutClient{MockS3Client: mockClient(t, sb), failCalls: 1} + sb.s3 = &s3.S3Service{Client: failing, Bucket: sb.s3.Bucket} + + const digest = "sha256:failedsbomupload" + + first := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), testVulnReport) + if first.SbomBlob != nil { + t.Fatalf("SBOM upload was supposed to fail, but the record has a ref") + } + if first.VulnReportBlob == nil { + t.Fatal("the vulnerability report upload should still have succeeded") + } + + before := len(failing.PutObjectCalls) + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), testVulnReport) + + if got := len(failing.PutObjectCalls) - before; got != 1 { + t.Errorf("rescan uploaded %d objects, want 1: the report is unchanged "+ + "and already stored, but the missing SBOM has to be written", got) + } + if second.SbomBlob == nil { + t.Error("the rescan did not heal the missing SBOM ref") + } + if !blobStored(t, sb, second.SbomBlob) { + t.Error("the healed SBOM ref points at nothing in S3") + } + if second.VulnReportBlob == nil || second.VulnReportBlob.Ref != first.VulnReportBlob.Ref { + t.Error("the already-stored vulnerability report was uploaded again") + } +} + +// TestScanHandleResult_ReuploadsWhenTheStoredBlobIsGone is the safety valve. A +// record can outlive its blob: object lifecycle rules, a bucket restore, a GC +// pass that did not know these blobs were reachable. Deduplication that trusts +// the record alone would leave the manifest pointing at a 404 forever, because +// every later rescan would make the same decision. +func TestScanHandleResult_ReuploadsWhenTheStoredBlobIsGone(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + const digest = "sha256:vanishedblob" + + first := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), testVulnReport) + if !blobStored(t, sb, first.SbomBlob) || !blobStored(t, sb, first.VulnReportBlob) { + t.Fatal("first scan did not land both blobs in S3") + } + + // Both objects disappear from under the record. + m := mockClient(t, sb) + for k := range m.Objects { + delete(m.Objects, k) + } + before := putCount(t, sb) + + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), testVulnReport) + + if got := putCount(t, sb) - before; got != 2 { + t.Errorf("rescan uploaded %d objects, want 2: the record's blobs are "+ + "gone from S3 and reusing the refs would serve a 404 forever", got) + } + if !blobStored(t, sb, second.SbomBlob) { + t.Error("SBOM ref still points at nothing in S3") + } + if !blobStored(t, sb, second.VulnReportBlob) { + t.Error("vulnerability report ref still points at nothing in S3") + } +} + +// TestScanHandleResult_ReuploadsWhenTheScannerVersionChanged guards the one way +// an identical report can hide a materially different SBOM. The report lists +// matched packages, not every package, so a scanner upgrade that starts +// cataloguing a new ecosystem with no known vulnerabilities changes the SBOM +// and leaves the report byte-identical. Content is pinned by the manifest +// digest, so a version change is the only way that can happen — refusing to +// reuse across versions closes it. +func TestScanHandleResult_ReuploadsWhenTheScannerVersionChanged(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + const digest = "sha256:scannerupgraded" + + first := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), testVulnReport) + + // Pretend the record on file was written by an older scanner build. + _, record, err := sb.pds.GetScanRecord(context.Background(), digest) + if err != nil { + t.Fatalf("get scan record: %v", err) + } + record.ScannerVersion = "atcr-scanner-v0.9.0" + if _, _, err := sb.pds.CreateScanRecord(context.Background(), record); err != nil { + t.Fatalf("rewrite scanner version: %v", err) + } + before := putCount(t, sb) + + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), testVulnReport) + + if got := putCount(t, sb) - before; got != 2 { + t.Errorf("rescan uploaded %d objects, want 2: a different scanner "+ + "version can produce a different SBOM behind an identical report", got) + } + if second.SbomBlob == nil || second.SbomBlob.Ref == first.SbomBlob.Ref { + t.Error("the SBOM from the old scanner build was kept") + } +} + +// TestScanHandleResult_WithoutAReportCannotDeduplicate documents the gap the +// rule leaves open rather than pretending it does not exist. A scanner running +// with vulnerability scanning off sends an SBOM and no report, so there is no +// stable digest to compare and every rescan uploads a new SBOM. Nothing here is +// safe to deduplicate on: the SBOM's own bytes move on every run by design. +func TestScanHandleResult_WithoutAReportCannotDeduplicate(t *testing.T) { + sb := newRecordingScanBroadcaster(t) + sub := newTestScanSubscriber(t, sb, 4) + + const digest = "sha256:novulnscanning" + + deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-01T10:00:00Z"), "") + before := putCount(t, sb) + + second := deliverScanResult(t, sb, sub, digest, sbomAt("2026-09-08T10:00:00Z"), "") + + if got := putCount(t, sb) - before; got != 1 { + t.Errorf("rescan uploaded %d objects, want 1", got) + } + if second.SbomBlob == nil { + t.Error("the rescan lost the SBOM ref") + } + if second.VulnReportBlob != nil { + t.Error("a record with no report must not gain one") + } +} + +// failFirstPutClient fails the first n PutObject calls and then behaves like the +// mock it wraps. It stands in for an S3 endpoint that was briefly unavailable +// while the SBOM was going up. +type failFirstPutClient struct { + *s3.MockS3Client + + mu sync.Mutex + failCalls int +} + +func (c *failFirstPutClient) PutObject(ctx context.Context, input *awss3.PutObjectInput, opts ...func(*awss3.Options)) (*awss3.PutObjectOutput, error) { + c.mu.Lock() + fail := c.failCalls > 0 + if fail { + c.failCalls-- + } + c.mu.Unlock() + + if fail { + // Deliberately not recorded on the mock: PutObjectCalls is the count of + // objects that actually landed, which is what the tests assert on. + return nil, errors.New("s3 unavailable") + } + return c.MockS3Client.PutObject(ctx, input, opts...) +}