amadan
console+cli: scan metering and the queue-depth surfaces
Task 17's code half. Written by codex (gpt-5.6-sol); the mutation testing and this message are mine. The DOCS half is deliberately absent — see the end. ON-DEMAND WORK ONLY IS METERED, and that is the line worth guarding because getting it wrong bills a tenant for work they never asked for. `Drained.Scanned` is exactly the distinction: false means the answer came back from a tag the ordinary hourly sweep had already paid for. Mutating the gate away so every outcome counts is caught. The fold accumulates per ACCOUNT across every app in a tick and writes ONCE at the end of it, CAS-merged the way UpsertEdgeDay already merges — not a write per object, and not per app. METER NAMES FREEZE AT MERGE, so they are chosen here rather than inherited from whatever the first surface happened to call them: `scan_objects` counts engine scans, `scan_bytes` keeps the observed byte count without an early conversion that would round small scans away. Both are recorded and explicitly unpriced, which is the shape MeterBackhaulGB already set. The depth surfaces are named for what they COUNT. Per app it is items; deployment-wide it is `queue_apps_with_work`, because that number counts apps rather than items AND an app's entry survives one extra tick after its last item drains — a consumed page reports "not empty", so the entry goes on the first tick that lists nothing. Called a queue length it would read as off by one for one tick and somebody would "fix" the drain to match. One existing test was renamed, not weakened: the assertions in TestScanStatusReportsTheDeploymentSwitch are untouched, it is now TestStoreScanStatusReportsDeploymentState to match the name the plan gives it. TWO TESTS ARE MISSING ON PURPOSE. TestPrintScanStatusShowsQueueDepth and TestPrintSystemAntivirusStatusShowsQueueDepth need printer lines, and a printer line is a sentence a person reads. This project reviews user-facing language with a human before it ships, so the implementer stopped and said which sentences it would have written rather than writing them. That is the right instinct and the reason it was told to. They land with the docs, after the copy is approved.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
10 files changed,
+295
−11
apimode.go+7 −5internal/console/apiscan.go+5 −0internal/console/apiscan_test.go+18 −1internal/console/apisystem.go+3 −2internal/console/apisystem_test.go+14 −0internal/console/scandrain.go+17 −3internal/console/scandrain_test.go+77 −0internal/usage/scan.go+77 −0internal/usage/scan_test.go+72 −0internal/usage/usage.go+5 −0
CI passed — run details
diff --git a/apimode.go b/apimode.go| index e937fb90e..1709fd2a3 100644 |
| --- a/apimode.go |
| +++ b/apimode.go |
| @@ -1629,11 +1629,12 @@ type apiStoreReply struct { |
| // with the engine unreachable, and an operator debugging "nothing is scanning" |
| // needs to know which half is missing. |
| type apiSystemAntivirusReply struct { |
| - Wired bool `json:"wired"` |
| - Enabled bool `json:"enabled"` |
| - Engine string `json:"engine,omitempty"` // the engine the RECORD names |
| - At string `json:"at,omitempty"` |
| - By string `json:"by,omitempty"` |
| + Wired bool `json:"wired"` |
| + Enabled bool `json:"enabled"` |
| + QueueAppsWithWork int `json:"queue_apps_with_work"` |
| + Engine string `json:"engine,omitempty"` // the engine the RECORD names |
| + At string `json:"at,omitempty"` |
| + By string `json:"by,omitempty"` |
| // EngineWired is what the wired engine calls itself, which can differ from |
| // the record's name — that difference is engine drift, and it is why the |
| // two are separate fields rather than one. |
| @@ -1769,6 +1770,7 @@ type apiScanReply struct { |
| MaxSignatureAgeSeconds int64 `json:"max_signature_age_seconds"` |
| Scanning *apiScanBlock `json:"scanning"` |
| Coverage *apiScanCoverage `json:"coverage"` |
| + QueueDepth int `json:"queue_depth"` |
| } |
| // scanCall is every scan door's round trip: they all answer the same shape, so |
diff --git a/internal/console/apiscan.go b/internal/console/apiscan.go| index 57bad93e4..1815cf395 100644 |
| --- a/internal/console/apiscan.go |
| +++ b/internal/console/apiscan.go |
| @@ -107,12 +107,17 @@ func (s *Server) scanReply(ctx context.Context, acct *Account, app string, rec * |
| // fail the whole door. |
| log.Printf("console: api scan status %s/%s: sidecar: %v", acct.Sqid, app, err) |
| } |
| + queueDepth, err := objstore.QueueDepth(ctx, s.cfg.Releases, acct.Sqid, app) |
| + if err != nil { |
| + return nil, err |
| + } |
| out := map[string]any{ |
| "app": app, |
| "backend": s.cfg.StoreDeployment.Backend, |
| "scanning": rec.Scanning, |
| "coverage": side, |
| "max_signature_age_seconds": int64(s.scanMaxSigAge() / time.Second), |
| + "queue_depth": queueDepth, |
| } |
| // THE DEPLOYMENT SWITCH IS PART OF THE ANSWER, not context somebody else |
| // holds. A per-app record saying `enabled: clamav` while the deployment is |
diff --git a/internal/console/apiscan_test.go b/internal/console/apiscan_test.go| index 76eeba6c5..ee5b1994e 100644 |
| --- a/internal/console/apiscan_test.go |
| +++ b/internal/console/apiscan_test.go |
| @@ -503,7 +503,7 @@ func disableDeploymentScanning(t *testing.T, fx *consoleFixture) { |
| // this reply, so omitting the deployment half is a new way for the status |
| // output to lie — the same lie the engine_error branch already refuses to tell |
| // about the wiring half. |
| -func TestScanStatusReportsTheDeploymentSwitch(t *testing.T) { |
| +func TestStoreScanStatusReportsDeploymentState(t *testing.T) { |
| ctx := context.Background() |
| fx, _, token := scanArmed(t, "code-scan-depstate") |
| requestScanning(t, fx, token, false) |
| @@ -547,3 +547,20 @@ func TestScanStatusReportsTheDeploymentSwitch(t *testing.T) { |
| t.Fatal("the per-app record vanished from the reply") |
| } |
| } |
| + |
| +func TestStoreScanStatusReportsQueueDepth(t *testing.T) { |
| + fx, _, token := scanArmed(t, "code-scan-queue-depth") |
| + requestScanning(t, fx, token, false) |
| + if _, err := objstore.Enqueue(context.Background(), fx.rel, fx.srv.scanQueue, |
| + fixtureAcct, "titogo", &objstore.QueueItem{Key: "uploads/a.pdf", By: "test"}, time.Now()); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + status, out := getJSON(t, scanURL(fx, "titogo"), token) |
| + if status != http.StatusOK { |
| + t.Fatalf("status = %d %v", status, out) |
| + } |
| + if got := int(out["queue_depth"].(float64)); got != 1 { |
| + t.Fatalf("queue_depth = %d, want 1", got) |
| + } |
| +} |
diff --git a/internal/console/apisystem.go b/internal/console/apisystem.go| index faf97b79b..1838ce1cb 100644 |
| --- a/internal/console/apisystem.go |
| +++ b/internal/console/apisystem.go |
| @@ -153,8 +153,9 @@ func (s *Server) rotateWebhookBudget(ctx context.Context, pub *ecdh.PublicKey) ( |
| // how an operator ends up debugging the wrong half. |
| func (s *Server) systemAntivirusReply(ctx context.Context) map[string]any { |
| out := map[string]any{ |
| - "wired": s.cfg.StoreScanner != nil, |
| - "enabled": false, |
| + "wired": s.cfg.StoreScanner != nil, |
| + "enabled": false, |
| + "queue_apps_with_work": s.ScanQueueDepth(), |
| } |
| av, _, err := release.GetAntivirus(ctx, s.cfg.Releases) |
| if err != nil { |
diff --git a/internal/console/apisystem_test.go b/internal/console/apisystem_test.go| index b4412fdf7..6b700cc96 100644 |
| --- a/internal/console/apisystem_test.go |
| +++ b/internal/console/apisystem_test.go |
| @@ -102,6 +102,20 @@ func TestSystemAntivirusStatusSeparatesWiringFromPolicy(t *testing.T) { |
| } |
| } |
| +func TestSystemAntivirusStatusReportsQueueAppsWithWork(t *testing.T) { |
| + fx, token := systemArmed(t, "code-sys-queue-depth") |
| + fx.srv.scanQueue.Bump(objstore.AppRef{Account: fixtureAcct, App: fixtureApp}) |
| + fx.srv.scanQueue.Bump(objstore.AppRef{Account: fixtureAcct, App: "second"}) |
| + |
| + status, out := getJSON(t, fx.ts.URL+systemAntivirusURL, token) |
| + if status != http.StatusOK { |
| + t.Fatalf("status = %d %v", status, out) |
| + } |
| + if got := int(out["queue_apps_with_work"].(float64)); got != 2 { |
| + t.Fatalf("queue_apps_with_work = %d, want 2", got) |
| + } |
| +} |
| + |
| // The record is the audit trail: who turned it on, and when. |
| func TestSystemAntivirusRecordsWhoAndWhen(t *testing.T) { |
| fx, token := systemArmed(t, "code-sys-audit") |
diff --git a/internal/console/scandrain.go b/internal/console/scandrain.go| index 82cff6ff9..fda6c4a4f 100644 |
| --- a/internal/console/scandrain.go |
| +++ b/internal/console/scandrain.go |
| @@ -22,6 +22,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/objstore" |
| + "github.com/carlosframework/platform/internal/usage" |
| ) |
| // scanDrainInterval is the tick. 30s, against the sweep's hour: this path |
| @@ -177,13 +178,20 @@ func (s *Server) scanDrainTick(ctx context.Context) error { |
| return nil |
| } |
| now := time.Now() |
| + deltas := map[string]usage.ScanRow{} |
| for _, ref := range s.scanQueue.Apps() { |
| if err := ctx.Err(); err != nil { |
| return err |
| } |
| - s.drainScanApp(ctx, ref, now) |
| + delta := s.drainScanApp(ctx, ref, now) |
| + if delta.Objects != 0 { |
| + row := deltas[ref.Account] |
| + row.Objects += delta.Objects |
| + row.Bytes += delta.Bytes |
| + deltas[ref.Account] = row |
| + } |
| } |
| - return nil |
| + return usage.UpsertScanDay(ctx, s.cfg.Releases, now, deltas) |
| } |
| // drainScanApp drains one app and tells whoever is waiting. |
| @@ -191,7 +199,8 @@ func (s *Server) scanDrainTick(ctx context.Context) error { |
| // 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) { |
| +func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time.Time) usage.ScanRow { |
| + var delta usage.ScanRow |
| // ONE APP'S TURN IS BOUNDED. clamd's own read timeout is minutes, and the |
| // object body stream carries no deadline of its own, so fifty slow objects |
| // belonging to one tenant could hold the whole deployment's drain for |
| @@ -213,6 +222,10 @@ func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time |
| // error path would make a long-poll wait out its full window for an answer |
| // that already exists. |
| for _, d := range done { |
| + if d.Scanned { |
| + delta.Objects++ |
| + delta.Bytes += d.Size |
| + } |
| s.scanNotify(ref, d.Key) |
| for _, path := range d.CallbackPaths { |
| if startRec != nil && startRec.Scanning != nil { |
| @@ -237,6 +250,7 @@ func (s *Server) drainScanApp(ctx context.Context, ref objstore.AppRef, now time |
| log.Printf("console: scan drain: %s/%s: %v", ref.Account, ref.App, err) |
| } |
| s.collectScanResults(ctx, ref.Account, ref.App, now) |
| + return delta |
| } |
| // collectScanResults ages out an app's fallback result objects. |
diff --git a/internal/console/scandrain_test.go b/internal/console/scandrain_test.go| index 71b230755..c230d07d7 100644 |
| --- a/internal/console/scandrain_test.go |
| +++ b/internal/console/scandrain_test.go |
| @@ -2,16 +2,19 @@ package console |
| import ( |
| "context" |
| + "encoding/json" |
| "errors" |
| "fmt" |
| "os" |
| "path/filepath" |
| + "strings" |
| "sync/atomic" |
| "testing" |
| "time" |
| "github.com/carlosframework/platform/internal/objstore" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/usage" |
| ) |
| // drainFixture is a console whose release store is a PLAIN DirStore. |
| @@ -90,6 +93,18 @@ type countingStore struct { |
| lists atomic.Int64 |
| } |
| +type scanUsageCountingStore struct { |
| + *release.DirStore |
| + puts atomic.Int64 |
| +} |
| + |
| +func (s *scanUsageCountingStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if strings.HasPrefix(key, usage.ScanPrefix) { |
| + s.puts.Add(1) |
| + } |
| + return s.DirStore.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| func (c *countingStore) List(ctx context.Context, prefix string) ([]string, error) { |
| c.lists.Add(1) |
| return c.DirStore.List(ctx, prefix) |
| @@ -658,3 +673,65 @@ func TestStartScanDrainIsInertWithoutAnEngine(t *testing.T) { |
| t.Fatal("StartScanDrain kept ticking on a deployment with no engine") |
| } |
| } |
| + |
| +func TestScanUsageCountsOnDemandOnly(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) |
| + |
| + tagged := objstore.ScanObject{Key: "uploads/tagged.pdf", ETag: "etag-tagged", Modified: mod, Size: 100} |
| + sc.PutObject(bucket, tagged.Key, tagged.ETag, tagged.Modified, tagged.Size) |
| + sc.Put(bucket, tagged.Key, tagged.Version(), objstore.VerdictClean) |
| + sc.PutObject(bucket, "uploads/new.pdf", "etag-new", mod, 25) |
| + enqueueFor(t, fx, fixtureAcct, fixtureApp, tagged.Key, mod) |
| + enqueueFor(t, fx, fixtureAcct, fixtureApp, "uploads/new.pdf", mod.Add(time.Millisecond)) |
| + |
| + if err := fx.srv.scanDrainTick(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + raw, _, err := fx.rel.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now()))) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var day usage.ScanDay |
| + if err := json.Unmarshal(raw, &day); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 1, Bytes: 25}) { |
| + t.Fatalf("usage = %+v, want only the newly scanned object", got) |
| + } |
| +} |
| + |
| +func TestScanUsageFoldsPerTick(t *testing.T) { |
| + ctx := context.Background() |
| + fx, sc, dir := drainFixture(t) |
| + counting := &scanUsageCountingStore{DirStore: release.NewDirStore(dir)} |
| + fx.rel = counting |
| + fx.srv.cfg.Releases = counting |
| + enableStore(t, fx, fixtureAcct, fixtureApp, true) |
| + enableStore(t, fx, fixtureAcct, "second", true) |
| + for app, key := range map[string]string{fixtureApp: "uploads/a.pdf", "second": "uploads/b.pdf"} { |
| + sc.PutObject(drainBucket(fixtureAcct, app), key, "etag-1", time.Now(), 10) |
| + enqueueFor(t, fx, fixtureAcct, app, key, time.Now()) |
| + } |
| + |
| + if err := fx.srv.scanDrainTick(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := counting.puts.Load(); got != 1 { |
| + t.Fatalf("usage writes = %d, want one for the whole tick", got) |
| + } |
| + raw, _, err := counting.Get(ctx, usage.ScanDayKey(usage.DayOf(time.Now()))) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var day usage.ScanDay |
| + if err := json.Unmarshal(raw, &day); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := day.Accounts[fixtureAcct]; got != (usage.ScanRow{Objects: 2, Bytes: 20}) { |
| + t.Fatalf("folded usage = %+v", got) |
| + } |
| +} |
diff --git a/internal/usage/scan.go b/internal/usage/scan.go| new file mode 100644 |
| index 000000000..5de0c2018 |
| --- /dev/null |
| +++ b/internal/usage/scan.go |
| @@ -0,0 +1,77 @@ |
| +package usage |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "os" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// ScanRow is one account's on-demand scanning work for a UTC day. |
| +type ScanRow struct { |
| + Objects int64 `json:"objects"` |
| + Bytes int64 `json:"bytes"` |
| +} |
| + |
| +// ScanDay is usage/scan/<date>.json, keyed by account. |
| +type ScanDay struct { |
| + V int `json:"v"` // 1 |
| + Date string `json:"date"` |
| + Accounts map[string]ScanRow `json:"accounts"` |
| +} |
| + |
| +func ScanDayKey(date string) string { return "usage/scan/" + date + ".json" } |
| + |
| +const ScanPrefix = "usage/scan/" |
| + |
| +func (d *ScanDay) AddDeltas(deltas map[string]ScanRow) { |
| + if d.Accounts == nil { |
| + d.Accounts = map[string]ScanRow{} |
| + } |
| + for account, delta := range deltas { |
| + row := d.Accounts[account] |
| + row.Objects += delta.Objects |
| + row.Bytes += delta.Bytes |
| + d.Accounts[account] = row |
| + } |
| +} |
| + |
| +// UpsertScanDay folds one drain tick into its UTC day. Empty ticks write |
| +// nothing; a compare-and-swap loss re-reads before adding so concurrent |
| +// consoles cannot silently replace each other's counts. |
| +func UpsertScanDay(ctx context.Context, s release.Store, now time.Time, deltas map[string]ScanRow) error { |
| + if len(deltas) == 0 { |
| + return nil |
| + } |
| + date := DayOf(now) |
| + key := ScanDayKey(date) |
| + for attempt := 0; attempt < 3; attempt++ { |
| + day := ScanDay{V: 1, Date: date} |
| + raw, etag, err := s.Get(ctx, key) |
| + switch { |
| + case err == nil: |
| + if err := json.Unmarshal(raw, &day); err != nil { |
| + return fmt.Errorf("usage: parse %s: %w", key, err) |
| + } |
| + case errors.Is(err, os.ErrNotExist): |
| + etag = "" |
| + default: |
| + return err |
| + } |
| + day.AddDeltas(deltas) |
| + out, err := json.Marshal(day) |
| + if err != nil { |
| + return err |
| + } |
| + if _, err := s.Put(ctx, key, out, etag); err == nil { |
| + return nil |
| + } else if !errors.Is(err, release.ErrCASConflict) { |
| + return err |
| + } |
| + } |
| + return fmt.Errorf("usage: %s: lost CAS race three times", key) |
| +} |
diff --git a/internal/usage/scan_test.go b/internal/usage/scan_test.go| new file mode 100644 |
| index 000000000..7332e1034 |
| --- /dev/null |
| +++ b/internal/usage/scan_test.go |
| @@ -0,0 +1,72 @@ |
| +package usage |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +func TestUpsertScanDayUsesUTCDate(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + ctx := context.Background() |
| + at := time.Date(2026, 9, 8, 23, 59, 0, 0, time.UTC) |
| + |
| + if err := UpsertScanDay(ctx, s, at, map[string]ScanRow{ |
| + "bab": {Objects: 2, Bytes: 30}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := UpsertScanDay(ctx, s, at.Add(time.Minute), map[string]ScanRow{ |
| + "bab": {Objects: 1, Bytes: 7}, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + for date, want := range map[string]ScanRow{ |
| + "2026-09-08": {Objects: 2, Bytes: 30}, |
| + "2026-09-09": {Objects: 1, Bytes: 7}, |
| + } { |
| + raw, _, err := s.Get(ctx, ScanDayKey(date)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var day ScanDay |
| + if err := json.Unmarshal(raw, &day); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := day.Accounts["bab"]; got != want { |
| + t.Fatalf("%s row = %+v, want %+v", date, got, want) |
| + } |
| + } |
| +} |
| + |
| +func TestUpsertScanDayMergesAccountsAndTicks(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + ctx := context.Background() |
| + at := time.Date(2026, 9, 8, 10, 0, 0, 0, time.UTC) |
| + for _, deltas := range []map[string]ScanRow{ |
| + {"bab": {Objects: 2, Bytes: 20}}, |
| + {"bab": {Objects: 1, Bytes: 5}, "ops": {Objects: 3, Bytes: 30}}, |
| + } { |
| + if err := UpsertScanDay(ctx, s, at, deltas); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + raw, _, err := s.Get(ctx, ScanDayKey("2026-09-08")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var day ScanDay |
| + if err := json.Unmarshal(raw, &day); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := day.Accounts["bab"]; got != (ScanRow{Objects: 3, Bytes: 25}) { |
| + t.Fatalf("bab = %+v", got) |
| + } |
| + if got := day.Accounts["ops"]; got != (ScanRow{Objects: 3, Bytes: 30}) { |
| + t.Fatalf("ops = %+v", got) |
| + } |
| +} |
diff --git a/internal/usage/usage.go b/internal/usage/usage.go| index dd0ee44a9..e00b6bf86 100644 |
| --- a/internal/usage/usage.go |
| +++ b/internal/usage/usage.go |
| @@ -36,6 +36,11 @@ const ( |
| // requests that crossed a region before being served. Informational, |
| // unpriced, same posture. |
| MeterBackhaulRequests1k = "backhaul_requests_1k" |
| + // Scan meters stay in the units the scanner observes. Converting bytes to |
| + // decimal gigabytes at collection time would discard small scans forever, |
| + // and neither meter is priced today. |
| + MeterScanObjects = "scan_objects" |
| + MeterScanBytes = "scan_bytes" |
| ) |
| // DayOf is the UTC date label used in every usage key. |