amadan

carlos / platform Public

console: the scan drain, its own goroutine and its own clock

Task 12, spec §7.2. A 30s ticker, round-robin across the apps the index
names, 50 items per app per turn, and a boot rebuild that runs before the
first tick.

ITS OWN GOROUTINE, which is the whole of §7.2 and the reason this is not
three functions in scanpass.go. A queue phase inside the existing pass
cannot deliver seconds of latency: scanPass sweeps apps sequentially and
each gets a full budget, so an app that uploads a file and waits would be
waiting on a walk of somebody else's bucket. The test says that in the
only form that can fail — the sweep is PARKED inside its lister for the
whole duration of a drain, and the drain completes anyway.
`TestDrainIsItsOwnGoroutine` as a name would have passed against a queue
phase inside the sweep, which is the design being ruled out.

THE BOOT REBUILD IS THE ONLY THING THAT COLLECTS A STRANDED ITEM. Enqueue
writes the object and then bumps the index; a crash between the two
leaves an item nothing points at, and the drain only visits apps the
index names. The test runs a tick BEFORE the rebuild and requires it to
scan nothing, so the rebuild is the only difference between the halves —
otherwise it would pass against a drain that walked every app every 30s,
which is the design the index exists to avoid. Expired items and expired
result objects go in the same pass, on the same TTL.

Fairness is asserted against a flood rather than described: one app
queues more than the cap, another queues one item, and the single item is
answered in the same tick. The index entry is released a tick later than
you would guess, and the test now says why — a consumed page reports "not
empty" because a full page may have more behind it, so an app is released
on the first tick that lists nothing.

Notification happens BEFORE the error is handled. DrainApp returns what
it managed before it stopped, and a waiter whose object was scanned must
be woken even though the app's next item failed; the other order makes a
long-poll wait out its whole window for an answer that already exists.

Two lines are documented as NOT being the guarantee they look like,
because mutation testing showed neither changes anything observable:
the tick's deployment-switch check and its engine check are both about
avoiding work, since DrainApp re-reads consent and refuses on its own.
The first is now asserted the only way it can be — by counting listings —
and the second is marked as unassertable rather than left looking tested.

Thirteen mutations run against the drain, twelve caught.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01ExnZtUYmt2h8QFcymjPaSL
Paul Campbell pushed by paul@keymail.dev 8105c88cc45759c44b48c6a339ee9afe750b02f0 parent 976ff88
5 files changed, +1029 −0
  • cmd/console/main.go +4 −0
  • internal/console/scandrain.go +294 −0
  • internal/console/scandrain_test.go +651 −0
  • internal/console/server.go +5 −0
  • internal/objstore/queue.go +75 −0

CI passed — run details

