amadan

carlos / platform Public

zdd-4 final review: recycles keep the drift clock and converge past ConvergeGrace (F2, R13)

A respawn onto the same captured build no longer resets staleSince; only an
exec from the live link (a converge, or a respawn with no captured build)
starts a fresh drift clock. recycleWorkerLocked, after the CoalesceRecycle
check, reads live/<host> as the sweep does (ConvergeDecision over a.liveVer
and spawnVersion), starting the clock on first sight, and once the drift has
lasted ConvergeGrace it converges (onto nil) instead of recycling the
captured build. A max_run shorter than ConvergeGrace, or a crash loop, which
skips the sweep's drift check, therefore still moves onto the installed
build. It replaces one spawn with one spawn, so it adds no load; the
per-sweep converge budget is local to sweepOnce and is not drawn on.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 7b239870f91a481c4e46bcfbf4fd7e4924734b71 parent 74a43e4
3 files changed, +105 −2
  • internal/activator/activator.go +9 −2
  • internal/activator/worker_identity_test.go +77 −0
  • internal/activator/workerbuild.go +19 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index c3b54bde7..de9eea175 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1984,8 +1984,15 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.workerToken = tok
e.spawnVersion = ident.Version
e.workerServing = ident
- e.workerVersion = "" // the new child has not reported yet
- e.staleSince = time.Time{} // fresh exec: it cannot be stale yet
+ e.workerVersion = "" // the new child has not reported yet
+ // Only an exec from the live link starts a fresh drift clock. A
+ // recycle onto the same captured build serves exactly what it
+ // served before, so a drift it was serving keeps its age, and a
+ // max_run shorter than ConvergeGrace cannot starve the converge
+ // (ruling R13).
+ if onto == nil {
+ e.staleSince = time.Time{}
+ }
e.spawnedAt = a.now()
e.lastActive = a.now()
a.mu.Unlock()
diff --git a/internal/activator/worker_identity_test.go b/internal/activator/worker_identity_test.go
index da987c191..0eceabf61 100644
--- a/internal/activator/worker_identity_test.go
+++ b/internal/activator/worker_identity_test.go
@@ -283,3 +283,80 @@ func TestWorkerRecycleFailingForAnotherReasonKeepsTheCapturedBuild(t *testing.T)
t.Fatalf("workerStarts = %d; want the worker running again", b.workerStarts())
}
}
+
+// Ruling R13: a max_run shorter than ConvergeGrace must not starve the
+// converge. Recycles onto the captured build keep the drift clock, and the
+// recycle that falls due past ConvergeGrace converges instead.
+func TestWorkerMaxRunRecyclesConvergeOnceTheGraceHasPassed(t *testing.T) {
+ a, b, clk, _, ver := stagedActivator(t)
+ host := "bot.x.internal"
+ rt := alwaysOnWorkerRoute(host)
+ rt.MaxRunSeconds = 30
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ ver.Store("v2")
+ converged := -1
+ for i := 1; i <= 6 && converged < 0; i++ {
+ clk.advance(31 * time.Second)
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, i+1)
+ if b.calls()[i] == nil {
+ converged = i
+ }
+ }
+ calls := b.calls()
+ if converged < 0 {
+ t.Fatalf("spawns = %v; six max_run recycles never moved the worker onto v2", calls)
+ }
+ for i := 1; i < converged; i++ {
+ if calls[i] == nil || *calls[i] != fakeIdentity("v1") {
+ t.Fatalf("recycle %d asked for %v; inside the grace a recycle keeps the captured v1", i, calls[i])
+ }
+ }
+ if elapsed := time.Duration(converged) * 31 * time.Second; elapsed < ConvergeGrace {
+ t.Fatalf("converged after %s; want the drift to have lasted ConvergeGrace first", elapsed)
+ }
+ if got := a.entryServing(host); got != fakeIdentity("v2") {
+ t.Fatalf("serving after the converge = %+v, want v2", got)
+ }
+ // No drift now: the next max_run recycles the captured v2.
+ clk.advance(31 * time.Second)
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, converged+2)
+ if c := b.calls()[converged+1]; c == nil || *c != fakeIdentity("v2") {
+ t.Fatalf("recycle after the converge asked for %v; want the captured v2", c)
+ }
+}
+
+// Ruling R13: an exited worker skips the sweep's drift check, so the crash
+// restart measures the drift itself, and a crash loop converges once the
+// grace has passed.
+func TestWorkerCrashLoopConvergesOnceTheGraceHasPassed(t *testing.T) {
+ a, b, clk, _, ver := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ ver.Store("v2")
+ converged := -1
+ for i := 1; i <= 6 && converged < 0; i++ {
+ b.lastWorker().kill()
+ clk.advance(sidecarRestartMax + time.Second)
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, i+1)
+ if b.calls()[i] == nil {
+ converged = i
+ }
+ }
+ calls := b.calls()
+ if converged < 0 {
+ t.Fatalf("spawns = %v; a crash loop never moved the worker onto v2", calls)
+ }
+ if converged < 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
+ t.Fatalf("spawns = %v; the first crash after the install must recycle the captured v1 (the drift has only just been seen)", calls)
+ }
+ if a.entrySpawnVersion(host) != "v2" {
+ t.Fatalf("spawn version = %q after the converge, want v2", a.entrySpawnVersion(host))
+ }
+}
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go
index 7d1f605ff..cd3ae341b 100644
--- a/internal/activator/workerbuild.go
+++ b/internal/activator/workerbuild.go
@@ -43,6 +43,17 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st
// admissible now, that attempt makes the one spawn and the recycle is
// skipped. A worker spawned by a backend without the staged capability has
// no captured build, and keeps today's respawn from the live link.
+//
+// Then the drift (ruling R13). Until plan 5 queues it as a release attempt,
+// the converge is the one path that moves a worker onto an installed build,
+// so a recycle must not starve it: an exited worker never reaches the
+// sweep's drift check, and a max_run shorter than ConvergeGrace would recycle
+// before the check fired. So the recycle reads live/<host> as the sweep does
+// (ConvergeDecision over a.liveVer and spawnVersion), starting the drift
+// clock if this is the first sight of it, and once the drift has lasted
+// ConvergeGrace it converges instead (onto nil). That replaces one spawn
+// with one spawn, so it adds no load, and it does not draw on the sweep's
+// per-sweep converge budget, which is local to sweepOnce.
func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) {
if hook := a.CoalesceRecycle; hook != nil && hook(e.host) {
log.Printf("activator: worker %s %s; a queued release attempt makes the spawn instead of a recycle", e.host, why)
@@ -50,6 +61,14 @@ func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) {
}
var onto *release.StagedIdentity
if e.workerServing.StagedPath != "" {
+ cur, readErr := a.liveVer(e.host)
+ converge, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, a.now(), ConvergeGrace)
+ e.staleSince = next
+ if converge {
+ log.Printf("activator: worker %s %s; it serves %s but live/%s has named %s for at least %s, so it converges onto the new build instead of recycling", e.host, why, e.spawnVersion, e.host, cur, ConvergeGrace)
+ a.respawnWorkerLocked(e, stopOld, "converging onto a new build", nil)
+ return
+ }
id := e.workerServing
onto = &id
} else {