amadan

carlos / platform Public

console: the scan status door, long-polling and honest

Task 13, spec §5.2. `GET .../store/scan/object?key=&wait=` answers what
is known about one object: clean, infected, skipped, failed, denied,
pending, gone, or unavailable.

THE TAG FIRST, THEN THE RESULT OBJECT, and the order is not a
preference. A result object exists only because a verdict could not be
written onto the object; if the object now carries one, it is both more
recent and authoritative, and the drain deletes the result the moment
that happens. The test stages a case where the two DISAGREE, which is
the only arrangement in which the order is observable at all.

FAILED AND DENIED ARE REPORTED AS THEMSELVES. The landed tag reader maps
both back to unknown so the sweep retries them — right as retry policy,
wrong as an answer to a question about an object. Applied here, a failed
scan would read as pending forever: the drain deletes the item as
handled, and nothing changes the answer until a sweep happens to re-fail
it.

The version comparison never leaves the console. An app that
reconstructed `etag@lastmodified` and compared it wrongly gets the
laundering bug as a correctness bug — upload v2 over a key whose v1 was
tagged clean, read the stale tag, serve the payload.

Waiters are capped per app AND globally, and the per-app cap is not
implied by the global one: without it a single tenant fills the budget
and every other tenant's wait degrades to polling. Over the cap nothing
is refused — the read still happens and still answers truthfully, so a
full console degrades to polling rather than to errors. An over-long
`wait` is clamped rather than rejected, because a client asking for five
minutes wants the longest wait it can have.

TWO GAPS, RECORDED RATHER THAN PAPERED OVER, both also in the plan:

`at`, `engine` and `signature` are absent from the TAG answer. §5.2's
example shows them, but the `Status` seam that same section specifies
returns the verdict and the object and nothing else. The door reports
provenance where it genuinely has it — a result object carries all three
— and says nothing in the tag branch, because the only value available
there is this deployment's CURRENT engine, which may not be the one that
judged the object. That would be a guess dressed as a record, in the one
feature whose entire purpose is not claiming coverage it does not have.
Closing it means widening a reviewed interface, which is its own task.

The register-before-read ordering is not asserted: catching the wrong
order needs a drain to land between two adjacent statements and there is
no seam to hold it there. Losing that race costs one expired wait, never
a wrong answer — the door re-reads on its timer and never answers from
the notification itself.

Eleven mutations, all 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 ba07db58f16d2dc93263cce62d9624a4f96c82ed parent 8105c88
6 files changed, +769 −21
  • docs/superpowers/plans/2026-09-02-scan-on-demand.md +50 −11
  • internal/console/apiscanstatus.go +239 −0
  • internal/console/apiscanstatus_test.go +430 −0
  • internal/console/authcli.go +3 −0
  • internal/console/scandrain.go +35 −7
  • internal/console/scandrain_test.go +12 −3

CI passed — run details

