amadan

carlos / platform Public

activator: clear a worker's persisted version on every new incarnation

Paul Campbell pushed by paul@keymail.dev a22c66ec21a0bf9df51e9f19041033ffaf689e08 parent 7e7896e
3 files changed, +191 −11
  • internal/activator/activator.go +28 −0
  • internal/activator/activator_test.go +153 −8
  • internal/host/agent.go +10 −3
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 22df6d142..9d2e6fe04 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -863,6 +863,21 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// full window. Publishing early gives nothing away: a wake that fails
// deletes its entry, so the token still dies with it.
a.setWorkerToken(e, token)
+ // Clear any version this row carries from a previous incarnation
+ // BEFORE spawning: this wake is a new incarnation, and until it
+ // reports for itself over the agent socket, §9.2's honesty rule says
+ // the row must read unknown — never the outgoing (or, on a fresh
+ // boot, simply stale) incarnation's claim. Ordered before StartWorker
+ // deliberately: NoteWorkerReady runs on a completely different
+ // goroutine (the agent socket's HTTP handler), which can observe the
+ // spawn and report moments after it starts — clearing AFTER the spawn
+ // could land behind a fast child's own report and erase it. prev
+ // mirrors NoteWorkerReady's own convention (e.route.Version, not a
+ // fresh registry read) so the two writers agree on what "previous"
+ // means.
+ if cerr := a.due.SetVersions(host, "", rt.Version); cerr != nil {
+ log.Printf("activator: clear worker version for %s: %v", host, cerr)
+ }
proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, token))
} else {
proc, spawnVer, err = a.backend.Start(ctx, rt)
@@ -1294,6 +1309,19 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
// of blocking on it. The timeout is the slow-materialize bound;
// the parent is the "are we still here at all" one.
ctx, cancel := context.WithTimeout(a.lifeCtx, workerSpawnTimeout)
+ // Clear the persisted running version BEFORE spawning: a respawn
+ // (crash restart or build convergence) is a new incarnation on
+ // the SAME entry, and the outgoing child's version must not go
+ // on claiming to be current once its process is gone. Ordered
+ // before StartWorker for the same race NoteWorkerReady is on the
+ // other side of — a fast /v1/ready from the freshly spawned
+ // child must land AFTER this clear, never be overwritten by it.
+ // Placed inside `wanted` deliberately: an abandoned respawn (the
+ // entry was stopped or replaced before we got here) must not
+ // clear a row that isn't this attempt's to touch.
+ if cerr := a.due.SetVersions(host, "", rt.Version); cerr != nil {
+ log.Printf("activator: clear worker version for %s: %v", host, cerr)
+ }
if tok, err = newWorkerToken(); err == nil {
proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok))
}
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 417ce31a4..d70444402 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -2654,29 +2654,74 @@ func TestNoteWorkerReadyPersistsVersionToRegistry(t *testing.T) {
// than write an empty string over whatever was there. Recording "" IS a
// version stamp, and an empty stamp reads as "verified empty" rather than
// "never reported" — the wrong claim.
+// TestNoteWorkerReadyEmptyVersionNeverStampsUnknown seeds the row through a
+// LEGITIMATE report, not a pre-wake SetVersions call: since review round 2
+// (TestFreshWorkerWakeClearsAPreviousIncarnationsVersion), a fresh wake
+// itself clears any pre-existing version — correctly, that's the whole
+// fix for a stale version surviving into a new incarnation — so seeding
+// before WakeWorker would be testing the wake's clear, not NoteWorkerReady's
+// own empty-report guard. This isolates the guard: a real v8 report, then a
+// SECOND call with no version claim, must leave v8 in place.
func TestNoteWorkerReadyEmptyVersionNeverStampsUnknown(t *testing.T) {
a, b, _, reg := testActivator(t, time.Hour)
host := "bot.x.internal"
- // Seed a version on the row BEFORE the wake, the way a prior adopt pass
- // would have — an empty report must not clobber it either.
if err := reg.Upsert(testWorkerRoute(host)); err != nil {
t.Fatalf("upsert: %v", err)
}
- if err := reg.SetVersions(host, "v8", ""); err != nil {
- t.Fatalf("seed version: %v", err)
- }
-
if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
t.Fatalf("WakeWorker: %v", err)
}
- noteReadyAsChild(t, a, b, host, "")
+ noteReadyAsChild(t, a, b, host, "v8")
+ if rt, err := reg.Get(host); err != nil || rt == nil || rt.Version != "v8" {
+ t.Fatalf("get %s after the first report: %+v, %v; want v8", host, rt, err)
+ }
+
+ // A second call reporting no version claim at all — the honesty rule
+ // says "" must never itself be a stamp.
+ a.NoteWorkerReady(host, "")
rt, err := reg.Get(host)
if err != nil || rt == nil {
t.Fatalf("get %s: %+v, %v", host, rt, err)
}
if rt.Version != "v8" {
- t.Fatalf("registry Version = %q after an empty ready report, want the seeded %q untouched", rt.Version, "v8")
+ t.Fatalf("registry Version = %q after a second, empty ready report, want the previously confirmed %q untouched", rt.Version, "v8")
+ }
+}
+
+// TestFreshWorkerWakeClearsAPreviousIncarnationsVersion is review round 1's
+// third finding: a fresh wake is equally a new incarnation (doWake, not just
+// respawnWorkerLocked's crash/converge paths), so a hibernating worker that
+// slept after reporting v1 and wakes again without ever calling /v1/ready
+// this time (it merely survives the settle window — TestWorkerSurviving-
+// SettleWindowIsAwake's shape) must read unknown, not its previous
+// incarnation's v1.
+func TestFreshWorkerWakeClearsAPreviousIncarnationsVersion(t *testing.T) {
+ a, _, _, reg := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ rt := testWorkerRoute(host) // Hibernate: true
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ // Seed the row the way a PREVIOUS incarnation's own report would have
+ // left it before the worker slept.
+ if err := reg.SetVersions(host, "v1", ""); err != nil {
+ t.Fatalf("seed version: %v", err)
+ }
+
+ old := sidecarSettle
+ sidecarSettle = 20 * time.Millisecond
+ defer func() { sidecarSettle = old }()
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+
+ got, err := reg.Get(host)
+ if err != nil || got == nil {
+ t.Fatalf("get %s: %+v, %v", host, got, err)
+ }
+ if got.Version != "" {
+ t.Fatalf("registry Version = %q after a fresh wake that never reported, want empty (unknown) — not the previous incarnation's %q", got.Version, "v1")
}
}
@@ -2734,6 +2779,46 @@ func TestAlwaysOnWorkerCrashRestartsWithGrowingBackoff(t *testing.T) {
waitForRestart(t, a, b, "bot.x.internal", 4)
}
+// TestAlwaysOnWorkerCrashRestartClearsVersionUntilTheNewChildReports is
+// review round 1's finding, the crash-restart twin of the converge-respawn
+// test: restartWorkerLocked also replaces the child on the SAME entry
+// through respawnWorkerLocked, so a crash must clear the row's version too —
+// the dead child's last report must not go on looking current once its
+// process is gone and a new one has taken its place.
+func TestAlwaysOnWorkerCrashRestartClearsVersionUntilTheNewChildReports(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ rt := alwaysOnWorkerRoute(host)
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b, host, "v1")
+
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b, host, 2)
+
+ got, err := reg.Get(host)
+ if err != nil || got == nil {
+ t.Fatalf("get %s: %+v, %v", host, got, err)
+ }
+ if got.Version != "" {
+ t.Fatalf("registry Version = %q immediately after a crash restart, want empty (unknown) until the restarted child reports — not the dead child's v1", got.Version)
+ }
+
+ noteReadyAsChild(t, a, b, host, "v2")
+ got, err = reg.Get(host)
+ if err != nil || got == nil {
+ t.Fatalf("get %s: %+v, %v", host, got, err)
+ }
+ if got.Version != "v2" {
+ t.Fatalf("registry Version = %q after the restarted child reported, want v2", got.Version)
+ }
+}
+
// waitForRestart waits until a supervised respawn has both spawned its child
// and INSTALLED it — the install and the interlock clear happen in one
// critical section, so waiting on the pair is what makes the next kill land on
@@ -3151,6 +3236,19 @@ func shippingActivator(t *testing.T, idle time.Duration) (*Service, *fakeBackend
return a, b, clk, &ver
}
+// shippingActivatorWithRegistry is shippingActivator with the registry handle
+// kept — the version-clear-on-respawn tests need to read routes.version back,
+// which shippingActivator's callers have never needed before.
+func shippingActivatorWithRegistry(t *testing.T, idle time.Duration) (*Service, *fakeBackend, *fakeClock, *registry.Registry, *atomic.Value) {
+ t.Helper()
+ a, b, clk, reg := testActivator(t, idle)
+ var ver atomic.Value
+ ver.Store("v1")
+ b.execVerFn = func() string { return ver.Load().(string) }
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ return a, b, clk, reg, &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,
@@ -3200,6 +3298,53 @@ func TestAlwaysOnWorkerRespawnsOntoANewBuild(t *testing.T) {
}
}
+// TestWorkerConvergeRespawnClearsVersionUntilTheNewChildReports is review
+// round 1's finding: a converge respawn execs a NEW build on the SAME entry
+// (respawnWorkerLocked, reached via convergeWorkerLocked), and the registry
+// row must stop claiming the OUTGOING child's version the moment that
+// happens — reading unknown until the new incarnation reports for itself —
+// rather than going on citing a build that is no longer running.
+func TestWorkerConvergeRespawnClearsVersionUntilTheNewChildReports(t *testing.T) {
+ a, b, clk, reg, ver := shippingActivatorWithRegistry(t, time.Hour)
+ host := "bot.x.internal"
+ rt := alwaysOnWorkerRoute(host)
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b, host, "v1")
+ if got, err := reg.Get(host); err != nil || got == nil || got.Version != "v1" {
+ t.Fatalf("get %s after the first report: %+v, %v; want Version v1", host, got, err)
+ }
+
+ ver.Store("v2") // a ship repoints the symlink under the running child
+ a.sweepOnce() // first sighting: starts the grace clock
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce() // the respawn onto v2 fires
+ waitFor(t, "the respawn onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" })
+
+ // The registry must already read unknown — the new child has not been
+ // given any chance to report yet, and the row must not go on citing v1.
+ got, err := reg.Get(host)
+ if err != nil || got == nil {
+ t.Fatalf("get %s: %+v, %v", host, got, err)
+ }
+ if got.Version != "" {
+ t.Fatalf("registry Version = %q immediately after a converge respawn, want empty (unknown) until the new incarnation reports — not the outgoing child's v1", got.Version)
+ }
+
+ noteReadyAsChild(t, a, b, host, "v2")
+ got, err = reg.Get(host)
+ if err != nil || got == nil {
+ t.Fatalf("get %s: %+v, %v", host, got, err)
+ }
+ if got.Version != "v2" {
+ t.Fatalf("registry Version = %q after the new incarnation reported, want v2", got.Version)
+ }
+}
+
// 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) {
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 863ec22e1..a3b837067 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -377,9 +377,16 @@ func RunAgent(args []string) {
svc.Shutdown(ctx)
// The agent socket's own defer (above, at its construction) stops it
// once this function returns — after Shutdown, per the LIFO ordering
- // explained there. Service is single-use once Shutdown has run (its
- // lifeCtx is cancelled), so there is nothing left to authorize a call
- // against by the time the socket actually stops either way.
+ // explained there. That gap is real, not zero-width: svc.Shutdown
+ // returning here and agentSockCancel actually firing are two
+ // different points in the stack unwind, so the socket is technically
+ // still accepting connections for the width of that unwind. It is
+ // harmless: Shutdown has already stopped every entry and set
+ // shuttingDown, so a call landing in the gap finds no live entry —
+ // WorkerToken returns "" (authorize's unmatchable() then refuses it),
+ // and NoteWorkerContact/NoteWorkerReady/SleepWorker are all no-ops on
+ // a nil entry lookup. Nothing in the gap can reactivate a stopped
+ // worker or authorize against a dead token.
}
}