amadan
console: a keyed run create is admitted once, settled by the instance record, and a retry gets the same run (PRO-6214)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
16 files changed,
+1582
−31
cmd/console/main.go+1 −0internal/console/apiruns.go+222 −17internal/console/apiruns_key_test.go+627 −0internal/console/apiruns_service_test.go+54 −0internal/console/authcli.go+1 −0internal/console/claim_invariant_fixtures_test.go+6 −0internal/console/server.go+7 −0internal/console/svcroutes_test.go+4 −2internal/console/testdata/claim-routes.txt+1 −0internal/provision/provision.go+9 −0internal/provision/run.go+84 −3internal/provision/run_keys_test.go+359 −0internal/runs/keys.go+94 −0internal/runs/keys_test.go+91 −0internal/runs/runs.go+17 −4internal/runs/runs_test.go+5 −5
diff --git a/cmd/console/main.go b/cmd/console/main.go| index c95cfa05f..8df9b8aea 100644 |
| --- a/cmd/console/main.go |
| +++ b/cmd/console/main.go |
| @@ -382,6 +382,7 @@ func serve() { |
| // The carlos db door's data bucket and this box's placement |
| // (dbDoorWiring). |
| HostClaim: hostClaim, |
| + RunKeysEnabled: os.Getenv("CARLOS_RUN_KEYS") == "on", // keyed run creates, once every box carries run_keys |
| HostConsoleID: os.Getenv("CARLOS_CONSOLE_ID"), |
| DBVersions: dbVersions, |
| HibernateBucket: dbBucket, |
diff --git a/internal/console/apiruns.go b/internal/console/apiruns.go| index 798d6093d..559368696 100644 |
| --- a/internal/console/apiruns.go |
| +++ b/internal/console/apiruns.go |
| @@ -48,6 +48,10 @@ const ( |
| runCodeRunEnded = "run_ended" |
| runCodeRunNotFinished = "run_not_finished" |
| runCodeLastSeqMismatch = "last_seq_mismatch" |
| + runCodeKeyInvalid = "idempotency_key_invalid" |
| + runCodeKeyReused = "idempotency_key_reused" |
| + runCodeKeysNotReady = "keys_not_ready" |
| + runCodeFileHashMismatch = "file_hash_mismatch" |
| ) |
| // runRefusalCodes is every code above, the one list the docs and the tests |
| @@ -58,6 +62,7 @@ var runRefusalCodes = []string{ |
| runCodeRunOpen, runCodeMalformed, runCodeInvalidRequest, runCodeFilesTooLarge, |
| runCodeImportUncommitted, runCodeCommandNotGranted, runCodeOnBehalfOfInvalid, |
| runCodeRunStarted, runCodeRunEnded, runCodeRunNotFinished, runCodeLastSeqMismatch, |
| + runCodeKeyInvalid, runCodeKeyReused, runCodeKeysNotReady, runCodeFileHashMismatch, |
| } |
| // runCaller reports whether p may use the run doors at all: a person (whose |
| @@ -178,6 +183,68 @@ const runRecordFailedMsg = "could not record the run. try again" |
| const runCommandNotGrantedMsg = "this credential may not run that command. it runs only the commands it was minted with" |
| +// runKeyHeader is the name of a create's idempotency key (service runs spec |
| +// § 3). |
| +const runKeyHeader = "Idempotency-Key" |
| + |
| +const runKeysNotReadyMsg = "this console does not take an Idempotency-Key on runs yet. try again later" |
| + |
| +const runKeyInvalidMsg = "an Idempotency-Key is 1 to 128 characters: letters, digits, dots, underscores, colons and hyphens" |
| + |
| +const runKeyReusedMsg = "this Idempotency-Key was used for a different run request. use a new key for a new request" |
| + |
| +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) |
| +} |
| + |
| +func runFileHashMismatchMsg(name string) string { |
| + return fmt.Sprintf("%s does not match the sha256 declared for it", name) |
| +} |
| + |
| +func runNoKeyMsg(host string) string { |
| + return fmt.Sprintf("%s has no run started with that Idempotency-Key", host) |
| +} |
| + |
| +// runKeyOf reads the create's Idempotency-Key. sent reports whether the |
| +// header was there at all, so an empty, repeated or malformed key is |
| +// refused (ok false), never taken for an unkeyed create. |
| +func runKeyOf(r *http.Request) (key string, sent, ok bool) { |
| + vals := r.Header.Values(runKeyHeader) |
| + switch { |
| + case len(vals) == 0: |
| + return "", false, false |
| + case len(vals) != 1 || !runs.ValidKey(vals[0]): |
| + return "", true, false |
| + } |
| + return vals[0], true, true |
| +} |
| + |
| +// runFileHashRefusal is "" when meta's declared sha256s may stand: each one |
| +// declared is 64 lowercase hex digits, and a keyed create declares them all. |
| +func runFileHashRefusal(m runs.Meta, keyed bool) string { |
| + for _, f := range m.Files { |
| + switch { |
| + case f.SHA256 == "" && keyed: |
| + return runFileHashRequiredMsg |
| + case f.SHA256 != "" && !runs.ValidFileHash(f.SHA256): |
| + return runFileHashShapeMsg(f.Name) |
| + } |
| + } |
| + return "" |
| +} |
| + |
| +// runReplayed answers a create whose key already admitted run id: 202 with |
| +// that run, replayed, and its state while the record still names it. |
| +func runReplayed(w http.ResponseWriter, rec *provision.Record, host, id string) { |
| + state := "" |
| + if rec != nil && rec.ActiveRun != nil && rec.ActiveRun.ID == id { |
| + state = rec.ActiveRun.State |
| + } |
| + writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: state, Replayed: true}) |
| +} |
| + |
| // runMetaRefusalCode is the code for a refusal runMetaRefusal named. |
| func runMetaRefusalCode(msg string) string { |
| if msg == runFilesTooLargeMsg { |
| @@ -288,6 +355,14 @@ var errRunMalformed = errors.New("console: run request disagrees with its meta") |
| // handleAPIRunCreate is POST …/instances/{host}/runs (run spec §5.1). The |
| // refusals come in the spec's order; nothing is written before every one |
| // that needs no body has passed. |
| +// |
| +// A create that carries an Idempotency-Key (service runs spec § 3) reads its |
| +// meta straight after the record, and answers a key the record already |
| +// holds before any other refusal and without reading a file: the same |
| +// request is the run that key admitted, a different one is refused. Only |
| +// then come the record's refusals, the command check and the files, whose |
| +// bytes must match the sha256 meta declares. The admission records the key |
| +// in the same write. An unkeyed create keeps the order it had before keys. |
| func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| p, ok := principalFrom(r.Context()) |
| if !ok || !runCaller(p) { |
| @@ -306,6 +381,17 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg) |
| return |
| } |
| + key, keyed, keyOK := runKeyOf(r) |
| + if keyed && !s.cfg.RunKeysEnabled { |
| + // Never run as unkeyed: the caller would retry it into a second run. |
| + jsonErrorCode(w, http.StatusServiceUnavailable, runCodeKeysNotReady, runKeysNotReadyMsg) |
| + return |
| + } |
| + scope := callerScope(p) |
| + if keyed && (!keyOK || scope == "") { |
| + jsonErrorCode(w, http.StatusBadRequest, runCodeKeyInvalid, runKeyInvalidMsg) |
| + return |
| + } |
| if s.cfg.Releases == nil { |
| jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runRecordFailedMsg) |
| return |
| @@ -327,16 +413,24 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| return |
| } |
| - if msg := runClassRefusal(*rec); msg != "" { |
| - jsonErrorCode(w, http.StatusConflict, runCodeNotRunnable, msg) |
| - return |
| - } |
| - if rec.Move != nil { |
| - jsonErrorCode(w, http.StatusConflict, runCodeMoving, runMovingMsg(host)) |
| - return |
| + // recordRefused writes refusals 5 to 7 from the record (the class, a move |
| + // in flight, an open run) and reports whether it wrote one. |
| + recordRefused := func() bool { |
| + if msg := runClassRefusal(*rec); msg != "" { |
| + jsonErrorCode(w, http.StatusConflict, runCodeNotRunnable, msg) |
| + return true |
| + } |
| + if rec.Move != nil { |
| + jsonErrorCode(w, http.StatusConflict, runCodeMoving, runMovingMsg(host)) |
| + return true |
| + } |
| + if rec.ActiveRun.Open() { |
| + jsonErrorCode(w, http.StatusConflict, runCodeRunOpen, s.runOpenRefusal(ctx, p, acct.Sqid, app, host, *rec.ActiveRun)) |
| + return true |
| + } |
| + return false |
| } |
| - if rec.ActiveRun.Open() { |
| - jsonErrorCode(w, http.StatusConflict, runCodeRunOpen, s.runOpenRefusal(ctx, p, acct.Sqid, app, host, *rec.ActiveRun)) |
| + if !keyed && recordRefused() { |
| return |
| } |
| r.Body = http.MaxBytesReader(w, r.Body, runs.MaxFileBytes+runs.MetaAllowance) |
| @@ -364,13 +458,37 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| jsonErrorCode(w, http.StatusBadRequest, runCodeOnBehalfOfInvalid, msg) |
| return |
| } |
| + if msg := runFileHashRefusal(meta, keyed); msg != "" { |
| + jsonErrorCode(w, http.StatusBadRequest, runCodeInvalidRequest, msg) |
| + return |
| + } |
| + var scoped, digest string |
| + if keyed { |
| + scoped, digest = runs.ScopedKey(scope, key), runs.MetaDigest(meta) |
| + if k, ok := rec.RunKeyFor(scoped, time.Now()); ok { |
| + if k.Digest != digest { |
| + jsonErrorCode(w, http.StatusConflict, runCodeKeyReused, runKeyReusedMsg) |
| + return |
| + } |
| + // The files are never read: the run that key admitted has them. |
| + runReplayed(w, rec, host, k.RunID) |
| + return |
| + } |
| + if recordRefused() { |
| + return |
| + } |
| + if rec.ImportUncommitted() { |
| + jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host)) |
| + return |
| + } |
| + } |
| // A run service runs only what it was granted, refused before any |
| // object is written. A person runs any command. |
| if p.Kind == PrincipalService && !runs.CommandGranted(meta.Argv, p.Commands) { |
| jsonErrorCode(w, http.StatusForbidden, runCodeCommandNotGranted, runCommandNotGrantedMsg) |
| return |
| } |
| - if rec.ImportUncommitted() { |
| + if !keyed && rec.ImportUncommitted() { |
| jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host)) |
| return |
| } |
| @@ -389,6 +507,12 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| jsonErrorCode(w, http.StatusBadRequest, runCodeMalformed, runMalformedMsg) |
| return |
| } |
| + for i, f := range files { |
| + if want := meta.Files[i].SHA256; want != "" && f.SHA256 != want { |
| + jsonErrorCode(w, http.StatusBadRequest, runCodeFileHashMismatch, runFileHashMismatchMsg(f.Name)) |
| + return |
| + } |
| + } |
| id, err := runs.MintID() |
| if err != nil { |
| jsonErrorCode(w, http.StatusBadGateway, runCodeRecordFailed, runRecordFailedMsg) |
| @@ -423,7 +547,19 @@ 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) |
| + 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. |
| + 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) |
| + } |
| + cur, _ := provision.GetRecord(wctx, s.cfg.Releases, acct.Sqid, app, host) |
| + runReplayed(w, cur, host, 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. |
| @@ -450,6 +586,8 @@ func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) { |
| jsonErrorCode(w, http.StatusConflict, runCodeMoving, runMovingMsg(host)) |
| case errors.Is(err, provision.ErrRunImport): |
| jsonErrorCode(w, http.StatusConflict, runCodeImportUncommitted, runImportMsg(host)) |
| + case errors.Is(err, provision.ErrRunKeyReused): |
| + jsonErrorCode(w, http.StatusConflict, runCodeKeyReused, runKeyReusedMsg) |
| case errors.Is(err, os.ErrNotExist): |
| jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, notInstance) |
| default: |
| @@ -467,29 +605,96 @@ 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. A RunOpenError naming id itself means an |
| +// 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 |
| // earlier attempt of this same call committed (updateRecord retries after a |
| // lost-response conflict, and finds its own run queued): that is admitted. |
| -func (s *Server) admitRun(ctx context.Context, sqid, app, host, id, by string) (string, error) { |
| +// 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) { |
| 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)})) |
| + provision.RunAdmit(provision.RunAdmission{ID: id, By: by, Deadline: time.Now().UTC().Add(runs.PickupWindow), Key: key, Digest: digest})) |
| var open *provision.RunOpenError |
| + var replay *provision.RunKeyReplay |
| switch { |
| case err == nil: |
| return got.ActiveRun.State, nil |
| case errors.As(err, &open) && open.Run.ID == id: |
| return open.Run.State, nil |
| + case errors.As(err, &replay) && replay.RunID == id: |
| + // This request's own admission committed; delete nothing. The state |
| + // is the record's, "" if it cannot be read now. |
| + state := "" |
| + if cur, rerr := provision.GetRecord(ctx, s.cfg.Releases, sqid, app, host); rerr == nil && cur.ActiveRun != nil && cur.ActiveRun.ID == id { |
| + state = cur.ActiveRun.State |
| + } |
| + return state, nil |
| } |
| return "", err |
| } |
| // runAdmitRefused reports whether err is the record saying no (another run, |
| -// a move, an uncommitted import, no record) rather than a failure whose |
| -// outcome the record has to settle. |
| +// a move, an uncommitted import, a key reused, no record) rather than a |
| +// failure whose outcome the record has to settle. |
| func runAdmitRefused(err error) bool { |
| var open *provision.RunOpenError |
| return errors.As(err, &open) || errors.Is(err, provision.ErrRunMoving) || |
| - errors.Is(err, provision.ErrRunImport) || errors.Is(err, os.ErrNotExist) |
| + errors.Is(err, provision.ErrRunImport) || errors.Is(err, provision.ErrRunKeyReused) || |
| + errors.Is(err, os.ErrNotExist) |
| +} |
| + |
| +// handleAPIRunKeyLookup is GET …/instances/{host}/runs?key=<key> (service |
| +// runs spec § 3): the run the caller's own key admitted on this instance, |
| +// from the record, or 404. Another caller's key is never found, because the |
| +// record holds keys scoped to their caller. It is the run doors' gate, then |
| +// a read. |
| +func (s *Server) handleAPIRunKeyLookup(w http.ResponseWriter, r *http.Request) { |
| + p, ok := principalFrom(r.Context()) |
| + if !ok || !runCaller(p) { |
| + http.NotFound(w, r) |
| + return |
| + } |
| + _, acct, app, ok := s.apiAccountApp(w, r) |
| + if !ok { |
| + return |
| + } |
| + if s.cfg.HostClaim { |
| + jsonErrorCode(w, http.StatusConflict, runCodeHostClaim, claimRunUnsupportedMsg) |
| + return |
| + } |
| + if !s.cli.runs.allow(hashToken(bearer(r)), time.Now()) { |
| + jsonErrorCode(w, http.StatusTooManyRequests, runCodeRateLimited, runTooManyMsg) |
| + return |
| + } |
| + keys, scope := r.URL.Query()["key"], callerScope(p) |
| + if len(keys) != 1 || !runs.ValidKey(keys[0]) || scope == "" { |
| + jsonErrorCode(w, http.StatusBadRequest, runCodeKeyInvalid, runKeyInvalidMsg) |
| + return |
| + } |
| + if s.cfg.Releases == nil { |
| + jsonErrorCode(w, http.StatusServiceUnavailable, runCodeUnavailable, runReadFailedMsg) |
| + return |
| + } |
| + host := r.PathValue("host") |
| + if !provision.ValidHostShape(host) { |
| + jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, fmt.Sprintf("%s is not an instance of %s", host, app)) |
| + return |
| + } |
| + rec, err := provision.GetRecord(r.Context(), s.cfg.Releases, acct.Sqid, app, host) |
| + switch { |
| + case errors.Is(err, os.ErrNotExist): |
| + jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, fmt.Sprintf("%s is not an instance of %s", host, app)) |
| + return |
| + case err != nil: |
| + jsonErrorCode(w, http.StatusBadGateway, runCodeReadFailed, runReadFailedMsg) |
| + return |
| + } |
| + k, found := rec.RunKeyFor(runs.ScopedKey(scope, keys[0]), time.Now()) |
| + if !found { |
| + jsonErrorCode(w, http.StatusNotFound, runCodeNotFound, runNoKeyMsg(host)) |
| + return |
| + } |
| + writeJSON(w, http.StatusOK, runs.KeyAnswer{RunID: k.RunID}) |
| } |
| // writeRunObjects writes request.json first (the orphan sweep ages a run |
diff --git a/internal/console/apiruns_key_test.go b/internal/console/apiruns_key_test.go| new file mode 100644 |
| index 000000000..d04543788 |
| --- /dev/null |
| +++ b/internal/console/apiruns_key_test.go |
| @@ -0,0 +1,627 @@ |
| +package console |
| + |
| +// A keyed run create (service runs spec § 3): admitted once, settled by the |
| +// instance record, and a retry with the same key gets the same run. |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "mime/multipart" |
| + "net/http" |
| + "net/http/httptest" |
| + "net/url" |
| + "slices" |
| + "strings" |
| + "sync" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/provision" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/runs" |
| + "github.com/carlosframework/platform/internal/svcrole" |
| +) |
| + |
| +const testKey = "fleet:import:acct-42" |
| + |
| +// keyedFixture is runFixture with keyed creates turned on. |
| +func keyedFixture(t *testing.T) *runFx { |
| + t.Helper() |
| + fx := runFixture(t) |
| + fx.srv.cfg.RunKeysEnabled = true |
| + return fx |
| +} |
| + |
| +func sha(s string) string { |
| + h := sha256.Sum256([]byte(s)) |
| + return hex.EncodeToString(h[:]) |
| +} |
| + |
| +// fileMeta is okMeta with one declared file, x.csv holding body. |
| +func fileMeta(body string) runs.Meta { |
| + m := okMeta() |
| + m.Files = []runs.MetaFile{{Name: "x.csv", Size: int64(len(body)), SHA256: sha(body)}} |
| + return m |
| +} |
| + |
| +func xcsv(body string) []struct{ name, body string } { |
| + return []struct{ name, body string }{{"x.csv", body}} |
| +} |
| + |
| +// postKeyed is postRun with an Idempotency-Key header ("" sends none). |
| +func postKeyed(t *testing.T, u, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) (int, string) { |
| + t.Helper() |
| + var buf bytes.Buffer |
| + mw := multipart.NewWriter(&buf) |
| + mp, err := mw.CreateFormField("meta") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := json.NewEncoder(mp).Encode(meta); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, p := range parts { |
| + fp, err := mw.CreateFormFile("file", p.name) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + io.WriteString(fp, p.body) |
| + } |
| + mw.Close() |
| + req, err := http.NewRequest(http.MethodPost, u, &buf) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + req.Header.Set("Content-Type", mw.FormDataContentType()) |
| + req.Header.Set("Authorization", "Bearer "+bearer) |
| + if key != "" { |
| + req.Header.Set("Idempotency-Key", key) |
| + } |
| + res, err := http.DefaultClient.Do(req) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer res.Body.Close() |
| + out, _ := io.ReadAll(res.Body) |
| + return res.StatusCode, string(out) |
| +} |
| + |
| +// keyedCreated is the 202 body, with replayed. |
| +type keyedCreated struct { |
| + runs.Created |
| + Replayed *bool `json:"replayed"` |
| +} |
| + |
| +func createdOf(t *testing.T, status int, body string) keyedCreated { |
| + t.Helper() |
| + if status != http.StatusAccepted { |
| + t.Fatalf("create: %d %s", status, body) |
| + } |
| + var c keyedCreated |
| + if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) { |
| + t.Fatalf("created %s: %v", body, err) |
| + } |
| + return c |
| +} |
| + |
| +func (c keyedCreated) replayed() bool { return c.Replayed != nil && *c.Replayed } |
| + |
| +// createKeyed is a keyed create that must answer 202. |
| +func (fx *runFx) createKeyed(t *testing.T, bearer, key string, meta runs.Meta, parts []struct{ name, body string }) keyedCreated { |
| + t.Helper() |
| + status, body := postKeyed(t, fx.runsURL(runTestHost), bearer, key, meta, parts) |
| + return createdOf(t, status, body) |
| +} |
| + |
| +// runIDsWithObjects is every run id with objects under the shop app. |
| +func (fx *runFx) runIDsWithObjects(t *testing.T) []string { |
| + t.Helper() |
| + prefix := runs.RunsPrefix(release.OpsAccount, "shop") |
| + keys, err := fx.rel.List(context.Background(), prefix) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var ids []string |
| + for _, k := range keys { |
| + id := strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0] |
| + if !slices.Contains(ids, id) { |
| + ids = append(ids, id) |
| + } |
| + } |
| + return ids |
| +} |
| + |
| +func (fx *runFx) instRecord(t *testing.T) *provision.Record { |
| + t.Helper() |
| + rec, err := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return rec |
| +} |
| + |
| +func (fx *runFx) runStarts(t *testing.T) int { |
| + t.Helper() |
| + events, _ := fx.st.Events(OpsAccountID, "shop", 100) |
| + n := 0 |
| + for _, e := range events { |
| + if e.Action == "run-start" { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| +// finishAndAck takes id from queued to acknowledged through the doors a box |
| +// and the CLI use. |
| +func (fx *runFx) finishAndAck(t *testing.T, bearer, id string) { |
| + t.Helper() |
| + fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour))) |
| + if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "shop", id, runs.Result{Exit: 0, LastSeq: 0}); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.transition(t, provision.RunFinish(id)) |
| + if status, body := fx.followAs(t, bearer, "ack", id); status != http.StatusOK { |
| + t.Fatalf("ack: %d %s", status, body) |
| + } |
| +} |
| + |
| +func TestRunKeyedCreateThenRetryReturnsTheSameRun(t *testing.T) { |
| + fx := keyedFixture(t) |
| + first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c")) |
| + if first.replayed() || first.State != provision.RunQueued { |
| + t.Fatalf("a fresh admission: %+v", first) |
| + } |
| + again := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c")) |
| + if again.ID != first.ID || !again.replayed() || again.Host != runTestHost { |
| + t.Fatalf("the retry: %+v, want %s replayed", again, first.ID) |
| + } |
| + if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID { |
| + t.Fatalf("runs with objects %v, want only %s", ids, first.ID) |
| + } |
| + if n := fx.runStarts(t); n != 1 { |
| + t.Errorf("%d run-start rows, want 1", n) |
| + } |
| + rec := fx.instRecord(t) |
| + if len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != first.ID || rec.RunKeys[0].Key != runs.ScopedKey("user:alice@keymail.dev", testKey) { |
| + t.Fatalf("run_keys %+v", rec.RunKeys) |
| + } |
| + req, err := runs.ReadRequest(context.Background(), fx.rel, release.OpsAccount, "shop", first.ID) |
| + if err != nil || len(req.Files) != 1 || req.Files[0].SHA256 != sha("a,b,c") { |
| + t.Fatalf("request %+v, %v", req, err) |
| + } |
| +} |
| + |
| +func TestRunKeyedRetryWhileOpenReturnsTheRun(t *testing.T) { |
| + fx := keyedFixture(t) |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + first := fx.createKeyed(t, runner, testKey, serviceMeta(), nil) |
| + fx.transition(t, provision.RunPick(first.ID, "box-1", "boot-1", time.Now().Add(time.Hour))) |
| + again := fx.createKeyed(t, runner, testKey, serviceMeta(), nil) |
| + if again.ID != first.ID || !again.replayed() || again.State != provision.RunPicked { |
| + t.Fatalf("the retry while open: %+v, want %s picked and replayed", again, first.ID) |
| + } |
| + if n := fx.runStarts(t); n != 1 { |
| + t.Errorf("%d run-start rows, want 1", n) |
| + } |
| +} |
| + |
| +func TestRunKeyedRetryAfterAckStillReturnsTheRun(t *testing.T) { |
| + fx := keyedFixture(t) |
| + first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + fx.finishAndAck(t, fx.member, first.ID) |
| + again := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + if again.ID != first.ID || !again.replayed() || again.State != "" { |
| + t.Fatalf("the retry after ack: %+v, want %s replayed, with no record state", again, first.ID) |
| + } |
| + // A later unkeyed run takes the record; the key still answers. |
| + other := fx.startRun(t) |
| + again = fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + if again.ID != first.ID || !again.replayed() { |
| + t.Fatalf("the retry after %s replaced it: %+v", other, again) |
| + } |
| + if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != other { |
| + t.Fatalf("runs with objects %v, want only %s", ids, other) |
| + } |
| +} |
| + |
| +func TestRunKeyReusedWithADifferentRequest(t *testing.T) { |
| + fx := keyedFixture(t) |
| + first := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + before := fx.instRecord(t) |
| + other := okMeta() |
| + other.Argv = []string{"import", "y.csv"} |
| + status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, other, nil) |
| + if status != http.StatusConflict || codeOf(t, body) != runCodeKeyReused || errorOf(t, body) == "" { |
| + t.Fatalf("the same key, another request: %d %s, want 409 %s", status, body, runCodeKeyReused) |
| + } |
| + if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != first.ID { |
| + t.Fatalf("runs with objects %v, want only %s", ids, first.ID) |
| + } |
| + if after := fx.instRecord(t); after.RecordSeq != before.RecordSeq { |
| + t.Fatalf("the refusal wrote the record: seq %d -> %d", before.RecordSeq, after.RecordSeq) |
| + } |
| +} |
| + |
| +// barrierStore holds every admission's record write until want request.json |
| +// writes have landed, so that concurrent creates are all past the door's key |
| +// lookup and have all written their objects before any is admitted. |
| +type barrierStore struct { |
| + release.Store |
| + recordKey string |
| + want int |
| + mu sync.Mutex |
| + requests int |
| + ready chan struct{} |
| +} |
| + |
| +func (b *barrierStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) { |
| + if key == b.recordKey { |
| + select { |
| + case <-b.ready: |
| + case <-time.After(10 * time.Second): |
| + return "", errors.New("barrier: the other create never wrote its objects") |
| + } |
| + } |
| + etag, err := b.Store.Put(ctx, key, body, ifMatch) |
| + if err == nil && strings.HasSuffix(key, "/request.json") { |
| + b.mu.Lock() |
| + b.requests++ |
| + if b.requests == b.want { |
| + close(b.ready) |
| + } |
| + b.mu.Unlock() |
| + } |
| + return etag, err |
| +} |
| + |
| +func TestRunConcurrentIdenticalKeyedCreatesMakeOneRun(t *testing.T) { |
| + fx := keyedFixture(t) |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + fx.srv.cfg.Releases = &barrierStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), want: 2, ready: make(chan struct{})} |
| + meta := serviceMeta() |
| + meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5, SHA256: sha("a,b,c")}} |
| + var wg sync.WaitGroup |
| + statuses, bodies := make([]int, 2), make([]string, 2) |
| + for i := range 2 { |
| + wg.Add(1) |
| + go func() { |
| + defer wg.Done() |
| + statuses[i], bodies[i] = postKeyed(t, fx.runsURL(runTestHost), runner, testKey, meta, xcsv("a,b,c")) |
| + }() |
| + } |
| + wg.Wait() |
| + a, b := createdOf(t, statuses[0], bodies[0]), createdOf(t, statuses[1], bodies[1]) |
| + if a.ID != b.ID { |
| + t.Fatalf("two runs for one key: %s and %s", a.ID, b.ID) |
| + } |
| + if a.replayed() == b.replayed() { |
| + t.Fatalf("want one fresh admission and one replay: %s / %s", bodies[0], bodies[1]) |
| + } |
| + if ids := fx.runIDsWithObjects(t); len(ids) != 1 || ids[0] != a.ID { |
| + t.Fatalf("runs with objects %v, want only the winner %s: the loser's objects must be gone", ids, a.ID) |
| + } |
| + if n := fx.runStarts(t); n != 1 { |
| + t.Errorf("%d run-start rows, want 1", n) |
| + } |
| + if rec := fx.instRecord(t); rec.ActiveRun == nil || rec.ActiveRun.ID != a.ID || len(rec.RunKeys) != 1 { |
| + t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys) |
| + } |
| +} |
| + |
| +// lookupKey is GET …/runs?key=<key>. |
| +func (fx *runFx) lookupKey(t *testing.T, bearer, key string) (int, string) { |
| + t.Helper() |
| + status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(key), bearer, "") |
| + return status, body |
| +} |
| + |
| +func TestRunKeyOfAnotherCallerIsInvisible(t *testing.T) { |
| + fx := keyedFixture(t) |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + person := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + if status, body := fx.lookupKey(t, runner, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound { |
| + t.Fatalf("the service looking up the person's key: %d %s, want 404", status, body) |
| + } |
| + fx.finishAndAck(t, fx.member, person.ID) |
| + svc := fx.createKeyed(t, runner, testKey, serviceMeta(), nil) |
| + if svc.ID == person.ID || svc.replayed() { |
| + t.Fatalf("the service's create with the person's key: %+v, want a fresh run", svc) |
| + } |
| + for _, c := range []struct { |
| + bearer, want string |
| + }{{fx.member, person.ID}, {runner, svc.ID}} { |
| + status, body := fx.lookupKey(t, c.bearer, testKey) |
| + var got struct { |
| + RunID string `json:"run_id"` |
| + } |
| + if status != http.StatusOK || json.Unmarshal([]byte(body), &got) != nil || got.RunID != c.want { |
| + t.Errorf("lookup: %d %s, want %s", status, body, c.want) |
| + } |
| + } |
| +} |
| + |
| +func TestRunKeyedHashMismatchWritesNothing(t *testing.T) { |
| + fx := keyedFixture(t) |
| + dir := spoolDir(t) |
| + status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,d")) |
| + if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch { |
| + t.Fatalf("a file that is not its declared sha256: %d %s, want 400 %s", status, body, runCodeFileHashMismatch) |
| + } |
| + fx.noRunObjects(t, "a hash mismatch") |
| + assertNoSpool(t, dir, "a hash mismatch") |
| + if rec := fx.instRecord(t); len(rec.RunKeys) != 0 { |
| + t.Fatalf("run_keys %+v", rec.RunKeys) |
| + } |
| + |
| + // A keyed create must declare every file's sha256. |
| + undeclared := fileMeta("a,b,c") |
| + undeclared.Files[0].SHA256 = "" |
| + status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, undeclared, xcsv("a,b,c")) |
| + if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest { |
| + t.Fatalf("a keyed file with no sha256: %d %s, want 400 %s", status, body, runCodeInvalidRequest) |
| + } |
| + // A declared sha256 is checked whether or not the create is keyed. |
| + status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", fileMeta("a,b,c"), xcsv("a,b,d")) |
| + if status != http.StatusBadRequest || codeOf(t, body) != runCodeFileHashMismatch { |
| + t.Fatalf("an unkeyed mismatch: %d %s", status, body) |
| + } |
| + badShape := fileMeta("a,b,c") |
| + badShape.Files[0].SHA256 = "ABC" |
| + status, body = postKeyed(t, fx.runsURL(runTestHost), fx.member, "", badShape, xcsv("a,b,c")) |
| + if status != http.StatusBadRequest || codeOf(t, body) != runCodeInvalidRequest { |
| + t.Fatalf("a malformed sha256: %d %s", status, body) |
| + } |
| + fx.noRunObjects(t, "the hash refusals") |
| +} |
| + |
| +// failAfter is a request body that serves head and then fails every read. |
| +type failAfter struct { |
| + head io.Reader |
| + hit bool |
| +} |
| + |
| +func (f *failAfter) Read(p []byte) (int, error) { |
| + if n, err := f.head.Read(p); n > 0 || err != io.EOF { |
| + return n, err |
| + } |
| + f.hit = true |
| + return 0, errors.New("failAfter: the file part must not be read") |
| +} |
| + |
| +// serveBrokenBody sends a create whose body is the meta part and the start |
| +// of a file part, after which every read fails, straight to the handler. |
| +func (fx *runFx) serveBrokenBody(t *testing.T, bearer, key string, meta runs.Meta) *httptest.ResponseRecorder { |
| + t.Helper() |
| + var buf bytes.Buffer |
| + mw := multipart.NewWriter(&buf) |
| + mp, _ := mw.CreateFormField("meta") |
| + if err := json.NewEncoder(mp).Encode(meta); err != nil { |
| + t.Fatal(err) |
| + } |
| + fp, _ := mw.CreateFormFile("file", "x.csv") |
| + io.WriteString(fp, "a") |
| + req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: &buf}) |
| + req.Header.Set("Content-Type", mw.FormDataContentType()) |
| + req.Header.Set("Authorization", "Bearer "+bearer) |
| + if key != "" { |
| + req.Header.Set("Idempotency-Key", key) |
| + } |
| + w := httptest.NewRecorder() |
| + fx.srv.ServeHTTP(w, req) |
| + return w |
| +} |
| + |
| +func TestRunKnownKeyAnsweredWithoutReadingFiles(t *testing.T) { |
| + fx := keyedFixture(t) |
| + first := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c")) |
| + w := fx.serveBrokenBody(t, fx.member, testKey, fileMeta("a,b,c")) |
| + again := createdOf(t, w.Code, w.Body.String()) |
| + if again.ID != first.ID || !again.replayed() { |
| + t.Fatalf("the retry: %+v, want %s replayed", again, first.ID) |
| + } |
| + // The same broken body with a fresh key reaches the file part and fails. |
| + fx.finishAndAck(t, fx.member, first.ID) |
| + w = fx.serveBrokenBody(t, fx.member, "another-key", fileMeta("a,b,c")) |
| + if w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed { |
| + t.Fatalf("a fresh key with a broken file part: %d %s, want 400 malformed", w.Code, w.Body.String()) |
| + } |
| +} |
| + |
| +func TestRunRefusedKeyIsReevaluated(t *testing.T) { |
| + fx := keyedFixture(t) |
| + open := fx.startRun(t) |
| + status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, okMeta(), nil) |
| + if status != http.StatusConflict || codeOf(t, body) != runCodeRunOpen { |
| + t.Fatalf("a keyed create over an open run: %d %s, want 409 run_open", status, body) |
| + } |
| + fx.finishAndAck(t, fx.member, open) |
| + c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + if c.replayed() || c.ID == open { |
| + t.Fatalf("the retry once the other run was acked: %+v, want a fresh run", c) |
| + } |
| +} |
| + |
| +func TestRunKeyLookup(t *testing.T) { |
| + fx := keyedFixture(t) |
| + if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound { |
| + t.Fatalf("an unknown key: %d %s, want 404", status, body) |
| + } |
| + c := fx.createKeyed(t, fx.member, testKey, okMeta(), nil) |
| + status, body := fx.lookupKey(t, fx.member, testKey) |
| + if status != http.StatusOK || body != fmt.Sprintf("{\"run_id\":%q}\n", c.ID) { |
| + t.Fatalf("a known key: %d %s, want %s", status, body, c.ID) |
| + } |
| + for _, bad := range []string{"", "a b", strings.Repeat("k", 129)} { |
| + status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+url.QueryEscape(bad), fx.member, "") |
| + if status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid { |
| + t.Errorf("key %q: %d %s, want 400 %s", bad, status, body, runCodeKeyInvalid) |
| + } |
| + } |
| + status, _, body = rawReq(t, http.MethodGet, fx.ts.URL+"/api/cli/apps/"+release.OpsAccount+"/shop/instances/nobody.example.com/runs?key="+testKey, fx.member, "") |
| + if status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound { |
| + t.Fatalf("an unknown host: %d %s", status, body) |
| + } |
| + // The shared gates hold here as at every run door. |
| + stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-key-stranger") |
| + status, h, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"?key="+testKey, stranger, "") |
| + assertNotFound(t, "a stranger's lookup", canonicalNotFound(t), status, h, body) |
| + fx.srv.cfg.HostClaim = true |
| + if status, body := fx.lookupKey(t, fx.member, testKey); status != http.StatusConflict || codeOf(t, body) != runCodeHostClaim { |
| + t.Fatalf("a lookup on a host-claim console: %d %s", status, body) |
| + } |
| +} |
| + |
| +// The key lookup is a run door: the run role reaches it, every other role |
| +// gets the role gate's 403, Admin included. |
| +func TestRunKeyLookupRoleMatrix(t *testing.T) { |
| + fx := keyedFixture(t) |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + u := fx.runsURL(runTestHost) + "?key=" + testKey |
| + if status, _, body := rawReq(t, http.MethodGet, u, runner, ""); status != http.StatusNotFound || codeOf(t, body) != runCodeNotFound { |
| + t.Fatalf("a run service: %d %s, want the lookup's own 404", status, body) |
| + } |
| + for _, r := range svcrole.All() { |
| + if r == svcrole.Run { |
| + continue |
| + } |
| + app, rungs := "shop", []string(nil) |
| + if r == svcrole.Admin { |
| + app = "" |
| + } |
| + if r == svcrole.Deploy { |
| + rungs = []string{"edge"} |
| + } |
| + tok, err := fx.st.NewServicePrincipalScoped(OpsAccountID, "role-"+string(r), app, r, rungs, nil, "paul@keymail.dev", time.Now()) |
| + if err != nil { |
| + t.Fatalf("mint %s: %v", r, err) |
| + } |
| + raw, _ := json.Marshal(map[string]string{"error": fmt.Sprintf("this credential holds role %q; this door needs one of [run]", r)}) |
| + if status, _, body := rawReq(t, http.MethodGet, u, tok, ""); status != http.StatusForbidden || body != string(raw)+"\n" { |
| + t.Errorf("role %s: %d %q, want 403 %s", r, status, body, raw) |
| + } |
| + } |
| +} |
| + |
| +// TestRunUnkeyedOrderUnchanged: an unkeyed create over an open run is |
| +// refused before its body is read, as before keys. A keyed one reads its |
| +// meta first, so the same broken body is malformed. |
| +func TestRunUnkeyedOrderUnchanged(t *testing.T) { |
| + fx := keyedFixture(t) |
| + fx.startRun(t) |
| + broken := func(key string) *httptest.ResponseRecorder { |
| + req := httptest.NewRequest(http.MethodPost, fx.runsURL(runTestHost), &failAfter{head: strings.NewReader("")}) |
| + req.Header.Set("Content-Type", "multipart/form-data; boundary=x") |
| + req.Header.Set("Authorization", "Bearer "+fx.member) |
| + if key != "" { |
| + req.Header.Set("Idempotency-Key", key) |
| + } |
| + w := httptest.NewRecorder() |
| + fx.srv.ServeHTTP(w, req) |
| + return w |
| + } |
| + if w := broken(""); w.Code != http.StatusConflict || codeOf(t, w.Body.String()) != runCodeRunOpen { |
| + t.Fatalf("unkeyed over an open run: %d %s, want 409 run_open before the body", w.Code, w.Body.String()) |
| + } |
| + if w := broken(testKey); w.Code != http.StatusBadRequest || codeOf(t, w.Body.String()) != runCodeMalformed { |
| + t.Fatalf("keyed over an open run with a broken body: %d %s, want 400 malformed", w.Code, w.Body.String()) |
| + } |
| +} |
| + |
| +// The keyed form of TestAPIRunCreateAdmissionWhoseAnswerWasLost: the retry |
| +// inside updateRecord finds this request's own key, which is a fresh |
| +// admission, and the run's objects stay. |
| +func TestAPIRunCreateAdmissionWhoseAnswerWasLostKeyed(t *testing.T) { |
| + for name, failure := range map[string]error{ |
| + "retried write": fmt.Errorf("s3store: record: %w", release.ErrCASConflict), |
| + "transport error": errors.New("connection reset by peer"), |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + fx := keyedFixture(t) |
| + fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), err: failure} |
| + c := fx.createKeyed(t, fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c")) |
| + if c.replayed() || c.State != provision.RunQueued { |
| + t.Fatalf("created %+v, want a fresh admission", c) |
| + } |
| + rec := fx.instRecord(t) |
| + if rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || len(rec.RunKeys) != 1 || rec.RunKeys[0].RunID != c.ID { |
| + t.Fatalf("record run %+v keys %+v", rec.ActiveRun, rec.RunKeys) |
| + } |
| + keys, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "shop", c.ID)) |
| + if len(keys) != 2 { |
| + t.Fatalf("the admitted 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) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// admitRun itself answers the retry inside updateRecord that finds this |
| +// request's own key: admitted, with the record's state, and no error for the |
| +// handler to settle. |
| +func TestAdmitRunOwnKeyReplayIsAdmitted(t *testing.T) { |
| + fx := keyedFixture(t) |
| + fx.srv.cfg.Releases = &commitThenFailStore{Store: fx.rel, recordKey: provision.RecordKey(release.OpsAccount, "shop", runTestHost), |
| + err: fmt.Errorf("s3store: record: %w", release.ErrCASConflict)} |
| + id, err := runs.MintID() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + state, err := fx.srv.admitRun(context.Background(), release.OpsAccount, "shop", runTestHost, id, "alice@keymail.dev", "K", "D") |
| + if err != nil || state != provision.RunQueued { |
| + t.Fatalf("admitRun = %q, %v; want queued and no error", state, err) |
| + } |
| +} |
| + |
| +func TestKeyedCreateRefusedUntilKeysAreOn(t *testing.T) { |
| + fx := runFixture(t) |
| + status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, testKey, fileMeta("a,b,c"), xcsv("a,b,c")) |
| + if status != http.StatusServiceUnavailable || codeOf(t, body) != runCodeKeysNotReady || errorOf(t, body) == "" { |
| + t.Fatalf("a keyed create with keys off: %d %s, want 503 %s", status, body, runCodeKeysNotReady) |
| + } |
| + fx.noRunObjects(t, "a keyed create with keys off") |
| + if rec := fx.instRecord(t); len(rec.RunKeys) != 0 { |
| + t.Fatalf("run_keys %+v", rec.RunKeys) |
| + } |
| + // Even an invalid key is refused as not ready, never run unkeyed. |
| + if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a b", okMeta(), nil); status != http.StatusServiceUnavailable { |
| + t.Fatalf("an invalid key with keys off: %d %s", status, body) |
| + } |
| + c := fx.createKeyed(t, fx.member, "", okMeta(), nil) |
| + if c.replayed() { |
| + t.Fatalf("an unkeyed create: %+v", c) |
| + } |
| + if rec := fx.instRecord(t); len(rec.RunKeys) != 0 { |
| + t.Fatalf("an unkeyed create recorded a key: %+v", rec.RunKeys) |
| + } |
| + // With keys on, an invalid key is refused by its shape. |
| + fx.srv.cfg.RunKeysEnabled = true |
| + if status, body := postKeyed(t, fx.runsURL(runTestHost), fx.member, "a/b", okMeta(), nil); status != http.StatusBadRequest || codeOf(t, body) != runCodeKeyInvalid { |
| + t.Fatalf("an invalid key: %d %s, want 400 %s", status, body, runCodeKeyInvalid) |
| + } |
| +} |
| + |
| +// The codes this task adds are in the one list. |
| +func TestRunKeyCodesAreListed(t *testing.T) { |
| + for _, c := range []string{runCodeKeyReused, runCodeKeyInvalid, runCodeKeysNotReady, runCodeFileHashMismatch} { |
| + if !slices.Contains(runRefusalCodes, c) { |
| + t.Errorf("%s is not in runRefusalCodes", c) |
| + } |
| + } |
| + for c, want := range map[string]string{ |
| + runCodeKeyReused: "idempotency_key_reused", runCodeKeyInvalid: "idempotency_key_invalid", |
| + runCodeKeysNotReady: "keys_not_ready", runCodeFileHashMismatch: "file_hash_mismatch", |
| + } { |
| + if c != want { |
| + t.Errorf("code %s, want %s", c, want) |
| + } |
| + } |
| +} |
diff --git a/internal/console/apiruns_service_test.go b/internal/console/apiruns_service_test.go| index 24dabae57..7e7f91fe4 100644 |
| --- a/internal/console/apiruns_service_test.go |
| +++ b/internal/console/apiruns_service_test.go |
| @@ -793,3 +793,57 @@ func TestCallerScopeNamesTheKind(t *testing.T) { |
| t.Fatal("a member does not see a service's run") |
| } |
| } |
| + |
| +// A run credential minted in another account, for an app of the same name, |
| +// is a stranger at this account's run doors: the membership 404, and nothing |
| +// written. |
| +func TestRunServiceOfAnotherAccountRefused(t *testing.T) { |
| + fx := runFixture(t) |
| + const otherID = 4242 |
| + now := time.Now() |
| + if err := fx.st.CreateAccount(otherID, "cdef", "Other", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := fx.st.CreateAppIn(otherID, "shop", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + other, err := fx.st.NewServicePrincipalScoped(otherID, "ci", "shop", svcrole.Run, nil, importClassic, "paul@keymail.dev", now) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + id := fx.startAs(t, runner, serviceMeta()) |
| + want := canonicalNotFound(t) |
| + for _, d := range fx.runDoorURLs(id) { |
| + status, h, body := rawReq(t, d.method, d.u, other, d.body) |
| + assertNotFound(t, "another account's run service at "+d.method+" "+d.u, want, status, h, body) |
| + } |
| + status, h, body := postRunH(t, fx.runsURL(runTestHost), other, serviceMeta(), nil) |
| + assertNotFound(t, "another account's run service at create", want, status, h, body) |
| + if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "shop", runTestHost); rec == nil || rec.ActiveRun == nil || rec.ActiveRun.ID != id || rec.ActiveRun.State != provision.RunQueued { |
| + t.Fatalf("another account's run service changed the run: %+v", rec) |
| + } |
| +} |
| + |
| +// A granted run service on a host-claim console gets the refusal a person |
| +// gets, at every run door. |
| +func TestRunGrantedServiceOnAHostClaimConsoleRefused(t *testing.T) { |
| + fx := runFixture(t) |
| + runner := runServiceToken(t, fx.st, "ci", "shop", importClassic) |
| + id := fx.startAs(t, runner, serviceMeta()) |
| + fx.srv.cfg.HostClaim = true |
| + for _, d := range fx.runDoorURLs(id) { |
| + pStatus, _, pBody := rawReq(t, d.method, d.u, fx.member, d.body) |
| + sStatus, _, sBody := rawReq(t, d.method, d.u, runner, d.body) |
| + if pStatus != http.StatusConflict || codeOf(t, pBody) != runCodeHostClaim || errorOf(t, pBody) != claimRunUnsupportedMsg { |
| + t.Errorf("a person at %s %s = %d %s, want the host-claim refusal", d.method, d.u, pStatus, pBody) |
| + } |
| + if sStatus != pStatus || sBody != pBody { |
| + t.Errorf("the run service at %s %s = %d %q, want what a person gets, %d %q", d.method, d.u, sStatus, sBody, pStatus, pBody) |
| + } |
| + } |
| + status, body := postRun(t, fx.runsURL(runTestHost), runner, serviceMeta(), nil) |
| + if status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg { |
| + t.Errorf("a granted create on a host-claim console = %d %s", status, body) |
| + } |
| +} |
diff --git a/internal/console/authcli.go b/internal/console/authcli.go| index 63ba5c45f..87ecc1b47 100644 |
| --- a/internal/console/authcli.go |
| +++ b/internal/console/authcli.go |
| @@ -738,6 +738,7 @@ func (s *Server) registerCLIRoutes() { |
| // join: a credential that may provision instances cannot run commands on |
| // them unless it was minted for that. |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs", svcrole.Exactly(svcrole.Run), s.handleAPIRunCreate) |
| + s.cliRoute("GET /api/cli/apps/{account}/{app}/instances/{host}/runs", svcrole.Exactly(svcrole.Run), s.handleAPIRunKeyLookup) |
| s.cliRoute("GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}", svcrole.Exactly(svcrole.Run), s.handleAPIRunGet) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel", svcrole.Exactly(svcrole.Run), s.handleAPIRunCancel) |
| s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/expire", svcrole.Exactly(svcrole.Run), s.handleAPIRunExpire) |
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go| index 20b1b8dab..92749cd45 100644 |
| --- a/internal/console/claim_invariant_fixtures_test.go |
| +++ b/internal/console/claim_invariant_fixtures_test.go |
| @@ -1491,6 +1491,12 @@ var claimFixtures = map[string]claimFixture{ |
| status: http.StatusConflict, |
| msg: claimRunUnsupportedMsg, |
| }}, |
| + "route GET /api/cli/apps/{account}/{app}/instances/{host}/runs": {refused: &claimRefusalFixture{ |
| + apps: claimRefs("bab/titogo"), |
| + do: claimRefusedAPI(http.MethodGet, "/api/cli/apps/bab/titogo/instances/acme.example.com/runs?key=k", ""), |
| + status: http.StatusConflict, |
| + msg: claimRunUnsupportedMsg, |
| + }}, |
| "route GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}": {refused: &claimRefusalFixture{ |
| apps: claimRefs("bab/titogo"), |
| do: claimRefusedAPI(http.MethodGet, "/api/cli/apps/bab/titogo/instances/acme.example.com/runs/run_aaaaaaaaaaaaaaaaaaaaaaaaaa", ""), |
diff --git a/internal/console/server.go b/internal/console/server.go| index e7c940087..f36a2183e 100644 |
| --- a/internal/console/server.go |
| +++ b/internal/console/server.go |
| @@ -484,6 +484,13 @@ type Config struct { |
| HostClaim bool |
| WakeRetryReset *WakeRetryResetDoor |
| + // RunKeysEnabled admits keyed run creates (CARLOS_RUN_KEYS=on; service |
| + // runs spec § 3). Off, a create that carries an Idempotency-Key is |
| + // refused 503 keys_not_ready before anything is written, because a box |
| + // binary older than the record's run_keys would drop the key on its |
| + // next record write. It is turned on once every box runs a build that |
| + // carries run_keys through. |
| + RunKeysEnabled bool |
| // SeedScratchDir is <console data>/scratch/seeds: --from restores and |
| // import spools. Start-up removes the pool's own leftovers from it. |
| // Empty: the seed doors answer 507. |
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go| index 182e00302..497b2060e 100644 |
| --- a/internal/console/svcroutes_test.go |
| +++ b/internal/console/svcroutes_test.go |
| @@ -221,7 +221,7 @@ var closedDoors = []string{ |
| // same door's gate for an instance in another region and relays its |
| // rows: the same cut. |
| "POST /api/cli/apps/{account}/{app}/db/{host}/reads", |
| - // The five `carlos run` doors are not here: since the service runs spec |
| + // The six `carlos run` doors are not here: since the service runs spec |
| // (2026-10-24 §1, amending run spec §2 ruling 1) they take the run role |
| // and no other, Admin included, through svcrole.Exactly. runDoors below |
| // pins that. |
| @@ -279,12 +279,14 @@ var closedDoors = []string{ |
| "POST /api/cli/apps/{account}/{app}/releases/{version}/authorize", |
| } |
| -// runDoors are the five `carlos run` doors (service runs spec §1). They are |
| +// runDoors are the six `carlos run` doors (service runs spec §1, and § 3's |
| +// key lookup). They are |
| // the one family registered with svcrole.Exactly: the run role reaches them |
| // and NO other role does, Admin included, so a credential that may provision |
| // instances cannot run commands on them unless it was minted for that. |
| var runDoors = []string{ |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/runs", |
| + "GET /api/cli/apps/{account}/{app}/instances/{host}/runs", |
| "GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}", |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel", |
| "POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/expire", |
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt| index 92a03ad37..8e4029792 100644 |
| --- a/internal/console/testdata/claim-routes.txt |
| +++ b/internal/console/testdata/claim-routes.txt |
| @@ -59,6 +59,7 @@ route GET /api/cli/apps/{account}/{app}/instances |
| route GET /api/cli/apps/{account}/{app}/instances/{host} |
| route GET /api/cli/apps/{account}/{app}/instances/{host}/move |
| route GET /api/cli/apps/{account}/{app}/instances/{host}/move/offline |
| +route GET /api/cli/apps/{account}/{app}/instances/{host}/runs |
| route GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id} |
| route GET /api/cli/apps/{account}/{app}/ledgers |
| route GET /api/cli/apps/{account}/{app}/ledgers/{ledger}/head |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index a5657b0cb..64b8ec448 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -75,6 +75,15 @@ type Record struct { |
| // (run.go) write it. Additive and omitempty. |
| ActiveRun *RunState `json:"run,omitempty"` |
| + // RunKeys is the instance's recent keyed run admissions (service runs |
| + // spec § 3): each Idempotency-Key, scoped to its caller and hashed, |
| + // with the run it admitted. RunAdmit writes it in the same |
| + // compare-and-swap as the admission, so admission and dedupe are one |
| + // write; every other writer carries it through unchanged. Additive and |
| + // omitempty. A writer built before it drops it, which is why keyed |
| + // creates wait for CARLOS_RUN_KEYS=on. |
| + RunKeys []RunKey `json:"run_keys,omitempty"` |
| + |
| // Seed is where this instance's database started (seeded-instances spec |
| // §3.2): a sibling's replica (--from) or an uploaded snapshot (import). |
| // Additive and omitempty. Like Maintenance, it is preserved by ordinary |
diff --git a/internal/provision/run.go b/internal/provision/run.go| index b40a226e9..4cf0a4b8f 100644 |
| --- a/internal/provision/run.go |
| +++ b/internal/provision/run.go |
| @@ -4,9 +4,11 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "sort" |
| "time" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/runs" |
| ) |
| // RunState is a `carlos run` on the instance record (run spec §5.1). Box |
| @@ -83,17 +85,89 @@ func SetRecordRun(ctx context.Context, s release.Store, acct, app, host string, |
| return updateRecord(ctx, s, acct, app, host, mut) |
| } |
| -// RunAdmission is what the console decided for a new run. |
| +// 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. |
| type RunAdmission struct { |
| - ID, By string |
| - Deadline time.Time // the pickup deadline |
| + ID, By string |
| + Deadline time.Time // the pickup deadline |
| + Key, Digest string |
| +} |
| + |
| +// RunKey is one keyed admission on the record: the scoped key, the run it |
| +// admitted, the request's digest and when. |
| +type RunKey struct { |
| + Key string `json:"key"` |
| + RunID string `json:"run_id"` |
| + Digest string `json:"digest"` |
| + At time.Time `json:"at"` |
| +} |
| + |
| +// RunKeyReplay is RunAdmit finding its key already admitted with the same |
| +// digest: RunID is the run that admission made, and this one admitted |
| +// nothing. It is the caller's own earlier write when RunID is the id it |
| +// minted (updateRecord's retry after a lost answer), and a twin's otherwise. |
| +type RunKeyReplay struct{ RunID string } |
| + |
| +func (e *RunKeyReplay) Error() string { |
| + return "provision: the idempotency key already admitted " + e.RunID |
| +} |
| + |
| +// ErrRunKeyReused: the key was admitted for a different request. |
| +var ErrRunKeyReused = errors.New("provision: the idempotency key was used for a different request") |
| + |
| +// RunKeyFor is the record's entry for the scoped key, when one younger than |
| +// runs.KeyTTL is there. |
| +func (r Record) RunKeyFor(key string, now time.Time) (RunKey, bool) { |
| + for _, k := range r.RunKeys { |
| + if k.Key == key && now.Sub(k.At) < runs.KeyTTL { |
| + return k, true |
| + } |
| + } |
| + 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. |
| +func prunedRunKeys(keys []RunKey, now time.Time) []RunKey { |
| + var out []RunKey |
| + for _, k := range keys { |
| + if now.Sub(k.At) < runs.KeyTTL { |
| + out = append(out, k) |
| + } |
| + } |
| + 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:] |
| + } |
| + return out |
| } |
| // RunAdmit admits a run: refused while a move is in flight, while a run is |
| // open, or while the record is an uncommitted import. A terminal run is |
| // replaced. |
| +// |
| +// A keyed admission checks its key first, before every refusal, so a retry |
| +// 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. |
| 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 == "" { |
| + return fmt.Errorf("%w: malformed admission", ErrRunConflict) |
| + } |
| + if k, ok := (Record{RunKeys: keys}).RunKeyFor(p.Key, now); ok { |
| + if k.Digest != p.Digest { |
| + return ErrRunKeyReused |
| + } |
| + return &RunKeyReplay{RunID: k.RunID} |
| + } |
| + } |
| switch { |
| case p.ID == "" || p.Deadline.IsZero(): |
| return fmt.Errorf("%w: malformed admission", ErrRunConflict) |
| @@ -110,6 +184,13 @@ func RunAdmit(p RunAdmission) func(*Record) error { |
| return ErrRunImport |
| } |
| 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) |
| + } |
| + if len(keys) == 0 { |
| + keys = nil // an empty list stays absent from the record |
| + } |
| + rec.RunKeys = keys |
| return nil |
| } |
| } |
diff --git a/internal/provision/run_keys_test.go b/internal/provision/run_keys_test.go| new file mode 100644 |
| index 000000000..3f879de11 |
| --- /dev/null |
| +++ b/internal/provision/run_keys_test.go |
| @@ -0,0 +1,359 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "fmt" |
| + "go/ast" |
| + "go/parser" |
| + "go/token" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/runs" |
| + "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} |
| +} |
| + |
| +func wantReplay(t *testing.T, what string, err error, id string) { |
| + t.Helper() |
| + var replay *RunKeyReplay |
| + if !errors.As(err, &replay) || replay.RunID != id { |
| + t.Fatalf("%s: %v, want a RunKeyReplay of %s", what, err, id) |
| + } |
| +} |
| + |
| +func TestRunAdmitRecordsAndReplaysAKey(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + got, err := setRun(t, s, rec, RunAdmit(keyed("run_a", "K", "D"))) |
| + if err != nil || got.ActiveRun.ID != "run_a" { |
| + t.Fatalf("keyed admit: %+v, %v", got, err) |
| + } |
| + if len(got.RunKeys) != 1 || got.RunKeys[0].Key != "K" || got.RunKeys[0].RunID != "run_a" || got.RunKeys[0].Digest != "D" || got.RunKeys[0].At.IsZero() { |
| + t.Fatalf("the admission must record its key: %+v", got.RunKeys) |
| + } |
| + |
| + // While run_a is still open, the replay is answered before the open-run |
| + // refusal. |
| + _, err = setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D"))) |
| + wantReplay(t, "the same key and digest while open", err, "run_a") |
| + if _, err := setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D2"))); !errors.Is(err, ErrRunKeyReused) { |
| + t.Fatalf("the same key with another digest: %v, want ErrRunKeyReused", err) |
| + } |
| + // Another key still meets the open run. |
| + if _, err := setRun(t, s, rec, RunAdmit(keyed("run_b", "K2", "D"))); !errors.Is(err, ErrRunOpen) { |
| + t.Fatalf("another key while open: %v, want ErrRunOpen", err) |
| + } |
| + |
| + // Picked, finished and cleared (the ack): the key still replays. |
| + for _, step := range []func(*Record) error{ |
| + RunPick("run_a", "box-1", "boot-1", time.Now().Add(time.Hour)), |
| + RunFinish("run_a"), |
| + RunClear("run_a"), |
| + } { |
| + if _, err := setRun(t, s, rec, step); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, err = setRun(t, s, rec, RunAdmit(keyed("run_b", "K", "D"))) |
| + wantReplay(t, "after a run transition", err, "run_a") |
| + } |
| + |
| + // A later, unkeyed run replaces it in the record; the key still replays. |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_c", Deadline: time.Now().Add(time.Minute)})); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := setRun(t, s, rec, RunExpire("run_c")); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, err = setRun(t, s, rec, RunAdmit(keyed("run_d", "K", "D"))) |
| + wantReplay(t, "after a later run replaced it", err, "run_a") |
| + |
| + // A refused replay wrote nothing. |
| + cur, err := GetRecord(context.Background(), s, rec.Account, rec.App, rec.Host) |
| + if err != nil || cur.ActiveRun.ID != "run_c" || len(cur.RunKeys) != 1 { |
| + t.Fatalf("record after the replays: run %+v keys %+v, %v", cur.ActiveRun, cur.RunKeys, err) |
| + } |
| +} |
| + |
| +// putRunKeys installs keys on the record directly, as an earlier console's |
| +// admissions would have left them. |
| +func putRunKeys(t *testing.T, s release.Store, rec Record, keys []RunKey) { |
| + t.Helper() |
| + if _, err := updateRecord(context.Background(), s, rec.Account, rec.App, rec.Host, func(r *Record) error { |
| + r.RunKeys = keys |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +func TestRunKeysPrunedByAgeAndCount(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + 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)}) |
| + } |
| + putRunKeys(t, s, rec, keys) |
| + |
| + // An expired key is not a replay: it admits a fresh run. |
| + got, err := setRun(t, s, rec, RunAdmit(keyed("run_new", "old", "D"))) |
| + 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) |
| + } |
| + kept := map[string]bool{} |
| + for _, k := range got.RunKeys { |
| + kept[k.Key+"/"+k.RunID] = true |
| + } |
| + if kept["old/run_old"] { |
| + t.Error("the entry older than KeyTTL survived") |
| + } |
| + 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++ { |
| + 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) |
| + } |
| + } |
| + |
| + // Under the count, age alone prunes. |
| + 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)}, |
| + }) |
| + 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" { |
| + t.Fatalf("an expired entry under the count: %+v, %v; want a then b", got.RunKeys, err) |
| + } |
| + |
| + // The record's lookup ignores an expired entry too. |
| + r := Record{RunKeys: []RunKey{{Key: "x", RunID: "run_x", Digest: "D", At: now.Add(-runs.KeyTTL - time.Second)}}} |
| + if _, ok := r.RunKeyFor("x", now); ok { |
| + t.Error("RunKeyFor answered an expired key") |
| + } |
| + r.RunKeys[0].At = now.Add(-time.Minute) |
| + if k, ok := r.RunKeyFor("x", now); !ok || k.RunID != "run_x" { |
| + t.Errorf("RunKeyFor = %+v, %v", k, ok) |
| + } |
| +} |
| + |
| +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) |
| + } |
| +} |
| + |
| +// keysJSON is the record's run_keys as stored, byte for byte. |
| +func keysJSON(t *testing.T, s release.Store, rec Record) string { |
| + t.Helper() |
| + raw, _, err := s.Get(context.Background(), RecordKey(rec.Account, rec.App, rec.Host)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + var m map[string]json.RawMessage |
| + if err := json.Unmarshal(raw, &m); err != nil { |
| + t.Fatal(err) |
| + } |
| + return string(m["run_keys"]) |
| +} |
| + |
| +// TestRunKeysSurviveEveryRecordMutation: every run transition, the |
| +// ordinary writers, the move transitions and the import transitions carry |
| +// run_keys through unchanged. Only a keyed admission adds to them. |
| +func TestRunKeysSurviveEveryRecordMutation(t *testing.T) { |
| + ctx := context.Background() |
| + at := time.Now().UTC().Truncate(time.Second) |
| + seeded := []RunKey{{Key: strings.Repeat("a", 64), RunID: "run_k", Digest: strings.Repeat("d", 64), At: at}} |
| + fresh := func(t *testing.T, r Record) (release.Store, Record, string) { |
| + t.Helper() |
| + s := release.NewDirStore(t.TempDir()) |
| + if err := CreateRecord(ctx, s, r); err != nil { |
| + t.Fatal(err) |
| + } |
| + putRunKeys(t, s, r, seeded) |
| + want := keysJSON(t, s, r) |
| + if !strings.Contains(want, "run_k") { |
| + t.Fatalf("run_keys = %s", want) |
| + } |
| + return s, r, want |
| + } |
| + base := func() Record { |
| + return Record{Schema: 1, Host: "a.go.example.com", Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + } |
| + type step struct { |
| + name string |
| + apply func(release.Store, Record) error |
| + } |
| + run := func(mut func(*Record) error) func(release.Store, Record) error { |
| + return func(s release.Store, r Record) error { |
| + _, err := SetRecordRun(ctx, s, r.Account, r.App, r.Host, mut) |
| + return err |
| + } |
| + } |
| + move := func(mut func(*Record) error) func(release.Store, Record) error { |
| + return func(s release.Store, r Record) error { |
| + _, err := SetRecordMove(ctx, s, r.Account, r.App, r.Host, mut) |
| + return err |
| + } |
| + } |
| + seed := func(mut func(*Record) error) func(release.Store, Record) error { |
| + return func(s release.Store, r Record) error { |
| + _, err := SetRecordSeed(ctx, s, r.Account, r.App, r.Host, mut) |
| + return err |
| + } |
| + } |
| + deadline := at.Add(time.Hour) |
| + moveID := "mv_" + strings.Repeat("a", 26) |
| + seedID, err := seeds.NewID() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + importRecord := base() |
| + importRecord.Seed = &SeedSource{ID: seedID, Kind: seeds.KindImport, State: seeds.StateAwaitingUpload} |
| + abortRecord := base() |
| + abortRecord.Seed = &SeedSource{ID: seedID, Kind: seeds.KindImport, State: seeds.StateAwaitingUpload} |
| + |
| + for _, path := range []struct { |
| + name string |
| + rec Record |
| + steps []step |
| + }{ |
| + {"the six run transitions", base(), []step{ |
| + {"RunAdmit, unkeyed", run(RunAdmit(RunAdmission{ID: "run_a", Deadline: deadline}))}, |
| + {"RunPick", run(RunPick("run_a", "box-1", "boot-1", deadline))}, |
| + {"RunCancelTransition, picked", run(RunCancelTransition("run_a"))}, |
| + {"RunFinish", run(RunFinish("run_a"))}, |
| + {"RunClear", run(RunClear("run_a"))}, |
| + {"RunAdmit, unkeyed, again", run(RunAdmit(RunAdmission{ID: "run_b", Deadline: deadline}))}, |
| + {"RunExpire", run(RunExpire("run_b"))}, |
| + {"RunAdmit replacing a terminal run", run(RunAdmit(RunAdmission{ID: "run_c", Deadline: deadline}))}, |
| + {"RunCancelTransition, queued", run(RunCancelTransition("run_c"))}, |
| + {"RunClear, cancelled", run(RunClear("run_c"))}, |
| + }}, |
| + {"the ordinary writers", base(), []step{ |
| + {"SetRecordRegion", func(s release.Store, r Record) error { return SetRecordRegion(ctx, s, r.Account, r.App, r.Host, "uk") }}, |
| + {"UpdateRecord", func(s release.Store, r Record) error { |
| + cur, etag, err := GetRecordETag(ctx, s, r.Account, r.App, r.Host) |
| + if err != nil { |
| + return err |
| + } |
| + cur.Channel = "beta" |
| + return UpdateRecord(ctx, s, *cur, etag) |
| + }}, |
| + }}, |
| + {"a legacy move to its end", base(), []step{ |
| + {"MoveRequest", move(MoveRequest("uk", at))}, |
| + {"MovePromoteReady", move(MovePromoteReady(at))}, |
| + {"MoveMarkCutting", move(MoveMarkCutting())}, |
| + {"MoveComplete", move(MoveComplete())}, |
| + }}, |
| + {"a legacy move aborted", base(), []step{ |
| + {"MoveRequest", move(MoveRequest("uk", at))}, |
| + {"MoveAbort", move(MoveAbort())}, |
| + {"MoveAbortComplete", move(MoveAbortComplete())}, |
| + }}, |
| + {"a phase-1 move start", base(), []step{ |
| + {"MoveStartTransition", move(MoveStartTransition(MoveStart{ID: moveID, Incarnation: strings.Repeat("a", 64), From: "ie", To: "uk", Cut: "auto", Now: at}))}, |
| + }}, |
| + {"an offline move aborted", base(), []step{ |
| + {"OfflineStart", move(OfflineStart(moveID, "ie", "uk", "paul", at))}, |
| + {"OfflineAbort", move(OfflineAbort(moveID))}, |
| + {"OfflineFinish", move(OfflineFinish(moveID, at))}, |
| + }}, |
| + {"an import to its commit", importRecord, []step{ |
| + {"SeedUploaded", seed(SeedUploaded(seedID))}, |
| + {"SeedStopped", seed(SeedStopped(seedID))}, |
| + {"SeedSwapped", seed(SeedSwapped(seedID))}, |
| + {"SeedPlace", seed(SeedPlace(seedID))}, |
| + {"SeedCommit", seed(SeedCommit(seedID))}, |
| + }}, |
| + {"an import aborted", abortRecord, []step{ |
| + {"SeedAbort", seed(SeedAbort(seedID))}, |
| + }}, |
| + } { |
| + t.Run(path.name, func(t *testing.T) { |
| + s, r, want := fresh(t, path.rec) |
| + for _, st := range path.steps { |
| + if err := st.apply(s, r); err != nil { |
| + t.Fatalf("%s: %v", st.name, err) |
| + } |
| + if got := keysJSON(t, s, r); got != want { |
| + t.Fatalf("%s: run_keys %s, want %s", st.name, got, want) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// TestOnlyRunAdmitWritesRunKeys reads the package's source: RunAdmit is the |
| +// one function that assigns RunKeys, and no function overwrites a whole |
| +// record through a pointer, which would drop run_keys with everything else |
| +// it did not copy. Every other writer mutates the record it read in place, |
| +// which is what TestRunKeysSurviveEveryRecordMutation relies on. |
| +func TestOnlyRunAdmitWritesRunKeys(t *testing.T) { |
| + files, err := filepath.Glob("*.go") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + fset := token.NewFileSet() |
| + for _, file := range files { |
| + if strings.HasSuffix(file, "_test.go") { |
| + continue |
| + } |
| + src, err := os.ReadFile(file) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + f, err := parser.ParseFile(fset, file, src, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, decl := range f.Decls { |
| + fn, ok := decl.(*ast.FuncDecl) |
| + if !ok || fn.Body == nil { |
| + continue |
| + } |
| + ast.Inspect(fn.Body, func(n ast.Node) bool { |
| + as, ok := n.(*ast.AssignStmt) |
| + if !ok { |
| + return true |
| + } |
| + for _, lhs := range as.Lhs { |
| + switch l := lhs.(type) { |
| + case *ast.SelectorExpr: |
| + if l.Sel.Name == "RunKeys" && !(file == "run.go" && fn.Name.Name == "RunAdmit") { |
| + t.Errorf("%s: %s assigns RunKeys; only RunAdmit may", fset.Position(l.Pos()), fn.Name.Name) |
| + } |
| + case *ast.StarExpr: |
| + if id, ok := l.X.(*ast.Ident); ok && (id.Name == "rec" || id.Name == "r") { |
| + t.Errorf("%s: %s overwrites a whole record through *%s", fset.Position(l.Pos()), fn.Name.Name, id.Name) |
| + } |
| + } |
| + } |
| + return true |
| + }) |
| + } |
| + } |
| +} |
diff --git a/internal/runs/keys.go b/internal/runs/keys.go| new file mode 100644 |
| index 000000000..0f24183cc |
| --- /dev/null |
| +++ b/internal/runs/keys.go |
| @@ -0,0 +1,94 @@ |
| +package runs |
| + |
| +import ( |
| + "crypto/sha256" |
| + "encoding/binary" |
| + "encoding/hex" |
| + "hash" |
| + "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. |
| +const ( |
| + KeyTTL = 24 * time.Hour |
| + MaxRunKeys = 128 |
| + // MaxKeyBytes is the longest Idempotency-Key a caller may send. |
| + MaxKeyBytes = 128 |
| +) |
| + |
| +// ValidKey reports whether k is an Idempotency-Key: 1 to 128 bytes of |
| +// [A-Za-z0-9._:-]. |
| +func ValidKey(k string) bool { |
| + if len(k) == 0 || len(k) > MaxKeyBytes { |
| + return false |
| + } |
| + for i := 0; i < len(k); i++ { |
| + switch c := k[i]; { |
| + case 'a' <= c && c <= 'z', 'A' <= c && c <= 'Z', '0' <= c && c <= '9': |
| + case c == '.', c == '_', c == ':', c == '-': |
| + default: |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// ScopedKey is the key as the record stores it: hex(sha256(scope, NUL, |
| +// key)). scope names the caller ("svc:" and a credential id, or "user:" |
| +// and an address), so one caller's key can never find, probe or collide |
| +// with another's, and the record never holds the caller's own words. |
| +func ScopedKey(scope, key string) string { |
| + h := sha256.Sum256([]byte(scope + "\x00" + key)) |
| + return hex.EncodeToString(h[:]) |
| +} |
| + |
| +// ValidFileHash reports whether h is a sha256 as a create declares it: 64 |
| +// lowercase hex digits. |
| +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') { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// 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. |
| +func MetaDigest(m Meta) string { |
| + h := sha256.New() |
| + str(h, "carlos-run-meta-v1") |
| + num(h, uint64(len(m.Argv))) |
| + for _, a := range m.Argv { |
| + str(h, a) |
| + } |
| + num(h, uint64(m.TimeoutS)) |
| + str(h, m.OnBehalfOf) |
| + num(h, uint64(len(m.Files))) |
| + for _, f := range m.Files { |
| + str(h, f.Name) |
| + num(h, uint64(f.Size)) |
| + str(h, f.SHA256) |
| + } |
| + return hex.EncodeToString(h.Sum(nil)) |
| +} |
| + |
| +func num(h hash.Hash, n uint64) { |
| + var b [8]byte |
| + binary.BigEndian.PutUint64(b[:], n) |
| + h.Write(b[:]) |
| +} |
| + |
| +func str(h hash.Hash, s string) { |
| + num(h, uint64(len(s))) |
| + h.Write([]byte(s)) |
| +} |
diff --git a/internal/runs/keys_test.go b/internal/runs/keys_test.go| new file mode 100644 |
| index 000000000..7e8d44ef1 |
| --- /dev/null |
| +++ b/internal/runs/keys_test.go |
| @@ -0,0 +1,91 @@ |
| +package runs |
| + |
| +import ( |
| + "strings" |
| + "testing" |
| +) |
| + |
| +func TestValidKey(t *testing.T) { |
| + for _, k := range []string{ |
| + "a", "Z", "0", ".", "_", ":", "-", |
| + "fleet:import:acct-42:2026-10-24T12.00", |
| + strings.Repeat("k", 128), |
| + "AZaz09._:-", |
| + } { |
| + if !ValidKey(k) { |
| + t.Errorf("ValidKey(%q) = false, want true", k) |
| + } |
| + } |
| + for _, k := range []string{ |
| + "", |
| + strings.Repeat("k", 129), |
| + "a/b", |
| + "a b", |
| + "a\x00b", |
| + "a\nb", |
| + "café", |
| + "a,b", |
| + "a+b", |
| + } { |
| + if ValidKey(k) { |
| + t.Errorf("ValidKey(%q) = true, want false", k) |
| + } |
| + } |
| +} |
| + |
| +func TestScopedKeySeparatesCallers(t *testing.T) { |
| + svc := ScopedKey("svc:cred_1", "K") |
| + if len(svc) != 64 || strings.Trim(svc, "0123456789abcdef") != "" { |
| + t.Fatalf("ScopedKey = %q, want 64 lowercase hex digits", svc) |
| + } |
| + if ScopedKey("svc:cred_1", "K") != svc { |
| + t.Fatal("ScopedKey is not deterministic") |
| + } |
| + for _, other := range []string{ |
| + ScopedKey("svc:cred_2", "K"), |
| + ScopedKey("user:paul@keymail.dev", "K"), |
| + ScopedKey("svc:cred_1", "K2"), |
| + // The separator keeps the scope and the key apart. |
| + ScopedKey("svc:cred_", "1K"), |
| + ScopedKey("svc:cred_1K", ""), |
| + } { |
| + if other == svc { |
| + t.Errorf("two different callers or keys share the scoped key %s", svc) |
| + } |
| + } |
| +} |
| + |
| +func TestMetaDigestCoversEveryField(t *testing.T) { |
| + hash := strings.Repeat("a", 64) |
| + base := func() Meta { |
| + return Meta{ |
| + Argv: []string{"import", "x.csv"}, |
| + TimeoutS: 60, |
| + OnBehalfOf: "paul@example.com", |
| + Files: []MetaFile{{Name: "x.csv", Size: 5, SHA256: hash}}, |
| + } |
| + } |
| + want := MetaDigest(base()) |
| + if MetaDigest(base()) != want { |
| + t.Fatal("MetaDigest is not deterministic") |
| + } |
| + for name, change := range map[string]func(*Meta){ |
| + "argv": func(m *Meta) { m.Argv[1] = "y.csv" }, |
| + "argv split": func(m *Meta) { m.Argv = []string{"importx.csv"} }, |
| + "argv joined": func(m *Meta) { m.Argv = []string{"import", "x", ".csv"} }, |
| + "timeout": func(m *Meta) { m.TimeoutS = 61 }, |
| + "on_behalf_of": func(m *Meta) { m.OnBehalfOf = "bob@example.com" }, |
| + "no requester": func(m *Meta) { m.OnBehalfOf = "" }, |
| + "file name": func(m *Meta) { m.Files[0].Name = "y.csv" }, |
| + "file size": func(m *Meta) { m.Files[0].Size = 6 }, |
| + "file sha256": func(m *Meta) { m.Files[0].SHA256 = strings.Repeat("b", 64) }, |
| + "no files": func(m *Meta) { m.Files = nil }, |
| + "a second file": func(m *Meta) { m.Files = append(m.Files, MetaFile{Name: "y.csv", Size: 1, SHA256: hash}) }, |
| + } { |
| + m := base() |
| + change(&m) |
| + if MetaDigest(m) == want { |
| + t.Errorf("changing the %s left the digest unchanged", name) |
| + } |
| + } |
| +} |
diff --git a/internal/runs/runs.go b/internal/runs/runs.go| index fd402da6d..9c07386fc 100644 |
| --- a/internal/runs/runs.go |
| +++ b/internal/runs/runs.go |
| @@ -156,6 +156,10 @@ type Meta struct { |
| type MetaFile struct { |
| Name string `json:"name"` |
| Size int64 `json:"size"` |
| + // SHA256 is the file's hex sha256 as the caller declares it. A keyed |
| + // create must declare it, and the console checks the uploaded bytes |
| + // against it whenever it is declared (service runs spec § 3). |
| + SHA256 string `json:"sha256,omitempty"` |
| } |
| // File is one uploaded file as request.json records it. |
| @@ -246,11 +250,20 @@ type Ack struct { |
| LastSeq int64 `json:"last_seq"` |
| } |
| -// Created is the 202 to a create. |
| +// Created is the 202 to a create. State is the record's state for the run, |
| +// "" when the record no longer names it. Replayed says the run was admitted |
| +// earlier under the same Idempotency-Key, and this create started nothing. |
| type Created struct { |
| - ID string `json:"id"` |
| - Host string `json:"host"` |
| - State string `json:"state"` |
| + ID string `json:"id"` |
| + Host string `json:"host"` |
| + State string `json:"state"` |
| + Replayed bool `json:"replayed,omitempty"` |
| +} |
| + |
| +// KeyAnswer is GET …/runs?key=<key>: the run admitted under the caller's |
| +// key. |
| +type KeyAnswer struct { |
| + RunID string `json:"run_id"` |
| } |
| // StateAnswer is the answer to cancel, expire and ack. |
diff --git a/internal/runs/runs_test.go b/internal/runs/runs_test.go| index 4f76a1a88..feb207a3f 100644 |
| --- a/internal/runs/runs_test.go |
| +++ b/internal/runs/runs_test.go |
| @@ -71,22 +71,22 @@ func TestCheckArgvAndTimeout(t *testing.T) { |
| } |
| func TestCheckFiles(t *testing.T) { |
| - if err := CheckFiles([]MetaFile{{"a.csv", 1}, {"b.json", 2}}); err != nil { |
| + if err := CheckFiles([]MetaFile{{Name: "a.csv", Size: 1}, {Name: "b.json", Size: 2}}); err != nil { |
| t.Fatal(err) |
| } |
| - if err := CheckFiles([]MetaFile{{"a", MaxFileBytes}, {"b", 1}}); !errors.Is(err, ErrFilesTooLarge) { |
| + if err := CheckFiles([]MetaFile{{Name: "a", Size: MaxFileBytes}, {Name: "b", Size: 1}}); !errors.Is(err, ErrFilesTooLarge) { |
| t.Errorf("over the cap: %v", err) |
| } |
| for _, name := range []string{"", ".", "..", "a/b", "../x", "a\x00b", "a\nb", strings.Repeat("n", 256)} { |
| var fe *FileNameError |
| - if err := CheckFiles([]MetaFile{{name, 1}}); !errors.Is(err, ErrFileName) || !errors.As(err, &fe) || fe.Name != name { |
| + if err := CheckFiles([]MetaFile{{Name: name, Size: 1}}); !errors.Is(err, ErrFileName) || !errors.As(err, &fe) || fe.Name != name { |
| t.Errorf("name %q: %v", name, err) |
| } |
| } |
| - if err := CheckFiles([]MetaFile{{"a", 1}, {"a", 1}}); !errors.Is(err, ErrFileTwice) { |
| + if err := CheckFiles([]MetaFile{{Name: "a", Size: 1}, {Name: "a", Size: 1}}); !errors.Is(err, ErrFileTwice) { |
| t.Errorf("twice: %v", err) |
| } |
| - if err := CheckFiles([]MetaFile{{"a", -1}}); !errors.Is(err, ErrFilesTooLarge) { |
| + if err := CheckFiles([]MetaFile{{Name: "a", Size: -1}}); !errors.Is(err, ErrFilesTooLarge) { |
| t.Errorf("a negative size must be refused: %v", err) |
| } |
| } |