amadan

carlos / platform Public

Converge an always-on sidecar in place, and refuse its /v1/idle

Paul Campbell pushed by paul@keymail.dev 7eaf3e719c0a8b45a6f645716a5500cb27d38705 parent 715df60
2 files changed, +345 −29
  • internal/activator/activator.go +117 −18
  • internal/activator/activator_test.go +228 −11
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 2c4d8730c..4b8534902 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -330,6 +330,11 @@ type actEntry struct {
// run that lasted sidecarHealthyRun.
workerBackoff time.Duration
workerRetryAt time.Time
+ // workerRespawning is set while convergeWorkerLocked's goroutine owns this
+ // entry's child: the sweeper skips the entry entirely so the moment
+ // between the old child's stop and the new one's spawn — when the proc
+ // reads as Exited — is not mistaken for a crash.
+ workerRespawning bool
}
// Sidecar restart pacing — Restart=always semantics without a supervisor: the
@@ -1173,6 +1178,77 @@ func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) {
e.lastActive = now
}
+// convergeWorkerLocked replaces an always-on worker's child with one exec'd
+// from the build live/<host> now points at, on the SAME entry — the caller
+// holds a.mu and has already established the drift persisted past
+// ConvergeGrace and that the per-sweep budget allows it.
+//
+// Stop THEN spawn, and both off the lock, on their own goroutine: the old
+// child's SIGTERM drain is bounded but not instant, and nothing may hold a.mu
+// across it — every request wake on the box queues behind that lock. Stopping
+// first (rather than spawning first and stopping after) is deliberate: a
+// sidecar is an outbound worker, so two live incarnations would double its
+// work — a bot posting twice — where a brief gap costs only latency.
+//
+// e.workerRespawning is the interlock. While it is set the sweeper skips the
+// entry entirely, so the mid-respawn moment when the old proc reads as Exited
+// is not mistaken for a crash (which would spawn a SECOND child). If the
+// entry is stopped or replaced while the respawn is in flight, the new child
+// is stopped instead of installed — a stop always wins.
+//
+// A failed spawn deliberately leaves the dead child in place: the next sweep
+// sees Exited and the crash-restart path takes over, with its backoff. The
+// convergence attempt degrades into supervision rather than into a hole.
+func (a *Service) convergeWorkerLocked(e *actEntry) {
+ e.workerRespawning = true
+ old, rt, host := e.proc, e.route, e.host
+ go func() {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ if err := old.Stop(stopCtx); err != nil {
+ log.Printf("activator: worker %s converge: stopping the old child: %v", host, err)
+ }
+ cancel()
+
+ var (
+ proc Proc
+ spawnVer string
+ )
+ tok, err := newWorkerToken()
+ if err == nil {
+ proc, spawnVer, err = a.backend.StartWorker(context.Background(), rt, workerEnv(a.agentSocket(), host, tok))
+ }
+
+ a.mu.Lock()
+ e.workerRespawning = false
+ live := a.entries[host] == e && e.state == actAwake
+ if err != nil {
+ a.mu.Unlock()
+ log.Printf("activator: WORKER %s failed to respawn onto the new build: %v (the sweep's crash restart takes it from here)", host, err)
+ emitEvent(a.Logs, rt, "error", "sidecar could not restart onto the new build — retrying")
+ return
+ }
+ if !live {
+ // Stopped or replaced under us — a stop always wins. Shed the
+ // child we just made rather than installing it over the top.
+ a.mu.Unlock()
+ log.Printf("activator: worker %s stopped mid-converge — discarding the respawned child", host)
+ shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
+ _ = proc.Stop(shedCtx)
+ shedCancel()
+ return
+ }
+ e.proc = proc
+ e.workerToken = tok
+ e.spawnVersion = spawnVer
+ e.staleSince = time.Time{} // fresh exec: it cannot be stale yet
+ e.spawnedAt = a.now()
+ e.lastActive = a.now()
+ a.mu.Unlock()
+ log.Printf("activator: worker %s respawned onto %s", host, spawnVer)
+ emitEvent(a.Logs, rt, "info", "sidecar restarted onto a new build")
+ }()
+}
+
// NoteWorkerContact stamps last-activity from an agent-socket call. It is a
// worker's only sign of life: nothing routes to a sidecar, so no request will
// ever stamp one, and without contact the idle clock runs to its window and
@@ -1218,18 +1294,31 @@ func (a *Service) SetWorkerNextDue(host string, at int64) error {
return a.due.SetNextDue(host, at)
}
-// SleepWorker stops an awake worker now — the /v1/idle call, "I have nothing
-// to do". It is a stop REQUESTED immediately rather than an idle timestamp
-// nudged backwards, so the ordinary stop path (park the DB, release the lease)
-// does the work. Anything other than an awake worker is a no-op: this must
-// never start something, and never touch an instance.
+// SleepWorker stops an awake HIBERNATING worker now — the /v1/idle call, "I
+// have nothing to do". It is a stop REQUESTED immediately rather than an idle
+// timestamp nudged backwards, so the ordinary stop path (park the DB, release
+// the lease) does the work. Anything other than an awake worker is a no-op:
+// this must never start something, and never touch an instance.
+//
+// An ALWAYS-ON worker's request is logged and REFUSED. Its contract is "the
+// record exists, so the process runs" — and obeying would strand it: the stop
+// deletes the entry, so not even the crash-restart sweep has anything left to
+// see, and it stays down until the agent restarts. A sidecar that genuinely
+// wants to stop exits its own process, which the sweeper then restarts (or
+// its operator deletes the record, which retires it for good).
func (a *Service) SleepWorker(host string) {
a.mu.Lock()
defer a.mu.Unlock()
- if e := a.entries[host]; e != nil && e.worker && e.state == actAwake {
- emitEvent(a.Logs, e.route, "info", "sidecar reported idle — stopping")
- a.beginStopLocked(e)
+ e := a.entries[host]
+ if e == nil || !e.worker || e.state != actAwake {
+ return
}
+ if !e.route.Hibernate {
+ log.Printf("activator: worker %s reported idle but is always-on (hibernate:false) — ignoring", e.host)
+ return
+ }
+ emitEvent(a.Logs, e.route, "info", "sidecar reported idle — stopping")
+ a.beginStopLocked(e)
}
// WorkerToken reports the live per-wake token for host, or "" — the agent
@@ -1381,6 +1470,12 @@ func (a *Service) sweepOnce() {
if e.state != actAwake {
continue
}
+ // A converge respawn owns this entry's child until it finishes: its
+ // old proc is stopped and its new one not yet installed, which every
+ // check below would misread. One tick of patience, not a policy.
+ if e.workerRespawning {
+ continue
+ }
if err := e.rep.Err(); err != nil {
log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
a.beginStopLocked(e)
@@ -1455,17 +1550,21 @@ func (a *Service) sweepOnce() {
cur, readErr := a.liveVer(e.host)
cycle, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, ConvergeGrace)
e.staleSince = next
- // An always-on worker is NOT cycled here: the cycle's second half is
- // "something wakes it back up", and for a worker with no request path
- // and no schedule there is nothing to. It would convergence-cycle
- // itself off the box. It converges instead at the agent's next restart
- // or through `carlos restart` (Task 11) — a real gap, deliberately
- // left visible rather than closed with a stop that strands a member's
- // bot.
- if cycle && e.worker && !e.route.Hibernate {
- cycle = false
- }
if cycle && cycled < maxConvergeCyclesPerSweep {
+ // An always-on worker converges as one MOVE, not as a cycle: the
+ // cycle's second half is "something wakes it back up", and a
+ // worker with no request path and no schedule has nothing that
+ // would. Stop-then-respawn on the same entry is that second half,
+ // performed here. Both existing guards still apply — this only
+ // runs once the drift has persisted past ConvergeGrace, and it
+ // consumes the same one-per-sweep budget, so a bad build cannot
+ // respawn the fleet at once.
+ if e.worker && !e.route.Hibernate {
+ log.Printf("activator: worker %s serving %s but live symlink is now %s — respawning onto the new build", e.host, e.spawnVersion, cur)
+ a.convergeWorkerLocked(e)
+ cycled++
+ continue
+ }
log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur)
a.beginStopLocked(e)
cycled++
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 4cdd233f6..ef4d5fbf5 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -87,12 +87,24 @@ func (p *fakeProc) Stop(context.Context) error {
}
p.mu.Lock()
p.stopped = true
+ // A stopped process HAS exited: modelling that is what lets a test see the
+ // window a converge respawn opens — old child down, new one not yet
+ // installed — which the sweeper must not mistake for a crash.
+ p.dead = true
p.mu.Unlock()
if p.rec != nil {
p.rec.record("stop:" + p.label)
}
return nil
}
+
+// setStopGate installs (or clears) the gate Stop blocks on, under the proc's
+// own lock — the sweeper may be reading it concurrently.
+func (p *fakeProc) setStopGate(gate chan struct{}) {
+ p.mu.Lock()
+ p.stopGate = gate
+ p.mu.Unlock()
+}
func (p *fakeProc) Exited() bool {
p.mu.Lock()
defer p.mu.Unlock()
@@ -187,6 +199,10 @@ type fakeBackend struct {
// that crashes on boot, which the settle window exists to catch.
workerDeadOnStart bool
workerSpawnVer string // what StartWorker reports as the build it execed
+ // workerStartDelay holds StartWorker open for a beat, so a test can land a
+ // sweep INSIDE a converge respawn's window (old child stopped, new one not
+ // yet installed) rather than hoping to hit it.
+ workerStartDelay time.Duration
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
@@ -307,6 +323,12 @@ func (b *fakeBackend) setSidecarErr(err error) {
}
func (b *fakeBackend) StartWorker(_ context.Context, rt registry.Route, env map[string]string) (Proc, string, error) {
+ b.mu.Lock()
+ delay := b.workerStartDelay
+ b.mu.Unlock()
+ if delay > 0 {
+ time.Sleep(delay) // outside the fake's own lock: a slow exec, not a stuck fake
+ }
b.mu.Lock()
defer b.mu.Unlock()
b.workerStartN++
@@ -317,7 +339,14 @@ func (b *fakeBackend) StartWorker(_ context.Context, rt registry.Route, env map[
}
p := &fakeProc{label: "worker", rec: b, dead: b.workerDeadOnStart}
b.workers = append(b.workers, p)
- return p, b.workerSpawnVer, nil
+ // Same symlink model as Start: execVerFn, when wired, is the live/<host>
+ // target this exec would have read — so a convergence test can move the
+ // symlink and have the respawned child report the new build.
+ ver := b.workerSpawnVer
+ if b.execVerFn != nil {
+ ver = b.execVerFn()
+ }
+ return p, ver, nil
}
// lastWorker is the most recently spawned worker child.
@@ -1658,6 +1687,27 @@ func TestOwnsCoversEntryLifetimeAnyState(t *testing.T) {
// entryAwake reports whether host has a live entry in the awake state — how the
// due tests observe an async due-time wake resolving.
+// entryRespawning: whether a converge respawn currently owns this entry's
+// child — the interlock the sweeper honours.
+func (a *Service) entryRespawning(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && e.workerRespawning
+}
+
+// entrySpawnVersion: the build the host's live child was exec'd from, "" if
+// there is no entry — how a test sees that a respawn actually landed on the
+// new binary rather than merely happening.
+func (a *Service) entrySpawnVersion(host string) string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e := a.entries[host]; e != nil {
+ return e.spawnVersion
+ }
+ return ""
+}
+
func (a *Service) entryAwake(host string) bool {
a.mu.Lock()
defer a.mu.Unlock()
@@ -2607,27 +2657,194 @@ func TestAlwaysOnWorkerIgnoresIdleClockAndMaxRun(t *testing.T) {
}
}
-// Convergence cycles a stale child by STOPPING it and letting the next wake
-// exec the new build. An always-on worker has no next wake, so cycling it
-// would take it off the box for good; it is left running instead (it converges
-// at the agent's next restart).
-func TestAlwaysOnWorkerIsNotConvergenceCycled(t *testing.T) {
- a, b, clk, _ := testActivator(t, time.Hour)
+// --- always-on workers: converging onto a shipped build -----------------------
+
+// shippingActivator wires the fake backend and the sweep's liveVer seam to ONE
+// modelled live/<host> symlink, so a test can "ship" by moving it: a child
+// execs whatever it points at now, and the sweep reads the same source.
+func shippingActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeClock, *atomic.Value) {
+ t.Helper()
+ a, b, clk, _ := testActivator(t, idle)
var ver atomic.Value
ver.Store("v1")
- b.workerSpawnVer = "v1"
+ b.execVerFn = func() string { return ver.Load().(string) }
a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ return a, b, clk, &ver
+}
+// A running child never re-reads live/<host>, and an always-on worker never
+// sleeps, so without this it would serve the old binary forever while
+// `carlos deploy` reported success. It converges as one MOVE — stop, respawn,
+// same entry, same lease — because the ordinary cycle's second half ("a
+// request wakes it onto the new build") does not exist for a worker.
+func TestAlwaysOnWorkerRespawnsOntoANewBuild(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
t.Fatalf("WakeWorker: %v", err)
}
+ if got := a.entrySpawnVersion("bot.x.internal"); got != "v1" {
+ t.Fatalf("spawn version = %q, want v1", got)
+ }
+ first := b.lastWorker()
+
ver.Store("v2") // a ship repoints the symlink under the running child
- a.sweepOnce() // first sighting: starts the grace clock, never cycles
+ a.sweepOnce() // first sighting: starts the grace clock, never acts
+ // entryRespawning, not a spawn count: convergeWorkerLocked sets the
+ // interlock synchronously, under the same lock the sweep holds, so this
+ // sees a respawn that has merely been STARTED — a count would race the
+ // goroutine and pass whether or not the grace window was honoured.
+ if a.entryRespawning("bot.x.internal") {
+ t.Fatal("a respawn began on the drift's first sighting — the ConvergeGrace window must pass first")
+ }
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("workerStarts = %d — the grace window must pass before anything is respawned", n)
+ }
clk.advance(ConvergeGrace + time.Minute)
- a.sweepOnce() // the drift has now persisted past the grace
+ a.sweepOnce()
+ waitFor(t, "the respawn onto v2", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
+ if n := b.workerStarts(); n != 2 {
+ t.Fatalf("workerStarts = %d, want 2 (one respawn)", n)
+ }
+ if !first.isStopped() {
+ t.Fatal("the old child was left running beside its replacement — a sidecar would double its work")
+ }
if !a.entryAwake("bot.x.internal") {
- t.Fatal("an always-on worker was convergence-cycled — nothing would wake it onto the new build")
+ t.Fatal("the worker is not awake after converging — the respawn must land on the SAME entry")
+ }
+ if b.lastRep().isClosed() || b.releaseCount() != 0 {
+ t.Fatal("converging respawned through a stop: the lease must not churn for an always-on worker")
+ }
+ // The new child holds the live token, so the agent socket still knows it.
+ if tok := a.WorkerToken("bot.x.internal"); tok == "" || b.lastWorkerEnv()["CARLOS_SIDECAR_TOKEN"] != tok {
+ t.Fatal("the respawned child was not handed the token the socket now expects")
+ }
+}
+
+// The one-per-sweep budget still applies: a bad ship must not respawn every
+// sidecar on the box at once.
+func TestWorkerConvergeRespawnIsRateLimitedPerSweep(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
+ for _, host := range []string{"bot-a.x.internal", "bot-b.x.internal"} {
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker %s: %v", host, err)
+ }
+ }
+ ver.Store("v2")
+ a.sweepOnce() // grace clocks start for both
+ clk.advance(ConvergeGrace + time.Minute)
+
+ a.sweepOnce()
+ waitFor(t, "one respawn", func() bool { return b.workerStarts() == 3 })
+ time.Sleep(100 * time.Millisecond) // give a second respawn every chance to appear
+ if n := b.workerStarts(); n != 3 {
+ t.Fatalf("workerStarts = %d after one sweep, want 3 — two workers, ONE respawn per tick", n)
+ }
+
+ // The next tick takes the other one.
+ a.sweepOnce()
+ waitFor(t, "the second respawn", func() bool { return b.workerStarts() == 4 })
+}
+
+// A HIBERNATING worker's convergence is unchanged: it is CYCLED (stopped, DB
+// parked, lease released) and comes back on its schedule, exactly as before.
+func TestHibernatingWorkerStillConvergenceCycles(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
+ if err := a.WakeWorker(testWorkerRoute("bot.x.internal")); err != nil { // Hibernate: true
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ ver.Store("v2")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+
+ waitFor(t, "the convergence cycle", func() bool { return a.entryCount() == 0 })
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("workerStarts = %d — a hibernating worker is cycled, not respawned in place", n)
+ }
+ if !b.lastRep().isClosed() {
+ t.Fatal("the cycle did not release the lease")
+ }
+}
+
+// A respawn stops the old child before spawning the new one, so for a moment
+// the entry's proc reads as exited. A sweep landing in that window must do
+// NOTHING: read as a crash it would start a second child, and the box would
+// end up running two of a bot that must run once.
+func TestWorkerConvergeRespawnIsNotMistakenForACrash(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ b.mu.Lock()
+ b.workerStartDelay = 300 * time.Millisecond // hold the respawn's spawn open
+ b.mu.Unlock()
+
+ ver.Store("v2")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce() // starts the respawn
+
+ waitFor(t, "the respawn to be in flight", func() bool { return a.entryRespawning("bot.x.internal") })
+ a.sweepOnce() // a tick inside the window
+ if n := b.workerStarts(); n != 1 {
+ t.Fatalf("workerStarts = %d during a respawn — the sweep spawned a second child for a worker mid-converge", n)
+ }
+
+ waitFor(t, "the respawn to land", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
+ if n := b.workerStarts(); n != 2 {
+ t.Fatalf("workerStarts = %d after the respawn, want exactly 2", n)
+ }
+}
+
+// A respawn whose spawn FAILS must not leave a hole: the dead child stays in
+// place, and the ordinary crash-restart path picks it up on a later tick with
+// its backoff. Convergence degrades into supervision.
+func TestFailedConvergeRespawnFallsBackToCrashRestart(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ b.mu.Lock()
+ b.workerErr = errors.New("exec: live symlink gone")
+ b.mu.Unlock()
+
+ ver.Store("v2")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the failed respawn to finish", func() bool { return !a.entryRespawning("bot.x.internal") })
+ if !a.entryAwake("bot.x.internal") {
+ t.Fatal("a failed respawn tore the entry down — the crash restart would have nothing left to fix")
+ }
+
+ // The build comes back; the crash-restart path takes it from here.
+ b.mu.Lock()
+ b.workerErr = nil
+ b.mu.Unlock()
+ clk.advance(sidecarRestartBase + time.Second)
+ a.sweepOnce()
+ waitFor(t, "the crash restart to recover it", func() bool { return a.entrySpawnVersion("bot.x.internal") == "v2" })
+}
+
+// --- /v1/idle and the always-on contract --------------------------------------
+
+// "I have nothing to do" is not a request an always-on worker gets to make:
+// the stop would delete its entry, leaving nothing — not even the crash
+// restart — to bring it back before the agent restarts.
+func TestSleepWorkerIgnoresAnAlwaysOnWorker(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+
+ a.SleepWorker("bot.x.internal")
+
+ if !a.entryAwake("bot.x.internal") {
+ t.Fatal("an always-on worker obeyed /v1/idle — nothing would ever start it again")
+ }
+ if b.releaseCount() != 0 {
+ t.Fatal("an always-on worker's lease was released on a /v1/idle it should have refused")
}
}