amadan

carlos / platform Public

objstore: the sweep walk, resumable and safe against teardown

A reconciler, not a pipeline. Nothing in CARLOS sees a PUT, so there is
no upload to hook — but what a walk can do that a pipeline cannot is
count what it has NOT seen. A pipeline's misses are invisible by
construction; a walk that lists the bucket knows exactly how many
objects carry no verdict. That number is the whole deliverable.

The pre-write re-read is not belt-and-braces. A fresh sweep's first
write is create-only by construction, so the CAS retry guard cannot see
this case at all: the sweep reads an enabled store.json, spends seconds
listing and scanning, a teardown purges the whole prefix, and the
create-only write then SUCCEEDS. The orphan that leaves is expensive —
the app-name claim gate refuses any name whose prefix still holds a
key, rename is refused, and box-side auto-purge never reaches it
because the tombstone went last. The app name is gone for good.
Verified by disabling the re-read: the test writes the orphan.

On a lost CAS the guard looks before retrying. A conditional write
against a deleted key also reads as ErrCASConflict, so a blind retry
would fall back to create-only and resurrect the sidecar teardown just
removed.

Skipped counts as unscanned, infected counts as scanned. An object over
the size ceiling was never judged; an infected one we know exactly what
it is. Getting that backwards would make the audit sentence false in
precisely the direction that matters.

StartAfter, not ContinuationToken: a token's cross-process lifetime is
unspecified and, on Ceph, unmeasured. A key resumes after a console
restart or a change of S3 vendor.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01YJkJkxZZLgW8X8A5tk69Sq
Paul Campbell pushed by paul@keymail.dev de4e2f023495957d14506a1f690c4ce622b452a8 parent 829c8f1
3 files changed, +688 −2
  • internal/objstore/memscanner.go +19 −2
  • internal/objstore/sweep.go +319 −0
  • internal/objstore/sweep_test.go +350 −0