diff --git a/cmd/console/main.go b/cmd/console/main.go
index b8642dc85..57f73f4e5 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -340,6 +340,10 @@ func serve() {
// here, never in NewServer. It returns immediately when no engine is
// configured, which is every deployment that has not opted in.
go srv.StartScanPass(context.Background())
+ // The on-demand drain, on its own clock (spec §7.2). Separate goroutine
+ // from the sweep above and not a phase inside it: a queue phase behind a
+ // sequential per-app budget would starve for as long as the sweep takes.
+ go srv.StartScanDrain(context.Background())
addr := envOr("CONSOLE_ADDR", "127.0.0.1:8090")
ln, err := net.Listen("tcp", addr)
if err != nil {
diff --git a/internal/console/scandrain.go b/internal/console/scandrain.go
new file mode 100644
index 000000000..cc2232178
--- /dev/null
+++ b/internal/console/scandrain.go
@@ -0,0 +1,294 @@
+// The scan drain (spec §7.2): the goroutine that empties the on-demand queue.
+//
+// ITS OWN GOROUTINE AND ITS OWN CLOCK, which is the whole reason this file is
+// not three functions inside scanpass.go. Draft 1 put a queue phase inside the
+// existing pass; that cannot deliver seconds of latency, because scanPass
+// sweeps apps sequentially and each gets a full budget (scanpass.go), so a
+// queue phase behind it starves for up to N × budget. An app that uploads a
+// file and waits for a verdict would be waiting on a walk of somebody else's
+// bucket.
+//
+// The only state it shares with the sweep is object tags, which are idempotent
+// per version: two writers agreeing on the same answer for the same bytes is
+// not a race. It never touches the sidecar.
+package console
+
+import (
+ "context"
+ "errors"
+ "log"
+ "strings"
+ "sync"
+ "time"
+
+ "github.com/carlosframework/platform/internal/objstore"
+)
+
+// scanDrainInterval is the tick. 30s, against the sweep's hour: this path
+// exists because waiting an hour for a verdict on an upload is useless.
+const scanDrainInterval = 30 * time.Second
+
+// scanDrainPerApp is §7.3's fairness cap — how many items ONE app gets in one
+// tick, not how many the deployment gets.
+//
+// 50 at a 30s tick is 100/minute, close to the enqueue door's own 120/minute
+// limit, so an app inside its rate limit is never throttled by the drain. Not
+// FIFO across the deployment, deliberately: FIFO lets one tenant's ten
+// thousand queued objects delay every other tenant's single urgent upload.
+const scanDrainPerApp = 50
+
+// StartScanDrain rebuilds the index and then drains every 30s until ctx is
+// done. serve() starts it; constructing a Server starts nothing.
+//
+// THE REBUILD RUNS FIRST AND BLOCKS THE FIRST TICK, because a tick before it
+// would drain only the apps this process happens to have heard from — which,
+// one second after a restart, is none of them.
+func (s *Server) StartScanDrain(ctx context.Context) {
+ if s.cfg.Releases == nil || s.cfg.StoreScanner == nil {
+ return // no bucket, or no engine: nothing to drain into
+ }
+ if err := s.rebuildScanQueue(ctx); err != nil {
+ // Not fatal. A console that cannot rebuild still drains what it is
+ // told about from here on, and the next restart tries again; refusing
+ // to start the goroutine would turn a listing hiccup into an outage
+ // of the whole on-demand path.
+ log.Printf("console: scan queue rebuild: %v", err)
+ }
+ t := time.NewTicker(scanDrainInterval)
+ defer t.Stop()
+ for {
+ if err := s.scanDrainTick(ctx); err != nil {
+ log.Printf("console: scan drain: %v", err)
+ }
+ select {
+ case <-ctx.Done():
+ return
+ case <-t.C:
+ }
+ }
+}
+
+// rebuildScanQueue is the one full enumeration (§7.1): every app on the
+// deployment, so that an item stranded by a crash between the object write and
+// the index bump is found, and everything past its TTL is collected.
+//
+// Same enumeration as scanPass, and the same accepted cost — but ONCE at boot
+// rather than hourly, so it is the cheap one of the two.
+//
+// One app's failure does not stop the others: this pass exists to recover what
+// it can, and an account whose listing fails should not cost every account
+// after it its queue.
+func (s *Server) rebuildScanQueue(ctx context.Context) error {
+ if s.cfg.Releases == nil {
+ return nil
+ }
+ now := time.Now()
+ acctPrefixes, err := s.cfg.Releases.ListDirs(ctx, "apps/")
+ if err != nil {
+ return err
+ }
+ for _, acctPrefix := range acctPrefixes {
+ acct := strings.TrimSuffix(strings.TrimPrefix(acctPrefix, "apps/"), "/")
+ appPrefixes, err := s.cfg.Releases.ListDirs(ctx, acctPrefix)
+ if err != nil {
+ log.Printf("console: scan queue rebuild: %s: %v", acct, err)
+ continue
+ }
+ for _, appPrefix := range appPrefixes {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ app := strings.TrimSuffix(strings.TrimPrefix(appPrefix, acctPrefix), "/")
+ if _, err := objstore.RebuildQueue(ctx, s.cfg.Releases, s.scanQueue, acct, app, now); err != nil {
+ log.Printf("console: scan queue rebuild: %s/%s: %v", acct, app, err)
+ }
+ // The fallback results are collected in the SAME pass, because
+ // this is the only place that visits an app which has stopped
+ // enqueuing. The per-tick collector below only reaches apps that
+ // still have work.
+ s.collectScanResults(ctx, acct, app, now)
+ }
+ }
+ return nil
+}
+
+// scanDrainTick gives every app with work one capped turn.
+//
+// Round-robin is what iterating the index IS: every app named in it gets a
+// turn each tick and none gets more than scanDrainPerApp, so an app cannot
+// buy latency from another by queueing more.
+//
+// The deployment switch is read every tick, exactly as scanPass reads it, so
+// `carlos system antivirus enable` takes effect without a restart. Disabled
+// means this tick does nothing — not that there is no goroutine, which is what
+// would make enabling need one.
+func (s *Server) scanDrainTick(ctx context.Context) error {
+ // An early-out for a direct caller, not a guarantee: StartScanDrain never
+ // starts without an engine, and DrainApp refuses with ErrNoEngine before
+ // it lists anything. Removing this line changes nothing observable — it
+ // only stops a tick from logging one refusal per app. Kept because
+ // scanPass guards the same way, and said here so nobody looks for the
+ // test that pins it.
+ if s.cfg.Releases == nil || s.cfg.StoreScanner == nil {
+ return nil
+ }
+ enabled, err := s.antivirusEnabled(ctx)
+ if err != nil {
+ return err
+ }
+ if !enabled {
+ // NOT THE SAFETY NET, and worth being exact about: DrainApp re-reads
+ // consent for itself and refuses a disabled deployment on its own, so
+ // deleting this check changes no verdict and loses no item. What it
+ // changes is WORK — with it a disabled deployment costs one record
+ // read per tick, without it every app with an index entry is listed
+ // every 30s for as long as it stays off.
+ //
+ // Items are KEPT, not dropped: this refusal is reversible, and the
+ // TTL is what eventually clears a deployment left off for a day.
+ return nil
+ }
+ now := time.Now()
+ for _, ref := range s.scanQueue.Apps() {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ s.drainScanApp(ctx, ref, now)
+ }
+ return nil
+}
+
+// drainScanApp drains one app and tells whoever is waiting.
+//
+// Every outcome is logged and none is returned: one app's engine trouble must
+// not end the tick for the apps after it in the round-robin, which is the
+// posture scanPass already takes for the same reason.
+func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time.Time) {
+ done, _, err := objstore.DrainApp(ctx, s.cfg.Releases, s.cfg.StoreScanner, s.scanQueue,
+ ref.Account, ref.App, scanDrainPerApp, now)
+
+ // NOTIFY BEFORE HANDLING THE ERROR. DrainApp returns the items it managed
+ // before it stopped, and a waiter whose object was actually scanned should
+ // be woken even though the app's next item failed. Dropping these on the
+ // error path would make a long-poll wait out its full window for an answer
+ // that already exists.
+ for _, d := range done {
+ s.scanNotify(ref, d.Key)
+ }
+
+ switch {
+ case err == nil:
+ case errors.Is(err, objstore.ErrQueueNotPermitted):
+ // Not an incident: consent was withdrawn, or the app was trashed, and
+ // DrainApp has already deleted the items and the results. Logged at
+ // all only because "the queue emptied without scanning" is a thing an
+ // operator reading a support question wants to be able to see.
+ log.Printf("console: scan drain: %s/%s: queue dropped — this store may no longer be scanned",
+ ref.Account, ref.App)
+ case errors.Is(err, objstore.ErrScanningDisabled):
+ // A disable that landed between the tick's check and this app. The
+ // items stay; the next enabled tick has them.
+ default:
+ log.Printf("console: scan drain: %s/%s: %v", ref.Account, ref.App, err)
+ }
+ s.collectScanResults(ctx, ref.Account, ref.App, now)
+}
+
+// collectScanResults ages out an app's fallback result objects.
+//
+// PER VISITED APP rather than in one deployment-wide sweep, so that a console
+// which never restarts does not accumulate them — nothing else deletes a
+// result for an object that stays permanently untaggable, because the drain
+// only removes one when a later scan records a real verdict on the same key.
+func (s *Server) collectScanResults(ctx context.Context, acct, app string, now time.Time) {
+ n, err := objstore.CollectScanResults(ctx, s.cfg.Releases, acct, app, now.Add(-objstore.QueueTTL))
+ switch {
+ case errors.Is(err, objstore.ErrNoModLister):
+ // A store that cannot report modification times, which is a test fake
+ // rather than a deployment. Skipped, not failed — storagePass treats
+ // its own optional interface the same way.
+ case err != nil:
+ log.Printf("console: scan result collect: %s/%s: %v", acct, app, err)
+ case n > 0:
+ log.Printf("console: scan result collect: %s/%s: dropped %d expired result(s)", acct, app, n)
+ }
+}
+
+// ScanQueueDepth is how many apps have on-demand work waiting — the
+// deployment-wide number §4.4 wants on `carlos system antivirus status`.
+//
+// APPS, not items, and named for what it counts. The per-app item count is
+// objstore.QueueDepth, which costs a listing; this one is free and is the
+// number that answers "is the drain keeping up".
+func (s *Server) ScanQueueDepth() int {
+ if s.scanQueue == nil {
+ return 0
+ }
+ return s.scanQueue.Len()
+}
+
+// ---- waking the long-poll ---------------------------------------------------
+
+// scanWaitKey is one object, for one app.
+type scanWaitKey struct {
+ ref objstore.AppRef
+ key string
+}
+
+// scanWaiters is the in-process notification the status door's `wait` uses
+// (§5.2), so a long-poll costs a channel rather than a tag read per second per
+// waiter.
+//
+// IN-PROCESS ONLY, and that is a correctness statement rather than a caveat: a
+// waiter on console A is never woken by a drain on console B. The door must
+// therefore treat a wake as an OPTIMISATION and still read the tag when its
+// timer fires — never as the only way an answer arrives.
+type scanWaiters struct {
+ mu sync.Mutex
+ m map[scanWaitKey]map[chan struct{}]struct{}
+}
+
+// scanWait registers interest in one object and returns the channel to select
+// on, plus the cancel that must be deferred.
+//
+// The channel is BUFFERED to depth one and the notifier never blocks on it, so
+// a waiter that has already given up cannot stall the drain — the goroutine
+// whose latency this whole path exists to protect.
+func (s *Server) scanWait(ref objstore.AppRef, key string) (<-chan struct{}, func()) {
+ w := &s.scanWaits
+ k := scanWaitKey{ref: ref, key: key}
+ ch := make(chan struct{}, 1)
+ w.mu.Lock()
+ if w.m == nil {
+ w.m = map[scanWaitKey]map[chan struct{}]struct{}{}
+ }
+ if w.m[k] == nil {
+ w.m[k] = map[chan struct{}]struct{}{}
+ }
+ w.m[k][ch] = struct{}{}
+ w.mu.Unlock()
+ return ch, func() {
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ if set := w.m[k]; set != nil {
+ delete(set, ch)
+ if len(set) == 0 {
+ // The map must not grow one entry per key ever waited on.
+ delete(w.m, k)
+ }
+ }
+ }
+}
+
+// scanNotify wakes everyone waiting on one object. Never blocks.
+func (s *Server) scanNotify(ref objstore.AppRef, key string) {
+ w := &s.scanWaits
+ w.mu.Lock()
+ defer w.mu.Unlock()
+ for ch := range w.m[scanWaitKey{ref: ref, key: key}] {
+ select {
+ case ch <- struct{}{}:
+ default: // already woken and not yet read; one wake is all it needs
+ }
+ }
+}
diff --git a/internal/console/scandrain_test.go b/internal/console/scandrain_test.go
new file mode 100644
index 000000000..a07deb79f
--- /dev/null
+++ b/internal/console/scandrain_test.go
@@ -0,0 +1,651 @@
+package console
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/objstore"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// drainFixture is a console whose release store is a PLAIN DirStore.
+//
+// The shared fixture wraps its store in a struct that embeds release.Store as
+// an INTERFACE, which hides the two optional interfaces this whole file is
+// about: PageLister, which is how the drain caps one app's turn (§7.3), and
+// ModLister, which is how expired result objects are collected. A test on the
+// wrapper would silently exercise the degraded fallback in both cases and
+// prove nothing about the path a deployment actually runs.
+//
+// It returns the bucket root because two tests have to AGE a file, and mtime
+// is what DirStore's ListMod reads.
+func drainFixture(t *testing.T) (*consoleFixture, *objstore.MemScanner, string) {
+ t.Helper()
+ fx := newConsoleFixture(t)
+ dir := t.TempDir()
+ rel := release.NewDirStore(dir)
+ fx.rel = rel
+ fx.srv.cfg.Releases = rel
+ sc := objstore.NewMemScanner()
+ fx.srv.cfg.StoreScanner = sc
+ return fx, sc, dir
+}
+
+func drainBucket(acct, app string) string { return "carlos-" + acct + "-" + app }
+
+// enqueueFor puts one item in the bucket AND in the server's index, which is
+// what the door does.
+func enqueueFor(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
+ t.Helper()
+ if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, acct, app,
+ &objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
+ t.Fatalf("enqueue %s: %v", key, err)
+ }
+}
+
+// enqueueStranded is the crash §7.1 describes: the item object is written and
+// the process dies before the index is bumped. Modelled by bumping a THROWAWAY
+// index, so the bucket has the item and this console has never heard of it.
+func enqueueStranded(t *testing.T, fx *consoleFixture, acct, app, key string, at time.Time) {
+ t.Helper()
+ if _, err := objstore.Enqueue(context.Background(), fx.rel, objstore.NewQueueIndex(), acct, app,
+ &objstore.QueueItem{Key: key, By: "test"}, at); err != nil {
+ t.Fatalf("enqueue %s: %v", key, err)
+ }
+}
+
+func queueLen(t *testing.T, fx *consoleFixture, acct, app string) int {
+ t.Helper()
+ n, err := objstore.QueueDepth(context.Background(), fx.rel, acct, app)
+ if err != nil {
+ t.Fatalf("queue depth: %v", err)
+ }
+ return n
+}
+
+func resultKeys(t *testing.T, fx *consoleFixture, acct, app string) []string {
+ t.Helper()
+ keys, err := fx.rel.List(context.Background(), objstore.ScanResultsPrefix(acct, app))
+ if err != nil {
+ t.Fatalf("list results: %v", err)
+ }
+ return keys
+}
+
+// countingStore counts listings, which is the only way to see the difference
+// between "the drain refused" and "the drain did the work and objstore
+// refused".
+//
+// It embeds the CONCRETE *release.DirStore rather than the interface, so
+// ListMod and ListDirs are still promoted — embedding the interface is exactly
+// the trap drainFixture's comment describes.
+type countingStore struct {
+ *release.DirStore
+ lists atomic.Int64
+}
+
+func (c *countingStore) List(ctx context.Context, prefix string) ([]string, error) {
+ c.lists.Add(1)
+ return c.DirStore.List(ctx, prefix)
+}
+
+func (c *countingStore) ListPage(ctx context.Context, prefix, startAfter string, limit int) ([]string, error) {
+ c.lists.Add(1)
+ return c.DirStore.ListPage(ctx, prefix, startAfter, limit)
+}
+
+// setDeploymentScanning flips the deployment switch either way.
+//
+// enableDeploymentScanning is CREATE-ONLY — it returns early when a record
+// already exists, which is right for a fixture arming a fresh deployment and
+// useless for re-enabling one that was turned off.
+func setDeploymentScanning(t *testing.T, fx *consoleFixture, on bool) {
+ t.Helper()
+ ctx := context.Background()
+ av, etag, err := release.GetAntivirus(ctx, fx.rel)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if av == nil {
+ av = &release.AntivirusSettings{Engine: "clamav"}
+ }
+ av.Enabled = on
+ av.At = time.Now().UTC()
+ av.By = "test"
+ if _, err := release.PutAntivirus(ctx, fx.rel, av, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// revokeScanning is the operator taking the per-store grant away AFTER the
+// record exists, which enableStore cannot do: it writes with an empty etag and
+// so conflicts with anything already there.
+func revokeScanning(t *testing.T, fx *consoleFixture, acct, app string) {
+ t.Helper()
+ ctx := context.Background()
+ rec, etag, err := objstore.Get(ctx, fx.rel, acct, app)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if rec == nil {
+ t.Fatalf("no record for %s/%s to revoke", acct, app)
+ }
+ rec.Scanning = nil
+ if _, err := objstore.Put(ctx, fx.rel, acct, app, rec, etag); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestBootRebuildFindsStrandedItem is the whole argument for the rebuild
+// existing, and it is written so that the rebuild is the ONLY difference
+// between the two halves.
+//
+// A tick runs first and must do nothing: the drain only visits apps the index
+// names, and nothing named this one. Then the rebuild runs, and the same tick
+// scans. Without the "before" half this would pass against a drain that walked
+// every app on the deployment every 30s — which is the design the index exists
+// to avoid.
+func TestBootRebuildFindsStrandedItem(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
+ enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
+
+ if d := fx.srv.ScanQueueDepth(); d != 0 {
+ t.Fatalf("index depth %d before the rebuild — the item was not stranded, so this test proves nothing", d)
+ }
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := sc.Scans("uploads/x.pdf"); n != 0 {
+ t.Fatalf("a tick scanned %d time(s) for an app the index never named", n)
+ }
+
+ if err := fx.srv.rebuildScanQueue(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if d := fx.srv.ScanQueueDepth(); d != 1 {
+ t.Fatalf("index depth %d after the rebuild, want 1 — nothing would ever list that prefix again", d)
+ }
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := sc.Scans("uploads/x.pdf"); n != 1 {
+ t.Fatalf("scanned %d times after the rebuild, want 1", n)
+ }
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
+ t.Errorf("%d item(s) left after a successful drain", n)
+ }
+}
+
+// TestBootRebuildCollectsExpiredItemsAndResults covers BOTH prefixes on the
+// same 24h TTL, because they expire for the same reason: after a sweep cycle
+// the ordinary walk has had its chance at the object, and a stale on-demand
+// answer is worth less than none.
+//
+// The result object matters more than the item. A stranded item is drained on
+// the next tick; a stranded result is found by nothing, blocks the app-name
+// claim gate, and would sit in the bucket forever.
+func TestBootRebuildCollectsExpiredItemsAndResults(t *testing.T) {
+ ctx := context.Background()
+ fx, _, dir := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+
+ old := time.Now().Add(-objstore.QueueTTL - time.Hour)
+ enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/stale.pdf", old)
+ if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
+ Key: "uploads/untaggable.pdf", Verdict: "clean", Version: "etag-9@x",
+ Engine: "clamav", Signature: "sig", At: old.UTC(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ // ListMod reads MTIME, not the At field inside the JSON, so the file has
+ // to actually be old.
+ resKey := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/untaggable.pdf")
+ if err := os.Chtimes(filepath.Join(dir, filepath.FromSlash(resKey)), old, old); err != nil {
+ t.Fatal(err)
+ }
+
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
+ t.Fatalf("%d item(s) staged, want 1", n)
+ }
+ if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 1 {
+ t.Fatalf("%d result(s) staged, want 1", len(got))
+ }
+
+ if err := fx.srv.rebuildScanQueue(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
+ t.Errorf("%d expired item(s) survived the rebuild", n)
+ }
+ if got := resultKeys(t, fx, fixtureAcct, fixtureApp); len(got) != 0 {
+ t.Errorf("expired result(s) survived the rebuild: %v", got)
+ }
+ // And nothing is left pointing at an app with no work.
+ if d := fx.srv.ScanQueueDepth(); d != 0 {
+ t.Errorf("index depth %d after collecting everything, want 0", d)
+ }
+}
+
+// TestRebuildKeepsAnItemInsideItsTTL is the other half of the test above, and
+// it is here because "collects expired items" passes trivially against a
+// rebuild that deletes everything it finds.
+func TestRebuildKeepsAnItemInsideItsTTL(t *testing.T) {
+ ctx := context.Background()
+ fx, _, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/fresh.pdf", time.Now().Add(-time.Hour))
+
+ if err := fx.srv.rebuildScanQueue(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
+ t.Fatalf("%d item(s) after the rebuild, want the one that is an hour old", n)
+ }
+ if d := fx.srv.ScanQueueDepth(); d != 1 {
+ t.Errorf("index depth %d, want 1 — a kept item with no index entry is stranded again", d)
+ }
+}
+
+// TestDrainRunsWhileSweepIsBlocked is §7.2 stated as something observable.
+//
+// "TestDrainIsItsOwnGoroutine" as a name would pass against anything at all —
+// against a queue phase inside scanPass, even. So the sweep is PARKED, inside
+// the lister, for the whole duration of a drain, and the drain has to complete
+// anyway. That is the only formulation that fails if the two are ever folded
+// together, which is what draft 1 proposed and what would cost this path the
+// seconds of latency it exists for.
+func TestDrainRunsWhileSweepIsBlocked(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())
+
+ entered := make(chan struct{}, 1)
+ release := make(chan struct{})
+ fx.srv.storeListerFor = func(context.Context, *objstore.Provisioned) (objstore.ObjectLister, error) {
+ select {
+ case entered <- struct{}{}:
+ default:
+ }
+ <-release
+ return &countingLister{}, nil
+ }
+ swept := make(chan struct{})
+ go func() {
+ _ = fx.srv.scanPass(ctx)
+ close(swept)
+ }()
+
+ select {
+ case <-entered:
+ case <-time.After(5 * time.Second):
+ close(release)
+ t.Fatal("the sweep never reached the lister — the test never parked it")
+ }
+
+ // The sweep is now stopped dead inside scanOneApp. The drain must not be.
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatalf("drain tick while the sweep is parked: %v", err)
+ }
+ if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
+ t.Fatalf("scanned %d times while the sweep held its budget, want 1", n)
+ }
+
+ close(release)
+ <-swept
+}
+
+// TestDrainRoundRobinFairness: one app's flood must not delay another app's
+// single item beyond one tick.
+//
+// FIFO across the deployment is the design this refuses. The flooder is given
+// more items than the per-app cap on purpose, so that a drain which emptied
+// one app before moving on would still be inside the flooder when the tick
+// ended, and the other app's item would go unanswered.
+func TestDrainRoundRobinFairness(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ enableStore(t, fx, fixtureAcct, "flooder", true)
+
+ flood := scanDrainPerApp + 10
+ for i := 0; i < flood; i++ {
+ key := fmt.Sprintf("uploads/flood-%03d.bin", i)
+ sc.PutObject(drainBucket(fixtureAcct, "flooder"), key, "etag-f", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, "flooder", key, time.Now())
+ }
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/urgent.pdf", "etag-u", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/urgent.pdf", time.Now())
+
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := sc.Scans("uploads/urgent.pdf"); n != 1 {
+ t.Fatalf("the urgent item scanned %d times in the tick that also met a flood of %d, want 1", n, flood)
+ }
+ // And the cap held: the flooder did not get the whole queue in one turn.
+ left := queueLen(t, fx, fixtureAcct, "flooder")
+ if left != flood-scanDrainPerApp {
+ t.Errorf("flooder has %d item(s) left, want %d — the per-app cap is not being applied",
+ left, flood-scanDrainPerApp)
+ }
+ // A SECOND tick, because the release rule needs one: DrainApp reports a
+ // consumed page as "not empty" — a full page may have more behind it — so
+ // an app's entry goes on the first tick that lists nothing, not on the
+ // tick that emptied it. After tick two the urgent app is released and the
+ // flooder, which still had a partial page, is not.
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := queueLen(t, fx, fixtureAcct, "flooder"); n != 0 {
+ t.Errorf("flooder has %d item(s) after two ticks, want 0", n)
+ }
+ if d := fx.srv.ScanQueueDepth(); d != 1 {
+ t.Errorf("index depth %d after two ticks, want 1 — the drained app should have been released and the flooder kept", d)
+ }
+}
+
+// TestDrainAgeSweepsResultObjects: per VISITED app, so a console that never
+// restarts does not accumulate them.
+//
+// Nothing else deletes these. The drain removes a result only when a later
+// scan records a real verdict on the same key, and for an object that is
+// permanently untaggable — the ten-tag case — that never happens.
+func TestDrainAgeSweepsResultObjects(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, dir := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/live.pdf", "etag-l", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/live.pdf", time.Now())
+
+ old := time.Now().Add(-objstore.QueueTTL - time.Hour)
+ for _, k := range []string{"uploads/old.pdf", "uploads/recent.pdf"} {
+ if err := objstore.WriteScanResult(ctx, fx.rel, fixtureAcct, fixtureApp, &objstore.ScanResult{
+ Key: k, Verdict: "clean", Version: "etag-x@y", Engine: "clamav", Signature: "sig",
+ At: time.Now().UTC(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ }
+ oldPath := filepath.Join(dir, filepath.FromSlash(
+ objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/old.pdf")))
+ if err := os.Chtimes(oldPath, old, old); err != nil {
+ t.Fatal(err)
+ }
+
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ got := resultKeys(t, fx, fixtureAcct, fixtureApp)
+ if len(got) != 1 {
+ t.Fatalf("results after the tick = %v, want only the recent one", got)
+ }
+ // The one inside its TTL survives — a collector that dropped both would
+ // pass a test that only counted the expired one.
+ if want := objstore.ScanResultKey(fixtureAcct, fixtureApp, "uploads/recent.pdf"); got[0] != want {
+ t.Errorf("survivor = %q, want %q", got[0], want)
+ }
+}
+
+// TestQueueDepthReported: §4.4 wants it deployment-wide and §10 wants it per
+// app, and they are different numbers measured in different ways.
+func TestQueueDepthReported(t *testing.T) {
+ ctx := context.Background()
+ fx, _, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ enableStore(t, fx, fixtureAcct, "second", true)
+
+ if d := fx.srv.ScanQueueDepth(); d != 0 {
+ t.Fatalf("depth %d on an idle deployment, want 0", d)
+ }
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "a.pdf", time.Now())
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "b.pdf", time.Now())
+ enqueueFor(t, fx, fixtureAcct, "second", "c.pdf", time.Now())
+
+ // Deployment-wide counts APPS with work, not items: two apps, three items.
+ if d := fx.srv.ScanQueueDepth(); d != 2 {
+ t.Errorf("deployment depth = %d, want 2 apps with work", d)
+ }
+ // Per app counts ITEMS.
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 2 {
+ t.Errorf("per-app depth = %d, want 2 items", n)
+ }
+ if n := queueLen(t, fx, fixtureAcct, "second"); n != 1 {
+ t.Errorf("per-app depth = %d, want 1 item", n)
+ }
+ // An app that has never enqueued has no prefix, and that is depth zero
+ // rather than an error.
+ if n := queueLen(t, fx, fixtureAcct, "never-used"); n != 0 {
+ t.Errorf("depth %d for an app with no queue prefix, want 0", n)
+ }
+ _ = ctx
+}
+
+// TestDrainDoesNothingWhileTheDeploymentIsDisabled, and specifically KEEPS the
+// items: the deployment switch is the one reversible refusal, and an operator
+// running `enable` should find the queue still there. The TTL is what clears a
+// deployment left off for a day, not the tick.
+func TestDrainDoesNothingWhileTheDeploymentIsDisabled(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, dir := drainFixture(t)
+ counting := &countingStore{DirStore: release.NewDirStore(dir)}
+ fx.rel = counting
+ fx.srv.cfg.Releases = counting
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
+ setDeploymentScanning(t, fx, false)
+
+ // The tick's own switch check is not what keeps the bytes safe — DrainApp
+ // re-reads consent and refuses on its own, so removing the tick's check
+ // changes no verdict and no item. What it changes is WORK: with it, a
+ // disabled deployment costs one record read; without it, every app with a
+ // queue entry is listed every 30s forever. Listings are therefore what
+ // this asserts, because they are the only observable difference.
+ counting.lists.Store(0)
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := counting.lists.Load(); n != 0 {
+ t.Errorf("%d listing(s) on a disabled deployment — the tick is doing per-app work it can never use", n)
+ }
+ if n := sc.Scans("uploads/x.pdf"); n != 0 {
+ t.Errorf("scanned %d times with the deployment switch off", n)
+ }
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
+ t.Fatalf("%d item(s) left, want the item KEPT — this refusal is reversible", n)
+ }
+
+ // And it drains once the operator turns it back on, without a restart.
+ setDeploymentScanning(t, fx, true)
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := sc.Scans("uploads/x.pdf"); n != 1 {
+ t.Errorf("scanned %d times after re-enabling, want 1", n)
+ }
+}
+
+// TestDrainDropsTheQueueWhenConsentIsGone is §7.5's other side: the permanent
+// refusals delete, because none of them can change back in a way that makes
+// the item useful, and an item that outlives consent is a request to read
+// bytes nobody may read.
+func TestDrainDropsTheQueueWhenConsentIsGone(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
+ // The operator revokes the per-store grant between enqueue and drain.
+ revokeScanning(t, fx, fixtureAcct, fixtureApp)
+
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if n := sc.Scans("uploads/x.pdf"); n != 0 {
+ t.Errorf("read bytes %d time(s) from a store that may not be scanned", n)
+ }
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 0 {
+ t.Errorf("%d item(s) survived a permanent refusal", n)
+ }
+}
+
+// TestScanNotifyWakesAWaiter is the long-poll's half of the drain (§5.2). The
+// door lands in task 13; the wake it selects on is delivered here, so it has
+// something to be written against.
+func TestScanNotifyWakesAWaiter(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
+
+ ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
+ woken, cancel := fx.srv.scanWait(ref, "uploads/x.pdf")
+ defer cancel()
+ // A waiter on a DIFFERENT key must not be woken by this drain.
+ other, cancelOther := fx.srv.scanWait(ref, "uploads/other.pdf")
+ defer cancelOther()
+
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+ select {
+ case <-woken:
+ default:
+ t.Error("the drain scanned the object and nothing woke the waiter")
+ }
+ select {
+ case <-other:
+ t.Error("a waiter on another key was woken")
+ default:
+ }
+
+ // Cancelling removes the registration rather than leaking one entry per
+ // key ever waited on.
+ cancel()
+ fx.srv.scanWaits.mu.Lock()
+ n := len(fx.srv.scanWaits.m[scanWaitKey{ref: ref, key: "uploads/x.pdf"}])
+ fx.srv.scanWaits.mu.Unlock()
+ if n != 0 {
+ t.Errorf("%d waiter(s) still registered after cancel", n)
+ }
+}
+
+// TestDrainNotifiesEvenWhenALaterItemFails: DrainApp returns what it managed
+// before it stopped, and a waiter whose object WAS scanned must be woken even
+// though the app's next item failed. Notifying after the error check would
+// make a long-poll wait out its whole window for an answer that already
+// exists on the object.
+//
+// Reaching that state takes some care with the fake. The FIRST object is
+// pre-tagged at its current version, so Status answers from the tag and Scan
+// is never called for it — which leaves the one-shot Scan failure below for
+// the SECOND object. done therefore carries the first while err carries the
+// second, which is exactly the shape being asserted.
+func TestDrainNotifiesEvenWhenALaterItemFails(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ bucket := drainBucket(fixtureAcct, fixtureApp)
+
+ mod := time.Now().Truncate(time.Second)
+ first := objstore.ScanObject{Key: "uploads/first.pdf", ETag: "etag-1", Modified: mod, Size: 10}
+ sc.PutObject(bucket, first.Key, first.ETag, mod, first.Size)
+ sc.Put(bucket, first.Key, first.Version(), objstore.VerdictClean)
+ sc.PutObject(bucket, "uploads/second.pdf", "etag-2", mod, 10)
+
+ // Ids sort by time, so the first item is drained first.
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, first.Key, time.Now())
+ enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/second.pdf", time.Now().Add(2*time.Millisecond))
+
+ ref := objstore.AppRef{Account: fixtureAcct, App: fixtureApp}
+ woken, cancel := fx.srv.scanWait(ref, first.Key)
+ defer cancel()
+
+ sc.Fail("Scan", errors.New("engine had a bad moment"))
+
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err) // the tick logs per-app failures rather than returning them
+ }
+ if sc.Scans(first.Key) != 0 {
+ t.Fatal("the pre-tagged object was rescanned, so the Scan failure did not reach the second item")
+ }
+ select {
+ case <-woken:
+ default:
+ t.Error("the first object was answered and its waiter was not woken, because a later item failed")
+ }
+ // The failed item is KEPT — an engine failure is about the engine, not
+ // about whether this request was legitimate.
+ if n := queueLen(t, fx, fixtureAcct, fixtureApp); n != 1 {
+ t.Errorf("%d item(s) left, want the failed one kept for the next tick", n)
+ }
+}
+
+// TestStartScanDrainRebuildsBeforeItsFirstTick covers the goroutine itself,
+// which the tick-level tests above do not: they call rebuildScanQueue and
+// scanDrainTick by hand, so a StartScanDrain that never ran the rebuild would
+// pass every one of them and strand every item across every restart.
+//
+// It is also the only test of "the first tick is immediate". A loop that
+// waited a full interval before its first pass would make every restart cost
+// 30s of latency on work that was already queued.
+func TestStartScanDrainRebuildsBeforeItsFirstTick(t *testing.T) {
+ fx, sc, _ := drainFixture(t)
+ enableStore(t, fx, fixtureAcct, fixtureApp, true)
+ sc.PutObject(drainBucket(fixtureAcct, fixtureApp), "uploads/x.pdf", "etag-1", time.Now(), 10)
+ enqueueStranded(t, fx, fixtureAcct, fixtureApp, "uploads/x.pdf", time.Now())
+
+ ctx, cancel := context.WithCancel(context.Background())
+ defer cancel()
+ stopped := make(chan struct{})
+ go func() {
+ fx.srv.StartScanDrain(ctx)
+ close(stopped)
+ }()
+
+ deadline := time.Now().Add(5 * time.Second)
+ for sc.Scans("uploads/x.pdf") == 0 {
+ if time.Now().After(deadline) {
+ cancel()
+ <-stopped
+ t.Fatal("the goroutine did not rebuild and drain a stranded item within 5s — well inside its first tick")
+ }
+ time.Sleep(2 * time.Millisecond)
+ }
+ cancel()
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Fatal("StartScanDrain did not return when its context was cancelled")
+ }
+}
+
+// TestStartScanDrainIsInertWithoutAnEngine: scanning is opt-in, and the
+// ordinary deployment has no clamd. The goroutine must return rather than tick
+// forever over a deployment that can never drain anything.
+func TestStartScanDrainIsInertWithoutAnEngine(t *testing.T) {
+ fx, _, _ := drainFixture(t)
+ fx.srv.cfg.StoreScanner = nil
+
+ stopped := make(chan struct{})
+ go func() {
+ fx.srv.StartScanDrain(context.Background())
+ close(stopped)
+ }()
+ select {
+ case <-stopped:
+ case <-time.After(5 * time.Second):
+ t.Fatal("StartScanDrain kept ticking on a deployment with no engine")
+ }
+}
diff --git a/internal/console/server.go b/internal/console/server.go
index f1e393f74..6982631b1 100644
--- a/internal/console/server.go
+++ b/internal/console/server.go
@@ -695,6 +695,11 @@ type Server struct {
// first use would be the race, and a Server that always has one cannot
// have it.
scanQueue *objstore.QueueIndex
+ // scanWaits is the in-process wake the status door's long-poll selects on
+ // (scandrain.go). Zero value is usable, which is why it is a value rather
+ // than a pointer: a Server built by a test that never starts the drain
+ // still answers scanWait and scanNotify rather than panicking.
+ scanWaits scanWaiters
// ceCosts is billPass's test/fake seam for Cost Explorer (cepull.go,
// economicspass.go): nil means "use the real, lazily-built CE client"
// (s.realCECosts); tests inject a fake so a pull can be proven to
diff --git a/internal/objstore/queue.go b/internal/objstore/queue.go
index f1bdda53e..eb5895426 100644
--- a/internal/objstore/queue.go
+++ b/internal/objstore/queue.go
@@ -203,6 +203,81 @@ func Enqueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app st
return id, nil
}
+// ---- the boot rebuild -------------------------------------------------------
+
+// RebuildQueue restores one app's index entry from what is actually in the
+// bucket, and drops what expired while nobody was draining.
+//
+// THIS IS THE ONLY THING THAT COLLECTS A STRANDED ITEM, so it is not optional.
+// Enqueue writes the object and then bumps the index; a crash between the two
+// leaves an item in the bucket with nothing pointing at it, and the drain only
+// ever looks at apps the index names. Nothing else would ever list that
+// prefix again.
+//
+// UNCAPPED, unlike the drain's listing, and deliberately: the drain's cap is
+// fairness between tenants within a tick, while this runs once and must see
+// everything or the stranding it exists to fix survives it.
+//
+// live counts what a drain could still act on, and an item whose bytes could
+// not be READ counts as live. An unreadable item is deleted (no tick will ever
+// parse it), but a STORE error is the store having a bad moment: keeping the
+// index entry costs one empty listing next tick, while dropping it strands the
+// item until the next restart. Absence, never error — §7.5.
+func RebuildQueue(ctx context.Context, s release.Store, idx *QueueIndex, acct, app string, now time.Time) (live int, err error) {
+ keys, err := s.List(ctx, QueuePrefix(acct, app))
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return 0, nil
+ }
+ return 0, err
+ }
+ for _, qk := range keys {
+ item, rerr := readQueueItem(ctx, s, qk)
+ switch {
+ case errors.Is(rerr, os.ErrNotExist):
+ continue
+ case errors.Is(rerr, errQueueItemUnreadable):
+ if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) {
+ return live, derr
+ }
+ continue
+ case rerr != nil:
+ log.Printf("objstore: scan queue rebuild: read %s: %v", qk, rerr)
+ live++
+ continue
+ }
+ if now.Sub(item.At) > QueueTTL {
+ if derr := s.Delete(ctx, qk); derr != nil && !errors.Is(derr, os.ErrNotExist) {
+ return live, derr
+ }
+ continue
+ }
+ live++
+ }
+ if live > 0 {
+ idx.Bump(AppRef{Account: acct, App: app})
+ }
+ return live, nil
+}
+
+// QueueDepth is how many items are waiting for one app — the per-app half of
+// the depth §4.4 and §10 both want on a status page.
+//
+// A full listing, not the index: the index counts APPS with work, which is the
+// deployment-wide number, and says nothing about how much any one of them has.
+// Absence is depth zero, not an error: an app that has never enqueued has no
+// prefix.
+func QueueDepth(ctx context.Context, s release.Store, acct, app string) (int, error) {
+ keys, err := s.List(ctx, QueuePrefix(acct, app))
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return 0, nil
+ }
+ return 0, err
+ }
+ return len(keys), nil
+}
+
// ---- draining ---------------------------------------------------------------
// Drained is one item's outcome, everything console needs and nothing it does