amadan

carlos / platform Public

console: follow, cancel, expire and ack a run; run-end once (run spec §5.6, §5.7, §6, codex 5)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude bec020493d338b4c7b2fe4f14c60b1b7d4b00a05 parent 76acbe7
8 files changed, +723 −3
  • approved_run_copy_test.go +1 −1
  • internal/console/apirunsfollow.go +349 −0
  • internal/console/apirunsfollow_test.go +333 −0
  • internal/console/authcli.go +4 −0
  • internal/console/claim_invariant_fixtures_test.go +24 −0
  • internal/console/svcroutes_test.go +6 −2
  • internal/console/testdata/claim-authority-sites.txt +2 −0
  • internal/console/testdata/claim-routes.txt +4 −0
diff --git a/approved_run_copy_test.go b/approved_run_copy_test.go
index 215b523b7..6cdae40a3 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-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"}
+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-C36", "RUN-C37", "RUN-C38", "RUN-C39", "RUN-C40", "RUN-C41", "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/apirunsfollow.go b/internal/console/apirunsfollow.go
new file mode 100644
index 000000000..bf28905af
--- /dev/null
+++ b/internal/console/apirunsfollow.go
@@ -0,0 +1,349 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+const runReadFailedMsg = "could not read the run. try again"
+
+// runCall is one follow route's resolved request.
+type runCall struct {
+ actor string
+ acct *Account
+ app string
+ host string
+ id string
+}
+
+func noRunMsg(host, id string) string { return fmt.Sprintf("%s has no run %s", host, id) }
+
+// runGate is the four follow routes' shared gate, in the create door's
+// order: a person, a member (the byte-identical 404), the host-claim
+// refusal, the runs budget, then a run id of the right shape.
+func (s *Server) runGate(w http.ResponseWriter, r *http.Request) (runCall, bool) {
+ if p, ok := principalFrom(r.Context()); !ok || p.Kind != PrincipalUser {
+ http.NotFound(w, r)
+ return runCall{}, false
+ }
+ actor, acct, app, ok := s.apiAccountApp(w, r)
+ if !ok {
+ return runCall{}, false
+ }
+ if s.cfg.HostClaim {
+ jsonError(w, http.StatusConflict, claimRunUnsupportedMsg)
+ return runCall{}, false
+ }
+ if !s.cli.runs.allow(hashToken(bearer(r)), time.Now()) {
+ jsonError(w, http.StatusTooManyRequests, runTooManyMsg)
+ return runCall{}, false
+ }
+ c := runCall{actor: actor, acct: acct, app: app, host: r.PathValue("host"), id: r.PathValue("id")}
+ if s.cfg.Releases == nil {
+ jsonError(w, http.StatusServiceUnavailable, runReadFailedMsg)
+ return runCall{}, false
+ }
+ if !provision.ValidHostShape(c.host) || !runs.ValidID(c.id) {
+ jsonError(w, http.StatusNotFound, noRunMsg(c.host, c.id))
+ return runCall{}, false
+ }
+ return c, true
+}
+
+// runOf is the record's run when it is c.id, nil otherwise (the record is
+// gone, or names another run). A read failure is an error.
+func (s *Server) runOf(ctx context.Context, c runCall) (*provision.RunState, error) {
+ rec, err := provision.GetRecord(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ if rec.ActiveRun != nil && rec.ActiveRun.ID == c.id {
+ return rec.ActiveRun, nil
+ }
+ return nil, nil
+}
+
+// requestOnHost reports whether the run's objects exist and belong to c.host.
+// A run id names one run in the app's subtree; the host in the path must be
+// the host the run was admitted for. A read failure is an error.
+func (s *Server) requestOnHost(ctx context.Context, c runCall) (bool, error) {
+ req, err := runs.ReadRequest(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
+ if errors.Is(err, os.ErrNotExist) {
+ return false, nil
+ }
+ if err != nil {
+ return false, err
+ }
+ return req.Host == c.host, nil
+}
+
+// runWriteCtx is the context for a record write that must run to its end
+// even if the caller has gone: a cancelled conditional write may or may not
+// have landed, and what follows (an audit row, a delete) depends on which.
+func runWriteCtx(ctx context.Context) (context.Context, context.CancelFunc) {
+ return context.WithTimeout(context.WithoutCancel(ctx), 30*time.Second)
+}
+
+// handleAPIRunGet is GET …/runs/{id}?after=<seq> (run spec §5.6). It reads
+// result.json BEFORE it lists chunks: the box writes the result only after
+// every chunk is durable, so a page that carries the result also carries
+// (or pages on to) every chunk up to its last_seq. Reading them the other
+// way round could list the chunks, then see a result written after, and
+// tell the caller the run is complete with chunks missing.
+func (s *Server) handleAPIRunGet(w http.ResponseWriter, r *http.Request) {
+ c, ok := s.runGate(w, r)
+ if !ok {
+ return
+ }
+ ctx := r.Context()
+ var after int64
+ if q := r.URL.Query().Get("after"); q != "" {
+ var err error
+ if after, err = strconv.ParseInt(q, 10, 64); err != nil || after < 0 {
+ jsonError(w, http.StatusBadRequest, runMalformedMsg)
+ return
+ }
+ }
+ run, err := s.runOf(ctx, c)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ }
+ if run == nil {
+ // The record no longer names the run (cleared, or another run since):
+ // its objects are what is left of it.
+ found, err := s.requestOnHost(ctx, c)
+ switch {
+ case err != nil:
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ case !found:
+ jsonError(w, http.StatusNotFound, noRunMsg(c.host, c.id))
+ return
+ }
+ }
+ res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ }
+ keys, err := runs.ListChunks(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id, after, runs.PageChunks+1)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ }
+ p := runs.Page{ID: c.id, Result: res, Chunks: []runs.Chunk{}}
+ if run != nil {
+ p.State = run.State
+ }
+ out := runs.OutPrefix(c.acct.Sqid, c.app, c.id)
+ size := 0
+ for i, k := range keys {
+ if i == runs.PageChunks {
+ p.More = true
+ break
+ }
+ seq, stream, ok := runs.ParseChunkKey(k, out)
+ if !ok {
+ continue
+ }
+ raw, _, err := s.cfg.Releases.Get(ctx, k)
+ if err != nil {
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ }
+ if size+len(raw) > runs.PageBytes && len(p.Chunks) > 0 {
+ p.More = true
+ break
+ }
+ size += len(raw)
+ p.Chunks = append(p.Chunks, runs.Chunk{Seq: seq, Stream: stream, Data: raw})
+ }
+ if res != nil {
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "result", res)
+ }
+ writeJSON(w, http.StatusOK, p)
+}
+
+// handleAPIRunCancel is POST …/runs/{id}/cancel (run spec §5.7): queued
+// becomes cancelled (terminal; the run-end row is written now), picked
+// becomes cancel and the serial is bumped so the owning box hears it.
+func (s *Server) handleAPIRunCancel(w http.ResponseWriter, r *http.Request) {
+ c, ok := s.runGate(w, r)
+ if !ok {
+ return
+ }
+ ctx, cancel := runWriteCtx(r.Context())
+ defer cancel()
+ got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunCancelTransition(c.id))
+ if err != nil {
+ s.runTransitionRefused(w, ctx, c, err)
+ return
+ }
+ switch got.ActiveRun.State {
+ case provision.RunCancelled:
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "cancelled", nil)
+ case provision.RunCancel:
+ s.bumpSerial(ctx)
+ }
+ writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State})
+}
+
+// handleAPIRunExpire is POST …/runs/{id}/expire (run spec §5.2): the CLI's
+// "not picked up", asked for sixty seconds after its request. The route
+// applies the transition and does not judge the clock; the sweep, which
+// expires on its own, decides by the record's pickup Deadline (stamped at
+// admission, after the upload), never by request.json's requested_at. A
+// pickup that landed first wins.
+func (s *Server) handleAPIRunExpire(w http.ResponseWriter, r *http.Request) {
+ c, ok := s.runGate(w, r)
+ if !ok {
+ return
+ }
+ ctx, cancel := runWriteCtx(r.Context())
+ defer cancel()
+ got, err := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunExpire(c.id))
+ if err != nil {
+ s.runTransitionRefused(w, ctx, c, err)
+ return
+ }
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "expired", nil)
+ writeJSON(w, http.StatusOK, runs.StateAnswer{State: got.ActiveRun.State})
+}
+
+// runTransitionRefused names why a cancel or expire did not apply, from a
+// fresh read of the record. A run the record shows cancelled or expired
+// gets its run-end row here too (written once, so a no-op when the call
+// that ended it already wrote it): a cancel whose response was lost and was
+// retried finds its own terminal state, and the row must not depend on the
+// first attempt having got that far.
+func (s *Server) runTransitionRefused(w http.ResponseWriter, ctx context.Context, c runCall, err error) {
+ if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("console: api run %s/%s/%s %s: %v", c.acct.Sqid, c.app, c.host, c.id, err)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ run, rerr := s.runOf(ctx, c)
+ switch {
+ case rerr != nil:
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ case run == nil:
+ jsonError(w, http.StatusNotFound, noRunMsg(c.host, c.id))
+ case run.Open():
+ jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has already started", c.id))
+ default:
+ if run.State == provision.RunCancelled || run.State == provision.RunExpired {
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, run.State, nil)
+ }
+ jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has already ended", c.id))
+ }
+}
+
+// handleAPIRunAck is POST …/runs/{id}/ack (run spec §5.6, §5.8, codex 5):
+// refused unless the result exists and last_seq is the result's. Then the
+// record's run is cleared and the run's objects are deleted, in that order,
+// and only for a run the record does not hold in a live state.
+func (s *Server) handleAPIRunAck(w http.ResponseWriter, r *http.Request) {
+ c, ok := s.runGate(w, r)
+ if !ok {
+ return
+ }
+ var ack runs.Ack
+ if err := json.NewDecoder(http.MaxBytesReader(w, r.Body, 4<<10)).Decode(&ack); err != nil {
+ jsonError(w, http.StatusBadRequest, runMalformedMsg)
+ return
+ }
+ ctx, cancel := runWriteCtx(r.Context())
+ defer cancel()
+ found, err := s.requestOnHost(ctx, c)
+ switch {
+ case err != nil:
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ case !found:
+ jsonError(w, http.StatusNotFound, noRunMsg(c.host, c.id))
+ return
+ }
+ res, err := runs.ReadResult(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id)
+ switch {
+ case err != nil:
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ case res == nil:
+ jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has not finished yet", c.id))
+ return
+ case res.LastSeq != ack.LastSeq:
+ jsonError(w, http.StatusConflict, fmt.Sprintf("run %s ended at chunk %d, not %d", c.id, res.LastSeq, ack.LastSeq))
+ return
+ }
+ run, err := s.runOf(ctx, c)
+ switch {
+ case err != nil:
+ jsonError(w, http.StatusBadGateway, runReadFailedMsg)
+ return
+ case run.Open():
+ // The box has written the result and not yet moved the run to done.
+ // Its objects are the box's until it has.
+ jsonError(w, http.StatusConflict, fmt.Sprintf("run %s has not finished yet", c.id))
+ return
+ }
+ s.auditRunEnd(ctx, c.acct, c.app, c.host, c.id, "result", res)
+ if run != nil {
+ // Clear the record first. If the CAS fails, or its outcome is unknown,
+ // the record is asked again before anything is deleted: a run the
+ // record still names keeps its objects, and the caller may ack again.
+ _, cerr := provision.SetRecordRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.host, provision.RunClear(c.id))
+ still, rerr := s.runOf(ctx, c)
+ if rerr != nil || still != nil {
+ log.Printf("console: api run %s/%s/%s %s: clear after ack: %v; record read: %v", c.acct.Sqid, c.app, c.host, c.id, cerr, rerr)
+ jsonError(w, http.StatusBadGateway, runRecordFailedMsg)
+ return
+ }
+ }
+ if err := runs.DeleteRun(ctx, s.cfg.Releases, c.acct.Sqid, c.app, c.id); err != nil {
+ // The sweep deletes what is left (Task 5); the ack itself stands.
+ log.Printf("console: api run %s/%s/%s %s: delete after ack: %v", c.acct.Sqid, c.app, c.host, c.id, err)
+ }
+ writeJSON(w, http.StatusOK, runs.StateAnswer{State: "acknowledged"})
+}
+
+// auditRunEnd writes the run-end row exactly once per run (run spec §6):
+// whoever creates runs/<id>/audited first writes it, from the CLI's poll,
+// a cancel, an expiry or the sweep. The actor is the run's caller. It runs
+// to its end even if the caller has gone, so a cancelled request cannot
+// leave the marker without the row's chance.
+func (s *Server) auditRunEnd(ctx context.Context, acct *Account, app, host, id, outcome string, res *runs.Result) {
+ ctx, cancel := runWriteCtx(ctx)
+ defer cancel()
+ if _, err := s.cfg.Releases.Put(ctx, runs.AuditedKey(acct.Sqid, app, id), []byte(outcome), ""); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ log.Printf("console: api run %s/%s/%s %s: audit marker: %v", acct.Sqid, app, host, id, err)
+ }
+ return // written already, or the bucket refused: either way not twice
+ }
+ by := ""
+ if req, err := runs.ReadRequest(ctx, s.cfg.Releases, acct.Sqid, app, id); err == nil {
+ by = req.By
+ }
+ d := struct {
+ ID string `json:"id"`
+ Outcome string `json:"outcome"`
+ Result *runs.Result `json:"result,omitempty"`
+ }{id, outcome, res}
+ raw, _ := json.Marshal(d)
+ s.auditActivity(acct, app, by, Activity{Action: "run-end", TargetType: "instance", Target: host, Detail: string(raw)})
+}
diff --git a/internal/console/apirunsfollow_test.go b/internal/console/apirunsfollow_test.go
new file mode 100644
index 000000000..921fded45
--- /dev/null
+++ b/internal/console/apirunsfollow_test.go
@@ -0,0 +1,333 @@
+package console
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "net/http"
+ "strings"
+ "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"
+)
+
+// startRun admits one run through the door and returns its id.
+func (fx *runFx) startRun(t *testing.T) string {
+ t.Helper()
+ status, body := postRun(t, fx.runsURL(runTestHost), fx.member, okMeta(), nil)
+ if status != http.StatusAccepted {
+ t.Fatalf("create: %d %s", status, body)
+ }
+ var c runs.Created
+ if err := json.Unmarshal([]byte(body), &c); err != nil {
+ t.Fatal(err)
+ }
+ return c.ID
+}
+
+func (fx *runFx) transition(t *testing.T, mut func(*provision.Record) error) {
+ t.Helper()
+ if _, err := provision.SetRecordRun(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost, mut); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (fx *runFx) chunk(t *testing.T, id string, seq int64, stream, data string) {
+ t.Helper()
+ if _, err := fx.rel.Put(context.Background(), runs.ChunkKey(release.OpsAccount, "titogo", id, seq, stream), []byte(data), ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (fx *runFx) page(t *testing.T, id string, after int64) runs.Page {
+ t.Helper()
+ status, _, body := rawReq(t, http.MethodGet, fmt.Sprintf("%s/%s?after=%d", fx.runsURL(runTestHost), id, after), fx.member, "")
+ if status != http.StatusOK {
+ t.Fatalf("GET page: %d %s", status, body)
+ }
+ var p runs.Page
+ if err := json.Unmarshal([]byte(body), &p); err != nil {
+ t.Fatal(err)
+ }
+ return p
+}
+
+func (fx *runFx) post(t *testing.T, id, verb, body string) (int, string) {
+ t.Helper()
+ status, _, out := rawReq(t, http.MethodPost, fx.runsURL(runTestHost)+"/"+id+"/"+verb, fx.member, body)
+ return status, out
+}
+
+func TestAPIRunGetPagesChunksAndTheResult(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ fx.chunk(t, id, 1, runs.StreamOut, "hello ")
+ fx.chunk(t, id, 2, runs.StreamErr, "warn")
+ fx.chunk(t, id, 3, runs.StreamOut, "world")
+ p := fx.page(t, id, 0)
+ if p.State != provision.RunPicked || len(p.Chunks) != 3 || p.Result != nil || string(p.Chunks[1].Data) != "warn" || p.Chunks[1].Stream != runs.StreamErr {
+ t.Fatalf("page %+v", p)
+ }
+ if p := fx.page(t, id, 2); len(p.Chunks) != 1 || p.Chunks[0].Seq != 3 {
+ t.Fatalf("after 2: %+v", p)
+ }
+}
+
+func TestAPIRunGetPagesSeveralPagesAfterTheResult(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ for seq := int64(1); seq <= 3*runs.PageChunks; seq++ {
+ fx.chunk(t, id, seq, runs.StreamOut, "x")
+ }
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 0, LastSeq: 3 * runs.PageChunks}); err != nil {
+ t.Fatal(err)
+ }
+ var after int64
+ pages := 0
+ for {
+ p := fx.page(t, id, after)
+ pages++
+ for _, c := range p.Chunks {
+ if c.Seq != after+1 {
+ t.Fatalf("seq %d after %d", c.Seq, after)
+ }
+ after = c.Seq
+ }
+ if p.Result == nil {
+ t.Fatal("every page must carry the result once it exists")
+ }
+ if !p.More {
+ break
+ }
+ }
+ if after != 3*runs.PageChunks || pages != 3 {
+ t.Fatalf("read to %d in %d pages", after, pages)
+ }
+}
+
+func TestAPIRunGetReadsTheResultBeforeTheChunks(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ fx.chunk(t, id, 1, runs.StreamOut, "a")
+ // The box writes the last chunk, then the result. A store that shows
+ // the result to the console's read must also show every chunk to the
+ // listing after it.
+ order := []string{}
+ fx.srv.cfg.Releases = &orderedStore{Store: fx.rel, seen: &order}
+ fx.page(t, id, 0)
+ result, list := -1, -1
+ for i, call := range order {
+ if result < 0 && strings.HasSuffix(call, "/result.json") {
+ result = i
+ }
+ if list < 0 && strings.HasPrefix(call, "list ") && strings.HasSuffix(call, "/out/") {
+ list = i
+ }
+ }
+ if result < 0 || list < 0 || result > list {
+ t.Fatalf("store calls %q: the result must be read before the chunks are listed", order)
+ }
+}
+
+// orderedStore records Get and List calls in order.
+type orderedStore struct {
+ release.Store
+ seen *[]string
+}
+
+func (o *orderedStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ *o.seen = append(*o.seen, "get "+key)
+ return o.Store.Get(ctx, key)
+}
+func (o *orderedStore) List(ctx context.Context, prefix string) ([]string, error) {
+ *o.seen = append(*o.seen, "list "+prefix)
+ return o.Store.List(ctx, prefix)
+}
+
+func TestAPIRunAckChecksLastSeqThenCleansUp(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has not finished yet" {
+ t.Fatalf("ack before the result: %d %s", status, body)
+ }
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ fx.chunk(t, id, 1, runs.StreamOut, "a")
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
+ t.Fatal(err)
+ }
+ fx.transition(t, provision.RunFinish(id))
+ if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" ended at chunk 1, not 0" {
+ t.Fatalf("wrong last_seq: %d %s", status, body)
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
+ t.Fatal("a refused ack deleted the run")
+ }
+ if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusOK {
+ t.Fatalf("ack: %d %s", status, body)
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) != 0 {
+ t.Fatalf("ack left %q", left)
+ }
+ if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost); rec.ActiveRun != nil {
+ t.Fatalf("ack left the run on the record: %+v", rec.ActiveRun)
+ }
+}
+
+func TestAPIRunCancelBeforeAndAfterPickup(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ if status, body := fx.post(t, id, "cancel", ""); status != http.StatusOK || !strings.Contains(body, `"cancelled"`) {
+ t.Fatalf("queued cancel: %d %s", status, body)
+ }
+ if status, body := fx.post(t, id, "cancel", ""); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has already ended" {
+ t.Fatalf("cancel a cancelled run: %d %s", status, body)
+ }
+ id2 := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id2, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ if status, body := fx.post(t, id2, "cancel", ""); status != http.StatusOK || !strings.Contains(body, `"cancel"`) {
+ t.Fatalf("picked cancel: %d %s", status, body)
+ }
+ events, _ := fx.st.Events(OpsAccountID, "titogo", 20)
+ ends := 0
+ for _, e := range events {
+ if e.Action == "run-end" && strings.Contains(e.Detail, id) && strings.Contains(e.Detail, `"cancelled"`) {
+ ends++
+ }
+ }
+ if ends != 1 {
+ t.Errorf("a run cancelled before pickup must write one run-end row, got %d", ends)
+ }
+}
+
+func TestAPIRunExpireRacesAPickup(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ if status, body := fx.post(t, id, "expire", ""); status != http.StatusOK || !strings.Contains(body, `"expired"`) {
+ t.Fatalf("expire: %d %s", status, body)
+ }
+ id2 := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id2, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ if status, body := fx.post(t, id2, "expire", ""); status != http.StatusConflict || errorOf(t, body) != "run "+id2+" has already started" {
+ t.Fatalf("expire after pickup: %d %s", status, body)
+ }
+}
+
+func TestAPIRunRoutesNameAMissingRun(t *testing.T) {
+ fx := runFixture(t)
+ missing := "run_bbbbbbbbbbbbbbbbbbbbbbbbbb"
+ status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/"+missing, fx.member, "")
+ if status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run "+missing {
+ t.Fatalf("GET: %d %s", status, body)
+ }
+ for _, verb := range []string{"cancel", "expire", "ack"} {
+ if status, body := fx.post(t, missing, verb, `{"last_seq":0}`); status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run "+missing {
+ t.Errorf("%s: %d %s", verb, status, body)
+ }
+ }
+}
+
+func TestAPIRunGetWritesRunEndOnce(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ 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, "titogo", id, runs.Result{Exit: 7, Box: "box-1"}); err != nil {
+ t.Fatal(err)
+ }
+ fx.page(t, id, 0)
+ fx.page(t, id, 0)
+ events, _ := fx.st.Events(OpsAccountID, "titogo", 20)
+ n := 0
+ for _, e := range events {
+ if e.Action == "run-end" && strings.Contains(e.Detail, id) {
+ n++
+ if !strings.Contains(e.Detail, `"exit":7`) || e.Address != "alice@keymail.dev" {
+ t.Errorf("run-end %+v", e)
+ }
+ }
+ }
+ if n != 1 {
+ t.Fatalf("run-end rows %d, want 1", n)
+ }
+}
+
+func TestAPIRunAckKeepsTheObjectsOfALiveRun(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ fx.transition(t, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ fx.chunk(t, id, 1, runs.StreamOut, "a")
+ // The box has written the result and not yet moved the run to done.
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 0, LastSeq: 1}); err != nil {
+ t.Fatal(err)
+ }
+ if status, body := fx.post(t, id, "ack", `{"last_seq":1}`); status != http.StatusConflict || errorOf(t, body) != "run "+id+" has not finished yet" {
+ t.Fatalf("ack of a picked run: %d %s", status, body)
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
+ t.Fatal("an ack deleted the objects of a run the record holds as picked")
+ }
+ if rec, _ := provision.GetRecord(context.Background(), fx.rel, release.OpsAccount, "titogo", runTestHost); rec.ActiveRun == nil || rec.ActiveRun.State != provision.RunPicked {
+ t.Fatalf("record %+v", rec.ActiveRun)
+ }
+}
+
+func TestAPIRunAckAfterTheRecordMovedOnDeletesTheObjects(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ 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, "titogo", id, runs.Result{Exit: 0}); err != nil {
+ t.Fatal(err)
+ }
+ fx.transition(t, provision.RunFinish(id))
+ fx.transition(t, provision.RunClear(id))
+ // A repeated ack, the record already clear: the objects go, the answer stands.
+ if status, body := fx.post(t, id, "ack", `{"last_seq":0}`); status != http.StatusOK {
+ t.Fatalf("ack: %d %s", status, body)
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) != 0 {
+ t.Fatalf("left %q", left)
+ }
+}
+
+func TestAPIRunRoutesRefuseAnotherHostsRun(t *testing.T) {
+ fx := runFixture(t)
+ id := fx.startRun(t)
+ other := "other.go.tito.io"
+ putDBRecord(t, fx.consoleFixture, provision.Record{Host: other, Channel: "stable", CreatedAt: time.Now().UTC()})
+ if err := runs.WriteResult(context.Background(), fx.rel, release.OpsAccount, "titogo", id, runs.Result{Exit: 0}); err != nil {
+ t.Fatal(err)
+ }
+ want := other + " has no run " + id
+ status, _, body := rawReq(t, http.MethodGet, fx.runsURL(other)+"/"+id, fx.member, "")
+ if status != http.StatusNotFound || errorOf(t, body) != want {
+ t.Errorf("GET: %d %s", status, body)
+ }
+ for _, verb := range []string{"cancel", "expire", "ack"} {
+ _, _, body := rawReq(t, http.MethodPost, fx.runsURL(other)+"/"+id+"/"+verb, fx.member, `{"last_seq":0}`)
+ if errorOf(t, body) != want {
+ t.Errorf("%s: %s", verb, body)
+ }
+ }
+ if left, _ := fx.rel.List(context.Background(), runs.Prefix(release.OpsAccount, "titogo", id)); len(left) == 0 {
+ t.Fatal("another host's ack deleted the run")
+ }
+}
+
+func TestAPIRunRoutesRefuseAServiceCredentialAndABadID(t *testing.T) {
+ fx := runFixture(t)
+ svc := serviceToken(t, fx.st, OpsAccountID, "ci", "titogo", svcrole.Instance)
+ id := "run_aaaaaaaaaaaaaaaaaaaaaaaaaa"
+ if status, _, _ := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/"+id, svc, ""); status != http.StatusForbidden {
+ t.Errorf("service credential: %d, want 403", status)
+ }
+ status, _, body := rawReq(t, http.MethodGet, fx.runsURL(runTestHost)+"/not-a-run", fx.member, "")
+ if status != http.StatusNotFound || errorOf(t, body) != runTestHost+" has no run not-a-run" {
+ t.Errorf("bad id: %d %s", status, body)
+ }
+}
diff --git a/internal/console/authcli.go b/internal/console/authcli.go
index 93b2c0067..1015b357c 100644
--- a/internal/console/authcli.go
+++ b/internal/console/authcli.go
@@ -690,6 +690,10 @@ func (s *Server) registerCLIRoutes() {
}
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)
+ s.cliRoute("GET /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}", svcrole.Set{}, s.handleAPIRunGet)
+ s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel", svcrole.Set{}, s.handleAPIRunCancel)
+ s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/expire", svcrole.Set{}, s.handleAPIRunExpire)
+ s.cliRoute("POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/ack", svcrole.Set{}, s.handleAPIRunAck)
// 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 f497d1370..fad8d15f4 100644
--- a/internal/console/claim_invariant_fixtures_test.go
+++ b/internal/console/claim_invariant_fixtures_test.go
@@ -1156,6 +1156,30 @@ var claimFixtures = map[string]claimFixture{
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.go.tito.io/runs/run_aaaaaaaaaaaaaaaaaaaaaaaaaa", ""),
+ status: http.StatusConflict,
+ msg: claimRunUnsupportedMsg,
+ }},
+ "route POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel": {refused: &claimRefusalFixture{
+ apps: claimRefs("bab/titogo"),
+ do: claimRefusedAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/acme.go.tito.io/runs/run_aaaaaaaaaaaaaaaaaaaaaaaaaa/cancel", `{}`),
+ status: http.StatusConflict,
+ msg: claimRunUnsupportedMsg,
+ }},
+ "route POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/expire": {refused: &claimRefusalFixture{
+ apps: claimRefs("bab/titogo"),
+ do: claimRefusedAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/acme.go.tito.io/runs/run_aaaaaaaaaaaaaaaaaaaaaaaaaa/expire", `{}`),
+ status: http.StatusConflict,
+ msg: claimRunUnsupportedMsg,
+ }},
+ "route POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/ack": {refused: &claimRefusalFixture{
+ apps: claimRefs("bab/titogo"),
+ do: claimRefusedAPI(http.MethodPost, "/api/cli/apps/bab/titogo/instances/acme.go.tito.io/runs/run_aaaaaaaaaaaaaaaaaaaaaaaaaa/ack", `{}`),
+ 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/svcroutes_test.go b/internal/console/svcroutes_test.go
index 60cb60665..3bda8da4a 100644
--- a/internal/console/svcroutes_test.go
+++ b/internal/console/svcroutes_test.go
@@ -189,6 +189,10 @@ var closedDoors = []string{
// `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",
+ "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",
+ "POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/ack",
// 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
@@ -902,8 +906,8 @@ func TestEveryClosedDoorRefusesAServiceCredential(t *testing.T) {
}
seen++
}
- 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)",
+ if seen != len(closedDoors) || seen != 47 {
+ t.Errorf("walked %d closed doors, want all %d (and 47 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 five run doors)",
seen, len(closedDoors))
}
}
diff --git a/internal/console/testdata/claim-authority-sites.txt b/internal/console/testdata/claim-authority-sites.txt
index bd112c37e..8313a1e89 100644
--- a/internal/console/testdata/claim-authority-sites.txt
+++ b/internal/console/testdata/claim-authority-sites.txt
@@ -28,6 +28,8 @@
../mail/aws.go|NewAWSProvisioner|sts.NewFromConfig(cfg)
apiruns.go|handleAPIRunCreate|rec.ActiveRun.Open()
apiruns.go|writeRunObjects|os.Open(f.path)
+apirunsfollow.go|handleAPIRunAck|run.Open()
+apirunsfollow.go|runTransitionRefused|run.Open()
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 0acd96388..c0a5abf5e 100644
--- a/internal/console/testdata/claim-routes.txt
+++ b/internal/console/testdata/claim-routes.txt
@@ -57,6 +57,7 @@ route GET /api/cli/apps/{account}/{app}/geo
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}/runs/{id}
route GET /api/cli/apps/{account}/{app}/ledgers
route GET /api/cli/apps/{account}/{app}/ledgers/{ledger}/head
route GET /api/cli/apps/{account}/{app}/logs
@@ -173,6 +174,9 @@ 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}/instances/{host}/runs/{id}/ack
+route POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/cancel
+route POST /api/cli/apps/{account}/{app}/instances/{host}/runs/{id}/expire
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