diff --git a/internal/objstore/memscanner.go b/internal/objstore/memscanner.go
index 3c19340f4..550cb2dd4 100644
--- a/internal/objstore/memscanner.go
+++ b/internal/objstore/memscanner.go
@@ -13,6 +13,7 @@ type MemScanner struct {
enabled map[string]bool
verdicts map[string]scanRecord
failures map[string]error
+ scanAs map[string]Verdict
}
// scanRecord is one object's verdict AND the etag it was judged against. The
@@ -103,8 +104,24 @@ func (s *MemScanner) Scan(ctx context.Context, bucket, key, etag string, size in
if err := s.takeFailure("Scan"); err != nil {
return VerdictUnknown, err
}
- s.verdicts[memScanKey(bucket, key)] = scanRecord{etag: etag, verdict: VerdictClean}
- return VerdictClean, nil
+ v := VerdictClean
+ if got, ok := s.scanAs[key]; ok {
+ v = got
+ }
+ s.verdicts[memScanKey(bucket, key)] = scanRecord{etag: etag, verdict: v}
+ return v, nil
+}
+
+// ScanAs makes Scan answer v for key rather than clean, so a test can stage a
+// skipped or infected object without a real engine. Skipped in particular is
+// not a hypothetical: it is every object over the size ceiling.
+func (s *MemScanner) ScanAs(key string, v Verdict) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if s.scanAs == nil {
+ s.scanAs = map[string]Verdict{}
+ }
+ s.scanAs[key] = v
}
func (s *MemScanner) Enforcement() DenyTag {
diff --git a/internal/objstore/sweep.go b/internal/objstore/sweep.go
new file mode 100644
index 000000000..edce0db1b
--- /dev/null
+++ b/internal/objstore/sweep.go
@@ -0,0 +1,319 @@
+// The sweep (spec §7): a resumable reconciliation pass over one bucket.
+//
+// It is a reconciler, not a pipeline. Nothing in CARLOS sees a PUT — the
+// control plane hands an app its own credentials and then never touches the
+// data plane — so there is no upload to hook. What a reconciler can do that a
+// pipeline cannot is COUNT WHAT IT HAS NOT SEEN: a pipeline's misses are
+// invisible by construction, while a walk that lists the bucket knows exactly
+// how many objects carry no verdict. That number is the whole point. An audit
+// answer needs two sentences, and the second one is the honest one.
+package objstore
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "time"
+
+ awss3 "github.com/aws/aws-sdk-go-v2/service/s3"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// sweepPageSize is how many keys one listing page asks for. The page is the
+// unit of durable progress: everything on it is scanned, then the sidecar is
+// written once. Smaller pages mean more sidecar writes; larger ones mean more
+// work lost when a tick's budget runs out mid-page.
+const sweepPageSize = 1000
+
+// ScanObject is one key as the listing reports it. Size and ETag come free on
+// the listing page, which is why the sweep never needs a HeadObject: the size
+// decides the ceiling before any fetch, and the ETag binds the verdict.
+type ScanObject struct {
+ Key string
+ ETag string
+ Size int64
+}
+
+// ObjectLister is the one listing call the sweep makes, narrowed so the walk
+// is testable without S3 or credentials.
+type ObjectLister interface {
+ // ListPage returns up to limit keys ordered after startAfter, and whether
+ // more remain.
+ ListPage(ctx context.Context, bucket, startAfter string, limit int32) (objs []ScanObject, truncated bool, err error)
+}
+
+// s3Lister adapts an S3 client to ObjectLister.
+type s3Lister struct{ s3 listPageAPI }
+
+type listPageAPI interface {
+ ListObjectsV2(ctx context.Context, in *awss3.ListObjectsV2Input, optFns ...func(*awss3.Options)) (*awss3.ListObjectsV2Output, error)
+}
+
+// NewObjectLister adapts an S3 client to the sweep's listing seam.
+func NewObjectLister(c listPageAPI) ObjectLister { return &s3Lister{s3: c} }
+
+func (l *s3Lister) ListPage(ctx context.Context, bucket, startAfter string, limit int32) ([]ScanObject, bool, error) {
+ in := &awss3.ListObjectsV2Input{Bucket: &bucket, MaxKeys: &limit}
+ // StartAfter, not ContinuationToken. A token's lifetime across processes
+ // is unspecified and, on Ceph, unmeasured; a key is a portable resume
+ // point that survives a console restart and a change of S3 vendor.
+ if startAfter != "" {
+ in.StartAfter = &startAfter
+ }
+ out, err := l.s3.ListObjectsV2(ctx, in)
+ if err != nil {
+ return nil, false, fmt.Errorf("objstore: list %s: %w", bucket, err)
+ }
+ objs := make([]ScanObject, 0, len(out.Contents))
+ for _, o := range out.Contents {
+ obj := ScanObject{Key: aws2s(o.Key), ETag: normalizeETag(aws2s(o.ETag))}
+ if o.Size != nil {
+ obj.Size = *o.Size
+ }
+ objs = append(objs, obj)
+ }
+ truncated := out.IsTruncated != nil && *out.IsTruncated
+ return objs, truncated, nil
+}
+
+// errStandDown is not a failure. It is this sweep noticing that someone else —
+// another driver, or a teardown — is ahead of it, and that the right move is
+// to stop writing. RunSweep converts it to a nil error.
+var errStandDown = errors.New("objstore: sweep standing down")
+
+// RunSweep advances one bucket's coverage by up to budget of work and returns
+// the sidecar as it now stands.
+//
+// A nil sidecar with a nil error means there was nothing to do: no store, no
+// scanning enabled, or an app in the trash.
+func RunSweep(ctx context.Context, s release.Store, lister ObjectLister, sc Scanner,
+ acct, app string, budget time.Duration) (*ScanSidecar, error) {
+
+ rec, _, err := Get(ctx, s, acct, app)
+ if err != nil {
+ return nil, err
+ }
+ if rec == nil || rec.Provisioned == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil {
+ return nil, nil
+ }
+
+ // A trashed store is destroyed within 30 days. Scanning it bills for
+ // egress on data that is on its way out, and reports coverage for an app
+ // nobody can reach.
+ tomb, err := release.ReadAppTombstone(ctx, s, acct, app)
+ if err != nil {
+ return nil, err
+ }
+ if tomb != nil {
+ return nil, nil
+ }
+
+ // A scanner cannot read ciphertext. Sweeping a client-encrypted store
+ // would report every object clean, which is worse than reporting nothing:
+ // it is a true-looking sentence about bytes nobody examined. Mark it and
+ // stop.
+ if rec.Scanning.Requested != nil && rec.Scanning.Requested.Encrypted {
+ return markNotApplicable(ctx, s, 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)
+ }
+
+ engine, signature, err := sc.Provenance(ctx)
+ if err != nil {
+ return nil, fmt.Errorf("objstore: %s/%s: engine unreachable: %w", acct, app, err)
+ }
+ // Engine drift refuses rather than guesses. Verdicts already in the bucket
+ // were written by a different engine under its own conventions, and
+ // recounting them as unscanned would destroy the coverage history. An
+ // operator changes engine by revoking and re-granting, deliberately.
+ if rec.Scanning.Enabled.Engine != engine {
+ return nil, fmt.Errorf("objstore: %s/%s: record says engine %q, configured engine is %q",
+ acct, app, rec.Scanning.Enabled.Engine, engine)
+ }
+
+ side, etag, err := GetSidecar(ctx, s, acct, app)
+ if err != nil {
+ return nil, err
+ }
+ if side == nil {
+ side = &ScanSidecar{}
+ }
+ side.NotApplicable = false
+ if side.Current == nil {
+ side.Current = &Sweep{StartedAt: time.Now().UTC()}
+ }
+ side.Current.Engine = engine
+ side.Current.Signature = signature
+
+ bucket := rec.Provisioned.Bucket
+ deadline := time.Now().Add(budget)
+
+ for {
+ pageStart := side.Current.After
+ objs, truncated, err := lister.ListPage(ctx, bucket, pageStart, sweepPageSize)
+ if err != nil {
+ return nil, err
+ }
+
+ for _, o := range objs {
+ if err := ctx.Err(); err != nil {
+ return nil, err
+ }
+ v, err := sc.Verdict(ctx, bucket, o.Key, o.ETag)
+ if err != nil {
+ return nil, err
+ }
+ if v == VerdictUnknown {
+ // A scan error here is the ENGINE failing, not this object:
+ // the next object would fail identically. Abandon the page
+ // without writing. Nothing is lost but the listing — every
+ // verdict already reached is a tag in the bucket, so the
+ // re-run reads them back without re-scanning a byte.
+ if v, err = sc.Scan(ctx, bucket, o.Key, o.ETag, o.Size); err != nil {
+ return nil, err
+ }
+ }
+ side.Current.count(v, o.Size)
+ }
+
+ side.Current.At = time.Now().UTC()
+ if len(objs) > 0 {
+ side.Current.After = objs[len(objs)-1].Key
+ }
+
+ last := !truncated
+ if last {
+ // The completed sweep becomes the answer, and Current clears. A
+ // single field would have the next tick overwrite the compliance
+ // artifact with page-one partials.
+ side.Current.After = ""
+ side.LastComplete = side.Current
+ side.Current = nil
+ }
+
+ newETag, err := writeSidecar(ctx, s, acct, app, side, etag, pageStart)
+ if errors.Is(err, errStandDown) {
+ return side, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ etag = newETag
+
+ if last || time.Now().After(deadline) {
+ return side, nil
+ }
+ }
+}
+
+// count folds one object's verdict into the rollup.
+//
+// Scanned and Unscanned are not complements, and the arithmetic here is the
+// reason the audit answer needs two sentences. An infected object WAS scanned
+// — we know what it is. A skipped one was not judged at all.
+func (sw *Sweep) count(v Verdict, size int64) {
+ sw.Objects++
+ sw.Bytes += size
+ switch v {
+ case VerdictClean:
+ sw.Scanned++
+ case VerdictInfected:
+ sw.Scanned++
+ sw.Infected++
+ case VerdictSkipped:
+ sw.Skipped++
+ sw.Unscanned++
+ case VerdictDenied:
+ sw.Denied++
+ sw.Unscanned++
+ case VerdictFailed:
+ sw.Failed++
+ sw.Unscanned++
+ default:
+ sw.Unscanned++
+ }
+}
+
+// markNotApplicable records that no coverage is claimed for this store, and
+// does no listing at all.
+func markNotApplicable(ctx context.Context, s release.Store, acct, app string) (*ScanSidecar, error) {
+ side, etag, err := GetSidecar(ctx, s, acct, app)
+ if err != nil {
+ return nil, err
+ }
+ if side != nil && side.NotApplicable && side.Current == nil {
+ return side, nil // already recorded; no write, no CAS churn
+ }
+ if side == nil {
+ side = &ScanSidecar{}
+ }
+ side.NotApplicable = true
+ side.Current = nil
+ if _, err := writeSidecar(ctx, s, acct, app, side, etag, ""); err != nil {
+ if errors.Is(err, errStandDown) {
+ return side, nil
+ }
+ return nil, err
+ }
+ return side, nil
+}
+
+// writeSidecar performs the pre-write re-read and the CAS retry guard.
+//
+// The re-read closes a hole the retry guard alone cannot. A FRESH sweep's
+// first write is create-only by construction — no sidecar exists, so the etag
+// is "". Sequence: the sweep reads an enabled store.json, spends seconds
+// listing and scanning a page, a teardown runs and PurgeApp deletes the whole
+// prefix, and the sweep then writes scan.json create-only, which SUCCEEDS. The
+// result is an orphan under a purged app, and that orphan is expensive: the
+// app-name claim gate refuses any name whose prefix still holds a key, rename
+// is refused, and box-side auto-purge never reaches it because the tombstone
+// was written last. The app name becomes permanently unclaimable.
+//
+// The one-round-trip race this still leaves is the same one
+// deleteRecordIfUnchanged already accepts.
+func writeSidecar(ctx context.Context, s release.Store, acct, app string,
+ side *ScanSidecar, etag, pageStart string) (string, error) {
+
+ rec, _, err := Get(ctx, s, acct, app)
+ if err != nil {
+ return "", err
+ }
+ if rec == nil || rec.Scanning == nil || rec.Scanning.Enabled == nil {
+ return "", errStandDown
+ }
+
+ newETag, err := PutSidecar(ctx, s, acct, app, side, etag)
+ if err == nil {
+ return newETag, nil
+ }
+ if !errors.Is(err, release.ErrCASConflict) {
+ return "", err
+ }
+
+ // Someone else wrote. A conditional write against a DELETED key also
+ // reads as ErrCASConflict, so a blind retry would fall back to a
+ // create-only write and resurrect a sidecar teardown just removed. Look
+ // before retrying.
+ current, freshETag, gerr := GetSidecar(ctx, s, acct, app)
+ if gerr != nil {
+ return "", gerr
+ }
+ if current == nil || current.Current == nil || current.Current.After != pageStart {
+ // Deleted, completed by another driver, or advanced past us. In every
+ // case this sweep is no longer the one making progress.
+ return "", errStandDown
+ }
+ newETag, err = PutSidecar(ctx, s, acct, app, side, freshETag)
+ if err != nil {
+ if errors.Is(err, release.ErrCASConflict) {
+ return "", errStandDown
+ }
+ return "", err
+ }
+ return newETag, nil
+}
diff --git a/internal/objstore/sweep_test.go b/internal/objstore/sweep_test.go
new file mode 100644
index 000000000..4d86bbd2c
--- /dev/null
+++ b/internal/objstore/sweep_test.go
@@ -0,0 +1,350 @@
+package objstore
+
+import (
+ "context"
+ "crypto/md5"
+ "encoding/hex"
+ "errors"
+ "os"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// memLister lists a MemProvisioner bucket, paginated, with real content-hash
+// ETags so the etag binding in Verdict is exercised rather than stubbed away.
+type memLister struct {
+ p *MemProvisioner
+ pageSize int
+ calls int
+ // 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) ListPage(_ context.Context, bucket, startAfter string, _ int32) ([]ScanObject, bool, error) {
+ l.calls++
+ if l.before != nil {
+ l.before()
+ }
+ keys := make([]string, 0, len(l.p.Buckets[bucket]))
+ for k := range l.p.Buckets[bucket] {
+ if k > startAfter {
+ keys = append(keys, k)
+ }
+ }
+ sort.Strings(keys)
+
+ size := l.pageSize
+ if size <= 0 {
+ size = sweepPageSize
+ }
+ truncated := len(keys) > size
+ if truncated {
+ keys = keys[:size]
+ }
+ objs := make([]ScanObject, 0, len(keys))
+ for _, k := range keys {
+ body := l.p.Buckets[bucket][k]
+ sum := md5.Sum(body)
+ objs = append(objs, ScanObject{
+ Key: k,
+ ETag: hex.EncodeToString(sum[:]),
+ Size: int64(len(body)),
+ })
+ }
+ return objs, truncated, nil
+}
+
+// refuseLister fails the test if it is ever asked for a page — for the cases
+// where the right amount of work is none.
+type refuseLister struct{ t *testing.T }
+
+func (l *refuseLister) ListPage(context.Context, string, string, int32) ([]ScanObject, bool, error) {
+ l.t.Fatal("listed a bucket that should never have been swept")
+ return nil, false, nil
+}
+
+// sweepFixture stands up a provisioned, scanning-enabled store with n objects.
+func sweepFixture(t *testing.T, n int) (release.Store, *MemProvisioner, *MemScanner, string) {
+ t.Helper()
+ s := release.NewDirStore(t.TempDir())
+ _, pub := testSealKeypair(t)
+ p := NewMemProvisioner("eu-west-1")
+ rec := rotateFixture(t, s, pub, p)
+ enableScanning(t, s, "bab", "titogo")
+
+ bucket := rec.Provisioned.Bucket
+ for i := 0; i < n; i++ {
+ p.Buckets[bucket][objectName(i)] = []byte(strings.Repeat("x", i+1))
+ }
+ return s, p, NewMemScanner(), bucket
+}
+
+func objectName(i int) string {
+ return "uploads/" + string(rune('a'+i)) + ".txt"
+}
+
+// sweepToCompletion drives RunSweep until a sweep finishes, so a paginated
+// test does not have to guess how many ticks that takes. It bounds the loop:
+// a sweep that never completes is a bug, not a slow test.
+func sweepToCompletion(t *testing.T, s release.Store, l ObjectLister, sc Scanner) *ScanSidecar {
+ t.Helper()
+ var side *ScanSidecar
+ for i := 0; i < 50; i++ {
+ var err error
+ side, err = RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
+ if err != nil {
+ t.Fatalf("sweep %d: %v", i, err)
+ }
+ if side != nil && side.LastComplete != nil {
+ return side
+ }
+ }
+ t.Fatal("sweep never completed in 50 ticks")
+ return nil
+}
+
+// A synchronous engine's FIRST pass reports objects as scanned. Counting them
+// unscanned because "the next sweep reads the result" is GuardDuty's shape,
+// and applying it to ClamAV would make the audit sentence false for a full
+// re-sweep interval.
+func TestSweepCountsSynchronousVerdicts(t *testing.T) {
+ s, p, sc, _ := sweepFixture(t, 3)
+
+ side, err := RunSweep(context.Background(), s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if side.LastComplete == nil {
+ t.Fatal("a sweep that listed every object in one page did not complete")
+ }
+ got := side.LastComplete
+ if got.Objects != 3 || got.Scanned != 3 || got.Unscanned != 0 {
+ t.Fatalf("objects=%d scanned=%d unscanned=%d, want 3/3/0", got.Objects, got.Scanned, got.Unscanned)
+ }
+ if got.Engine != "clamav" || got.Signature == "" {
+ t.Fatalf("no provenance recorded: engine=%q signature=%q", got.Engine, got.Signature)
+ }
+ if side.Current != nil {
+ t.Fatal("a completed sweep left Current set")
+ }
+}
+
+// LastComplete survives the tick after completion. A single Sweep field would
+// overwrite the audit answer with page-one partials every cycle — the
+// compliance artifact destroyed on a timer.
+func TestSweepLastCompleteSurvivesNextTick(t *testing.T) {
+ s, p, sc, _ := sweepFixture(t, 5)
+ l := &memLister{p: p, pageSize: 2}
+
+ side := sweepToCompletion(t, s, l, sc)
+ if side.LastComplete.Objects != 5 {
+ t.Fatalf("completed sweep counted %d objects, want 5", side.LastComplete.Objects)
+ }
+
+ // One more tick: a fresh sweep starts and gets exactly one page in.
+ next, err := RunSweep(context.Background(), s, l, sc, "bab", "titogo", 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if next.Current == nil || next.Current.Objects != 2 {
+ t.Fatalf("the next tick did not start a fresh partial sweep: %+v", next.Current)
+ }
+ if next.LastComplete == nil || next.LastComplete.Objects != 5 {
+ t.Fatalf("page-one partials overwrote the audit answer: %+v", next.LastComplete)
+ }
+}
+
+// Interrupted mid-bucket and re-run, a sweep reaches the same rollup without
+// double-counting what it already saw.
+func TestSweepResumesWithoutDoubleCounting(t *testing.T) {
+ s, p, sc, _ := sweepFixture(t, 5)
+ l := &memLister{p: p, pageSize: 2}
+
+ side := sweepToCompletion(t, s, l, sc)
+ if l.calls < 3 {
+ t.Fatalf("the fixture did not actually paginate: %d listing calls", l.calls)
+ }
+ if side.LastComplete.Objects != 5 || side.LastComplete.Scanned != 5 {
+ t.Fatalf("resumed sweep counted objects=%d scanned=%d, want 5/5",
+ side.LastComplete.Objects, side.LastComplete.Scanned)
+ }
+}
+
+// A fresh sweep over a bucket whose app is torn down between the listing and
+// the write must leave NOTHING behind: an orphan scan.json under a purged
+// prefix makes the app name permanently unclaimable, because the claim gate
+// refuses any name whose prefix still holds a key and box-side auto-purge
+// never reaches it.
+//
+// The first write of a fresh sweep is create-only by construction, so the CAS
+// retry guard cannot catch this on its own — only the pre-write re-read can.
+func TestSweepDoesNotResurrectTornDownApp(t *testing.T) {
+ ctx := context.Background()
+ s, p, sc, _ := sweepFixture(t, 3)
+
+ l := &memLister{p: p}
+ l.before = func() {
+ // Teardown lands while this sweep is listing.
+ if err := s.Delete(ctx, Key("bab", "titogo")); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ side, err := RunSweep(ctx, s, l, sc, "bab", "titogo", time.Minute)
+ if err != nil {
+ t.Fatalf("standing down is not an error: %v", err)
+ }
+ _ = side
+ if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the sweep wrote an orphan sidecar under a purged app: %v", err)
+ }
+}
+
+// A client-encrypted store is marked not_applicable and does no work. A
+// scanner cannot read ciphertext, and reporting every object clean would be a
+// true-looking sentence about bytes nobody examined.
+func TestSweepSkipsEncryptedStore(t *testing.T) {
+ ctx := context.Background()
+ s, _, sc, _ := sweepFixture(t, 3)
+
+ rec, etag, err := Get(ctx, s, "bab", "titogo")
+ if err != nil {
+ t.Fatal(err)
+ }
+ rec.Scanning.Requested = &ScanRequested{At: time.Now().UTC(), By: "m@example.com", Encrypted: true}
+ if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
+ t.Fatal(err)
+ }
+
+ side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if side == nil || !side.NotApplicable {
+ t.Fatalf("an encrypted store was not marked not_applicable: %+v", side)
+ }
+ if side.LastComplete != nil {
+ t.Fatal("coverage was claimed for a store nobody could read")
+ }
+}
+
+// A trashed app is not swept: it is destroyed within 30 days and scanning it
+// bills for egress on data on its way out.
+func TestSweepSkipsTrashedApp(t *testing.T) {
+ ctx := context.Background()
+ s, _, sc, _ := sweepFixture(t, 3)
+ if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+
+ side, err := RunSweep(ctx, s, &refuseLister{t: t}, sc, "bab", "titogo", time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if side != nil {
+ t.Fatalf("a trashed app was swept: %+v", side)
+ }
+}
+
+// Skipped is not clean, and both are visible separately. An object over the
+// size ceiling was never judged; folding it into Scanned would make the audit
+// sentence false in exactly the direction that matters.
+func TestSweepCountsSkippedSeparately(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.Objects != 3 || got.Scanned != 2 || got.Skipped != 1 {
+ t.Fatalf("objects=%d scanned=%d skipped=%d, want 3/2/1", got.Objects, got.Scanned, got.Skipped)
+ }
+ if got.Unscanned != 1 {
+ t.Fatalf("unscanned=%d, want 1 — a skipped object is not covered", got.Unscanned)
+ }
+}
+
+// An infected object WAS scanned. We know exactly what it is; it is coverage,
+// not a gap. This is the one place the two counts could plausibly have been
+// wired the other way round.
+func TestSweepCountsInfectedAsScanned(t *testing.T) {
+ s, p, sc, _ := sweepFixture(t, 2)
+ sc.ScanAs(objectName(0), VerdictInfected)
+
+ 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.Infected != 1 || got.Scanned != 2 || got.Unscanned != 0 {
+ t.Fatalf("infected=%d scanned=%d unscanned=%d, want 1/2/0",
+ got.Infected, got.Scanned, got.Unscanned)
+ }
+}
+
+// A record naming another engine refuses rather than guessing. Verdicts in the
+// bucket were written under conventions this engine does not read, and
+// recounting them as unscanned would destroy the coverage history.
+func TestSweepRefusesOnEngineDrift(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.Enabled.Engine = "guardduty"
+ if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
+ t.Fatal(err)
+ }
+
+ _, err = RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute)
+ if err == nil {
+ t.Fatal("a sweep ran against a record naming a different engine")
+ }
+ if !strings.Contains(err.Error(), "guardduty") {
+ t.Fatalf("the refusal does not name the engine on the record: %v", err)
+ }
+}
+
+// An engine that goes away mid-page fails the pass and writes NOTHING. Every
+// verdict already reached is a tag in the bucket, so the re-run reads them
+// back without re-scanning a byte — but a half-written rollup would be a
+// number nobody could interpret.
+func TestSweepEngineFailureWritesNoRollup(t *testing.T) {
+ ctx := context.Background()
+ s, p, sc, _ := sweepFixture(t, 3)
+ sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused"))
+
+ if _, err := RunSweep(ctx, s, &memLister{p: p}, sc, "bab", "titogo", time.Minute); err == nil {
+ t.Fatal("an unreachable engine reported a successful sweep")
+ }
+ if _, _, err := s.Get(ctx, ScanKey("bab", "titogo")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a failed sweep left a rollup behind: %v", err)
+ }
+}
+
+// A store with no scanning enabled is not swept at all — which is every store
+// on a deployment that has not opted in.
+func TestSweepIgnoresStoreWithoutScanning(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ _, pub := testSealKeypair(t)
+ p := NewMemProvisioner("eu-west-1")
+ rotateFixture(t, s, pub, p)
+
+ side, err := RunSweep(context.Background(), s, &refuseLister{t: t}, NewMemScanner(),
+ "bab", "titogo", time.Minute)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if side != nil {
+ t.Fatalf("a store that never opted in was swept: %+v", side)
+ }
+}