console: rate-limit ledger writes, delimiter-aware listing, honest error statuses, publish audit
Review findings 1-4 on the ledger CLI API: - Every ledger WRITE door now passes the s.cli.write budget its promote/rollback/new-account siblings pass, keyed on the address. The two GET doors stay unlimited, like their read-only siblings. Blobs take the limiter and MaxBytesReader but NOT shipSem: 4 MiB is not the unbounded-buffer hazard that semaphore exists for, and sharing its two slots would let blob traffic starve the deploy path. - release.Store grows ListDirs, the delimiter-aware listing (S3 CommonPrefixes; one os.ReadDir for DirStore). GET /ledgers used to walk every entry and blob ever written under an app to learn a handful of names; it now costs the number of chains. - ledger.ErrBusy marks CAS exhaustion, so contention (503, retry-worthy) and a broken bucket (500) stop sharing a status — and no store error text, bucket name or request id reaches a client. It goes to the log instead. - Publishing flips what the world can see, so it now leaves an audit line naming who, the flag and the ledgers, in the shape promote/rollback/ship already use.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
8 files changed,
+485
−22
internal/console/apiledger.go+94 −19internal/console/apiledger_test.go+178 −0internal/ledger/append.go+8 −1internal/ledger/append_test.go+39 −0internal/release/qualified_test.go+3 −2internal/release/scopedpromote_test.go+5 −0internal/release/store.go+71 −0internal/release/store_test.go+87 −0
diff --git a/internal/console/apiledger.go b/internal/console/apiledger.go| index 8b4c27f86..2195a0fa6 100644 |
| --- a/internal/console/apiledger.go |
| +++ b/internal/console/apiledger.go |
| @@ -22,21 +22,56 @@ package console |
| import ( |
| "encoding/json" |
| "errors" |
| + "fmt" |
| "io" |
| "log" |
| "net/http" |
| "os" |
| + "strconv" |
| "strings" |
| "time" |
| "github.com/carlosframework/platform/internal/ledger" |
| + "github.com/carlosframework/platform/internal/release" |
| ) |
| +// ledgerWriteAllowed is the write budget every ledger WRITE door passes, in |
| +// the same shape and with the same key as promote/rollback/new-account: the |
| +// principal a write budget belongs to is the PERSON, not the terminal they |
| +// happen to be typing on. The two GET doors below are unlimited, matching |
| +// their read-only siblings (channels, apps). |
| +func (s *Server) ledgerWriteAllowed(w http.ResponseWriter, addr string) bool { |
| + if !s.cli.write.allow(addr, time.Now()) { |
| + jsonError(w, http.StatusTooManyRequests, "too many requests — wait a minute and try again") |
| + return false |
| + } |
| + return true |
| +} |
| + |
| +// ledgerStoreError answers a failure that is NOT the caller's fault. Two |
| +// things matter here. First, contention is retry-worthy and a broken bucket |
| +// is not, so they must not share a status: a 503 tells a CLI to try again, |
| +// a 500 tells it to stop. Second, the store's own error text — bucket names, |
| +// key paths, AWS request ids — never reaches the client; it goes to the log, |
| +// where the operator who can act on it will see it. |
| +func (s *Server) ledgerStoreError(w http.ResponseWriter, what string, err error) { |
| + log.Printf("console: %s: %v", what, err) |
| + if errors.Is(err, ledger.ErrBusy) || errors.Is(err, release.ErrCASConflict) { |
| + jsonError(w, http.StatusServiceUnavailable, |
| + "the ledger is busy right now — try again in a moment") |
| + return |
| + } |
| + jsonError(w, http.StatusInternalServerError, "could not reach the ledger store") |
| +} |
| + |
| func (s *Server) handleAPILedgerAppend(w http.ResponseWriter, r *http.Request) { |
| addr, acct, app, ok := s.apiAccountApp(w, r) |
| if !ok { |
| return |
| } |
| + if !s.ledgerWriteAllowed(w, addr) { |
| + return |
| + } |
| name := r.PathValue("ledger") |
| if !ledger.ValidName(name) { |
| jsonError(w, http.StatusBadRequest, "invalid ledger name") |
| @@ -61,11 +96,22 @@ func (s *Server) handleAPILedgerAppend(w http.ResponseWriter, r *http.Request) { |
| jsonError(w, http.StatusBadRequest, "missing payload") |
| return |
| } |
| + // The cap, checked HERE so an over-cap payload is the caller's 413 and |
| + // not swept into ledger.Append's generic error return below. A body |
| + // under MaxBytesReader's allowance can still carry an over-cap payload: |
| + // the reader bounds the request, this bounds the entry. |
| + if len(in.Payload) > ledger.MaxPayload { |
| + jsonError(w, http.StatusRequestEntityTooLarge, "payload exceeds the 64 KiB cap") |
| + return |
| + } |
| a, err := ledger.Append(r.Context(), s.cfg.Releases, acct.Sqid, app, name, addr, |
| in.Payload, time.Now) |
| if err != nil { |
| - log.Printf("console: ledger append %s/%s/%s: %v", acct.Sqid, app, name, err) |
| - jsonError(w, http.StatusBadRequest, err.Error()) |
| + // Every caller-caused failure has been answered above — an invalid |
| + // name, a payload over the cap, a body that isn't JSON. What's left |
| + // is contention or a broken store, neither of which is the caller's |
| + // fault and neither of which they should read raw. |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger append %s/%s/%s", acct.Sqid, app, name), err) |
| return |
| } |
| w.Header().Set("Content-Type", "application/json") |
| @@ -88,18 +134,27 @@ func (s *Server) handleAPILedgerHead(w http.ResponseWriter, r *http.Request) { |
| return |
| } |
| if err != nil { |
| - jsonError(w, http.StatusInternalServerError, "could not read ledger head") |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger head %s/%s/%s", acct.Sqid, app, name), err) |
| return |
| } |
| w.Header().Set("Content-Type", "application/json") |
| w.Write(b) |
| } |
| +// handleAPILedgerBlob takes raw bytes. It is rate-limited on the write |
| +// budget and hard-bounded at 4 MiB by MaxBytesReader, and deliberately does |
| +// NOT take ship's semaphore: a blob is 64× smaller than a ship's cap, so a |
| +// bounded 4 MiB read is not the unbounded-buffer hazard shipSem exists for, |
| +// and sharing that two-slot semaphore would let blob traffic starve the |
| +// deploy path. |
| func (s *Server) handleAPILedgerBlob(w http.ResponseWriter, r *http.Request) { |
| - _, acct, app, ok := s.apiAccountApp(w, r) |
| + addr, acct, app, ok := s.apiAccountApp(w, r) |
| if !ok { |
| return |
| } |
| + if !s.ledgerWriteAllowed(w, addr) { |
| + return |
| + } |
| name := r.PathValue("ledger") |
| if !ledger.ValidName(name) { |
| jsonError(w, http.StatusBadRequest, "invalid ledger name") |
| @@ -107,12 +162,17 @@ func (s *Server) handleAPILedgerBlob(w http.ResponseWriter, r *http.Request) { |
| } |
| body, err := io.ReadAll(http.MaxBytesReader(w, r.Body, ledger.MaxBlob)) |
| if err != nil { |
| - jsonError(w, http.StatusRequestEntityTooLarge, "blob exceeds the 4 MiB cap") |
| + var tooBig *http.MaxBytesError |
| + if errors.As(err, &tooBig) { |
| + jsonError(w, http.StatusRequestEntityTooLarge, "blob exceeds the 4 MiB cap") |
| + return |
| + } |
| + jsonError(w, http.StatusBadRequest, "could not read the request body") |
| return |
| } |
| sha, err := ledger.PutBlob(r.Context(), s.cfg.Releases, acct.Sqid, app, name, body) |
| if err != nil { |
| - jsonError(w, http.StatusBadRequest, err.Error()) |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger blob %s/%s/%s", acct.Sqid, app, name), err) |
| return |
| } |
| w.Header().Set("Content-Type", "application/json") |
| @@ -132,32 +192,42 @@ func (s *Server) handleAPILedgerList(w http.ResponseWriter, r *http.Request) { |
| case errors.Is(err, os.ErrNotExist): |
| // no manifest yet — null in the response |
| default: |
| - jsonError(w, http.StatusInternalServerError, "could not read ledger manifest") |
| - return |
| - } |
| - // Distinct ledger names, derived from the HEAD keys under this app. |
| - keys, err := s.cfg.Releases.List(r.Context(), "ledgers/"+acct.Sqid+"/"+app+"/") |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger manifest %s/%s", acct.Sqid, app), err) |
| + return |
| + } |
| + // The ledger names are the immediate child directories of the app's |
| + // prefix — one delimiter-aware listing, whose cost is the number of |
| + // chains rather than the total length of their histories. A flat List |
| + // here would walk every entry and every blob ever written under this |
| + // app just to learn a handful of names. Blobs live INSIDE a chain's |
| + // directory, so the children are exactly the chain names; ledgers.json |
| + // is a file at this level, and ListDirs never returns files. |
| + prefix := "ledgers/" + acct.Sqid + "/" + app + "/" |
| + dirs, err := s.cfg.Releases.ListDirs(r.Context(), prefix) |
| if err != nil { |
| - jsonError(w, http.StatusInternalServerError, "could not list ledgers") |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger list %s/%s", acct.Sqid, app), err) |
| return |
| } |
| var names []string |
| - for _, k := range keys { |
| - if !strings.HasSuffix(k, "/HEAD.json") { |
| + for _, d := range dirs { |
| + n := strings.TrimSuffix(strings.TrimPrefix(d, prefix), "/") |
| + if n == "" { |
| continue |
| } |
| - rest := strings.TrimPrefix(k, "ledgers/"+acct.Sqid+"/"+app+"/") |
| - names = append(names, strings.TrimSuffix(rest, "/HEAD.json")) |
| + names = append(names, n) |
| } |
| w.Header().Set("Content-Type", "application/json") |
| json.NewEncoder(w).Encode(map[string]any{"manifest": manifest, "ledgers": names}) |
| } |
| func (s *Server) handleAPILedgerPublish(w http.ResponseWriter, r *http.Request) { |
| - _, acct, app, ok := s.apiAccountApp(w, r) |
| + addr, acct, app, ok := s.apiAccountApp(w, r) |
| if !ok { |
| return |
| } |
| + if !s.ledgerWriteAllowed(w, addr) { |
| + return |
| + } |
| var in ledger.Manifest |
| r.Body = http.MaxBytesReader(w, r.Body, 8<<10) |
| if err := json.NewDecoder(r.Body).Decode(&in); err != nil { |
| @@ -171,10 +241,15 @@ func (s *Server) handleAPILedgerPublish(w http.ResponseWriter, r *http.Request) |
| } |
| } |
| if err := ledger.WriteManifest(r.Context(), s.cfg.Releases, acct.Sqid, app, in); err != nil { |
| - log.Printf("console: ledger publish %s/%s: %v", acct.Sqid, app, err) |
| - jsonError(w, http.StatusInternalServerError, "could not write manifest") |
| + s.ledgerStoreError(w, fmt.Sprintf("ledger publish %s/%s", acct.Sqid, app), err) |
| return |
| } |
| + // Flipping an app's ledgers public is exactly the kind of act the audit |
| + // trail exists for, and the ONLY one on this API that changes what the |
| + // world can see — the same trail promote, rollback and ship write to. |
| + s.audit(acct, app, addr, "ledger-publish", |
| + "set ledgers public="+strconv.FormatBool(in.Public)+ |
| + " ["+strings.Join(in.Ledgers, " ")+"] (cli)") |
| w.Header().Set("Content-Type", "application/json") |
| w.Write([]byte("{}")) |
| } |
diff --git a/internal/console/apiledger_test.go b/internal/console/apiledger_test.go| index fefab06c5..050c81e09 100644 |
| --- a/internal/console/apiledger_test.go |
| +++ b/internal/console/apiledger_test.go |
| @@ -12,6 +12,7 @@ import ( |
| "bytes" |
| "context" |
| "encoding/json" |
| + "errors" |
| "io" |
| "net/http" |
| "strings" |
| @@ -19,6 +20,7 @@ import ( |
| "time" |
| "github.com/carlosframework/platform/internal/ledger" |
| + "github.com/carlosframework/platform/internal/release" |
| ) |
| // postBody is postJSON's raw-body cousin: these tests send bodies that are |
| @@ -246,6 +248,18 @@ func TestAPILedgerPublishAndList(t *testing.T) { |
| t.Fatalf("publish = %d %s", status, out) |
| } |
| + // A second entry and a blob under the same chain: the listing is |
| + // delimiter-aware, so neither the entries/ nor the blobs/ directory may |
| + // surface as a ledger name, however much history accrues. |
| + if status, out := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/carbon/append", token, |
| + `{"payload":{"n":2}}`); status != http.StatusOK { |
| + t.Fatalf("second append = %d %s", status, out) |
| + } |
| + if _, err := fx.rel.Put(context.Background(), |
| + ledger.BlobKey("bab", "titogo", "carbon", strings.Repeat("a", 64)), []byte("x"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| status, body = getRaw(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers", token) |
| if status != http.StatusOK || !strings.Contains(body, `"carbon"`) { |
| t.Fatalf("list = %d %s, want carbon among the ledgers", status, body) |
| @@ -264,6 +278,33 @@ func TestAPILedgerPublishAndList(t *testing.T) { |
| if post.Manifest.Schema != ledger.Schema { |
| t.Fatalf("manifest schema = %d, want %d (stamped server-side)", post.Manifest.Schema, ledger.Schema) |
| } |
| + if len(post.Ledgers) != 1 || post.Ledgers[0] != "carbon" { |
| + t.Fatalf("ledgers = %v, want exactly [carbon] — entries/ and blobs/ are not ledgers", post.Ledgers) |
| + } |
| + |
| + // Publishing is a visibility change, so it leaves an audit line naming |
| + // who did it, the flag, and the ledgers — the same trail promote, |
| + // rollback and ship write to. |
| + events, err := fx.st.Events(OpsAccountID, "titogo", 10) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var pub *Event |
| + for i := range events { |
| + if events[i].Action == "ledger-publish" { |
| + pub = &events[i] |
| + } |
| + } |
| + if pub == nil { |
| + t.Fatalf("events = %v, want a ledger-publish row", events) |
| + } |
| + if pub.Address != "alice@keymail.dev" { |
| + t.Fatalf("ledger-publish address = %q, want the authenticated identity", pub.Address) |
| + } |
| + if !strings.Contains(pub.Detail, "public=true") || !strings.Contains(pub.Detail, "carbon") || |
| + !strings.Contains(pub.Detail, "(cli)") { |
| + t.Fatalf("ledger-publish detail = %q, want the flag, the ledgers and the CLI marker", pub.Detail) |
| + } |
| // A manifest naming a ledger that isn't a ledger name is refused. |
| if status, out := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/publish", token, |
| @@ -332,3 +373,140 @@ func TestAPILedgerBlob(t *testing.T) { |
| t.Fatalf("blob with a bad ledger name = %d %s, want 400", status, out) |
| } |
| } |
| + |
| +// TestAPILedgerWritesAreRateLimited: every WRITE door here passes the same |
| +// write budget its promote/rollback/new-account siblings do, keyed on the |
| +// address. Without it, the one API that appends an object per call is the |
| +// one API with no bound on how fast a token can fill a bucket. |
| +func TestAPILedgerWritesAreRateLimited(t *testing.T) { |
| + fx, token := ledgerFixture(t, "code-ledger-ratelimit") |
| + |
| + limited := false |
| + for i := 0; i < cliWriteBurst+3; i++ { |
| + status, _ := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/carbon/append", token, |
| + `{"payload":{"n":1}}`) |
| + if status == http.StatusTooManyRequests { |
| + limited = true |
| + break |
| + } |
| + } |
| + if !limited { |
| + t.Fatalf("append was never rate-limited in %d attempts", cliWriteBurst+3) |
| + } |
| + |
| + // The budget is shared across the write doors — an exhausted address |
| + // cannot simply switch verb. (Same limiter, same key: this is what |
| + // proves publish and blobs pass it too, not just append.) |
| + if status, _ := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/publish", token, |
| + `{"public":true,"ledgers":["carbon"]}`); status != http.StatusTooManyRequests { |
| + t.Fatalf("publish on an exhausted budget = %d, want 429", status) |
| + } |
| + req, err := http.NewRequest(http.MethodPost, |
| + fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/prompts/blobs", bytes.NewReader([]byte("corpus"))) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + req.Header.Set("Authorization", "Bearer "+token) |
| + res, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + res.Body.Close() |
| + if res.StatusCode != http.StatusTooManyRequests { |
| + t.Fatalf("blob on an exhausted budget = %d, want 429", res.StatusCode) |
| + } |
| + |
| + // The READ doors are unlimited, like their read-only siblings — an |
| + // exhausted write budget must not blind a CLI to what it already wrote. |
| + if status, body := getRaw(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers", token); status != http.StatusOK { |
| + t.Fatalf("list on an exhausted write budget = %d %s, want 200", status, body) |
| + } |
| + if status, _ := getRaw(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/carbon/head", token); status != http.StatusOK { |
| + t.Fatalf("head on an exhausted write budget = %d, want 200", status) |
| + } |
| +} |
| + |
| +// casStore loses every HEAD race, the way a chain under more concurrent |
| +// writers than the retry loop can outlast behaves. |
| +type casStore struct{ release.Store } |
| + |
| +func (c casStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if strings.HasSuffix(key, "/HEAD.json") || strings.HasSuffix(key, "/ledgers.json") { |
| + return "", release.ErrCASConflict |
| + } |
| + return c.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// brokenStore is a bucket the console cannot reach, whose error text is |
| +// exactly the kind that must never reach a client. |
| +type brokenStore struct{ release.Store } |
| + |
| +const brokenStoreErr = "AccessDenied: s3://carlos-prod-secret-bucket/ledgers request id ABC123" |
| + |
| +func (b brokenStore) Put(context.Context, string, []byte, string) (string, error) { |
| + return "", errors.New(brokenStoreErr) |
| +} |
| + |
| +func (b brokenStore) ListDirs(context.Context, string) ([]string, error) { |
| + return nil, errors.New(brokenStoreErr) |
| +} |
| + |
| +// TestAPILedgerStoreFailureStatuses: a failure that is not the caller's |
| +// fault must not be reported as if it were. Contention is retry-worthy |
| +// (503), a broken bucket is not (500), and neither answer may echo the |
| +// store's own error text — bucket names and request ids are for the log. |
| +func TestAPILedgerStoreFailureStatuses(t *testing.T) { |
| + t.Run("contention is 503", func(t *testing.T) { |
| + fx, token := ledgerFixture(t, "code-ledger-busy") |
| + fx.srv.cfg.Releases = casStore{fx.rel} |
| + |
| + status, out := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/carbon/append", token, |
| + `{"payload":{"n":1}}`) |
| + if status != http.StatusServiceUnavailable { |
| + t.Fatalf("append under total contention = %d %s, want 503", status, out) |
| + } |
| + if !strings.Contains(string(out), "try again") { |
| + t.Fatalf("contention error = %s, want it to invite a retry", out) |
| + } |
| + if strings.Contains(string(out), "CAS") || strings.Contains(string(out), "HEAD.json") { |
| + t.Fatalf("contention error leaks store internals: %s", out) |
| + } |
| + // Publish's own lost race is the same retry-worthy answer. |
| + if status, out := postBody(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/publish", token, |
| + `{"public":true,"ledgers":["carbon"]}`); status != http.StatusServiceUnavailable { |
| + t.Fatalf("publish under total contention = %d %s, want 503", status, out) |
| + } |
| + }) |
| + |
| + t.Run("a broken bucket is 500 and says nothing", func(t *testing.T) { |
| + fx, token := ledgerFixture(t, "code-ledger-broken") |
| + fx.srv.cfg.Releases = brokenStore{fx.rel} |
| + |
| + type answer struct { |
| + what string |
| + status int |
| + body string |
| + } |
| + appendStatus, appendBody := postBody(t, |
| + fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/carbon/append", token, `{"payload":{"n":1}}`) |
| + publishStatus, publishBody := postBody(t, |
| + fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers/publish", token, `{"public":true,"ledgers":["carbon"]}`) |
| + listStatus, listBody := getRaw(t, fx.ts.URL+"/api/cli/apps/bab/titogo/ledgers", token) |
| + cases := []answer{ |
| + {"append", appendStatus, string(appendBody)}, |
| + {"publish", publishStatus, string(publishBody)}, |
| + {"list", listStatus, listBody}, |
| + } |
| + |
| + for _, c := range cases { |
| + if c.status != http.StatusInternalServerError { |
| + t.Errorf("%s against a broken bucket = %d %s, want 500", c.what, c.status, c.body) |
| + } |
| + if strings.Contains(c.body, "AccessDenied") || |
| + strings.Contains(c.body, "carlos-prod-secret-bucket") || |
| + strings.Contains(c.body, "ABC123") { |
| + t.Errorf("%s leaked the store's error text to the client: %s", c.what, c.body) |
| + } |
| + } |
| + }) |
| +} |
diff --git a/internal/ledger/append.go b/internal/ledger/append.go| index 43dd13b8d..536575451 100644 |
| --- a/internal/ledger/append.go |
| +++ b/internal/ledger/append.go |
| @@ -22,6 +22,13 @@ type Appended struct { |
| // attempts outlasts any plausible burst without ever spinning. |
| const appendAttempts = 4 |
| +// ErrBusy marks the one Append failure that is neither the caller's fault |
| +// nor a broken store: too many writers accruing the same chain at once, so |
| +// every CAS attempt lost. A caller that retries later succeeds — which is |
| +// exactly what a caller cannot work out from an error STRING, hence a |
| +// sentinel to match with errors.Is. |
| +var ErrBusy = errors.New("ledger: append contention") |
| + |
| // Append validates, chains, and writes one entry, advancing HEAD with CAS. |
| // A lost HEAD race re-reads and re-chains (up to 4 attempts). An entry |
| // object orphaned by a lost race is harmless: nothing references it, and |
| @@ -98,5 +105,5 @@ func Append(ctx context.Context, st release.Store, sqid, app, name, writer strin |
| } |
| return Appended{Key: key, SHA256: sha, Count: nh.Count}, nil |
| } |
| - return Appended{}, fmt.Errorf("ledger: append lost %d CAS races: %w", appendAttempts, lastErr) |
| + return Appended{}, fmt.Errorf("ledger: append lost %d CAS races: %w: %w", appendAttempts, ErrBusy, lastErr) |
| } |
diff --git a/internal/ledger/append_test.go b/internal/ledger/append_test.go| index 53b3aaedd..3438c8847 100644 |
| --- a/internal/ledger/append_test.go |
| +++ b/internal/ledger/append_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "encoding/json" |
| "errors" |
| "os" |
| + "strings" |
| "testing" |
| "time" |
| @@ -128,3 +129,41 @@ func TestAppendRetriesLostRace(t *testing.T) { |
| t.Fatal("retried entry did not re-chain onto the rival's head") |
| } |
| } |
| + |
| +// alwaysCASStore loses every HEAD race — the shape of a chain under more |
| +// concurrent writers than the retry loop can outlast. |
| +type alwaysCASStore struct{ release.Store } |
| + |
| +func (a alwaysCASStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if strings.HasSuffix(key, "/HEAD.json") { |
| + return "", release.ErrCASConflict |
| + } |
| + return a.Store.Put(ctx, key, body, ifMatch) |
| +} |
| + |
| +// TestAppendExhaustionIsErrBusy: losing every race is neither the caller's |
| +// fault nor a broken store — it is retry-worthy, and a caller must be able |
| +// to tell WITHOUT matching on error text, because callers that match on text |
| +// break the first time the sentence changes. |
| +func TestAppendExhaustionIsErrBusy(t *testing.T) { |
| + st := alwaysCASStore{Store: release.NewDirStore(t.TempDir())} |
| + _, err := Append(context.Background(), st, "bab", "platform", "carbon", "me", |
| + json.RawMessage(`{"n":1}`), fixedNow) |
| + if err == nil { |
| + t.Fatal("Append succeeded against a store that loses every HEAD race") |
| + } |
| + if !errors.Is(err, ErrBusy) { |
| + t.Fatalf("exhaustion err = %v, want it to wrap ErrBusy", err) |
| + } |
| + // The underlying cause is still reachable — the sentinel adds to the |
| + // chain, it does not replace it. |
| + if !errors.Is(err, release.ErrCASConflict) { |
| + t.Fatalf("exhaustion err = %v, want it to still wrap ErrCASConflict", err) |
| + } |
| + // A caller-caused failure must NOT look busy. |
| + _, err = Append(context.Background(), st, "bab", "platform", "Not Valid", "me", |
| + json.RawMessage(`{"n":1}`), fixedNow) |
| + if err == nil || errors.Is(err, ErrBusy) { |
| + t.Fatalf("invalid-name err = %v, want a non-ErrBusy failure", err) |
| + } |
| +} |
diff --git a/internal/release/qualified_test.go b/internal/release/qualified_test.go| index 50fe5071c..03dd8ebbf 100644 |
| --- a/internal/release/qualified_test.go |
| +++ b/internal/release/qualified_test.go |
| @@ -23,8 +23,9 @@ func (e *erroringStore) Get(context.Context, string) ([]byte, string, error) { |
| func (e *erroringStore) Put(context.Context, string, []byte, string) (string, error) { |
| return "", e.err |
| } |
| -func (e *erroringStore) List(context.Context, string) ([]string, error) { return nil, e.err } |
| -func (e *erroringStore) Delete(context.Context, string) error { return e.err } |
| +func (e *erroringStore) List(context.Context, string) ([]string, error) { return nil, e.err } |
| +func (e *erroringStore) ListDirs(context.Context, string) ([]string, error) { return nil, e.err } |
| +func (e *erroringStore) Delete(context.Context, string) error { return e.err } |
| // legacyTripwire fails the test if anything asks the store for a key or |
| // prefix under one app's LEGACY unqualified prefix. It is how "the runtime |
diff --git a/internal/release/scopedpromote_test.go b/internal/release/scopedpromote_test.go| index b9ced5060..fe89f1767 100644 |
| --- a/internal/release/scopedpromote_test.go |
| +++ b/internal/release/scopedpromote_test.go |
| @@ -29,6 +29,11 @@ func (s trapStore) List(_ context.Context, prefix string) ([]string, error) { |
| return nil, nil |
| } |
| +func (s trapStore) ListDirs(_ context.Context, prefix string) ([]string, error) { |
| + s.t.Fatalf("out-of-scope promote listed %s — it must refuse before touching the store", prefix) |
| + return nil, nil |
| +} |
| + |
| func (s trapStore) Delete(_ context.Context, key string) error { |
| s.t.Fatalf("out-of-scope promote DELETED %s — the scope check is not a gate", key) |
| return nil |
diff --git a/internal/release/store.go b/internal/release/store.go| index 094c63427..f8b5cbd7d 100644 |
| --- a/internal/release/store.go |
| +++ b/internal/release/store.go |
| @@ -40,6 +40,14 @@ type Store interface { |
| Get(ctx context.Context, key string) ([]byte, string, error) // body, etag |
| Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) // new etag |
| List(ctx context.Context, prefix string) ([]string, error) |
| + // ListDirs returns the immediate child "directories" of prefix — the |
| + // delimiter-aware listing, so a caller that only wants the names one |
| + // level down does not pay for every object beneath them. Each result is |
| + // a FULL prefix ending in "/" (S3 CommonPrefixes semantics), never a |
| + // bare name. A prefix that does not end in "/" is treated as though it |
| + // did, so the two implementations answer identically. Objects at the |
| + // level itself (files, not directories) are never included. |
| + ListDirs(ctx context.Context, prefix string) ([]string, error) |
| Delete(ctx context.Context, key string) error |
| } |
| @@ -158,6 +166,42 @@ func (d *DirStore) List(_ context.Context, prefix string) ([]string, error) { |
| return out, err |
| } |
| +// ListDirs is DirStore's delimiter-aware listing: one directory read, no |
| +// walk. A missing directory lists empty, like a bucket prefix with nothing |
| +// under it. Only directories are returned, so a file sitting at this level |
| +// (ledgers.json among them) is never mistaken for a child prefix. |
| +func (d *DirStore) ListDirs(_ context.Context, prefix string) ([]string, error) { |
| + if prefix != "" && !strings.HasSuffix(prefix, "/") { |
| + prefix += "/" |
| + } |
| + dir := d.root |
| + if trimmed := strings.TrimSuffix(prefix, "/"); trimmed != "" { |
| + p, err := d.path(trimmed) |
| + if err != nil { |
| + return nil, err |
| + } |
| + dir = p |
| + } |
| + d.mu.Lock() |
| + defer d.mu.Unlock() |
| + ents, err := os.ReadDir(dir) |
| + if err != nil { |
| + if errors.Is(err, fs.ErrNotExist) { |
| + return nil, nil |
| + } |
| + return nil, err |
| + } |
| + var out []string |
| + for _, e := range ents { |
| + if !e.IsDir() { |
| + continue |
| + } |
| + out = append(out, prefix+e.Name()+"/") |
| + } |
| + sort.Strings(out) |
| + return out, nil |
| +} |
| + |
| func (d *DirStore) Delete(_ context.Context, key string) error { |
| p, err := d.path(key) |
| if err != nil { |
| @@ -297,6 +341,33 @@ func (s *S3Store) List(ctx context.Context, prefix string) ([]string, error) { |
| return out, nil |
| } |
| +// ListDirs is the Delimiter:"/" listing: S3 rolls everything below one level |
| +// up into a CommonPrefix, so the response is bounded by the number of child |
| +// prefixes rather than by the total object count beneath them. Objects AT |
| +// this level come back in Contents, which this deliberately ignores. |
| +func (s *S3Store) ListDirs(ctx context.Context, prefix string) ([]string, error) { |
| + if prefix != "" && !strings.HasSuffix(prefix, "/") { |
| + prefix += "/" |
| + } |
| + var out []string |
| + p := awss3.NewListObjectsV2Paginator(s.Client, &awss3.ListObjectsV2Input{ |
| + Bucket: &s.Bucket, |
| + Prefix: &prefix, |
| + Delimiter: aws.String("/"), |
| + }) |
| + for p.HasMorePages() { |
| + page, err := p.NextPage(ctx) |
| + if err != nil { |
| + return nil, err |
| + } |
| + for _, cp := range page.CommonPrefixes { |
| + out = append(out, aws.ToString(cp.Prefix)) |
| + } |
| + } |
| + sort.Strings(out) |
| + return out, nil |
| +} |
| + |
| func (s *S3Store) Delete(ctx context.Context, key string) error { |
| _, err := s.Client.DeleteObject(ctx, &awss3.DeleteObjectInput{ |
| Bucket: aws.String(s.Bucket), |
diff --git a/internal/release/store_test.go b/internal/release/store_test.go| index a8c2bd78c..a7a5f7726 100644 |
| --- a/internal/release/store_test.go |
| +++ b/internal/release/store_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "errors" |
| "fmt" |
| "os" |
| + "strings" |
| "testing" |
| "time" |
| ) |
| @@ -112,6 +113,65 @@ func TestDirStoreDelete(t *testing.T) { |
| } |
| } |
| +// TestDirStoreListDirs: the delimiter-aware listing returns the immediate |
| +// child prefixes and nothing else — not the objects beneath them (which is |
| +// the whole point: cost bounded by the number of children, not by their |
| +// histories), and not the FILES sitting at the same level, which is what |
| +// keeps a ledger's ledgers.json out of its own list of ledgers. |
| +func TestDirStoreListDirs(t *testing.T) { |
| + s := NewDirStore(t.TempDir()) |
| + ctx := context.Background() |
| + for _, k := range []string{ |
| + "ledgers/bab/titogo/carbon/HEAD.json", |
| + "ledgers/bab/titogo/carbon/entries/20260803T000000Z-aaaaaaaaaaaa.json", |
| + "ledgers/bab/titogo/carbon/entries/20260803T000001Z-bbbbbbbbbbbb.json", |
| + "ledgers/bab/titogo/carbon/blobs/deadbeef", |
| + "ledgers/bab/titogo/prompts/HEAD.json", |
| + "ledgers/bab/titogo/ledgers.json", // a FILE at this level |
| + "ledgers/bab/otherapp/elsewhere/HEAD.json", |
| + } { |
| + if _, err := s.Put(ctx, k, []byte(`{}`), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + |
| + got, err := s.ListDirs(ctx, "ledgers/bab/titogo/") |
| + if err != nil { |
| + t.Fatalf("ListDirs: %v", err) |
| + } |
| + want := []string{"ledgers/bab/titogo/carbon/", "ledgers/bab/titogo/prompts/"} |
| + if len(got) != len(want) { |
| + t.Fatalf("ListDirs = %v, want %v (no files, no grandchildren)", got, want) |
| + } |
| + for i := range want { |
| + if got[i] != want[i] { |
| + t.Fatalf("ListDirs = %v, want %v", got, want) |
| + } |
| + } |
| + // A prefix without the trailing slash answers identically — the two |
| + // implementations must not disagree about that. |
| + bare, err := s.ListDirs(ctx, "ledgers/bab/titogo") |
| + if err != nil || len(bare) != len(want) || bare[0] != want[0] { |
| + t.Fatalf("ListDirs without a trailing slash = %v, %v; want %v", bare, err, want) |
| + } |
| + // One level up sees the apps, not the chains under them. |
| + up, err := s.ListDirs(ctx, "ledgers/bab/") |
| + if err != nil || len(up) != 2 || up[0] != "ledgers/bab/otherapp/" || up[1] != "ledgers/bab/titogo/" { |
| + t.Fatalf("ListDirs one level up = %v, %v", up, err) |
| + } |
| + // A prefix with nothing under it lists empty, like an empty bucket |
| + // prefix — not an error. |
| + empty, err := s.ListDirs(ctx, "ledgers/nosuch/") |
| + if err != nil || len(empty) != 0 { |
| + t.Fatalf("ListDirs of a missing prefix = %v, %v; want empty and no error", empty, err) |
| + } |
| + // The root lists the top-level prefixes. |
| + root, err := s.ListDirs(ctx, "") |
| + if err != nil || len(root) != 1 || root[0] != "ledgers/" { |
| + t.Fatalf("ListDirs of the root = %v, %v; want [ledgers/]", root, err) |
| + } |
| +} |
| + |
| // TestS3StoreLive exercises the production store against a real bucket — |
| // only when CARLOS_TEST_S3_BUCKET is set; the type otherwise just compiles. |
| func TestS3StoreLive(t *testing.T) { |
| @@ -152,4 +212,31 @@ func TestS3StoreLive(t *testing.T) { |
| if err != nil || len(keys) != 1 || keys[0] != key { |
| t.Fatalf("List = %v, %v; want [%s]", keys, err, key) |
| } |
| + |
| + // ListDirs against the same live bucket: the object written above sits |
| + // one level below its parent prefix, so the parent's parent must see the |
| + // child PREFIX (CommonPrefixes) and not the object itself. |
| + parent := key[:strings.LastIndex(key, "/")+1] // .../<nanos>/ |
| + grandparent := parent[:strings.LastIndex(parent[:len(parent)-1], "/")+1] |
| + dirs, err := s.ListDirs(ctx, grandparent) |
| + if err != nil { |
| + t.Fatalf("ListDirs: %v", err) |
| + } |
| + found := false |
| + for _, d := range dirs { |
| + if d == parent { |
| + found = true |
| + } |
| + if d == key { |
| + t.Fatalf("ListDirs returned an OBJECT key %q, want only child prefixes", d) |
| + } |
| + } |
| + if !found { |
| + t.Fatalf("ListDirs(%q) = %v, want %q among them", grandparent, dirs, parent) |
| + } |
| + // And at the object's own level there are no child prefixes at all. |
| + leaf, err := s.ListDirs(ctx, parent) |
| + if err != nil || len(leaf) != 0 { |
| + t.Fatalf("ListDirs(%q) = %v, %v; want empty (the only child is an object)", parent, leaf, err) |
| + } |
| } |