diff --git a/docs/superpowers/plans/2026-09-02-scan-on-demand.md b/docs/superpowers/plans/2026-09-02-scan-on-demand.md
index 7019e2d25..85ff870d0 100644
--- a/docs/superpowers/plans/2026-09-02-scan-on-demand.md
+++ b/docs/superpowers/plans/2026-09-02-scan-on-demand.md
@@ -426,19 +426,19 @@ thing that collects a stranded item, so it is not optional" and task 9's
**Files:** `internal/console/scandrain.go` (new) + test.
-- [ ] `TestBootRebuildFindsStrandedItem` — an item written with no index entry
+- [x] `TestBootRebuildFindsStrandedItem` — an item written with no index entry
is drained after a rebuild.
-- [ ] `TestBootRebuildCollectsExpiredItemsAndResults` — both prefixes, 24h TTL.
-- [ ] `TestDrainRunsWhileSweepIsBlocked` — **observable**, not nominal:
+- [x] `TestBootRebuildCollectsExpiredItemsAndResults` — both prefixes, 24h TTL.
+- [x] `TestDrainRunsWhileSweepIsBlocked` — **observable**, not nominal:
`TestDrainIsItsOwnGoroutine` as named would pass against anything. Park
`scanPass` in a blocking lister and assert an item still drains. `scanPass`
sweeps apps sequentially for a full budget each (`scanpass.go:86-103`), so a
queue phase behind it starves for up to N × 60s.
-- [ ] `TestDrainRoundRobinFairness` — one app's flood does not delay another's
+- [x] `TestDrainRoundRobinFairness` — one app's flood does not delay another's
item beyond one tick.
-- [ ] `TestDrainAgeSweepsResultObjects` — per visited app, so a console that
+- [x] `TestDrainAgeSweepsResultObjects` — per visited app, so a console that
never restarts does not accumulate them.
-- [ ] `TestQueueDepthReported` — §4.4 and §10 both want it; **it is delivered
+- [x] `TestQueueDepthReported` — §4.4 and §10 both want it; **it is delivered
here** and surfaced in task 17. Task 6 lands before this and cannot surface
it.
@@ -449,22 +449,61 @@ waiters, deleting the result object when a later scan records a real verdict.
---
+**Landed** as `console: the scan drain, its own goroutine and its own clock`.
+
+Two lines in the tick are documented as NOT being the guarantee they look
+like, because mutation testing showed neither changes anything observable: the
+deployment-switch check and the engine check are both about avoiding work,
+since `DrainApp` re-reads consent and refuses on its own. The first is asserted
+the only way it can be — by counting listings — and the second is marked
+unassertable in the comment rather than left looking tested.
+
+`TestDrainRoundRobinFairness` also pinned down something the spec left
+implicit: an app's index entry is released on the first tick that lists
+NOTHING, not on the tick that empties it, because a consumed page reports "not
+empty". Task 17's queue-depth surface should say "apps with work" rather than
+"apps with items", or it will read as off-by-one for one tick.
+
+---
+
## Task 13: the status door and long-poll
**Files:** `internal/console/apiscan.go` + test.
-- [ ] `TestStatusDoorReadsTagThenResultObject`.
-- [ ] `TestStatusDoorRefusesStaleResultObject` — the door-level half of task
+- [x] `TestStatusDoorReadsTagThenResultObject`.
+- [x] `TestStatusDoorRefusesStaleResultObject` — the door-level half of task
9's unit test.
-- [ ] `TestStatusDoorReportsFailedAndDenied` — as themselves, never `pending`.
-- [ ] `TestStatusDoorGoneForMissingObject`.
-- [ ] `TestLongPollWakesOnDrain`, `TestLongPollCapsWaiters` — global **and**
+- [x] `TestStatusDoorReportsFailedAndDenied` — as themselves, never `pending`.
+- [x] `TestStatusDoorGoneForMissingObject`.
+- [x] `TestLongPollWakesOnDrain`, `TestLongPollCapsWaiters` — global **and**
per-app.
**Commit:** `console: the scan status door, long-polling and honest`
---
+**Landed** as `console: the scan status door, long-polling and honest`.
+Eleven mutations, all caught. Two gaps are recorded rather than papered over:
+
+- [ ] **`at`, `engine` and `signature` are missing from the TAG answer.** §5.2's
+ example shows `{"state":"clean","at":"...","engine":"clamav","signature":"..."}`,
+ but the `Scanner.Status` seam §5.2 itself specifies returns the verdict and
+ the object and nothing else. The door reports provenance where it genuinely
+ has it — a result object records all three — and says NOTHING in the tag
+ branch, because the only value available there is this deployment's CURRENT
+ engine, which may not be the one that judged the object. That would be a
+ guess dressed as a record, in the one feature whose whole purpose is not
+ reporting coverage it does not have. Closing this means widening `Status` to
+ return the tag's provenance, which is a change to a reviewed interface and
+ belongs in its own task rather than smuggled into this one.
+- The register-before-read ordering in the long-poll is not asserted: catching
+ the wrong order needs a drain to land between two adjacent statements and
+ there is no seam to hold it there. Noted in the code. Losing that race costs
+ one expired wait, never a wrong answer — the door re-reads on its timer and
+ never answers from the notification itself.
+
+---
+
## Task 14: callback signing (pure)
**Files:** `internal/console/scanhook.go` + test.
diff --git a/internal/console/apiscanstatus.go b/internal/console/apiscanstatus.go
new file mode 100644
index 000000000..96e0656a3
--- /dev/null
+++ b/internal/console/apiscanstatus.go
@@ -0,0 +1,239 @@
+// The on-demand READ door (spec §5.2): what is known about one object right
+// now, optionally waiting a little for it.
+//
+// Its own file rather than more of apiscan.go, because the scan family there
+// is about POLICY — who may scan, whether enforcement is on — and this one
+// answers a question about a single object for the app that owns it. They
+// share the consent chain and nothing else.
+package console
+
+import (
+ "context"
+ "errors"
+ "log"
+ "net/http"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/objstore"
+)
+
+// scanWaitMax is the long-poll ceiling. Beyond this a client should be
+// polling: a request parked longer than half a minute is holding a console
+// goroutine against an answer that may never come in this process at all.
+const scanWaitMax = 30 * time.Second
+
+// scanWaitersPerApp and scanWaitersGlobal cap parked requests.
+//
+// BOTH, and the per-app one is not implied by the global one: without it a
+// single tenant fills the global budget and every other tenant's wait
+// degrades to polling. Without the GLOBAL one, enough tenants at their
+// per-app limit still add up to more parked goroutines than a console should
+// hold. Neither is a refusal — over the cap the door answers `pending`
+// immediately, which is TRUE, and the client polls.
+const (
+ scanWaitersPerApp = 32
+ scanWaitersGlobal = 256
+)
+
+// scanStatusOf reads what is known about one object, in the order §5.2 sets:
+// the object's own tag first, the fallback result second.
+//
+// THE TAG WINS, and the order is not a preference. A result object exists only
+// because a verdict could not be written onto the object; if the object now
+// carries one, the object is the more recent and more authoritative answer,
+// and the drain deletes the result the moment that happens.
+//
+// The version comparison happens inside the scanner and inside
+// ReadScanResult. It never reaches this door and never reaches an app: the
+// token is an internal format, and an app comparing it wrongly gets the
+// laundering bug as a correctness bug — upload v2 over a key whose v1 was
+// tagged clean, read the stale tag, serve the payload.
+func (s *Server) scanStatusOf(ctx context.Context, acct, app, bucket, key string) (map[string]string, error) {
+ verdict, obj, err := s.cfg.StoreScanner.Status(ctx, bucket, key)
+ if errors.Is(err, objstore.ErrObjectGone) {
+ return map[string]string{"state": "gone"}, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ if verdict != objstore.VerdictUnknown {
+ // REPORTED AS THEMSELVES. `failed` and `denied` are verdicts, not
+ // absences: the landed tag reader maps them back to unknown so the
+ // SWEEP retries them, which is right as retry policy and wrong as an
+ // answer to a question about an object. Left applied here, a failed
+ // scan would read as `pending` forever — the drain deletes the item
+ // as handled, and nothing changes the answer until a sweep re-fails
+ // it.
+ return map[string]string{"state": string(verdict)}, nil
+ }
+ // No usable tag. The object may still have an answer that could not be
+ // written onto it.
+ if obj == (objstore.ScanObject{}) {
+ // A denied HeadObject: no version, so no result object can be matched
+ // against it either. Status already said what this is.
+ return map[string]string{"state": "pending"}, nil
+ }
+ res, err := objstore.ReadScanResult(ctx, s.cfg.Releases, acct, app, key, obj.Version())
+ if err != nil {
+ return nil, err
+ }
+ if res == nil {
+ // Absent, or recorded against different bytes — ReadScanResult makes
+ // no distinction and neither should this door. Nothing is known about
+ // the object that is there NOW.
+ return map[string]string{"state": "pending"}, nil
+ }
+ out := map[string]string{"state": res.Verdict}
+ // Provenance, but only where it is actually known. A result object records
+ // the engine and signature that judged it; a TAG, through the Status seam
+ // as it stands, yields the verdict alone. Reporting this deployment's
+ // current engine for a tag would be a guess dressed as a record — the
+ // engine may have been replaced since — so the tag branch above says
+ // nothing rather than something plausible.
+ if res.Engine != "" {
+ out["engine"] = res.Engine
+ }
+ if res.Signature != "" {
+ out["signature"] = res.Signature
+ }
+ if !res.At.IsZero() {
+ out["at"] = res.At.UTC().Format(time.RFC3339)
+ }
+ // The verdict is on a result object rather than the object itself, which
+ // the app has to know: an infected object answered this way is NOT covered
+ // by the enforcement Deny, because that matches on a tag that is not
+ // there.
+ out["recorded"] = "false"
+ return out, nil
+}
+
+// handleAPIScanObject answers GET .../store/scan/object?key=&wait=.
+//
+// `skipped`, `gone` and `unavailable` are NOT `clean`, and they are distinct
+// states rather than an error field, because an app treating any of them as
+// success reproduces exactly the dishonesty the whole feature exists to
+// prevent: reporting coverage that does not exist.
+func (s *Server) handleAPIScanObject(w http.ResponseWriter, r *http.Request) {
+ _, acct, app, ok := s.apiAccountApp(w, r)
+ if !ok {
+ return
+ }
+ key := strings.TrimSpace(r.URL.Query().Get("key"))
+ if key == "" {
+ jsonError(w, http.StatusBadRequest, `name the object, e.g. ?key=uploads/x.pdf`)
+ return
+ }
+ wait, err := scanWaitFor(r.URL.Query().Get("wait"))
+ if err != nil {
+ jsonError(w, http.StatusBadRequest, err.Error())
+ return
+ }
+ if s.cfg.Releases == nil {
+ scanNoEngine(w)
+ return
+ }
+ ctx, cancel := context.WithTimeout(r.Context(), wait+storeTimeout)
+ defer cancel()
+
+ rec, refusal, err := objstore.QueuePermitted(ctx, s.cfg.Releases, acct.Sqid, app)
+ if err != nil {
+ log.Printf("console: api scan object %s/%s: consent: %v", acct.Sqid, app, err)
+ jsonError(w, http.StatusBadGateway, "the store record could not be read — try again")
+ return
+ }
+ // The SAME mapping the enqueue door uses, which is the point of it being a
+ // table: two doors that answered the same refusal differently would let an
+ // app conclude something about its store from the disagreement.
+ if ans := scanRefusalAnswer(refusal); ans.refused() {
+ if ans.unknown {
+ log.Printf("console: api scan object %s/%s: unhandled refusal %q", acct.Sqid, app, refusal)
+ }
+ if ans.notFound {
+ http.NotFound(w, r)
+ return
+ }
+ scanUnavailable(w, ans.status, string(refusal), ans.msg)
+ return
+ }
+ if s.cfg.StoreScanner == nil {
+ scanNoEngine(w)
+ return
+ }
+ bucket := rec.Provisioned.Bucket
+
+ // REGISTERED BEFORE THE FIRST READ. A drain that finishes between the read
+ // and the registration would wake nobody, and this request would then park
+ // for its whole window on an answer that had already landed.
+ //
+ // NOT ASSERTED BY ANY TEST, and said so rather than left looking covered:
+ // catching the wrong order needs the drain to land inside the gap between
+ // the two statements, and there is no seam here to hold it there. The
+ // window is real but small, and the cost of losing the race is one wait
+ // that expires rather than a wrong answer — the door re-reads the object
+ // when its timer fires and never answers from the notification itself.
+ var woken <-chan struct{}
+ if wait > 0 {
+ ch, release, ok := s.scanWait(objstore.AppRef{Account: acct.Sqid, App: app}, key)
+ if ok {
+ defer release()
+ woken = ch
+ }
+ // Not ok means the caps are full. Nothing is refused: the read below
+ // still happens and still answers truthfully; only the WAITING is
+ // skipped.
+ }
+
+ deadline := time.NewTimer(wait)
+ defer deadline.Stop()
+ for {
+ out, err := s.scanStatusOf(ctx, acct.Sqid, app, bucket, key)
+ if err != nil {
+ log.Printf("console: api scan object %s/%s %q: %v", acct.Sqid, app, key, err)
+ jsonError(w, http.StatusBadGateway, "the verdict could not be read — try again")
+ return
+ }
+ if out["state"] != "pending" || woken == nil {
+ writeJSON(w, http.StatusOK, out)
+ return
+ }
+ select {
+ case <-woken:
+ // Round again and re-read. The wake says "something happened",
+ // never what — the answer comes from the object, not from the
+ // notification, so a spurious wake costs one read and nothing
+ // else.
+ case <-deadline.C:
+ writeJSON(w, http.StatusOK, out)
+ return
+ case <-ctx.Done():
+ writeJSON(w, http.StatusOK, out)
+ return
+ }
+ }
+}
+
+// scanWaitFor parses the wait parameter and bounds it.
+//
+// An over-long wait is CLAMPED rather than refused: a client asking for five
+// minutes wants the longest wait it can have, and answering 400 turns a
+// reasonable request into an error the app author has to special-case. A
+// malformed one is a 400, because that is a bug in the caller rather than an
+// ambitious number.
+func scanWaitFor(raw string) (time.Duration, error) {
+ raw = strings.TrimSpace(raw)
+ if raw == "" {
+ return 0, nil
+ }
+ d, err := time.ParseDuration(raw)
+ if err != nil {
+ return 0, errors.New(`wait must be a duration such as "15s"`)
+ }
+ if d < 0 {
+ return 0, errors.New("wait cannot be negative")
+ }
+ if d > scanWaitMax {
+ return scanWaitMax, nil
+ }
+ return d, nil
+}
diff --git a/internal/console/apiscanstatus_test.go b/internal/console/apiscanstatus_test.go
new file mode 100644
index 000000000..8164a5d86
--- /dev/null
+++ b/internal/console/apiscanstatus_test.go
@@ -0,0 +1,430 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "net/http"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/objstore"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// statusArmed is enqueueArmed plus a handle on the engine, because every test
+// here has to stage what the object's tag says.
+func statusArmed(t *testing.T, code string) (*consoleFixture, *objstore.MemScanner, string, string) {
+ t.Helper()
+ fx, _, token, _ := storeArmed(t, code)
+ grantStore(t, fx, token)
+ requestScanning(t, fx, token, false)
+ sc := objstore.NewMemScanner()
+ fx.srv.cfg.StoreScanner = sc
+ enableDeploymentScanning(t, fx)
+ fx.srv.cfg.StoreOperators = []string{"alice@keymail.dev"}
+ if s, o := postJSON(t, scanURL(fx, "titogo")+"/grant", map[string]any{}, token); s != http.StatusOK {
+ t.Fatalf("scan grant = %d %v", s, o)
+ }
+ rec, _, err := objstore.Get(context.Background(), fx.rel, release.OpsAccount, "titogo")
+ if err != nil || rec == nil || rec.Provisioned == nil {
+ t.Fatalf("no provisioned record to read a bucket from: %v", err)
+ }
+ return fx, sc, appToken(t, fx, "titobot"), rec.Provisioned.Bucket
+}
+
+// getStatus asks the door about one key and decodes the answer.
+func getStatus(t *testing.T, fx *consoleFixture, tok, query string) (int, map[string]any) {
+ t.Helper()
+ status, _, body := rawReq(t, http.MethodGet, enqueueURL(fx, "titogo")+"?"+query, tok, "")
+ out := map[string]any{}
+ if body != "" {
+ if err := json.Unmarshal([]byte(body), &out); err != nil {
+ t.Fatalf("decode %q: %v", body, err)
+ }
+ }
+ return status, out
+}
+
+// seedObject puts an object in the engine's world and returns its version, so
+// a test can stage a tag or a result against the same bytes.
+func seedObject(sc *objstore.MemScanner, bucket, key, etag string) string {
+ mod := time.Now().Truncate(time.Second)
+ sc.PutObject(bucket, key, etag, mod, 10)
+ return objstore.ScanObject{Key: key, ETag: etag, Modified: mod, Size: 10}.Version()
+}
+
+// TestStatusDoorReadsTagThenResultObject pins the ORDER, which is the whole of
+// §5.2's read rule: the object's own tag is the more recent and more
+// authoritative answer, and a result object exists only because a tag could
+// not be written.
+func TestStatusDoorReadsTagThenResultObject(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, tok, bucket := statusArmed(t, "code-status-order")
+
+ // (a) a tag and nothing else.
+ vTagged := seedObject(sc, bucket, "uploads/tagged.pdf", "etag-t")
+ sc.Put(bucket, "uploads/tagged.pdf", vTagged, objstore.VerdictClean)
+ if status, out := getStatus(t, fx, tok, "key=uploads/tagged.pdf"); status != http.StatusOK || out["state"] != "clean" {
+ t.Fatalf("tagged = %d %v, want 200 clean", status, out)
+ }
+
+ // (b) a result object and no tag.
+ vResult := seedObject(sc, bucket, "uploads/untaggable.pdf", "etag-u")
+ if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
+ Key: "uploads/untaggable.pdf", Verdict: "infected", Version: vResult,
+ Engine: "clamav", Signature: "Eicar-Test", At: time.Now().UTC(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ status, out := getStatus(t, fx, tok, "key=uploads/untaggable.pdf")
+ if status != http.StatusOK || out["state"] != "infected" {
+ t.Fatalf("result-object read = %d %v, want 200 infected", status, out)
+ }
+ if out["engine"] != "clamav" || out["signature"] != "Eicar-Test" {
+ t.Errorf("provenance missing from a result-object answer: %v", out)
+ }
+ // The app must be told the verdict is NOT on the object: an infected
+ // object answered this way is not covered by the enforcement Deny, which
+ // matches on a tag that is not there.
+ if out["recorded"] != "false" {
+ t.Errorf("recorded = %v, want %q — the app cannot tell this verdict is not enforceable", out["recorded"], "false")
+ }
+
+ // (c) BOTH, disagreeing. The tag wins.
+ vBoth := seedObject(sc, bucket, "uploads/both.pdf", "etag-b")
+ sc.Put(bucket, "uploads/both.pdf", vBoth, objstore.VerdictClean)
+ if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
+ Key: "uploads/both.pdf", Verdict: "infected", Version: vBoth,
+ Engine: "clamav", At: time.Now().UTC(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if status, out := getStatus(t, fx, tok, "key=uploads/both.pdf"); out["state"] != "clean" {
+ t.Fatalf("with both present = %d %v, want the TAG's answer (clean)", status, out)
+ }
+}
+
+// TestStatusDoorRefusesStaleResultObject is the laundering bug as a
+// correctness bug: v1 was judged, v2 was uploaded over it, and the recorded
+// answer is about bytes that are no longer there.
+func TestStatusDoorRefusesStaleResultObject(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, tok, bucket := statusArmed(t, "code-status-stale")
+
+ if err := objstore.WriteScanResult(ctx, fx.rel, release.OpsAccount, "titogo", &objstore.ScanResult{
+ Key: "uploads/x.pdf", Verdict: "clean", Version: "etag-v1@2020-01-01T00:00:00Z",
+ Engine: "clamav", At: time.Now().UTC(),
+ }); err != nil {
+ t.Fatal(err)
+ }
+ // The object now holds DIFFERENT bytes.
+ seedObject(sc, bucket, "uploads/x.pdf", "etag-v2")
+
+ status, out := getStatus(t, fx, tok, "key=uploads/x.pdf")
+ if status != http.StatusOK {
+ t.Fatalf("status = %d %v", status, out)
+ }
+ if out["state"] != "pending" {
+ t.Fatalf("state = %v, want pending — a verdict about replaced bytes is not an answer about this object", out["state"])
+ }
+}
+
+// TestStatusDoorReportsFailedAndDenied: as themselves, never as pending.
+//
+// The landed TAG READER maps both back to unknown so the sweep retries them,
+// which is right as retry policy and wrong as an answer to a question. If this
+// door applied that rule, a failed scan would read as pending forever — the
+// drain deletes the item as handled, and nothing changes the answer until a
+// sweep happens to re-fail it.
+func TestStatusDoorReportsFailedAndDenied(t *testing.T) {
+ fx, sc, tok, bucket := statusArmed(t, "code-status-fd")
+
+ v := seedObject(sc, bucket, "uploads/failed.pdf", "etag-f")
+ sc.Put(bucket, "uploads/failed.pdf", v, objstore.VerdictFailed)
+ if _, out := getStatus(t, fx, tok, "key=uploads/failed.pdf"); out["state"] != "failed" {
+ t.Errorf("state = %v, want failed", out["state"])
+ }
+
+ seedObject(sc, bucket, "uploads/denied.pdf", "etag-d")
+ sc.DenyHead(bucket, "uploads/denied.pdf")
+ if _, out := getStatus(t, fx, tok, "key=uploads/denied.pdf"); out["state"] != "denied" {
+ t.Errorf("state = %v, want denied", out["state"])
+ }
+
+ // And skipped is not clean. An app treating it as success reproduces the
+ // dishonesty the two-number rule exists to prevent.
+ vs := seedObject(sc, bucket, "uploads/huge.bin", "etag-h")
+ sc.Put(bucket, "uploads/huge.bin", vs, objstore.VerdictSkipped)
+ if _, out := getStatus(t, fx, tok, "key=uploads/huge.bin"); out["state"] != "skipped" {
+ t.Errorf("state = %v, want skipped", out["state"])
+ }
+}
+
+// TestStatusDoorGoneForMissingObject: an object that is not there is `gone`,
+// which is distinct from both `pending` and `clean`.
+func TestStatusDoorGoneForMissingObject(t *testing.T) {
+ fx, _, tok, _ := statusArmed(t, "code-status-gone")
+
+ status, out := getStatus(t, fx, tok, "key=uploads/never-existed.pdf")
+ if status != http.StatusOK || out["state"] != "gone" {
+ t.Fatalf("missing object = %d %v, want 200 gone", status, out)
+ }
+}
+
+// TestStatusDoorPendingForAnUnjudgedObject is the positive control for
+// everything above: without it, a door that answered `pending` to absolutely
+// everything would pass the stale test and the long-poll tests.
+func TestStatusDoorPendingForAnUnjudgedObject(t *testing.T) {
+ fx, sc, tok, bucket := statusArmed(t, "code-status-pending")
+ seedObject(sc, bucket, "uploads/new.pdf", "etag-n")
+
+ if _, out := getStatus(t, fx, tok, "key=uploads/new.pdf"); out["state"] != "pending" {
+ t.Fatalf("state = %v, want pending", out["state"])
+ }
+}
+
+// TestStatusDoorSharesTheConsentChain: the same table the enqueue door reads,
+// so the two doors cannot answer one refusal differently — a disagreement is
+// itself something an app could conclude from.
+func TestStatusDoorSharesTheConsentChain(t *testing.T) {
+ fx, _, member, _ := storeArmed(t, "code-status-consent")
+ grantStore(t, fx, member)
+ fx.srv.cfg.StoreScanner = objstore.NewMemScanner()
+ enableDeploymentScanning(t, fx)
+ tok := appToken(t, fx, "bot")
+
+ status, out := getStatus(t, fx, tok, "key=uploads/x.pdf")
+ if status != http.StatusConflict || out["reason"] != "not_requested" {
+ t.Fatalf("status door = %d %v, want the enqueue door's 409 not_requested", status, out)
+ }
+ if out["state"] != "unavailable" {
+ t.Errorf("state = %v, want unavailable", out["state"])
+ }
+}
+
+// TestStatusDoorNeedsAKey and rejects a malformed wait.
+func TestStatusDoorArgumentErrors(t *testing.T) {
+ fx, _, tok, _ := statusArmed(t, "code-status-args")
+
+ if status, _ := getStatus(t, fx, tok, ""); status != http.StatusBadRequest {
+ t.Errorf("no key = %d, want 400", status)
+ }
+ if status, _ := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=soon"); status != http.StatusBadRequest {
+ t.Errorf("malformed wait = %d, want 400", status)
+ }
+ if status, _ := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=-5s"); status != http.StatusBadRequest {
+ t.Errorf("negative wait = %d, want 400", status)
+ }
+}
+
+// TestScanWaitForClamps: an over-long wait is the client asking for the
+// longest it can have, not an error to special-case.
+func TestScanWaitForClamps(t *testing.T) {
+ for _, tc := range []struct {
+ in string
+ want time.Duration
+ bad bool
+ }{
+ {in: "", want: 0},
+ {in: "5s", want: 5 * time.Second},
+ {in: "30s", want: scanWaitMax},
+ {in: "5m", want: scanWaitMax},
+ {in: "24h", want: scanWaitMax},
+ {in: "soon", bad: true},
+ {in: "-1s", bad: true},
+ } {
+ got, err := scanWaitFor(tc.in)
+ if tc.bad {
+ if err == nil {
+ t.Errorf("scanWaitFor(%q) = %v, want an error", tc.in, got)
+ }
+ continue
+ }
+ if err != nil {
+ t.Errorf("scanWaitFor(%q): %v", tc.in, err)
+ continue
+ }
+ if got != tc.want {
+ t.Errorf("scanWaitFor(%q) = %v, want %v", tc.in, got, tc.want)
+ }
+ }
+}
+
+// TestLongPollWakesOnDrain: a request parked on `wait` returns as soon as the
+// drain records a verdict, rather than sitting out its window.
+//
+// The wait is long and the assertion is that the answer arrives EARLY, which
+// is the only formulation that fails if the wake is never delivered — a test
+// that merely checked the final answer would pass against a door that always
+// slept the full duration.
+func TestLongPollWakesOnDrain(t *testing.T) {
+ ctx := context.Background()
+ fx, sc, tok, bucket := statusArmed(t, "code-status-poll")
+ seedObject(sc, bucket, "uploads/x.pdf", "etag-x")
+ if _, err := objstore.Enqueue(ctx, fx.rel, fx.srv.scanQueue, release.OpsAccount, "titogo",
+ &objstore.QueueItem{Key: "uploads/x.pdf", By: "test"}, time.Now()); err != nil {
+ t.Fatal(err)
+ }
+
+ type reply struct {
+ out map[string]any
+ took time.Duration
+ state string
+ }
+ done := make(chan reply, 1)
+ go func() {
+ start := time.Now()
+ _, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=25s")
+ st, _ := out["state"].(string)
+ done <- reply{out: out, took: time.Since(start), state: st}
+ }()
+
+ // Give the request time to park before the drain runs, so the wake is
+ // what returns it rather than the first read.
+ time.Sleep(150 * time.Millisecond)
+ if err := fx.srv.scanDrainTick(ctx); err != nil {
+ t.Fatal(err)
+ }
+
+ select {
+ case got := <-done:
+ if got.state != "clean" {
+ t.Fatalf("state = %v (%v), want clean", got.state, got.out)
+ }
+ if got.took > 20*time.Second {
+ t.Errorf("took %v against a 25s wait — it sat out the window rather than being woken", got.took)
+ }
+ case <-time.After(20 * time.Second):
+ t.Fatal("the parked request was never woken by the drain")
+ }
+}
+
+// TestLongPollReturnsPendingWhenNothingHappens: the window is a ceiling, not a
+// promise, and it must end.
+func TestLongPollReturnsPendingWhenNothingHappens(t *testing.T) {
+ fx, sc, tok, bucket := statusArmed(t, "code-status-timeout")
+ seedObject(sc, bucket, "uploads/quiet.pdf", "etag-q")
+
+ start := time.Now()
+ status, out := getStatus(t, fx, tok, "key=uploads/quiet.pdf&wait=300ms")
+ if status != http.StatusOK || out["state"] != "pending" {
+ t.Fatalf("quiet wait = %d %v, want 200 pending", status, out)
+ }
+ if took := time.Since(start); took < 250*time.Millisecond {
+ t.Errorf("returned after %v — it did not wait the window it was given", took)
+ }
+}
+
+// TestLongPollCapsWaiters covers BOTH caps, and the per-app one is not implied
+// by the global one: without it a single tenant fills the global budget and
+// every other tenant's wait degrades to polling.
+func TestLongPollCapsWaiters(t *testing.T) {
+ fx, _, _, _ := statusArmed(t, "code-status-caps")
+
+ // Per app: one app's own keys, up to its cap and then no further.
+ ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}
+ var release1 []func()
+ for i := 0; i < scanWaitersPerApp; i++ {
+ _, rel, ok := fx.srv.scanWait(ref, string(rune('a'+i%26))+"/x")
+ if !ok {
+ t.Fatalf("refused waiter %d, below the per-app cap of %d", i, scanWaitersPerApp)
+ }
+ release1 = append(release1, rel)
+ }
+ if _, _, ok := fx.srv.scanWait(ref, "one/too/many"); ok {
+ t.Errorf("parked a waiter past the per-app cap of %d", scanWaitersPerApp)
+ }
+ // A DIFFERENT app is unaffected — which is the whole point of the per-app
+ // cap existing beside the global one.
+ other := objstore.AppRef{Account: release.OpsAccount, App: "otherapp"}
+ if _, rel, ok := fx.srv.scanWait(other, "x"); !ok {
+ t.Errorf("one app at its cap blocked another app entirely")
+ } else {
+ rel()
+ }
+ for _, rel := range release1 {
+ rel()
+ }
+ // Releasing gives the capacity back rather than burning it.
+ if _, rel, ok := fx.srv.scanWait(ref, "after/release"); !ok {
+ t.Error("capacity was not returned when the waiters were released")
+ } else {
+ rel()
+ }
+
+ // Global: enough distinct apps that no per-app cap is ever reached.
+ var release2 []func()
+ for i := 0; ; i++ {
+ app := "app" + string(rune('A'+i%26)) + string(rune('A'+(i/26)%26))
+ r := objstore.AppRef{Account: release.OpsAccount, App: app}
+ ch, rel, ok := fx.srv.scanWait(r, "k")
+ if !ok {
+ break
+ }
+ _ = ch
+ release2 = append(release2, rel)
+ if i > scanWaitersGlobal+8 {
+ t.Fatalf("parked %d waiters with a global cap of %d", i, scanWaitersGlobal)
+ }
+ }
+ if len(release2) != scanWaitersGlobal {
+ t.Errorf("global cap admitted %d waiters, want %d", len(release2), scanWaitersGlobal)
+ }
+ for _, rel := range release2 {
+ rel()
+ }
+}
+
+// TestReleaseIsIdempotent: the door defers the release and a caller may call it
+// early. A second call that decremented again would leak capacity downwards
+// until the console stopped parking anything at all.
+func TestReleaseIsIdempotent(t *testing.T) {
+ fx, _, _, _ := statusArmed(t, "code-status-idem")
+ ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}
+
+ _, rel, ok := fx.srv.scanWait(ref, "k")
+ if !ok {
+ t.Fatal("idle console refused a waiter")
+ }
+ rel()
+ rel()
+ rel()
+
+ fx.srv.scanWaits.mu.Lock()
+ total := fx.srv.scanWaits.total
+ fx.srv.scanWaits.mu.Unlock()
+ if total != 0 {
+ t.Errorf("total = %d after releasing one waiter three times, want 0", total)
+ }
+}
+
+// TestStatusDoorAnswersWhenTheWaitCapIsFull: over the cap nothing is refused.
+// The read still happens and still answers truthfully; only the waiting is
+// skipped, so a full console degrades to polling rather than to errors.
+func TestStatusDoorAnswersWhenTheWaitCapIsFull(t *testing.T) {
+ fx, sc, tok, bucket := statusArmed(t, "code-status-full")
+ v := seedObject(sc, bucket, "uploads/x.pdf", "etag-x")
+ sc.Put(bucket, "uploads/x.pdf", v, objstore.VerdictClean)
+
+ ref := objstore.AppRef{Account: release.OpsAccount, App: "titogo"}
+ var rels []func()
+ for i := 0; i < scanWaitersPerApp; i++ {
+ if _, rel, ok := fx.srv.scanWait(ref, "filler/"+string(rune('a'+i%26))); ok {
+ rels = append(rels, rel)
+ }
+ }
+ defer func() {
+ for _, rel := range rels {
+ rel()
+ }
+ }()
+
+ start := time.Now()
+ status, out := getStatus(t, fx, tok, "key=uploads/x.pdf&wait=10s")
+ if status != http.StatusOK || out["state"] != "clean" {
+ t.Fatalf("over the cap = %d %v, want the truthful 200 clean", status, out)
+ }
+ if took := time.Since(start); took > 5*time.Second {
+ t.Errorf("took %v — a request that cannot park should answer at once", took)
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index c6e05b4b5..93f8179ad 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -765,6 +765,9 @@ func (s *Server) registerCLIRoutes() {
// store was requested, granted and enabled, so the worst a stolen service
// token achieves is queueing scans of its own app's objects.
s.cliRoute("POST /api/cli/apps/{account}/{app}/store/scan/object", svcrole.Allow(svcrole.Service), s.handleAPIScanEnqueue)
+ // And reading the answer back (spec §5.2), same role and same reasoning:
+ // an app asking about its own object.
+ s.cliRoute("GET /api/cli/apps/{account}/{app}/store/scan/object", svcrole.Allow(svcrole.Service), s.handleAPIScanObject)
// The deployment-wide doors (internal/console/apisystem.go). No {account}
// segment, so no service credential can be scoped to them at all: an empty
// Set shuts them to every credential, and closedDoors pins that shut.
diff --git a/internal/console/scandrain.go b/internal/console/scandrain.go
index cc2232178..f28ea490a 100644
--- a/internal/console/scandrain.go
+++ b/internal/console/scandrain.go
@@ -246,6 +246,12 @@ type scanWaitKey struct {
type scanWaiters struct {
mu sync.Mutex
m map[scanWaitKey]map[chan struct{}]struct{}
+ // perApp and total are the two caps (§5.2). Counted rather than derived
+ // from m, because m is keyed by OBJECT: one app parking a thousand
+ // requests across a thousand keys is a thousand map entries of one, which
+ // a per-key length would read as no load at all.
+ perApp map[objstore.AppRef]int
+ total int
}
// scanWait registers interest in one object and returns the channel to select
@@ -254,30 +260,52 @@ type scanWaiters struct {
// 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()) {
+func (s *Server) scanWait(ref objstore.AppRef, key string) (ch <-chan struct{}, release func(), ok bool) {
w := &s.scanWaits
k := scanWaitKey{ref: ref, key: key}
- ch := make(chan struct{}, 1)
+ c := make(chan struct{}, 1)
w.mu.Lock()
+ defer w.mu.Unlock()
+ if w.total >= scanWaitersGlobal || w.perApp[ref] >= scanWaitersPerApp {
+ // Refused a PARK, not refused an answer. The caller reads the object
+ // and replies truthfully; only the waiting is skipped, so a full
+ // console degrades to polling rather than to errors.
+ return nil, func() {}, false
+ }
if w.m == nil {
w.m = map[scanWaitKey]map[chan struct{}]struct{}{}
+ w.perApp = map[objstore.AppRef]int{}
}
if w.m[k] == nil {
w.m[k] = map[chan struct{}]struct{}{}
}
- w.m[k][ch] = struct{}{}
- w.mu.Unlock()
- return ch, func() {
+ w.m[k][c] = struct{}{}
+ w.perApp[ref]++
+ w.total++
+ released := false
+ return c, func() {
w.mu.Lock()
defer w.mu.Unlock()
+ // IDEMPOTENT. The status door defers this and the test calls it early;
+ // a second call that decremented again would leak capacity downwards
+ // until the console stopped parking anything at all.
+ if released {
+ return
+ }
+ released = true
if set := w.m[k]; set != nil {
- delete(set, ch)
+ delete(set, c)
if len(set) == 0 {
// The map must not grow one entry per key ever waited on.
delete(w.m, k)
}
}
- }
+ w.perApp[ref]--
+ if w.perApp[ref] <= 0 {
+ delete(w.perApp, ref)
+ }
+ w.total--
+ }, true
}
// scanNotify wakes everyone waiting on one object. Never blocks.
diff --git a/internal/console/scandrain_test.go b/internal/console/scandrain_test.go
index a07deb79f..71b230755 100644
--- a/internal/console/scandrain_test.go
+++ b/internal/console/scandrain_test.go
@@ -510,10 +510,16 @@ func TestScanNotifyWakesAWaiter(t *testing.T) {
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")
+ woken, cancel, ok := fx.srv.scanWait(ref, "uploads/x.pdf")
+ if !ok {
+ t.Fatal("an idle console refused to park the first waiter")
+ }
defer cancel()
// A waiter on a DIFFERENT key must not be woken by this drain.
- other, cancelOther := fx.srv.scanWait(ref, "uploads/other.pdf")
+ other, cancelOther, ok := fx.srv.scanWait(ref, "uploads/other.pdf")
+ if !ok {
+ t.Fatal("an idle console refused to park a second waiter")
+ }
defer cancelOther()
if err := fx.srv.scanDrainTick(ctx); err != nil {
@@ -569,7 +575,10 @@ func TestDrainNotifiesEvenWhenALaterItemFails(t *testing.T) {
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)
+ woken, cancel, ok := fx.srv.scanWait(ref, first.Key)
+ if !ok {
+ t.Fatal("an idle console refused to park a waiter")
+ }
defer cancel()
sc.Fail("Scan", errors.New("engine had a bad moment"))