amadan
objstore: the re-review's blocker — denied verdicts must not be sticky
A fix from the previous round introduced this one. Classifying an unreadable object as VerdictDenied stopped the sweep stalling, but the verdict was written as a tag AND read back as an answer — so the object was never retried. Its bytes do not change, so its version never changes, so a permissions fix would never take. The spec says the opposite in as many words: "failed and denied are re-issued every sweep — both are usually transient or a permissions fix away". The SSE-KMS case is the one that bites. Tagging needs no KMS access, so the denied tag lands cleanly on exactly the objects that can never be read, and they stay denied for good. Glacier is the same shape: a restore does not change LastModified. denied and failed now read back as unknown; the tag stays as an audit trace of what happened last time. Three more from the same review: **tag() blind-wrote when its own tag READ failed** — destroying exactly the foreign tags the read-merge-write exists to protect, on the one path where we know least about what is there. It fails the page now and lets the next tick retry. **The ten-tag cap turned a working scan into a permanent stall.** An object with eight foreign tags produces an eleven-tag write, which S3 refuses, which abandons the page unwritten every tick forever — the exact stall the denied classification was added to remove. The verdict is recorded for the sweep and the write skipped. **Fresh/Reused split Objects, not Scanned**, so "N of M scanned, of those X judged now" printed visibly wrong arithmetic the moment one object was skipped. Also: a record with an engine enabled and no attestation is refused rather than swept. Unreachable now that the grant door checks, but sweeping reads tenant bytes and the attestation is the consent for that — guessing "probably plaintext" is not this walk's call. The sweep-level version binding had no test: mutating Version() back to ETag passed everything, because the test lister never set LastModified. It does now, and that mutation fails.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YJkJkxZZLgW8X8A5tk69Sq
6 files changed,
+272
−26
docs/superpowers/plans/2026-09-01-carloku-scanning-rollout.md+1 −1docs/superpowers/plans/2026-09-01-object-storage-virus-scanning.md+2 −2internal/objstore/clamav.go+54 −18internal/objstore/clamav_test.go+93 −0internal/objstore/sweep.go+24 −2internal/objstore/sweep_test.go+98 −3
diff --git a/docs/superpowers/plans/2026-09-01-carloku-scanning-rollout.md b/docs/superpowers/plans/2026-09-01-carloku-scanning-rollout.md| index 49673fdc2..e26f04610 100644 |
| --- a/docs/superpowers/plans/2026-09-01-carloku-scanning-rollout.md |
| +++ b/docs/superpowers/plans/2026-09-01-carloku-scanning-rollout.md |
| @@ -108,7 +108,7 @@ step 2, and a `last complete` timestamp. Then verify the two properties that |
| are easy to assume and worth proving: |
| - **Verdicts are on the objects.** `aws s3api get-object-tagging` on any |
| - scanned key shows `carlos-scan`, `carlos-scan-etag` and `carlos-scan-at`, |
| + scanned key shows `carlos-scan`, `carlos-scan-version` and `carlos-scan-at`, |
| and the etag matches the object's current one. |
| - **A rewrite invalidates the verdict.** Overwrite one object, re-run |
| `scan status` — that object must count as unscanned until the next sweep |
diff --git a/docs/superpowers/plans/2026-09-01-object-storage-virus-scanning.md b/docs/superpowers/plans/2026-09-01-object-storage-virus-scanning.md| index f31ffbdff..f02c1f6db 100644 |
| --- a/docs/superpowers/plans/2026-09-01-object-storage-virus-scanning.md |
| +++ b/docs/superpowers/plans/2026-09-01-object-storage-virus-scanning.md |
| @@ -218,7 +218,7 @@ type DenyTag struct { |
| // the bytes it judged. |
| const ( |
| TagVerdict = "carlos-scan" |
| - TagETag = "carlos-scan-etag" |
| + TagETag = "carlos-scan-version" |
| TagAt = "carlos-scan-at" |
| ) |
| ``` |
| @@ -1278,7 +1278,7 @@ func TestClamAVSkipsOversizeWithoutFetching(t *testing.T) { /* stub s3API, asser |
| // A verdict is written with the etag it judged, so a later rewrite reads as |
| // unknown rather than inheriting the old verdict. |
| -func TestClamAVTagsVerdictWithETag(t *testing.T) { /* assert PutObjectTagging carries carlos-scan-etag */ } |
| +func TestClamAVTagsVerdictWithETag(t *testing.T) { /* assert PutObjectTagging carries carlos-scan-version */ } |
| // clamd being unreachable is a PASS error, not thousands of failed tags. |
| // "The engine is unreachable" and "the engine read this object and failed on |
diff --git a/internal/objstore/clamav.go b/internal/objstore/clamav.go| index a31246fbe..8a5470f48 100644 |
| --- a/internal/objstore/clamav.go |
| +++ b/internal/objstore/clamav.go |
| @@ -148,6 +148,19 @@ func (c *clamavScanner) Verdict(ctx context.Context, bucket, key, version string |
| if verdict == "" || taggedVersion == "" { |
| return VerdictUnknown, nil |
| } |
| + // denied and failed are RE-ISSUED every sweep, never carried forward. |
| + // |
| + // Both are transient by nature — a permissions fix, a Glacier restore, an |
| + // engine that was having a bad day — and the tag is an audit trace of what |
| + // happened last time, not a verdict to inherit. Reading them back as |
| + // answers would make them permanent: the object is not rewritten, so its |
| + // version never changes, so it would never be retried and the repair would |
| + // never take. An SSE-KMS denial is the case that bites, because the |
| + // tagging call needs no KMS access and therefore succeeds. |
| + switch Verdict(verdict) { |
| + case VerdictDenied, VerdictFailed: |
| + return VerdictUnknown, nil |
| + } |
| if taggedVersion != version { |
| // The bytes changed under the verdict. Not a failure — just unknown |
| // again, which is exactly what makes the next sweep re-scan it. |
| @@ -182,12 +195,12 @@ func (c *clamavScanner) Scan(ctx context.Context, bucket, key, version string, s |
| // account in an adopted bucket, would stall that store's coverage |
| // forever while every surface reported a sweep merely "in progress". |
| if isDenied(err) { |
| - if terr := c.tag(ctx, bucket, key, version, VerdictDenied); terr != nil { |
| - // Tagging it is best effort: if we cannot read the object we |
| - // may well not be able to tag it either. The verdict still |
| - // stands for this sweep's count. |
| - return VerdictDenied, nil |
| - } |
| + // Tagging is best effort and its failure is deliberately ignored: |
| + // if we may not read the object we may well not be able to tag it, |
| + // and the verdict still stands for this sweep's count. The tag is |
| + // only an audit trace here — Verdict re-issues denied rather than |
| + // reading it back, so nothing depends on this landing. |
| + _ = c.tag(ctx, bucket, key, version, VerdictDenied) |
| return VerdictDenied, nil |
| } |
| // Deleted between the listing and the fetch. Routine on a bucket |
| @@ -260,30 +273,53 @@ func (c *clamavScanner) tag(ctx context.Context, bucket, key, version string, v |
| {Key: s2aws(TagAt), Value: s2aws(time.Now().UTC().Format(time.RFC3339))}, |
| } |
| // Read-merge-write. One extra call per object per LIFETIME, not per sweep: |
| - // an object whose verdict still matches its ETag never reaches Scan again. |
| - if existing, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{ |
| + // an object whose verdict still matches its version never reaches Scan |
| + // again. |
| + existing, err := c.s3.GetObjectTagging(ctx, &awss3.GetObjectTaggingInput{ |
| Bucket: &bucket, |
| Key: &key, |
| - }); err == nil { |
| - for _, t := range existing.TagSet { |
| - if !isScanTag(aws2s(t.Key)) { |
| - set = append(set, t) |
| - } |
| + }) |
| + if err != nil { |
| + // A tag set we cannot read must NOT be written over. PutObjectTagging |
| + // replaces the whole set, so proceeding here would destroy exactly the |
| + // foreign tags this merge exists to protect — silently, and on the one |
| + // path where we know least about what is there. The sweep read this |
| + // same object's tags moments ago through Verdict, so a failure now is |
| + // transient: fail the page and let the next tick retry. |
| + if isDenied(err) || isGone(err) { |
| + return nil // nothing we could ever write, and not a pass failure |
| + } |
| + return fmt.Errorf("objstore: read tags before writing %s/%s: %w", bucket, key, err) |
| + } |
| + for _, t := range existing.TagSet { |
| + if !isScanTag(aws2s(t.Key)) { |
| + set = append(set, t) |
| } |
| } |
| - // A tag set we cannot read is not a reason to refuse to record a verdict: |
| - // the verdict is the point, and S3 caps the set at ten anyway. |
| - _, err := c.s3.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{ |
| + |
| + // S3 caps an object at ten tags. An object already carrying eight foreign |
| + // ones cannot also hold our three, and erroring here would abandon the |
| + // page unwritten every single tick — the exact permanent stall the denied |
| + // classification was added to remove. Record the verdict for this sweep |
| + // and move on; the object is simply re-scanned next time rather than |
| + // blocking every object behind it. |
| + if len(set) > maxObjectTags { |
| + return nil |
| + } |
| + |
| + if _, err := c.s3.PutObjectTagging(ctx, &awss3.PutObjectTaggingInput{ |
| Bucket: &bucket, |
| Key: &key, |
| Tagging: &s3types.Tagging{TagSet: set}, |
| - }) |
| - if err != nil { |
| + }); err != nil { |
| return fmt.Errorf("objstore: tag %s/%s: %w", bucket, key, err) |
| } |
| return nil |
| } |
| +// maxObjectTags is S3's per-object tag limit. |
| +const maxObjectTags = 10 |
| + |
| // isScanTag is the set this scanner owns and may overwrite. Everything else on |
| // the object belongs to whoever put it there. |
| func isScanTag(key string) bool { |
diff --git a/internal/objstore/clamav_test.go b/internal/objstore/clamav_test.go| index 21b042a09..b0ee1490a 100644 |
| --- a/internal/objstore/clamav_test.go |
| +++ b/internal/objstore/clamav_test.go |
| @@ -411,6 +411,9 @@ func TestClamAVUnreadableObjectIsDeniedNotAPassFailure(t *testing.T) { |
| if v != VerdictDenied { |
| t.Fatalf("verdict = %q, want denied", v) |
| } |
| + // The tag is an audit trace, not an answer: the next sweep |
| + // re-issues denied rather than reading it back. See |
| + // TestClamAVDeniedAndFailedAreReIssuedNotInherited. |
| if wrote[TagVerdict] != string(VerdictDenied) { |
| t.Fatalf("tags = %v", wrote) |
| } |
| @@ -498,3 +501,93 @@ func TestClamAVPreservesForeignTags(t *testing.T) { |
| t.Fatalf("the scanner did not overwrite its own stale etag tag: %v", wrote) |
| } |
| } |
| + |
| +// denied and failed are RE-ISSUED every sweep, never carried forward. |
| +// |
| +// Both are transient — a permissions fix, a Glacier restore, an engine having |
| +// a bad day. Reading them back as answers would make them permanent: the |
| +// object is not rewritten, so its version never changes, so it would never be |
| +// retried and the repair would never take. The SSE-KMS case is the one that |
| +// bites, because tagging needs no KMS access and so the tag does land. |
| +func TestClamAVDeniedAndFailedAreReIssuedNotInherited(t *testing.T) { |
| + for _, stuck := range []Verdict{VerdictDenied, VerdictFailed} { |
| + t.Run(string(stuck), func(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(TagVerdict, string(stuck), TagVersion, "v1"), nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + |
| + v, err := c.Verdict(context.Background(), "b", "k", "v1") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictUnknown { |
| + t.Fatalf("verdict = %q, want unknown — %q must be retried, not inherited", v, stuck) |
| + } |
| + }) |
| + } |
| + |
| + // clean and infected ARE carried forward: they are answers about bytes |
| + // that have not changed. |
| + for _, keep := range []Verdict{VerdictClean, VerdictInfected, VerdictSkipped} { |
| + t.Run(string(keep), func(t *testing.T) { |
| + s3 := &stubScanS3{t: t} |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(TagVerdict, string(keep), TagVersion, "v1"), nil |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{}} |
| + v, err := c.Verdict(context.Background(), "b", "k", "v1") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != keep { |
| + t.Fatalf("verdict = %q, want %q carried forward", v, keep) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A tag set we cannot READ must never be written over. PutObjectTagging |
| +// replaces the whole set, so writing blind here destroys exactly the foreign |
| +// tags the merge exists to protect — on the one path where we know least about |
| +// what is there. |
| +func TestClamAVDoesNotBlindWriteWhenTagReadFails(t *testing.T) { |
| + s3 := &stubScanS3{t: t} // putTagging unset: any write fails the test |
| + s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) { |
| + return objectBody("hello"), nil |
| + } |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return nil, apiErr("SlowDown") |
| + } |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| + |
| + if _, err := c.Scan(context.Background(), "b", "k", "v1", 5); err == nil { |
| + t.Fatal("an unreadable tag set was written over anyway") |
| + } |
| +} |
| + |
| +// An object already carrying eight foreign tags cannot also hold our three — |
| +// S3 caps an object at ten. Erroring there would abandon the page unwritten |
| +// every tick forever, which is the exact permanent stall the denied |
| +// classification was added to remove. Record the verdict, skip the write. |
| +func TestClamAVSkipsTagWriteRatherThanStallAtTheTagLimit(t *testing.T) { |
| + var foreign []string |
| + for i := 0; i < 8; i++ { |
| + foreign = append(foreign, "own-"+string(rune('a'+i)), "v") |
| + } |
| + s3 := &stubScanS3{t: t} // putTagging unset: an 11-tag write fails the test |
| + s3.getObject = func(*awss3.GetObjectInput) (*awss3.GetObjectOutput, error) { |
| + return objectBody("hello"), nil |
| + } |
| + s3.allowTagRead(foreign...) |
| + c := &clamavScanner{s3: s3, engine: &fakeEngine{res: clamd.Result{Status: clamd.StatusClean}}, maxBytes: 1 << 20} |
| + |
| + v, err := c.Scan(context.Background(), "b", "k", "v1", 5) |
| + if err != nil { |
| + t.Fatalf("a heavily tagged object stalled the sweep: %v", err) |
| + } |
| + if v != VerdictClean { |
| + t.Fatalf("verdict = %q, want clean — it was scanned, just not recorded", v) |
| + } |
| +} |
diff --git a/internal/objstore/sweep.go b/internal/objstore/sweep.go| index aa834015d..53f589fa0 100644 |
| --- a/internal/objstore/sweep.go |
| +++ b/internal/objstore/sweep.go |
| @@ -148,6 +148,17 @@ func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scan |
| return markNotApplicable(ctx, s, acct, app) |
| } |
| + // An engine enabled with no attestation on record is refused rather than |
| + // swept. The grant door will not create that state any more, so this is |
| + // unreachable in practice — but sweeping reads tenant object bytes, and |
| + // the attestation is the consent for that. Guessing "probably plaintext" |
| + // is not a decision this walk gets to make on the member's behalf. |
| + if rec.Scanning.Requested == nil { |
| + return nil, fmt.Errorf("objstore: %s/%s: scanning is enabled but no attestation is on record "+ |
| + "— the member runs `carlos store scan request` to say whether these objects are readable", |
| + acct, app) |
| + } |
| + |
| if sc == nil { |
| return nil, fmt.Errorf("objstore: %s/%s has scanning enabled but no engine is configured "+ |
| "(set CARLOS_STORE_SCANNER)", acct, app) |
| @@ -228,8 +239,10 @@ func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scan |
| if err != nil { |
| return nil, err |
| } |
| - fresh++ |
| - } else { |
| + if isCoverage(v) { |
| + fresh++ |
| + } |
| + } else if isCoverage(v) { |
| reused++ |
| } |
| side.Current.count(v, o.Size) |
| @@ -267,6 +280,15 @@ func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scan |
| } |
| } |
| +// isCoverage is a verdict that counts as Scanned. Fresh and Reused split THAT |
| +// number, not the object count: an object that was skipped or denied was never |
| +// judged at all, so calling its verdict "fresh" would make the printed |
| +// arithmetic — "N of M scanned, of those X judged now" — visibly wrong the |
| +// moment a single object is too large. |
| +func isCoverage(v Verdict) bool { |
| + return v == VerdictClean || v == VerdictInfected |
| +} |
| + |
| // count folds one object's verdict into the rollup. |
| // |
| // Scanned and Unscanned are not complements, and the arithmetic here is the |
diff --git a/internal/objstore/sweep_test.go b/internal/objstore/sweep_test.go| index 0b7975a5a..2ea2ade96 100644 |
| --- a/internal/objstore/sweep_test.go |
| +++ b/internal/objstore/sweep_test.go |
| @@ -20,11 +20,22 @@ type memLister struct { |
| p *MemProvisioner |
| pageSize int |
| calls int |
| + // modified is each key's last-modified time. It is half of the version a |
| + // verdict binds to, so a lister that left it zero would make every sweep |
| + // test blind to that half. |
| + modified map[string]time.Time |
| // before runs at the top of each ListPage, so a test can make the world |
| // change underneath a sweep at the one moment that matters. |
| before func() |
| } |
| +func (l *memLister) touch(key string, at time.Time) { |
| + if l.modified == nil { |
| + l.modified = map[string]time.Time{} |
| + } |
| + l.modified[key] = at |
| +} |
| + |
| func (l *memLister) ListPage(_ context.Context, bucket, startAfter string, _ int32) ([]ScanObject, bool, error) { |
| l.calls++ |
| if l.before != nil { |
| @@ -50,10 +61,15 @@ func (l *memLister) ListPage(_ context.Context, bucket, startAfter string, _ int |
| for _, k := range keys { |
| body := l.p.Buckets[bucket][k] |
| sum := md5.Sum(body) |
| + mod := l.modified[k] |
| + if mod.IsZero() { |
| + mod = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC) |
| + } |
| objs = append(objs, ScanObject{ |
| - Key: k, |
| - ETag: hex.EncodeToString(sum[:]), |
| - Size: int64(len(body)), |
| + Key: k, |
| + ETag: hex.EncodeToString(sum[:]), |
| + Modified: mod, |
| + Size: int64(len(body)), |
| }) |
| } |
| return objs, truncated, nil |
| @@ -471,3 +487,82 @@ func (g *goneScanner) Scan(_ context.Context, _, key, _ string, _ int64) (Verdic |
| func (g *goneScanner) Provenance(context.Context) (string, string, error) { |
| return "clamav", "test/0/never", nil |
| } |
| + |
| +// The sweep binds verdicts to the object VERSION, not the ETag alone. |
| +// |
| +// Same bytes, later last-modified: an overwrite that happened to produce the |
| +// same content hash. A chosen-prefix MD5 collision is exactly this shape, and |
| +// it is the case enforcement exists for — an app that accepts uploads and |
| +// serves them back. Passing only the ETag here would let the payload inherit |
| +// the clean verdict of the object it replaced, and every existing sweep test |
| +// would still have passed. |
| +func TestSweepRescansWhenOnlyLastModifiedChanges(t *testing.T) { |
| + ctx := context.Background() |
| + s, p, sc, bucket := sweepFixture(t, 1) |
| + l := &memLister{p: p} |
| + |
| + first, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if first.LastComplete.Fresh != 1 { |
| + t.Fatalf("setup: fresh = %d, want 1", first.LastComplete.Fresh) |
| + } |
| + |
| + // The bytes are untouched — so the ETag is identical — but the object was |
| + // rewritten. |
| + _ = bucket |
| + l.touch(objectName(0), time.Date(2026, 9, 2, 0, 0, 0, 0, time.UTC)) |
| + |
| + second, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if second.LastComplete.Fresh != 1 || second.LastComplete.Reused != 0 { |
| + t.Fatalf("fresh=%d reused=%d — an overwritten object inherited its old verdict", |
| + second.LastComplete.Fresh, second.LastComplete.Reused) |
| + } |
| +} |
| + |
| +// Fresh and Reused split SCANNED, not the object count. An object that was |
| +// skipped was never judged, so counting its verdict as "judged during this |
| +// sweep" makes the printed arithmetic — "N of M scanned, of those X judged |
| +// now" — visibly wrong the moment one object is too large. |
| +func TestSweepFreshAndReusedSplitScannedNotObjects(t *testing.T) { |
| + s, p, sc, _ := sweepFixture(t, 3) |
| + sc.ScanAs(objectName(1), VerdictSkipped) |
| + |
| + side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + got := side.LastComplete |
| + if got.Fresh+got.Reused != got.Scanned { |
| + t.Fatalf("fresh(%d)+reused(%d) = %d, but scanned = %d of %d objects", |
| + got.Fresh, got.Reused, got.Fresh+got.Reused, got.Scanned, got.Objects) |
| + } |
| + if got.Fresh != 2 { |
| + t.Fatalf("fresh = %d, want 2 — the skipped object was never judged", got.Fresh) |
| + } |
| +} |
| + |
| +// An engine enabled with no attestation on record is refused, not swept. |
| +// Sweeping reads tenant object bytes and the attestation is the consent for |
| +// that; guessing "probably plaintext" is not this walk's decision to make. |
| +func TestSweepRefusesWhenNoAttestationIsOnRecord(t *testing.T) { |
| + ctx := context.Background() |
| + s, p, sc, _ := sweepFixture(t, 1) |
| + |
| + rec, etag, err := Get(ctx, s, "bab", "titogo") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + rec.Scanning.Requested = nil |
| + if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil { |
| + t.Fatal("a store with no attestation on record was swept anyway") |
| + } |
| +} |