amadan

carlos / platform Public

objstore: the scan queue, its index and its drain

One object per item, not one queue object: a shared object would put
every enqueue in a CAS race with every other and losing would be the
normal case, which is the problem the sidecar avoided by having a single
writer. Distinct keys never collide.

Under the app's prefix, so PurgeApp removes them and the claim gate sees
them. A top-level queue would be found by one listing instead of an
index — and would survive app deletion invisibly, because PurgeApp lists
only the app prefix. The index is the price of keeping items where
deletion can find them.

Release needs BOTH an unchanged generation and an empty listing, and the
second condition is the blocker that forced the plan's rewrite. The
drain caps work per tick, so on a flooded app it deliberately leaves
items behind with no enqueue to bump anything; releasing on generation
alone deleted the entry pointing at exactly those items, and they would
sit until the next boot with the callback and long-poll wake lost — for
the app the cap exists for. Mutation-checked, both halves.

Deleting versus keeping is the other pairing that has to be right.
Missing attestation, attested encrypted, engine not enabled, app in the
trash: permanent, so the items go — leaving them in the trash case would
block the app's name for thirty days. Deployment disabled, engine drift,
engine unreachable, read error: reversible, so items and index entry
both stay.

The drain re-issues failed and denied itself. Without it an object the
sweep last tagged failed would be answered from that tag forever and
never rescanned on request, making on-demand strictly less able than the
hourly pass it exists to improve on.

The id alphabet is base32-HEX, and the test caught why. Standard base32
puts digits at values above the letters while ASCII sorts them below,
so encoded ids would not sort in timestamp order and a prefix listing
would stop being the queue in order.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev ada1d2c0434e546d1ea64a102420d4c1a46734f3 parent 229a58d
2 files changed, +894 −0
  • internal/objstore/queue.go +439 −0
  • internal/objstore/queue_test.go +455 −0
