amadan

carlos / platform Public

zdd-4 second fix wave: the activator publishes every worker's captured build in the generations file (N1, R17)

Flag-independent worker pin publication, whatever CARLOS_SWAP says. The
activator writes release.GenerationsPath(<data dir>) with
release.WriteGenerations, listing for every worker entry it holds with a
captured serving build (waking, awake, mid-respawn or stopping) that
build's version under its host. An identity with no version, or a name
that is not release.ValidVersion, is skipped: it came from a link outside
the version store, which no pass restages.

Service.GenerationsFile (set before Start; empty publishes nothing) names
the file. persistGenerations takes genMu, snapshots under a.mu, and writes
off a.mu, so an older snapshot never lands over a newer one. It runs before
the first worker spawn (the file exists, even empty, before any worker
runs), after a wake's spawn captures an identity (published while the
entry is still waking), after every respawn installs, and after a worker's
failed wake or stop removes its entry. A failed write is logged, marks the
publication dirty, and the sweep retries it; a spawn never waits on it.

internal/host wires svc.GenerationsFile = release.GenerationsPath(dd),
with dd = dataDir(): the same data dir root's RunAdopt and the agent's
passes hand release.BakeDir, which is where the pass reads the file
(/data/zdd/generations.json in production).

Tests: the file is written before the first spawn; it names the captured
build after a wake, keeps it across a crash recycle, follows a converge and
drops it after a stop; it lists only version-store builds; a write failure
does not block the spawn and the sweep retries it. End to end with a real
ExecBackend worker, release.Ship, signed pointers and release.Pass with
BakeDir: a worker captured on v1 survives two promotions, v1 is re-shipped
with different bytes for another host, the pass refuses with
ErrVersionReused and leaves the copy untouched, and the crash recycle onto
v1 verifies and keeps running.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 889f686854e415c6a3d33a62c888c3f90bd8f955 parent 7ea7df0
6 files changed, +532 −5
  • internal/activator/activator.go +41 −0
  • internal/activator/generations.go +103 −0
  • internal/activator/worker_generations_test.go +374 −0
  • internal/host/agent.go +7 −0
  • internal/release/adopt.go +3 −2
  • internal/release/pins.go +4 −3
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index f2b571e3f..2b425dd2e 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -694,6 +694,21 @@ type Service struct {
// 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
+
+ // GenerationsFile, when set, is where the activator publishes every
+ // worker's captured serving build for root's pin set
+ // (release.GenerationsPath of the data dir root's passes use as BakeDir;
+ // see generations.go). Empty publishes nothing. Set before Start and
+ // before any wake, like OnWake.
+ GenerationsFile string
+
+ // genMu serialises the generations file's writers (persistGenerations).
+ // It is taken BEFORE a.mu, never while holding it. genWritten (a write
+ // has succeeded) and genDirty (the last write failed; the sweep retries)
+ // are read and written under genMu only.
+ genMu sync.Mutex
+ genWritten bool
+ genDirty bool
}
// New wires the state machine to its side-effect backend and due-time store.
@@ -1031,6 +1046,11 @@ func (a *Service) wake(e *actEntry) {
}
a.mu.Unlock()
close(e.ready)
+ if err != nil && e.worker {
+ // The failed wake's entry is gone, and with it any build its spawn
+ // published (generations.go).
+ _ = a.persistGenerations()
+ }
}
// wakeEventMessage is the member-facing wake line, split out so the one thing
@@ -1281,9 +1301,22 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
// 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.
+ //
+ // The generations file exists before any worker runs (ruling R17).
+ a.publishGenerationsBeforeSpawn()
var ident release.StagedIdentity
proc, ident, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, token), nil)
spawnVer, w.spawnIdentity = ident.Version, ident
+ if err == nil {
+ // Published while the entry is still waking: the child is
+ // running these bytes from this moment, settle window or not,
+ // so root must already count them as referenced. The awake
+ // transition stores the same identity again.
+ a.mu.Lock()
+ e.workerServing = ident
+ a.mu.Unlock()
+ _ = a.persistGenerations()
+ }
} else {
proc, spawnVer, err = a.backend.Start(ctx, rt)
}
@@ -1990,6 +2023,8 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.spawnedAt = a.now()
e.lastActive = a.now()
a.mu.Unlock()
+ // The captured build may have changed (a converge): publish it.
+ _ = a.persistGenerations()
log.Printf("activator: worker %s respawned (%s), now running %s", host, why, ident.Version)
emitEvent(a.Logs, rt, "info", "sidecar restarted")
}
@@ -2368,6 +2403,10 @@ func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
a.rememberStopLocked(e.host, outcome)
delete(a.entries, e.host)
a.mu.Unlock()
+ if e.worker {
+ // Its child is gone, so its build is no longer referenced by it.
+ _ = a.persistGenerations()
+ }
close(e.stopped)
log.Printf("activator: %s hibernated", e.host)
}
@@ -2802,6 +2841,8 @@ func (a *Service) sweepOnce() {
started++
}
a.mu.Unlock()
+ // A generations file write that failed is retried here, off a.mu.
+ a.retryGenerations()
}
// RetireExcept stops (final checkpoint + lease release) every live entry whose
diff --git a/internal/activator/generations.go b/internal/activator/generations.go
new file mode 100644
index 000000000..090048d0e
--- /dev/null
+++ b/internal/activator/generations.go
@@ -0,0 +1,103 @@
+package activator
+
+import (
+ "log"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Worker pin publication (zero-downtime deploys spec §4.2, §8.5; ruling R17).
+//
+// A worker recycles onto the build it was exec'd from, its captured serving
+// identity, long after live/<host> and the registry row have moved on. Root's
+// staging must never overwrite that staged copy, and root is a separate
+// process that cannot see this one's memory, so the activator publishes it:
+// the generations file (release.WriteGenerations, read by root through
+// release.GenerationPins) lists, for every worker entry it holds with a
+// captured serving build, that build's version under the worker's host. It
+// does so whatever CARLOS_SWAP says; this flag-independent writer is the one
+// plan 5 extends.
+//
+// The file is written before the first worker spawn (so it exists, even
+// empty, before any worker runs), after every worker spawn or respawn
+// captures an identity, and after every worker stop removes one. A write
+// that fails is logged and marks the publication dirty, and the sweep
+// retries it. It never blocks a spawn.
+
+// generationsSnapshotLocked is what the generations file should say now: for
+// every worker entry (waking, awake, mid-respawn or stopping, since its child
+// may still run) whose captured serving build has a version in the version
+// store, that version. An identity with no version, or one that is not a
+// valid version name, came from a link outside the version store, which no
+// pass restages, so there is nothing to protect. The caller holds a.mu.
+func (a *Service) generationsSnapshotLocked() release.Generations {
+ hosts := map[string][]string{}
+ for host, e := range a.entries {
+ if !e.worker {
+ continue
+ }
+ if v := e.workerServing.Version; v != "" && release.ValidVersion(v) {
+ hosts[host] = []string{v}
+ }
+ }
+ return release.Generations{Hosts: hosts}
+}
+
+// persistGenerations writes the generations file from a snapshot of the
+// entries. Never call it with a.mu held: it takes genMu first, then a.mu only
+// for the snapshot, and does the file I/O off a.mu. genMu serialises the
+// writers, and each one snapshots after taking it, so an older snapshot can
+// never land over a newer one. A no-op when GenerationsFile is empty.
+func (a *Service) persistGenerations() error {
+ if a.GenerationsFile == "" {
+ return nil
+ }
+ a.genMu.Lock()
+ defer a.genMu.Unlock()
+ return a.persistGenerationsGenLocked()
+}
+
+// persistGenerationsGenLocked is persistGenerations' body. The caller holds
+// genMu and not a.mu.
+func (a *Service) persistGenerationsGenLocked() error {
+ a.mu.Lock()
+ g := a.generationsSnapshotLocked()
+ a.mu.Unlock()
+ if err := release.WriteGenerations(a.GenerationsFile, g); err != nil {
+ a.genDirty = true
+ log.Printf("activator: publish worker generations to %s: %v; the sweep retries", a.GenerationsFile, err)
+ return err
+ }
+ a.genWritten = true
+ a.genDirty = false
+ return nil
+}
+
+// publishGenerationsBeforeSpawn makes sure the generations file has been
+// written once before a worker is spawned. A failure is logged and left to
+// the sweep; the spawn goes ahead. Never call it with a.mu held.
+func (a *Service) publishGenerationsBeforeSpawn() {
+ if a.GenerationsFile == "" {
+ return
+ }
+ a.genMu.Lock()
+ defer a.genMu.Unlock()
+ if a.genWritten {
+ return
+ }
+ _ = a.persistGenerationsGenLocked()
+}
+
+// retryGenerations rewrites the generations file if the last write failed.
+// The sweep calls it after releasing a.mu.
+func (a *Service) retryGenerations() {
+ if a.GenerationsFile == "" {
+ return
+ }
+ a.genMu.Lock()
+ defer a.genMu.Unlock()
+ if !a.genDirty {
+ return
+ }
+ _ = a.persistGenerationsGenLocked()
+}
diff --git a/internal/activator/worker_generations_test.go b/internal/activator/worker_generations_test.go
new file mode 100644
index 000000000..131258718
--- /dev/null
+++ b/internal/activator/worker_generations_test.go
@@ -0,0 +1,374 @@
+package activator
+
+import (
+ "bytes"
+ "context"
+ "os"
+ "path/filepath"
+ "reflect"
+ "runtime"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// readGenerations is the generations file's hosts, or nil when it is absent.
+func readGenerations(t *testing.T, path string) map[string][]string {
+ t.Helper()
+ g, err := release.ReadGenerations(path)
+ if err != nil {
+ t.Fatalf("ReadGenerations: %v", err)
+ }
+ if g == nil {
+ return nil
+ }
+ return g.Hosts
+}
+
+func waitForGenerations(t *testing.T, path string, want map[string][]string) {
+ t.Helper()
+ waitFor(t, "generations file naming "+stringOf(want), func() bool {
+ g, err := release.ReadGenerations(path)
+ return err == nil && g != nil && reflect.DeepEqual(g.Hosts, want)
+ })
+}
+
+func stringOf(m map[string][]string) string {
+ var b strings.Builder
+ for h, vs := range m {
+ b.WriteString(h + "=" + strings.Join(vs, ",") + " ")
+ }
+ return b.String()
+}
+
+// Ruling R17: the file exists, even empty, before any worker runs.
+func TestWorkerGenerationsFileIsWrittenBeforeTheFirstSpawn(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ seen := "spawn never ran"
+ b.workerStartHook = func(registry.Route) {
+ g, err := release.ReadGenerations(path)
+ switch {
+ case err != nil:
+ seen = "unreadable: " + err.Error()
+ case g == nil:
+ seen = "absent"
+ default:
+ seen = "hosts=" + stringOf(g.Hosts)
+ }
+ }
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if seen != "hosts=" {
+ t.Fatalf("at the first spawn the generations file was %s; want it written, with no hosts", seen)
+ }
+}
+
+// Ruling R17: the file names a worker's captured build after its wake, keeps
+// naming it across a crash recycle while live/<host> names another build,
+// follows a converge, and drops it once the worker stops.
+func TestWorkerGenerationsFileFollowsTheCapturedBuild(t *testing.T) {
+ a, b, clk, _, ver := stagedActivator(t)
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if got := readGenerations(t, path); !reflect.DeepEqual(got, map[string][]string{host: {"v1"}}) {
+ t.Fatalf("after the wake the file names %v; want %s: [v1]", got, host)
+ }
+
+ ver.Store("v2") // root installs v2 under the running child
+ b.lastWorker().kill()
+ a.sweepOnce()
+ waitForRestart(t, a, b.fakeBackend, host, 2)
+ if calls := b.calls(); calls[1] == nil || *calls[1] != fakeIdentity("v1") {
+ t.Fatalf("crash recycle asked for %v; want the captured v1", calls)
+ }
+ waitForGenerations(t, path, map[string][]string{host: {"v1"}})
+
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })
+ waitForGenerations(t, path, map[string][]string{host: {"v2"}})
+
+ // The stop every path takes (a hibernation, a stand-down, a shutdown).
+ a.mu.Lock()
+ a.beginStopLocked(a.entries[host])
+ a.mu.Unlock()
+ waitFor(t, "the worker's stop", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return a.entries[host] == nil
+ })
+ waitForGenerations(t, path, map[string][]string{})
+}
+
+// A captured build with no version, or a version name the store could never
+// hold, came from a link outside the version store: nothing restages it, so
+// it is not listed, and it never makes the file unwritable.
+func TestWorkerGenerationsListOnlyVersionStoreBuilds(t *testing.T) {
+ a, _, _, _, ver := stagedActivator(t)
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ ver.Store("")
+ if err := a.WakeWorker(alwaysOnWorkerRoute("empty.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ ver.Store("not/a/version")
+ if err := a.WakeWorker(alwaysOnWorkerRoute("odd.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ ver.Store("v3")
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ if got := readGenerations(t, path); !reflect.DeepEqual(got, map[string][]string{"bot.x.internal": {"v3"}}) {
+ t.Fatalf("file names %v; want only bot.x.internal: [v3]", got)
+ }
+}
+
+// Ruling R17: a write that fails never blocks a spawn, and the sweep retries
+// it.
+func TestWorkerGenerationsWriteFailureDoesNotBlockTheSpawnAndTheSweepRetries(t *testing.T) {
+ a, b, _, _, _ := stagedActivator(t)
+ data := filepath.Join(t.TempDir(), "data")
+ if err := os.WriteFile(data, []byte("a file where the data dir should be"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ path := release.GenerationsPath(data)
+ a.GenerationsFile = path
+ host := "bot.x.internal"
+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {
+ t.Fatalf("WakeWorker with an unwritable generations file: %v", err)
+ }
+ if b.workerStarts() != 1 || b.lastWorker().Exited() {
+ t.Fatalf("workerStarts = %d; a failed publication must not block the spawn", b.workerStarts())
+ }
+ if _, err := os.Stat(path); err == nil {
+ t.Fatal("the generations file exists under a file; the test's failure did not happen")
+ }
+ a.sweepOnce() // still failing: retried and logged, nothing else
+ if err := os.Remove(data); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(data, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ a.sweepOnce()
+ if got := readGenerations(t, path); !reflect.DeepEqual(got, map[string][]string{host: {"v1"}}) {
+ t.Fatalf("after the sweep's retry the file names %v; want %s: [v1]", got, host)
+ }
+}
+
+// realWorkerBackend is the fake's lease, restore and replication around a
+// REAL ExecBackend worker spawn: the child is a process exec'd by staged path,
+// and a recycle runs release.VerifyStaged on real bytes.
+type realWorkerBackend struct {
+ *fakeBackend
+ exec *ExecBackend
+}
+
+func (b *realWorkerBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
+ return b.exec.StartWorkerStaged(ctx, rt, env, onto)
+}
+
+// captureStdout runs fn with os.Stdout redirected, and returns what it
+// printed: the adoption pass reports its refusals there.
+func captureStdout(t *testing.T, fn func()) string {
+ t.Helper()
+ r, w, err := os.Pipe()
+ if err != nil {
+ t.Fatal(err)
+ }
+ orig := os.Stdout
+ os.Stdout = w
+ done := make(chan struct{})
+ var buf bytes.Buffer
+ go func() {
+ buf.ReadFrom(r)
+ close(done)
+ }()
+ func() {
+ defer func() { os.Stdout = orig }()
+ fn()
+ }()
+ w.Close()
+ <-done
+ return buf.String()
+}
+
+// Ruling R17, end to end (spec §4.2 "a staged copy that something refers to
+// never changes underneath it"). A worker is captured on build A (v1). Root
+// installs B and then C on its host, so the live link names C and the row
+// says Version C, PrevVersion B: only the generations file still names A.
+// A is then re-shipped under the same name with different bytes, and a pass
+// wants it for another host. The pass refuses with ErrVersionReused and A's
+// staged bytes do not change, so the worker's crash recycle onto A passes
+// VerifyStaged and it keeps running.
+func TestWorkerCapturedBuildSurvivesARestageOfItsReusedName(t *testing.T) {
+ ctx := context.Background()
+ eb := testExecBackend(t)
+ opt, dataDir := eb.OptDir, t.TempDir()
+ fb := &fakeBackend{}
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, &realWorkerBackend{fakeBackend: fb, exec: eb}, reg)
+ a.liveVer = eb.LiveVersion
+ a.GenerationsFile = release.GenerationsPath(dataDir)
+ t.Cleanup(func() { a.Shutdown(context.Background()) })
+
+ store := release.NewDirStore(t.TempDir())
+ seed, pubB64, err := release.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ priv, err := release.ParseSigningKey(seed, "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ pub, err := release.ParsePublicKey(pubB64, "test")
+ if err != nil {
+ t.Fatal(err)
+ }
+ plat := runtime.GOOS + "-" + runtime.GOARCH
+ argv0 := filepath.Join(t.TempDir(), "argv0")
+ build := func(name string) []byte {
+ return []byte("#!/bin/sh\n# build " + name + "\nprintf '%s\\n' \"$0\" > '" + argv0 + ".tmp' && mv '" + argv0 + ".tmp' '" + argv0 + "'\nexec sleep 30\n")
+ }
+ ship := func(ver string, body []byte) {
+ t.Helper()
+ m := release.Manifest{App: "app", Version: ver, Kind: "binary", Artifacts: map[string]release.Artifact{plat: {}}}
+ if err := release.Ship(ctx, store, "bab", m, map[string][]byte{plat: body}); err != nil {
+ t.Fatalf("Ship %s: %v", ver, err)
+ }
+ }
+ promote := func(channel, ver string) {
+ t.Helper()
+ if err := release.Promote(ctx, store, priv, "bab", "app", channel, ver); err != nil {
+ t.Fatalf("Promote %s %s: %v", channel, ver, err)
+ }
+ }
+ pass := func() string {
+ t.Helper()
+ return captureStdout(t, func() {
+ if err := release.Pass(ctx, reg, store, pub, opt, func(string) error { return nil }, release.BakeDir(dataDir)); err != nil {
+ t.Fatalf("Pass: %v", err)
+ }
+ })
+ }
+
+ host := "bot.example.com"
+ rt := registry.Route{Host: host, Kind: "sidecar", App: "app", Channel: "canary/x", Args: `["sidecar","run"]`, Status: "ready"}
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ stagedA := release.VersionPath(opt, "bab", "v1", "app")
+
+ // A (v1) is installed and the worker is captured on it.
+ buildA := build("A")
+ ship("v1", buildA)
+ promote("canary/x", "v1")
+ pass()
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ // The child's own startup write, waited out so it cannot land after the
+ // removal below and pass for the recycle's.
+ if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != stagedA {
+ t.Fatalf("the wake exec'd %q; want A's staged path %s", got, stagedA)
+ }
+ captured := a.entryServing(host)
+ if captured.Version != "v1" || captured.StagedPath != stagedA {
+ t.Fatalf("captured %+v; want v1 at %s", captured, stagedA)
+ }
+ if got := readGenerations(t, a.GenerationsFile); !reflect.DeepEqual(got, map[string][]string{host: {"v1"}}) {
+ t.Fatalf("generations file names %v; want %s: [v1]", got, host)
+ }
+
+ // B (v2), then C (v3): the link and the row move on.
+ ship("v2", build("B"))
+ promote("canary/x", "v2")
+ pass()
+ ship("v3", build("C"))
+ promote("canary/x", "v3")
+ pass()
+ row, err := reg.Get(host)
+ if err != nil || row == nil || row.Version != "v3" || row.PrevVersion != "v2" {
+ t.Fatalf("row = %+v, %v; want Version v3, PrevVersion v2", row, err)
+ }
+ if lv, _ := release.LiveVersion(opt, host); lv != "v3" {
+ t.Fatalf("live = %q, want v3", lv)
+ }
+ rows, err := reg.List()
+ if err != nil {
+ t.Fatal(err)
+ }
+ if pins, err := release.BoxPins(opt, rows); err != nil || pins.Has(stagedA) {
+ t.Fatalf("BoxPins = %v, %v; want A unpinned by links and rows, so only the generations file protects it", pins, err)
+ }
+
+ // A is pruned from the store and re-shipped under the same name with
+ // different bytes, and another host wants it.
+ for _, key := range []string{release.ManifestKey("bab", "app", "v1"), release.ArtifactKey("bab", "app", "v1", plat)} {
+ if err := store.Delete(ctx, key); err != nil {
+ t.Fatalf("prune %s: %v", key, err)
+ }
+ }
+ ship("v1", build("A, re-shipped"))
+ // Hibernating, so the sweep's always-on arm never starts it: only the
+ // pass acts on it here.
+ other := registry.Route{Host: "bot2.example.com", Kind: "sidecar", App: "app", Channel: "canary/y", Args: `["sidecar","run"]`, Status: "ready", Hibernate: true}
+ if err := reg.Upsert(other); err != nil {
+ t.Fatal(err)
+ }
+ promote("canary/y", "v1")
+ out := pass()
+ if !strings.Contains(out, release.ErrVersionReused.Error()) {
+ t.Fatalf("pass printed %q; want the refusal %q", out, release.ErrVersionReused)
+ }
+ if b, err := os.ReadFile(stagedA); err != nil || !bytes.Equal(b, buildA) {
+ t.Fatalf("A's staged copy = %q, %v; want A's bytes untouched", b, err)
+ }
+ if r2, _ := reg.Get(other.Host); r2 == nil || r2.Version == "v1" {
+ t.Fatalf("%s row = %+v; the reused name must not install", other.Host, r2)
+ }
+
+ // The worker crashes, and its recycle onto A verifies and runs.
+ os.Remove(argv0)
+ a.mu.Lock()
+ old := a.entries[host].proc
+ a.mu.Unlock()
+ if err := old.Stop(ctx); err != nil {
+ t.Fatalf("crash the worker: %v", err)
+ }
+ a.sweepOnce()
+ if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != stagedA {
+ t.Fatalf("the recycle exec'd %q; want A's staged path %s", got, stagedA)
+ }
+ waitFor(t, "the recycle to land", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && !e.workerRespawning && e.proc != old
+ })
+ a.mu.Lock()
+ e := a.entries[host]
+ serving, proc := e.workerServing, e.proc
+ a.mu.Unlock()
+ if serving != captured {
+ t.Fatalf("serving after the recycle = %+v; want the captured A %+v", serving, captured)
+ }
+ time.Sleep(100 * time.Millisecond)
+ if proc.Exited() {
+ t.Fatal("the worker recycled onto A is not running")
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index c713732d2..875d1b50d 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -387,6 +387,13 @@ func runAgent(args []string, buildClaims ClaimFactory) {
// (design §6.8). The registry satisfies RouteLookup with its own
// Get, exactly as DueStore above.
svc.Routes = reg
+ // Every worker's captured serving build, published for root's pin
+ // set (zero-downtime deploys spec §4.2, ruling R17), whatever
+ // CARLOS_SWAP says. dd is dataDir(), the same data dir root's
+ // RunAdopt and this agent's passes hand release.BakeDir, which is
+ // where the pass reads release.GenerationsPath from. Set before
+ // Start and before any wake.
+ svc.GenerationsFile = release.GenerationsPath(dd)
if fleetStore != nil {
// MUST be assigned before Start() below spawns the sweeper
// goroutine, which reads svc.OnWake on every successful wake from
diff --git a/internal/release/adopt.go b/internal/release/adopt.go
index b2224db81..800b76e85 100644
--- a/internal/release/adopt.go
+++ b/internal/release/adopt.go
@@ -397,8 +397,9 @@ func PassFilter(ctx context.Context, reg *registry.Registry, s Store, pub ed2551
if err != nil {
return nil, err
}
- // Live and pending generations, from the file plan 5's agent
- // writes into the data dir (index amendments). Every production
+ // Live and pending generations, from the file the agent's
+ // activator writes into the data dir (index amendments; its
+ // workers' captured builds since ruling R17). Every production
// pass passes the data dir as BakeDir: root's RunAdopt, the
// agent's loops, and cold materialization. A pass with no data
// dir (tests) has no agent beside it to ask.
diff --git a/internal/release/pins.go b/internal/release/pins.go
index 0e37c1883..598bcafb6 100644
--- a/internal/release/pins.go
+++ b/internal/release/pins.go
@@ -82,9 +82,10 @@ func BoxPins(optDir string, routes []registry.Route) (StagedPaths, error) {
// generation (current, successor, retiring, spawning: spec §8.5), the version
// each one runs. Live generations exist only in the agent's memory, while the
// pin set is also needed by root's pass, a separate process; this file is how
-// root learns them (index amendments, 2026-10-07). Plan 5's agent writes it
-// with WriteGenerations whenever its operation set changes, and before its
-// first spawn.
+// root learns them (index amendments, 2026-10-07). The agent's activator
+// writes it with WriteGenerations, listing every worker's captured serving
+// build, before its first worker spawn and whenever a spawn or stop changes
+// that set (ruling R17); plan 5 extends the same writer to its operation set.
//
// On disk: {"schema":1,"hosts":{"<host>":["<version>",...]}}
type Generations struct {