amadan

carlos / platform Public

host: the runs worker: pickup by CAS, agent-restart close-out, the boot pass (run spec §5.2, codex 3, R2-6)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude d963bc0625789effbb8f76573273397cb1b148b2 parent 623d89e
2 files changed, +527 −0
  • internal/host/runs.go +225 −0
  • internal/host/runs_test.go +302 −0

CI passed — run details

diff --git a/internal/host/runs.go b/internal/host/runs.go
new file mode 100644
index 000000000..418af4382
--- /dev/null
+++ b/internal/host/runs.go
@@ -0,0 +1,225 @@
+package host
+
+import (
+ "context"
+ "crypto/rand"
+ "encoding/hex"
+ "errors"
+ "log"
+ "os"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/runs"
+)
+
+// runLauncher is what the runs worker needs from the activator.
+type runLauncher interface {
+ RunCmd(ctx context.Context, rt registry.Route, spec activator.CmdSpec) (activator.CmdHandle, error)
+}
+
+var _ runLauncher = (*activator.Service)(nil)
+
+// runHandle is the worker's line to one executing run.
+type runHandle struct {
+ cancel chan struct{}
+ once sync.Once
+}
+
+func newRunHandle() *runHandle { return &runHandle{cancel: make(chan struct{})} }
+
+func (h *runHandle) requestCancel() { h.once.Do(func() { close(h.cancel) }) }
+
+// runsWorker is the box's half of carlos run (run spec §5.2). boot is this
+// agent process's own id: a run picked under another boot of this box lost
+// its unit when the edge stopped (BindsTo=carlos-edge.service).
+type runsWorker struct {
+ store release.Store
+ routes func(host string) (*registry.Route, error)
+ launch runLauncher
+ dataDir string
+ box string
+ boot string
+ now func() time.Time
+ exec func(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle)
+ // start runs a picked run's exec: on its own goroutine in the agent,
+ // inline in tests that count launches.
+ start func(func())
+
+ mu sync.Mutex
+ noted map[string]provision.Record // host -> the record reconcile read
+ active map[string]*runHandle // run id -> the run this agent is executing
+}
+
+func newRunsWorker(store release.Store, routes func(string) (*registry.Route, error), launch runLauncher, dataDir, box, boot string) *runsWorker {
+ w := &runsWorker{store: store, routes: routes, launch: launch, dataDir: dataDir, box: box, boot: boot,
+ now: time.Now, noted: map[string]provision.Record{}, active: map[string]*runHandle{},
+ start: func(f func()) { go f() }}
+ w.exec = w.execute
+ return w
+}
+
+// newBootID is a fresh id for this agent process.
+func newBootID() string {
+ var b [16]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ return time.Now().UTC().Format("20060102T150405.000000000")
+ }
+ return hex.EncodeToString(b[:])
+}
+
+// armedRuns is the worker the agent armed; nil until then (and on a box
+// with no activator, which cannot run anything).
+var armedRuns atomic.Pointer[runsWorker]
+
+// note is ReconcileConfig.RunSeen: remember, never act, never block.
+func (w *runsWorker) note(rec provision.Record) {
+ w.mu.Lock()
+ w.noted[rec.Host] = rec
+ w.mu.Unlock()
+}
+
+// pass acts on every record noted since the last pass.
+func (w *runsWorker) pass(ctx context.Context) {
+ w.mu.Lock()
+ noted := w.noted
+ w.noted = map[string]provision.Record{}
+ w.mu.Unlock()
+ for _, rec := range noted {
+ w.consider(ctx, rec)
+ }
+}
+
+// bootPass runs once at agent start (index item 2: reconcile's first pass
+// is a minute away, and the CLI expires a run at 60 s). It deletes every
+// .carlos-run directory first (run spec §5.8: no run is open on the box
+// yet), then reads every record once and acts on each run.
+func (w *runsWorker) bootPass(ctx context.Context) {
+ if err := runs.SweepRunDirs(w.dataDir); err != nil {
+ log.Printf("agent: runs: boot sweep of .carlos-run: %v", err)
+ }
+ recs, err := provision.ListAllRecords(ctx, w.store)
+ if err != nil {
+ log.Printf("agent: runs: boot pass: %v", err)
+ return
+ }
+ for _, rec := range recs {
+ if rec.ActiveRun != nil {
+ w.consider(ctx, rec)
+ }
+ }
+}
+
+// owned is the row this box runs the record's instance from, when it is
+// this box's to act on: provisioned for this record's account and app,
+// and one the activator manages.
+func (w *runsWorker) owned(rec provision.Record) (registry.Route, bool) {
+ rt, err := w.routes(rec.Host)
+ if err != nil || rt == nil || !rt.Provisioned || rt.Account != rec.Account || rt.App != rec.App || !activator.ActivatorManaged(*rt) {
+ return registry.Route{}, false
+ }
+ return *rt, true
+}
+
+// consider acts on one record's run (run spec §5.2).
+func (w *runsWorker) consider(ctx context.Context, rec provision.Record) {
+ r := rec.ActiveRun
+ if r == nil {
+ return
+ }
+ switch r.State {
+ case provision.RunQueued:
+ rt, ok := w.owned(rec)
+ if !ok || w.now().After(r.Deadline) {
+ return
+ }
+ w.pickup(ctx, rec, rt)
+ case provision.RunPicked, provision.RunCancel:
+ if r.Box != w.box {
+ return
+ }
+ w.mu.Lock()
+ h := w.active[r.ID]
+ w.mu.Unlock()
+ switch {
+ case h != nil && r.State == provision.RunCancel:
+ h.requestCancel()
+ case h == nil && r.BootID != w.boot:
+ w.closeInterrupted(ctx, rec)
+ }
+ }
+}
+
+// pickup takes a queued run with one CAS from queued to picked. Only the
+// box whose CAS lands launches; any other outcome (another box, a cancel,
+// an expiry) leaves the run alone.
+func (w *runsWorker) pickup(ctx context.Context, rec provision.Record, rt registry.Route) {
+ id := rec.ActiveRun.ID
+ req, err := runs.ReadRequest(ctx, w.store, rec.Account, rec.App, id)
+ if err != nil {
+ // The console writes request.json before the record names the run;
+ // without it there is nothing to run, and the expiry ends it.
+ log.Printf("agent: runs: %s %s: request: %v", rec.Host, id, err)
+ return
+ }
+ h := newRunHandle()
+ w.mu.Lock()
+ if w.active[id] != nil {
+ w.mu.Unlock()
+ return
+ }
+ w.active[id] = h
+ w.mu.Unlock()
+ got, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host,
+ provision.RunPick(id, w.box, w.boot, w.now().Add(req.Timeout())))
+ if err != nil {
+ w.forget(id)
+ if !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: pickup: %v", rec.Host, id, err)
+ }
+ return
+ }
+ log.Printf("agent: runs: %s: picked run %s", rec.Host, id)
+ w.start(func() { w.exec(context.Background(), *got, *req, rt, h) })
+}
+
+func (w *runsWorker) forget(id string) {
+ w.mu.Lock()
+ delete(w.active, id)
+ w.mu.Unlock()
+}
+
+// closeInterrupted ends a run this box picked under an earlier boot (run
+// spec §5.2): its unit went with the edge, so it is never executed again.
+// The result names the last chunk the old boot wrote.
+func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record) {
+ id := rec.ActiveRun.ID
+ last, err := runs.LastChunkSeq(ctx, w.store, rec.Account, rec.App, id)
+ if err != nil {
+ log.Printf("agent: runs: %s %s: list chunks: %v", rec.Host, id, err)
+ return
+ }
+ res := runs.Result{Interrupted: runs.InterruptedAgentRestart, Box: w.box, Ended: w.now().UTC(), LastSeq: last}
+ if err := runs.WriteResult(ctx, w.store, rec.Account, rec.App, id, res); err != nil {
+ log.Printf("agent: runs: %s %s: result: %v", rec.Host, id, err)
+ return
+ }
+ if _, err := provision.SetRecordRun(ctx, w.store, rec.Account, rec.App, rec.Host, provision.RunFinish(id)); err != nil &&
+ !errors.Is(err, provision.ErrRunConflict) && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: done: %v", rec.Host, id, err)
+ }
+ if err := runs.RemoveRunDir(provision.InstanceDir(w.dataDir, rec.Host), id); err != nil && !errors.Is(err, os.ErrNotExist) {
+ log.Printf("agent: runs: %s %s: remove run dir: %v", rec.Host, id, err)
+ }
+ log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id)
+}
+
+// execute runs one picked run; Task 5 gives it its body.
+func (w *runsWorker) execute(ctx context.Context, rec provision.Record, req runs.Request, rt registry.Route, h *runHandle) {
+ defer w.forget(req.ID)
+}
diff --git a/internal/host/runs_test.go b/internal/host/runs_test.go
new file mode 100644
index 000000000..2cb483c26
--- /dev/null
+++ b/internal/host/runs_test.go
@@ -0,0 +1,302 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "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"
+)
+
+const runHost = "acme.go.tito.io"
+
+func runRecord() provision.Record {
+ return provision.Record{Schema: 1, Host: runHost, Slug: "acme", App: "titogo", Account: "bab", Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
+}
+
+func runOwnedRoute(host string) func(string) (*registry.Route, error) {
+ return func(h string) (*registry.Route, error) {
+ if h != host {
+ return nil, nil
+ }
+ return &registry.Route{Host: host, Kind: "instance", Hibernate: true, Socket: "/run/x.sock", Provisioned: true, App: "titogo", Account: "bab"}, nil
+ }
+}
+
+func noRoutes(string) (*registry.Route, error) { return nil, nil }
+
+// queueRun writes a record whose run is queued, with its request.json.
+func queueRun(t *testing.T, s release.Store, deadline time.Time) (provision.Record, string) {
+ t.Helper()
+ rec := runRecord()
+ if err := provision.CreateRecord(context.Background(), s, rec); err != nil {
+ t.Fatal(err)
+ }
+ id := "run_abcdefghijklmnopqrstuvwxyz"
+ inc, _ := hostclaimIncarnation(rec)
+ req := runs.Request{Schema: 1, ID: id, Account: rec.Account, App: rec.App, Host: rec.Host, Incarnation: inc, Argv: []string{"import"}, TimeoutS: 60, RequestedAt: time.Now()}
+ putJSONObject(t, s, runs.RequestKey(rec.Account, rec.App, id), req)
+ got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunAdmit(provision.RunAdmission{ID: id, By: "alice@keymail.dev", Deadline: deadline}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return *got, id
+}
+
+type execLog struct {
+ mu sync.Mutex
+ ids []string
+}
+
+func (l *execLog) exec(_ context.Context, _ provision.Record, req runs.Request, _ registry.Route, _ *runHandle) {
+ l.mu.Lock()
+ l.ids = append(l.ids, req.ID)
+ l.mu.Unlock()
+}
+func (l *execLog) count() int { l.mu.Lock(); defer l.mu.Unlock(); return len(l.ids) }
+
+// testWorker execs inline, so a test counts launches as soon as consider
+// returns.
+func testWorker(s release.Store, routes func(string) (*registry.Route, error), box, boot string, log *execLog) *runsWorker {
+ w := newRunsWorker(s, routes, nil, "/data", box, boot)
+ w.exec = log.exec
+ w.start = func(f func()) { f() }
+ return w
+}
+
+func currentRun(t *testing.T, s release.Store) *provision.RunState {
+ t.Helper()
+ rec, err := provision.GetRecord(context.Background(), s, "bab", "titogo", runHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return rec.ActiveRun
+}
+
+func TestRunsWorkerPicksOnlyRunsOnItsOwnRows(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, s, time.Now().Add(time.Minute))
+ var other, mine execLog
+ testWorker(s, noRoutes, "box-2", "boot-2", &other).consider(context.Background(), rec)
+ if other.count() != 0 || currentRun(t, s).State != provision.RunQueued {
+ t.Fatal("a box that does not own the row acted on the run")
+ }
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &mine).consider(context.Background(), rec)
+ r := currentRun(t, s)
+ if mine.count() != 1 || r.State != provision.RunPicked || r.Box != "box-1" || r.BootID != "boot-1" || r.ID != id {
+ t.Fatalf("pickup: execs %d, run %+v", mine.count(), r)
+ }
+ if d := time.Until(r.Deadline); d < 50*time.Second || d > 61*time.Second {
+ t.Errorf("the picked deadline is the run's timeout (60 s), got %s", d)
+ }
+}
+
+func TestRunsWorkerPickupRaceLaunchesOnce(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
+ var a, b execLog
+ wa := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &a)
+ wb := testWorker(s, runOwnedRoute(runHost), "box-2", "boot-2", &b)
+ var wg sync.WaitGroup
+ for _, w := range []*runsWorker{wa, wb} {
+ wg.Add(1)
+ go func() { defer wg.Done(); w.consider(context.Background(), rec) }()
+ }
+ wg.Wait()
+ if a.count()+b.count() != 1 {
+ t.Fatalf("launched %d times, want once", a.count()+b.count())
+ }
+}
+
+func TestRunsWorkerLeavesARunPastItsPickupDeadline(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, s, time.Now().Add(-time.Second))
+ var l execLog
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
+ if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
+ t.Fatal("a box picked a run past its pickup deadline: that run is the expiry's")
+ }
+}
+
+func TestRunsWorkerNeverPicksACancelledOrExpiredRun(t *testing.T) {
+ for _, mut := range []func(string) func(*provision.Record) error{provision.RunCancelTransition, provision.RunExpire} {
+ s := release.NewDirStore(t.TempDir())
+ rec, id := queueRun(t, s, time.Now().Add(time.Minute))
+ if _, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, mut(id)); err != nil {
+ t.Fatal(err)
+ }
+ var l execLog
+ // rec is the stale read the pass noted, still queued: the CAS refuses.
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), rec)
+ if l.count() != 0 {
+ t.Fatal("a cancelled or expired run was launched (codex R2-6)")
+ }
+ }
+}
+
+func TestRunsWorkerClosesOutARunPickedUnderAnEarlierBoot(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ _, id := queueRun(t, s, time.Now().Add(time.Minute))
+ got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-1", "boot-old", time.Now().Add(time.Hour)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(context.Background(), runs.ChunkKey("bab", "titogo", id, 4, runs.StreamOut), []byte("x"), ""); err != nil {
+ t.Fatal(err)
+ }
+ var l execLog
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
+ if l.count() != 0 {
+ t.Fatal("a run picked before an agent restart was run again")
+ }
+ res, err := runs.ReadResult(context.Background(), s, "bab", "titogo", id)
+ if err != nil || res == nil || res.Interrupted != runs.InterruptedAgentRestart || res.LastSeq != 4 || res.Box != "box-1" {
+ t.Fatalf("result %+v, %v", res, err)
+ }
+ if r := currentRun(t, s); r.State != provision.RunDone {
+ t.Fatalf("run %+v, want done", r)
+ }
+}
+
+func TestRunsWorkerForwardsACancelToItsRun(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ _, id := queueRun(t, s, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ h := newRunHandle()
+ w.active[id] = h
+ got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got, err = provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunCancelTransition(id)); err != nil {
+ t.Fatal(err)
+ }
+ w.consider(context.Background(), *got)
+ select {
+ case <-h.cancel:
+ default:
+ t.Fatal("the cancel did not reach the run")
+ }
+}
+
+func TestBootPassPicksARunQueuedBeforeStart(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ queueRun(t, s, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.dataDir = t.TempDir()
+ w.bootPass(context.Background())
+ if l.count() != 1 {
+ t.Fatal("the boot pass did not pick a run queued before agent start")
+ }
+}
+
+func TestRunsWorkerNotesAndPasses(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.note(rec)
+ w.note(rec)
+ w.pass(context.Background())
+ w.pass(context.Background())
+ if l.count() != 1 {
+ t.Fatalf("execs %d, want 1: a noted record is acted on once", l.count())
+ }
+}
+
+func TestRunDirNamesAgree(t *testing.T) {
+ if runs.DirName != activator.CmdRunDirName {
+ t.Fatalf("runs.DirName %q, activator.CmdRunDirName %q", runs.DirName, activator.CmdRunDirName)
+ }
+}
+
+func hostclaimIncarnation(rec provision.Record) (string, error) {
+ return hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+}
+
+func putJSONObject(t *testing.T, s release.Store, key string, v any) {
+ t.Helper()
+ raw, err := json.Marshal(v)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(context.Background(), key, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// A run another box picked is that box's: this box neither closes it out
+// nor forwards anything to it, whatever boot picked it.
+func TestRunsWorkerLeavesARunAnotherBoxPicked(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ _, id := queueRun(t, s, time.Now().Add(time.Minute))
+ got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-2", "boot-old", time.Now().Add(time.Hour)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l execLog
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-new", &l).consider(context.Background(), *got)
+ if res, err := runs.ReadResult(context.Background(), s, "bab", "titogo", id); err != nil || res != nil {
+ t.Fatalf("a box wrote a result for a run another box picked: %+v, %v", res, err)
+ }
+ if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked || r.Box != "box-2" {
+ t.Fatalf("execs %d, run %+v", l.count(), r)
+ }
+}
+
+// A run this box picked under its own boot is not an agent restart, even
+// when the worker holds no handle for it (a stale record noted after the
+// run's goroutine let go).
+func TestRunsWorkerLeavesItsOwnRunUnderThisBoot(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ _, id := queueRun(t, s, time.Now().Add(time.Minute))
+ got, err := provision.SetRecordRun(context.Background(), s, "bab", "titogo", runHost, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(time.Hour)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l execLog
+ testWorker(s, runOwnedRoute(runHost), "box-1", "boot-1", &l).consider(context.Background(), *got)
+ if res, err := runs.ReadResult(context.Background(), s, "bab", "titogo", id); err != nil || res != nil {
+ t.Fatalf("a run of this boot was closed out as interrupted: %+v, %v", res, err)
+ }
+ if r := currentRun(t, s); l.count() != 0 || r.State != provision.RunPicked {
+ t.Fatalf("execs %d, run %+v", l.count(), r)
+ }
+}
+
+// Only a provisioned row for the record's account and app that the
+// activator manages is this box's to pick for.
+func TestRunsWorkerPicksOnlyForActivatorManagedRowsOfTheRecord(t *testing.T) {
+ for name, mut := range map[string]func(*registry.Route){
+ "unprovisioned": func(rt *registry.Route) { rt.Provisioned = false },
+ "another account": func(rt *registry.Route) { rt.Account = "other" },
+ "another app": func(rt *registry.Route) { rt.App = "other" },
+ "not hibernating": func(rt *registry.Route) { rt.Hibernate = false },
+ "unit-backed": func(rt *registry.Route) { rt.Backing = "unit" },
+ "not an instance row": func(rt *registry.Route) { rt.Kind = "service" },
+ } {
+ s := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, s, time.Now().Add(time.Minute))
+ routes := func(h string) (*registry.Route, error) {
+ rt, err := runOwnedRoute(runHost)(h)
+ if rt != nil {
+ mut(rt)
+ }
+ return rt, err
+ }
+ var l execLog
+ testWorker(s, routes, "box-1", "boot-1", &l).consider(context.Background(), rec)
+ if l.count() != 0 || currentRun(t, s).State != provision.RunQueued {
+ t.Errorf("%s: the box picked a run for a row it does not run", name)
+ }
+ }
+}