diff --git a/internal/objstore/queue.go b/internal/objstore/queue.go
new file mode 100644
index 000000000..cfbdc10bd
--- /dev/null
+++ b/internal/objstore/queue.go
@@ -0,0 +1,439 @@
+// The scan queue: apps/<acct>/<app>/scan-queue/<id>.json, one object per item.
+//
+// ONE OBJECT PER ITEM, not one queue object. A shared queue object would put
+// every enqueue in a compare-and-swap race with every other, and losing that
+// race would be the normal case rather than the exception — the same problem
+// the sidecar design avoided by having a single writer. Distinct keys never
+// collide, so an enqueue needs no coordination at all.
+//
+// UNDER THE APP'S PREFIX, so PurgeApp removes the items with everything else
+// and the name-claim gate can see them. A top-level scan-queue/<acct>/<app>/
+// would be found by one listing instead of an index — but PurgeApp lists only
+// the app prefix, and the claim gate never looks outside it, so items would
+// survive app deletion invisibly. The index below is the price of keeping them
+// where deletion can find them.
+//
+// The walk lives here and the goroutine lives in console, mirroring how
+// RunSweep and scanPass already divide: objstore knows what to do to a bucket,
+// console knows when.
+package objstore
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/base32"
+ "encoding/binary"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "os"
+ "sort"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// QueueSchema is pinned; an unknown schema refuses.
+const QueueSchema = 1
+
+// QueueTTL is how long an item may sit before it is dropped rather than
+// scanned. One sweep cycle: after that the ordinary sweep has had its chance
+// to reach the object, and a stale on-demand answer is worth less than none.
+const QueueTTL = 24 * time.Hour
+
+// QueueItem is one app's request that a named object be scanned soon.
+type QueueItem struct {
+ Schema int `json:"schema"`
+ Key string `json:"key"`
+ At time.Time `json:"at"`
+ By string `json:"by"`
+ // CallbackPath is the app's own path to POST the result to, validated at
+ // the door. Never a host: the platform knows where the app lives, and a
+ // caller-supplied host would be an SSRF against the console.
+ CallbackPath string `json:"callback_path,omitempty"`
+}
+
+// AppRef names one app, for the index.
+type AppRef struct{ Account, App string }
+
+// queueIDEncoding is lowercase base32-HEX without padding.
+//
+// The hex alphabet, not the standard one, and the difference is the whole
+// point: base32's usual alphabet puts the digits 2-7 at values 26-31, ABOVE
+// the letters, while in ASCII digits sort BELOW them. Encoded ids would then
+// not sort in the same order as the timestamps they encode, and a prefix
+// listing would stop being the queue in order. base32-hex is monotonic —
+// 0-9 then a-v, in both value and byte order.
+var queueIDEncoding = base32.NewEncoding("0123456789abcdefghijklmnopqrstuv").WithPadding(base32.NoPadding)
+
+// NewQueueID is a time-ordered id: milliseconds big-endian, then randomness.
+//
+// Hand-rolled because there is no ULID in this repo and adding a dependency is
+// a decision, not a default. The properties that matter are that ids sort by
+// time — so a prefix listing IS the queue in order — and that two enqueues in
+// the same millisecond do not collide.
+func NewQueueID(now time.Time) string {
+ var b [16]byte
+ binary.BigEndian.PutUint64(b[:8], uint64(now.UnixMilli()))
+ if _, err := rand.Read(b[8:]); err != nil {
+ // crypto/rand does not fail in practice; if it ever did, a
+ // time-only id still sorts correctly and merely risks a collision
+ // that the create-only write would catch.
+ binary.BigEndian.PutUint64(b[8:], uint64(now.UnixNano()))
+ }
+ return queueIDEncoding.EncodeToString(b[:])
+}
+
+// QueuePrefix is where one app's items live.
+func QueuePrefix(acct, app string) string {
+ return release.AppPrefix(acct, app) + "scan-queue/"
+}
+
+// QueueKey is one item's object key.
+func QueueKey(acct, app, id string) string { return QueuePrefix(acct, app) + id + ".json" }
+
+// ---- the index --------------------------------------------------------------
+
+// QueueIndex remembers which apps have work, so the drain lists only those
+// prefixes instead of walking every app on the deployment every tick.
+//
+// SOFT STATE, deliberately. Losing it costs latency, never items: the boot
+// rebuild finds anything stranded, and the ordinary sweep still reaches every
+// object on its own cycle. What it must never do is claim an app is empty when
+// it is not, which is what the generation counter below is for.
+type QueueIndex struct {
+ mu sync.Mutex
+ gen map[AppRef]uint64
+}
+
+func NewQueueIndex() *QueueIndex { return &QueueIndex{gen: map[AppRef]uint64{}} }
+
+// Bump marks an app as having work. Called AFTER the item object is written:
+// an index entry pointing at nothing is harmless, an item nothing points at is
+// stranded until the next boot.
+func (i *QueueIndex) Bump(ref AppRef) {
+ i.mu.Lock()
+ defer i.mu.Unlock()
+ if i.gen == nil {
+ i.gen = map[AppRef]uint64{}
+ }
+ i.gen[ref]++
+}
+
+// Observe reads the generation a drain is about to work against.
+func (i *QueueIndex) Observe(ref AppRef) uint64 {
+ i.mu.Lock()
+ defer i.mu.Unlock()
+ return i.gen[ref]
+}
+
+// Release drops an app's entry, and reports whether it did.
+//
+// TWO CONDITIONS, and the second is the one an earlier draft of this design
+// got wrong. The generation must be unchanged — no enqueue landed while we
+// worked — AND the listing must have come back empty. Generation alone is not
+// enough: the drain caps how many items it handles per tick, so on a flooded
+// app it deliberately leaves items behind with no enqueue to bump anything,
+// and releasing there would delete the entry pointing at them.
+func (i *QueueIndex) Release(ref AppRef, observed uint64, listedEmpty bool) bool {
+ if !listedEmpty {
+ return false
+ }
+ i.mu.Lock()
+ defer i.mu.Unlock()
+ if i.gen[ref] != observed {
+ return false
+ }
+ delete(i.gen, ref)
+ return true
+}
+
+// Apps lists the apps with work, for the drain's round-robin.
+func (i *QueueIndex) Apps() []AppRef {
+ i.mu.Lock()
+ defer i.mu.Unlock()
+ out := make([]AppRef, 0, len(i.gen))
+ for ref := range i.gen {
+ out = append(out, ref)
+ }
+ sort.Slice(out, func(a, b int) bool {
+ if out[a].Account != out[b].Account {
+ return out[a].Account < out[b].Account
+ }
+ return out[a].App < out[b].App
+ })
+ return out
+}
+
+// Len is how many apps have work — the deployment-wide queue depth surface.
+func (i *QueueIndex) Len() int {
+ i.mu.Lock()
+ defer i.mu.Unlock()
+ return len(i.gen)
+}
+
+// ---- enqueue ----------------------------------------------------------------
+
+// Enqueue writes one item and then marks the app.
+//
+// ORDER MATTERS: the object first, the index second. A crash between the two
+// strands that item until the next boot rebuild, which is recoverable. The
+// other order would mark an app whose item was never written, and a caller
+// answered 202 in between would stop retrying an item nothing points at.
+func Enqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem) (string, error) {
+ item.Schema = QueueSchema
+ body, err := json.MarshalIndent(item, "", " ")
+ if err != nil {
+ return "", err
+ }
+ id := NewQueueID(item.At)
+ if _, err := s.Put(ctx, QueueKey(acct, app, id), append(body, '\n'), ""); err != nil {
+ return "", fmt.Errorf("objstore: enqueue %s/%s: %w", acct, app, err)
+ }
+ idx.Bump(AppRef{Account: acct, App: app})
+ return id, nil
+}
+
+// ---- draining ---------------------------------------------------------------
+
+// Drained is one item's outcome, everything console needs and nothing it does
+// not: the long-poll wake, the callback body, and the per-tick meter fold.
+type Drained struct {
+ Key, Version, ETag string
+ Verdict Verdict
+ // Recorded is whether the verdict reached the object itself. When false
+ // the caller has written a result object instead, and the app must be told
+ // so — an infected object with Recorded false is NOT covered by the
+ // enforcement deny, because the deny matches on a tag that is not there.
+ Recorded bool
+ // Scanned is whether the engine actually ran, as against the verdict being
+ // read back from an existing tag. The meter counts on-demand work; an
+ // answer the sweep already paid for is not the tenant's to be charged.
+ Scanned bool
+ Size int64
+ Engine string
+ Signature string
+ CallbackPath string
+}
+
+// ErrQueueNotPermitted is a queue that must not be drained: the record is
+// gone, the app is in the trash, or the consent chain no longer holds. The
+// caller deletes the items rather than retrying them.
+var ErrQueueNotPermitted = errors.New("objstore: this store may not be scanned")
+
+// DrainApp scans up to cap of one app's queued items.
+//
+// It owns Observe and Release. Console cannot: Observe must happen BEFORE the
+// listing, and that ordering is exactly what the generation counter protects —
+// a caller that could get it wrong would be a caller that eventually does.
+func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex,
+ acct, app string, cap int, now time.Time) (done []Drained, listedEmpty bool, err error) {
+
+ ref := AppRef{Account: acct, App: app}
+ observed := idx.Observe(ref)
+
+ keys, err := listQueue(ctx, s, acct, app, cap)
+ if err != nil {
+ return nil, false, err
+ }
+ listedEmpty = len(keys) == 0
+ defer func() {
+ if err == nil {
+ idx.Release(ref, observed, listedEmpty)
+ }
+ }()
+ if listedEmpty {
+ return nil, true, nil
+ }
+
+ // The consent chain, re-checked here and not only at the door: the record
+ // can change between enqueue and drain, and THIS is where bytes are read.
+ rec, permitted, err := queuePermitted(ctx, s, acct, app)
+ if err != nil {
+ return nil, false, err
+ }
+ if !permitted {
+ // Not a retry — none of these can change back in a way that makes the
+ // items useful. Drop them and say so.
+ for _, k := range keys {
+ if derr := s.Delete(ctx, k); derr != nil && !errors.Is(derr, os.ErrNotExist) {
+ return nil, false, derr
+ }
+ }
+ return nil, false, ErrQueueNotPermitted
+ }
+
+ engine, signature, err := sc.Provenance(ctx)
+ if err != nil {
+ return nil, false, fmt.Errorf("%w (%s/%s): %v", ErrEngineUnreachable, acct, app, err)
+ }
+ if rec.Scanning.Enabled.Engine != engine {
+ return nil, false, fmt.Errorf("%w (%s/%s): record says %q, configured engine is %q",
+ ErrEngineDrift, acct, app, rec.Scanning.Enabled.Engine, engine)
+ }
+
+ bucket := rec.Provisioned.Bucket
+ seen := map[string]bool{}
+ for _, qk := range keys {
+ item, ierr := readQueueItem(ctx, s, qk)
+ if ierr != nil {
+ // An item we cannot read is not one we can act on, and leaving it
+ // would block the app's entry forever.
+ if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) {
+ return done, false, derr
+ }
+ continue
+ }
+ // Expired, or a key already handled this tick. Both are deletions
+ // rather than work: re-scanning the same key twice in one pass buys
+ // nothing, and the dedupe is what stops one app's flood being
+ // amplified into repeated fetches.
+ if now.Sub(item.At) > QueueTTL || seen[item.Key] {
+ if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) {
+ return done, false, derr
+ }
+ continue
+ }
+ seen[item.Key] = true
+
+ out, derr := drainOne(ctx, s, sc, acct, app, bucket, engine, signature, item)
+ if errors.Is(derr, ErrObjectGone) {
+ if delErr := s.Delete(ctx, qk); delErr != nil && !errors.Is(delErr, os.ErrNotExist) {
+ return done, false, delErr
+ }
+ continue
+ }
+ if derr != nil {
+ // An engine failure is about the ENGINE, not this item. Keep the
+ // item and the index entry, and let the next tick retry.
+ return done, false, derr
+ }
+ if delErr := s.Delete(ctx, qk); delErr != nil && !errors.Is(delErr, os.ErrNotExist) {
+ return done, false, delErr
+ }
+ done = append(done, out)
+ }
+ // A page that was fully consumed may still have more behind it.
+ listedEmpty = false
+ return done, listedEmpty, nil
+}
+
+// drainOne scans a single item and records the outcome wherever it can go.
+func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucket, engine, signature string,
+ item *QueueItem) (Drained, error) {
+
+ verdict, obj, err := sc.Status(ctx, bucket, item.Key)
+ if err != nil {
+ return Drained{}, err
+ }
+ out := Drained{
+ Key: item.Key, Version: obj.Version(), ETag: obj.ETag, Size: obj.Size,
+ Engine: engine, Signature: signature, CallbackPath: item.CallbackPath,
+ }
+
+ // The drain applies the re-issue rule ITSELF: denied and failed are
+ // transient, and answering an on-demand request from such a tag forever
+ // would make this path strictly less able than the hourly sweep it exists
+ // to improve on. A zero ScanObject means HeadObject was denied, so there
+ // is no version or size to scan with.
+ needsScan := verdict == VerdictUnknown || verdict == VerdictFailed || verdict == VerdictDenied
+ if !needsScan || obj == (ScanObject{}) {
+ out.Verdict, out.Recorded = verdict, true
+ return out, nil
+ }
+
+ v, recorded, err := sc.Scan(ctx, bucket, item.Key, obj.Version(), obj.Size)
+ if err != nil {
+ return Drained{}, err
+ }
+ out.Verdict, out.Recorded, out.Scanned = v, recorded, true
+
+ if recorded {
+ // A real verdict is on the object now, so any fallback must go.
+ if err := DeleteScanResult(ctx, s, acct, app, item.Key); err != nil {
+ return Drained{}, err
+ }
+ return out, nil
+ }
+
+ // The verdict could not be written to the object. Re-read the record
+ // first: the writeSidecar race applies, and a result object stranded under
+ // a purged prefix is found by nothing and resurrects the app.
+ if _, permitted, err := queuePermitted(ctx, s, acct, app); err != nil || !permitted {
+ return out, err
+ }
+ if err := WriteScanResult(ctx, s, acct, app, &ScanResult{
+ Key: item.Key, Verdict: string(v), Version: obj.Version(),
+ Engine: engine, Signature: signature, At: time.Now().UTC(),
+ }); err != nil {
+ return Drained{}, err
+ }
+ return out, nil
+}
+
+// queuePermitted is the consent chain, in one place so the door and the drain
+// cannot answer it differently.
+func queuePermitted(ctx context.Context, s release.Store, acct, app string) (*Record, bool, error) {
+ rec, _, err := Get(ctx, s, acct, app)
+ if err != nil {
+ return nil, false, err
+ }
+ if rec == nil || rec.Provisioned == nil {
+ return nil, false, nil
+ }
+ tomb, err := release.ReadAppTombstone(ctx, s, acct, app)
+ if err != nil {
+ return nil, false, err
+ }
+ if tomb != nil {
+ return nil, false, nil
+ }
+ if rec.Scanning == nil || rec.Scanning.Requested == nil ||
+ rec.Scanning.Requested.Encrypted || rec.Scanning.Enabled == nil {
+ return nil, false, nil
+ }
+ on, err := release.AntivirusEnabled(ctx, s)
+ if err != nil {
+ return nil, false, err
+ }
+ if !on {
+ // Reversible: the caller KEEPS the items. Distinguished from the
+ // refusals above by the record still being intact.
+ return rec, false, ErrScanningDisabled
+ }
+ return rec, true, nil
+}
+
+func listQueue(ctx context.Context, s release.Store, acct, app string, cap int) ([]string, error) {
+ prefix := QueuePrefix(acct, app)
+ if pl, ok := s.(release.PageLister); ok {
+ return pl.ListPage(ctx, prefix, "", cap)
+ }
+ // A store that cannot page still works, just less cheaply — the same
+ // degradation storagePass accepts for SizeLister.
+ keys, err := s.List(ctx, prefix)
+ if err != nil {
+ return nil, err
+ }
+ sort.Strings(keys)
+ if cap > 0 && len(keys) > cap {
+ keys = keys[:cap]
+ }
+ return keys, nil
+}
+
+func readQueueItem(ctx context.Context, s release.Store, key string) (*QueueItem, error) {
+ raw, _, err := s.Get(ctx, key)
+ if err != nil {
+ return nil, err
+ }
+ var item QueueItem
+ if err := json.Unmarshal(raw, &item); err != nil {
+ return nil, err
+ }
+ if item.Schema != QueueSchema {
+ return nil, fmt.Errorf("objstore: queue item %s: schema %d is not %d", key, item.Schema, QueueSchema)
+ }
+ return &item, nil
+}
diff --git a/internal/objstore/queue_test.go b/internal/objstore/queue_test.go
new file mode 100644
index 000000000..877dc3c0f
--- /dev/null
+++ b/internal/objstore/queue_test.go
@@ -0,0 +1,455 @@
+package objstore
+
+import (
+ "context"
+ "errors"
+ "os"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// queueFixture is a provisioned, scanning-enabled store with an armed engine
+// and n objects the scanner knows about.
+func queueFixture(t *testing.T, n int) (release.Store, *MemScanner, *QueueIndex, 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")
+
+ sc := NewMemScanner()
+ bucket := rec.Provisioned.Bucket
+ for i := 0; i < n; i++ {
+ sc.PutObject(bucket, objectName(i), "etag-"+objectName(i), time.Now().UTC(), 10)
+ }
+ return s, sc, NewQueueIndex(), bucket
+}
+
+func enqueueN(t *testing.T, s release.Store, idx *QueueIndex, n int) {
+ t.Helper()
+ for i := 0; i < n; i++ {
+ if _, err := Enqueue(context.Background(), s, idx, "bab", "titogo", &QueueItem{
+ Key: objectName(i), At: time.Now().UTC(), By: "app",
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// Ids sort by time, so a prefix listing IS the queue in order.
+func TestQueueIDSortsByTime(t *testing.T) {
+ base := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC)
+ var prev string
+ for i := 0; i < 200; i++ {
+ id := NewQueueID(base.Add(time.Duration(i) * time.Millisecond))
+ if id <= prev {
+ t.Fatalf("id %q does not sort after %q", id, prev)
+ }
+ prev = id
+ }
+ // Two in the same millisecond must not collide.
+ seen := map[string]bool{}
+ for i := 0; i < 1000; i++ {
+ id := NewQueueID(base)
+ if seen[id] {
+ t.Fatal("two ids in the same millisecond collided")
+ }
+ seen[id] = true
+ }
+}
+
+// The object first, the index second. A crash between the two strands one item
+// until the boot rebuild; the other order would mark an app whose item was
+// never written, and a 202 sent in between would stop the client retrying.
+func TestEnqueueWritesObjectBeforeBump(t *testing.T) {
+ ctx := context.Background()
+ s, _, idx, _ := queueFixture(t, 1)
+
+ id, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
+ Key: "uploads/a", At: time.Now().UTC(), By: "app",
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := s.Get(ctx, QueueKey("bab", "titogo", id)); err != nil {
+ t.Fatalf("item object was not written: %v", err)
+ }
+ if idx.Len() != 1 {
+ t.Fatal("the app was not marked")
+ }
+}
+
+// THE test this design needed. The drain caps work per tick, so on a flooded
+// app it deliberately leaves items behind with no enqueue to bump anything —
+// and an earlier draft released the entry whenever the generation was
+// unchanged, deleting the pointer to exactly those items. They would then sit
+// until the next console boot, and the callback and long-poll wake would be
+// lost for the app the cap exists for.
+func TestIndexEntrySurvivesPartialDrain(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 5)
+ enqueueN(t, s, idx, 5)
+
+ done, listedEmpty, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 2 {
+ t.Fatalf("drained %d, want the cap of 2", len(done))
+ }
+ if listedEmpty {
+ t.Fatal("a capped listing reported itself empty")
+ }
+ if idx.Len() != 1 {
+ t.Fatal("the entry pointing at the remaining items was released")
+ }
+
+ // Successive calls finish the job.
+ for i := 0; i < 5 && idx.Len() > 0; i++ {
+ if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 2, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if idx.Len() != 0 {
+ t.Fatal("successive drains never emptied the queue")
+ }
+ left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if err != nil || len(left) != 0 {
+ t.Fatalf("items left behind: %v, %v", left, err)
+ }
+}
+
+// An enqueue landing mid-drain must not have its entry released underneath it.
+func TestIndexEntrySurvivesEnqueueDuringDrain(t *testing.T) {
+ ref := AppRef{Account: "bab", App: "titogo"}
+ idx := NewQueueIndex()
+ idx.Bump(ref)
+
+ observed := idx.Observe(ref)
+ idx.Bump(ref) // an enqueue lands while the drain works
+
+ if idx.Release(ref, observed, true) {
+ t.Fatal("released an entry an enqueue had bumped")
+ }
+ if idx.Len() != 1 {
+ t.Fatal("the entry is gone")
+ }
+}
+
+// Release needs BOTH conditions. Generation alone is the bug above.
+func TestIndexReleasedOnlyOnEmptyListing(t *testing.T) {
+ ref := AppRef{Account: "bab", App: "titogo"}
+ idx := NewQueueIndex()
+ idx.Bump(ref)
+ observed := idx.Observe(ref)
+
+ if idx.Release(ref, observed, false) {
+ t.Fatal("released on a non-empty listing")
+ }
+ if !idx.Release(ref, observed, true) {
+ t.Fatal("did not release on an empty listing with an unchanged generation")
+ }
+}
+
+// The consent chain, re-checked at the drain because the record can change
+// between enqueue and drain and THIS is where bytes are read.
+func TestDrainAppRefusesAndDeletesWhenNotPermitted(t *testing.T) {
+ ctx := context.Background()
+
+ for name, break_ := range map[string]func(t *testing.T, s release.Store){
+ "no attestation": func(t *testing.T, s release.Store) {
+ rec, etag, _ := Get(ctx, s, "bab", "titogo")
+ rec.Scanning.Requested = nil
+ if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
+ t.Fatal(err)
+ }
+ },
+ "attested encrypted": func(t *testing.T, s release.Store) {
+ rec, etag, _ := Get(ctx, s, "bab", "titogo")
+ rec.Scanning.Requested.Encrypted = true
+ if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
+ t.Fatal(err)
+ }
+ },
+ "engine not enabled for the store": func(t *testing.T, s release.Store) {
+ rec, etag, _ := Get(ctx, s, "bab", "titogo")
+ rec.Scanning.Enabled = nil
+ if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil {
+ t.Fatal(err)
+ }
+ },
+ "app in the trash": func(t *testing.T, s release.Store) {
+ if err := release.TrashApp(ctx, s, "bab", "titogo", "op@example.com", time.Now().UTC()); err != nil {
+ t.Fatal(err)
+ }
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ s, sc, idx, _ := queueFixture(t, 2)
+ enqueueN(t, s, idx, 2)
+ break_(t, s)
+
+ _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if !errors.Is(err, ErrQueueNotPermitted) {
+ t.Fatalf("err = %v, want ErrQueueNotPermitted", err)
+ }
+ // The items are DELETED, not kept: none of these can change back
+ // in a way that makes them useful, and leaving them would block
+ // the app's name for thirty days in the trash case.
+ left, lerr := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if lerr != nil || len(left) != 0 {
+ t.Fatalf("items survived a permanent refusal: %v, %v", left, lerr)
+ }
+ })
+ }
+}
+
+// Disabled deployment-wide is REVERSIBLE, so the items are kept. The refusals
+// above are not, so they are deleted. Getting these two the same way round
+// would either strand a tenant's queue or lose it.
+func TestDrainAppKeepsItemsWhenDeploymentDisabled(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 2)
+ enqueueN(t, s, idx, 2)
+
+ av, etag, err := release.GetAntivirus(ctx, s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ av.Enabled = false
+ if _, err := release.PutAntivirus(ctx, s, av, etag); err != nil {
+ t.Fatal(err)
+ }
+
+ if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrScanningDisabled) {
+ t.Fatalf("err = %v, want ErrScanningDisabled", err)
+ }
+ left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if err != nil || len(left) != 2 {
+ t.Fatalf("a reversible refusal dropped the queue: %v, %v", left, err)
+ }
+ if idx.Len() != 1 {
+ t.Fatal("the index entry was released on a reversible refusal")
+ }
+}
+
+// The drain re-issues failed and denied itself. Without this, an object the
+// sweep last tagged failed would be answered from that tag forever and never
+// rescanned on request — on-demand strictly less able than the hourly pass it
+// exists to improve on.
+func TestDrainAppReIssuesFailedAndDenied(t *testing.T) {
+ ctx := context.Background()
+ for _, stuck := range []Verdict{VerdictFailed, VerdictDenied} {
+ t.Run(string(stuck), func(t *testing.T) {
+ s, sc, idx, bucket := queueFixture(t, 1)
+ key := objectName(0)
+ // A stale verdict already on the object, at its current version.
+ _, obj, err := sc.Status(ctx, bucket, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sc.Put(bucket, key, obj.Version(), stuck)
+ enqueueN(t, s, idx, 1)
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 1 {
+ t.Fatalf("drained %d items", len(done))
+ }
+ if !done[0].Scanned {
+ t.Fatalf("a %q verdict was answered from the tag rather than rescanned", stuck)
+ }
+ if done[0].Verdict != VerdictClean {
+ t.Fatalf("verdict = %q, want the rescan's answer", done[0].Verdict)
+ }
+ })
+ }
+}
+
+// A verdict already recorded at the current version costs a tag read and no
+// fetch. That is the amplification guard: re-requesting a judged key must not
+// re-stream its bytes.
+func TestDrainAppReusesAJudgedVerdict(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, bucket := queueFixture(t, 1)
+ key := objectName(0)
+ _, obj, err := sc.Status(ctx, bucket, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ sc.Put(bucket, key, obj.Version(), VerdictClean)
+ enqueueN(t, s, idx, 1)
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 1 || done[0].Scanned {
+ t.Fatalf("a judged key was rescanned: %+v", done)
+ }
+ if done[0].Verdict != VerdictClean {
+ t.Fatalf("verdict = %q", done[0].Verdict)
+ }
+}
+
+// A verdict the object cannot carry goes to a result object instead, and the
+// outcome says so — an infected object with Recorded false is NOT covered by
+// the enforcement deny, because the deny matches a tag that is not there.
+func TestDrainAppWritesResultWhenUnrecorded(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, bucket := queueFixture(t, 1)
+ key := objectName(0)
+ sc.Unrecordable(key)
+ enqueueN(t, s, idx, 1)
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 1 || done[0].Recorded {
+ t.Fatalf("outcome = %+v, want recorded false", done)
+ }
+ _, obj, err := sc.Status(ctx, bucket, key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ res, err := ReadScanResult(ctx, s, "bab", "titogo", key, obj.Version())
+ if err != nil || res == nil {
+ t.Fatalf("no result object for an unrecorded verdict: %+v, %v", res, err)
+ }
+ if res.Verdict != string(VerdictClean) {
+ t.Fatalf("result verdict = %q", res.Verdict)
+ }
+}
+
+// And when a later scan DOES persist a verdict, the fallback goes — it must
+// not outlive the real answer.
+func TestDrainAppDeletesResultWhenLaterRecorded(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, bucket := queueFixture(t, 1)
+ key := objectName(0)
+
+ sc.Unrecordable(key)
+ enqueueN(t, s, idx, 1)
+ if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); err != nil {
+ t.Fatalf("setup: no result object: %v", err)
+ }
+
+ // The tag limit clears, and the object is re-requested.
+ sc2 := NewMemScanner()
+ _, obj, _ := sc.Status(ctx, bucket, key)
+ sc2.PutObject(bucket, key, obj.ETag, obj.Modified, obj.Size)
+ if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
+ Key: key, At: time.Now().UTC(), By: "app",
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := DrainApp(ctx, s, sc2, idx, "bab", "titogo", 10, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+ if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", key)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the fallback outlived the real verdict: %v", err)
+ }
+}
+
+// Expired items are dropped rather than scanned: after a sweep cycle the
+// ordinary pass has had its chance, and a stale on-demand answer is worth less
+// than none.
+func TestDrainAppDropsExpiredItems(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 1)
+ if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
+ Key: objectName(0), At: time.Now().Add(-48 * time.Hour).UTC(), By: "app",
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 0 {
+ t.Fatalf("an expired item was scanned: %+v", done)
+ }
+ left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if err != nil || len(left) != 0 {
+ t.Fatalf("expired item not collected: %v, %v", left, err)
+ }
+}
+
+// The same key twice in one tick is scanned once. This is what stops one app's
+// flood being amplified into repeated fetches of the same bytes.
+func TestDrainAppDedupesByKey(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 1)
+ for i := 0; i < 3; i++ {
+ if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
+ Key: objectName(0), At: time.Now().UTC(), By: "app",
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(done) != 1 {
+ t.Fatalf("drained %d outcomes for one key", len(done))
+ }
+ left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if err != nil || len(left) != 0 {
+ t.Fatalf("duplicate items left behind: %v, %v", left, err)
+ }
+}
+
+// An object deleted between enqueue and drain is dropped, not an error.
+func TestDrainAppDropsGoneObjects(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 0) // the scanner knows about no objects
+ if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{
+ Key: "uploads/vanished", At: time.Now().UTC(), By: "app",
+ }); err != nil {
+ t.Fatal(err)
+ }
+
+ done, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now())
+ if err != nil {
+ t.Fatalf("a vanished object failed the drain: %v", err)
+ }
+ if len(done) != 0 {
+ t.Fatalf("outcomes for an object that is not there: %+v", done)
+ }
+ left, _ := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if len(left) != 0 {
+ t.Fatal("the item for a vanished object was kept")
+ }
+}
+
+// An engine that goes away keeps the items AND the entry: it is a fact about
+// the engine, not about these objects, and the next tick should retry.
+func TestDrainAppKeepsItemsOnEngineFailure(t *testing.T) {
+ ctx := context.Background()
+ s, sc, idx, _ := queueFixture(t, 1)
+ enqueueN(t, s, idx, 1)
+ sc.Fail("Scan", errors.New("dial unix /run/clamav/clamd.ctl: connection refused"))
+
+ if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err == nil {
+ t.Fatal("an unreachable engine reported a successful drain")
+ }
+ left, err := s.List(ctx, QueuePrefix("bab", "titogo"))
+ if err != nil || len(left) != 1 {
+ t.Fatalf("an engine failure dropped the item: %v, %v", left, err)
+ }
+ if idx.Len() != 1 {
+ t.Fatal("an engine failure released the index entry")
+ }
+}