amadan

carlos / platform Public

console: run keys are capped per caller, recorded on the request, and settle a lost admission exactly (PRO-6214)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Paul Campbell pushed by claude 6dc07d3804d331a4ca7f7be0d55780f6842c3379 parent bbe84d1
8 files changed, +366 −64
  • docs/superpowers/specs/2026-10-24-service-runs-design.md +12 −6
  • internal/console/apiruns.go +45 −18
  • internal/console/apiruns_key_test.go +148 −1
  • internal/provision/run.go +34 −16
  • internal/provision/run_keys_test.go +62 −12
  • internal/runs/keys.go +28 −11
  • internal/runs/keys_test.go +32 −0
  • internal/runs/runs.go +5 −0
diff --git a/docs/superpowers/specs/2026-10-24-service-runs-design.md b/docs/superpowers/specs/2026-10-24-service-runs-design.md
index 2472482ee..26488812d 100644
--- a/docs/superpowers/specs/2026-10-24-service-runs-design.md
+++ b/docs/superpowers/specs/2026-10-24-service-runs-design.md
@@ -144,7 +144,7 @@ readers ignore them:
| `by_service` | empty | the service name, for display |
| `by_credential` | empty | the credential's id, which the own-runs check uses |
| `on_behalf_of` | empty | the email the service gave, or empty for a scheduled run |
-| `idempotency_key` | optional (§ 3) | optional, and expected from Fleet |
+| `idempotency_key` | the key's scoped hash (§ 3), when one was sent | the key's scoped hash, expected from Fleet |
Every place the run is shown or audited (the CLI's follow output, the
console's run history, the account's audit log) shows a service's run as
@@ -167,8 +167,8 @@ the create response can't tell "started" from "never arrived". So:
- **The proof that a key was admitted lives in the instance record, written
by the same compare-and-swap that admits the run** (my call, after Astra's
review). The record gains `run_keys`: a short list of recent admissions,
- each with a hash of (caller scope, key), the run id, the request digest and
- the time. `provision.RunAdmit` refuses to admit when the list already holds
+ each with a hash of (caller scope, key), the run id, the request digest, a
+ hash of the caller scope (which the per-caller cap counts by) and the time. `provision.RunAdmit` refuses to admit when the list already holds
this caller's key, and returns that run instead.
- **Admission and dedupe are therefore one atomic write.** Two identical
creates at once can both write their objects, but only one is admitted.
@@ -205,9 +205,15 @@ the create response can't tell "started" from "never arrived". So:
- A 5xx, a timeout or a dropped connection means unknown, and the caller
retries with the same key.
- **How long a key lasts:** each admission prunes `run_keys` to entries younger
- than 24 hours, and to the newest 128 (the builder may tune both and must
- document them). A caller must not retry a key older than that. Fleet retries
- within minutes.
+ than 24 hours (`runs.KeyTTL`), then keeps the newest 32 entries of each
+ caller (`runs.MaxRunKeysPerCaller`). The cap is per caller, so one caller's
+ admissions can never evict another's key. A key is honoured for 24 hours,
+ or until that same caller has made 32 newer keyed admissions on this
+ instance, whichever comes first. A caller must not retry a key past that.
+ Fleet retries within minutes.
+- **`request.json` records the key** as `idempotency_key`: its scoped hash,
+ never the caller's own words. It links the run to its key after the record
+ has pruned the entry.
- **Mixed versions:** an older box binary rewriting the record would drop
`run_keys`. Keyed creates stay off (`503` `keys_not_ready`) until the console
is told, by config, that every record writer preserves the field.
diff --git a/internal/console/apiruns.go b/internal/console/apiruns.go
index 559368696..db80bf6e1 100644
--- a/internal/console/apiruns.go
+++ b/internal/console/apiruns.go
@@ -196,7 +196,7 @@ const runKeyReusedMsg = "this Idempotency-Key was used for a different run reque
const runFileHashRequiredMsg = "a run created with an Idempotency-Key must declare the sha256 of each file"
func runFileHashShapeMsg(name string) string {
- return fmt.Sprintf("the sha256 declared for %s is not 64 lowercase hex digits", name)
+ return fmt.Sprintf("the sha256 declared for %s is not 64 hex digits", name)
}
func runFileHashMismatchMsg(name string) string {
@@ -222,7 +222,7 @@ func runKeyOf(r *http.Request) (key string, sent, ok bool) {
}
// runFileHashRefusal is "" when meta's declared sha256s may stand: each one
-// declared is 64 lowercase hex digits, and a keyed create declares them all.
+// declared is 64 hex digits, and a keyed create declares them all.
func runFileHashRefusal(m runs.Meta, keyed bool) string {
for _, f := range m.Files {
switch {
@@ -450,6 +450,11 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg)
return
}
+ // A declared sha256 is compared and digested in the lowercase form the
+ // spool computes, so either case declares the same file.
+ for i := range meta.Files {
+ meta.Files[i].SHA256 = runs.NormalFileHash(meta.Files[i].SHA256)
+ }
if msg := runMetaRefusal(meta); msg != "" {
jsonErrorCode(w, http.StatusBadRequest, runMetaRefusalCode(msg), msg)
return
@@ -462,9 +467,9 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
jsonErrorCode(w, http.StatusBadRequest, runCodeInvalidRequest, msg)
return
}
- var scoped, digest string
+ var scoped, digest, caller string
if keyed {
- scoped, digest = runs.ScopedKey(scope, key), runs.MetaDigest(meta)
+ scoped, digest, caller = runs.ScopedKey(scope, key), runs.MetaDigest(meta), runs.CallerOf(scope)
if k, ok := rec.RunKeyFor(scoped, time.Now()); ok {
if k.Digest != digest {
jsonErrorCode(w, http.StatusConflict, runCodeKeyReused, runKeyReusedMsg)
@@ -525,7 +530,7 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
}
now := time.Now().UTC()
req := runs.Request{Schema: 1, ID: id, Account: acct.Sqid, App: app, Host: host, Incarnation: inc,
- Argv: meta.Argv, TimeoutS: meta.TimeoutS, By: actor, RequestedAt: now}
+ Argv: meta.Argv, TimeoutS: meta.TimeoutS, By: actor, RequestedAt: now, IdempotencyKey: scoped}
if p.Kind == PrincipalService {
req.ByService, req.ByCredential, req.OnBehalfOf = p.Name, p.CredentialID, meta.OnBehalfOf
}
@@ -547,22 +552,27 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
// read. The outcome is then settled from the record, not from the error.
wctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
defer cancel()
- state, err := s.admitRun(wctx, acct.Sqid, app, host, id, actor, scoped, digest)
- var replay *provision.RunKeyReplay
- if errors.As(err, &replay) {
- // A twin with the same key and request was admitted first (admitRun
- // answered a replay of this request's own id as its admission). The
- // twin's run is the answer; this request's own objects are no one's.
+ state, err := s.admitRun(wctx, acct.Sqid, app, host, id, actor, scoped, digest, caller)
+ // twinWon answers with the run a twin (the same key and request) had
+ // admitted first. This request's own objects are no one's.
+ twinWon := func(cur *provision.Record, twin string) {
if derr := runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id); derr != nil {
- log.Printf("console: api run %s/%s/%s: delete %s after its twin %s won: %v", acct.Sqid, app, host, id, replay.RunID, derr)
+ log.Printf("console: api run %s/%s/%s: delete %s after its twin %s won: %v", acct.Sqid, app, host, id, twin, derr)
}
+ runReplayed(w, cur, host, twin)
+ }
+ var replay *provision.RunKeyReplay
+ if errors.As(err, &replay) {
+ // admitRun answered a replay of this request's own id as its
+ // admission, so this one names a twin.
cur, _ := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host)
- runReplayed(w, cur, host, replay.RunID)
+ twinWon(cur, replay.RunID)
return
}
if err != nil && !runAdmitRefused(err) {
// The outcome is unknown (a lost response, a cancelled request, a
- // transport error after the write): ask the record whose run it holds.
+ // transport error after the write, or both CAS attempts lost): ask
+ // the record whose run it holds.
cur, rerr := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host)
switch {
case rerr != nil:
@@ -572,6 +582,23 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
log.Printf("console: api run %s/%s/%s: admit %s: %v; record unreadable: %v", acct.Sqid, app, host, id, err, rerr)
jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg)
return
+ case keyed:
+ // A keyed admission is settled by its key, which commits with it
+ // or not at all, never by ActiveRun, which a box or an ack may
+ // already have moved on: a committed run keeps its objects.
+ k, found := cur.RunKeyFor(scoped, time.Now())
+ switch {
+ case found && k.RunID == id:
+ state, err = "", nil
+ if cur.ActiveRun != nil && cur.ActiveRun.ID == id {
+ state = cur.ActiveRun.State
+ }
+ case found && k.Digest == digest:
+ twinWon(cur, k.RunID)
+ return
+ case found:
+ err = provision.ErrRunKeyReused
+ }
case cur.ActiveRun != nil && cur.ActiveRun.ID == id:
state, err = cur.ActiveRun.State, nil
}
@@ -605,15 +632,15 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: state})
}
-// admitRun is the admission CAS, with the caller's scoped key and request
-// digest when the create is keyed. A RunOpenError naming id itself means an
+// admitRun is the admission CAS, with the caller's scoped key, request
+// digest and runs.CallerOf when the create is keyed. A RunOpenError naming id itself means an
// earlier attempt of this same call committed (updateRecord retries after a
// lost-response conflict, and finds its own run queued): that is admitted.
// So is a RunKeyReplay naming id, the same retry finding its own key; any
// other RunKeyReplay is returned for the caller to answer.
-func (s *Server) admitRun(ctx context.Context, sqid, app, host, id, by, key, digest string) (string, error) {
+func (s *Server) admitRun(ctx context.Context, sqid, app, host, id, by, key, digest, caller string) (string, error) {
got, err := provision.SetRecordRun(ctx, s.cfg.Releases, sqid, app, host,
- provision.RunAdmit(provision.RunAdmission{ID: id, By: by, Deadline: time.Now().UTC().Add(runs.PickupWindow), Key: key, Digest: digest}))
+ provision.RunAdmit(provision.RunAdmission{ID: id, By: by, Deadline: time.Now().UTC().Add(runs.PickupWindow), Key: key, Digest: digest, Caller: caller}))
var open *provision.RunOpenError
var replay *provision.RunKeyReplay
switch {
diff --git a/internal/console/apiruns_key_test.go b/internal/console/apiruns_key_test.go
index d04543788..98d4ba761 100644
--- a/internal/console/apiruns_key_test.go
+++ b/internal/console/apiruns_key_test.go
@@ -196,6 +196,15 @@ func TestRunKeyedCreateThenRetryReturnsTheSameRun(t *testing.T) {
if err != nil || len(req.Files) != 1 || req.Files[0].SHA256 != sha("a,b,c") {
t.Fatalf("request %+v, %v", req, err)
}
+ // request.json links the run to its key by the scoped hash, never the
+ // caller's own words.
+ if req.IdempotencyKey != runs.ScopedKey("user:alice@keymail.dev", testKey) {
+ t.Fatalf("request.json idempotency_key %q, want the scoped hash", req.IdempotencyKey)
+ }
+ raw, _, err := fx.rel.Get(context.Background(), runs.RequestKey(release.OpsAccount, "shop", first.ID))
+ if err != nil || strings.Contains(string(raw), testKey) {
+ t.Fatalf("request.json holds the raw key: %s, %v", raw, err)
+ }
}
func TestRunKeyedRetryWhileOpenReturnsTheRun(t *testing.T) {
@@ -575,7 +584,7 @@ func TestAdmitRunOwnKeyReplayIsAdmitted(t *testing.T) {
if err != nil {
t.Fatal(err)
}
- state, err := fx.srv.admitRun(context.Background(), release.OpsAccount, "shop", runTestHost, id, "alice@keymail.dev", "K", "D")
+ state, err := fx.srv.admitRun(context.Background(), release.OpsAccount, "shop", runTestHost, id, "alice@keymail.dev", "K", "D", "C")
if err != nil || state != provision.RunQueued {
t.Fatalf("admitRun = %q, %v; want queued and no error", state, err)
}
@@ -599,6 +608,9 @@ func TestKeyedCreateRefusedUntilKeysAreOn(t *testing.T) {
if c.replayed() {
t.Fatalf("an unkeyed create: %+v", c)
}
+ if req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", c.ID); err != nil || req.IdempotencyKey != "" {
+ t.Fatalf("an unkeyed request: %+v, %v", req, err)
+ }
if rec := fx.instRecord(t); len(rec.RunKeys) != 0 {
t.Fatalf("an unkeyed create recorded a key: %+v", rec.RunKeys)
}
@@ -625,3 +637,138 @@ func TestRunKeyCodesAreListed(t *testing.T) {
}
}
}
+
+// movedOnStore commits the admission's record write, then has the run picked,
+// finished and cleared (a box and an ack, faster than the answer), and
+// answers err: the record's ActiveRun no longer names the run when the door
+// settles the lost answer.
+type movedOnStore struct {
+ release.Store
+ recordKey string
+ err error
+ once sync.Once
+ t *testing.T
+}
+
+func (m *movedOnStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ etag, err := m.Store.Put(ctx, key, body, ifMatch)
+ if err != nil || key != m.recordKey {
+ return etag, err
+ }
+ m.once.Do(func() {
+ rec, gerr := provision.DecodeRecord(body)
+ if gerr != nil || rec.ActiveRun == nil {
+ m.t.Errorf("movedOnStore: %v %+v", gerr, rec.ActiveRun)
+ return
+ }
+ id := rec.ActiveRun.ID
+ for _, mut := range []func(*provision.Record) error{
+ provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)),
+ provision.RunFinish(id),
+ provision.RunClear(id),
+ } {
+ if _, e := provision.SetRecordRun(context.Background(), m.Store, release.OpsAccount, "shop", runTestHost, mut); e != nil {
+ m.t.Errorf("movedOnStore: %v", e)
+ }
+ }
+ err = m.err
+ })
+ return etag, err
+}
+
+// A keyed admission whose answer was lost is settled by its key, not by
+// ActiveRun: the record has moved on, the key still names this run, so the
+// door answers it admitted and keeps its objects.
+func TestRunKeyedLostAdmissionSettledByItsKey(t *testing.T) {
+ fx := keyedFixture(t)
+ fx.srv.cfg.Releases = &movedOnStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost),
+ err: errors.New("connection reset by peer"), t: t}
+ c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
+ if c.replayed() || c.State != "" {
+ t.Fatalf("created %+v, want a fresh admission whose run the record no longer names", c)
+ }
+ rec := fx.instRecord(t)
+ if rec.ActiveRun != nil || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID {
+ t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys)
+ }
+ if keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)); len(keys) != 2 {
+ t.Fatalf("the committed run's objects are %q, want request.json and in/x.csv", keys)
+ }
+ if n := fx.runStarts(t); n != 1 {
+ t.Errorf("%d run-start rows, want 1", n)
+ }
+}
+
+// twinOnRetryStore loses both of updateRecord's attempts: the first record
+// write is refused as a conflict without landing, and before the second a
+// twin's keyed admission lands, so the second conflicts for real.
+type twinOnRetryStore struct {
+ release.Store
+ recordKey string
+ twin provision.RunAdmission
+ puts int
+ t *testing.T
+}
+
+func (s *twinOnRetryStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if key == s.recordKey {
+ s.puts++
+ switch s.puts {
+ case 1:
+ return "", fmt.Errorf("s3store: record: %w", release.ErrCASConflict)
+ case 2:
+ if _, err := provision.SetRecordRun(context.Background(), s.Store, release.OpsAccount, "shop", runTestHost, provision.RunAdmit(s.twin)); err != nil {
+ s.t.Errorf("twin admission: %v", err)
+ }
+ }
+ }
+ return s.Store.Put(ctx, key, body, ifMatch)
+}
+
+// A keyed create that loses both CAS attempts to a twin answers with the
+// twin's run, replayed, and deletes only its own objects: not a 502.
+func TestRunKeyedCreateLosingBothAttemptsAnswersTheTwin(t *testing.T) {
+ fx := keyedFixture(t)
+ scope := "user:alice@keymail.dev"
+ twinID, err := runs.MintID()
+ if err != nil {
+ t.Fatal(err)
+ }
+ st := &twinOnRetryStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), t: t,
+ twin: provision.RunAdmission{ID: twinID, By: "alice@keymail.dev", Deadline: time.Now().Add(time.Minute),
+ Key: runs.ScopedKey(scope, testKey), Digest: runs.MetaDigest(okMeta()), Caller: runs.CallerOf(scope)}}
+ fx.srv.cfg.Releases = st
+ c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil)
+ if st.puts != 2 {
+ t.Fatalf("%d record writes, want the two lost attempts", st.puts)
+ }
+ if c.ID != twinID || !c.replayed() || c.State != provision.RunQueued {
+ t.Fatalf("created %+v, want the twin %s, replayed and queued", c, twinID)
+ }
+ if ids := fx.runIDsWithObjects(t); len(ids) != 0 {
+ t.Fatalf("runs with objects %v: the loser's objects must be gone", ids)
+ }
+ if n := fx.runStarts(t); n != 0 {
+ t.Errorf("%d run-start rows, want none from the loser", n)
+ }
+}
+
+// A declared sha256 may be in either case: it is lowercased before it is
+// compared or digested, so an uppercase declaration is not refused and a
+// retry in the other case is the same request.
+func TestRunDeclaredHashIsCaseInsensitive(t *testing.T) {
+ fx := keyedFixture(t)
+ upper := fileMeta("a,b,c")
+ upper.Files[0].SHA256 = strings.ToUpper(upper.Files[0].SHA256)
+ plain := fx.createKeyed(t, fx.member, "", upper, xcsv("a,b,c"))
+ fx.finishAndAck(t, fx.member, plain.ID)
+ first := fx.createKeyed(t, fx.member, testKey, upper, xcsv("a,b,c"))
+ again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c"))
+ if again.ID != first.ID || !again.replayed() {
+ t.Fatalf("a lowercase retry of an uppercase create: %+v, want %s replayed", again, first.ID)
+ }
+ req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID)
+ if err != nil || req.Files[0].SHA256 != sha("a,b,c") {
+ t.Fatalf("request %+v, %v", req, err)
+ }
+}
diff --git a/internal/provision/run.go b/internal/provision/run.go
index 4cf0a4b8f..51d334f42 100644
--- a/internal/provision/run.go
+++ b/internal/provision/run.go
@@ -87,19 +87,22 @@ func SetRecordRun(ctx context.Context, s release.Store, acct, app, host string,
// RunAdmission is what the console decided for a new run. Key is the
// caller's scoped Idempotency-Key (runs.ScopedKey), "" for an unkeyed
-// create; Digest is the request's runs.MetaDigest, set with Key.
+// create; Digest is the request's runs.MetaDigest and Caller the caller's
+// runs.CallerOf, both set with Key.
type RunAdmission struct {
- ID, By string
- Deadline time.Time // the pickup deadline
- Key, Digest string
+ ID, By string
+ Deadline time.Time // the pickup deadline
+ Key, Digest, Caller string
}
// RunKey is one keyed admission on the record: the scoped key, the run it
-// admitted, the request's digest and when.
+// admitted, the request's digest, the caller (runs.CallerOf, which the
+// per-caller cap counts by) and when.
type RunKey struct {
Key string `json:"key"`
RunID string `json:"run_id"`
Digest string `json:"digest"`
+ Caller string `json:"caller"`
At time.Time `json:"at"`
}
@@ -127,18 +130,33 @@ func (r Record) RunKeyFor(key string, now time.Time) (RunKey, bool) {
return RunKey{}, false
}
-// prunedRunKeys is keys without the entries runs.KeyTTL old or older, in
-// age order (oldest first), trimmed to the newest runs.MaxRunKeys.
+// prunedRunKeys is keys without the entries runs.KeyTTL old or older, then
+// with only the newest runs.MaxRunKeysPerCaller of each caller, in age order
+// (oldest first). The cap is per caller so that no caller's admissions can
+// evict another caller's key: a key is honoured for runs.KeyTTL, or until
+// that same caller makes runs.MaxRunKeysPerCaller newer keyed admissions on
+// this instance, whichever comes first.
func prunedRunKeys(keys []RunKey, now time.Time) []RunKey {
- var out []RunKey
+ var fresh []RunKey
for _, k := range keys {
if now.Sub(k.At) < runs.KeyTTL {
- out = append(out, k)
+ fresh = append(fresh, k)
+ }
+ }
+ sort.SliceStable(fresh, func(i, j int) bool { return fresh[i].At.Before(fresh[j].At) })
+ kept := make([]bool, len(fresh))
+ per := map[string]int{}
+ for i := len(fresh) - 1; i >= 0; i-- {
+ if per[fresh[i].Caller] < runs.MaxRunKeysPerCaller {
+ per[fresh[i].Caller]++
+ kept[i] = true
}
}
- sort.SliceStable(out, func(i, j int) bool { return out[i].At.Before(out[j].At) })
- if len(out) > runs.MaxRunKeys {
- out = out[len(out)-runs.MaxRunKeys:]
+ var out []RunKey
+ for i, k := range fresh {
+ if kept[i] {
+ out = append(out, k)
+ }
}
return out
}
@@ -151,14 +169,14 @@ func prunedRunKeys(keys []RunKey, now time.Time) []RunKey {
// is answered even while its own run is open: the key already admitted with
// the same digest is a *RunKeyReplay, with another digest ErrRunKeyReused.
// A keyed admission that succeeds records its key in the same write.
-// run_keys is pruned by age on every admission and by count when one is
-// added; nothing else changes it.
+// run_keys is pruned by age on every admission and by each caller's count
+// when one is added; nothing else changes it.
func RunAdmit(p RunAdmission) func(*Record) error {
return func(rec *Record) error {
now := time.Now().UTC()
keys := prunedRunKeys(rec.RunKeys, now)
if p.Key != "" {
- if p.Digest == "" {
+ if p.Digest == "" || p.Caller == "" {
return fmt.Errorf("%w: malformed admission", ErrRunConflict)
}
if k, ok := (Record{RunKeys: keys}).RunKeyFor(p.Key, now); ok {
@@ -185,7 +203,7 @@ func RunAdmit(p RunAdmission) func(*Record) error {
}
rec.ActiveRun = &RunState{ID: p.ID, State: RunQueued, By: p.By, Deadline: p.Deadline.UTC()}
if p.Key != "" {
- keys = prunedRunKeys(append(keys, RunKey{Key: p.Key, RunID: p.ID, Digest: p.Digest, At: now}), now)
+ keys = prunedRunKeys(append(keys, RunKey{Key: p.Key, RunID: p.ID, Digest: p.Digest, Caller: p.Caller, At: now}), now)
}
if len(keys) == 0 {
keys = nil // an empty list stays absent from the record
diff --git a/internal/provision/run_keys_test.go b/internal/provision/run_keys_test.go
index 3f879de11..d63d3b225 100644
--- a/internal/provision/run_keys_test.go
+++ b/internal/provision/run_keys_test.go
@@ -19,9 +19,12 @@ import (
"github.com/carlosframework/platform/internal/seeds"
)
-// keyed is an admission that carries key k and digest d.
-func keyed(id, k, d string) RunAdmission {
- return RunAdmission{ID: id, By: "svc", Deadline: time.Now().Add(time.Minute), Key: k, Digest: d}
+// keyed is caller A's admission that carries key k and digest d.
+func keyed(id, k, d string) RunAdmission { return keyedBy("A", id, k, d) }
+
+// keyedBy is keyed for caller.
+func keyedBy(caller, id, k, d string) RunAdmission {
+ return RunAdmission{ID: id, By: "svc", Deadline: time.Now().Add(time.Minute), Key: k, Digest: d, Caller: caller}
}
func wantReplay(t *testing.T, what string, err error, id string) {
@@ -102,9 +105,9 @@ func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
rec := runTestRecord(t, s)
now := time.Now().UTC()
var keys []RunKey
- keys = append(keys, RunKey{Key: "old", RunID: "run_old", Digest: "D", At: now.Add(-runs.KeyTTL - time.Minute)})
- for i := 0; i < runs.MaxRunKeys+2; i++ {
- keys = append(keys, RunKey{Key: fmt.Sprintf("k%03d", i), RunID: fmt.Sprintf("run_%03d", i), Digest: "D", At: now.Add(-time.Hour + time.Duration(i)*time.Second)})
+ keys = append(keys, RunKey{Key: "old", RunID: "run_old", Digest: "D", Caller: "A", At: now.Add(-runs.KeyTTL - time.Minute)})
+ for i := 0; i < runs.MaxRunKeysPerCaller+2; i++ {
+ keys = append(keys, RunKey{Key: fmt.Sprintf("k%03d", i), RunID: fmt.Sprintf("run_%03d", i), Digest: "D", Caller: "A", At: now.Add(-time.Hour + time.Duration(i)*time.Second)})
}
putRunKeys(t, s, rec, keys)
@@ -113,8 +116,8 @@ func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
if err != nil || got.ActiveRun.ID != "run_new" {
t.Fatalf("an expired key must admit afresh: %+v, %v", got, err)
}
- if len(got.RunKeys) != runs.MaxRunKeys {
- t.Fatalf("%d keys kept, want %d", len(got.RunKeys), runs.MaxRunKeys)
+ if len(got.RunKeys) != runs.MaxRunKeysPerCaller {
+ t.Fatalf("%d keys kept, want %d", len(got.RunKeys), runs.MaxRunKeysPerCaller)
}
kept := map[string]bool{}
for _, k := range got.RunKeys {
@@ -126,8 +129,8 @@ func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
if !kept["old/run_new"] {
t.Error("the newest entry was not kept")
}
- // 130 fresh + 1 new, newest 128 kept: k000, k001 and k002 go.
- for i := 0; i < runs.MaxRunKeys+2; i++ {
+ // 34 fresh + 1 new, the caller's newest 32 kept: k000, k001 and k002 go.
+ for i := 0; i < runs.MaxRunKeysPerCaller+2; i++ {
name := fmt.Sprintf("k%03d/run_%03d", i, i)
if want := i >= 3; kept[name] != want {
t.Errorf("%s kept %v, want %v", name, kept[name], want)
@@ -138,8 +141,8 @@ func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
s2 := release.NewDirStore(t.TempDir())
rec2 := runTestRecord(t, s2)
putRunKeys(t, s2, rec2, []RunKey{
- {Key: "old", RunID: "run_old", Digest: "D", At: now.Add(-runs.KeyTTL - time.Minute)},
- {Key: "a", RunID: "run_a", Digest: "D", At: now.Add(-time.Hour)},
+ {Key: "old", RunID: "run_old", Digest: "D", Caller: "A", At: now.Add(-runs.KeyTTL - time.Minute)},
+ {Key: "a", RunID: "run_a", Digest: "D", Caller: "A", At: now.Add(-time.Hour)},
})
got, err = setRun(t, s2, rec2, RunAdmit(keyed("run_b", "b", "D")))
if err != nil || len(got.RunKeys) != 2 || got.RunKeys[0].Key != "a" || got.RunKeys[1].Key != "b" {
@@ -157,12 +160,59 @@ func TestRunKeysPrunedByAgeAndCount(t *testing.T) {
}
}
+// TestRunKeysCappedPerCaller: the count cap is each caller's own, so caller
+// B's admissions never evict caller A's key, and A's own 33rd admission
+// evicts A's oldest.
+func TestRunKeysCappedPerCaller(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec := runTestRecord(t, s)
+ admit := func(a RunAdmission) {
+ t.Helper()
+ if _, err := setRun(t, s, rec, RunAdmit(a)); err != nil {
+ t.Fatalf("admit %s: %v", a.ID, err)
+ }
+ if _, err := setRun(t, s, rec, RunExpire(a.ID)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ admit(keyedBy("A", "run_a0", "KA", "D"))
+ for i := range 40 {
+ admit(keyedBy("B", fmt.Sprintf("run_b%02d", i), fmt.Sprintf("KB%02d", i), "D"))
+ }
+ _, err := setRun(t, s, rec, RunAdmit(keyedBy("A", "run_x", "KA", "D")))
+ wantReplay(t, "A's key after B's 40 admissions", err, "run_a0")
+ cur, err := GetRecord(context.Background(), s, rec.Account, rec.App, rec.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ per := map[string]int{}
+ for _, k := range cur.RunKeys {
+ per[k.Caller]++
+ }
+ if per["A"] != 1 || per["B"] != runs.MaxRunKeysPerCaller {
+ t.Fatalf("entries per caller %v, want A 1 and B %d", per, runs.MaxRunKeysPerCaller)
+ }
+ for i := 1; i < runs.MaxRunKeysPerCaller; i++ {
+ admit(keyedBy("A", fmt.Sprintf("run_a%02d", i), fmt.Sprintf("KA%02d", i), "D"))
+ }
+ _, err = setRun(t, s, rec, RunAdmit(keyedBy("A", "run_y", "KA", "D")))
+ wantReplay(t, "A's key with 32 of A's own", err, "run_a0")
+ admit(keyedBy("A", "run_a32", "KA32", "D")) // A's 33rd
+ got, err := setRun(t, s, rec, RunAdmit(keyedBy("A", "run_z", "KA", "D")))
+ if err != nil || got.ActiveRun.ID != "run_z" {
+ t.Fatalf("A's oldest key after A's 33rd admission: %+v, %v; want it evicted and a fresh run", got, err)
+ }
+}
+
func TestRunAdmitRefusesAMalformedKeyedAdmission(t *testing.T) {
s := release.NewDirStore(t.TempDir())
rec := runTestRecord(t, s)
if _, err := setRun(t, s, rec, RunAdmit(keyed("run_a", "K", ""))); !errors.Is(err, ErrRunConflict) {
t.Fatalf("a key without a digest: %v, want ErrRunConflict", err)
}
+ if _, err := setRun(t, s, rec, RunAdmit(keyedBy("", "run_a", "K", "D"))); !errors.Is(err, ErrRunConflict) {
+ t.Fatalf("a key without a caller: %v, want ErrRunConflict", err)
+ }
}
// keysJSON is the record's run_keys as stored, byte for byte.
diff --git a/internal/runs/keys.go b/internal/runs/keys.go
index 0f24183cc..e697bd388 100644
--- a/internal/runs/keys.go
+++ b/internal/runs/keys.go
@@ -5,16 +5,20 @@ import (
"encoding/binary"
"encoding/hex"
"hash"
+ "strings"
"time"
)
// A keyed create's bounds (service runs spec § 3). Each admission prunes the
// record's run_keys to entries younger than KeyTTL, then keeps the newest
-// MaxRunKeys. A caller must not retry a key older than KeyTTL: it is a fresh
-// run then.
+// MaxRunKeysPerCaller entries of each caller. So a key is honoured for
+// KeyTTL, or until that same caller makes MaxRunKeysPerCaller newer keyed
+// admissions on this instance, whichever comes first. One caller's
+// admissions never evict another's key. A caller must not retry a key past
+// that: it is a fresh run then.
const (
- KeyTTL = 24 * time.Hour
- MaxRunKeys = 128
+ KeyTTL = 24 * time.Hour
+ MaxRunKeysPerCaller = 32
// MaxKeyBytes is the longest Idempotency-Key a caller may send.
MaxKeyBytes = 128
)
@@ -45,25 +49,38 @@ func ScopedKey(scope, key string) string {
return hex.EncodeToString(h[:])
}
-// ValidFileHash reports whether h is a sha256 as a create declares it: 64
-// lowercase hex digits.
+// CallerOf is the caller's name as a run_keys entry stores it:
+// hex(sha256("caller", NUL, scope)). It groups a caller's entries for the
+// per-caller cap without the record holding the caller's identity.
+func CallerOf(scope string) string {
+ h := sha256.Sum256([]byte("caller\x00" + scope))
+ return hex.EncodeToString(h[:])
+}
+
+// ValidFileHash reports whether h is a sha256 as a create may declare it: 64
+// hex digits, in either case. The console lowercases it (NormalFileHash)
+// before comparing it or digesting it.
func ValidFileHash(h string) bool {
if len(h) != 64 {
return false
}
for i := 0; i < len(h); i++ {
- if c := h[i]; !('0' <= c && c <= '9' || 'a' <= c && c <= 'f') {
+ if c := h[i]; !('0' <= c && c <= '9' || 'a' <= c && c <= 'f' || 'A' <= c && c <= 'F') {
return false
}
}
return true
}
+// NormalFileHash is a declared sha256 in the lowercase form the console
+// computes.
+func NormalFileHash(h string) string { return strings.ToLower(h) }
+
// MetaDigest is the hex sha256 of a canonical encoding of what a create
// asks for: argv, timeout_s, on_behalf_of, and each file's name, size and
-// sha256, in meta's order. Every string and count is length-prefixed, so no
-// two different requests share an encoding. A retry with the same key must
-// have the same digest.
+// sha256 (lowercased), in meta's order. Every string and count is
+// length-prefixed, so no two different requests share an encoding. A retry
+// with the same key must have the same digest.
func MetaDigest(m Meta) string {
h := sha256.New()
str(h, "carlos-run-meta-v1")
@@ -77,7 +94,7 @@ func MetaDigest(m Meta) string {
for _, f := range m.Files {
str(h, f.Name)
num(h, uint64(f.Size))
- str(h, f.SHA256)
+ str(h, NormalFileHash(f.SHA256))
}
return hex.EncodeToString(h.Sum(nil))
}
diff --git a/internal/runs/keys_test.go b/internal/runs/keys_test.go
index 7e8d44ef1..120991aef 100644
--- a/internal/runs/keys_test.go
+++ b/internal/runs/keys_test.go
@@ -89,3 +89,35 @@ func TestMetaDigestCoversEveryField(t *testing.T) {
}
}
}
+
+func TestValidFileHash(t *testing.T) {
+ lower := strings.Repeat("0123456789abcdef", 4)
+ for _, h := range []string{lower, strings.ToUpper(lower), "ABCDEF" + lower[6:]} {
+ if !ValidFileHash(h) {
+ t.Errorf("ValidFileHash(%q) = false", h)
+ }
+ }
+ for _, h := range []string{"", lower[:63], lower + "0", "g" + lower[1:], " " + lower[1:]} {
+ if ValidFileHash(h) {
+ t.Errorf("ValidFileHash(%q) = true", h)
+ }
+ }
+ if NormalFileHash(strings.ToUpper(lower)) != lower {
+ t.Error("NormalFileHash does not lowercase")
+ }
+ m := Meta{Argv: []string{"a"}, TimeoutS: 1, Files: []MetaFile{{Name: "x", Size: 1, SHA256: lower}}}
+ u := m
+ u.Files = []MetaFile{{Name: "x", Size: 1, SHA256: strings.ToUpper(lower)}}
+ if MetaDigest(m) != MetaDigest(u) {
+ t.Error("a declared sha256's case changes the digest")
+ }
+}
+
+func TestCallerOfSeparatesCallers(t *testing.T) {
+ if CallerOf("svc:a") == CallerOf("user:a") || CallerOf("svc:a") != CallerOf("svc:a") || len(CallerOf("x")) != 64 {
+ t.Fatal("CallerOf does not name callers apart")
+ }
+ if CallerOf("svc:a") == ScopedKey("svc:a", "") {
+ t.Fatal("CallerOf collides with a scoped key")
+ }
+}
diff --git a/internal/runs/runs.go b/internal/runs/runs.go
index 9c07386fc..f0e68b526 100644
--- a/internal/runs/runs.go
+++ b/internal/runs/runs.go
@@ -190,6 +190,11 @@ type Request struct {
ByService string `json:"by_service,omitempty"`
ByCredential string `json:"by_credential,omitempty"`
OnBehalfOf string `json:"on_behalf_of,omitempty"`
+ // IdempotencyKey is a keyed create's key as the record's run_keys holds
+ // it, scoped to its caller and hashed (ScopedKey), never the caller's
+ // own words. It links the run to its key after the record has pruned
+ // the entry (service runs spec §§ 2, 3). Empty for an unkeyed create.
+ IdempotencyKey string `json:"idempotency_key,omitempty"`
}
// Timeout is the run's timeout as a duration.