provision: the run on the instance record, serialized with moves and imports (run spec §5.1, codex 1, R1-1, R2-2, R2-6)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+441
−1
approved_run_copy_test.go+1 −1internal/provision/provision.go+20 −0internal/provision/run.go+211 −0internal/provision/run_test.go+209 −0
diff --git a/approved_run_copy_test.go b/approved_run_copy_test.go| index 5a534e988..86006821f 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"} |
| +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"} |
| func mustParseRunRecord(t *testing.T) []copyRow { |
| t.Helper() |
diff --git a/internal/provision/provision.go b/internal/provision/provision.go| index 71345a85a..95eaa79db 100644 |
| --- a/internal/provision/provision.go |
| +++ b/internal/provision/provision.go |
| @@ -65,6 +65,20 @@ type Record struct { |
| // absent on every record at rest on main. |
| LastMove *MoveOutcome `json:"last_move,omitempty"` |
| + // ActiveRun is the instance's `carlos run`, if any (run spec §5.1): |
| + // queued, picked or cancelling while open, then done, cancelled or |
| + // expired until the console clears it. Runs, moves and imports all |
| + // change this one object, so they serialize on its ETag. Ordinary |
| + // writers carry it through unchanged; only the Run* transitions |
| + // (run.go) write it. Additive and omitempty. |
| + ActiveRun *RunState `json:"run,omitempty"` |
| + |
| + // Seed is the record's seed provenance (seeded-instances spec §3.2). |
| + // This series reads Kind and State only, to refuse runs on an import |
| + // that is not committed (run spec §5.1 step 9); the seeded-instances |
| + // plan adds the rest of SeedSource. Additive and omitempty. |
| + Seed *SeedSource `json:"seed,omitempty"` |
| + |
| // --- shape fields (2026-08-10 route-records design §3) -------------- |
| // Every one is omitempty with a zero value that reproduces the mint |
| // reconcile hardcoded before they existed: a record written by an older |
| @@ -1145,6 +1159,9 @@ func SetRecordMove(ctx context.Context, s release.Store, acct, app, host string, |
| // flight — the caller aborts first or waits. |
| func MoveRequest(to string, now time.Time) func(*Record) error { |
| return func(rec *Record) error { |
| + if rec.ActiveRun.Open() { |
| + return &MoveRunOpenError{Run: *rec.ActiveRun} |
| + } |
| if rec.Move != nil { |
| return fmt.Errorf("%w: a move to %s is already %s", ErrMoveConflict, rec.Move.To, rec.Move.State) |
| } |
| @@ -1176,6 +1193,9 @@ type MoveStart struct { |
| // without an id has no journal to run. |
| func MoveStartTransition(p MoveStart) func(*Record) error { |
| return func(rec *Record) error { |
| + if rec.ActiveRun.Open() { |
| + return &MoveRunOpenError{Run: *rec.ActiveRun} |
| + } |
| if rec.Move != nil { |
| return fmt.Errorf("%w: a move to %s is already %s", ErrMoveConflict, rec.Move.To, rec.Move.State) |
| } |
diff --git a/internal/provision/run.go b/internal/provision/run.go| new file mode 100644 |
| index 000000000..aa0518a1c |
| --- /dev/null |
| +++ b/internal/provision/run.go |
| @@ -0,0 +1,211 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// RunState is a `carlos run` on the instance record (run spec §5.1). Box |
| +// and BootID name the box and agent boot that picked it; Deadline is the |
| +// pickup deadline while queued and the run's own deadline once picked. |
| +// By is the caller, so a refusal can name who holds the instance (an |
| +// addition to the spec's struct, index item 3). |
| +type RunState struct { |
| + ID string `json:"id"` |
| + State string `json:"state"` |
| + By string `json:"by,omitempty"` |
| + Box string `json:"box,omitempty"` |
| + BootID string `json:"boot,omitempty"` |
| + Deadline time.Time `json:"deadline"` |
| +} |
| + |
| +// The run's states. queued, picked and cancel are open; the rest are |
| +// terminal, and a terminal run never blocks a run, a move or an import. |
| +const ( |
| + RunQueued = "queued" |
| + RunPicked = "picked" |
| + RunCancel = "cancel" |
| + RunDone = "done" |
| + RunCancelled = "cancelled" |
| + RunExpired = "expired" |
| +) |
| + |
| +// Open reports whether r holds the instance: queued, picked or cancel. |
| +func (r *RunState) Open() bool { |
| + return r != nil && (r.State == RunQueued || r.State == RunPicked || r.State == RunCancel) |
| +} |
| + |
| +// SeedSource is the part of the seeded-instances spec's SeedSource that |
| +// runs read (its §3.2). The seeded-instances plan extends this type; Kind |
| +// and State keep these json tags. |
| +type SeedSource struct { |
| + Kind string `json:"kind"` |
| + State string `json:"state"` |
| +} |
| + |
| +const ( |
| + SeedKindImport = "import" |
| + SeedCommitted = "committed" |
| +) |
| + |
| +// ImportUncommitted reports whether the record's seed is an import that is |
| +// not committed yet (seeded spec §5.4): a run is refused then, at the door |
| +// and again at launch (run spec §5.1 step 9, codex R2-2). |
| +func (r Record) ImportUncommitted() bool { |
| + return r.Seed != nil && r.Seed.Kind == SeedKindImport && r.Seed.State != SeedCommitted |
| +} |
| + |
| +var ( |
| + // ErrRunOpen: a run is open on the instance. A RunOpenError carries it. |
| + ErrRunOpen = errors.New("provision: a run is open on the instance") |
| + // ErrRunConflict: the record's run is not in the state the transition |
| + // needs (another box took it, it was cancelled or expired, or it is |
| + // another run). |
| + ErrRunConflict = errors.New("provision: run state conflict") |
| + // ErrRunMoving: a move is in flight. |
| + ErrRunMoving = errors.New("provision: the instance is moving") |
| + // ErrRunImport: the instance is an import that is not committed. |
| + ErrRunImport = errors.New("provision: the instance is an import that is not committed") |
| +) |
| + |
| +// RunOpenError is a refusal because Run holds the instance. |
| +type RunOpenError struct{ Run RunState } |
| + |
| +func (e *RunOpenError) Error() string { return ErrRunOpen.Error() + ": " + e.Run.ID } |
| +func (e *RunOpenError) Is(target error) bool { return target == ErrRunOpen } |
| + |
| +// MoveRunOpenError is a move transition's refusal while a run is open: a |
| +// move conflict, so both move doors answer 409. The legacy door answers |
| +// Error(), which carries Sentence (RUN-C43). |
| +type MoveRunOpenError struct{ Run RunState } |
| + |
| +func (e *MoveRunOpenError) Error() string { return ErrMoveConflict.Error() + ": " + e.Sentence() } |
| +func (e *MoveRunOpenError) Unwrap() error { return ErrMoveConflict } |
| +func (e *MoveRunOpenError) Sentence() string { |
| + return fmt.Sprintf("a run is open on this instance: %s. wait for it to finish or cancel it, then move the instance", e.Run.ID) |
| +} |
| + |
| +// SetRecordRun applies one run transition under the record's ETag, with |
| +// updateRecord's one retry: a transition re-checks its own state on the |
| +// fresh read, so a lost race refuses rather than overwrites. |
| +func SetRecordRun(ctx context.Context, s release.Store, acct, app, host string, mut func(*Record) error) (*Record, error) { |
| + return updateRecord(ctx, s, acct, app, host, mut) |
| +} |
| + |
| +// RunAdmission is what the console decided for a new run. |
| +type RunAdmission struct { |
| + ID, By string |
| + Deadline time.Time // the pickup deadline |
| +} |
| + |
| +// 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. |
| +func RunAdmit(p RunAdmission) func(*Record) error { |
| + return func(rec *Record) error { |
| + switch { |
| + case p.ID == "" || p.Deadline.IsZero(): |
| + return fmt.Errorf("%w: malformed admission", ErrRunConflict) |
| + case rec.Move != nil: |
| + return ErrRunMoving |
| + case rec.ActiveRun.Open(): |
| + return &RunOpenError{Run: *rec.ActiveRun} |
| + case rec.ImportUncommitted(): |
| + return ErrRunImport |
| + } |
| + rec.ActiveRun = &RunState{ID: p.ID, State: RunQueued, By: p.By, Deadline: p.Deadline.UTC()} |
| + return nil |
| + } |
| +} |
| + |
| +// runIn returns a copy of the record's run when it is id in one of states. |
| +func runIn(rec *Record, id string, states ...string) (RunState, error) { |
| + r := rec.ActiveRun |
| + if r == nil || r.ID != id { |
| + return RunState{}, ErrRunConflict |
| + } |
| + for _, s := range states { |
| + if r.State == s { |
| + return *r, nil |
| + } |
| + } |
| + return RunState{}, ErrRunConflict |
| +} |
| + |
| +// RunPick moves a queued run to picked by box under boot, with the run's |
| +// own deadline (run spec §5.2). Only the box whose CAS lands launches. |
| +func RunPick(id, box, boot string, deadline time.Time) func(*Record) error { |
| + return func(rec *Record) error { |
| + r, err := runIn(rec, id, RunQueued) |
| + if err != nil { |
| + return err |
| + } |
| + r.State, r.Box, r.BootID, r.Deadline = RunPicked, box, boot, deadline.UTC() |
| + rec.ActiveRun = &r |
| + return nil |
| + } |
| +} |
| + |
| +// RunCancelTransition cancels: queued becomes cancelled, which is terminal, |
| +// so no box ever launches it; picked becomes cancel, and the owning box |
| +// stops the unit (run spec §5.7). cancel stays cancel. |
| +func RunCancelTransition(id string) func(*Record) error { |
| + return func(rec *Record) error { |
| + r, err := runIn(rec, id, RunQueued, RunPicked, RunCancel) |
| + if err != nil { |
| + return err |
| + } |
| + if r.State == RunQueued { |
| + r.State = RunCancelled |
| + } else { |
| + r.State = RunCancel |
| + } |
| + rec.ActiveRun = &r |
| + return nil |
| + } |
| +} |
| + |
| +// RunExpire expires a queued run (run spec §5.2, §5.8): terminal, so a box |
| +// that turns up late can never pick it. A picked run is not expired: the |
| +// pickup won. |
| +func RunExpire(id string) func(*Record) error { |
| + return func(rec *Record) error { |
| + r, err := runIn(rec, id, RunQueued) |
| + if err != nil { |
| + return err |
| + } |
| + r.State = RunExpired |
| + rec.ActiveRun = &r |
| + return nil |
| + } |
| +} |
| + |
| +// RunFinish marks a picked or cancelling run done: the box after it wrote |
| +// result.json, or the console's sweep for a lost run. |
| +func RunFinish(id string) func(*Record) error { |
| + return func(rec *Record) error { |
| + r, err := runIn(rec, id, RunPicked, RunCancel) |
| + if err != nil { |
| + return err |
| + } |
| + r.State = RunDone |
| + rec.ActiveRun = &r |
| + return nil |
| + } |
| +} |
| + |
| +// RunClear removes a terminal run from the record, only while the record |
| +// still names this run (run spec §5.8). |
| +func RunClear(id string) func(*Record) error { |
| + return func(rec *Record) error { |
| + if _, err := runIn(rec, id, RunDone, RunCancelled, RunExpired); err != nil { |
| + return err |
| + } |
| + rec.ActiveRun = nil |
| + return nil |
| + } |
| +} |
diff --git a/internal/provision/run_test.go b/internal/provision/run_test.go| new file mode 100644 |
| index 000000000..419494e82 |
| --- /dev/null |
| +++ b/internal/provision/run_test.go |
| @@ -0,0 +1,209 @@ |
| +package provision |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "errors" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +func runTestRecord(t *testing.T, s release.Store) Record { |
| + t.Helper() |
| + rec := Record{Schema: 1, Host: "a.go.example.com", Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()} |
| + if err := CreateRecord(context.Background(), s, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + return rec |
| +} |
| + |
| +func setRun(t *testing.T, s release.Store, rec Record, mut func(*Record) error) (*Record, error) { |
| + t.Helper() |
| + return SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, mut) |
| +} |
| + |
| +func TestRunAdmitPickFinishClear(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + at := time.Now().Add(time.Minute) |
| + got, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", By: "alice@keymail.dev", Deadline: at})) |
| + if err != nil || got.ActiveRun.State != RunQueued || got.ActiveRun.By != "alice@keymail.dev" || !got.ActiveRun.Open() { |
| + t.Fatalf("admit: %+v, %v", got, err) |
| + } |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_b", Deadline: at})); !errors.Is(err, ErrRunOpen) { |
| + t.Fatalf("second admit: %v, want ErrRunOpen", err) |
| + } |
| + var open *RunOpenError |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_b", Deadline: at})); !errors.As(err, &open) || open.Run.ID != "run_a" || open.Run.By != "alice@keymail.dev" { |
| + t.Fatalf("the refusal must name the open run and its caller: %v", err) |
| + } |
| + if got, err = setRun(t, s, rec, RunPick("run_a", "box-1", "boot-1", at.Add(time.Hour))); err != nil || got.ActiveRun.State != RunPicked || got.ActiveRun.Box != "box-1" { |
| + t.Fatalf("pick: %+v, %v", got, err) |
| + } |
| + if _, err := setRun(t, s, rec, RunPick("run_a", "box-2", "boot-2", at)); !errors.Is(err, ErrRunConflict) { |
| + t.Fatalf("a second pick: %v, want ErrRunConflict", err) |
| + } |
| + if _, err := setRun(t, s, rec, RunClear("run_a")); !errors.Is(err, ErrRunConflict) { |
| + t.Fatalf("clearing an open run: %v, want ErrRunConflict", err) |
| + } |
| + if got, err = setRun(t, s, rec, RunFinish("run_a")); err != nil || got.ActiveRun.State != RunDone || got.ActiveRun.Open() { |
| + t.Fatalf("finish: %+v, %v", got, err) |
| + } |
| + if got, err = setRun(t, s, rec, RunClear("run_a")); err != nil || got.ActiveRun != nil { |
| + t.Fatalf("clear: %+v, %v", got, err) |
| + } |
| +} |
| + |
| +func TestRunCancelBeforeAndAfterPickup(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + at := time.Now().Add(time.Minute) |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", Deadline: at})); err != nil { |
| + t.Fatal(err) |
| + } |
| + got, err := setRun(t, s, rec, RunCancelTransition("run_a")) |
| + if err != nil || got.ActiveRun.State != RunCancelled || got.ActiveRun.Open() { |
| + t.Fatalf("queued cancel: %+v, %v", got, err) |
| + } |
| + if _, err := setRun(t, s, rec, RunPick("run_a", "box-1", "boot-1", at)); !errors.Is(err, ErrRunConflict) { |
| + t.Fatal("a cancelled run was picked (codex R2-6)") |
| + } |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_b", Deadline: at})); err != nil { |
| + t.Fatalf("a terminal run must not block the next: %v", err) |
| + } |
| + if _, err := setRun(t, s, rec, RunPick("run_b", "box-1", "boot-1", at)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, err = setRun(t, s, rec, RunCancelTransition("run_b")); err != nil || got.ActiveRun.State != RunCancel || !got.ActiveRun.Open() { |
| + t.Fatalf("picked cancel: %+v, %v", got, err) |
| + } |
| + if got, err = setRun(t, s, rec, RunCancelTransition("run_b")); err != nil || got.ActiveRun.State != RunCancel { |
| + t.Fatalf("a second cancel is idempotent: %+v, %v", got, err) |
| + } |
| + if got, err = setRun(t, s, rec, RunFinish("run_b")); err != nil || got.ActiveRun.State != RunDone { |
| + t.Fatalf("finish after cancel: %+v, %v", got, err) |
| + } |
| +} |
| + |
| +func TestRunExpireOnlyFromQueued(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + at := time.Now() |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", Deadline: at})); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got, err := setRun(t, s, rec, RunExpire("run_a")); err != nil || got.ActiveRun.State != RunExpired { |
| + t.Fatalf("expire: %+v, %v", got, err) |
| + } |
| + if _, err := setRun(t, s, rec, RunPick("run_a", "late-box", "boot", at)); !errors.Is(err, ErrRunConflict) { |
| + t.Fatal("an expired run was picked by a late box (codex R2-6)") |
| + } |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_b", Deadline: at})); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := setRun(t, s, rec, RunPick("run_b", "box-1", "boot-1", at)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := setRun(t, s, rec, RunExpire("run_b")); !errors.Is(err, ErrRunConflict) { |
| + t.Fatal("a picked run was expired: the late pickup must win") |
| + } |
| +} |
| + |
| +func TestRunAdmitRefusesAMoveAndAnUncommittedImport(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + at := time.Now() |
| + if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, MoveRequest("uk", at)); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", Deadline: at})); !errors.Is(err, ErrRunMoving) { |
| + t.Fatalf("admit while moving: %v, want ErrRunMoving", err) |
| + } |
| + s2 := release.NewDirStore(t.TempDir()) |
| + imp := runTestRecord(t, s2) |
| + for state, refused := range map[string]bool{"swapped": true, "placed": true, "committed": false} { |
| + if _, err := updateRecord(context.Background(), s2, imp.Account, imp.App, imp.Host, func(r *Record) error { |
| + r.Seed, r.ActiveRun = &SeedSource{Kind: SeedKindImport, State: state}, nil |
| + return nil |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + _, err := setRun(t, s2, imp, RunAdmit(RunAdmission{ID: "run_" + state, Deadline: at})) |
| + if refused != errors.Is(err, ErrRunImport) { |
| + t.Errorf("import %s: err %v, refused want %v (codex R2-2)", state, err, refused) |
| + } |
| + } |
| +} |
| + |
| +// A move request is refused while a run is open, in both move transitions, |
| +// and the refusal is a move conflict whose text is RUN-C43 (codex 1, R1-1). |
| +func TestMoveTransitionsRefuseWhileARunIsOpen(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", Deadline: time.Now()})); err != nil { |
| + t.Fatal(err) |
| + } |
| + inc := strings.Repeat("a", 64) |
| + for name, mut := range map[string]func(*Record) error{ |
| + "legacy": MoveRequest("uk", time.Now()), |
| + "phase 1": MoveStartTransition(MoveStart{ID: "mv_" + strings.Repeat("a", 26), Incarnation: inc, From: "ie", To: "uk", Cut: "auto", Now: time.Now()}), |
| + } { |
| + _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, mut) |
| + var mro *MoveRunOpenError |
| + if !errors.Is(err, ErrMoveConflict) || !errors.As(err, &mro) || !strings.Contains(err.Error(), "a run is open on this instance: run_a") { |
| + t.Errorf("%s: %v", name, err) |
| + } |
| + } |
| + if _, err := setRun(t, s, rec, RunExpire("run_a")); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := SetRecordMove(context.Background(), s, rec.Account, rec.App, rec.Host, MoveRequest("uk", time.Now())); err != nil { |
| + t.Fatalf("a terminal run must not block a move: %v", err) |
| + } |
| +} |
| + |
| +// Ordinary record writers carry the run through unchanged, as they carry |
| +// Maintenance (run spec §5.1). |
| +func TestOrdinaryWritersCarryActiveRun(t *testing.T) { |
| + s := release.NewDirStore(t.TempDir()) |
| + rec := runTestRecord(t, s) |
| + at := time.Date(2026, 10, 8, 12, 0, 0, 0, time.UTC) |
| + if _, err := setRun(t, s, rec, RunAdmit(RunAdmission{ID: "run_a", By: "alice@keymail.dev", Deadline: at})); err != nil { |
| + t.Fatal(err) |
| + } |
| + runJSON := func() string { |
| + 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"]) |
| + } |
| + want := runJSON() |
| + if !strings.Contains(want, `"state":"queued"`) { |
| + t.Fatalf("run = %s", want) |
| + } |
| + if err := SetRecordRegion(context.Background(), s, rec.Account, rec.App, rec.Host, "uk"); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := runJSON(); got != want { |
| + t.Fatalf("SetRecordRegion: run %s, want %s", got, want) |
| + } |
| + cur, etag, err := GetRecordETag(context.Background(), s, rec.Account, rec.App, rec.Host) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + cur.Channel = "beta" |
| + if err := UpdateRecord(context.Background(), s, *cur, etag); err != nil { |
| + t.Fatal(err) |
| + } |
| + if got := runJSON(); got != want { |
| + t.Fatalf("UpdateRecord: run %s, want %s", got, want) |
| + } |
| +} |