amadan

carlos / platform Public

console: the carlos run create door, refusals in spec order (run spec §5.1, §6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude f04a0581662d21f326052213d489b715b9edb4fe parent 139429c
9 files changed, +706 −3
  • approved_run_copy_test.go +1 −1
  • internal/console/apiruns.go +349 −0
  • internal/console/apiruns_test.go +329 −0
  • internal/console/authcli.go +9 −0
  • internal/console/claim_invariant_fixtures_test.go +8 −0
  • internal/console/claim_invariant_test.go +2 −0
  • internal/console/svcroutes_test.go +5 −2
  • internal/console/testdata/claim-authority-sites.txt +2 −0
  • internal/console/testdata/claim-routes.txt +1 −0
diff --git a/approved_run_copy_test.go b/approved_run_copy_test.go
index 48097ea28..215b523b7 100644
--- a/approved_run_copy_test.go
+++ b/approved_run_copy_test.go
@@ -33,7 +33,7 @@ var runCopyIDs = func() []string {
// runCopyBuilt is every deck id whose text the code or docs now carry. The
// task that writes a row's text adds its id here in the same commit.
-var runCopyBuilt = []string{"RUN-C1", "RUN-C2", "RUN-C3", "RUN-C4", "RUN-C5", "RUN-C6", "RUN-C7", "RUN-C8", "RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14", "RUN-C43", "RUN-C44", "RUN-C45", "RUN-C46", "RUN-C82"}
+var runCopyBuilt = []string{"RUN-C1", "RUN-C2", "RUN-C3", "RUN-C4", "RUN-C5", "RUN-C6", "RUN-C7", "RUN-C8", "RUN-C9", "RUN-C10", "RUN-C11", "RUN-C12", "RUN-C13", "RUN-C14", "RUN-C15", "RUN-C16", "RUN-C17", "RUN-C18", "RUN-C19", "RUN-C20", "RUN-C21", "RUN-C23", "RUN-C24", "RUN-C25", "RUN-C26", "RUN-C27", "RUN-C28", "RUN-C29", "RUN-C30", "RUN-C31", "RUN-C32", "RUN-C33", "RUN-C34", "RUN-C35", "RUN-C42", "RUN-C43", "RUN-C44", "RUN-C45", "RUN-C46", "RUN-C82"}
func mustParseRunRecord(t *testing.T) []copyRow {
t.Helper()
diff --git a/internal/console/apiruns.go b/internal/console/apiruns.go
new file mode 100644
index 000000000..98eb80505
--- /dev/null
+++ b/internal/console/apiruns.go
@@ -0,0 +1,349 @@
+package console
+
+import (
+ "context"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "mime/multipart"
+ "net/http"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+// claimRunUnsupportedMsg is every run route's answer on a host-claim
+// console (index item 12, RUN-C42).
+const claimRunUnsupportedMsg = "carlos run is not available on this console yet. it needs the host-claim launch protocol"
+
+const runTooManyMsg = "too many run requests. wait a minute and try again"
+
+// runOpenMsg is refusal 7 (RUN-C24), at the door and at admission.
+func runOpenMsg(host string, run provision.RunState) string {
+ return fmt.Sprintf("%s already has a run open: %s, started by %s", host, run.ID, run.By)
+}
+
+// runMovingMsg is refusal 6 (RUN-C23), at the door and at admission.
+func runMovingMsg(host string) string {
+ return fmt.Sprintf("%s is moving. run the command once the move has finished", host)
+}
+
+// runImportMsg is refusal 9 (RUN-C33), at the door and at admission.
+func runImportMsg(host string) string {
+ return fmt.Sprintf("%s is an import that is not committed yet. finish or abort the import first", host)
+}
+
+// runFilesTooLargeMsg is refusal 8 for the file total (RUN-C30), whether meta
+// declares it or the body grows past it.
+const runFilesTooLargeMsg = "the files come to more than 64 MB"
+
+const runMalformedMsg = "the run request is malformed"
+
+const runRecordFailedMsg = "could not record the run. try again"
+
+// runClassRefusal is spec §5.1 refusal 5 as a sentence, "" for a
+// platform-run, hibernating exec instance on a platform box.
+func runClassRefusal(rec provision.Record) string {
+ switch {
+ case rec.Kind != "" && rec.Kind != "instance":
+ return fmt.Sprintf("%s is a %s route. carlos run works on instances", rec.Host, rec.Kind)
+ case rec.Addr != "" || len(rec.Addrs) > 0:
+ return fmt.Sprintf("%s is served from its own address, so the platform does not run its process", rec.Host)
+ case !rec.HibernateOrDefault():
+ return fmt.Sprintf("%s does not sleep, so the platform does not run its process", rec.Host)
+ case rec.Pool != "":
+ return fmt.Sprintf("%s is a pooled instance. carlos run works on instances with a process of their own", rec.Host)
+ }
+ if fleet, ok := registry.FleetPlacement(rec.Region); ok {
+ return fmt.Sprintf("%s runs on your fleet %s. carlos run works on platform boxes only", rec.Host, fleet)
+ }
+ return ""
+}
+
+// runMetaRefusal is refusal 8 as a sentence, "" when meta is valid.
+func runMetaRefusal(m runs.Meta) string {
+ switch err := runs.CheckArgv(m.Argv); {
+ case errors.Is(err, runs.ErrNoArgs):
+ return "name the command to run after --"
+ case errors.Is(err, runs.ErrTooManyArgs):
+ return "a run takes at most 256 arguments"
+ case errors.Is(err, runs.ErrArgvTooLarge):
+ return "the arguments come to more than 64 KB"
+ case errors.Is(err, runs.ErrArgNUL):
+ return "an argument holds a NUL byte"
+ }
+ if runs.CheckTimeout(time.Duration(m.TimeoutS)*time.Second) != nil {
+ return "the timeout must be between 1 second and 6 hours"
+ }
+ err := runs.CheckFiles(m.Files)
+ var fe *runs.FileNameError
+ switch {
+ case errors.Is(err, runs.ErrFilesTooLarge):
+ return runFilesTooLargeMsg
+ case errors.As(err, &fe) && fe.Twice:
+ return fmt.Sprintf("two files are both named %s", fe.Name)
+ case errors.As(err, &fe):
+ return fmt.Sprintf("%s is not a plain file name", fe.Name)
+ }
+ return ""
+}
+
+// spooled is one uploaded file on the console's disk.
+type spooled struct {
+ runs.File
+ path string
+}
+
+// spoolRunFiles reads the file parts, one per meta file in meta's order,
+// into temp files, checking each name and size and hashing it. The caller
+// removes the files. A short, long, misnamed or missing part is
+// malformed; a body past the cap is the 64 MB refusal.
+func spoolRunFiles(mr *multipart.Reader, files []runs.MetaFile) ([]spooled, error) {
+ var out []spooled
+ fail := func(err error) ([]spooled, error) {
+ for _, f := range out {
+ os.Remove(f.path)
+ }
+ return nil, err
+ }
+ for _, want := range files {
+ part, err := mr.NextPart()
+ if err != nil {
+ return fail(err)
+ }
+ if part.FormName() != "file" || part.FileName() != want.Name {
+ return fail(errRunMalformed)
+ }
+ tmp, err := os.CreateTemp("", "carlos-run-*")
+ if err != nil {
+ return fail(err)
+ }
+ h := sha256.New()
+ n, err := io.Copy(io.MultiWriter(tmp, h), io.LimitReader(part, want.Size+1))
+ cerr := tmp.Close()
+ out = append(out, spooled{File: runs.File{Name: want.Name, Size: n, SHA256: hex.EncodeToString(h.Sum(nil))}, path: tmp.Name()})
+ if err != nil {
+ return fail(err)
+ }
+ if cerr != nil {
+ return fail(cerr)
+ }
+ if n != want.Size {
+ return fail(errRunMalformed)
+ }
+ }
+ if _, err := mr.NextPart(); err != io.EOF {
+ return fail(errRunMalformed)
+ }
+ return out, nil
+}
+
+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.
+func (s *Server) handleAPIRunCreate(w http.ResponseWriter, r *http.Request) {
+ if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
+ http.NotFound(w, r)
+ return
+ }
+ actor, acct, app, ok := s.apiAccountApp(w, r)
+ if !ok {
+ return
+ }
+ if s.cfg.HostClaim {
+ jsonError(w, http.StatusConflict, claimRunUnsupportedMsg)
+ return
+ }
+ if !s.cli.instances.allow(hashToken(bearer(r)), time.Now()) {
+ jsonError(w, http.StatusTooManyRequests, runTooManyMsg)
+ return
+ }
+ if s.cfg.Releases == nil {
+ jsonError(w, http.StatusServiceUnavailable, runRecordFailedMsg)
+ return
+ }
+ ctx := r.Context()
+ host := r.PathValue("host")
+ notInstance := fmt.Sprintf("%s is not an instance of %s", host, app)
+ if !provision.ValidHostShape(host) {
+ jsonError(w, http.StatusNotFound, notInstance)
+ return
+ }
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, acct.Sqid, app, host)
+ if errors.Is(err, os.ErrNotExist) {
+ jsonError(w, http.StatusNotFound, notInstance)
+ return
+ }
+ if err != nil {
+ log.Printf("console: api run %s/%s/%s: read record: %v", acct.Sqid, app, host, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ if msg := runClassRefusal(*rec); msg != "" {
+ jsonError(w, http.StatusConflict, msg)
+ return
+ }
+ if rec.Move != nil {
+ jsonError(w, http.StatusConflict, runMovingMsg(host))
+ return
+ }
+ if rec.ActiveRun.Open() {
+ jsonError(w, http.StatusConflict, runOpenMsg(host, *rec.ActiveRun))
+ return
+ }
+ r.Body = http.MaxBytesReader(w, r.Body, runs.MaxFileBytes+runs.MetaAllowance)
+ mr, err := r.MultipartReader()
+ if err != nil {
+ jsonError(w, http.StatusBadRequest, runMalformedMsg)
+ return
+ }
+ var meta runs.Meta
+ part, err := mr.NextPart()
+ if err == nil && part.FormName() == "meta" {
+ err = json.NewDecoder(io.LimitReader(part, runs.MetaAllowance)).Decode(&meta)
+ } else if err == nil {
+ err = errRunMalformed
+ }
+ if err != nil {
+ jsonError(w, http.StatusBadRequest, runMalformedMsg)
+ return
+ }
+ if msg := runMetaRefusal(meta); msg != "" {
+ jsonError(w, http.StatusBadRequest, msg)
+ return
+ }
+ if rec.ImportUncommitted() {
+ jsonError(w, http.StatusConflict, runImportMsg(host))
+ return
+ }
+ files, err := spoolRunFiles(mr, meta.Files)
+ defer func() {
+ for _, f := range files {
+ os.Remove(f.path)
+ }
+ }()
+ var tooLarge *http.MaxBytesError
+ switch {
+ case errors.As(err, &tooLarge):
+ jsonError(w, http.StatusRequestEntityTooLarge, runFilesTooLargeMsg)
+ return
+ case err != nil:
+ jsonError(w, http.StatusBadRequest, runMalformedMsg)
+ return
+ }
+ id, err := runs.MintID()
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ 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}
+ for _, f := range files {
+ req.Files = append(req.Files, f.File)
+ }
+ if err := s.writeRunObjects(ctx, req, files); err != nil {
+ log.Printf("console: api run %s/%s/%s: write %s: %v", acct.Sqid, app, host, id, err)
+ _ = runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ // The pickup window opens at admission, not at the first byte: a slow
+ // upload must not eat the box's 60 seconds.
+ got, err := provision.SetRecordRun(ctx, s.cfg.Releases, acct.Sqid, app, host,
+ provision.RunAdmit(provision.RunAdmission{ID: id, By: actor, Deadline: time.Now().UTC().Add(runs.PickupWindow)}))
+ if err != nil {
+ _ = runs.DeleteRun(context.WithoutCancel(ctx), s.cfg.Releases, acct.Sqid, app, id)
+ var open *provision.RunOpenError
+ switch {
+ case errors.As(err, &open):
+ jsonError(w, http.StatusConflict, runOpenMsg(host, open.Run))
+ case errors.Is(err, provision.ErrRunMoving):
+ jsonError(w, http.StatusConflict, runMovingMsg(host))
+ case errors.Is(err, provision.ErrRunImport):
+ jsonError(w, http.StatusConflict, runImportMsg(host))
+ case errors.Is(err, os.ErrNotExist):
+ jsonError(w, http.StatusNotFound, notInstance)
+ default:
+ log.Printf("console: api run %s/%s/%s: admit %s: %v", acct.Sqid, app, host, id, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ }
+ return
+ }
+ s.bumpSerial(ctx)
+ s.auditActivity(acct, app, actor, Activity{Action: "run-start", TargetType: "instance", Target: host,
+ Detail: runStartDetail(req, s.runVersion(ctx, acct.Sqid, app, rec.Channel))})
+ writeJSON(w, http.StatusAccepted, runs.Created{ID: id, Host: host, State: got.ActiveRun.State})
+}
+
+// writeRunObjects writes request.json first (the orphan sweep ages a run
+// directory by it, Task 5), then the files. The record CAS comes after
+// both, so no box ever sees a run whose objects are partial.
+func (s *Server) writeRunObjects(ctx context.Context, req runs.Request, files []spooled) error {
+ body, err := json.Marshal(req)
+ if err != nil {
+ return err
+ }
+ if _, err := s.cfg.Releases.Put(ctx, runs.RequestKey(req.Account, req.App, req.ID), body, ""); err != nil {
+ return err
+ }
+ for _, f := range files {
+ fh, err := os.Open(f.path)
+ if err != nil {
+ return err
+ }
+ _, err = release.PutObjectFrom(ctx, s.cfg.Releases, runs.InputKey(req.Account, req.App, req.ID, f.Name), fh, f.Size, "")
+ fh.Close()
+ if err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// runVersion is the release the instance's channel serves, "" when it
+// cannot be read: the audit row's version is a record, not a gate.
+func (s *Server) runVersion(ctx context.Context, sqid, app, channel string) string {
+ ptr, err := release.Channel(ctx, s.cfg.Releases, s.cfg.ReleasePub, sqid, app, channel)
+ if err != nil || ptr == nil {
+ return ""
+ }
+ return ptr.Version
+}
+
+// runStartDetail is the run-start row's detail (run spec §6): the run id,
+// argv, the file names and sizes and the version, as JSON. Output never
+// goes in the audit log; argv does, and the docs say so.
+func runStartDetail(req runs.Request, version string) string {
+ type file struct {
+ Name string `json:"name"`
+ Size int64 `json:"size"`
+ }
+ d := struct {
+ ID string `json:"id"`
+ Argv []string `json:"argv"`
+ Files []file `json:"files"`
+ Version string `json:"version"`
+ }{ID: req.ID, Argv: req.Argv, Version: version}
+ for _, f := range req.Files {
+ d.Files = append(d.Files, file{f.Name, f.Size})
+ }
+ raw, _ := json.Marshal(d)
+ return string(raw)
+}
diff --git a/internal/console/apiruns_test.go b/internal/console/apiruns_test.go
new file mode 100644
index 000000000..d5025b6f1
--- /dev/null
+++ b/internal/console/apiruns_test.go
@@ -0,0 +1,329 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "io"
+ "mime/multipart"
+ "net/http"
+ "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 runTestHost = "acme.go.tito.io"
+
+type runFx struct {
+ *consoleFixture
+ member string
+}
+
+func runFixture(t *testing.T) *runFx {
+ t.Helper()
+ fx, member := instanceFixture(t, "code-run")
+ putDBRecord(t, fx, provision.Record{Host: runTestHost, Channel: "stable", CreatedAt: time.Now().UTC()})
+ return &runFx{consoleFixture: fx, member: member}
+}
+
+func (fx *runFx) runsURL(host string) string {
+ return fx.ts.URL + "/api/cli/apps/" + release.OpsAccount + "/titogo/instances/" + host + "/runs"
+}
+
+// postRun sends the multipart create; parts lets a test send parts that
+// disagree with meta.
+func postRun(t *testing.T, u, bearer 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())
+ if bearer != "" {
+ req.Header.Set("Authorization", "Bearer "+bearer)
+ }
+ 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)
+}
+
+func okMeta() runs.Meta { return runs.Meta{Argv: []string{"import", "x.csv"}, TimeoutS: 60} }
+
+func errorOf(t *testing.T, body string) string {
+ t.Helper()
+ var e struct{ Error string }
+ if err := json.Unmarshal([]byte(body), &e); err != nil {
+ t.Fatalf("body %q: %v", body, err)
+ }
+ return e.Error
+}
+
+func TestAPIRunCreateAdmitsAndRecords(t *testing.T) {
+ fx := runFixture(t)
+ meta := okMeta()
+ meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
+ status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
+ if status != http.StatusAccepted {
+ t.Fatalf("status %d %s", status, body)
+ }
+ var c runs.Created
+ if err := json.Unmarshal([]byte(body), &c); err != nil || !runs.ValidID(c.ID) || c.State != provision.RunQueued {
+ t.Fatalf("created %+v, %v", c, err)
+ }
+ ctx := context.Background()
+ rec, err := provision.GetRecord(ctx, fx.rel, release.OpsAccount, "titogo", runTestHost)
+ if err != nil || rec.ActiveRun == nil || rec.ActiveRun.ID != c.ID || rec.ActiveRun.By != "alice@keymail.dev" {
+ t.Fatalf("record run %+v, %v", rec.ActiveRun, err)
+ }
+ if d := time.Until(rec.ActiveRun.Deadline); d < 50*time.Second || d > 61*time.Second {
+ t.Errorf("pickup deadline %s away, want about 60 s", d)
+ }
+ req, err := runs.ReadRequest(ctx, fx.rel, release.OpsAccount, "titogo", c.ID)
+ if err != nil || req.Host != runTestHost || len(req.Incarnation) != 64 || req.TimeoutS != 60 || len(req.Files) != 1 || len(req.Files[0].SHA256) != 64 {
+ t.Fatalf("request %+v, %v", req, err)
+ }
+ raw, _, err := fx.rel.Get(ctx, runs.InputKey(release.OpsAccount, "titogo", c.ID, "x.csv"))
+ if err != nil || string(raw) != "a,b,c" {
+ t.Fatalf("input %q, %v", raw, err)
+ }
+ events, _ := fx.st.Events(OpsAccountID, "titogo", 10)
+ found := false
+ for _, e := range events {
+ if e.Action == "run-start" && e.Target == runTestHost && e.Address == "alice@keymail.dev" && strings.Contains(e.Detail, c.ID) && strings.Contains(e.Detail, "x.csv") {
+ found = true
+ }
+ }
+ if !found {
+ t.Errorf("no run-start audit row in %+v", events)
+ }
+}
+
+// Each refusal in spec §5.1's order: every case is valid in every later
+// respect, so the status and message show which check refused it.
+func TestAPIRunCreateRefusesInOrder(t *testing.T) {
+ fx := runFixture(t)
+ ctx := context.Background()
+ svc := serviceToken(t, fx.st, OpsAccountID, "ci", "titogo", svcrole.Instance)
+ stranger := cliToken(t, fx.consoleFixture, "mallory@keymail.dev", "code-run-stranger")
+
+ if status, _ := postRun(t, fx.runsURL(runTestHost), svc, okMeta(), nil); status != http.StatusForbidden {
+ t.Errorf("service credential: %d, want 403", status)
+ }
+ status, body := postRun(t, fx.runsURL(runTestHost), stranger, okMeta(), nil)
+ want := canonicalNotFound(t)
+ if status != want.Status || body != want.Body {
+ t.Errorf("stranger: %d %q, want the byte-identical 404", status, body)
+ }
+ if status, body := postRun(t, fx.runsURL("nobody.go.tito.io"), fx.member, okMeta(), nil); status != http.StatusNotFound || errorOf(t, body) != "nobody.go.tito.io is not an instance of titogo" {
+ t.Errorf("no record: %d %s", status, body)
+ }
+ for host, c := range map[string]struct {
+ rec provision.Record
+ msg string
+ }{
+ "static.go.tito.io": {provision.Record{Kind: "static"}, "static.go.tito.io is a static route. carlos run works on instances"},
+ "addr.go.tito.io": {provision.Record{Addr: "127.0.0.1:9"}, "addr.go.tito.io is served from its own address, so the platform does not run its process"},
+ "pool.go.tito.io": {provision.Record{Pool: "tito"}, "pool.go.tito.io is a pooled instance. carlos run works on instances with a process of their own"},
+ "fleet.go.tito.io": {provision.Record{Region: "fleet/pi"}, "fleet.go.tito.io runs on your fleet pi. carlos run works on platform boxes only"},
+ } {
+ c.rec.Host = host
+ putDBRecord(t, fx.consoleFixture, c.rec)
+ if status, body := postRun(t, fx.runsURL(host), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != c.msg {
+ t.Errorf("%s: %d %s", host, status, body)
+ }
+ }
+ no := false
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "awake.go.tito.io", Hibernate: &no})
+ if status, body := postRun(t, fx.runsURL("awake.go.tito.io"), fx.member, okMeta(), nil); status != http.StatusConflict || !strings.Contains(errorOf(t, body), "does not sleep") {
+ t.Errorf("always-on: %d %s", status, body)
+ }
+ if _, err := provision.SetRecordMove(ctx, fx.rel, release.OpsAccount, "titogo", runTestHost, provision.MoveRequest("uk", time.Now())); err != nil {
+ t.Fatal(err)
+ }
+ if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != runTestHost+" is moving. run the command once the move has finished" {
+ t.Errorf("moving: %d %s", status, body)
+ }
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "busy.go.tito.io", ActiveRun: &provision.RunState{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", State: provision.RunPicked, By: "bob@keymail.dev"}})
+ if status, body := postRun(t, fx.runsURL("busy.go.tito.io"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "busy.go.tito.io already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev" {
+ t.Errorf("open run: %d %s", status, body)
+ }
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "fresh.go.tito.io"})
+ bad := []struct {
+ meta runs.Meta
+ msg string
+ }{
+ {runs.Meta{TimeoutS: 60}, "name the command to run after --"},
+ {runs.Meta{Argv: make([]string, 257), TimeoutS: 60}, "a run takes at most 256 arguments"},
+ {runs.Meta{Argv: []string{strings.Repeat("a", 64<<10+1)}, TimeoutS: 60}, "the arguments come to more than 64 KB"},
+ {runs.Meta{Argv: []string{"a\x00"}, TimeoutS: 60}, "an argument holds a NUL byte"},
+ {runs.Meta{Argv: []string{"a"}, TimeoutS: 6*3600 + 1}, "the timeout must be between 1 second and 6 hours"},
+ {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 65 << 20}}}, "the files come to more than 64 MB"},
+ {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "../x", Size: 1}}}, "../x is not a plain file name"},
+ {runs.Meta{Argv: []string{"a"}, TimeoutS: 60, Files: []runs.MetaFile{{Name: "a", Size: 1}, {Name: "a", Size: 1}}}, "two files are both named a"},
+ }
+ for i := range bad[1].meta.Argv {
+ bad[1].meta.Argv[i] = "x"
+ }
+ for _, c := range bad {
+ if status, body := postRun(t, fx.runsURL("fresh.go.tito.io"), fx.member, c.meta, nil); status != http.StatusBadRequest || errorOf(t, body) != c.msg {
+ t.Errorf("%q: %d %s", c.msg, status, body)
+ }
+ }
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: "import.go.tito.io", Seed: &provision.SeedSource{Kind: provision.SeedKindImport, State: "placed"}})
+ if status, body := postRun(t, fx.runsURL("import.go.tito.io"), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != "import.go.tito.io is an import that is not committed yet. finish or abort the import first" {
+ t.Errorf("import: %d %s", status, body)
+ }
+}
+
+func TestAPIRunCreateSpendsTheInstanceBudget(t *testing.T) {
+ fx := runFixture(t)
+ // A window of one: the first hit in a window is always allowed
+ // (rateLimiter.allow, authcli.go:226), the second is over. The budget
+ // is refusal 3, so it comes before refusal 7 (the run the first opened).
+ fx.srv.cli.instances = newRateLimiter(1, cliRateWindow)
+ if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusAccepted {
+ t.Fatalf("first: %d %s", status, body)
+ }
+ if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusTooManyRequests || errorOf(t, body) != "too many run requests. wait a minute and try again" {
+ t.Fatalf("over budget: %d %s", status, body)
+ }
+}
+
+func TestAPIRunCreateRefusesAPartThatDisagreesWithMeta(t *testing.T) {
+ fx := runFixture(t)
+ meta := okMeta()
+ meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
+ for name, parts := range map[string][]struct{ name, body string }{
+ "short": {{"x.csv", "abc"}},
+ "long": {{"x.csv", "abcdefg"}},
+ "wrong name": {{"y.csv", "abcde"}},
+ "missing": nil,
+ } {
+ status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, parts)
+ if status != http.StatusBadRequest || errorOf(t, body) != "the run request is malformed" {
+ t.Errorf("%s: %d %s", name, status, body)
+ }
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "titogo")); len(left) != 0 {
+ t.Fatalf("a refused upload left %q", left)
+ }
+ if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost); rec.ActiveRun != nil {
+ t.Fatal("a refused upload admitted a run")
+ }
+}
+
+// Two members at once: one run is admitted, the other is told who holds the
+// instance, and its objects are gone (codex R1-1).
+func TestAPIRunCreateRaceAdmitsOne(t *testing.T) {
+ fx := runFixture(t)
+ var wg sync.WaitGroup
+ statuses := make([]int, 2)
+ bodies := make([]string, 2)
+ for i := range 2 {
+ wg.Add(1)
+ go func() {
+ defer wg.Done()
+ statuses[i], bodies[i] = postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil)
+ }()
+ }
+ wg.Wait()
+ won, lost := 0, 0
+ for i := range 2 {
+ switch statuses[i] {
+ case http.StatusAccepted:
+ won++
+ case http.StatusConflict:
+ lost++
+ if !strings.Contains(errorOf(t, bodies[i]), "already has a run open") {
+ t.Errorf("loser said %s", bodies[i])
+ }
+ }
+ }
+ if won != 1 || lost != 1 {
+ t.Fatalf("statuses %v %v", statuses, bodies)
+ }
+ // DirStore keeps emptied directories, so count the runs that still own
+ // objects rather than the directories.
+ prefix := runs.RunsPrefix(release.OpsAccount, "titogo")
+ keys, _ := fx.rel.List(context.Background(), prefix)
+ owners := map[string]bool{}
+ for _, k := range keys {
+ owners[strings.SplitN(strings.TrimPrefix(k, prefix), "/", 2)[0]] = true
+ }
+ if len(owners) != 1 {
+ t.Fatalf("runs with objects %v, want only the winner's", owners)
+ }
+}
+
+// rivalStore admits another run on the instance record the moment the door
+// writes request.json: after the door's own checks passed and its objects
+// began to land, before its admission. That is the race, made certain.
+type rivalStore struct {
+ release.Store
+ t *testing.T
+ once sync.Once
+}
+
+func (r *rivalStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ if strings.HasSuffix(key, "/request.json") {
+ r.once.Do(func() {
+ _, err := provision.SetRecordRun(ctx, r.Store, release.OpsAccount, "titogo", runTestHost,
+ provision.RunAdmit(provision.RunAdmission{ID: "run_aaaaaaaaaaaaaaaaaaaaaaaaaa", By: "bob@keymail.dev", Deadline: time.Now().Add(time.Minute)}))
+ if err != nil {
+ r.t.Error(err)
+ }
+ })
+ }
+ return r.Store.Put(ctx, key, body, ifMatch)
+}
+
+// The loser of an admission that happens after its objects began to land is
+// told who holds the instance, and nothing of its own is left in the bucket.
+func TestAPIRunCreateLoserAfterUploadLeavesNothing(t *testing.T) {
+ fx := runFixture(t)
+ fx.srv.cfg.Releases = &rivalStore{Store: fx.rel, t: t}
+ meta := okMeta()
+ meta.Files = []runs.MetaFile{{Name: "x.csv", Size: 5}}
+ status, body := postRun(t, fx.runsURL(runTestHost), fx.member, meta, []struct{ name, body string }{{"x.csv", "a,b,c"}})
+ if want := "already has a run open: run_aaaaaaaaaaaaaaaaaaaaaaaaaa, started by bob@keymail.dev"; status != http.StatusConflict || !strings.HasSuffix(errorOf(t, body), want) {
+ t.Fatalf("%d %s", status, body)
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.RunsPrefix(release.OpsAccount, "titogo")); len(left) != 0 {
+ t.Fatalf("the loser left %q", left)
+ }
+}
+
+func TestAPIRunCreateOnAHostClaimConsole(t *testing.T) {
+ fx := runFixture(t)
+ fx.srv.cfg.HostClaim = true
+ if status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil); status != http.StatusConflict || errorOf(t, body) != claimRunUnsupportedMsg {
+ t.Fatalf("%d %s", status, body)
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index d7aa40b9a..93b2c0067 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -99,6 +99,9 @@ const (
// token's noise would starve the app's own signups, and a busy signup
// hour would refuse its minter's promotes.
cliInstanceBurst = 60
+ // cliRunsBurst is the follow routes' per-token budget per window: a
+ // once-a-second poll, its acks and cancels, with room.
+ cliRunsBurst = 600
// cliCreateWindow is app creation's own budget window — a DAY, the
// longest here, because claiming an app name is the rarest deliberate
@@ -335,6 +338,10 @@ type cliLimits struct {
// apidb.go): a restore is dozens of reads, a different shape from the
// instance verbs.
db *rateLimiter
+ // runs is the follow routes' budget (run spec §5.6): the CLI polls once
+ // a second, which the 60-a-minute instance window would refuse. The
+ // create door spends instances, as spec refusal 3 says.
+ runs *rateLimiter
// svcCreate, svcWrite and svcShip are the SERVICE side of create/write/
// ship — separate counters, separate keys (principalKey's NAME, not the
// person's ADDRESS or the ship-person's token hash), so a credential's
@@ -380,6 +387,7 @@ func newCLILimits() *cliLimits {
// budget is the wrong principal for it in both directions.
instances: newRateLimiter(cliInstanceBurst, cliRateWindow),
db: newRateLimiter(cliDBBurst, cliRateWindow),
+ runs: newRateLimiter(cliRunsBurst, cliRateWindow),
// Create gets its own budget on its own (day-long) window, keyed by
// address — the stolen-token ceiling handleAPICreateApp explains.
create: newRateLimiter(cliCreateBurst, cliCreateWindow),
@@ -681,6 +689,7 @@ func (s *Server) registerCLIRoutes() {
s.registerMovesPhase1Routes()
}
s.cliRoute("DELETE /api/cli/apps/{account}/{app}/instances/{host}", svcrole.Allow(svcrole.Instance), s.handleAPIInstanceDelete)
+ s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs", svcrole.Set{}, s.handleAPIRunCreate)
// The `carlos db` replica door (apidb.go; spec 2026-09-02 §6): an
// instance's replicated LTX files, by coordinate, for a member to
// restore on their own machine. The zero Set, deliberately: the payload
diff --git a/internal/console/claim_invariant_fixtures_test.go b/internal/console/claim_invariant_fixtures_test.go
index ee95ba83d..f497d1370 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1148,6 +1148,14 @@ var claimFixtures = map[string]claimFixture{
},
do: claimAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/web.go.tito.io/channel", `{"channel":"beta"}`),
}},
+ // carlos run waits for the claim's reservation protocol (#250): every
+ // run route refuses on a host-claim console (claimUnsupported).
+ "route POST /api/cli/apps/{account}/{app}/instances/{host}/runs": {refused: &claimRefusalFixture{
+ apps: claimRefs("bab/titogo"),
+ do: claimRefusedAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/acme.go.tito.io/runs", `{}`),
+ status: http.StatusConflict,
+ msg: claimRunUnsupportedMsg,
+ }},
// Instance moves are intentionally unsupported (claimUnsupported): the
// request is valid (a staging region exists), the cut follows a
// promote-ready move.
diff --git a/internal/console/claim_invariant_test.go b/internal/console/claim_invariant_test.go
index 7720daa00..996f11de3 100644
--- a/internal/console/claim_invariant_test.go
+++ b/internal/console/claim_invariant_test.go
@@ -92,6 +92,8 @@ type claimRefusalFixture struct {
// these may be a refused fixture; every other app mutator is a door that
// must hold the gate and reach its effect.
var claimUnsupported = map[string]string{
+ claimRunUnsupportedMsg: "carlos run launches a cmd unit that the claim's isolated root protocol has no reservation for (#250); " +
+ "every run route refuses before it reads or writes anything",
claimHostResolveUnsupportedMsg: "host-loss declaration/readmission changes claim roots; the existing service has no scoped lifecycle grant in this slice",
claimMoveUnsupportedMsg: "instance move request, cut and abort rewrite the record's move journal, " +
"which only the unbuilt host-claim Moves protocol may do under its own operation",
diff --git a/internal/console/svcroutes_test.go b/internal/console/svcroutes_test.go
index 04175b8e6..60cb60665 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -186,6 +186,9 @@ var closedDoors = []string{
// §6, "Who may open it").
"GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}",
"GET /api/cli/apps/{account}/{app}/db/{host}/ltx/{level}/{span}",
+ // `carlos run` (run spec §2 ruling 1): a person who is a member, the same
+ // rule as the db door. A service credential cannot run commands.
+ "POST /api/cli/apps/{account}/{app}/instances/{host}/runs",
// The build probe is what `carlos deploy`'s watch TRUSTS in place of the
// X-Carlos-Version header — the header being the thing that reports a
// release before any process has restarted. A credential that could
@@ -899,8 +902,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
}
seen++
}
- if seen != len(closedDoors) || seen != 42 {
- t.Errorf("walked %d closed doors, want all %d (and 42 today: instance-token, three door-key doors, the routes door, instances-manifest, sidecars-manifest, the route-channel move, two db replica doors, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read, the layout read, seven system doors, two deployment-wide antivirus, three platform geoip doors, and four store-entitlement doors)",
+ if seen != len(closedDoors) || seen != 43 {
+ t.Errorf("walked %d closed doors, want all %d (and 43 today: instance-token, three door-key doors, the routes door, instances-manifest, sidecars-manifest, the route-channel move, two db replica doors, six fleets, four services, two email credentials, email pause, the build probe, the live tenancy read, the layout read, seven system doors, two deployment-wide antivirus, three platform geoip doors, four store-entitlement doors, and the run create door)",
seen, len(closedDoors))
}
}
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index 3911358f0..bd112c37e 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -26,6 +26,8 @@
../mail/aws.go|NewAWSProvisioner|iam.NewFromConfig(cfg)
../mail/aws.go|NewAWSProvisioner|sesv2.NewFromConfig(cfg)
../mail/aws.go|NewAWSProvisioner|sts.NewFromConfig(cfg)
+apiruns.go|handleAPIRunCreate|rec.ActiveRun.Open()
+apiruns.go|writeRunObjects|os.Open(f.path)
cepull.go|realCECosts|costexplorer.NewFromConfig(cfg)
claim_alias.go|attachDomainClaim|a.Control.RegisterStart(ctx, acct.Sqid, app, appID, id, reg, time.Now())
claim_app_admission.go|bindClaimAppIdentity|s.claimEnrollmentContext(ctx, acct, app, appID)
diff --git a/internal/console/testdata/claim-routes.txt b/internal/console/testdata/claim-routes.txt
index 6148a6bc8..0acd96388 100644
--- a/internal/console/testdata/claim-routes.txt
+++ b/internal/console/testdata/claim-routes.txt
@@ -172,6 +172,7 @@ route POST /api/cli/apps/{account}/{app}/instances
route POST /api/cli/apps/{account}/{app}/instances/{host}/channel
route POST /api/cli/apps/{account}/{app}/instances/{host}/move
route POST /api/cli/apps/{account}/{app}/instances/{host}/move/cut
+route POST /api/cli/apps/{account}/{app}/instances/{host}/runs
route POST /api/cli/apps/{account}/{app}/keep
route POST /api/cli/apps/{account}/{app}/ledgers/publish
route POST /api/cli/apps/{account}/{app}/ledgers/{ledger}/append