amadan

carlos / platform Public

plan 5 (scheduler) complete: 17 tasks

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 024b629da7d5865164133da2f7653e816d01c4e4 parent 775d8f0
1 file changed, +1862 −7
  • docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md +1862 −7

CI passed — run details

diff --git a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md
index 6431eeade..29bded50c 100644
--- a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md
+++ b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md
@@ -73,7 +73,7 @@ From the index (exact names):
## Review Focus
-1. **Two releases promoted 20 s apart** (the index's cross-plan item, owned here). The second supersedes queued attempts for the first, per instance; a swap in flight onto the first finishes or is abandoned at its re-check and never switches to it. Pinned in Task 8 (`TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst`) and in Task 15's supersession invariant.
+1. **Two releases promoted 20 s apart** (the index's cross-plan item, owned here). The second supersedes queued attempts for the first, per instance; a swap in flight onto the first finishes or is abandoned at its re-check and never switches to it. Pinned in Task 8 (`TestTwoReleasesTwentySecondsApartNeverSwitchToTheFirst`) and in Task 16's supersession invariant (I4).
2. **A bucket blip while the app policy is read.** A policy read once keeps its last value; one never read gives `hold`, never `overlap`. Pinned in Task 3 (`TestPolicyCacheNeverGuessesOverlap`).
3. **An agent restart with a failure record on disk and the failed build still installed.** The next wake after the restart boots the good build. Pinned in Task 4 (`TestFailureBookSurvivesReopen`) and Task 10 (`TestPlanBootsTheGoodBuildForAHeldInstallation`).
4. **A version name re-shipped with different bytes.** The canonical digest check refuses the attempt as a staging error: no failure record, the stage relay touched, the instance left on its serving build. Pinned in Task 8 (`TestStagingRefusalWritesNoFailureRecord`).
@@ -95,7 +95,8 @@ From the index (exact names):
| `internal/rollout/scheduler.go` (create) | the per-app goroutine and `execute` |
| `internal/rollout/wakeplan.go` (create) | `decideWake`, `Manager.Plan`, `Manager.OnGen` |
| `internal/rollout/worker.go` (create) | socketless attempts, `CoalesceRecycle` |
-| `internal/rollout/*_test.go` (create) | tests per task, the fakes, the simulation |
+| `internal/rollout/status.go` (create) | `Manager.Status`, `reasonWord` |
+| `internal/rollout/*_test.go` (create) | tests per task, the fakes, the simulation (`sim_test.go`) |
| `internal/registry/registry.go` (modify) | four columns, `GoodBuild`, `SetGoodBuild`, `SetServingVersion` |
| `internal/appconfig/appconfig.go` (modify) | `Manifest.Swap`, `Manifest.SwapConcurrency`, validators |
| `internal/appconfig/claimsync.go`, `syncexec.go` (modify) | `Callbacks.Delivered` |
@@ -4085,6 +4086,7 @@ git commit -m "zdd-5 task 8: the manager and the per-app scheduler"
// Service gains: Planner func(ctx context.Context, rt registry.Route) WakePlan; OnGen func(GenEvent); OnDrift func(host string)
func (a *Service) Current(host string) (Gen, bool)
func (a *Service) BounceFor(host string, cause Cause) bool
+ func bounceText(worker bool, cause Cause) string // plan 6 rewords it
func (b *SplitBackend) KillGen(ctx context.Context, rt registry.Route, g Gen) error // only if plan 2 did not add it
// startPinned gains a last parameter: healthBy time.Time (zero: no cap)
var firstWakeBudget, firstWakeKillReserve, firstWakeKillBound, firstWakeSettle time.Duration // 55 s, 25 s, 2 s, 10 s
@@ -4604,13 +4606,20 @@ func (a *Service) BounceFor(host string, cause Cause) bool {
if e == nil || e.state != actAwake {
return false
}
+ emitEvent(a.Logs, e.route, "info", bounceText(e.worker, cause))
+ a.beginStopLocked(e)
+ return true
+}
+
+// bounceText is the one place BounceFor words its event. Draft wording:
+// plan 6 rewrites it through copy review. The flag-off literal in
+// bounceLocked is never edited.
+func bounceText(worker bool, cause Cause) string {
noun := "instance"
- if e.worker {
+ if worker {
noun = "sidecar"
}
- emitEvent(a.Logs, e.route, "info", noun+" restarting ("+string(cause)+")")
- a.beginStopLocked(e)
- return true
+ return noun + " restarting (" + string(cause) + ")"
}
// killFailed ends a failed first-wake build outright (spec §8.2 step 1). A
@@ -5382,4 +5391,1850 @@ git commit -m "zdd-5 task 10: the wake planner and planned-wake outcomes"
---
-<!-- CONTINUE -->
+### Task 11: Socketless attempts and the recycle hook
+
+**Files:**
+- Create: `internal/rollout/worker.go`, `internal/rollout/worker_test.go`
+- Modify: `internal/rollout/scheduler.go` (`execute`)
+
+**Interfaces:**
+- Consumes: Task 8's `Activator.BounceFor`, `Activator.WakeWorker`; Task 10's `workerWait`, `Manager.workers`, `OnGen`; Task 6's `gate`, `start`.
+- Produces:
+ ```go
+ const workerAttemptTimeout = 2 * time.Minute
+ func (s *scheduler) executeWorker(ctx context.Context, d dispatch, rt registry.Route) activator.SwapResult
+ func (q *appQueue) admitOne(host string, now time.Time) (dispatch, bool)
+ func (m *Manager) CoalesceRecycle(host string) bool // plan 4's Service.CoalesceRecycle
+ ```
+
+Spec §6.5 says a failed socketless `release`, `install` or `retry` attempt "is treated as a failed first wake". So a worker attempt is a stop (`BounceFor`) and a wake (`WakeWorker`), and the wake runs Task 9's planned path: Task 10's planner boots the attempt's build, the settle window is the readiness check, and an overlap failure starts the good build in its place. The stop releases and re-takes the lease, unlike §7.9's same-entry respawn (see the report).
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/rollout/worker_test.go`:
+
+```go
+package rollout
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// workerHarness turns the harness's hosts into always-on sidecars.
+func workerHarness(t *testing.T, hosts ...string) *harness {
+ h := newHarness(t, hosts...)
+ for _, host := range hosts {
+ r := h.rows.get(host)
+ r.Kind, r.Hibernate = "sidecar", false
+ h.rows.put(r)
+ }
+ return h
+}
+
+func TestAWorkerReleaseIsAStopAndAFirstWake(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ h.act.wake = func(rt registry.Route) error {
+ h.act.serve(rt.Host, "v2")
+ g, _ := h.act.Current(rt.Host)
+ go func() {
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: g, FirstWake: true})
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: g, FirstWake: true})
+ }()
+ return nil
+ }
+ h.install("v2", "bot")
+ h.pass()
+ waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" })
+ calls, _, _ := h.act.snapshot()
+ if len(calls) != 0 {
+ t.Fatal("a socketless attempt went through Swap; it is stop-then-start (spec §7.9)")
+ }
+ h.act.mu.Lock()
+ defer h.act.mu.Unlock()
+ if len(h.act.bounced) != 1 || h.act.bounced[0] != "bot" {
+ t.Fatalf("bounced %v", h.act.bounced)
+ }
+}
+
+func TestAFailedWorkerReleaseFallsBackAndHalts(t *testing.T) {
+ h := workerHarness(t, "bot")
+ h.stage("v2")
+ h.act.wake = func(rt registry.Route) error {
+ h.act.serve(rt.Host, "v1")
+ g, _ := h.act.Current(rt.Host)
+ go h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, Gen: g, FirstWake: true,
+ Mode: activator.ModeOverlap, Failed: release.StagedIdentity{Version: "v2"}, Reason: "exited within its settle window"})
+ return nil
+ }
+ h.install("v2", "bot")
+ h.pass()
+ waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok })
+ if r := h.rows.get("bot"); r.GoodVersion != "v1" {
+ t.Fatalf("good build = %q after a failed worker release", r.GoodVersion)
+ }
+}
+
+// Spec §7.9: a recycle that falls due coalesces with a queued release attempt
+// only when the scheduler would admit that attempt now.
+func TestARecycleCoalescesOnlyWithAnAdmissibleRelease(t *testing.T) {
+ q := awakeQueue(noBlocks, "a", "b")
+ releaseAll(q, "v2", "a", "b")
+ canary, ok := q.admitOne("a", t0)
+ if !ok || !canary.Canary {
+ t.Fatalf("admitOne(a) = %+v %v, want a's canary", canary, ok)
+ }
+ if _, ok := q.admitOne("b", t0); ok {
+ t.Fatal("coalesced past swap_concurrency 1 while another worker's canary runs")
+ }
+ q.finish(canary, activator.SwapDone)
+ if d, ok := q.admitOne("b", t0); !ok || d.ID.Build != "v2" {
+ t.Fatalf("admitOne(b) = %+v %v after the canary settled", d, ok)
+ }
+ r := awakeQueue(noBlocks, "c")
+ r.wantRestart("c", "r1", t0)
+ if _, ok := r.admitOne("c", t0); ok {
+ t.Fatal("a restart is not a release attempt; the recycle runs on the serving build")
+ }
+ h := newHarness(t, "a")
+ if h.m.CoalesceRecycle("nobody") {
+ t.Fatal("an unknown host coalesced")
+ }
+}
+```
+
+- [ ] **Step 2: Run them to see them fail**
+
+Run: `go test ./internal/rollout/ -run 'TestAWorkerRelease|TestAFailedWorkerRelease|TestARecycleCoalesces' -count=1`
+Expected: FAIL, `q.admitOne undefined`.
+
+- [ ] **Step 3: Write the worker path**
+
+Create `internal/rollout/worker.go`:
+
+```go
+package rollout
+
+import (
+ "context"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// workerAttemptTimeout bounds how long a socketless attempt waits for its
+// wake's outcome: the settle window, a fallback's settle and the wake's own
+// work fit inside it many times over.
+const workerAttemptTimeout = 2 * time.Minute
+
+// executeWorker runs one attempt on a socketless host (spec §6.5, §7.9): stop
+// it, then wake it. While the attempt is registered in m.workers, the
+// planner boots the attempt's build, and the wake's outcome comes back on the
+// channel. No outcome at all (the wake failed before anything executed: a
+// lease, a staging refusal) is not a build failure.
+func (s *scheduler) executeWorker(ctx context.Context, d dispatch, rt registry.Route) activator.SwapResult {
+ m := s.m
+ w := &workerWait{id: d.ID, ch: make(chan activator.GenEvent, 4)}
+ m.mu.Lock()
+ m.workers[rt.Host] = w
+ m.mu.Unlock()
+ defer func() {
+ m.mu.Lock()
+ if m.workers[rt.Host] == w {
+ delete(m.workers, rt.Host)
+ }
+ m.mu.Unlock()
+ }()
+ m.emit(rt, EvSwapping, "info", textSwapping(d.ID.Build, d.ID.Cause))
+ m.d.Act.BounceFor(rt.Host, d.ID.Cause)
+ werr := m.d.Act.WakeWorker(rt)
+ wait := workerAttemptTimeout
+ if werr != nil {
+ // A wake that failed reports its outcome, if any, right after it
+ // returns (Task 9's afterWake runs once e.ready has closed).
+ wait = time.Second
+ }
+ timer := time.NewTimer(wait)
+ defer timer.Stop()
+ for {
+ select {
+ case ev := <-w.ch:
+ switch ev.Kind {
+ case activator.GenHealthy:
+ if !ev.FirstWake {
+ return activator.SwapResult{Outcome: activator.SwapDone, Gen: ev.Gen}
+ }
+ case activator.GenSettled:
+ return activator.SwapResult{Outcome: activator.SwapDone, Gen: ev.Gen}
+ case activator.GenFellBack:
+ return activator.SwapResult{Outcome: activator.SwapFailed, Gen: ev.Gen, Reason: ev.Reason}
+ case activator.GenFailed:
+ return activator.SwapResult{Outcome: activator.SwapDown, Reason: ev.Reason}
+ }
+ case <-timer.C:
+ reason := "the worker's wake reported no outcome"
+ if werr != nil {
+ reason = werr.Error()
+ }
+ s.mu.Lock()
+ s.q.deferHost(rt.Host, m.d.Now().Add(stagingDeferral))
+ s.mu.Unlock()
+ return activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: reason}
+ case <-ctx.Done():
+ return activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: ctx.Err().Error()}
+ }
+ }
+}
+
+// admitOne is admit for one host, for the recycle hook: the host's pending
+// attempt, if it is a release-class attempt the queue would start right now.
+func (q *appQueue) admitOne(host string, now time.Time) (dispatch, bool) {
+ h := q.hosts[host]
+ if h == nil || h.running != nil || q.running >= q.concurrency {
+ return dispatch{}, false
+ }
+ d, ok := q.gate(h, now)
+ if !ok || d.Class != classAdopt {
+ return dispatch{}, false
+ }
+ q.start(h, d)
+ return d, true
+}
+
+// CoalesceRecycle is plan 4's Service.CoalesceRecycle (spec §7.9). It is
+// called under the activator's lock, so it calls nothing in the activator:
+// the attempt it admits runs on its own goroutine. True means the attempt
+// makes the spawn and the recycle is skipped.
+func (m *Manager) CoalesceRecycle(host string) bool {
+ s := m.schedulerOf(host)
+ if s == nil {
+ return false
+ }
+ s.mu.Lock()
+ d, ok := s.q.admitOne(host, m.d.Now())
+ s.mu.Unlock()
+ if !ok {
+ return false
+ }
+ go s.execute(m.ctx, d)
+ return true
+}
+```
+
+In `scheduler.go`'s `execute`, directly after the `rt, err := m.d.Routes.Get(...)` block, add:
+
+```go
+ if rt.IsSidecar() {
+ s.done(d, s.executeWorker(ctx, d, *rt), "")
+ return
+ }
+```
+
+- [ ] **Step 4: Run them to see them pass**
+
+Run: `go test ./internal/rollout/ -count=1 -race`
+Expected: `ok github.com/carlosframework/platform/internal/rollout`
+
+- [ ] **Step 5: Commit**
+
+```bash
+go vet ./internal/rollout/ && test -z "$(gofmt -l internal/rollout)"
+git add internal/rollout/worker.go internal/rollout/worker_test.go internal/rollout/scheduler.go
+git commit -m "zdd-5 task 11: socketless attempts and the recycle hook"
+```
+
+---
+
+### Task 12: The generations file, and the pins root reads
+
+**Files:**
+- Create: `internal/activator/genfile.go`, `internal/activator/genfile_test.go`
+- Modify: `internal/activator/activator.go` (`Service`, `Start`, end of `wake`, end of `finishStop` at 2355, end of `sweepOnce`), plan 3's `leaveSet` and `swapOverlap`, plan 4's recycle install in `respawnWorkerLocked`
+- Modify: `internal/release/pins.go` (plan 4's `BoxPins`), `internal/release/pins_test.go`
+- Modify: `internal/host/versionsweep.go`, `internal/host/versionsweep_test.go`
+
+**Interfaces:**
+- Consumes: plan 4's `release.Generations`, `GenerationsPath`, `WriteGenerations`, `ReadGenerations`, `GenerationPins`, `BoxPins`; plan 3's `e.ops.members`, `e.pin`; Task 9's `e.plan`.
+- Produces:
+ ```go
+ // Service gains: GenerationsFile string (empty: never written, the flag off)
+ func (a *Service) generationsLocked() release.Generations
+ func (a *Service) persistGenerations()
+ func sweepStagedVersionsWith(reg routeLister, optDir, dataDir string, grace time.Duration, now time.Time) // internal/host
+ ```
+
+Plan 4 already reads the file in every adopt pass with a data dir (`GenerationPins(GenerationsPath(cfg.bakeDir), …)`), so root's restage refusal needs nothing more from this plan than the file itself and `GoodVersion` in `BoxPins`. The hourly version sweep (`internal/host/versionsweep.go`) is separate and learns both here.
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/activator/genfile_test.go`:
+
+```go
+package activator
+
+import (
+ "context"
+ "path/filepath"
+ "slices"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestTheGenerationsFileFollowsEveryChange(t *testing.T) {
+ dir := t.TempDir()
+ good := stagedID(t, dir, "v1")
+ a, _, _ := planActivator(t, WakePlan{Boot: good, Mode: ModeOverlap})
+ a.GenerationsFile = release.GenerationsPath(dir)
+ a.persistGenerations()
+ g, err := release.ReadGenerations(a.GenerationsFile)
+ if err != nil || g == nil || len(g.Hosts) != 0 {
+ t.Fatalf("before any spawn: %+v, %v; want an empty file, written", g, err)
+ }
+ done, err := a.beginRequest(context.Background(), testRoute("one.example.com"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ done()
+ g, err = release.ReadGenerations(a.GenerationsFile)
+ if err != nil || !slices.Equal(g.Hosts["one.example.com"], []string{"v1"}) {
+ t.Fatalf("awake: %+v, %v", g, err)
+ }
+ a.BounceFor("one.example.com", CauseRestart)
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ if g, _ = release.ReadGenerations(a.GenerationsFile); g != nil && len(g.Hosts) == 0 {
+ return
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ t.Fatalf("after the stop: %+v, want no generations", g)
+}
+
+func TestTheGenerationsFileNamesThePlannedBuildBeforeItSpawns(t *testing.T) {
+ dir := t.TempDir()
+ good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2")
+ a, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good})
+ a.GenerationsFile = filepath.Join(dir, "zdd", "generations.json")
+ b.gate = make(chan struct{})
+ go a.beginRequest(context.Background(), testRoute("one.example.com"))
+ deadline := time.Now().Add(5 * time.Second)
+ for time.Now().Before(deadline) {
+ g, _ := release.ReadGenerations(a.GenerationsFile)
+ if g != nil && slices.Equal(g.Hosts["one.example.com"], []string{"v1", "v2"}) {
+ close(b.gate)
+ return
+ }
+ time.Sleep(5 * time.Millisecond)
+ }
+ close(b.gate)
+ t.Fatal("the spawning build and its fallback were not pinned before the spawn")
+}
+```
+
+Append to plan 4's `internal/release/pins_test.go`:
+
+```go
+func TestBoxPinsNamesTheGoodBuild(t *testing.T) {
+ opt := t.TempDir()
+ if err := StageArtifact(opt, "bab", "v0", "km", []byte("v0")); err != nil {
+ t.Fatal(err)
+ }
+ pins, err := BoxPins(opt, []registry.Route{{Host: "b.example.com", Kind: "instance", App: "km", Account: "bab", Version: "v2", GoodVersion: "v0"}})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !pins.Has(VersionPath(opt, "bab", "v0", "km")) {
+ t.Fatal("GoodVersion is a pin (spec §8.5)")
+ }
+}
+```
+
+Append to `internal/host/versionsweep_test.go`:
+
+```go
+func TestSweepKeepsGoodBuildsAndGenerations(t *testing.T) {
+ opt, dd := t.TempDir(), t.TempDir()
+ old := time.Now().Add(-48 * time.Hour)
+ for _, v := range []string{"good", "gen", "gone"} {
+ if err := release.StageArtifact(opt, "", v, "km", []byte(v)); err != nil {
+ t.Fatal(err)
+ }
+ _ = os.Chtimes(release.VersionPath(opt, "", v, "km"), old, old)
+ }
+ rows := fakeLister{routes: []registry.Route{{Host: "a.example.com", Kind: "instance", App: "km", Version: "x", GoodVersion: "good"}}}
+ if err := release.WriteGenerations(release.GenerationsPath(dd), release.Generations{Hosts: map[string][]string{"a.example.com": {"gen"}}}); err != nil {
+ t.Fatal(err)
+ }
+ sweepStagedVersionsWith(rows, opt, dd, 6*time.Hour, time.Now())
+ for v, want := range map[string]bool{"good": true, "gen": true, "gone": false} {
+ _, err := os.Stat(release.VersionPath(opt, "", v, "km"))
+ if (err == nil) != want {
+ t.Errorf("%s present = %v, want %v", v, err == nil, want)
+ }
+ }
+}
+
+func TestAnUnreadableGenerationsFileSkipsTheSweep(t *testing.T) {
+ opt, dd := t.TempDir(), t.TempDir()
+ old := time.Now().Add(-48 * time.Hour)
+ if err := release.StageArtifact(opt, "", "gone", "km", []byte("x")); err != nil {
+ t.Fatal(err)
+ }
+ _ = os.Chtimes(release.VersionPath(opt, "", "gone", "km"), old, old)
+ path := release.GenerationsPath(dd)
+ if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.WriteFile(path, []byte("{half"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ sweepStagedVersionsWith(fakeLister{}, opt, dd, 6*time.Hour, time.Now())
+ if _, err := os.Stat(release.VersionPath(opt, "", "gone", "km")); err != nil {
+ t.Fatal("swept with an incomplete pin set (index amendments)")
+ }
+}
+```
+
+`fakeLister` is the existing test double in `versionsweep_test.go`; if its field is not named `routes`, use its name. Add `os`, `path/filepath`, `registry` and `release` to the file's imports as needed.
+
+- [ ] **Step 2: Run them to see them fail**
+
+Run: `TMPDIR=/tmp/k go test ./internal/activator/ ./internal/release/ ./internal/host/ -run 'TestTheGenerationsFile|TestBoxPinsNamesTheGoodBuild|TestSweepKeeps|TestAnUnreadableGenerations' -count=1`
+Expected: FAIL, `a.GenerationsFile undefined`, `undefined: sweepStagedVersionsWith`, and the `BoxPins` assertion.
+
+- [ ] **Step 3: Write the generations writer**
+
+Create `internal/activator/genfile.go`:
+
+```go
+package activator
+
+import (
+ "log"
+ "slices"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// genFileMu serialises writers of the generations file, each of which takes
+// its snapshot under it: a later snapshot can never be overwritten by an
+// earlier one. Never taken while holding a.mu.
+var genFileMu sync.Mutex
+
+// generationsLocked is every live or pending generation's build, per host
+// (spec §8.5: current, successor, retiring, spawning), in plan 4's format.
+// A wake's planned build and its fallback count from the moment they are
+// planned, so they are on disk before anything execs them. The caller holds
+// a.mu.
+func (a *Service) generationsLocked() release.Generations {
+ g := release.Generations{Schema: 1, Hosts: map[string][]string{}}
+ for host, e := range a.entries {
+ set := map[string]bool{e.gen.Build: true, e.workerServing.Version: true}
+ if e.pin != nil {
+ set[e.pin.Build] = true
+ }
+ if e.plan.planned() {
+ set[e.plan.Boot.Version] = true
+ if e.plan.Fallback != nil {
+ set[e.plan.Fallback.Version] = true
+ }
+ }
+ for _, m := range e.ops.members {
+ set[m.gen.Build] = true
+ }
+ delete(set, "")
+ if len(set) == 0 {
+ continue
+ }
+ vers := make([]string, 0, len(set))
+ for v := range set {
+ vers = append(vers, v)
+ }
+ slices.Sort(vers)
+ g.Hosts[host] = vers
+ }
+ return g
+}
+
+// persistGenerations rewrites the generations file (index amendments) with
+// plan 4's atomic writer. Called without a.mu, before the first spawn
+// (Start), before every planned spawn, and after every change to an
+// operation set. A failed write is logged: root then reads the previous
+// file, and the next change or sweep writes again.
+func (a *Service) persistGenerations() {
+ if a.GenerationsFile == "" {
+ return
+ }
+ genFileMu.Lock()
+ defer genFileMu.Unlock()
+ a.mu.Lock()
+ g := a.generationsLocked()
+ a.mu.Unlock()
+ if err := release.WriteGenerations(a.GenerationsFile, g); err != nil {
+ log.Printf("activator: write %s: %v", a.GenerationsFile, err)
+ }
+}
+```
+
+Add to `Service` in `activator.go`, beside `Planner`:
+
+```go
+ // GenerationsFile, when set, is where every live or pending generation's
+ // build is published for root's pin set (release.GenerationsPath). Set
+ // only with CARLOS_SWAP=overlap; empty writes nothing.
+ GenerationsFile string
+```
+
+Call `a.persistGenerations()` (never while holding `a.mu`) at each of these points:
+- `Start`: as its first statement, so the file exists before the first spawn.
+- `doWake`: directly after the planning block Task 9 added (after `a.mu.Unlock()` that stores `e.plan`), and before step 1. Add it after the `e.pin` hold path too: put the call outside the `if a.Planner != nil && e.pin == nil {` block so it runs for both.
+- `wake`: after the `a.afterWake(e, w, err)` line Task 9 added.
+- `finishStop`: after `close(e.stopped)`.
+- `sweepOnce`: as its last statement, after `a.mu.Unlock()`: one rewrite per sweep catches any change the points above miss.
+- plan 3's `swapOverlap`: directly before its `gb.StartGen(` call (the successor is already in the operation set), and plan 3's `leaveSet`: before `close(m.done)`.
+- plan 4's `respawnWorkerLocked` goroutine: after the `default:` arm installs the new child and unlocks (`a.mu.Unlock()` before its `log.Printf("activator: worker %s respawned …`).
+
+- [ ] **Step 4: Put the good build in the pin sets**
+
+In plan 4's `BoxPins` (`internal/release/pins.go`), change the per-row list from `[]string{rt.Version, rt.PrevVersion}` to `[]string{rt.Version, rt.PrevVersion, rt.GoodVersion}`, and add `GoodVersion` to the doc comment's list.
+
+In `internal/host/versionsweep.go`, rename `sweepStagedVersions` to `sweepStagedVersionsWith`, add a `dataDir string` parameter after `optDir`, and in its body add the good build to the per-row pins and the generations file to the set:
+
+```go
+ if r.GoodVersion != "" {
+ pinned[r.GoodVersion] = true
+ }
+```
+
+```go
+ // Live generations exist only in the agent's memory; it publishes them
+ // (zero-downtime plan 5). An unreadable file is an incomplete pin set,
+ // and an incomplete pin set deletes nothing (index amendments).
+ gens, err := release.ReadGenerations(release.GenerationsPath(dataDir))
+ if err != nil {
+ log.Printf("adopt: staged-version sweep skipped: %v", err)
+ return
+ }
+ if gens != nil {
+ for _, vers := range gens.Hosts {
+ for _, v := range vers {
+ pinned[v] = true
+ }
+ }
+ }
+```
+
+(the second block directly before `res, err := release.SweepStaged(...)`). Keep the old name as a wrapper, so existing tests compile unchanged:
+
+```go
+func sweepStagedVersions(reg routeLister, optDir string, grace time.Duration, now time.Time) {
+ sweepStagedVersionsWith(reg, optDir, "", grace, now)
+}
+```
+
+With an empty `dataDir`, `GenerationsPath("")` names `zdd/generations.json` relative to the working directory, which does not exist in a test: no generations. `sweepStagedVersionsPass` calls `sweepStagedVersionsWith(reg, release.OptDir(), dataDir(), versionSweepGrace(os.Getenv), time.Now())`.
+
+- [ ] **Step 5: Run the tests**
+
+Run: `TMPDIR=/tmp/k go test ./internal/activator/ ./internal/release/ ./internal/host/ -count=1`
+Expected: three `ok` lines.
+
+- [ ] **Step 6: Commit**
+
+```bash
+go build ./... && go vet ./internal/activator/ ./internal/release/ ./internal/host/ && test -z "$(gofmt -l internal)"
+git add internal/activator/ internal/release/pins.go internal/release/pins_test.go internal/host/versionsweep.go internal/host/versionsweep_test.go
+git commit -m "zdd-5 task 12: the generations file; GoodVersion and generations in root's pin sets"
+```
+
+---
+
+### Task 13: The box status report
+
+**Files:**
+- Modify: `internal/status/snapshot.go` (on `17362eaf`: `Probes` ends at line 168 after `ReleaseRefusals`; `BoxMetrics` near 419; `CollectBox`'s "something known" condition at 826 and its `BoxMetrics` literal near 860)
+- Test: `internal/status/snapshot_test.go` (append)
+- Create: `internal/rollout/status.go`, `internal/rollout/status_test.go`
+- Modify: `internal/rollout/manager.go` (`Deps.OnChange`, `Manager.observing`), `scheduler.go` (`done`), `wakeplan.go` (`OnGen`)
+
+**Interfaces:**
+- Produces (plan 6 reads these names; they match plan 6's assumed rows):
+ ```go
+ // status
+ type RolloutReport struct { Hosts map[string]HostRollout; Apps map[string]AppRollout; Failures []RolloutFailure }
+ type HostRollout struct { Account, App, Installed, Serving, Good string; Awake bool; Attempt, AttemptBuild, HeldBy, Reason string; ConfigPending bool; ConfigError string }
+ type AppRollout struct { Queued, Running, Settling int } // keyed "<acct>/<app>"
+ type RolloutFailure struct { Account, App, Build, Host, Mode, Reason string; PromotedAt, At time.Time; HeldDown []string }
+ const AttemptQueued, AttemptRunning, AttemptSettling, AttemptFailed, AttemptDown = "queued", "running", "settling", "failed", "down"
+ const ReasonHealth, ReasonSettle, ReasonExit, ReasonStaging, ReasonReused = "health", "settle", "exit", "staging", "reused"
+ // Probes.Rollout and BoxMetrics.Rollout *RolloutReport `json:"rollout,omitempty"`
+ // rollout
+ func (m *Manager) Status() status.RolloutReport
+ func reasonWord(text string) string
+ // Deps gains OnChange func(): called after every rollout transition
+ ```
+
+- [ ] **Step 1: Write the failing tests**
+
+Append to `internal/status/snapshot_test.go`:
+
+```go
+func TestTheRolloutReportRidesThePrivateMetrics(t *testing.T) {
+ r := &RolloutReport{Hosts: map[string]HostRollout{"a.example.com": {App: "blog", Installed: "v2", Serving: "v1", Attempt: AttemptFailed, HeldBy: "v2", Reason: ReasonHealth}}}
+ box := CollectBox(nil, Probes{Rollout: r}, time.Now())
+ if box.Metrics == nil || box.Metrics.Rollout == nil || box.Metrics.Rollout.Hosts["a.example.com"].HeldBy != "v2" {
+ t.Fatalf("metrics = %+v", box.Metrics)
+ }
+ if b := CollectBox(nil, Probes{}, time.Now()); b.Metrics != nil {
+ t.Fatal("the flag off publishes no rollout block and no Metrics because of it")
+ }
+}
+```
+
+Create `internal/rollout/status_test.go`:
+
+```go
+package rollout
+
+import (
+ "testing"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+func TestStatusReportsHeldAndRollingInstances(t *testing.T) {
+ h := newHarness(t, "a", "b")
+ h.stage("v2")
+ gate := h.act.hold("a")
+ h.install("v2", "a", "b")
+ h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v3", Host: "z", Reason: "health check failed", At: t0})
+ h.pass()
+ waitFor(t, "a's canary", func() bool { _, s, _ := h.act.snapshot(); return len(s) == 1 })
+ r := h.m.Status()
+ if got := r.Hosts["a"]; got.Attempt != status.AttemptSettling || got.AttemptBuild != "v2" || got.Installed != "v2" || got.Serving != "v1" {
+ t.Fatalf("a = %+v, want the canary settling onto v2", got)
+ }
+ if got := r.Hosts["b"]; got.Attempt != status.AttemptQueued {
+ t.Fatalf("b = %+v, want queued behind the canary", got)
+ }
+ if app := r.Apps["bab/blog"]; app.Settling != 1 || app.Queued != 1 {
+ t.Fatalf("app = %+v", app)
+ }
+ if len(r.Failures) != 1 || r.Failures[0].Build != "v3" {
+ t.Fatalf("failures = %+v", r.Failures)
+ }
+ close(gate)
+}
+
+func TestAFailedFirstWakeShowsAsHeld(t *testing.T) {
+ h := newHarness(t, "a")
+ h.stage("v2")
+ h.install("v2", "a")
+ h.pass()
+ h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: h.rows.get("a"), FirstWake: true,
+ Mode: activator.ModeOverlap, Gen: activator.Gen{Build: "v1", Digest: "sha-v1"}, Failed: release.StagedIdentity{Version: "v2"},
+ Reason: "exited within 10s of becoming healthy"})
+ got := h.m.Status().Hosts["a"]
+ if got.Attempt != status.AttemptFailed || got.AttemptBuild != "v2" || got.HeldBy != "v2" || got.Reason != status.ReasonExit {
+ t.Fatalf("a = %+v", got)
+ }
+}
+
+func TestReasonWords(t *testing.T) {
+ for text, want := range map[string]string{
+ "health check failed": status.ReasonHealth,
+ "swap: successor exited during its settle": status.ReasonSettle,
+ "exited within 10s of becoming healthy": status.ReasonExit,
+ "staging: digest mismatch": status.ReasonStaging,
+ "version name reused with different content, ship under a new name": status.ReasonReused,
+ "GET / answered 500": status.ReasonHealth,
+ } {
+ if got := reasonWord(text); got != want {
+ t.Errorf("reasonWord(%q) = %q, want %q", text, got, want)
+ }
+ }
+}
+```
+
+Add `"github.com/carlosframework/platform/internal/release"` to that file's imports.
+
+- [ ] **Step 2: Run them to see them fail**
+
+Run: `go test ./internal/status/ ./internal/rollout/ -run 'TestTheRolloutReport|TestStatusReports|TestAFailedFirstWakeShows|TestReasonWords' -count=1`
+Expected: FAIL, `undefined: RolloutReport`.
+
+- [ ] **Step 3: Add the report to status**
+
+In `internal/status/snapshot.go`, add after the `ReleaseRefusals` type (or anywhere at top level):
+
+```go
+// RolloutReport is this box's rollout scheduler state (zero-downtime spec
+// §8.4, §9.1), published only with CARLOS_SWAP=overlap. It names hosts, so
+// it lives in the private Metrics block alone.
+type RolloutReport struct {
+ Hosts map[string]HostRollout `json:"hosts,omitempty"`
+ Apps map[string]AppRollout `json:"apps,omitempty"` // keyed "<acct>/<app>"
+ Failures []RolloutFailure `json:"failures,omitempty"`
+}
+
+// HostRollout is one managed instance. Attempt and Reason are the closed
+// words below; plan 6 renders them.
+type HostRollout struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Installed string `json:"installed"`
+ Serving string `json:"serving,omitempty"`
+ Good string `json:"good,omitempty"`
+ Awake bool `json:"awake"`
+ Attempt string `json:"attempt,omitempty"`
+ AttemptBuild string `json:"attempt_build,omitempty"`
+ HeldBy string `json:"held_by,omitempty"`
+ Reason string `json:"reason,omitempty"`
+ ConfigPending bool `json:"config_pending,omitempty"`
+ ConfigError string `json:"config_error,omitempty"`
+}
+
+// AppRollout counts one app's attempts. Settling is canary settles and
+// first-wake observations under way.
+type AppRollout struct {
+ Queued int `json:"queued"`
+ Running int `json:"running"`
+ Settling int `json:"settling"`
+}
+
+// RolloutFailure is one failure record (spec §8.4).
+type RolloutFailure struct {
+ Account string `json:"account"`
+ App string `json:"app"`
+ Build string `json:"build"`
+ Host string `json:"host"`
+ Mode string `json:"mode,omitempty"`
+ Reason string `json:"reason"`
+ PromotedAt time.Time `json:"promoted_at"`
+ At time.Time `json:"at"`
+ HeldDown []string `json:"held_down,omitempty"`
+}
+
+// The attempt words.
+const (
+ AttemptQueued = "queued"
+ AttemptRunning = "running"
+ AttemptSettling = "settling"
+ AttemptFailed = "failed"
+ AttemptDown = "down"
+)
+
+// The reason words.
+const (
+ ReasonHealth = "health"
+ ReasonSettle = "settle"
+ ReasonExit = "exit"
+ ReasonStaging = "staging"
+ ReasonReused = "reused"
+)
+```
+
+Add `Rollout *RolloutReport` to `Probes` (after `ReleaseRefusals`) and `Rollout *RolloutReport \`json:"rollout,omitempty"\`` to `BoxMetrics` (after its `ReleaseRefusals`). In `CollectBox`, add `|| p.Rollout != nil` to the "something known" condition, and `Rollout: p.Rollout,` to the `BoxMetrics` literal.
+
+- [ ] **Step 4: Write the manager's report**
+
+In `manager.go`, add to `Deps`:
+
+```go
+ // OnChange, when set, is told after every rollout transition, so the box
+ // republishes its status within seconds (plan 6's deploy watch).
+ OnChange func()
+```
+
+add `observing map[string]bool // first wakes under their 10 s observation` to `Manager`, `observing: map[string]bool{}` to `NewManager`'s literal, and this method:
+
+```go
+func (m *Manager) changed() {
+ if m.d.OnChange != nil {
+ m.d.OnChange()
+ }
+}
+```
+
+Call `m.changed()` as the last statement of `scheduler.done` (before the deferred `s.poke()` runs, so add it as a `defer m.changed()` directly after `defer s.poke()`) and of `Manager.OnGen`. In `OnGen`, add first:
+
+```go
+ m.mu.Lock()
+ switch {
+ case ev.Kind == activator.GenHealthy && ev.FirstWake:
+ m.observing[rt.Host] = true
+ case ev.Kind != activator.GenHealthy:
+ delete(m.observing, rt.Host)
+ }
+ m.mu.Unlock()
+```
+
+Create `internal/rollout/status.go`:
+
+```go
+package rollout
+
+import (
+ "strings"
+
+ "github.com/carlosframework/platform/internal/status"
+)
+
+// reasonWord turns an attempt's diagnostic text into one of status's closed
+// reason words, so no free text reaches a member-facing surface. Anything
+// unrecognised is a failed health check: the swap's checks are health, the
+// settle and the exit watch.
+func reasonWord(text string) string {
+ t := strings.ToLower(text)
+ switch {
+ case strings.Contains(t, "reused"):
+ return status.ReasonReused
+ case strings.Contains(t, "staging"), strings.Contains(t, "staged"), strings.Contains(t, "digest"):
+ return status.ReasonStaging
+ case strings.Contains(t, "settle"):
+ return status.ReasonSettle
+ case strings.Contains(t, "exited"):
+ return status.ReasonExit
+ }
+ return status.ReasonHealth
+}
+
+// Status is the box's rollout report (spec §8.4, §9.1).
+func (m *Manager) Status() status.RolloutReport {
+ r := status.RolloutReport{Hosts: map[string]status.HostRollout{}, Apps: map[string]status.AppRollout{}}
+ for _, f := range m.d.Book.List() {
+ r.Failures = append(r.Failures, status.RolloutFailure{Account: f.Account, App: f.App, Build: f.Build, Host: f.Host,
+ Mode: f.Mode, Reason: reasonWord(f.Reason), PromotedAt: f.PromotedAt, At: f.At, HeldDown: f.HeldDown})
+ }
+ m.mu.Lock()
+ scheds := make([]*scheduler, 0, len(m.apps))
+ for _, s := range m.apps {
+ scheds = append(scheds, s)
+ }
+ observing := map[string]bool{}
+ for h := range m.observing {
+ observing[h] = true
+ }
+ cfg := map[string]configTrack{}
+ for h, t := range m.cfg {
+ cfg[h] = *t
+ }
+ m.mu.Unlock()
+ for _, s := range scheds {
+ app := status.AppRollout{}
+ s.mu.Lock()
+ for host, h := range s.q.hosts {
+ hr := status.HostRollout{Account: s.key.acct, App: s.key.app, Installed: h.wantBuild, Serving: h.serving,
+ Good: h.good, Awake: h.serving != ""}
+ if t, ok := cfg[host]; ok && t.delivered != "" && t.delivered != t.active {
+ hr.ConfigPending, hr.ConfigError = true, t.lastErr
+ }
+ switch {
+ case h.running != nil && (h.running.Canary || observing[host]):
+ hr.Attempt, hr.AttemptBuild = status.AttemptSettling, h.running.ID.Build
+ app.Settling++
+ case h.running != nil:
+ hr.Attempt, hr.AttemptBuild = status.AttemptRunning, h.running.ID.Build
+ app.Running++
+ case observing[host]:
+ hr.Attempt, hr.AttemptBuild = status.AttemptSettling, h.wantBuild
+ app.Settling++
+ default:
+ if d, ok := s.q.pending(h); ok {
+ hr.Attempt, hr.AttemptBuild = status.AttemptQueued, d.ID.Build
+ app.Queued++
+ }
+ }
+ if rec, ok := m.d.Book.Lookup(s.key.acct, s.key.app, h.wantBuild); ok && hr.Attempt == "" &&
+ m.d.Book.Blocks(s.key.acct, s.key.app, h.wantBuild, host, h.wantBuild == h.good) {
+ hr.Attempt, hr.AttemptBuild, hr.HeldBy, hr.Reason = status.AttemptFailed, rec.Build, rec.Build, reasonWord(rec.Reason)
+ if m.d.Book.HeldDown(s.key.acct, s.key.app, h.wantBuild, host) {
+ hr.Attempt = status.AttemptDown
+ }
+ }
+ r.Hosts[host] = hr
+ }
+ s.mu.Unlock()
+ r.Apps[s.key.acct+"/"+s.key.app] = app
+ }
+ return r
+}
+```
+
+- [ ] **Step 5: Run the tests**
+
+Run: `go test ./internal/status/ ./internal/rollout/ -count=1 -race`
+Expected: two `ok` lines.
+
+- [ ] **Step 6: Commit**
+
+```bash
+go build ./... && go vet ./internal/status/ ./internal/rollout/ && test -z "$(gofmt -l internal/status internal/rollout)"
+git add internal/status/snapshot.go internal/status/snapshot_test.go internal/rollout/
+git commit -m "zdd-5 task 13: the rollout report in box status"
+```
+
+---
+
+### Task 14: Wire `CARLOS_SWAP=overlap` into the agent
+
+**Files:**
+- Create: `internal/host/rollout.go`, `internal/host/rollout_test.go`
+- Modify: `internal/host/agent.go` (on `17362eaf`: `svc = activator.New(` at 378, `svc.Logs = logSink` at 393, `syncExecSerialized` at 1055, `convergeLegFor` call at 2205, `restartRequestLeg(ctx, routes` at 2265, `statusLoop` at 3057 and its `time.Sleep(delay)`, the probe line `p.ReleaseRefusals = releaseRefusalProbe()` at 3237, `adoptBounceFn` at 3665)
+- Modify: `internal/host/restart.go` (`restartRequestLeg` at 860)
+
+**Interfaces:**
+- Consumes: everything above; plan 1's `activator.SetGenerations`; plan 3's `Config.Swap`; `release.Channel`, `release.CanonicalDemand`, `appconfig.Get`, `appconfig.Callbacks.Delivered`.
+- Produces:
+ ```go
+ func armedRollout() *rollout.Manager // nil unless CARLOS_SWAP=overlap; plan 6 calls it in fastConvergeLoop
+ func armRollout(ctx context.Context, getenv func(string) string, svc *activator.Service, reg *registry.Registry, dd string, sink activator.EventSink, openStore func(context.Context) (release.Store, error)) *rollout.Manager
+ func rolloutLegFor(m *rollout.Manager, reg *registry.Registry) func()
+ func feedRestarts(routes []registry.Route, moved [][2]string, scoped []scopedRestart, gen map[[2]string]string, feed func(host, id string) bool) int
+ func rolloutStatusProbe() *status.RolloutReport
+ var statusNudge chan struct{} // cap 1
+ ```
+
+- [ ] **Step 1: Write the failing tests**
+
+Create `internal/host/rollout_test.go`:
+
+```go
+package host
+
+import (
+ "context"
+ "errors"
+ "path/filepath"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func rolloutFixture(t *testing.T) (*activator.Service, *registry.Registry, string) {
+ t.Helper()
+ dd := t.TempDir()
+ reg, err := registry.Open(filepath.Join(dd, "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ svc := activator.New(activator.Config{Bucket: "b", Idle: time.Hour}, &activator.SplitBackend{}, reg)
+ return svc, reg, dd
+}
+
+// Cross-plan review focus 1: the flag off changes nothing.
+func TestTheFlagOffArmsNothing(t *testing.T) {
+ svc, reg, dd := rolloutFixture(t)
+ opened := false
+ m := armRollout(context.Background(), func(string) string { return "" }, svc, reg, dd, nil,
+ func(context.Context) (release.Store, error) { opened = true; return nil, errors.New("unused") })
+ if m != nil || armedRollout() != nil || opened {
+ t.Fatal("CARLOS_SWAP unset armed the scheduler")
+ }
+ if svc.Planner != nil || svc.OnGen != nil || svc.OnDrift != nil || svc.CoalesceRecycle != nil || svc.GenerationsFile != "" {
+ t.Fatal("CARLOS_SWAP unset set an activator hook")
+ }
+ if adoptBounceFn(svc) == nil {
+ t.Fatal("the adopt bounce must stay BounceAdopted")
+ }
+}
+
+func TestTheFlagOnArmsTheScheduler(t *testing.T) {
+ svc, reg, dd := rolloutFixture(t)
+ t.Cleanup(func() { rolloutMgr.Store(nil) })
+ _, pub, err := release.GenerateKey()
+ if err != nil {
+ t.Fatal(err)
+ }
+ env := map[string]string{"CARLOS_SWAP": "overlap", "CARLOS_RELEASE_PUBKEY": pub}
+ store := release.NewDirStore(t.TempDir())
+ m := armRollout(context.Background(), func(k string) string { return env[k] }, svc, reg, dd, nil,
+ func(context.Context) (release.Store, error) { return store, nil })
+ if m == nil || armedRollout() != m {
+ t.Fatal("CARLOS_SWAP=overlap did not arm the scheduler")
+ }
+ if svc.Planner == nil || svc.OnGen == nil || svc.OnDrift == nil || svc.CoalesceRecycle == nil {
+ t.Fatal("a hook is missing")
+ }
+ if svc.GenerationsFile != release.GenerationsPath(dd) {
+ t.Fatalf("generations file = %q", svc.GenerationsFile)
+ }
+}
+
+func TestAnUnusableStoreLeavesTheBoxOnTodaysPaths(t *testing.T) {
+ svc, reg, dd := rolloutFixture(t)
+ m := armRollout(context.Background(), func(k string) string { return map[string]string{"CARLOS_SWAP": "overlap"}[k] }, svc, reg, dd, nil,
+ func(context.Context) (release.Store, error) { return nil, errors.New("no deployment store") })
+ if m != nil || svc.Planner != nil {
+ t.Fatal("armed without a deploy store: the canonical demand could not be read")
+ }
+}
+
+func TestRestartsFeedTheSchedulerWithTheirIdentity(t *testing.T) {
+ routes := []registry.Route{
+ {Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true},
+ {Host: "bot.internal", Kind: "sidecar", App: "blog", Account: "bab", Status: "ready"},
+ {Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"},
+ {Host: "o.example.com", Kind: "instance", App: "other", Account: "bab", Status: "ready", Hibernate: true},
+ }
+ k := [2]string{"bab", "blog"}
+ got := map[string]string{}
+ n := feedRestarts(routes, [][2]string{k}, nil, map[[2]string]string{k: "etag-7"}, func(h, id string) bool { got[h] = id; return true })
+ if n != 2 || got["a.example.com"] != "gen:etag-7" || got["bot.internal"] != "gen:etag-7" {
+ t.Fatalf("fed %d: %v; want the instance and the sidecar, never the named unit or another app", n, got)
+ }
+}
+```
+
+- [ ] **Step 2: Run them to see them fail**
+
+Run: `TMPDIR=/tmp/k go test ./internal/host/ -run 'TestTheFlagOff|TestTheFlagOn|TestAnUnusableStore|TestRestartsFeed' -count=1`
+Expected: FAIL, `undefined: armRollout`.
+
+- [ ] **Step 3: Write the arming**
+
+Create `internal/host/rollout.go`:
+
+```go
+package host
+
+import (
+ "context"
+ "log"
+ "runtime"
+ "sync/atomic"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/rollout"
+ "github.com/carlosframework/platform/internal/status"
+)
+
+var rolloutMgr atomic.Pointer[rollout.Manager]
+
+// armedRollout is the box's rollout manager, nil unless CARLOS_SWAP=overlap
+// armed one. Every leg that would bounce an exec child asks it first.
+func armedRollout() *rollout.Manager { return rolloutMgr.Load() }
+
+// statusNudge asks the status loop for an early tick: a rollout transition
+// should reach status/boxes/<label>.json in seconds, not at the next
+// five-minute tick. Never sent to with the flag off.
+var statusNudge = make(chan struct{}, 1)
+
+// armRollout builds the scheduler when CARLOS_SWAP=overlap (spec §15) and
+// wires it into svc, which must not have started yet. With the flag off it
+// returns nil and touches nothing. With the flag on but no deploy store, no
+// public key or an unreadable failure book, it logs why and returns nil: the
+// box stays on today's paths, because each of those is a fact the scheduler
+// cannot do without (the canonical demand, the retry signal, the records).
+func armRollout(ctx context.Context, getenv func(string) string, svc *activator.Service, reg *registry.Registry, dd string,
+ sink activator.EventSink, openStore func(context.Context) (release.Store, error)) *rollout.Manager {
+ if svc == nil || !rollout.Enabled(getenv) {
+ return nil
+ }
+ store, err := openStore(ctx)
+ if err != nil {
+ log.Printf("agent: %s=overlap but the deploy store is unavailable (%v): swaps stay stop-then-wake", rollout.SwapEnv, err)
+ return nil
+ }
+ pub, err := release.ParsePublicKey(getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
+ if err != nil {
+ log.Printf("agent: %s=overlap but %v: swaps stay stop-then-wake", rollout.SwapEnv, err)
+ return nil
+ }
+ book, err := rollout.OpenFailureBook(dd)
+ if err != nil {
+ log.Printf("agent: %v: refusing to arm overlap swaps, a lost failure record could move instances back onto a failed build", err)
+ return nil
+ }
+ platform := runtime.GOOS + "-" + runtime.GOARCH
+ m := rollout.NewManager(ctx, rollout.Deps{
+ Act: svc, Routes: reg, Book: book, OptDir: release.OptDir(), DataDir: dd,
+ Demand: func(ctx context.Context, acct, app, ver, path string) (string, string, error) {
+ return release.CanonicalDemand(ctx, store, acct, app, ver, platform, path)
+ },
+ Policy: func(ctx context.Context, acct, app string) (*appconfig.Manifest, error) {
+ return appconfig.Get(ctx, store, acct, app)
+ },
+ Pointer: func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error) {
+ return release.Channel(ctx, store, pub, acct, app, channel)
+ },
+ Events: sink,
+ Touch: func() { touchStageMarker(dd) },
+ OnChange: func() {
+ select {
+ case statusNudge <- struct{}{}:
+ default:
+ }
+ },
+ })
+ svc.Planner = m.Plan
+ svc.OnGen = m.OnGen
+ svc.OnDrift = m.Nudge
+ svc.CoalesceRecycle = m.CoalesceRecycle
+ svc.GenerationsFile = release.GenerationsPath(dd)
+ if routes, err := reg.List(); err == nil {
+ m.Bootstrap(routes) // spec §8.5: awake, healthy instances record their serving build
+ m.Pass(routes)
+ }
+ rolloutMgr.Store(m)
+ log.Printf("agent: overlap swaps armed (%s=overlap)", rollout.SwapEnv)
+ return m
+}
+
+// rolloutLegFor is the converge leg with the flag on (spec §10): the agent
+// pass of spec §6.1, every adoptedConvergeEvery, in place of convergeAdopted.
+func rolloutLegFor(m *rollout.Manager, reg *registry.Registry) func() {
+ return func() {
+ routes, err := reg.List()
+ if err != nil {
+ return
+ }
+ m.Pass(routes)
+ }
+}
+
+// feedRestarts is the restart leg's exec half with the flag on (spec §10):
+// each moved app's exec instances and sidecars, and each narrowed request's
+// hosts, get a restart attempt identified by the request (the app-wide
+// generation's etag, or the newest narrowed request's id). The root half,
+// named units and unit backing, is unchanged.
+func feedRestarts(routes []registry.Route, moved [][2]string, scoped []scopedRestart, gen map[[2]string]string, feed func(host, id string) bool) int {
+ want := map[[2]string]bool{}
+ for _, k := range moved {
+ want[k] = true
+ }
+ n := 0
+ for _, rt := range routes {
+ k := [2]string{release.RouteAccount(rt), rt.App}
+ if !want[k] || !restartEligible(rt) || !execBounceable(rt) {
+ continue
+ }
+ if feed(rt.Host, "gen:"+gen[k]) {
+ n++
+ }
+ }
+ for _, w := range scoped {
+ if len(w.requests) == 0 {
+ continue
+ }
+ id := "scope:" + w.requests[len(w.requests)-1].ID
+ for _, host := range scopedHosts(routes, w.app, w.requests, execBounceable) {
+ if feed(host, id) {
+ n++
+ }
+ }
+ }
+ return n
+}
+
+// rolloutStatusProbe feeds status.Probes.Rollout: nil with the flag off.
+func rolloutStatusProbe() *status.RolloutReport {
+ m := armedRollout()
+ if m == nil {
+ return nil
+ }
+ r := m.Status()
+ return &r
+}
+```
+
+`touchStageMarker` is Task 15's.
+
+- [ ] **Step 4: Wire it into `runAgent` and the legs**
+
+In `runAgent`, directly before `svc = activator.New(*cfg, &activator.SplitBackend{` (line 378), add:
+
+```go
+ // CARLOS_SWAP=overlap turns on generations (plan 1) and swaps
+ // (plan 3) before the Service exists, which reads both once.
+ swapOn := rollout.Enabled(os.Getenv)
+ activator.SetGenerations(swapOn)
+ cfg.Swap = swapOn
+```
+
+After `svc.Logs = logSink` (line 393, still before `startActivator(...)`), add:
+
+```go
+ var rolloutSink activator.EventSink
+ if logSink != nil { // a typed nil in the interface would not be nil
+ rolloutSink = logSink
+ }
+ armRollout(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
+ if armedRollout() != nil {
+ armStageRelay(reg, dd)
+ }
+```
+
+In `fastConvergeLoop`, replace
+
+```go
+ if leg := convergeLegFor(svc, reg); leg != nil {
+ go runConvergeLoop(leg, adoptedConvergeEvery, nil)
+ }
+```
+
+with:
+
+```go
+ if m := armedRollout(); m != nil {
+ go runConvergeLoop(rolloutLegFor(m, reg), adoptedConvergeEvery, nil)
+ } else if leg := convergeLegFor(svc, reg); leg != nil {
+ go runConvergeLoop(leg, adoptedConvergeEvery, nil)
+ }
+```
+
+In `adoptBounceFn`, before `return svc.BounceAdopted`, add:
+
+```go
+ if m := armedRollout(); m != nil {
+ return m.Adopted
+ }
+```
+
+In `syncExecSerialized`, before its final `return appconfig.SyncExec(...)`, add:
+
+```go
+ if m := armedRollout(); m != nil {
+ // The scheduler restarts exec instances, paced, and acknowledges a
+ // revision only once it is active (spec §8.1, §10).
+ cb := appconfig.Callbacks{
+ Restart: func(string) error { return nil },
+ Delivered: m.ConfigDelivered,
+ Ack: m.DeferAck(configAckFunc(ctx, store, "agent")),
+ }
+ return appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, dd, os.Getenv("PLATFORM_APPS_DOMAIN"), cb, opts...)
+ }
+```
+
+In `restart.go`'s `restartRequestLeg`, replace
+
+```go
+ bounceExecRoutes(routes, moved, bounce)
+ cycleWorkers(routes, moved, bounce, wake)
+```
+
+with
+
+```go
+ if m := armedRollout(); m != nil {
+ feedRestarts(routes, moved, scoped, mem.gen, m.Restart)
+ } else {
+ bounceExecRoutes(routes, moved, bounce)
+ cycleWorkers(routes, moved, bounce, wake)
+ }
+```
+
+and wrap the later `bounceScopedRoutes(routes, scoped, bounce)` the same way: `if armedRollout() == nil { bounceScopedRoutes(routes, scoped, bounce) }`.
+
+In `statusTick`'s probe block, after `p.ReleaseRefusals = releaseRefusalProbe()`, add `p.Rollout = rolloutStatusProbe()`. In `statusLoop`, replace `time.Sleep(delay)` with:
+
+```go
+ select {
+ case <-time.After(delay):
+ case <-statusNudge:
+ // A rollout transition (flag on only): publish within seconds,
+ // at most every two, so a busy rollout cannot hammer the bucket.
+ time.Sleep(2 * time.Second)
+ }
+```
+
+Add the `rollout` import to `agent.go`.
+
+- [ ] **Step 5: Run the tests**
+
+Run: `TMPDIR=/tmp/k go test ./internal/host/ -count=1`
+Expected: `ok github.com/carlosframework/platform/internal/host`
+
+- [ ] **Step 6: Commit**
+
+```bash
+go build ./... && go vet ./internal/host/ && test -z "$(gofmt -l internal/host)"
+git add internal/host/
+git commit -m "zdd-5 task 14: wire CARLOS_SWAP=overlap"
+```
+
+---
+
+### Task 15: Stage-relay retransmission
+
+**Files:**
+- Create: `internal/host/stagerelay.go`, `internal/host/stagerelay_test.go`
+- Modify: `internal/host/agent.go` (`stageRequester` at 1357)
+
+**Interfaces:**
+- Produces:
+ ```go
+ type stageRelay struct { /* pending map[string]string, get, touch */ }
+ func (r *stageRelay) note(host, ver string)
+ func (r *stageRelay) tick() bool
+ func armStageRelay(reg *registry.Registry, dd string)
+ func touchStageMarker(dd string)
+ ```
+
+Spec §7.7: "The agent touches it again every `stageRequestEvery` while any exec route's channel head isn't installed, so a touch missed during a running pass is repeated." The relay learns a channel head from the pass that could not stage it (`OnStageBlocked`), which runs only past bake holds; so it keeps asking until root's stamp shows that version installed, the route goes, or a newer head replaces it.
+
+- [ ] **Step 1: Write the failing test**
+
+Create `internal/host/stagerelay_test.go`:
+
+```go
+package host
+
+import (
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestTheRelayAsksAgainUntilTheHeadIsInstalled(t *testing.T) {
+ rows := map[string]*registry.Route{"a": {Host: "a", Version: "v1"}, "b": {Host: "b", Version: "v1"}}
+ touches := 0
+ r := &stageRelay{pending: map[string]string{},
+ get: func(h string) (*registry.Route, error) { return rows[h], nil },
+ touch: func() { touches++ }}
+ if r.tick() || touches != 0 {
+ t.Fatal("touched with nothing outstanding")
+ }
+ r.note("a", "v2")
+ r.note("b", "v2")
+ if !r.tick() || touches != 1 {
+ t.Fatal("one touch per tick while heads are outstanding")
+ }
+ rows["a"].Version = "v2"
+ delete(rows, "b")
+ if r.tick() || touches != 1 {
+ t.Fatal("kept asking after root installed a's head and b's route went")
+ }
+ r.note("a", "v3")
+ r.get = func(string) (*registry.Route, error) { return nil, errors.New("database is locked") }
+ if !r.tick() {
+ t.Fatal("an unreadable row is not proof the head is installed")
+ }
+}
+```
+
+- [ ] **Step 2: Run it to see it fail**
+
+Run: `TMPDIR=/tmp/k go test ./internal/host/ -run TestTheRelayAsksAgain -count=1`
+Expected: FAIL, `undefined: stageRelay`.
+
+- [ ] **Step 3: Write the relay**
+
+Create `internal/host/stagerelay.go`:
+
+```go
+package host
+
+import (
+ "log"
+ "os"
+ "path/filepath"
+ "sync"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// stageRelay retransmits the stage request (spec §7.7). carlos-stage.path
+// does not act on a change made while carlos-stage.service is running, so a
+// touch that lands mid-pass is lost; this asks again every stageRequestEvery
+// until root's stamp shows the head installed. Armed only with the flag on.
+type stageRelay struct {
+ mu sync.Mutex
+ pending map[string]string // host -> the channel head a pass could not stage
+ get func(host string) (*registry.Route, error)
+ touch func()
+}
+
+func (r *stageRelay) note(host, ver string) {
+ r.mu.Lock()
+ r.pending[host] = ver
+ r.mu.Unlock()
+}
+
+// tick touches the marker once if any head is still outstanding, and drops
+// the ones root has installed or whose route is gone.
+func (r *stageRelay) tick() bool {
+ r.mu.Lock()
+ snap := make(map[string]string, len(r.pending))
+ for h, v := range r.pending {
+ snap[h] = v
+ }
+ r.mu.Unlock()
+ outstanding := false
+ for host, ver := range snap {
+ rt, err := r.get(host)
+ if err != nil {
+ outstanding = true
+ continue
+ }
+ if rt == nil || rt.Version == ver {
+ r.mu.Lock()
+ if r.pending[host] == ver {
+ delete(r.pending, host)
+ }
+ r.mu.Unlock()
+ continue
+ }
+ outstanding = true
+ }
+ if outstanding {
+ r.touch()
+ }
+ return outstanding
+}
+
+var armedStageRelay atomic.Pointer[stageRelay]
+
+// touchStageMarker writes the stage-request marker, as stageRequester does.
+// A failed write costs latency; carlos-adopt.timer is the floor.
+func touchStageMarker(dd string) {
+ if err := os.WriteFile(filepath.Join(dd, stageRequestMarker), []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644); err != nil {
+ log.Printf("agent: stage relay: %v", err)
+ }
+}
+
+// armStageRelay starts the retransmission loop for the life of the process.
+func armStageRelay(reg *registry.Registry, dd string) {
+ r := &stageRelay{pending: map[string]string{}, get: reg.Get, touch: func() { touchStageMarker(dd) }}
+ armedStageRelay.Store(r)
+ go func() {
+ t := time.NewTicker(stageRequestEvery)
+ defer t.Stop()
+ for range t.C {
+ r.tick()
+ }
+ }()
+}
+```
+
+In `stageRequester`, change the returned closure from `return func(host, ver string, _ error) { t.request(host, ver) }` to:
+
+```go
+ return func(host, ver string, _ error) {
+ t.request(host, ver)
+ if r := armedStageRelay.Load(); r != nil {
+ r.note(host, ver)
+ }
+ }
+```
+
+- [ ] **Step 4: Run the tests**
+
+Run: `TMPDIR=/tmp/k go test ./internal/host/ -count=1`
+Expected: `ok github.com/carlosframework/platform/internal/host`
+
+- [ ] **Step 5: Commit**
+
+```bash
+go vet ./internal/host/ && test -z "$(gofmt -l internal/host)"
+git add internal/host/stagerelay.go internal/host/stagerelay_test.go internal/host/agent.go
+git commit -m "zdd-5 task 15: stage-relay retransmission"
+```
+
+---
+
+### Task 16: The invariant simulation, with mutation checks
+
+**Files:**
+- Create: `internal/rollout/sim_test.go`
+
+**Interfaces:**
+- Consumes: the real `appQueue` (Tasks 5, 6, 11), `FailureBook` in memory (Task 4), `failCohort` (Task 6), `decideWake` (Task 10). Only the activator is modelled.
+
+The invariants (spec §12 "Concurrency", and this plan's brief):
+- **I1:** never a release-cause failure that leaves an instance without a good build it can boot. Every wake of an instance that has a good build boots something, and no wake boots, for an instance that has not served it, a build that failed on this box.
+- **I2:** `swap_concurrency` is never exceeded.
+- **I3:** a failure record never blocks an instance already serving that build: its restart and config attempts are admitted whenever capacity and their own cohort allow, and no record is ever written against a build by an instance that serves it.
+- **I4:** a swap never switches to a superseded build (the index's review item 5).
+
+- [ ] **Step 1: Write the simulation**
+
+Create `internal/rollout/sim_test.go`:
+
+```go
+package rollout
+
+import (
+ "fmt"
+ "math/rand"
+ "os"
+ "strconv"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+type simHost struct {
+ name string
+ ch int
+ awake bool
+ serving, good string
+ served map[string]bool
+}
+
+type sim struct {
+ r *rand.Rand
+ q *appQueue
+ book *FailureBook
+ hosts []*simHost
+ byName map[string]*simHost
+ installed [2]string
+ bad map[string]bool // builds that fail when first run (ground truth)
+ failed map[string]bool // builds whose failure stands on this box (ground truth)
+ running map[string]dispatch
+ n int
+ now time.Time
+ step int
+}
+
+func newSim(seed int64) *sim {
+ s := &sim{r: rand.New(rand.NewSource(seed)), book: &FailureBook{}, byName: map[string]*simHost{},
+ bad: map[string]bool{}, failed: map[string]bool{}, running: map[string]dispatch{}, now: t0}
+ s.q = newAppQueue("bab", "blog", func(build, host string, served bool) bool {
+ return s.book.Blocks("bab", "blog", build, host, served)
+ })
+ s.q.concurrency = 1 + s.r.Intn(3)
+ s.installed = [2]string{"b0", "b0"}
+ for i := range 5 {
+ h := &simHost{name: "h" + strconv.Itoa(i), ch: i % 2, awake: s.r.Intn(2) == 0, served: map[string]bool{"b0": true}}
+ h.good = "b0"
+ if h.awake {
+ h.serving = "b0"
+ }
+ s.hosts = append(s.hosts, h)
+ s.byName[h.name] = h
+ s.q.observe(h.name, h.serving, h.good, s.now)
+ s.q.wantBuild(h.name, activator.CauseRelease, "b0", s.now)
+ }
+ return s
+}
+
+func (s *sim) failure(build, host string) {
+ if s.byName[host].served[build] {
+ panic("I3: a failure record against build " + build + ", which " + host + " serves")
+ }
+ failCohort(s.book, s.q, FailureRecord{Account: "bab", App: "blog", Build: build, Host: host, PromotedAt: s.now, At: s.now})
+ s.failed[build] = true // ground truth, whatever the book recorded
+}
+
+func (s *sim) pass() {
+ for _, h := range s.hosts {
+ if _, running := s.running[h.name]; !running {
+ s.q.observe(h.name, h.serving, h.good, s.now)
+ }
+ s.q.wantBuild(h.name, activator.CauseRelease, s.installed[h.ch], s.now)
+ }
+}
+
+func (s *sim) promote() {
+ s.n++
+ b := "b" + strconv.Itoa(s.n)
+ s.bad[b] = s.r.Intn(3) == 0
+ s.installed[s.r.Intn(2)] = b
+ s.pass()
+}
+
+func (s *sim) admit() {
+ view := map[string]string{}
+ for _, h := range s.hosts {
+ if s.r.Intn(10) == 0 {
+ continue // waking or stopping
+ }
+ view[h.name] = h.serving
+ }
+ for _, d := range s.q.admit(view, s.now) {
+ s.running[d.ID.Host] = d
+ }
+ if len(s.running) > s.q.concurrency || s.q.running > s.q.concurrency {
+ panic(fmt.Sprintf("I2: %d running with swap_concurrency %d", len(s.running), s.q.concurrency))
+ }
+ if s.q.running >= s.q.concurrency {
+ return
+ }
+ // I3, liveness: with capacity free, an awake instance's restart or config
+ // work onto the build it serves runs unless its own cohort's canary holds it.
+ for _, h := range s.hosts {
+ slot := s.q.hosts[h.name]
+ if _, seen := view[h.name]; !seen {
+ continue // waking or stopping this round: it keeps its place
+ }
+ if _, running := s.running[h.name]; running || !h.awake || slot.serving == "" {
+ continue
+ }
+ d, ok := s.q.pending(slot)
+ if !ok || d.ID.Build != slot.serving || s.now.Before(slot.configAfter) {
+ continue
+ }
+ c := s.q.cohorts[cohortKey(slot.serving, classRecover)]
+ if c == nil || c.canary == "" || c.settled {
+ panic("I3: " + h.name + "'s " + string(d.ID.Cause) + " onto the build it serves was held back")
+ }
+ }
+}
+
+func (s *sim) complete() {
+ if len(s.running) == 0 {
+ return
+ }
+ var host string
+ for host = range s.running {
+ break
+ }
+ d := s.running[host]
+ delete(s.running, host)
+ h := s.byName[host]
+ out := activator.SwapDone
+ switch {
+ case s.q.newest(host) != d.SchedRev:
+ out = activator.SwapAbandoned // Swap's re-check (spec §6.3)
+ case !h.served[d.ID.Build] && s.bad[d.ID.Build]:
+ out = activator.SwapFailed
+ case d.ID.Build == h.serving && s.r.Intn(8) == 0:
+ out = activator.SwapFailed // a restart or config attempt that fails
+ }
+ adopting := d.ID.Cause == activator.CauseRelease || d.ID.Cause == activator.CauseRetry
+ if out == activator.SwapDone && adopting && s.q.hosts[host].wantBuild != d.ID.Build {
+ panic("I4: " + host + " switched to " + d.ID.Build + " after its desired build moved to " + s.q.hosts[host].wantBuild)
+ }
+ transition := s.q.finish(d, out)
+ if out == activator.SwapDone {
+ h.serving, h.good = d.ID.Build, d.ID.Build
+ h.served[d.ID.Build] = true
+ s.q.observe(host, h.serving, h.good, s.now)
+ }
+ if transition {
+ s.failure(d.ID.Build, host)
+ }
+}
+
+func (s *sim) wake() {
+ h := s.hosts[s.r.Intn(len(s.hosts))]
+ if h.awake {
+ return
+ }
+ rt := registry.Route{Host: h.name, Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true,
+ Version: s.installed[h.ch], GoodVersion: h.good}
+ if h.good != "" {
+ rt.GoodDigest = "d-" + h.good
+ }
+ plan, _ := decideWake(wakeInputs{Route: rt, OptDir: "/opt/carlos", Policy: DefaultPolicy(), Book: s.book,
+ Demand: func(v, _ string) (string, string, error) { return "", "d-" + v, nil }})
+ boot := plan.Boot.Version
+ if h.good != "" && (plan.Refuse != "" || boot == "") {
+ panic("I1: " + h.name + " has good build " + h.good + " but its wake boots nothing: " + plan.Refuse)
+ }
+ if boot != "" && s.failed[boot] && !h.served[boot] {
+ panic("I1: " + h.name + " boots " + boot + ", which failed on this box")
+ }
+ if boot == "" {
+ return
+ }
+ if plan.FirstWake && s.bad[boot] {
+ if plan.Fallback == nil {
+ s.failure(boot, h.name)
+ return
+ }
+ h.awake, h.serving = true, plan.Fallback.Version
+ s.failure(boot, h.name)
+ } else {
+ h.awake, h.serving = true, boot
+ if plan.FirstWake {
+ h.served[boot], h.good = true, boot
+ }
+ }
+ s.q.observe(h.name, h.serving, h.good, s.now)
+}
+
+func (s *sim) sleep() {
+ h := s.hosts[s.r.Intn(len(s.hosts))]
+ if _, running := s.running[h.name]; running || !h.awake {
+ return
+ }
+ h.awake, h.serving = false, ""
+ s.q.observe(h.name, "", h.good, s.now)
+}
+
+func (s *sim) retry() {
+ for b := range s.failed {
+ if ok, _ := s.book.ClearOnPointer("bab", "blog", b, s.now.Add(time.Second)); ok {
+ s.q.unhalt(b)
+ delete(s.failed, b)
+ }
+ return
+ }
+}
+
+func (s *sim) run(steps int) {
+ for s.step = 0; s.step < steps; s.step++ {
+ s.now = s.now.Add(time.Second)
+ switch k := s.r.Intn(20); {
+ case k < 2:
+ s.promote()
+ case k < 7:
+ s.admit()
+ case k < 12:
+ s.complete()
+ case k < 14:
+ s.wake()
+ case k < 15:
+ s.sleep()
+ case k < 16:
+ h := s.hosts[s.r.Intn(len(s.hosts))]
+ if h.awake {
+ s.n++
+ s.q.wantRestart(h.name, "r"+strconv.Itoa(s.n), s.now)
+ }
+ case k < 17:
+ h := s.hosts[s.r.Intn(len(s.hosts))]
+ if h.awake {
+ s.n++
+ s.q.wantConfig(h.name, "c"+strconv.Itoa(s.n), s.now)
+ }
+ case k < 18:
+ s.retry()
+ default:
+ s.pass()
+ }
+ }
+}
+
+// TestSimInvariants drives the real queue, failure book, cohort-failure
+// transition and wake planner against a modelled activator. Seeds come from
+// ROLLOUT_SIM_SEEDS (default 200; the per-revision sweep is 3000).
+func TestSimInvariants(t *testing.T) {
+ seeds := 200
+ if v, err := strconv.Atoi(os.Getenv("ROLLOUT_SIM_SEEDS")); err == nil && v > 0 {
+ seeds = v
+ }
+ for seed := range int64(seeds) {
+ func() {
+ s := newSim(seed)
+ defer func() {
+ if r := recover(); r != nil {
+ t.Fatalf("seed %d step %d: %v", seed, s.step, r)
+ }
+ }()
+ s.run(400)
+ }()
+ }
+}
+```
+
+- [ ] **Step 2: Run it**
+
+Run: `uptime && nice -n 19 go test ./internal/rollout/ -run TestSimInvariants -count=1 -p 1`
+Expected: `ok github.com/carlosframework/platform/internal/rollout`. A failure names its seed and step: fix the code, never the invariant.
+
+- [ ] **Step 3: Run the per-revision sweep**
+
+Check `uptime` first; if the load average is above 12, wait. Then:
+
+```bash
+ROLLOUT_SIM_SEEDS=3000 nice -n 19 go test ./internal/rollout/ -run TestSimInvariants -count=1 -p 1 -timeout 30m
+```
+Expected: `ok`. One sweep per revision; deeper runs only when Paul asks.
+
+- [ ] **Step 4: Prove each invariant can fail (mutation checks)**
+
+Commit first (Step 5 below can come before this step), because the copy is made from `HEAD`. Each mutation edits a scratch copy under `$TMPDIR`, never the worktree, and each must make the simulation FAIL. Run them one at a time:
+
+```bash
+MUT="$TMPDIR/zdd5-mut"
+mutate() { # $1 name, $2 file, $3 sed program, $4 the invariant it should break
+ rm -rf "$MUT" && mkdir -p "$MUT" && git archive HEAD | tar -x -C "$MUT"
+ before=$(md5sum "$MUT/$2")
+ sed -i "$3" "$MUT/$2"
+ [ "$before" != "$(md5sum "$MUT/$2")" ] || { echo "$1: sed matched nothing, fix the program"; return 1; }
+ if (cd "$MUT" && GOFLAGS=-buildvcs=false nice -n 19 go test ./internal/rollout/ -run TestSimInvariants -count=1 -p 1 2>&1 | grep -q "$4"); then
+ echo "$1: killed ($4)"
+ else
+ echo "$1: SURVIVED"
+ fi
+}
+mutate M1 internal/rollout/queue.go 's/if q.running >= q.concurrency {/if false \&\& q.running >= q.concurrency {/' I2
+mutate M2 internal/rollout/queue.go 's/class := classRecover/class := classAdopt/' I3
+mutate M3 internal/rollout/queue.go 's/added, err = book.Record(rec)/added, err = true, nil/' I1
+mutate M4 internal/rollout/queue.go '/h.wantBuild, h.wantCause = build, cause/{n;s/h.buildRev = q.nextRev()/_ = q.nextRev()/}' I4
+mutate M5 internal/rollout/wakeplan.go 's/if in.Book.Blocks(acct, app, installed, host, false) {/if false \&\& in.Book.Blocks(acct, app, installed, host, false) {/' I1
+rm -rf "$MUT"
+```
+
+Expected: five lines ending `killed (I2)`, `killed (I3)`, `killed (I1)`, `killed (I4)`, `killed (I1)`. A `SURVIVED` line means the simulation cannot see that invariant break: strengthen the simulation (more steps, or the step that would exercise it) until it is killed. If a mutated copy does not compile, the sed changed the wrong line: fix the program, not the code.
+
+- [ ] **Step 5: Commit**
+
+```bash
+go vet ./internal/rollout/ && test -z "$(gofmt -l internal/rollout)"
+git add internal/rollout/sim_test.go
+git commit -m "zdd-5 task 16: invariant simulation for the scheduler"
+```
+
+---
+
+### Task 17: The gate, and handing the branch for review
+
+**Files:** none new.
+
+- [ ] **Step 1: Run the whole gate**
+
+```bash
+uptime
+mkdir -p /tmp/k
+go build ./... && go vet ./... && test -z "$(gofmt -l .)"
+TMPDIR=/tmp/k nice -n 19 go test ./... -count=1 -p 2
+GOOS=darwin GOARCH=arm64 go build ./... && GOOS=windows GOARCH=amd64 go build ./... && GOOS=linux GOARCH=arm64 go build ./...
+```
+Expected: no output from `gofmt`, every package `ok`, and three silent cross-builds.
+
+- [ ] **Step 2: Check the flag-off promise (cross-plan review focus 1)**
+
+```bash
+grep -rn 'CARLOS_SWAP\|rollout.Enabled' --include=*.go internal | grep -v _test.go
+```
+Expected: `rollout.Enabled` is read only in `internal/host/agent.go` (the `SetGenerations`/`Config.Swap` line) and `internal/host/rollout.go`, and `SwapEnv` only in `internal/rollout`. With the flag unset, `TestTheFlagOffArmsNothing` (Task 14) and `TestWithoutAPlannerTheWakeIsTodays` (Task 9) are the proof.
+
+- [ ] **Step 3: Push and describe**
+
+```bash
+git push origin zdd-5-scheduler
+amadan branch describe carlos/platform zdd-5-scheduler -body - <<'EOF'
+Zero-downtime deploys plan 5: the rollout scheduler, and the CARLOS_SWAP=overlap wiring.
+Flag off: byte-identical (Tasks 9 and 14 pin it). Flag on: per-app schedulers, canaries,
+swap_concurrency, failure records with promote --retry clearing, first-wake budget and
+fallback, config activation, the generations file root reads, stage-relay retransmission.
+Event wording is draft until plan 6's copy review; keep the flag off outside dev until then.
+Simulation: 3000 seeds, mutations M1-M5 killed.
+EOF
+amadan ci status carlos/platform -branch zdd-5-scheduler
+```
+Expected: `amadan ci status` exits 0 once CI finishes (2 while pending). Do not merge: the branch waits for review (a whole-branch review on the most capable model, per AGENTS.md's delegation policy), then squash and `amadan branch merge carlos/platform zdd-5-scheduler -expect <tip> -message "..."`.
+
+---
+
+## Self-review
+
+- **Spec coverage.** §4.1 policy keys: Task 3. §4.2 modes, canonical demand, retained identity, recovery mode with current policy, staging refusal and relay touch: Tasks 3, 8, 10. §6.1 one scheduler per app off the tick, per-instance desired builds, the four feeds: Tasks 5, 8, 14. §6.2 sleepers move at once, canary and settle, `swap_concurrency`, waking instances keep their place: Tasks 6, 8. §6.3 identity, dedupe, supersession, coalescing, newest check: Tasks 5, 8, 16. §6.4 first-wake canary, intentional stops, hold never falls back: Task 9. §6.5 socketless: Tasks 9, 11. §7.7 retransmission: Task 15. §7.9 coalescing and the drift check as a release attempt: Tasks 9, 11. §8.1 cohort-failure transition and its order, restart and config failures, delivered and active config with 1 to 30 minute backoff: Tasks 6, 7, 8. §8.2 the budget, the kill, the fallback, wake-time choice without a repoint: Tasks 9, 10. §8.3 hold failures held down: Tasks 8, 10. §8.4 records, scope, status, clearing by newer build, host retry, `promote --retry`: Tasks 4, 8, 13. §8.5 `GoodVersion` triple, expected-current writes, bootstrap, pins: Tasks 2, 8, 12. §9.4 events: Tasks 8, 9 (drafts). §10 converge leg, sweep, restart, config: Tasks 9, 14. §12 Scheduler, Config activation, Failure-hold scope, Failure without a repoint, Fallback, Bootstrap, Recovery mode, Lifecycle versus settle, Concurrency simulation: the tests named in those tasks.
+- **Not here:** surfaces, CLI and console doors, final event wording, the `retry` selector on the narrowed-restart log (plan 6); the swap mechanics (plan 3).
+- **Placeholder scan:** none. Steps that depend on plans 1 to 4 name the code to find and the exact change.
+- **Type consistency:** `Identity`, `dispatch`, `appQueue`, `FailureRecord`, `WakePlan`, `GenEvent`, `RolloutReport` and the `Deps` fields are used with one spelling throughout; Task 1 pins every outside name.
+- **Review Focus:** each of the five items has its test in the task named.