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...) +}