objstore: what two reviews of the queue found
Two independent reviews of tasks 1-10 landed on the same four defects in the drain, and both named the same root cause for two of them: a rule stated once in the spec and implemented in only one of the two places it applies. DELETION IS DRIVEN BY ABSENCE, NEVER BY ERROR (§7.5). queuePermitted honoured that for the record read; the item read did not. One transient GET failure on scan-queue/<id>.json deleted an accepted request outright — no scan, no callback, and a status door answering `pending` until some sweep happened to tag the object. Store errors now keep the item and the index entry; only absence and bytes no build will ever parse are deletions. DEDUPE THE WORK, NOT THE PROMISES. Two enqueues for one key with different callback paths were answered with two 202s and delivered one callback: the duplicate item was deleted before its path was read. The drain now groups a page by object key, reads the object once, and carries every distinct path into the outcome. Drained.CallbackPath becomes CallbackPaths. A DENIED HeadObject IS NOT A RECORDED VERDICT. With no version and no size there is nothing to scan and nothing was written; reporting recorded:true would have put "denied" in a signed callback beside a claim the object carries it, when the console never saw a tag and the enforcement deny matches one that is not there. ENQUEUE OWNS ITS CLOCK. The id is derived from At and the TTL is measured from it, so a door that forgot to stamp it would mint an id sorting at year one and an item the next drain discards as expired — a 202 that meant nothing. It is a parameter now, not a field a caller fills in. Also: a nil Scanner refuses in words rather than dereferencing (RunSweep's posture); a permanent refusal deletes the app's scan-results with its queue items, because a result nobody may answer from still blocks the name-claim gate and resurrects the app; and the two collectors task 9 specified but never got — DeleteScanResults and CollectScanResults, the latter degrading on a store with no ModLister the way storagePass degrades without SizeLister. The tests both reviews called vacuous are rewritten to fail under the mutation each was supposed to catch, and each was checked by making it: denied head says recorded -> TestDrainAppDeniedHeadIsNotRecorded delete on any read error -> TestDrainAppKeepsItemOnTransientReadError no purge re-read guard -> TestResultObjectNotWrittenUnderPurgedPrefix Observe after the listing -> TestIndexEntrySurvivesEnqueueDuringDrain bump before the write -> TestEnqueueWritesObjectBeforeBump dedupe drops later callbacks -> TestDrainAppDedupesWorkNotCallbacks results survive a refusal -> TestDrainAppDeletesResultsWhenNotPermitted TestIndexEntrySurvivesEnqueueDuringDrain drove Observe/Bump/Release by hand, so it proved the methods worked while leaving the ordering they exist to protect free to be wrong; it goes through DrainApp now. The result key is pinned to an exact string rather than compared with itself. Status is pinned against the re-issue rule on MemScanner, which is the only implementation the queue tests ever see. And the stale-tag test gains the case the version binding actually exists for: same ETag, later timestamp — single-part ETag is the content MD5 and chosen-prefix collisions are practical.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YJkJkxZZLgW8X8A5tk69Sq
7 files changed,
+888
−72
internal/objstore/clamav_test.go+19 −0internal/objstore/memscanner.go+44 −0internal/objstore/queue.go+156 −45internal/objstore/queue_test.go+466 −24internal/objstore/scanner_contract_test.go+57 −0internal/objstore/scanresult.go+67 −0internal/objstore/scanresult_test.go+79 −3
diff --git a/internal/objstore/clamav_test.go b/internal/objstore/clamav_test.go| index ac9c123df..c72978c59 100644 |
| --- a/internal/objstore/clamav_test.go |
| +++ b/internal/objstore/clamav_test.go |
| @@ -695,6 +695,25 @@ func TestStatusIgnoresStaleTag(t *testing.T) { |
| if v != VerdictUnknown { |
| t.Fatalf("a rewritten object reported %q from its predecessor's tag", v) |
| } |
| + |
| + // THE case the version binding exists for, and the one the check above |
| + // does not make: the ETag is IDENTICAL and only the timestamp moved. |
| + // Single-part ETag is the content MD5, and chosen-prefix MD5 collisions |
| + // are practical — so an attacker who can get one file scanned can upload a |
| + // second with the same ETag and inherit its clean verdict. Comparing only |
| + // the half before the "@" passes the first assertion and fails this one. |
| + s3.headOK("same-etag", mod.Add(time.Hour), 10) |
| + s3.getTagging = func(*awss3.GetObjectTaggingInput) (*awss3.GetObjectTaggingOutput, error) { |
| + return taggingOutput(TagVerdict, "clean", |
| + TagVersion, ScanObject{ETag: "same-etag", Modified: mod}.Version()), nil |
| + } |
| + v, _, err = c.Status(context.Background(), "b", "k") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictUnknown { |
| + t.Fatalf("an object rewritten with a colliding ETag reported %q — the timestamp half of the version is not being compared", v) |
| + } |
| } |
| // A denied HeadObject leaves no version and no size, so the caller must not go |
diff --git a/internal/objstore/memscanner.go b/internal/objstore/memscanner.go| index c70221f5c..6f2281cc3 100644 |
| --- a/internal/objstore/memscanner.go |
| +++ b/internal/objstore/memscanner.go |
| @@ -22,6 +22,14 @@ type MemScanner struct { |
| // cap, in production. A hook rather than a stub, because the case is the |
| // reason Scan reports `recorded` at all. |
| unrecordable map[string]bool |
| + // deniedHead are keys whose HeadObject is refused, so Status answers |
| + // denied with a ZERO ScanObject — the one shape that leaves a caller with |
| + // no version and no size. |
| + deniedHead map[string]bool |
| + // scans counts Scan calls per key, so a test can assert that deduping |
| + // really stopped the engine reading the same object twice rather than |
| + // merely folding two outcomes into one. |
| + scans map[string]int |
| } |
| // scanRecord is one object's verdict AND the object version it was judged |
| @@ -85,6 +93,13 @@ func (s *MemScanner) Status(ctx context.Context, bucket, key string) (Verdict, S |
| if err := s.takeFailure("Status"); err != nil { |
| return VerdictUnknown, ScanObject{}, err |
| } |
| + if s.deniedHead[memScanKey(bucket, key)] { |
| + // HeadObject itself refused: no version, no size, nothing to scan |
| + // with. The production shape is a bucket policy the console is not |
| + // covered by, as against the SSE-KMS case where the head succeeds and |
| + // the GET is what fails. |
| + return VerdictDenied, ScanObject{}, nil |
| + } |
| obj, ok := s.objects[memScanKey(bucket, key)] |
| if !ok { |
| return VerdictUnknown, ScanObject{}, ErrObjectGone |
| @@ -166,6 +181,10 @@ func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size |
| if err := s.takeFailure("Scan"); err != nil { |
| return VerdictUnknown, false, err |
| } |
| + if s.scans == nil { |
| + s.scans = map[string]int{} |
| + } |
| + s.scans[key]++ |
| v := VerdictClean |
| if got, ok := s.scanAs[key]; ok { |
| v = got |
| @@ -179,6 +198,23 @@ func (s *MemScanner) Scan(ctx context.Context, bucket, key, version string, size |
| return v, true, nil |
| } |
| +// DenyHead makes Status refuse the head of one key. |
| +func (s *MemScanner) DenyHead(bucket, key string) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + if s.deniedHead == nil { |
| + s.deniedHead = map[string]bool{} |
| + } |
| + s.deniedHead[memScanKey(bucket, key)] = true |
| +} |
| + |
| +// Scans is how many times the engine was asked to read one key. |
| +func (s *MemScanner) Scans(key string) int { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + return s.scans[key] |
| +} |
| + |
| // 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. |
| @@ -202,6 +238,14 @@ func (s *MemScanner) Capacity(ctx context.Context) (int, int, error) { return -1 |
| // and test fixtures write Engine: "clamav" — a fake that named itself would |
| // make every one of those fixtures refuse instead of sweep. |
| func (s *MemScanner) Provenance(ctx context.Context) (string, string, error) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + // Armable like the others: an engine that will not say what it is, is a |
| + // state the drain and the sweep both have to survive, and a fake that |
| + // could never be in it left that path untested. |
| + if err := s.takeFailure("Provenance"); err != nil { |
| + return "", "", err |
| + } |
| return "clamav", "MemScanner/0/never", nil |
| } |
diff --git a/internal/objstore/queue.go b/internal/objstore/queue.go| index cfbdc10bd..e696b5b4e 100644 |
| --- a/internal/objstore/queue.go |
| +++ b/internal/objstore/queue.go |
| @@ -26,6 +26,7 @@ import ( |
| "encoding/json" |
| "errors" |
| "fmt" |
| + "log" |
| "os" |
| "sort" |
| "sync" |
| @@ -181,8 +182,15 @@ func (i *QueueIndex) Len() int { |
| // 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) { |
| +// |
| +// ENQUEUE OWNS THE TIMESTAMP, which is why now is a parameter and At is not a |
| +// field the caller fills in. The id is derived from At and the TTL is measured |
| +// from it, so a door that forgot to stamp it would mint an id sorting at year |
| +// one and an item the very next drain discards as expired — a 202 that means |
| +// nothing. One writer, one clock. |
| +func Enqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, item *QueueItem, now time.Time) (string, error) { |
| item.Schema = QueueSchema |
| + item.At = now.UTC() |
| body, err := json.MarshalIndent(item, "", " ") |
| if err != nil { |
| return "", err |
| @@ -210,11 +218,17 @@ type Drained struct { |
| // 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 |
| + Scanned bool |
| + Size int64 |
| + Engine string |
| + Signature string |
| + // CallbackPaths is EVERY path that asked about this key in this page, not |
| + // one. The drain dedupes by object key so the engine reads an object once, |
| + // and an earlier draft deleted the duplicate items along with their |
| + // callbacks: two requests for the same key with different paths returned |
| + // two 202s and delivered one callback. Deduplicate the WORK, never the |
| + // promises made about it. |
| + CallbackPaths []string |
| } |
| // ErrQueueNotPermitted is a queue that must not be drained: the record is |
| @@ -222,6 +236,9 @@ type Drained struct { |
| // caller deletes the items rather than retrying them. |
| var ErrQueueNotPermitted = errors.New("objstore: this store may not be scanned") |
| +// ErrNoEngine is a drain asked to run with no scanner behind it. |
| +var ErrNoEngine = errors.New("objstore: no scanning engine") |
| + |
| // DrainApp scans up to cap of one app's queued items. |
| // |
| // It owns Observe and Release. Console cannot: Observe must happen BEFORE the |
| @@ -230,6 +247,13 @@ var ErrQueueNotPermitted = errors.New("objstore: this store may not be scanned") |
| 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) { |
| + // In words rather than as a nil dereference, the way RunSweep refuses |
| + // (sweep.go). The console gates its goroutine on a wired engine, but a |
| + // package that dereferences whatever it is handed is one call site away |
| + // from taking the control plane down. |
| + if sc == nil { |
| + return nil, false, ErrNoEngine |
| + } |
| ref := AppRef{Account: acct, App: app} |
| observed := idx.Observe(ref) |
| @@ -261,6 +285,14 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| return nil, false, derr |
| } |
| } |
| + // And the fallback results with them. A result object is a verdict no |
| + // door may answer from once the app may not be scanned, and it is the |
| + // worse orphan of the two: a stranded queue item is drained next tick, |
| + // while a stranded result is found by nothing, blocks the app-name |
| + // claim gate, and resurrects a deleted app on the next boot. |
| + if derr := DeleteScanResults(ctx, s, acct, app); derr != nil { |
| + return nil, false, derr |
| + } |
| return nil, false, ErrQueueNotPermitted |
| } |
| @@ -274,32 +306,14 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| } |
| 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) |
| + groups, err := groupQueue(ctx, s, keys, now) |
| + if err != nil { |
| + return nil, false, err |
| + } |
| + for _, g := range groups { |
| + out, derr := drainOne(ctx, s, sc, acct, app, bucket, engine, signature, g) |
| if errors.Is(derr, ErrObjectGone) { |
| - if delErr := s.Delete(ctx, qk); delErr != nil && !errors.Is(delErr, os.ErrNotExist) { |
| + if delErr := deleteQueueItems(ctx, s, g.queueKeys); delErr != nil { |
| return done, false, delErr |
| } |
| continue |
| @@ -309,7 +323,7 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| // 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) { |
| + if delErr := deleteQueueItems(ctx, s, g.queueKeys); delErr != nil { |
| return done, false, delErr |
| } |
| done = append(done, out) |
| @@ -319,31 +333,121 @@ func DrainApp(ctx context.Context, s release.Store, sc Scanner, idx *QueueIndex, |
| return done, listedEmpty, nil |
| } |
| -// drainOne scans a single item and records the outcome wherever it can go. |
| +// scanGroup is every queued request in this page that asks about ONE object |
| +// key: the object is read once, and every request that asked gets its answer. |
| +type scanGroup struct { |
| + key string |
| + // queueKeys is every item object standing for this key, all deleted |
| + // together once the group has an outcome. |
| + queueKeys []string |
| + // callbackPaths is one entry per DISTINCT path that asked. Empty when |
| + // every request was long-poll only. |
| + callbackPaths []string |
| +} |
| + |
| +// groupQueue reads a listed page into scan groups, deleting what can never be |
| +// acted on and KEEPING what a later tick might still read. |
| +// |
| +// The distinction is §7.5's, and getting it wrong loses a tenant's request for |
| +// good: deletion is driven by ABSENCE and by content that no tick will ever |
| +// parse, never by a store error. An earlier draft deleted on any read failure, |
| +// so one transient 500 on a GET discarded an accepted request silently. |
| +func groupQueue(ctx context.Context, s release.Store, keys []string, now time.Time) ([]*scanGroup, error) { |
| + var groups []*scanGroup |
| + byKey := map[string]*scanGroup{} |
| + for _, qk := range keys { |
| + item, err := readQueueItem(ctx, s, qk) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + continue // already gone; nothing to delete and nothing to do |
| + case errors.Is(err, errQueueItemUnreadable): |
| + // Unparseable, or a schema this build does not know. No tick will |
| + // ever read it, and leaving it would hold the app's index entry |
| + // open forever. |
| + if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) { |
| + return nil, derr |
| + } |
| + continue |
| + case err != nil: |
| + // A STORE error. Reversible: keep the item, keep the entry, and |
| + // let the next tick have it. The rest of the page still drains — |
| + // one bad round trip is not a reason to stall an app's queue. |
| + log.Printf("objstore: scan queue: read %s: %v", qk, err) |
| + continue |
| + } |
| + if now.Sub(item.At) > QueueTTL { |
| + if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) { |
| + return nil, derr |
| + } |
| + continue |
| + } |
| + g := byKey[item.Key] |
| + if g == nil { |
| + g = &scanGroup{key: item.Key} |
| + byKey[item.Key] = g |
| + groups = append(groups, g) |
| + } |
| + g.queueKeys = append(g.queueKeys, qk) |
| + if item.CallbackPath != "" && !containsString(g.callbackPaths, item.CallbackPath) { |
| + g.callbackPaths = append(g.callbackPaths, item.CallbackPath) |
| + } |
| + } |
| + return groups, nil |
| +} |
| + |
| +func containsString(list []string, want string) bool { |
| + for _, s := range list { |
| + if s == want { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func deleteQueueItems(ctx context.Context, s release.Store, keys []string) error { |
| + for _, k := range keys { |
| + if err := s.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// drainOne scans a single object 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) { |
| + g *scanGroup) (Drained, error) { |
| - verdict, obj, err := sc.Status(ctx, bucket, item.Key) |
| + verdict, obj, err := sc.Status(ctx, bucket, g.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, |
| + Key: g.key, Version: obj.Version(), ETag: obj.ETag, Size: obj.Size, |
| + Engine: engine, Signature: signature, CallbackPaths: g.callbackPaths, |
| } |
| // 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. |
| + // to improve on. |
| needsScan := verdict == VerdictUnknown || verdict == VerdictFailed || verdict == VerdictDenied |
| - if !needsScan || obj == (ScanObject{}) { |
| + if !needsScan { |
| + // The object carries this verdict, at this version, right now. |
| out.Verdict, out.Recorded = verdict, true |
| return out, nil |
| } |
| + if obj == (ScanObject{}) { |
| + // HeadObject itself was denied, so there is no version and no size to |
| + // scan with — and NOTHING WAS RECORDED. Saying recorded here would put |
| + // "denied" in a callback beside a claim that the object carries it, |
| + // when the console never even saw a tag; the enforcement deny matches |
| + // on a tag that is not there, and the app would be told a durable |
| + // answer exists that no door can produce. |
| + out.Verdict, out.Recorded = verdict, false |
| + return out, nil |
| + } |
| - v, recorded, err := sc.Scan(ctx, bucket, item.Key, obj.Version(), obj.Size) |
| + v, recorded, err := sc.Scan(ctx, bucket, g.key, obj.Version(), obj.Size) |
| if err != nil { |
| return Drained{}, err |
| } |
| @@ -351,7 +455,7 @@ func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucke |
| 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 { |
| + if err := DeleteScanResult(ctx, s, acct, app, g.key); err != nil { |
| return Drained{}, err |
| } |
| return out, nil |
| @@ -364,7 +468,7 @@ func drainOne(ctx context.Context, s release.Store, sc Scanner, acct, app, bucke |
| return out, err |
| } |
| if err := WriteScanResult(ctx, s, acct, app, &ScanResult{ |
| - Key: item.Key, Verdict: string(v), Version: obj.Version(), |
| + Key: g.key, Verdict: string(v), Version: obj.Version(), |
| Engine: engine, Signature: signature, At: time.Now().UTC(), |
| }); err != nil { |
| return Drained{}, err |
| @@ -423,6 +527,13 @@ func listQueue(ctx context.Context, s release.Store, acct, app string, cap int) |
| return keys, nil |
| } |
| +// errQueueItemUnreadable marks an item whose BYTES are the problem — bad JSON, |
| +// or a schema this build does not know. It is the one read failure that is |
| +// permanent, and the drain deletes on it; every other read failure is the |
| +// store having a bad moment and the item stays. Wrapped rather than compared |
| +// so the caller can tell the two apart with errors.Is and nothing else. |
| +var errQueueItemUnreadable = errors.New("objstore: unreadable queue item") |
| + |
| func readQueueItem(ctx context.Context, s release.Store, key string) (*QueueItem, error) { |
| raw, _, err := s.Get(ctx, key) |
| if err != nil { |
| @@ -430,10 +541,10 @@ func readQueueItem(ctx context.Context, s release.Store, key string) (*QueueItem |
| } |
| var item QueueItem |
| if err := json.Unmarshal(raw, &item); err != nil { |
| - return nil, err |
| + return nil, fmt.Errorf("%w %s: %v", errQueueItemUnreadable, key, err) |
| } |
| if item.Schema != QueueSchema { |
| - return nil, fmt.Errorf("objstore: queue item %s: schema %d is not %d", key, item.Schema, QueueSchema) |
| + return nil, fmt.Errorf("%w %s: schema %d is not %d", errQueueItemUnreadable, key, item.Schema, QueueSchema) |
| } |
| return &item, nil |
| } |
diff --git a/internal/objstore/queue_test.go b/internal/objstore/queue_test.go| index 877dc3c0f..1f1a8e429 100644 |
| --- a/internal/objstore/queue_test.go |
| +++ b/internal/objstore/queue_test.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "errors" |
| "os" |
| + "strings" |
| "testing" |
| "time" |
| @@ -32,8 +33,8 @@ 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 { |
| + Key: objectName(i), By: "app", |
| + }, time.Now()); err != nil { |
| t.Fatal(err) |
| } |
| } |
| @@ -64,22 +65,45 @@ func TestQueueIDSortsByTime(t *testing.T) { |
| // 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. |
| +// It is the ORDER that is under test, so the assertions have to be able to see |
| +// it. Asserting only on the end state passes just as happily with the two |
| +// swapped, which is the mutation that matters. |
| 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", |
| - }) |
| + // (a) At the instant the object is being written, the app is not yet |
| + // marked. Bump-first fails here. |
| + var markedDuringWrite int |
| + watching := &queueStore{Store: s, onPut: func(string) { markedDuringWrite = idx.Len() }} |
| + id, err := Enqueue(ctx, watching, idx, "bab", "titogo", &QueueItem{ |
| + Key: "uploads/a", By: "app", |
| + }, time.Now()) |
| if err != nil { |
| t.Fatal(err) |
| } |
| + if markedDuringWrite != 0 { |
| + t.Fatal("the index was bumped before the item object was written") |
| + } |
| 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") |
| } |
| + |
| + // (b) And a write that FAILS leaves the app unmarked, so nothing ever |
| + // points at an item that does not exist. |
| + fresh := NewQueueIndex() |
| + failing := &queueStore{Store: s, putErr: errors.New("the bucket said no")} |
| + if _, err := Enqueue(ctx, failing, fresh, "bab", "titogo", &QueueItem{ |
| + Key: "uploads/b", By: "app", |
| + }, time.Now()); err == nil { |
| + t.Fatal("a failed write reported success") |
| + } |
| + if fresh.Len() != 0 { |
| + t.Fatal("the app was marked for an item that was never written") |
| + } |
| } |
| // THE test this design needed. The drain caps work per tick, so on a flooded |
| @@ -123,19 +147,45 @@ func TestIndexEntrySurvivesPartialDrain(t *testing.T) { |
| } |
| // An enqueue landing mid-drain must not have its entry released underneath it. |
| +// |
| +// Driven through DRAINAPP, not through Observe and Release by hand: what the |
| +// generation counter protects is an ORDERING inside DrainApp — Observe before |
| +// the listing — and a test that calls the three methods itself proves the |
| +// methods work while leaving the ordering free to be wrong. Moving Observe |
| +// below the listing passes such a test and fails this one. |
| func TestIndexEntrySurvivesEnqueueDuringDrain(t *testing.T) { |
| - ref := AppRef{Account: "bab", App: "titogo"} |
| - idx := NewQueueIndex() |
| - idx.Bump(ref) |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 2) |
| + enqueueN(t, s, idx, 1) |
| - observed := idx.Observe(ref) |
| - idx.Bump(ref) // an enqueue lands while the drain works |
| + // Drain the one item. The entry stays, because a consumed page may have |
| + // more behind it. |
| + if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if idx.Len() != 1 { |
| + t.Fatal("a consumed page released the entry before an empty listing confirmed it") |
| + } |
| - if idx.Release(ref, observed, true) { |
| - t.Fatal("released an entry an enqueue had bumped") |
| + // Now the queue really is empty — except that an enqueue lands after the |
| + // listing has been taken and before the release decision. |
| + racing := &queueStore{Store: s} |
| + racing.afterList = func() { |
| + if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| + Key: objectName(1), By: "app", |
| + }, time.Now()); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + if _, listedEmpty, err := DrainApp(ctx, racing, sc, idx, "bab", "titogo", 10, time.Now()); err != nil || !listedEmpty { |
| + t.Fatalf("drain = %v, %v", listedEmpty, err) |
| } |
| if idx.Len() != 1 { |
| - t.Fatal("the entry is gone") |
| + t.Fatal("the entry for an item enqueued during the drain was released — nothing points at it until boot") |
| + } |
| + left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if err != nil || len(left) != 1 { |
| + t.Fatalf("the late item = %v, %v", left, err) |
| } |
| } |
| @@ -348,8 +398,8 @@ func TestDrainAppDeletesResultWhenLaterRecorded(t *testing.T) { |
| _, 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 { |
| + Key: key, By: "app", |
| + }, time.Now()); err != nil { |
| t.Fatal(err) |
| } |
| if _, _, err := DrainApp(ctx, s, sc2, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| @@ -367,8 +417,8 @@ 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 { |
| + Key: objectName(0), By: "app", |
| + }, time.Now().Add(-48*time.Hour)); err != nil { |
| t.Fatal(err) |
| } |
| @@ -387,13 +437,20 @@ func TestDrainAppDropsExpiredItems(t *testing.T) { |
| // 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) { |
| +// |
| +// BUT DEDUPING THE WORK IS NOT DEDUPING THE PROMISES. Every enqueue was |
| +// answered 202, and every one that named a callback path is owed a delivery. |
| +// An earlier draft deleted the duplicate items along with their paths, so two |
| +// requests for one key with different callbacks got one callback and the other |
| +// app endpoint waited forever — a silent loss the drain's own test blessed by |
| +// asserting only that one outcome came back. |
| +func TestDrainAppDedupesWorkNotCallbacks(t *testing.T) { |
| ctx := context.Background() |
| s, sc, idx, _ := queueFixture(t, 1) |
| - for i := 0; i < 3; i++ { |
| + for _, path := range []string{"/hooks/a", "/hooks/b", "/hooks/a", ""} { |
| if _, err := Enqueue(ctx, s, idx, "bab", "titogo", &QueueItem{ |
| - Key: objectName(0), At: time.Now().UTC(), By: "app", |
| - }); err != nil { |
| + Key: objectName(0), By: "app", CallbackPath: path, |
| + }, time.Now()); err != nil { |
| t.Fatal(err) |
| } |
| } |
| @@ -403,7 +460,16 @@ func TestDrainAppDedupesByKey(t *testing.T) { |
| t.Fatal(err) |
| } |
| if len(done) != 1 { |
| - t.Fatalf("drained %d outcomes for one key", len(done)) |
| + t.Fatalf("drained %d outcomes for one key — the object was read more than once", len(done)) |
| + } |
| + if sc.Scans(objectName(0)) != 1 { |
| + t.Fatalf("the engine read the object %d times for four requests", sc.Scans(objectName(0))) |
| + } |
| + // Both distinct paths survive; the repeat does not become a second |
| + // delivery, and the request with no path adds nothing. |
| + got := done[0].CallbackPaths |
| + if len(got) != 2 || !containsString(got, "/hooks/a") || !containsString(got, "/hooks/b") { |
| + t.Fatalf("callback paths = %v, want both distinct paths and no duplicate", got) |
| } |
| left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| if err != nil || len(left) != 0 { |
| @@ -416,8 +482,8 @@ 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 { |
| + Key: "uploads/vanished", By: "app", |
| + }, time.Now()); err != nil { |
| t.Fatal(err) |
| } |
| @@ -453,3 +519,379 @@ func TestDrainAppKeepsItemsOnEngineFailure(t *testing.T) { |
| t.Fatal("an engine failure released the index entry") |
| } |
| } |
| + |
| +// ---- the store wrapper the ordering and error tests need ------------------- |
| + |
| +// queueStore wraps a store so one operation can be failed, watched or raced. |
| +// |
| +// The optional listings are forwarded BY HAND. An embedded interface promotes |
| +// only release.Store's own methods, so a wrapper that did not forward would |
| +// quietly take the drain off PageLister and the collector off ModLister — and |
| +// every test using it would then exercise the fallback path instead of the one |
| +// it names. |
| +type queueStore struct { |
| + release.Store |
| + putErr error // every Put fails with this |
| + onPut func(key string) // runs BEFORE the write lands |
| + failGetPrefix string // reads of keys under this prefix fail |
| + failGetKey string // reads of exactly this key fail |
| + afterList func() // runs after the listing, before the caller sees it |
| +} |
| + |
| +func (q *queueStore) Put(ctx context.Context, key string, body []byte, etag string) (string, error) { |
| + if q.onPut != nil { |
| + q.onPut(key) |
| + } |
| + if q.putErr != nil { |
| + return "", q.putErr |
| + } |
| + return q.Store.Put(ctx, key, body, etag) |
| +} |
| + |
| +func (q *queueStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == q.failGetKey || (q.failGetPrefix != "" && strings.HasPrefix(key, q.failGetPrefix)) { |
| + return nil, "", errors.New("objstore test: the bucket had a bad moment") |
| + } |
| + return q.Store.Get(ctx, key) |
| +} |
| + |
| +func (q *queueStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| + keys, err := q.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit) |
| + if q.afterList != nil { |
| + f := q.afterList |
| + q.afterList = nil // once |
| + f() |
| + } |
| + return keys, err |
| +} |
| + |
| +func (q *queueStore) ListMod(ctx context.Context, prefix string) ([]release.KeyMod, error) { |
| + return q.Store.(release.ModLister).ListMod(ctx, prefix) |
| +} |
| + |
| +// Enqueue owns the timestamp, so a door cannot forget to stamp it. A zero At |
| +// would mint an id sorting at year one and an item the very next drain |
| +// discards as expired — a 202 that meant nothing. |
| +func TestEnqueueStampsItsOwnTimestamp(t *testing.T) { |
| + ctx := context.Background() |
| + s, _, idx, _ := queueFixture(t, 1) |
| + now := time.Date(2026, 9, 2, 9, 0, 0, 0, time.UTC) |
| + |
| + item := &QueueItem{Key: "uploads/a", By: "app"} // no At |
| + id, err := Enqueue(ctx, s, idx, "bab", "titogo", item, now) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if !item.At.Equal(now) { |
| + t.Fatalf("At = %v, want the clock the caller passed", item.At) |
| + } |
| + if want := NewQueueID(now); id[:12] != want[:12] { |
| + t.Fatalf("id %q does not carry the enqueue time (want the prefix of %q)", id, want) |
| + } |
| + stored, err := readQueueItem(ctx, s, QueueKey("bab", "titogo", id)) |
| + if err != nil || stored.At.IsZero() { |
| + t.Fatalf("stored item = %+v, %v", stored, err) |
| + } |
| +} |
| + |
| +// A HeadObject the console may not make is not a recorded verdict. |
| +// |
| +// Saying recorded here would put "denied" in a callback beside a claim that |
| +// the object carries it, when no tag was ever read or written — and the |
| +// enforcement deny matches on a tag that is not there, so the app would be |
| +// told a durable answer exists that no door can produce. |
| +func TestDrainAppDeniedHeadIsNotRecorded(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, bucket := queueFixture(t, 1) |
| + sc.DenyHead(bucket, objectName(0)) |
| + 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 outcomes", len(done)) |
| + } |
| + if done[0].Verdict != VerdictDenied { |
| + t.Fatalf("verdict = %q, want denied", done[0].Verdict) |
| + } |
| + if done[0].Recorded { |
| + t.Fatal("a denied HeadObject was reported as a recorded verdict") |
| + } |
| + if done[0].Scanned { |
| + t.Fatal("the engine was asked to read an object with no version") |
| + } |
| +} |
| + |
| +// §7.5: deletion is driven by ABSENCE, never by error. A bucket that fails one |
| +// GET must not cost a tenant an accepted request. |
| +func TestDrainAppKeepsItemOnTransientReadError(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + |
| + flaky := &queueStore{Store: s, failGetPrefix: QueuePrefix("bab", "titogo")} |
| + done, listedEmpty, err := DrainApp(ctx, flaky, sc, idx, "bab", "titogo", 10, time.Now()) |
| + if err != nil { |
| + t.Fatalf("one unreadable item failed the whole drain: %v", err) |
| + } |
| + if len(done) != 0 { |
| + t.Fatalf("outcomes for an item that was never read: %+v", done) |
| + } |
| + if listedEmpty { |
| + t.Fatal("a page with an item in it reported itself empty") |
| + } |
| + left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if err != nil || len(left) != 1 { |
| + t.Fatalf("a transient read error discarded the request: %v, %v", left, err) |
| + } |
| + if idx.Len() != 1 { |
| + t.Fatal("the entry pointing at the kept item was released") |
| + } |
| + |
| + // And it drains on the next tick, once the bucket is well again. |
| + done, _, err = DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| + if err != nil || len(done) != 1 { |
| + t.Fatalf("retry = %+v, %v", done, err) |
| + } |
| +} |
| + |
| +// The record itself failing to read is reversible too, and it is the SIBLING |
| +// of the check above: queuePermitted honours "absence, not error" for the |
| +// record, and an earlier draft honoured it there and nowhere else. |
| +func TestDrainAppKeepsItemsOnRecordReadError(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + |
| + broken := &queueStore{Store: s, failGetKey: Key("bab", "titogo")} |
| + if _, _, err := DrainApp(ctx, broken, sc, idx, "bab", "titogo", 10, time.Now()); err == nil { |
| + t.Fatal("an unreadable record reported a successful drain") |
| + } |
| + left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if err != nil || len(left) != 1 { |
| + t.Fatalf("an unreadable record discarded the queue: %v, %v", left, err) |
| + } |
| + if idx.Len() != 1 { |
| + t.Fatal("an unreadable record released the index entry") |
| + } |
| +} |
| + |
| +// Engine drift keeps the items too: the record naming a different engine is a |
| +// thing an operator reverses, not a fact about these objects. |
| +func TestDrainAppKeepsItemsOnEngineDrift(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| + rec.Scanning.Enabled.Engine = "someone-elses-scanner" |
| + if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| + if !errors.Is(err, ErrEngineDrift) { |
| + t.Fatalf("err = %v, want ErrEngineDrift", err) |
| + } |
| + left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if lerr != nil || len(left) != 1 { |
| + t.Fatalf("drift discarded the queue: %v, %v", left, lerr) |
| + } |
| + if idx.Len() != 1 { |
| + t.Fatal("drift released the index entry") |
| + } |
| +} |
| + |
| +// An engine that will not say what it is keeps them as well — the same |
| +// reasoning as an engine that will not scan. |
| +func TestDrainAppKeepsItemsOnProvenanceFailure(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + sc.Fail("Provenance", errors.New("dial: connection refused")) |
| + |
| + _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()) |
| + if !errors.Is(err, ErrEngineUnreachable) { |
| + t.Fatalf("err = %v, want ErrEngineUnreachable", err) |
| + } |
| + left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if lerr != nil || len(left) != 1 { |
| + t.Fatalf("an unreachable engine discarded the queue: %v, %v", left, lerr) |
| + } |
| +} |
| + |
| +// The one read failure that IS permanent: bytes no build will ever parse. |
| +// Leaving it would hold the app's index entry open forever. |
| +func TestDrainAppDeletesUnreadableItem(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + if _, err := s.Put(ctx, QueueKey("bab", "titogo", "0000000000000000000000000v"), |
| + []byte("{not json"), ""); 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("the readable item did not drain alongside the broken one: %+v", done) |
| + } |
| + left, err := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if err != nil || len(left) != 0 { |
| + t.Fatalf("the unparseable item was kept: %v, %v", left, err) |
| + } |
| +} |
| + |
| +// A result object is a verdict a door may answer from. When the app may no |
| +// longer be scanned at all, that verdict must go with the queue items — and it |
| +// is the worse orphan of the two: an item is drained next tick, a result is |
| +// found by nothing, blocks the app-name claim gate, and resurrects the app. |
| +func TestDrainAppDeletesResultsWhenNotPermitted(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", &ScanResult{ |
| + Key: objectName(0), Verdict: "clean", Version: "v1", At: time.Now().UTC(), |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + 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) |
| + } |
| + |
| + if _, _, err := DrainApp(ctx, s, sc, idx, "bab", "titogo", 10, time.Now()); !errors.Is(err, ErrQueueNotPermitted) { |
| + t.Fatalf("drain = %v, want ErrQueueNotPermitted", err) |
| + } |
| + left, err := s.List(ctx, ScanResultsPrefix("bab", "titogo")) |
| + if err != nil || len(left) != 0 { |
| + t.Fatalf("a result outlived the consent that allowed it: %v, %v", left, err) |
| + } |
| +} |
| + |
| +// A store that was never granted has no bucket to scan. The check has to be |
| +// the FIRST one, because everything after it dereferences Provisioned. |
| +func TestDrainAppRefusesWhenNotGranted(t *testing.T) { |
| + ctx := context.Background() |
| + |
| + for name, break_ := range map[string]func(t *testing.T, s release.Store){ |
| + "record absent": func(t *testing.T, s release.Store) { |
| + if err := s.Delete(ctx, Key("bab", "titogo")); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + "declared but never granted": func(t *testing.T, s release.Store) { |
| + rec, etag, _ := Get(ctx, s, "bab", "titogo") |
| + rec.Provisioned = nil |
| + if _, err := Put(ctx, s, "bab", "titogo", rec, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + }, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + 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) |
| + } |
| + 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) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// A drain with no engine refuses in words rather than panicking on a nil |
| +// interface, the way RunSweep does. The goroutine gates on this too, but a |
| +// package that dereferences whatever it is handed is one call site away from |
| +// taking the console down. |
| +func TestDrainAppRefusesWithoutAnEngine(t *testing.T) { |
| + ctx := context.Background() |
| + s, _, idx, _ := queueFixture(t, 1) |
| + enqueueN(t, s, idx, 1) |
| + |
| + _, _, err := DrainApp(ctx, s, nil, idx, "bab", "titogo", 10, time.Now()) |
| + if !errors.Is(err, ErrNoEngine) { |
| + t.Fatalf("err = %v, want ErrNoEngine", err) |
| + } |
| + left, lerr := s.List(ctx, QueuePrefix("bab", "titogo")) |
| + if lerr != nil || len(left) != 1 { |
| + t.Fatalf("a missing engine discarded the queue: %v, %v", left, lerr) |
| + } |
| +} |
| + |
| +// The purge race, at the drain. The re-read before writing a result is the |
| +// only thing between an unrecordable verdict and an orphan under a purged |
| +// prefix — and that orphan does not merely leak: the claim gate refuses the |
| +// app's name for good, and materialize counts any key under the prefix and |
| +// fails toward keeping the row, so the deleted app comes BACK on the next boot. |
| +func TestResultObjectNotWrittenUnderPurgedPrefix(t *testing.T) { |
| + ctx := context.Background() |
| + s, sc, idx, _ := queueFixture(t, 1) |
| + sc.Unrecordable(objectName(0)) // the ten-tag cap, in production |
| + enqueueN(t, s, idx, 1) |
| + |
| + // The teardown lands between the drain's own consent check (the first read |
| + // of store.json) and drainOne's re-read before writing (the second). The |
| + // re-read is the guard; without it the create-only write SUCCEEDS under a |
| + // prefix nothing will ever list again. |
| + staged := &recordRaceStore{Store: s, key: Key("bab", "titogo"), before: 2, race: func() { |
| + // What PurgeApp leaves behind: the prefix, emptied. Done directly |
| + // because PurgeApp itself refuses over a provisioned record, and the |
| + // teardown that clears it is not what this test is about. |
| + keys, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| + if err != nil { |
| + t.Error(err) |
| + return |
| + } |
| + for _, k := range keys { |
| + if err := s.Delete(ctx, k); err != nil { |
| + t.Error(err) |
| + } |
| + } |
| + }} |
| + |
| + if _, _, err := DrainApp(ctx, staged, sc, idx, "bab", "titogo", 10, time.Now()); err != nil { |
| + t.Fatalf("drain: %v", err) |
| + } |
| + left, err := s.List(ctx, release.AppPrefix("bab", "titogo")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(left) != 0 { |
| + t.Fatalf("the purged app's prefix still holds %v — its name can never be claimed again", left) |
| + } |
| +} |
| + |
| +// recordRaceStore runs race() just before the Nth read of one key, so a test |
| +// can put a competing writer at a chosen point in a walk. Same shape as |
| +// walk_test.go's raceOnReadStore, which fires on the first read only. |
| +type recordRaceStore struct { |
| + release.Store |
| + key string |
| + before int // fire immediately before this read, 1-based |
| + race func() |
| + seen int |
| +} |
| + |
| +func (r *recordRaceStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == r.key { |
| + r.seen++ |
| + if r.seen == r.before { |
| + r.race() |
| + } |
| + } |
| + return r.Store.Get(ctx, key) |
| +} |
| + |
| +func (r *recordRaceStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) { |
| + return r.Store.(release.PageLister).ListPage(ctx, prefix, startAfter, limit) |
| +} |
diff --git a/internal/objstore/scanner_contract_test.go b/internal/objstore/scanner_contract_test.go| index 02049e5e0..5554e404f 100644 |
| --- a/internal/objstore/scanner_contract_test.go |
| +++ b/internal/objstore/scanner_contract_test.go |
| @@ -2,7 +2,9 @@ package objstore |
| import ( |
| "context" |
| + "errors" |
| "testing" |
| + "time" |
| ) |
| // scannerContract is run against every Scanner implementation. Idempotence is |
| @@ -53,6 +55,61 @@ func TestMemScannerContract(t *testing.T) { |
| scannerContract(t, NewMemScanner(), "carlos-store-tito-bab-titogo") |
| } |
| +// Status and Verdict answer DIFFERENT QUESTIONS, and the fake has to hold that |
| +// line as firmly as the real engine does — otherwise the drain's own re-issue |
| +// test passes against a fake that already re-issued for it, and the split is |
| +// unpinned on the only implementation the queue tests ever see. |
| +// |
| +// Verdict applies the sweep's retry policy: a stored `failed` or `denied` |
| +// becomes Unknown so the next sweep re-issues it. Status reports what the tag |
| +// says, because "what is known about this object" is not "what should the |
| +// sweep do next" — left the other way, a failed scan reads as pending forever. |
| +func TestMemScannerStatusReportsWhatIsStored(t *testing.T) { |
| + ctx := context.Background() |
| + s := NewMemScanner() |
| + const bucket, key = "b", "uploads/x" |
| + mod := time.Date(2026, 9, 2, 8, 0, 0, 0, time.UTC) |
| + |
| + // An object the scanner has never heard of is GONE, not unknown: the |
| + // drain deletes the item rather than queueing an engine read. |
| + if _, _, err := s.Status(ctx, bucket, key); !errors.Is(err, ErrObjectGone) { |
| + t.Fatalf("Status on an absent object = %v, want ErrObjectGone", err) |
| + } |
| + |
| + s.PutObject(bucket, key, "etag-1", mod, 10) |
| + v, obj, err := s.Status(ctx, bucket, key) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if v != VerdictUnknown { |
| + t.Fatalf("an unscanned object = %q, want unknown", v) |
| + } |
| + // The object comes back so the caller can Scan it without a head of its |
| + // own — the whole reason Status returns one. |
| + if obj.ETag != "etag-1" || obj.Size != 10 || obj.Version() == "" { |
| + t.Fatalf("Status returned %+v, want the object a listing would have described", obj) |
| + } |
| + |
| + for _, stored := range []Verdict{VerdictFailed, VerdictDenied} { |
| + s.Put(bucket, key, obj.Version(), stored) |
| + if v, _, err := s.Status(ctx, bucket, key); err != nil || v != stored { |
| + t.Fatalf("Status over a stored %q = (%q, %v) — Status is applying the re-issue rule", stored, v, err) |
| + } |
| + if v, err := s.Verdict(ctx, bucket, key, obj.Version()); err != nil || v != VerdictUnknown { |
| + t.Fatalf("Verdict over a stored %q = (%q, %v), want unknown so the sweep re-issues it", stored, v, err) |
| + } |
| + } |
| + |
| + // A terminal verdict is reported by both, unchanged. |
| + s.Put(bucket, key, obj.Version(), VerdictInfected) |
| + if v, _, _ := s.Status(ctx, bucket, key); v != VerdictInfected { |
| + t.Fatalf("Status over infected = %q", v) |
| + } |
| + if v, _ := s.Verdict(ctx, bucket, key, obj.Version()); v != VerdictInfected { |
| + t.Fatalf("Verdict over infected = %q", v) |
| + } |
| +} |
| + |
| // The seeded-verdict path gets the same etag treatment as the scanned one, so |
| // a fixture cannot accidentally set up a verdict that outlives its bytes. |
| func TestMemScannerSeededVerdictIsETagBound(t *testing.T) { |
diff --git a/internal/objstore/scanresult.go b/internal/objstore/scanresult.go| index 95339b564..4c42d7389 100644 |
| --- a/internal/objstore/scanresult.go |
| +++ b/internal/objstore/scanresult.go |
| @@ -152,3 +152,70 @@ func DeleteScanResult(ctx context.Context, s release.Store, acct, app, objectKey |
| } |
| return nil |
| } |
| + |
| +// DeleteScanResults drops an app's whole fallback set — called when the app |
| +// may no longer be scanned at all. A result object outliving consent is a |
| +// verdict no door is allowed to answer from, and it is the orphan that blocks |
| +// the app-name claim gate. |
| +func DeleteScanResults(ctx context.Context, s release.Store, acct, app string) error { |
| + keys, err := s.List(ctx, ScanResultsPrefix(acct, app)) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return nil |
| + } |
| + return err |
| + } |
| + for _, k := range keys { |
| + if err := s.Delete(ctx, k); err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return err |
| + } |
| + } |
| + return nil |
| +} |
| + |
| +// CollectScanResults deletes an app's results older than before, and reports |
| +// how many it removed. |
| +// |
| +// WHY AGE AT ALL. A result exists because a verdict could not be written onto |
| +// the object, and the conditions that cause that — a ten-tag object, a denied |
| +// tagging call — do not clear themselves. Nothing else deletes these: the |
| +// drain only removes one when a LATER scan records a real verdict on the same |
| +// key, which for a permanently untaggable object never happens. Without a |
| +// collector an app accumulates one small object per such key forever. |
| +// |
| +// The TTL is the queue's, for the same reason the queue's is what it is: after |
| +// a sweep cycle the ordinary walk has had its chance at the object, and the |
| +// door's answer falls back to reading the tag itself. |
| +// |
| +// ModLister is optional (release/store.go:161), exactly as storagePass treats |
| +// SizeLister: a store that cannot answer is skipped, not an error — the caller |
| +// logs once rather than failing a tick over a capability a test fake lacks. |
| +func CollectScanResults(ctx context.Context, s release.Store, acct, app string, before time.Time) (int, error) { |
| + ml, ok := s.(release.ModLister) |
| + if !ok { |
| + return 0, ErrNoModLister |
| + } |
| + entries, err := ml.ListMod(ctx, ScanResultsPrefix(acct, app)) |
| + if err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + return 0, nil |
| + } |
| + return 0, err |
| + } |
| + n := 0 |
| + for _, e := range entries { |
| + if !e.Mod.Before(before) { |
| + continue |
| + } |
| + if err := s.Delete(ctx, e.Key); err != nil && !errors.Is(err, os.ErrNotExist) { |
| + return n, err |
| + } |
| + n++ |
| + } |
| + return n, nil |
| +} |
| + |
| +// ErrNoModLister is a store that cannot rank by modification time. The caller |
| +// skips the age sweep and says so once, rather than treating a missing |
| +// capability as a failure. |
| +var ErrNoModLister = errors.New("objstore: this store cannot list modification times") |
diff --git a/internal/objstore/scanresult_test.go b/internal/objstore/scanresult_test.go| index 7993429f7..93c07c84d 100644 |
| --- a/internal/objstore/scanresult_test.go |
| +++ b/internal/objstore/scanresult_test.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "errors" |
| "os" |
| + "path/filepath" |
| "testing" |
| "time" |
| @@ -113,9 +114,84 @@ func TestScanResultKeyIsUnderTheAppPrefix(t *testing.T) { |
| if ScanResultKey("bab", "titogo", "a") == ScanResultKey("bab", "titogo", "b") { |
| t.Fatal("two object keys share a result object") |
| } |
| - // The same key is stable, or the door could never find it again. |
| - if ScanResultKey("bab", "titogo", "a") != ScanResultKey("bab", "titogo", "a") { |
| - t.Fatal("result keys are not stable") |
| + // PINNED, not merely "deterministic". A key that is stable within one |
| + // build is stable enough to pass a self-comparison and still be a |
| + // different key from the one the last build wrote — and every result |
| + // written before the change becomes an object nothing will ever read or |
| + // delete, under the prefix that decides whether the app's name can be |
| + // claimed again. |
| + if want := release.AppPrefix("bab", "titogo") + |
| + "scan-results/c681b33175d57d52dc74238a2e327f0982614eeab6142d8e893dd83cb617488b.json"; got != want { |
| + t.Fatalf("key = %q, want %q — the key format moved and the old results are now unreachable", got, want) |
| + } |
| +} |
| + |
| +// The fallback set is deleted with the app's consent, and swept by age |
| +// otherwise. Nothing else removes one: the drain drops a result only when a |
| +// LATER scan records a real verdict on the same key, which for a permanently |
| +// untaggable object never happens. |
| +func TestResultObjectCollectedByAge(t *testing.T) { |
| + ctx := context.Background() |
| + dir := t.TempDir() |
| + s := release.NewDirStore(dir) |
| + |
| + for _, k := range []string{"uploads/old", "uploads/new"} { |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture(k, "clean", "v1")); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + old := time.Now().Add(-48 * time.Hour) |
| + aged := filepath.Join(dir, filepath.FromSlash(ScanResultKey("bab", "titogo", "uploads/old"))) |
| + if err := os.Chtimes(aged, old, old); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + n, err := CollectScanResults(ctx, s, "bab", "titogo", time.Now().Add(-QueueTTL)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if n != 1 { |
| + t.Fatalf("collected %d, want the one past its TTL", n) |
| + } |
| + if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", "uploads/old")); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("the aged result survived: %v", err) |
| + } |
| + if _, _, err := s.Get(ctx, ScanResultKey("bab", "titogo", "uploads/new")); err != nil { |
| + t.Fatalf("the age sweep took a result still inside its TTL: %v", err) |
| + } |
| +} |
| + |
| +// A store that cannot rank by modification time is SKIPPED, not an error — |
| +// the same degradation storagePass accepts for SizeLister. A tick that failed |
| +// over a capability a fake lacks would be a tick that stopped draining. |
| +func TestCollectScanResultsWithoutModLister(t *testing.T) { |
| + if _, err := CollectScanResults(context.Background(), plainStore{}, "bab", "titogo", time.Now()); !errors.Is(err, ErrNoModLister) { |
| + t.Fatalf("err = %v, want ErrNoModLister", err) |
| + } |
| +} |
| + |
| +// plainStore implements release.Store and nothing optional. |
| +type plainStore struct{ release.Store } |
| + |
| +// A whole app's results go when the app's consent does. |
| +func TestDeleteScanResults(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + for _, k := range []string{"uploads/a", "uploads/b"} { |
| + if err := WriteScanResult(ctx, s, "bab", "titogo", resultFixture(k, "clean", "v1")); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := DeleteScanResults(ctx, s, "bab", "titogo"); err != nil { |
| + t.Fatal(err) |
| + } |
| + left, err := s.List(ctx, ScanResultsPrefix("bab", "titogo")) |
| + if err != nil || len(left) != 0 { |
| + t.Fatalf("results survived: %v, %v", left, err) |
| + } |
| + // Deleting an empty set is not an error. |
| + if err := DeleteScanResults(ctx, s, "bab", "titogo"); err != nil { |
| + t.Fatal(err) |
| } |
| } |