amadan

carlos / platform Public

host: arm the runs worker; reconcile notes runs, the passes act, a boot pass at start

The runs passes do nothing until the boot pass (which sweeps every
.carlos-run directory) has finished, so no run is launched into a
directory about to be swept. follow's timer is capped at the run's own
deadline, so slow staging does not buy a full extra timeout.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 809329f4f2ae9c2d75efd979abf1ae1dc7252fb9 parent b225ba1
5 files changed, +197 −2
  • internal/host/agent.go +10 −0
  • internal/host/runexec.go +7 −1
  • internal/host/runexec_linux_test.go +23 −0
  • internal/host/runs.go +50 −1
  • internal/host/runs_test.go +107 −0

CI passed — run details

diff --git a/internal/host/agent.go b/internal/host/agent.go
index c713732d2..533e2a307 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -408,6 +408,7 @@ func runAgent(args []string, buildClaims ClaimFactory) {
eb.Mirror = m
}
armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes)
+ armRunsPass(svc, reg, dd)
// The startup sweep (design §6.7): a leftover instance unit from a
// previous agent process, one this box's registry no longer holds
// a live wake for, is stopped before this process starts spawning
@@ -1144,6 +1145,9 @@ func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Reg
// Move holds (moves spec §9.6): participant rows minted held and kept.
// Unchanged with the Phase 1 switch off.
cfg = withMovesHolds(cfg)
+ // Runs ride this pass's record reads (run spec §5.2): the worker notes
+ // every in-scope record with a run, and acts after the pass.
+ cfg = withRunsHook(cfg)
return errors.Join(
provision.Reconcile(ctx, s, reg, cfg),
provision.ReconcileSidecars(ctx, s, reg, cfg),
@@ -2209,6 +2213,9 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
// runs before the first moves pass, and an unbuilt view freezes every
// host (moves spec §9.4). Nothing with the Phase 1 switch off.
refreshMovesHolds(ctx, store)
+ // carlos run's boot pass (run spec §5.2, index item 2): reconcile's
+ // first pass is a minute away and the CLI expires a run at 60 s.
+ runRunsBootPass(ctx)
// The schedule converge's own seed, on restartSeed's precedent and for the
// same shape of reason: the leg below runs only on a serial MOVE, so
// without one pass here a box that booted after the bump would carry no
@@ -2250,6 +2257,8 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
mark("telemetry")
runMovesPass(ctx, store)
mark("moves")
+ runRunsPass(ctx)
+ mark("runs")
// Folded once per pass and handed in, never re-read inside the
// loop: this leg runs on every serial bump (a seconds cadence),
// and the fold is a LoadBoxes plus a List plus one Get per edges
@@ -2823,6 +2832,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config, configure ..
}
telemetryPass(ctx, store, reg)
runMovesPass(ctx, store)
+ runRunsPass(ctx)
deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil {
log.Printf("agent: instances alias pass: %v", err)
diff --git a/internal/host/runexec.go b/internal/host/runexec.go
index d0c83a5f5..10f22c2b2 100644
--- a/internal/host/runexec.go
+++ b/internal/host/runexec.go
@@ -311,7 +311,13 @@ func (w *runsWorker) follow(ctx context.Context, rec provision.Record, req runs.
}
}
}()
- timeout := time.NewTimer(req.Timeout())
+ // The timeout is the run's own deadline, set at pickup: staging that
+ // ran long leaves the command less, not a full timeout from launch.
+ limit := req.Timeout()
+ if r := rec.ActiveRun; r != nil && !r.Deadline.IsZero() {
+ limit = min(limit, r.Deadline.Sub(w.now()))
+ }
+ timeout := time.NewTimer(limit)
defer timeout.Stop()
recheck := time.NewTicker(runRecheckEvery)
defer recheck.Stop()
diff --git a/internal/host/runexec_linux_test.go b/internal/host/runexec_linux_test.go
index 43d965c58..6749ce62c 100644
--- a/internal/host/runexec_linux_test.go
+++ b/internal/host/runexec_linux_test.go
@@ -252,6 +252,29 @@ func TestRunTimesOut(t *testing.T) {
}
}
+// The run's deadline is set at pickup, so slow staging does not buy a full
+// extra timeout: a run whose deadline is nearer than its timeout times out
+// at the deadline, or the console sweep marks a still-running run lost.
+func TestRunTimesOutAtItsDeadlineNotAFullTimeoutLater(t *testing.T) {
+ s := ctxStore{release.NewDirStore(t.TempDir())}
+ rec, id := queueRun(t, s, time.Now().Add(time.Minute))
+ req, err := runs.ReadRequest(context.Background(), s, rec.Account, rec.App, id)
+ if err != nil {
+ t.Fatal(err)
+ }
+ req.TimeoutS = 60
+ got, err := provision.SetRecordRun(context.Background(), s, rec.Account, rec.App, rec.Host, provision.RunPick(id, "box-1", "boot-1", time.Now().Add(1500*time.Millisecond)))
+ if err != nil {
+ t.Fatal(err)
+ }
+ h := &fakeHandle{polls: make(chan instancectl.CmdStatus)}
+ start := time.Now()
+ res := runWith(t, s, &fakeLauncher{handle: h, out: "slow\n"}, *got, *req, newRunHandle())
+ if !res.TimedOut || time.Since(start) > 10*time.Second || h.stops == 0 {
+ t.Fatalf("result %+v after %s, stops %d: a run times out at its deadline, not %ds after launch", res, time.Since(start), h.stops, req.TimeoutS)
+ }
+}
+
// Review Focus 2: a cancel after the pickup CAS and before the unit starts
// ends the run cancelled with no unit started.
func TestRunCancelledBeforeLaunchNeverStarts(t *testing.T) {
diff --git a/internal/host/runs.go b/internal/host/runs.go
index c906e77ce..e98ed4098 100644
--- a/internal/host/runs.go
+++ b/internal/host/runs.go
@@ -51,6 +51,10 @@ type runsWorker struct {
// start runs a picked run's exec: on its own goroutine in the agent,
// inline in tests that count launches.
start func(func())
+ // booted is set when the boot pass has finished. The boot pass deletes
+ // every .carlos-run directory, so until then pass does nothing: a run
+ // it launched would have its directory swept from under it.
+ booted atomic.Bool
mu sync.Mutex
noted map[string]provision.Record // host -> the record reconcile read
@@ -86,8 +90,12 @@ func (w *runsWorker) note(rec provision.Record) {
w.mu.Unlock()
}
-// pass acts on every record noted since the last pass.
+// pass acts on every record noted since the last pass, once the boot pass
+// has finished; before that the notes wait.
func (w *runsWorker) pass(ctx context.Context) {
+ if !w.booted.Load() {
+ return
+ }
w.mu.Lock()
noted := w.noted
w.noted = map[string]provision.Record{}
@@ -102,6 +110,9 @@ func (w *runsWorker) pass(ctx context.Context) {
// .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) {
+ // Opens pass, whether or not the records could be read: runs the boot
+ // pass did not see are the next passes' to pick.
+ defer w.booted.Store(true)
if err := runs.SweepRunDirs(w.dataDir); err != nil {
log.Printf("agent: runs: boot sweep of .carlos-run: %v", err)
}
@@ -289,3 +300,41 @@ func (w *runsWorker) closeInterrupted(ctx context.Context, rec provision.Record)
}
log.Printf("agent: runs: %s: run %s was picked before this agent started; ended as interrupted", rec.Host, id)
}
+
+// armRunsPass arms the runs worker on a box with an activator. Thin, like
+// armMovesPass: the behaviour is the worker's, tested there.
+func armRunsPass(svc *activator.Service, reg *registry.Registry, dataDir string) {
+ if svc == nil || reg == nil {
+ return
+ }
+ store, err := openDeployStore(context.Background())
+ if err != nil {
+ log.Printf("agent: runs disabled: %v", err)
+ return
+ }
+ w := newRunsWorker(store, reg.Get, svc, dataDir, BoxLabel(), newBootID())
+ armedRuns.Store(w)
+ log.Printf("agent: runs armed (box %s, boot %s)", w.box, w.boot)
+}
+
+// withRunsHook gives reconcile the worker's note (run spec §5.2).
+func withRunsHook(cfg provision.ReconcileConfig) provision.ReconcileConfig {
+ if w := armedRuns.Load(); w != nil && cfg.RunSeen == nil {
+ cfg.RunSeen = w.note
+ }
+ return cfg
+}
+
+// runRunsPass acts on what the last reconcile pass noted.
+func runRunsPass(ctx context.Context) {
+ if w := armedRuns.Load(); w != nil {
+ w.pass(ctx)
+ }
+}
+
+// runRunsBootPass is the worker's one pass at agent start.
+func runRunsBootPass(ctx context.Context) {
+ if w := armedRuns.Load(); w != nil {
+ w.bootPass(ctx)
+ }
+}
diff --git a/internal/host/runs_test.go b/internal/host/runs_test.go
index d6b0e8c9f..2282f0d2c 100644
--- a/internal/host/runs_test.go
+++ b/internal/host/runs_test.go
@@ -7,6 +7,7 @@ import (
"errors"
"log"
"os"
+ "path/filepath"
"strings"
"sync"
"testing"
@@ -73,6 +74,7 @@ func testWorker(s release.Store, routes func(string) (*registry.Route, error), b
w := newRunsWorker(s, routes, nil, "/data", box, boot)
w.exec = log.exec
w.start = func(f func()) { f() }
+ w.booted.Store(true) // a test of the boot gate unsets it
return w
}
@@ -411,3 +413,108 @@ func TestRunsWorkerForgetsAPickupThatDidNotLand(t *testing.T) {
t.Fatalf("execs %d, run %+v, active %d", l.count(), currentRun(t, dir), len(w.active))
}
}
+
+func TestRunsHookIsOnlyThereWhenArmed(t *testing.T) {
+ armedRuns.Store(nil)
+ if cfg := withRunsHook(provision.ReconcileConfig{}); cfg.RunSeen != nil {
+ t.Fatal("an unarmed box must not hook reconcile")
+ }
+ runRunsPass(context.Background()) // no-op, no panic
+ runRunsBootPass(context.Background()) // ditto
+ var l execLog
+ w := testWorker(release.NewDirStore(t.TempDir()), noRoutes, "box-1", "boot-1", &l)
+ armedRuns.Store(w)
+ t.Cleanup(func() { armedRuns.Store(nil) })
+ cfg := withRunsHook(provision.ReconcileConfig{})
+ if cfg.RunSeen == nil {
+ t.Fatal("an armed box hooks reconcile")
+ }
+ cfg.RunSeen(runRecord())
+ if len(w.noted) != 1 {
+ t.Fatal("the hook does not reach the worker")
+ }
+}
+
+// listGate holds every List until released, so a test can act while the
+// boot pass is part-way through.
+type listGate struct {
+ release.Store
+ entered chan struct{}
+ proceed chan struct{}
+ once sync.Once
+}
+
+func (g *listGate) List(ctx context.Context, prefix string) ([]string, error) {
+ g.once.Do(func() { close(g.entered) })
+ <-g.proceed
+ return g.Store.List(ctx, prefix)
+}
+
+// The boot pass deletes every .carlos-run directory (run spec §5.8), so no
+// reconcile-driven pass may pick or launch anything before it has finished:
+// a run it launched would have its directory swept from under it.
+func TestRunsPassDoesNothingUntilTheBootPassHasFinished(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
+ gate := &listGate{Store: dir, entered: make(chan struct{}), proceed: make(chan struct{})}
+ var l execLog
+ w := testWorker(gate, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.booted.Store(false)
+ w.dataDir = t.TempDir()
+ stale := filepath.Join(provision.InstanceDir(w.dataDir, runHost), runs.DirName, "run_old")
+ if err := os.MkdirAll(stale, 0o700); err != nil {
+ t.Fatal(err)
+ }
+ armedRuns.Store(w)
+ t.Cleanup(func() { armedRuns.Store(nil) })
+
+ // Reconcile ran before the boot pass: the hook notes, the pass waits.
+ withRunsHook(provision.ReconcileConfig{}).RunSeen(rec)
+ runRunsPass(context.Background())
+ if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
+ t.Fatal("a pass picked a run before the boot pass ran")
+ }
+
+ done := make(chan struct{})
+ go func() { runRunsBootPass(context.Background()); close(done) }()
+ <-gate.entered // the sweep is done, the boot pass is reading records
+ if _, err := os.Stat(stale); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the boot pass did not sweep first: %v", err)
+ }
+ runRunsPass(context.Background())
+ if l.count() != 0 || currentRun(t, dir).State != provision.RunQueued {
+ t.Fatal("a pass picked a run while the boot pass was still running")
+ }
+ close(gate.proceed)
+ <-done
+ if l.count() != 1 {
+ t.Fatalf("execs %d, want 1: the boot pass picks the run queued before start", l.count())
+ }
+}
+
+// A boot pass that cannot read the records still opens the gate: the runs
+// it could not see are the next passes' to pick.
+func TestRunsPassActsAfterABootPassThatFailed(t *testing.T) {
+ dir := release.NewDirStore(t.TempDir())
+ rec, _ := queueRun(t, dir, time.Now().Add(time.Minute))
+ var l execLog
+ w := testWorker(listFails{dir}, runOwnedRoute(runHost), "box-1", "boot-1", &l)
+ w.booted.Store(false)
+ w.dataDir = t.TempDir()
+ w.note(rec)
+ w.pass(context.Background())
+ if l.count() != 0 {
+ t.Fatal("a pass acted before the boot pass")
+ }
+ w.bootPass(context.Background())
+ w.pass(context.Background())
+ if l.count() != 1 {
+ t.Fatalf("execs %d, want 1", l.count())
+ }
+}
+
+type listFails struct{ release.Store }
+
+func (listFails) List(context.Context, string) ([]string, error) {
+ return nil, errors.New("store down")
+}