amadan

carlos / platform Public

zdd-4 task 9: crash and max_run recycle the captured serving build; a worker's report is display-only

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude acc6de7aec91db54585077e0ab0e463ec21e1c30 parent b880fbb
4 files changed, +325 −274
  • internal/activator/activator.go +66 −85
  • internal/activator/activator_test.go +0 −189
  • internal/activator/worker_identity_test.go +210 −0
  • internal/activator/workerbuild.go +49 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f71e93bfd..dd00aac30 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -72,6 +72,7 @@ import (
"github.com/carlosframework/platform/internal/hostwrite"
"github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// SidecarHeader marks every request a co-spawned sidecar makes to its own
@@ -206,13 +207,6 @@ type Backend interface {
// in production (routes.next_due), so a sleeping host's timer survives agent
// restarts instead of dying with the process. *registry.Registry satisfies it
// directly.
-//
-// SetVersions rides along here rather than on a separate interface (Task 8):
-// the registry handle Service already holds for the due-time index is the
-// same one that owns routes.version/prev_version, and in every deployment
-// (production and every test in this package) the two are the same
-// *registry.Registry — a second interface would only name a distinction
-// that doesn't exist anywhere it's used. NoteWorkerReady is its one caller.
type DueStore interface {
SetNextDue(host string, due int64) error
// ClearNextDueIf consumes a FIRED due time — see the registry's own doc
@@ -224,7 +218,6 @@ type DueStore interface {
// so the sweep can start one that appeared after the agent's boot pass —
// which runs once, at arming, and never again.
AlwaysOnWorkers() ([]registry.Route, error)
- SetVersions(host, ver, prev string) error
// EarliestTimers is the hibernation guard's input: host -> soonest armed
// timer. Read with DueInstances, before the sweep takes a.mu.
EarliestTimers() (map[string]int64, error)
@@ -437,9 +430,20 @@ type actEntry struct {
// and never written again, like the other construction-time worker fields.
dueFired int64
- workerToken string // the per-wake agent-socket token
- spawnedAt time.Time // when this child spawned — the max_run and healthy-run clock
- workerVersion string // the build the child reported over the agent socket
+ workerToken string // the per-wake agent-socket token
+ spawnedAt time.Time // when this child spawned — the max_run and healthy-run clock
+ // workerVersion is what the child last reported over the agent socket:
+ // DISPLAY metadata only (spec §7.9). It can be empty or an app-chosen
+ // string, so it never identifies the executable and is never written to
+ // the registry, whose Version is root's install stamp.
+ workerVersion string
+ // workerServing is the build this worker's current child was exec'd
+ // from, captured by the backend at the exec (spec §7.9): staged path and
+ // sha256. Crash and max_run recycles respawn exactly this, never what
+ // live/<host> names by then. Zero when the backend could not report it.
+ // Written under a.mu at the awake transition and on every installed
+ // respawn.
+ workerServing release.StagedIdentity
// Crash-restart pacing for an always-on worker, the co-spawn supervisor's
// rule applied to a standalone member: doubling, capped, and reset by a
@@ -680,6 +684,16 @@ type Service struct {
// made a security-relevant refusal depend on an I/O call that can fail —
// and fail OPEN, since a read error must not stop legitimate bindings.
OnWake func(rt registry.Route)
+
+ // CoalesceRecycle, when set, is asked before a worker is recycled for a
+ // crash or max_run (spec §7.9). True means an attempt for that host (a
+ // queued release attempt admissible now, or one already running) will
+ // make the one spawn itself, so the recycle is skipped. The sweep asks
+ // again next time if the child is still due. Nil, the default and the
+ // whole of plan 4, recycles every time. It is called with a.mu held: it
+ // must answer from its own state, never block, and never call back into
+ // Service. Set before Start, like OnWake. Plan 5's scheduler fills it.
+ CoalesceRecycle func(host string) bool
}
// New wires the state machine to its side-effect backend and due-time store.
@@ -992,6 +1006,7 @@ func (a *Service) wake(e *actEntry) {
// the sweep would never converge it). Empty = unknown → the sweep won't
// cycle. Fresh wake, so it can't be stale yet.
e.spawnVersion = w.spawnVersion
+ e.workerServing = w.spawnIdentity
e.staleSince = time.Time{}
if a.fences[e.host] != 0 {
a.beginStopLocked(e)
@@ -1074,7 +1089,8 @@ type wokenInstance struct {
proc Proc
sidecar Proc
sidecarWanted bool
- spawnVersion string // live/<host> target read at exec — for build convergence
+ spawnVersion string // live/<host> target read at exec — for build convergence
+ spawnIdentity release.StagedIdentity // the build exec'd, as the backend captured it (workers; spec §7.9)
// spawnTook is how long Backend.Start took: exec to the socket accepting,
// which is the instance's OWN boot and nothing of the platform's. wakeTook
// is the whole wake around it, filled in by wake() rather than here
@@ -1262,22 +1278,12 @@ 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))
+ // No registry write (spec §7.9): the row's Version is root's install
+ // stamp, and the build this child runs is the identity the backend
+ // captures at the exec, not anything the child later says.
+ var ident release.StagedIdentity
+ proc, ident, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, token), nil)
+ spawnVer, w.spawnIdentity = ident.Version, ident
} else {
proc, spawnVer, err = a.backend.Start(ctx, rt)
}
@@ -1813,7 +1819,7 @@ func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) {
e.workerRetryAt = now.Add(e.workerBackoff)
log.Printf("activator: worker %s exited — restarting (backoff now %s)", e.host, e.workerBackoff)
// nil: this child is already dead, so there is nothing to retire first.
- a.respawnWorkerLocked(e, nil, "exited")
+ a.recycleWorkerLocked(e, nil, "exited")
}
// convergeWorkerLocked replaces an always-on worker's child with one exec'd
@@ -1821,8 +1827,12 @@ func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) {
// already established the drift persisted past ConvergeGrace and that the
// per-sweep budget allows it. The old child is retired first: it is alive, and
// two incarnations of an outbound worker would double its work.
+//
+// nil: a converge execs what live/<host> names now. Until plan 5 queues it as
+// a paced release attempt, this is the one path that moves a worker to a
+// different installed build.
func (a *Service) convergeWorkerLocked(e *actEntry) {
- a.respawnWorkerLocked(e, e.proc, "converging onto a new build")
+ a.respawnWorkerLocked(e, e.proc, "converging onto a new build", nil)
}
// workerSpawnTimeout bounds a supervised respawn's exec. It is generous
@@ -1838,6 +1848,12 @@ const workerSpawnTimeout = 2 * time.Minute
// lease and replication: only the process is new. The caller holds a.mu.
// stopOld is the child to retire first (nil when it is already dead).
//
+// onto is the build to exec: the captured serving build for a recycle
+// (recycleWorkerLocked), or nil for a converge, which execs what live/<host>
+// names now. A recycle onto a staged copy whose bytes have changed is refused
+// by the backend (release.ErrStagedCopyChanged) and lands in the failed-spawn
+// arm below, which leaves the crash restart to retry with its backoff.
+//
// Everything expensive happens OFF a.mu, on its own goroutine. That is not a
// stylistic preference: ExecBackend.StartWorker can materialize a build from
// the bucket, and the SIGTERM drain of the outgoing child is bounded but not
@@ -1865,7 +1881,7 @@ const workerSpawnTimeout = 2 * time.Minute
// 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. A failed
// converge degrades into supervision rather than into a hole.
-func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
+func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, onto *release.StagedIdentity) {
e.workerRespawning = true
e.respawnDone = make(chan struct{})
done, rt, host := e.respawnDone, e.route, e.host
@@ -1880,10 +1896,10 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
// it.
defer close(done)
var (
- proc Proc
- spawnVer string
- tok string
- err error
+ proc Proc
+ ident release.StagedIdentity
+ tok string
+ err error
// refused is the config gate's answer for this respawn (amadan
// #213). The lease is still held, so Acquire's gate is not in
// the way, and a respawn onto config this box cannot provide
@@ -1910,25 +1926,13 @@ 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 g, ok := a.backend.(configAdmitter); ok {
refused = g.AdmitConfig(ctx, rt)
}
if refused == nil {
- if cerr := a.due.SetVersions(host, "", rt.Version); cerr != nil {
- log.Printf("activator: clear worker version for %s: %v", host, cerr)
- }
+ // No registry write (spec §7.9): see doWake.
if tok, err = newWorkerToken(); err == nil {
- proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok))
+ proc, ident, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok), onto)
}
}
cancel()
@@ -1964,12 +1968,14 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
default:
e.proc = proc
e.workerToken = tok
- e.spawnVersion = spawnVer
+ 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.spawnedAt = a.now()
e.lastActive = a.now()
a.mu.Unlock()
- log.Printf("activator: worker %s respawned (%s), now running %s", host, why, spawnVer)
+ log.Printf("activator: worker %s respawned (%s), now running %s", host, why, ident.Version)
emitEvent(a.Logs, rt, "info", "sidecar restarted")
}
}()
@@ -1987,54 +1993,29 @@ func (a *Service) NoteWorkerContact(host string) {
}
}
-// NoteWorkerReady records a worker's own "I am up" — last-activity, the build
-// it reports running, and the end of its settle window. Deliberately the
-// child's claim, not ours: §9.2's honesty rule is that a sidecar which never
-// reports shows unknown, never "current".
+// NoteWorkerReady records a worker's own "I am up": last-activity, the end
+// of its settle window, and the build it SAYS it runs. That claim is display
+// metadata only (zero-downtime deploys spec §7.9): it can be empty or an
+// app-chosen string, so it never identifies the executable, never overwrites
+// registry Version (root's install stamp) and never decides a recycle. The
+// serving build is workerServing, captured at the exec.
//
-// It acts on a WAKING entry as well as an awake one — that is the whole point
+// It acts on a WAKING entry as well as an awake one, which is the whole point
// of the settle window: a child that boots fast says so and its wake resolves
// immediately instead of waiting out sidecarSettle.
-//
-// The registry write (Task 8) is the durable half of the same claim — the
-// in-memory e.workerVersion above is what the agent itself remembers between
-// restarts of nothing; the row is what `carlos sidecars ls` and everything
-// else off-process reads. It happens OUTSIDE a.mu: SetVersions is a SQL
-// write, and nothing under the entry lock should wait on disk.
-//
-// An EMPTY version is never written. §9.2's rule is that a sidecar which
-// never reports shows unknown — and the registry's own default for an
-// unset version IS the empty string (SetVersions is never called for a
-// fresh row), so "unknown" is represented by an ABSENT write, not by a
-// stamped "". Calling SetVersions with "" here would be a stamp: it would
-// overwrite whatever version the row already carried (from an earlier
-// report, or an adopt pass) with a value that reads as "verified empty"
-// rather than "this child said nothing this time" — the overclaim the rule
-// exists to forbid. So an empty report still ends the settle window and
-// stamps last-activity (the child IS up; it just isn't naming a build) and
-// simply skips the write.
func (a *Service) NoteWorkerReady(host, version string) {
a.mu.Lock()
+ defer a.mu.Unlock()
e := a.entries[host]
if e == nil || !e.worker {
- a.mu.Unlock()
return
}
e.lastActive = a.now()
e.workerVersion = version
- prev := e.route.Version
if !e.readyNoted {
e.readyNoted = true
close(e.readyc)
}
- a.mu.Unlock()
-
- if version == "" {
- return
- }
- if err := a.due.SetVersions(host, version, prev); err != nil {
- log.Printf("activator: record worker version for %s: %v", host, err)
- }
}
// SetWorkerNextDue records a sidecar's next scheduled work, straight to the
@@ -2567,7 +2548,7 @@ func (a *Service) sweepOnce() {
if e.worker && e.workerMaxRun > 0 && now.Sub(e.spawnedAt) >= e.workerMaxRun {
if !e.route.Hibernate {
log.Printf("activator: always-on sidecar %s hit its max_run of %s — recycling its child", e.host, e.workerMaxRun)
- a.respawnWorkerLocked(e, e.proc, "hit its max_run")
+ a.recycleWorkerLocked(e, e.proc, "hit its max_run")
continue
}
log.Printf("activator: %s hit its max_run of %s — stopping", e.host, e.workerMaxRun)
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index f6edb8d93..853563538 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -3326,108 +3326,6 @@ func TestWorkerReadyEndsSettleWindowEarly(t *testing.T) {
}
}
-// TestNoteWorkerReadyPersistsVersionToRegistry pins Task 8's registry half:
-// a worker's own claim over the agent socket must land on the row through
-// the registry's existing version setter, so `carlos sidecars ls` (and
-// anything else reading routes.version) can see it — not just the in-memory
-// entry the agent socket itself consults.
-func TestNoteWorkerReadyPersistsVersionToRegistry(t *testing.T) {
- a, b, _, reg := testActivator(t, time.Hour)
- host := "bot.x.internal"
- if err := reg.Upsert(testWorkerRoute(host)); err != nil {
- t.Fatalf("upsert: %v", err)
- }
- if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
- t.Fatalf("WakeWorker: %v", err)
- }
- noteReadyAsChild(t, a, b, host, "v9")
-
- rt, err := reg.Get(host)
- if err != nil || rt == nil {
- t.Fatalf("get %s: %+v, %v", host, rt, err)
- }
- if rt.Version != "v9" {
- t.Fatalf("registry Version = %q, want %q", rt.Version, "v9")
- }
-}
-
-// TestNoteWorkerReadyEmptyVersionNeverStampsUnknown pins §9.2's honesty rule:
-// a child that reports ready with no version claim must leave the registry
-// column exactly as it was (its default "" — displayed as unknown) rather
-// 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"
- if err := reg.Upsert(testWorkerRoute(host)); err != nil {
- t.Fatalf("upsert: %v", err)
- }
- if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
- t.Fatalf("WakeWorker: %v", err)
- }
- 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 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")
- }
-}
-
// --- always-on workers: supervision ------------------------------------------
func alwaysOnWorkerRoute(host string) registry.Route {
@@ -3482,46 +3380,6 @@ 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
@@ -4128,53 +3986,6 @@ 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/activator/worker_identity_test.go b/internal/activator/worker_identity_test.go
new file mode 100644
index 000000000..0771af5db
--- /dev/null
+++ b/internal/activator/worker_identity_test.go
@@ -0,0 +1,210 @@
+package activator
+
+import (
+ "path/filepath"
+ "sync/atomic"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// stagedActivator is shippingActivatorWithRegistry over stagedFake: ver is
+// what live/<host> names, and every spawn reports the identity it exec'd.
+func stagedActivator(t *testing.T) (*Service, *stagedFake, *fakeClock, *registry.Registry, *atomic.Value) {
+ t.Helper()
+ fb := &fakeBackend{}
+ b := &stagedFake{fakeBackend: fb}
+ clk := &fakeClock{t: time.Unix(1_700_000_000, 0)}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg)
+ a.now = clk.now
+ var ver atomic.Value
+ ver.Store("v1")
+ fb.execVerFn = func() string { return ver.Load().(string) }
+ a.liveVer = func(string) (string, error) { return ver.Load().(string), nil }
+ return a, b, clk, reg, &ver
+}
+
+func (a *Service) entryServing(host string) release.StagedIdentity {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e := a.entries[host]; e != nil {
+ return e.workerServing
+ }
+ return release.StagedIdentity{}
+}
+
+// Spec §7.9: a crash recycle respawns the serving build, not the installed one.
+func TestWorkerCrashRecyclesTheCapturedServingBuild(t *testing.T) {
+ a, b, _, _, ver := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if got := a.entryServing(host); got != fakeIdentity("v1") {
+ t.Fatalf("serving after wake = %+v, want the captured v1", got)
+ }
+ ver.Store("v2") // root installs v2 under the running child
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, 2)
+ calls := b.calls()
+ if len(calls) != 2 || calls[0] != nil {
+ t.Fatalf("spawns = %v; want a wake from the live link, then a recycle", calls)
+ }
+ if calls[1] == nil || *calls[1] != fakeIdentity("v1") {
+ t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls[1])
+ }
+ if got := a.entrySpawnVersion(host); got != "v1" {
+ t.Fatalf("spawn version = %q, want v1", got)
+ }
+}
+
+// Spec §7.9: so does an always-on worker's max_run recycle.
+func TestWorkerMaxRunRecyclesTheCapturedServingBuild(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")
+ clk.advance(31 * time.Second)
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, 2)
+ if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
+ t.Fatalf("max_run recycle asked for %v; want the captured v1 build", calls)
+ }
+}
+
+// Until plan 5 queues it, the drift check is the one path that moves a
+// worker to a different installed build, and it reads the live link.
+func TestWorkerConvergeStillMovesOntoTheInstalledBuild(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")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
+ if calls := b.calls(); len(calls) != 2 || calls[1] != nil {
+ t.Fatalf("converge asked for %v; want the live link (nil)", calls)
+ }
+ if got := a.entryServing(host); got != fakeIdentity("v2") {
+ t.Fatalf("serving after converge = %+v, want v2", got)
+ }
+}
+
+// Plan 5's hook: an admissible queued attempt takes the spawn.
+func TestWorkerRecycleDefersToCoalesceRecycle(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ var asked []string
+ a.CoalesceRecycle = func(h string) bool { asked = append(asked, h); return true }
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ b.lastWorker().kill()
+ a.sweepOnce()
+ if a.entryRespawning(host) || b.workerStarts() != 1 {
+ t.Fatalf("workerStarts = %d; the hook said a queued attempt makes the spawn", b.workerStarts())
+ }
+ if len(asked) != 1 || asked[0] != host {
+ t.Fatalf("hook asked about %v, want [%s]", asked, host)
+ }
+}
+
+// Review Focus 4, and spec §7.9: the report is display metadata; registry
+// Version is root's install stamp.
+func TestNoteWorkerReadyIsDisplayOnly(t *testing.T) {
+ a, b, _, reg := testActivator(t, time.Hour)
+ host := "bot.x.internal"
+ if err := reg.Upsert(testWorkerRoute(host)); err != nil {
+ t.Fatal(err)
+ }
+ if err := reg.SetVersions(host, "v1", ""); err != nil {
+ t.Fatal(err)
+ }
+ if err := a.WakeWorker(testWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b, host, "1.4.2-app-chosen")
+ if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" {
+ t.Fatalf("row = %+v; a worker's report must not overwrite root's stamp", rt)
+ }
+ if got := a.WorkerReportedVersion(host); got != "1.4.2-app-chosen" {
+ t.Fatalf("WorkerReportedVersion = %q", got)
+ }
+ a.NoteWorkerReady(host, "")
+ if rt, _ := reg.Get(host); rt == nil || rt.Version != "v1" {
+ t.Fatalf("row = %+v after an empty report", rt)
+ }
+ if got := a.WorkerReportedVersion("nobody.x.internal"); got != "" {
+ t.Fatalf("WorkerReportedVersion for no worker = %q", got)
+ }
+}
+
+// Spec §7.9: no spawn path writes registry Version: not a wake, a crash
+// recycle or a converge.
+func TestWorkerSpawnsLeaveTheInstalledStampAlone(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.Fatal(err)
+ }
+ if err := reg.SetVersions(host, "v1", "v0"); err != nil {
+ t.Fatal(err)
+ }
+ stamp := func(when string) {
+ t.Helper()
+ if got, _ := reg.Get(host); got == nil || got.Version != "v1" || got.PrevVersion != "v0" {
+ t.Fatalf("%s: row = %+v; want root's stamp v1 (prev v0) untouched", when, got)
+ }
+ }
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ stamp("after a wake")
+ noteReadyAsChild(t, a, b, host, "whatever-the-app-says")
+ stamp("after a ready report")
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b, host, 2)
+ stamp("after a crash recycle")
+ ver.Store("v2")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
+ stamp("after a converge")
+}
+
+// Review Focus 4: an app-chosen report changes nothing about what recycles,
+// and a new child's report starts out unknown.
+func TestWorkerReportingAnAppChosenVersionStillRecyclesItsCapturedBuild(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ noteReadyAsChild(t, a, b.fakeBackend, host, "latest")
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, 2)
+ if calls := b.calls(); len(calls) != 2 || calls[1] == nil || *calls[1] != fakeIdentity("v1") {
+ t.Fatalf("recycle asked for %v; want the captured v1, whatever the child said", calls)
+ }
+ if got := a.WorkerReportedVersion(host); got != "" {
+ t.Fatalf("WorkerReportedVersion = %q after a recycle; the new child has not reported", got)
+ }
+}
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go
index cef329906..7d1f605ff 100644
--- a/internal/activator/workerbuild.go
+++ b/internal/activator/workerbuild.go
@@ -2,6 +2,7 @@ package activator
import (
"context"
+ "log"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -21,3 +22,51 @@ var (
_ stagedWorkerStarter = (*ExecBackend)(nil)
_ stagedWorkerStarter = (*SplitBackend)(nil)
)
+
+// startWorker spawns through the staged capability when the backend has it,
+// and through StartWorker otherwise.
+func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
+ if s, ok := a.backend.(stagedWorkerStarter); ok {
+ return s.StartWorkerStaged(ctx, rt, env, onto)
+ }
+ p, ver, err := a.backend.StartWorker(ctx, rt, env)
+ return p, release.StagedIdentity{Version: ver}, err
+}
+
+// recycleWorkerLocked replaces a worker's child for a reason that is not a
+// build change: a crash, or an always-on worker's max_run. Automatic
+// recycling never changes a worker's build (spec §7.9): the child is
+// respawned from its SERVING build, the staged path and digest captured at
+// its exec, whatever live/<host> names by now. The caller holds a.mu.
+//
+// CoalesceRecycle is asked first. When it says a queued release attempt is
+// 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.
+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)
+ return
+ }
+ var onto *release.StagedIdentity
+ if e.workerServing.StagedPath != "" {
+ id := e.workerServing
+ onto = &id
+ } else {
+ log.Printf("activator: worker %s %s; no captured build to recycle onto, so it respawns from live/%s", e.host, why, e.host)
+ }
+ a.respawnWorkerLocked(e, stopOld, why, onto)
+}
+
+// WorkerReportedVersion is what host's worker last said it runs, over the
+// agent socket: display metadata only (spec §7.9). It may be empty or an
+// app-chosen string, and it never identifies the executable. "" when no
+// worker entry exists.
+func (a *Service) WorkerReportedVersion(host string) string {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ if e := a.entries[host]; e != nil && e.worker {
+ return e.workerVersion
+ }
+ return ""
+}