amadan
plans: Astra P2 findings fixed (3 partials completed, 8 new)
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+1146
−142
docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md+48 −3docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md+58 −5docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md+629 −61docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md+411 −73
CI passed — run details
diff --git a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md| index 866298032..ad0dfb9d8 100644 |
| --- a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md |
| +++ b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md |
| @@ -105,6 +105,7 @@ type SwapResult struct { |
| Gen Gen // the generation now current (on SwapDone) |
| Reason string |
| Drain time.Duration // added by plan 3: the retiring generation's drain, for the "swapped" event (spec §9.4) |
| + Mode SwapMode // added by plan 3: the mode the attempt actually ran in (a foreign row's overlap runs as hold, spec §2.9) |
| } |
| // Swap runs one attempt on an awake entry (spec §7.3–§7.5, §7.8). |
| @@ -325,6 +326,41 @@ names except where a finding said otherwise. |
| - **Test-type clash** (finding 16, plan 3 Task 4): plan 3's fixture is |
| `genServingProc`; plan 1's `servingProc` keeps its name. |
| +## Fixes for Astra's P2 review of the plans (2026-10-07) |
| + |
| +`CODEX-FINDINGS-ZDD-PLANS-P2.md`: three partial P1 fixes and eight new |
| +findings. |
| + |
| +- **P1 #1, completed** (plan 3 Task 2): the current generation and a failed |
| + wake's generation are held to the same proof as members. `stop()` and a |
| + pending stop's finish (`reapCurrentGen`) prove the child empty before the |
| + lease is released. `cleanupFailedWake` keeps the lease of a failed wake |
| + whose generation is unproven (`unprovenWakes`), refuses the host's next |
| + wake (`errWakeUnproven`), and the sweep retries. |
| +- **P1 #6, completed** (plan 5 Task 8): `done` builds the failure record |
| + before the lock (an unreadable row still records) and writes it with the |
| + halt in the critical section that frees the slot (`failCohort`). |
| +- **P1 #13, completed** (plan 1 Task 5): `dialGen` checks every new |
| + connection against `ServingGen`, so a pool can never reach a replacement |
| + on a shared foreign path. |
| +- **1** (plan 5 Tasks 13, 14): with the flag on and the store, key or |
| + failure book unavailable, `refuseManaged` refuses every managed wake, |
| + reports it (`RolloutReport.Unavailable`) and stands the bounce legs down. |
| +- **2** (plan 3 Task 9): a staged sidecar start leaves a populated host |
| + slice (the generation parent) alone, and is sent the generation's own |
| + tier. |
| +- **3** (plan 3 Task 7): a sidecar started for a retired generation is kept |
| + tracked and stopped synchronously before the replacement starts. |
| +- **4** (plan 3 Tasks 4, 6): the attempt's checks run again at the exec |
| + (overlap Step 3a; a hold's `pinCheck`) and before a hold's result counts. |
| +- **5, 7** (plan 5 Task 9): a first wake's observation is registered in the |
| + awake transition (`armFirstWatchLocked`); a planned worker is settled at |
| + its readiness. |
| +- **6** (plan 3 Task 6, plan 5 Tasks 8, 10): `SwapResult.Mode` is the mode |
| + that actually ran; `done` records a failed foreign hold as held down; the |
| + planner refuses a wake while the host's attempt is running. |
| +- **8** (plan 5 Task 12): the claimed-wake pin publication reads `e.plan`. |
| + |
| ## Cross-plan interface ledger |
| Every type, function, method or field that one plan defines and another |
| @@ -353,7 +389,7 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | `func (b *ExecBackend) prepareAt(rt, role instancectl.Role, sock string, gen uint64) (…)` (plan 2 adds `gen`) | P1 T2, P2 T11 | P3 T9 | |
| | `func (b *ExecBackend) normalizeSocketMode(rt registry.Route, sock string, p Proc)` | P1 T2 | P3 T1 (`GenSwitched`, `GenReady`) | |
| | `servingProc` (test type in `canonlink_test.go`) | P1 T4 | P1 only; plan 3's own fixture is `genServingProc` (P1 review finding 16) | |
| -| `GenPools` internals: `genPool{tr, gen}`, `dialGen`, `genMovedError` (each pool dials only its own generation's socket) | P1 T5 | P1 only; plans 1's Tasks 6 and 8 use only `NewGenPools`, `Retire`, `Len`, `AppSocketProxy` | |
| +| `GenPools` internals: `genPool{tr, gen}`, `dialGen` (dials only its own generation's socket, and checks each new connection against `ServingGen` so a shared foreign path never reaches another generation), `genMovedError` | P1 T5 | P1 only; plan 1's Tasks 6 and 8 use only `NewGenPools`, `Retire`, `Len`, `AppSocketProxy` | |
| | `func (b *ExecBackend) listenControl(rt registry.Route, control, envPath string) *edge.ControlListener` | P1 T2 | P3 T8 L3586, L3822 | |
| | `actEntry.gen Gen`; `wokenInstance.gen Gen` | P1 T3 | P3 T4 L1799, T6 L3048, T11 L5233; P5 T1 L188–189, T9 L4712, L5030 | |
| | `Service.gens bool` (read once in `New`); `func (a *Service) GenerationsOn() bool` | P1 T3 | P3 T2 L706, T4 L1761, T10 L4734, T11 L5236; P5 T1 L184, T9 L4340 | |
| @@ -384,11 +420,13 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | Interface | Defined | Used by | |
| |---|---|---| |
| | `Cause`, `SwapMode`, `Attempt`, `SwapOutcome` and their constants (index shared interfaces) | P3 T1 | P5 T1 L158–164, T5, T8 | |
| -| `type SwapResult struct{ Outcome SwapOutcome; Gen Gen; Reason string; Drain time.Duration }` | P3 T1 (Drain set in T4, T6) | P5 T1 L163, T8 L4113 (`textSwapped(build, res.Drain)`), T11 L5898 | |
| +| `type SwapResult struct{ Outcome SwapOutcome; Gen Gen; Reason string; Drain time.Duration; Mode SwapMode }` | P3 T1 (Drain set in T4, T6; Mode set by `Swap`, T6) | P5 T1 L163, T8 (`textSwapped(build, res.Drain)`; `done` records under `res.Mode` when set), T11 | |
| | `func (a *Service) Swap(ctx context.Context, at Attempt, newest func(host string) uint64) SwapResult` | P3 T4 | P5 T8 L2595, L3381 (the `Activator` interface), L4085 | |
| | `type SwapBackend interface{ StartStaged(ctx, rt, g Gen) (Gen, Proc, error); GenTier(rt, build string) cgroup.Limit; GenReady(rt, g Gen, p Proc); RaiseGenParent(ctx, rt, incumbent, successor cgroup.Limit) error; RestoreGenParent(ctx, rt, tier cgroup.Limit) error; GenSwitched(rt, g Gen, p Proc) error }`. `StartStaged` spawns and returns at once (no health wait); a foreign row spawns on its canonical socket | P3 T1 | P5 T1 L171, T9 (`planGenBackend` and `execStagedBackend` implement it) | |
| | `ExecBackend.TierOf func(rt, version string) cgroup.Limit`; `host.memTierOf` | P3 T1 | P3 only (wired in `agent.go` by plan 3) | |
| -| `type genKiller interface{ KillGen(ctx, rt registry.Route, n uint64) error }`; `func (a *Service) stopMember(e *actEntry, m *genMember) bool`; `genChildEmpty`; `reapExited` | P3 T2 | P5 T1 (surface pin), T9 (`killFailed`); plan 5 declares no `genKiller` of its own | |
| +| `type genKiller interface{ KillGen(ctx, rt registry.Route, n uint64) error }`; `func (a *Service) stopMember(e *actEntry, m *genMember) bool`; `func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool`; `reapExited`; `reapCurrentGen`; `unprovenWakes`, `errWakeUnproven`, `reapUnprovenWakesLocked`; `cleanupFailedWake(ctx, rt, w, lease)` (gains `rt`) | P3 T2 | P5 T1 (surface pin), T9 (`killFailed`); plan 5 declares no `genKiller` of its own | |
| +| `actEntry.pinCheck func() error`; `func (a *Service) startPinnedChecked(ctx, e *actEntry, rt, pin Gen) (Proc, string, Gen, error)` (doWake's pinned branch) | P3 T6 | P5 T9 (quotes the pinned branch; `startPinnedChecked` passes `time.Time{}` to `startPinned`) | |
| +| `actEntry.sidecarGen`, `wokenInstance.sidecarGen` (the generation a legacy sidecar runs); `startSidecarUnitFrom(ctx, rt, exe, tier cgroup.Limit)` | P3 T7, T9 | P3 only | |
| | `func (a *Service) foreignRow(host string) bool`: `Swap` runs a `ModeOverlap` attempt on a foreign row as a hold (its result reads as a hold's, `SwapDown` on failure) | P3 T6 | P5 T8 (`done` records the attempt's own mode, so no `HeldDown` for it) | |
| | `actEntry.settleCancel context.CancelFunc` (a hold canary's settle, cancelled by `beginStopLocked`) | P3 T6 | P5 T9 adds its own `firstWatch` cancel beside it in `beginStopLocked` | |
| | `instancectl.Client.StartSidecarExe(ctx, host string, memoryMax int64, x StagedExe) (string, error)`; `Server.ensureHostExeAt(host, x)` | P3 T9 | P3 only | |
| @@ -428,6 +466,9 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | `func bounceText(worker bool, cause Cause) string` (used by `BounceFor` and `ReplaceWorker`) | P5 T9 | P6 T14 L4471–4480 | |
| | `func (a *Service) ReplaceWorker(ctx, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) SwapResult`; `actEntry.attemptGate`, `attemptRestore`, `attemptVeto` (the attempt's checks, re-run by `respawnWorkerLocked` at the exec) | P5 T11 | P5 T8 (`Activator`, `fakeAct`), T11 (`executeWorker`) | |
| | `GenEventKind` `GenCancelled` (a first-wake observation ended with no verdict) | P5 T9 | P5 T13 (`OnGen` clears `observing`) | |
| +| `func (a *Service) armFirstWatchLocked(e *actEntry, w *wokenInstance)`; `actEntry.firstWatchCtx` (the observation registered in `wake()`'s awake transition) | P5 T9 | P5 only | |
| +| `status.RolloutReport.Unavailable string`; `refuseManaged`, `rolloutUnavailable`, `rolloutDown` (`internal/host`) | P5 T13, T14 | P6 may show `Unavailable` (not yet worded; copy review owns it) | |
| +| `func (m *Manager) attemptRunning(host string) bool` (`Plan` refuses while it is true) | P5 T10 | P5 only | |
| | `func plannedWorkerReady(ctx context.Context, p Proc) error` (spec §6.5 for planned socketless wakes) | P5 T9 | P5 only | |
| | `func (a *Service) persistGenerations() error`; `errGenerationsUnpublished` (a spawn whose build is not published does not run) | P5 T12 | P5 only | |
| | `status.RolloutReport`, `HostRollout`, `AppRollout`, `RolloutFailure`; `Attempt*` and `Reason*` words; `Probes.Rollout`, `BoxMetrics.Rollout`; `func reasonWord(text string) string`; `Deps.OnChange` | P5 T13 | P6 T2, T8 L2110–2183, T10 L2779, T11 L3670, L4022, T12 L4064 (through `rolloutView`), T14 L4553 | |
| @@ -441,3 +482,7 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | `release.ChannelPointer.PromotedAt`, `release.Channel`, `release.FileSHA256` | P5 T1, T8 | |
| | `host.BoxLabel() string` (`internal/host/metrics_loop.go`) | P5 T14 | |
| | `actEntry.respawnDone`, `actEntry.workerRespawning`, `procCrashed(p Proc) bool` | P5 T11 | |
| + |
| +- **Plan 6 copy review includes `RolloutReport.Unavailable`**, the reason |
| + managed starts are refused when, with the flag on, the box's required |
| + state can't be read (P2 #1). Plan 6's copy-review task adds the string. |
diff --git a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md| index 1c989087b..684f5c7c5 100644 |
| --- a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md |
| +++ b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md |
| @@ -2244,7 +2244,7 @@ git commit -m "zdd plan 1 task 4: the canonical socket path is a symlink to the |
| ``` |
| Test helpers (in `genproxy_test.go`, used by Tasks 6 and 7): `fakeGens` (`newFakeGens(on)`, `switchTo(host, GenInfo)`, `sleep(host)`, `askCount()`, fields `retired`, `onServing`), `genSockDir(t)`, `genApp` (fields `sock`, `opened`, `closed`, `srv`), `startGenApp(t, sock, h)`, `body(b, cc string) http.Handler`, `serve(t, h, url) *httptest.ResponseRecorder`. |
| - **A pool's connections are its own generation's** (spec §7.2, P1 review finding 13). The response's version header and the cache's generation must name the generation whose connection carried the request. So a generation's pool dials that generation's socket and nothing else (`dialGen`), and the response is attributed to the pool's generation, which is then exact. A switch between the pool choice and the dial is harmless: the retiring generation still accepts until its SIGTERM, which comes after the switch (plan 3's review focus 1). Only when the pool's own socket is gone *and* another generation is current does the dial give up with `genMovedError`, and `RoundTrip` sends the request once more on the current generation's pool if it can be replayed (no body, or `GetBody`). |
| + **A pool's connections are its own generation's** (spec §7.2, P1 review finding 13). The response's version header and the cache's generation must name the generation whose connection carried the request. So a generation's pool dials that generation's socket and nothing else (`dialGen`), and the response is attributed to the pool's generation, which is then exact. A switch between the pool choice and the dial is harmless: the retiring generation still accepts until its SIGTERM, which comes after the switch (plan 3's review focus 1). Only when the pool's own socket is gone *and* another generation is current does the dial give up with `genMovedError`, and `RoundTrip` sends the request once more on the current generation's pool if it can be replayed (no body, or `GetBody`). A foreign row's generations share one canonical path, so a path alone does not prove a generation (P2 review, partial P1 #13): every new connection is checked against `ServingGen` once connected, and one that may have reached another generation on the same path (another generation current on own's path, or none current) is closed and the request goes again on the current pool. An exclusive path (flat, nested) names its generation, so the retiring one is still accepted. |
| - [ ] **Step 1: Write the failing tests** |
| @@ -2541,6 +2541,38 @@ func TestAppSocketProxyAttributesTheConnectionsGeneration(t *testing.T) { |
| rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n) |
| } |
| }) |
| + // A foreign row's generations share one path. Generation 1's pool, chosen |
| + // an instant before the hold, dials that path after generation 2 has |
| + // bound it: the connection reaches 2, and a dial that trusts the path |
| + // stamps 2's response as 1 and keeps the connection in 1's pool. |
| + t.Run("a foreign row's path now belongs to the replacement", func(t *testing.T) { |
| + dir := genSockDir(t) |
| + path := filepath.Join(dir, "canonical.sock") |
| + a := startGenApp(t, path, body("A", "")) |
| + src := newFakeGens(true) |
| + src.switchTo(host, GenInfo{Socket: path, N: 1, Build: "v1"}) |
| + var once sync.Once |
| + var b *genApp |
| + src.onServing = func() { |
| + once.Do(func() { |
| + a.srv.Close() // the hold stops 1; its socket file goes |
| + b = startGenApp(t, path, body("B", "")) |
| + src.switchTo(host, GenInfo{Socket: path, N: 2, Build: "v2"}) |
| + }) |
| + } |
| + p := appSocketProxy(host, src, NewGenPools()) |
| + r, _ := withOutcome(httptest.NewRequest("GET", "http://inst.example.com/", nil)) |
| + rec := httptest.NewRecorder() |
| + p.ServeHTTP(rec, r) |
| + _, n, _ := servedGen(r) |
| + if rec.Body.String() != "B" || rec.Header().Get("X-Carlos-Version") != "v2" || n != 2 { |
| + t.Fatalf("served %q, version %q, generation %d: want B, v2, 2 (never attributed to the old generation)", |
| + rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n) |
| + } |
| + if b == nil || b.opened.Load() < 1 { |
| + t.Fatal("the replacement was never reached") |
| + } |
| + }) |
| } |
| // The generation that carried a request is recorded on its outcome, for the |
| @@ -2718,9 +2750,11 @@ func (e *genMovedError) Error() string { |
| return fmt.Sprintf("edge: generation moved to %d while dialling the one before it", e.to.N) |
| } |
| -// dialGen is a generation pool's dial: own's socket and nothing else, so |
| -// every connection in the pool is own's and a response through it is |
| -// attributed to own exactly (spec §7.2). The retiring generation still |
| +// dialGen is a generation pool's dial: own's socket and nothing else, and |
| +// a connection proven to be own's (a shared foreign path is checked |
| +// against the current generation once connected), so every connection in |
| +// the pool is own's and a response through it is attributed to own exactly |
| +// (spec §7.2). The retiring generation still |
| // accepts until its SIGTERM, which comes after the switch, so a request |
| // that chose own's pool an instant before the switch is served there. A |
| // retryable failure is retried inside the usual window while own is still |
| @@ -2732,7 +2766,26 @@ func dialGen(ctx context.Context, d *net.Dialer, src GenSource, host string, own |
| for { |
| conn, err := unixdial.Checked(ctx, d, own.Socket) |
| if err == nil { |
| - return conn, nil |
| + // A foreign row's generations share one canonical path (plan 1's |
| + // reading), so a path is not proof of a generation. Every new |
| + // connection is checked against the current generation after it |
| + // connects: if another generation now owns own's path, or none is |
| + // current, the connection may have reached the replacement, so it |
| + // is closed and the request goes again on the current pool. An |
| + // exclusive generation path (flat, nested) names its generation, |
| + // so the retiring one is still own and still accepts. |
| + cur, ok := src.ServingGen(host) |
| + if ok && cur.N == own.N { |
| + return conn, nil |
| + } |
| + if ok && cur.Socket != own.Socket { |
| + return conn, nil // own's exclusive socket: still own, retiring |
| + } |
| + conn.Close() |
| + if ok { |
| + return nil, &genMovedError{to: cur} |
| + } |
| + err = errNoGeneration |
| } |
| if ctx.Err() != nil || !isRetryableDialErr(err) || time.Now().After(deadline) { |
| return nil, err |
diff --git a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md| index 5fe09af8f..8ed44204f 100644 |
| --- a/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md |
| +++ b/docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md |
| @@ -127,7 +127,7 @@ The cross-plan items this plan owns (index review focus 2 and 4) are Task 12. |
| type SwapMode string; const ModeOverlap, ModeHold SwapMode |
| type Attempt struct{ Host string; Cause Cause; Build, StagedPath, Digest, Demand string; Mode SwapMode; ConfigRev, RestartID string; SchedRev uint64; Canary bool } |
| type SwapOutcome string; const SwapDone, SwapFailed, SwapAbandoned, SwapDown SwapOutcome |
| - type SwapResult struct{ Outcome SwapOutcome; Gen Gen; Reason string; Drain time.Duration } |
| + type SwapResult struct{ Outcome SwapOutcome; Gen Gen; Reason string; Drain time.Duration; Mode SwapMode } // Mode: the mode the attempt actually ran in (Task 6 sets it) |
| // Gen gains (index: "a plan may add fields"): Tier cgroup.Limit, the artifact's own tier, captured at attempt time |
| type SwapBackend interface { |
| // StartStaged spawns and returns at once: no health wait inside. The caller owns the health deadline and the termination. |
| @@ -555,6 +555,11 @@ type SwapResult struct { |
| // proven exit (an overlap), or the old generation's stop (a hold). Zero |
| // when nothing retired. The "swapped" event reports it (spec §9.4). |
| Drain time.Duration |
| + // Mode is the mode the attempt actually ran in. It differs from |
| + // Attempt.Mode when an overlap attempt on a foreign-layout row ran as a |
| + // hold (spec §2.9): the scheduler records a failure under this mode, so |
| + // a failed hold is held down (spec §6.4, §8.3). Swap sets it. |
| + Mode SwapMode |
| } |
| ``` |
| @@ -992,7 +997,9 @@ git commit -m "zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged |
| - `wake()`, lines 933–938 and 996; |
| - `stop()`, after line 2175 and at line 2268; |
| - `finishPendingStops`, lines 2301–2328; |
| - - `Shutdown`, lines 2921–2936. |
| + - `Shutdown`, lines 2921–2936; |
| + - `doWake`'s four `cleanupFailedWake` calls and its head (the unproven-wake refusal). |
| +- Modify: `internal/activator/maintenance.go` (`cleanupFailedWake`) |
| - Test: `internal/activator/opset_test.go` |
| **Interfaces:** |
| @@ -1011,12 +1018,20 @@ git commit -m "zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged |
| func (a *Service) isMember(e *actEntry, m *genMember) bool |
| func (a *Service) joinOps(e *actEntry) bool |
| func (a *Service) stopMember(e *actEntry, m *genMember) bool // leader exit AND an empty generation child |
| - func (a *Service) genChildEmpty(e *actEntry, m *genMember) bool |
| + func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool |
| + func (a *Service) reapCurrentGen(e *actEntry) // a pending stop's current generation, proven empty before finishStop |
| + type unprovenWake struct{ rt registry.Route; gen Gen; proc Proc; lease Lease; rep Rep; busy bool } |
| + var errWakeUnproven error |
| + func (a *Service) reapUnprovenWakesLocked() |
| + // actEntry gains genReaping bool; Service gains unprovenWakes map[string]*unprovenWake |
| + // maintenance.go: cleanupFailedWake(ctx, rt registry.Route, w *wokenInstance, lease Lease) (gains rt) |
| func (a *Service) dropMember(e *actEntry, m *genMember) bool |
| func (a *Service) leaveSet(e *actEntry, m *genMember) |
| func (a *Service) reapMembersLocked(e *actEntry) |
| func (a *Service) reapExited(e *actEntry, m *genMember) |
| ``` |
| + The same proof covers the **current** generation (P2 review, partial P1 #1): `stop()` and a pending stop's finish (`reapCurrentGen`) prove its child empty before the final checkpoint and the lease release, and `cleanupFailedWake` proves a failed wake's generation empty before it gives the lease back. Unproven, the lease is kept (a pending stop, or `unprovenWakes` for a failed wake, which also refuses the host's next wake) and the sweep retries. |
| + |
| A generation is terminated when its leader has exited **and** its generation child cgroup is proven empty (spec §7.3, §7.4). The flat `childProc.Stop` signals only the leader (`backend_exec.go`, `func (p *childProc) Stop`), and plan 2's exit goroutine only tries `RemoveChild`, which the kernel refuses while a forked descendant lives. So after the graceful drain, `stopMember` calls plan 2's `KillGen` for the member's generation, which kills what is left of that child and returns once it is empty. A failed kill keeps the member in the set, unproven, exactly as `ErrStillRunning` does. |
| - [ ] **Step 1: Write the failing tests** |
| @@ -1283,6 +1298,42 @@ func TestMemberStopWithoutContainmentAcceptsTheLeaderExit(t *testing.T) { |
| } |
| } |
| +// The current generation is held to the same proof (spec §7.4): a terminal |
| +// stop whose main process has exited keeps the lease while that |
| +// generation's child is not proven empty, and finishes once it is. A stop |
| +// that accepts the leader's exit alone releases the lease over a surviving |
| +// descendant. This is the witness for mutation M8. |
| +func TestSwapWitnessStopProvesTheCurrentGenerationEmpty(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + kb := &killGenBackend{fakeBackend: x.b, err: errors.New("the child still holds a process")} |
| + x.a.backend = kb |
| + x.a.mu.Lock() |
| + x.a.entries[x.host].gen.N = 1 << 41 // the woken generation, as plan 1's wake records it |
| + x.a.mu.Unlock() |
| + rep := x.b.lastRep() |
| + x.a.Bounce(x.host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + x.a.mu.Lock() |
| + defer x.a.mu.Unlock() |
| + e := x.a.entries[x.host] |
| + return e != nil && e.pendingStop |
| + }) |
| + if !x.b.lastProc().isStopped() { |
| + t.Fatal("the graceful stop never ran") |
| + } |
| + x.a.sweepOnce() // retries the proof, which still fails |
| + time.Sleep(50 * time.Millisecond) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while the current generation's child still held a process") |
| + } |
| + kb.setErr(nil) |
| + x.a.sweepOnce() |
| + waitFor(t, "the lease release", rep.isClosed) |
| + if k := kb.killed(); len(k) == 0 || k[len(k)-1] != 1<<41 { |
| + t.Fatalf("KillGen calls = %v, want the current generation's", k) |
| + } |
| +} |
| + |
| // With swaps on, Shutdown covers a waking entry (spec §7.4): it cancels the |
| // wake and waits for it. Its child is stopped and its lease given back |
| // before Shutdown returns. |
| @@ -1352,7 +1403,7 @@ func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) { |
| - [ ] **Step 2: Run the tests to verify they fail** |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1` |
| Expected: build failure: `undefined: genRole`, `undefined: genMember`, `x.a.entries[x.host].ops undefined`. |
| - [ ] **Step 3: Write `opset.go`** |
| @@ -1503,7 +1554,10 @@ func (a *Service) stopMember(e *actEntry, m *genMember) bool { |
| if !m.proc.Exited() { |
| return false |
| } |
| - return a.genChildEmpty(e, m) // swapsim:M7 |
| + a.mu.Lock() |
| + rt := e.route |
| + a.mu.Unlock() |
| + return a.genChildEmpty(rt, m) // swapsim:M7 |
| } |
| // genChildEmpty proves m's generation child empty, after its leader has |
| @@ -1517,7 +1571,7 @@ func (a *Service) stopMember(e *actEntry, m *genMember) bool { |
| // the host cgroup, and the terminal stop's own stop evidence on that cgroup |
| // is the proof (plan 2, "stop evidence on the parent"). Any other error is |
| // an unproven termination. |
| -func (a *Service) genChildEmpty(e *actEntry, m *genMember) bool { |
| +func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool { |
| if _, ok := m.proc.(*claimProcess); ok { |
| // A claims-on generation (Task 10): claimProcess.Stop already ran |
| // KillAndEmpty on its own child and succeeded, or Stop failed and |
| @@ -1528,9 +1582,6 @@ func (a *Service) genChildEmpty(e *actEntry, m *genMember) bool { |
| if !ok || m.gen.N == 0 { |
| return true |
| } |
| - a.mu.Lock() |
| - rt := e.route |
| - a.mu.Unlock() |
| ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| defer cancel() |
| err := k.KillGen(ctx, rt, m.gen.N) |
| @@ -1623,7 +1674,10 @@ func (a *Service) reapMembersLocked(e *actEntry) { |
| // the set only once its generation child is proven empty. A failed proof |
| // leaves it marked, and the next sweep tries the kill again. |
| func (a *Service) reapExited(e *actEntry, m *genMember) { |
| - if a.genChildEmpty(e, m) { |
| + a.mu.Lock() |
| + rt := e.route |
| + a.mu.Unlock() |
| + if a.genChildEmpty(rt, m) { |
| a.leaveSet(e, m) |
| return |
| } |
| @@ -1631,6 +1685,81 @@ func (a *Service) reapExited(e *actEntry, m *genMember) { |
| m.reaping, m.retriedAt = false, a.now() |
| a.mu.Unlock() |
| } |
| + |
| +// The CURRENT generation is held to the same proof (spec §7.4: "each is |
| +// tracked until its termination is proven"). stop() and a pending stop's |
| +// finish prove its child empty before the final checkpoint and the lease |
| +// release, and a failed wake proves its generation's child empty before it |
| +// gives the lease back. |
| + |
| +// reapCurrentGen finishes a pending stop whose main process has exited, once |
| +// the current generation's child is proven empty. A failed proof leaves the |
| +// stop pending, and the next sweep tries again. |
| +func (a *Service) reapCurrentGen(e *actEntry) { |
| + a.mu.Lock() |
| + rt, g, p := e.route, e.gen, e.proc |
| + a.mu.Unlock() |
| + if !a.genChildEmpty(rt, &genMember{gen: g, proc: p}) { |
| + a.mu.Lock() |
| + e.genReaping = false |
| + a.mu.Unlock() |
| + return |
| + } |
| + a.mu.Lock() |
| + e.genReaping, e.pendingStop = false, false |
| + outcome := e.stopOutcome |
| + a.mu.Unlock() |
| + if outcome == nil { |
| + outcome = &stopObservation{} |
| + } |
| + a.finishPendingStop(e, outcome) |
| +} |
| + |
| +// unprovenWake is a failed wake whose generation's child could not be |
| +// proven empty (spec §7.4). Its lease (and replica, if it got one) is kept |
| +// until the sweep proves the child empty; until then no new wake of the |
| +// host starts. |
| +type unprovenWake struct { |
| + rt registry.Route |
| + gen Gen |
| + proc Proc |
| + lease Lease |
| + rep Rep |
| + busy bool |
| +} |
| + |
| +// errWakeUnproven refuses a wake while the host's last failed wake is not |
| +// proven stopped. |
| +var errWakeUnproven = errors.New("the last failed wake's generation is not proven stopped") |
| + |
| +// reapUnprovenWakesLocked retries each unproven failed wake's proof and, |
| +// once it holds, gives the lease back. The caller holds a.mu. |
| +func (a *Service) reapUnprovenWakesLocked() { |
| + for host, u := range a.unprovenWakes { |
| + if u.busy { |
| + continue |
| + } |
| + u.busy = true |
| + go func(host string, u *unprovenWake) { |
| + proven := a.genChildEmpty(u.rt, &genMember{gen: u.gen, proc: u.proc}) |
| + if proven { |
| + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| + if u.rep != nil { |
| + _ = u.rep.Close(ctx) |
| + } else if u.lease != nil { |
| + _ = u.lease.Release(ctx) |
| + } |
| + cancel() |
| + } |
| + a.mu.Lock() |
| + if proven { |
| + delete(a.unprovenWakes, host) |
| + } |
| + u.busy = false |
| + a.mu.Unlock() |
| + }(host, u) |
| + } |
| +} |
| ``` |
| - [ ] **Step 4: Wire the set into `activator.go`** |
| @@ -1691,9 +1820,29 @@ At the end of `stop()`, replace `a.finishStop(e, outcome)` (line 2268) with: |
| a.mu.Unlock() |
| return |
| } |
| + if a.gens && e.proc != nil && e.proc.Exited() { |
| + // The current generation is a generation like any other (spec |
| + // §7.4): its leader's exit is not its termination until its child |
| + // is empty. Unproven, the stop stays pending with the lease held, |
| + // exactly as for ErrStillRunning, and the sweep retries the proof. |
| + a.mu.Lock() |
| + rt, g := e.route, e.gen |
| + a.mu.Unlock() |
| + if !a.genChildEmpty(rt, &genMember{gen: g, proc: e.proc}) { // swapsim:M8 |
| + log.Printf("activator: stop %s: generation %d's child is not proven empty; keeping the lease until it is", e.host, g.N) |
| + a.mu.Lock() |
| + e.pendingStop = true |
| + e.stopOutcome = outcome |
| + e.stopRetriedAt = a.now() |
| + a.mu.Unlock() |
| + return |
| + } |
| + } |
| a.finishStop(e, outcome) |
| ``` |
| +Add to `actEntry`, beside `ops`: `genReaping bool // reapCurrentGen is running for this pending stop; written under a.mu`. Add to `Service`: `unprovenWakes map[string]*unprovenWake // failed wakes whose generation is not proven stopped; under a.mu` (made lazily). |
| + |
| Replace the head of `finishPendingStops`'s loop (lines 2302–2305): |
| ```go |
| @@ -1722,6 +1871,59 @@ with |
| go a.finishStop(e, outcome) |
| continue |
| } |
| + if a.gens && e.proc.Exited() { |
| + // The main process has exited: its generation's child must be |
| + // proven empty before the stop finishes (spec §7.4). |
| + if !e.genReaping { |
| + e.genReaping = true |
| + go a.reapCurrentGen(e) |
| + } |
| + continue |
| + } |
| +``` |
| + |
| +and add, as the first statement of `finishPendingStops` (before the loop), `a.reapUnprovenWakesLocked()`. |
| + |
| +In `maintenance.go`, give `cleanupFailedWake` the route, `func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *wokenInstance, lease Lease)`, and pass `rt` at its four call sites in `doWake`. In its body, replace |
| + |
| +```go |
| + if w.proc != nil { |
| + err := w.proc.Stop(stopCtx) |
| + w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited() |
| + } |
| +``` |
| +with |
| +```go |
| + if w.proc != nil { |
| + err := w.proc.Stop(stopCtx) |
| + w.stopOutcome.mainStop, w.stopOutcome.mainExited = err == nil, w.proc.Exited() |
| + if a.gens && w.stopOutcome.mainExited && !a.genChildEmpty(rt, &genMember{gen: w.gen, proc: w.proc}) { |
| + // The failed generation's child is not proven empty (spec |
| + // §7.4): ownership and the lease are kept, the sweep retries |
| + // the proof (reapUnprovenWakesLocked), and no new wake of the |
| + // host starts until it holds. |
| + w.stopOutcome.mainExited = false |
| + log.Printf("activator: %s: failed wake's generation %d is not proven stopped; keeping the lease until it is", rt.Host, w.gen.N) |
| + a.mu.Lock() |
| + if a.unprovenWakes == nil { |
| + a.unprovenWakes = make(map[string]*unprovenWake) |
| + } |
| + a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, proc: w.proc, lease: lease, rep: w.rep} |
| + a.mu.Unlock() |
| + return |
| + } |
| + } |
| +``` |
| + |
| +and at the top of `doWake`, after the layout refusal (the `if c, ok := a.backend.(Classifier)` block), add: |
| + |
| +```go |
| + a.mu.Lock() |
| + _, unproven := a.unprovenWakes[host] |
| + a.mu.Unlock() |
| + if unproven { |
| + return w, fmt.Errorf("%s: refusing to wake: %w", host, errWakeUnproven) |
| + } |
| ``` |
| In `Shutdown`, replace lines 2921–2936 (from `var waits []chan struct{}` through `a.mu.Unlock()`) with: |
| @@ -1768,7 +1970,7 @@ In `Shutdown`, replace lines 2921–2936 (from `var waits []chan struct{}` throu |
| - [ ] **Step 5: Run the tests to verify they pass** |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1 -v 2>&1 | tail -20` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1 -v 2>&1 | tail -20` |
| Expected: every subtest `--- PASS`, then `ok`. |
| - [ ] **Step 6: Run the whole package (the flag-off paths are unchanged)** |
| @@ -1779,7 +1981,7 @@ Expected: `ok`. Every existing test passes without edits. |
| - [ ] **Step 7: Commit** |
| ```bash |
| -git add internal/activator/opset.go internal/activator/activator.go internal/activator/opset_test.go |
| +git add internal/activator/opset.go internal/activator/activator.go internal/activator/maintenance.go internal/activator/opset_test.go |
| git commit -m "zdd plan 3 task 2: the entry's operation set, the barrier in stop(), Shutdown joins waking entries" |
| ``` |
| @@ -2124,9 +2326,15 @@ type fakeSwapBackend struct { |
| // Task 7: legacy sidecars started from a generation. |
| sidecarGens []Gen |
| sidecarGenGate chan struct{} |
| + // sidecarStopGate, when set, is installed on every sidecar started from |
| + // then on: its Stop blocks until the gate closes. |
| + sidecarStopGate chan struct{} |
| // tiers is each build's own tier (GenTier); a build not here is the |
| // default tier. |
| tiers map[string]cgroup.Limit |
| + // onRaise, when set, runs inside RaiseGenParent: a test's window between |
| + // an attempt's admission and its exec. |
| + onRaise func() |
| } |
| func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen Gen) (Gen, Proc, error) { |
| @@ -2178,7 +2386,11 @@ func (g *fakeSwapBackend) RaiseGenParent(_ context.Context, _ registry.Route, in |
| g.record(fmt.Sprintf("raise:%d+%d", inc.MemoryMax, succ.MemoryMax)) |
| g.gmu.Lock() |
| g.raises++ |
| + hook := g.onRaise |
| g.gmu.Unlock() |
| + if hook != nil { |
| + hook() |
| + } |
| return nil |
| } |
| @@ -2573,6 +2785,23 @@ func TestSwapWitnessNoAdmissionOntoAStoppingEntry(t *testing.T) { |
| } |
| } |
| +// Spec §6.3, §7.4: the checks run again at the exec, not only at the |
| +// admission. A newer attempt that lands while the parent is being raised |
| +// (or, with plan 5, while the pins are being published) means nothing |
| +// execs, and the raise is undone. Checked only at the admission, the |
| +// superseded build execs and is abandoned only at the switch. |
| +func TestSwapOverlapReChecksAtTheExec(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.b.onRaise = func() { f.rev.Add(1) } |
| + res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| + if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| + t.Fatalf("Swap = %+v, want abandoned: superseded", res) |
| + } |
| + if gens, raises, restores := f.b.calls(); gens != 0 || raises != restores { |
| + t.Fatalf("spawned %d, raises %d, restores %d: want nothing exec'd and the raise undone", gens, raises, restores) |
| + } |
| +} |
| + |
| // A newer attempt that lands during the health check is seen at the switch. |
| // The successor is abandoned and stopped, and the incumbent keeps serving. |
| func TestSwapOverlapSupersededAtTheSwitch(t *testing.T) { |
| @@ -2601,7 +2830,8 @@ func TestSwapClaimsOn(t *testing.T) { |
| {"selector refuses", func() hostwrite.Selector { return &swapSelector{err: hostwrite.ErrUnavailable} }, 0}, |
| {"no selector", func() hostwrite.Selector { return nil }, 0}, |
| {"refused before the spawn", func() hostwrite.Selector { return &swapSelector{work: &swapWork{denyAt: 1}} }, 0}, |
| - {"refused at the switch", func() hostwrite.Selector { return &swapSelector{work: &swapWork{denyAt: 2}} }, 1}, |
| + {"refused at the exec", func() hostwrite.Selector { return &swapSelector{work: &swapWork{denyAt: 2}} }, 0}, |
| + {"refused at the switch", func() hostwrite.Selector { return &swapSelector{work: &swapWork{denyAt: 3}} }, 1}, |
| } { |
| t.Run(tc.name, func(t *testing.T) { |
| f := newSwapFixture(t) |
| @@ -2629,8 +2859,8 @@ func TestSwapClaimsOn(t *testing.T) { |
| if res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest); res.Outcome != SwapDone { |
| t.Fatalf("Swap = %+v", res) |
| } |
| - if w.withCalls() != 2 { |
| - t.Fatalf("With calls = %d, want 2 (before the spawn and at the switch)", w.withCalls()) |
| + if w.withCalls() != 3 { |
| + t.Fatalf("With calls = %d, want 3 (the admission, the exec and the switch)", w.withCalls()) |
| } |
| }) |
| } |
| @@ -2804,6 +3034,26 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| e.ops.raised = true |
| a.mu.Unlock() |
| + // Step 3a: the §7.4 checks again, at the exec (spec §6.3, §7.4). The |
| + // raise (and plan 5's pin publication, just above this step) took time: |
| + // a newer attempt, a halted cohort, a lost lease or a fence since |
| + // admission means nothing execs. Claims on, this runs inside the |
| + // selected work as the admission did. |
| + if err := a.admit(ctx, work, func() error { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + if err := a.swapChecksLocked(e, at, newest); err != nil { |
| + return err |
| + } |
| + if e.ops.members[m.gen.N] != m || mctx.Err() != nil { |
| + return errSwapStopped |
| + } |
| + return nil |
| + }); err != nil { |
| + a.dropMember(e, m) // restores the raised parent (leaveSet) |
| + return abandon(err) |
| + } |
| + |
| // Step 3: exec the attempt's staged build by path. The backend reports |
| // what it ran, and anything but the attempt's choice is refused. |
| g, proc, spawnErr := gb.StartStaged(mctx, rt, m.gen) |
| @@ -3204,6 +3454,8 @@ git commit -m "zdd plan 3 task 5: canary settle; an intentional stop cancels it |
| func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) uint64) SwapResult |
| func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) |
| func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) |
| + func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen) (Proc, string, Gen, error) // doWake's pinned branch: pinCheck, then startPinned |
| + // actEntry gains pinCheck func() error (the hold's attempt checks, re-run at the exec); swapHold re-runs swapChecksLocked before SwapDone |
| func (a *Service) lifeDone() <-chan struct{} |
| func (a *Service) foreignRow(host string) bool // Swap runs an overlap attempt on a foreign row as a hold |
| var errHoldNeedsPins error |
| @@ -3223,6 +3475,7 @@ import ( |
| "errors" |
| "os/exec" |
| "strings" |
| + "sync" |
| "testing" |
| "time" |
| @@ -3373,6 +3626,95 @@ func TestSwapHoldFailureIsDown(t *testing.T) { |
| } |
| } |
| +// acquireHookBackend runs hook inside Acquire: the window between a hold's |
| +// re-check at its stop and its pinned wake's exec. |
| +type acquireHookBackend struct { |
| + *fakeSwapBackend |
| + hook func() |
| +} |
| + |
| +func (b *acquireHookBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) { |
| + b.hook() |
| + return b.fakeSwapBackend.Acquire(ctx, rt) |
| +} |
| + |
| +// Spec §6.3, §7.4, §8.1: a hold's attempt checks run again at the pinned |
| +// wake's exec and once more before its result counts. Checked only when |
| +// the stop finished, a newer attempt (or a halted cohort, which bumps the |
| +// revision) landing during the wake still exec'd the superseded build and |
| +// answered SwapDone, which records it good. |
| +func TestSwapHoldReChecksAtTheExecAndBeforeTheResult(t *testing.T) { |
| + t.Run("superseded before the exec", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + var once sync.Once |
| + f.a.backend = &acquireHookBackend{fakeSwapBackend: f.b, hook: func() { once.Do(func() { f.rev.Add(1) }) }} |
| + res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) |
| + if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| + t.Fatalf("Swap = %+v, want abandoned: superseded", res) |
| + } |
| + if gens, _, _ := f.b.calls(); gens != 0 { |
| + t.Fatalf("the superseded hold build was exec'd (%d)", gens) |
| + } |
| + }) |
| + t.Run("superseded before the result", func(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.genHealth = func(context.Context, Gen) error { f.rev.Add(1); return nil } |
| + res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest) |
| + if res.Outcome != SwapAbandoned || !strings.Contains(res.Reason, errSwapSuperseded.Error()) { |
| + t.Fatalf("Swap = %+v, want abandoned: a superseded result is never done", res) |
| + } |
| + }) |
| +} |
| + |
| +// killingSwapBackend is fakeSwapBackend plus plan 2's KillGen. fail decides, |
| +// per generation, whether the child is still occupied. |
| +type killingSwapBackend struct { |
| + *fakeSwapBackend |
| + kmu sync.Mutex |
| + fail func(n uint64) bool |
| +} |
| + |
| +func (k *killingSwapBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error { |
| + k.kmu.Lock() |
| + defer k.kmu.Unlock() |
| + if k.fail != nil && k.fail(n) { |
| + return errors.New("the child still holds a process") |
| + } |
| + return nil |
| +} |
| + |
| +func (k *killingSwapBackend) setFail(f func(uint64) bool) { |
| + k.kmu.Lock() |
| + k.fail = f |
| + k.kmu.Unlock() |
| +} |
| + |
| +// A failed wake gives its lease back only once its generation's child is |
| +// proven empty (spec §7.4). Until then the host takes no new wake, and the |
| +// sweep retries the proof. cleanupFailedWake that accepts the leader's exit |
| +// releases the lease over a surviving descendant and lets a new wake start |
| +// beside it. |
| +func TestSwapHoldFailedWakeKeepsTheLeaseUntilItsChildIsEmpty(t *testing.T) { |
| + f := newSwapFixture(t) |
| + kb := &killingSwapBackend{fakeSwapBackend: f.b} |
| + kb.setFail(func(n uint64) bool { return n != f.incumbent.N }) // the hold's new generation only |
| + f.a.backend = kb |
| + f.a.genHealth = func(context.Context, Gen) error { return errors.New("unhealthy: HTTP 500") } |
| + before := f.b.releaseCount() |
| + if res := f.a.Swap(context.Background(), f.attempt(ModeHold), f.newest); res.Outcome != SwapDown { |
| + t.Fatalf("Swap = %+v, want down", res) |
| + } |
| + if got := f.b.releaseCount(); got != before+1 { |
| + t.Fatalf("lease releases %d -> %d: want the old generation's only, the failed wake's kept", before, got) |
| + } |
| + if _, err := f.a.beginRequest(context.Background(), testRoute(f.host)); err == nil || !strings.Contains(err.Error(), errWakeUnproven.Error()) { |
| + t.Fatalf("a new wake = %v, want refused while the failed generation is unproven", err) |
| + } |
| + kb.setFail(nil) |
| + f.a.sweepOnce() |
| + waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 }) |
| +} |
| + |
| // The re-check before the spawn happens when the stop finishes. A fence, a |
| // newer attempt or the agent's shutdown abandons the hold, and nothing is |
| // spawned. |
| @@ -3472,7 +3814,7 @@ func TestStartAdmittedRefusesAPinMismatch(t *testing.T) { |
| - [ ] **Step 2: Run the tests to verify they fail** |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessHold|TestSwapHold|TestStartAdmittedRefusesAPin|TestSwapOverlapOnAForeignRow' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessHold|TestSwapHold|TestStartAdmittedRefusesAPin|TestSwapOverlapOnAForeignRow' -count=1` (`TestSwapHold` covers `TestSwapHoldReChecksAtTheExecAndBeforeTheResult` and `TestSwapHoldFailedWakeKeepsTheLease…`) |
| Expected: build failure: `undefined: errHoldNeedsPins`, `unknown field Pin in struct literal of type StartPolicy`. |
| - [ ] **Step 3: Write `swaphold.go`** |
| @@ -3588,24 +3930,38 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) |
| g, werr, proc := hw.next.gen, hw.next.err, hw.next.proc |
| a.mu.Unlock() |
| switch { |
| - case werr == nil && at.Canary: |
| - // A canary must survive the settle before it counts (spec §6.2, |
| - // §8.5). Plan 5 does not observe a pinned wake itself. |
| - r := observeSettle(sctx, proc, swapSettle) |
| + case werr == nil: |
| + if at.Canary { |
| + // A canary must survive the settle before it counts (spec §6.2, |
| + // §8.5). Plan 5 does not observe a pinned wake itself. |
| + r := observeSettle(sctx, proc, swapSettle) |
| + a.mu.Lock() |
| + hw.next.settleCancel = nil |
| + a.mu.Unlock() |
| + switch r { |
| + case settleExited: |
| + // Nothing falls back from a hold (spec §8.3): the instance is |
| + // down until the sweep notices the exit and a later wake runs. |
| + return SwapResult{Outcome: SwapDown, Gen: g, Reason: errSettleExit.Error(), Drain: drain} |
| + case settleCancelled: |
| + return abandon(errSwapStopped) |
| + } |
| + } |
| + // The checks once more before the result counts (spec §6.3, §8.1): |
| + // a newer attempt or a halted cohort since the wake began means this |
| + // build is not recorded good. It keeps serving until the newer |
| + // attempt replaces it. |
| a.mu.Lock() |
| - hw.next.settleCancel = nil |
| + cerr := a.swapChecksLocked(hw.next, at, newest) |
| a.mu.Unlock() |
| - switch r { |
| - case settleExited: |
| - // Nothing falls back from a hold (spec §8.3): the instance is |
| - // down until the sweep notices the exit and a later wake runs. |
| - return SwapResult{Outcome: SwapDown, Gen: g, Reason: errSettleExit.Error(), Drain: drain} |
| - case settleCancelled: |
| - return abandon(errSwapStopped) |
| + if cerr != nil { |
| + return abandon(cerr) |
| } |
| return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain} |
| - case werr == nil: |
| - return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain} |
| + case errors.Is(werr, errSwapSuperseded) || errors.Is(werr, errSwapFenced) || errors.Is(werr, errSwapShutdown): |
| + // The pinned wake's own checks refused the exec (pinCheck): nothing |
| + // of the build ran, and this is no failure of it. |
| + return abandon(werr) |
| case errors.Is(werr, ErrFenced) || errors.Is(werr, context.Canceled): |
| // An intentional stop ended the pinned wake (spec §6.4). |
| return abandon(errSwapStopped) |
| @@ -3628,10 +3984,39 @@ func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) { |
| pin := h.gen |
| next := &actEntry{host: e.host, route: e.route, state: actWaking, ready: make(chan struct{}), pin: &pin, |
| settleCancel: h.settleCancel} |
| + // The same checks again at the pinned wake's exec (spec §6.3, §7.4): |
| + // the lease and the restore come between this moment and the exec. |
| + next.pinCheck = func() error { |
| + switch { |
| + case a.shuttingDown: |
| + return errSwapShutdown |
| + case a.fences[next.host] != 0: |
| + return errSwapFenced |
| + case h.newest(next.host) != h.at.SchedRev: |
| + return errSwapSuperseded |
| + } |
| + return nil |
| + } |
| a.entries[e.host] = next |
| return next, nil |
| } |
| +// startPinnedChecked is a hold's pinned spawn behind its attempt's checks, |
| +// run under a.mu at the exec (pinCheck, set by holdWakeLocked). A check |
| +// that fails spawns nothing. |
| +func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen) (Proc, string, Gen, error) { |
| + a.mu.Lock() |
| + var err error |
| + if e.pinCheck != nil { |
| + err = e.pinCheck() |
| + } |
| + a.mu.Unlock() |
| + if err != nil { |
| + return nil, "", pin, err |
| + } |
| + return a.startPinned(ctx, rt, pin) |
| +} |
| + |
| // startPinned is a hold's spawn: the attempt's staged build, verified and |
| // exec'd by path at its own tier, then the health check through the |
| // generation handle. StartStaged returns as soon as the process runs (Task |
| @@ -3676,13 +4061,20 @@ func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) ( |
| if a.foreignRow(at.Host) { |
| // A foreign row keeps its canonical socket (plan 1), so two |
| // generations of it cannot coexist: its replacement is a hold |
| - // (plan 1's reading, accepted in the index). Its result reads |
| - // as a hold's: SwapDown when the new build fails. |
| - return a.swapHold(ctx, at, newest) |
| - } |
| - return a.swapOverlap(ctx, gb, at, newest) |
| + // (spec §2.9, rev 17). Its result reads as a hold's, SwapDown |
| + // when the new build fails, and Mode tells the scheduler so: a |
| + // hold build that ran is held down (spec §6.4, §8.3). |
| + res := a.swapHold(ctx, at, newest) |
| + res.Mode = ModeHold |
| + return res |
| + } |
| + res := a.swapOverlap(ctx, gb, at, newest) |
| + res.Mode = ModeOverlap |
| + return res |
| case ModeHold: |
| - return a.swapHold(ctx, at, newest) |
| + res := a.swapHold(ctx, at, newest) |
| + res.Mode = ModeHold |
| + return res |
| } |
| ``` |
| @@ -3725,8 +4117,8 @@ func TestSwapOverlapOnAForeignRowRunsAsAHold(t *testing.T) { |
| f.b.layoutFn = func(registry.Route) (Layout, string) { return LayoutForeign, "hand-set socket" } |
| incumbent := f.b.lastProc() |
| res := f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) |
| - if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged { |
| - t.Fatalf("Swap = %+v, want done through the hold", res) |
| + if res.Outcome != SwapDone || res.Gen.StagedPath != f.staged || res.Mode != ModeHold { |
| + t.Fatalf("Swap = %+v, want done through the hold, reported as a hold (Mode)", res) |
| } |
| ev := f.b.events() |
| if i, j := slices.Index(ev, "stop:instance"), indexPrefix(ev, "startGen:"+f.staged); i < 0 || j < 0 || i > j { |
| @@ -3778,11 +4170,13 @@ func TestSwapOverlapOnAForeignRowRunsAsAHold(t *testing.T) { |
| ```go |
| } else if e.pin != nil { // swapsim:M3 |
| var g Gen |
| - proc, spawnVer, g, err = a.startPinned(ctx, rt, *e.pin) |
| + proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin) |
| w.gen = g |
| } else { |
| ``` |
| +Add to `actEntry`, beside `pin`: `pinCheck func() error // a hold's attempt checks, re-run under a.mu at the pinned exec; set at construction (holdWakeLocked)`. |
| + |
| `activator.go`, `finishStop`. Replace lines 2369–2375: |
| ```go |
| @@ -3842,6 +4236,20 @@ In `doClaimWake`, replace line 108, `policy := startPolicy{MaxStarts: 5, Window: |
| ```go |
| policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin} |
| + if e.pin != nil { |
| + // A claims-on hold's pinned wake: its attempt's checks again before |
| + // anything is admitted or exec'd (spec §6.3, §7.4), as |
| + // startPinnedChecked does for a claims-off one. |
| + a.mu.Lock() |
| + var perr error |
| + if e.pinCheck != nil { |
| + perr = e.pinCheck() |
| + } |
| + a.mu.Unlock() |
| + if perr != nil { |
| + return nil, perr |
| + } |
| + } |
| ``` |
| In `startAdmitted`, after the `if !s.valid() || ...` block (line 157–159), insert: |
| @@ -3899,7 +4307,7 @@ git commit -m "zdd plan 3 task 6: the hold swap, a pinned wake installed by fini |
| Three rules this task keeps (findings 2, 10 and 11 of the P1 review): |
| - **Placement.** `startSidecarChildFrom` is plan 2 Task 10's `startSidecarChild` body with one change: the binary. Plan 2's `placeFor(spawnSidecar, 0)`, `spawnInCgroupAt(rt, child, false, …)` and the goroutine's `RemoveChild(rt.Host, child)` stay. Going back to `spawnInCgroup` would put the tier and `memory.oom.group=1` back on the generation parent (spec §7.3) and, with controllers delegated, try to place a process in a parent that must hold none. |
| - **No live-link read for a nested sidecar.** The nested unit binds whatever root lays out, so a check of `LiveVersion` before the start is a race with a repoint, and after a fallback it refuses for ever. Until Task 9 the nested arm refuses outright (`errNestedSidecarGen`), the way Task 1's nested `StartStaged` does. Task 9 hands root the generation's captured staged executable, which root validates under the same rule as the run unit's. |
| - - **Replacement is never skipped.** A supervised restart in flight captured the generation it read when it began, which can be the one just retired. `replaceSidecar` waits for it (`sidecarStartDone`) and then replaces whatever it installed. The supervisor's own completion also discards a sidecar started for a generation that is no longer current. |
| + - **Replacement is never skipped, and never runs beside the old one.** A supervised restart in flight captured the generation it read when it began, which can be the one just retired. `replaceSidecar` waits for it (`sidecarStartDone`) and then replaces whatever it installed. The supervisor installs a sidecar it started for a retired generation (it does not discard it asynchronously), so it stays tracked: `replaceSidecar` stops it synchronously and starts the new one only once it has exited, and a terminal stop stops it like any sidecar. An old sidecar whose stop is unproven is kept as the entry's sidecar, nothing starts beside it, and the sweep retries the replacement (`actEntry.sidecarGen`, `wokenInstance.sidecarGen`: the generation a sidecar runs). |
| - [ ] **Step 1: Write the failing tests** |
| @@ -3915,7 +4323,10 @@ func (g *fakeSwapBackend) StartSidecarGen(_ context.Context, _ registry.Route, g |
| if gate != nil { |
| <-gate |
| } |
| - p := &fakeProc{label: "sidecar", rec: g.fakeBackend} |
| + g.gmu.Lock() |
| + stopGate := g.sidecarStopGate |
| + g.gmu.Unlock() |
| + p := &fakeProc{label: "sidecar", rec: g.fakeBackend, stopGate: stopGate} |
| g.mu.Lock() |
| g.sidecars = append(g.sidecars, p) |
| g.mu.Unlock() |
| @@ -4105,6 +4516,42 @@ func TestSwapSidecarReplacementJoinsAnInFlightSupervisor(t *testing.T) { |
| } |
| } |
| +// The supervisor's start for the retired generation is joined until that |
| +// process is proven gone (spec §7.4, §7.5): the replacement never starts |
| +// while it still runs. Discarding it asynchronously let the supervisor |
| +// close its join at once, so the replacement started beside a sidecar |
| +// still draining on the old build. |
| +func TestSwapSidecarNeverRunsBesideTheRetiredGenerationsSidecar(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.b.sidecarStaged = true |
| + f.a.sweepOnce() |
| + waitFor(t, "late attachment", func() bool { return f.a.sidecarSettled(f.host) && len(f.b.sidecarGenPaths()) == 1 }) |
| + startGate, stopGate := make(chan struct{}), make(chan struct{}) |
| + f.b.gmu.Lock() |
| + f.b.sidecarGenGate, f.b.sidecarStopGate = startGate, stopGate |
| + f.b.gmu.Unlock() |
| + f.b.lastSidecar().kill() |
| + f.clk.advance(3 * time.Second) |
| + f.a.sweepOnce() // the supervisor starts the incumbent's sidecar and blocks |
| + waitFor(t, "the supervisor's start", func() bool { return len(f.b.sidecarGenPaths()) == 2 }) |
| + results := make(chan SwapResult, 1) |
| + go func() { results <- f.a.Swap(context.Background(), f.attempt(ModeOverlap), f.newest) }() |
| + time.Sleep(50 * time.Millisecond) |
| + close(startGate) // the supervisor's sidecar starts; its Stop blocks on stopGate |
| + time.Sleep(100 * time.Millisecond) |
| + if n := len(f.b.sidecarGenPaths()); n != 2 { |
| + t.Fatalf("%d sidecar starts: the replacement started while the retired generation's sidecar still ran", n) |
| + } |
| + close(stopGate) |
| + if res := <-results; res.Outcome != SwapDone { |
| + t.Fatalf("Swap = %+v", res) |
| + } |
| + waitFor(t, "the replacement", func() bool { return len(f.b.sidecarGenPaths()) == 3 }) |
| + if p := f.b.sidecarGenPaths(); p[2] != f.staged { |
| + t.Fatalf("sidecar starts %v: want the new generation's last", p) |
| + } |
| +} |
| + |
| // Under plan 2's containment the sidecar keeps its role-sidecar child: no |
| // limit of its own, and the generation parent keeps memory.oom.group=0 and |
| // holds no process (spec §7.3, plan 2 D4). Spawning through spawnInCgroup |
| @@ -4192,7 +4639,7 @@ var _ SidecarGenStarter = (*SplitBackend)(nil) |
| - [ ] **Step 2: Run the tests to verify they fail** |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestLegacySidecar|TestSwapSidecar|TestExecBackendStartSidecarGen|TestStartSidecarGenKeeps' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestLegacySidecar|TestSwapSidecar|TestExecBackendStartSidecarGen|TestStartSidecarGenKeeps' -count=1` (`TestSwapSidecar` covers `TestSwapSidecarNeverRunsBesideTheRetiredGenerationsSidecar`) |
| Expected: build failure: `undefined: SidecarGenStarter`, `b.StartSidecarGen undefined`, `undefined: errNestedSidecarGen`. |
| - [ ] **Step 3: Write the backend side** |
| @@ -4334,18 +4781,36 @@ func (a *Service) replaceSidecar(e *actEntry) { |
| a.mu.Unlock() |
| return |
| } |
| + if e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen == e.gen.N { |
| + // Already on the current generation (the sweep's retry got here |
| + // first): nothing to replace. |
| + a.mu.Unlock() |
| + return |
| + } |
| e.sidecarStarting = true |
| done := make(chan struct{}) |
| e.sidecarStartDone = done |
| old, rt, g := e.sidecar, e.route, e.gen |
| a.mu.Unlock() |
| - defer close(done) |
| + defer close(done) // only after the old sidecar is proven gone, or retained |
| if old != nil { |
| ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| if err := old.Stop(ctx); err != nil { |
| log.Printf("activator: stop sidecar %s for its replacement: %v", e.host, err) |
| } |
| cancel() |
| + if !old.Exited() { |
| + // Unproven (spec §7.4, §7.5): it stays tracked as the entry's |
| + // sidecar, so a terminal stop still stops it, and no second |
| + // sidecar starts beside it. The sweep retries the replacement. |
| + a.mu.Lock() |
| + if a.entries[e.host] == e { |
| + e.sidecarStarting, e.sidecarStartDone, e.sidecarUncertain = false, nil, true |
| + } |
| + a.mu.Unlock() |
| + log.Printf("activator: SIDECAR for %s did not stop for its replacement; keeping it tracked (the sweep retries)", e.host) |
| + return |
| + } |
| } |
| ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| sc, err := a.startSidecar(ctx, rt, g) |
| @@ -4362,7 +4827,7 @@ func (a *Service) replaceSidecar(e *actEntry) { |
| stopDiscardedSidecar(sc) |
| return |
| } |
| - e.sidecar, e.sidecarStarted, e.sidecarBackoff = sc, a.now(), 0 |
| + e.sidecar, e.sidecarStarted, e.sidecarBackoff, e.sidecarGen = sc, a.now(), 0, g.N |
| if err != nil { |
| e.sidecarUncertain = true |
| log.Printf("activator: SIDECAR for %s failed to start on generation %d: %v (the sweep retries)", e.host, g.N, err) |
| @@ -4396,14 +4861,36 @@ In the same goroutine, directly after the existing `if e0.state != actAwake { |
| if a.gens && e0.gen.N != gen.N { |
| // A swap switched generations while this start was in |
| // flight, so sc runs the retired generation's build |
| - // (spec §7.5). The swap's replaceSidecar is waiting on |
| - // this goroutine and starts the current generation's |
| - // sidecar next; nothing old is installed meanwhile. |
| - stopDiscardedSidecar(sc) |
| + // (spec §7.5). It is NOT discarded asynchronously |
| + // (stopDiscardedSidecar): the swap's replaceSidecar, |
| + // joined on this goroutine, would then start the new |
| + // sidecar while this one still ran, and a terminal |
| + // stop could pass the join over it (§7.4). It is |
| + // installed, so it stays tracked: replaceSidecar stops |
| + // it synchronously and starts the current generation's |
| + // only once it is proven gone, and a terminal stop |
| + // stops it like any sidecar. |
| + e0.sidecar, e0.sidecarStarted, e0.sidecarGen = sc, a.now(), gen.N |
| + if err != nil { |
| + e0.sidecarUncertain = true |
| + } |
| return |
| } |
| ``` |
| +and in the same goroutine's two success installs (`e0.sidecar = sc` in the error and the success arms), record the generation too: `e0.sidecarGen = gen.N`. Do the same in `doWake` step 6 (`w.sidecarGen = w.gen.N`, copied into `e.sidecarGen` by `wake()` beside `e.sidecar`) and in `replaceSidecar` below. Add to `actEntry`: `sidecarGen uint64 // the generation whose build the legacy sidecar runs; written under a.mu`, and the same field to `wokenInstance`. |
| + |
| +In `sweepOnce`, beside the supervisor (inside the same per-entry block, before its `if e.sidecarWanted && (e.sidecar == nil || e.sidecar.Exited()) …`), add the retry for a retained one: |
| + |
| +```go |
| + if a.gens && e.state == actAwake && e.sidecarWanted && !e.sidecarStarting && |
| + e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen != e.gen.N { |
| + // A legacy sidecar still on a retired generation's build: a |
| + // replacement whose stop of it was unproven (spec §7.5). |
| + go a.replaceSidecar(e) |
| + } |
| +``` |
| + |
| `swap.go`, in `swapOverlap`, step 8: replace the final `return res` with: |
| ```go |
| @@ -4792,7 +5279,8 @@ A refusal answers 403. The client turns that into `ErrStagedExe`, and the swap i |
| // startRole gains a trailing exe *instancectl.StagedExe parameter (nil: plan 2's StartGen, the live link) |
| func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock string, gen uint64, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, string, error) // spawn only; startUnitAt = this + the health tail |
| // startRole gains a second trailing parameter, tier cgroup.Limit (zero: b.tier(rt)) |
| - func (b *ExecBackend) startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe) (Proc, error) |
| + func (b *ExecBackend) startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, error) // the generation's own tier |
| + // handleStart: a staged sidecar start (exe set) leaves a populated host slice's limit alone (the generation parent; P2 review finding 2) |
| // ExecBackend.StartSidecarGen's nested arm (Task 7's errNestedSidecarGen) now starts the sidecar unit on g's staged executable |
| ``` |
| **The nested legacy sidecar** (spec §7.5, "every legacy co-spawned sidecar runs its instance's serving artifact") gets the same treatment as the run unit. Its unit, `carlos-instance-sidecar@<id>`, binds `<ExeRoot>/<id>/app`, which root's `ensureHostExe` points at the live link. A `/start` for the sidecar role may now carry the generation's captured `exe`, `version` and `digest`. Root runs the same `validateStagedExe` (the host's own app's `VersionPath`, opened without following a symlink, the digest equal), answers 403 on a refusal, and only then points that `app` link at the staged build. The sidecar therefore never depends on the live link, so neither a fallback (serving differs from installed) nor a repoint during the start changes what it runs. A run-role start on the original template still refuses `exe`: only generation starts and the sidecar take one. |
| @@ -4884,6 +5372,24 @@ func TestValidateStagedExe(t *testing.T) { |
| } |
| } |
| +// ctlCalled reports whether any recorded systemctl call starts with prefix. |
| +func ctlCalled(c *fakeCtl, prefix string) bool { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + for _, l := range c.calls { |
| + if strings.HasPrefix(l, prefix) { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func (f *fakeCtl) reset() { |
| + f.mu.Lock() |
| + f.calls = nil |
| + f.mu.Unlock() |
| +} |
| + |
| // The nested legacy sidecar runs the serving generation's staged build |
| // (spec §7.5), validated by root under the run unit's rule before its |
| // executable mount is pointed at it. The live link names another build |
| @@ -4891,11 +5397,12 @@ func TestValidateStagedExe(t *testing.T) { |
| // leaves the mount alone. Before this task a sidecar /start could not name |
| // an executable at all, so the sidecar bound the live link. |
| func TestSidecarStartMountsTheValidatedStagedExe(t *testing.T) { |
| - opt, exeRoot := t.TempDir(), t.TempDir() |
| + opt, exeRoot, cgroupRoot := t.TempDir(), t.TempDir(), t.TempDir() |
| staged := release.VersionPath(opt, release.OpsAccount, "v2", "blog") |
| digest := writeStaged(t, staged, "build v2") |
| - _, sock := serveOnSocket(t, &fakeCtl{}, func(s *Server) { |
| - s.DropInDir, s.ExeRoot, s.OptDir = t.TempDir(), exeRoot, opt |
| + ctl := &fakeCtl{} |
| + _, sock := serveOnSocket(t, ctl, func(s *Server) { |
| + s.DropInDir, s.ExeRoot, s.OptDir, s.CgroupRoot = t.TempDir(), exeRoot, opt, cgroupRoot |
| s.HostApp = func(string) (string, string, error) { return release.OpsAccount, "blog", nil } |
| }) |
| c := NewClient(sock) |
| @@ -4918,6 +5425,35 @@ func TestSidecarStartMountsTheValidatedStagedExe(t *testing.T) { |
| t.Errorf("%s: a refused start moved the mount to %q", name, got) |
| } |
| } |
| + // The slice above was cold, so its tier was set. With a generation |
| + // running (populated), a staged sidecar start leaves the slice's limit |
| + // alone: during an overlap it is the raised parent of both generations |
| + // (spec §7.3), and a start that set one tier would put them both under |
| + // it. A flag-off start (no exe) sets the tier as today. |
| + if !ctlCalled(ctl, "set-property --runtime "+SliceName(genHost)) { |
| + t.Fatal("a cold slice got no tier") |
| + } |
| + events := filepath.Join(cgroupRoot, parentSlices, SliceName(genHost), "cgroup.events") |
| + if err := os.MkdirAll(filepath.Dir(events), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := os.WriteFile(events, []byte("populated 1\nfrozen 0\n"), 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + ctl.reset() |
| + if _, err := c.StartSidecarExe(context.Background(), genHost, cgroup.TierDefault.MemoryMax, |
| + StagedExe{Path: staged, Version: "v2", Digest: digest}); err != nil { |
| + t.Fatal(err) |
| + } |
| + if ctlCalled(ctl, "set-property") { |
| + t.Fatal("a staged sidecar start rewrote the populated generation parent's limit") |
| + } |
| + if _, err := c.Start(context.Background(), genHost, RoleSidecar, cgroup.TierDefault.MemoryMax); err != nil { |
| + t.Fatal(err) |
| + } |
| + if !ctlCalled(ctl, "set-property --runtime "+SliceName(genHost)) { |
| + t.Fatal("a flag-off sidecar start no longer sets the tier") |
| + } |
| // A run-role start on the original template names no executable. |
| q := hostRole(genHost, RoleRun) |
| q.Set("memory_max", strconv.FormatInt(cgroup.TierDefault.MemoryMax, 10)) |
| @@ -5039,6 +5575,7 @@ type exeControl struct { |
| sidecarExes []instancectl.StagedExe |
| plainStarts []instancectl.Role // Start calls: the live-link exe layout |
| tiers []int64 // StartGenExe's memoryMax, per call |
| + sidecarTiers []int64 // StartSidecarExe's memoryMax, per call |
| } |
| func (c *exeControl) Start(_ context.Context, _ string, role instancectl.Role, _ int64) (string, error) { |
| @@ -5047,10 +5584,11 @@ func (c *exeControl) Start(_ context.Context, _ string, role instancectl.Role, _ |
| c.plainStarts = append(c.plainStarts, role) |
| return "inv", nil |
| } |
| -func (c *exeControl) StartSidecarExe(_ context.Context, _ string, _ int64, x instancectl.StagedExe) (string, error) { |
| +func (c *exeControl) StartSidecarExe(_ context.Context, _ string, memoryMax int64, x instancectl.StagedExe) (string, error) { |
| c.mu.Lock() |
| defer c.mu.Unlock() |
| c.sidecarExes = append(c.sidecarExes, x) |
| + c.sidecarTiers = append(c.sidecarTiers, memoryMax) |
| return "inv-sc", nil |
| } |
| func (c *exeControl) Seed(context.Context, string, int64) (instancectl.SeedResponse, error) { |
| @@ -5108,7 +5646,9 @@ func TestStartSidecarGenNestedNamesTheStagedExe(t *testing.T) { |
| if err := os.WriteFile(b.pathsFor(rt).sidecarEnv, []byte("X=1\n"), 0o600); err != nil { |
| t.Fatal(err) |
| } |
| - g := Gen{N: 4, Build: "v2", StagedPath: "/opt/carlos/versions/v2/blog", Digest: strings.Repeat("b", 64)} |
| + // The generation's own tier: 1 GiB, while the live link's (b.tier) is |
| + // the default 512 MiB. After a fallback the two differ. |
| + g := Gen{N: 4, Build: "v2", StagedPath: "/opt/carlos/versions/v2/blog", Digest: strings.Repeat("b", 64), Tier: cgroup.TierChildren} |
| p, err := b.StartSidecarGen(context.Background(), rt, g) |
| if err != nil { |
| t.Fatalf("StartSidecarGen: %v", err) |
| @@ -5123,6 +5663,9 @@ func TestStartSidecarGenNestedNamesTheStagedExe(t *testing.T) { |
| if len(ctl.plainStarts) != 0 { |
| t.Fatalf("a live-link start ran too: %v", ctl.plainStarts) |
| } |
| + if ctl.sidecarTiers[0] != cgroup.TierChildren.MemoryMax { |
| + t.Fatalf("root was sent memory_max %d, want the generation's own %d", ctl.sidecarTiers[0], cgroup.TierChildren.MemoryMax) |
| + } |
| } |
| // Root refusing the staged path abandons the swap. It is not a failure of |
| @@ -5326,6 +5869,25 @@ with |
| if s.refuseStart(w, req.host, unit, func() error { return s.ensureHostExeAt(req.host, req.exe) }) { |
| ``` |
| +In the same original-template path, the tier write that follows, |
| + |
| +```go |
| + if err := s.setTier(ctx, req.host, req.tier); err != nil { |
| +``` |
| +becomes |
| +```go |
| + // A sidecar start that names a staged build comes from a box with |
| + // generations on, where the host slice is the generation parent (spec |
| + // §7.3). A swap may have raised it to two tiers, and lowering it here |
| + // would put both generations under one tier's limit. Its restore is the |
| + // swap owner's, so a populated slice is left alone, exactly as plan 2's |
| + // startGen does; a cold slice still gets the tier. A start without exe |
| + // (flag off) sets the tier as today. |
| + if req.exe.Path != "" && s.slicePopulated(req.host) { |
| + s.logf("instance-control: %s: slice populated, leaving its limit to the generation owner", req.host) |
| + } else if err := s.setTier(ctx, req.host, req.tier); err != nil { |
| +``` |
| + |
| The bind is made when the unit starts, so a later start for the same host (a seed, or a flag-off run unit) that points `app` back at the live link does not change what an already-running sidecar runs. |
| In plan 2's `startGen`, replace |
| @@ -5403,7 +5965,7 @@ In plan 2's `GenControl` interface, add: |
| Plan 2's test fake for `GenControl` (`fakeGenControl`) gains both methods: `StartGenExe` returns what its `StartGen` returns, and `StartSidecarExe` records `startsidecarexe <host> <max>` and returns `f.inv, f.startErr`. `instancectl.Client` satisfies the interface through Step 4. |
| -**The nested legacy sidecar.** In `backend_exec.go`, rename the body of `startSidecarUnit` to `startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe) (Proc, error)` and keep `startSidecarUnit` as a forwarder: `return b.startSidecarUnitFrom(ctx, rt, nil)`. In the body, replace |
| +**The nested legacy sidecar.** In `backend_exec.go`, rename the body of `startSidecarUnit` to `startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, error)` and keep `startSidecarUnit` as a forwarder: `return b.startSidecarUnitFrom(ctx, rt, nil, cgroup.Limit{})`. In the body, replace |
| ```go |
| inv, err := b.Control.Start(ctx, rt.Host, instancectl.RoleSidecar, b.tier(rt).MemoryMax) |
| @@ -5413,13 +5975,17 @@ with |
| var inv string |
| if exe != nil { |
| // The serving generation's captured staged build (zero-downtime |
| - // spec §7.5), which root validates before it lays out the mount. |
| + // spec §7.5), which root validates before it lays out the mount, |
| + // at that generation's own tier (Task 1), never the live link's. |
| gc, gerr := b.genControl() |
| if gerr != nil { |
| logs.Close() |
| return nil, gerr |
| } |
| - inv, err = gc.StartSidecarExe(ctx, rt.Host, b.tier(rt).MemoryMax, *exe) |
| + if tier == (cgroup.Limit{}) { |
| + tier = b.tier(rt) |
| + } |
| + inv, err = gc.StartSidecarExe(ctx, rt.Host, tier.MemoryMax, *exe) |
| } else { |
| inv, err = b.Control.Start(ctx, rt.Host, instancectl.RoleSidecar, b.tier(rt).MemoryMax) |
| } |
| @@ -5439,7 +6005,7 @@ In `genbackend.go`, Task 7's nested arm of `StartSidecarGen`, `return nil, errNe |
| return nil, errNestedSidecarGen |
| } |
| exe := instancectl.StagedExe{Path: g.StagedPath, Version: g.Build, Digest: g.Digest} |
| - return b.startSidecarUnitFrom(ctx, rt, &exe) |
| + return b.startSidecarUnitFrom(ctx, rt, &exe, g.Tier) |
| ``` |
| and `errNestedSidecarGen`'s message becomes `"nested sidecar: the generation has no staged executable"`. Update Task 7's `TestExecBackendStartSidecarGen` nested subtest to match: with `Gen{N: 3, Build: build}` (no staged path) it still answers `errNestedSidecarGen`. A root refusal (`ErrStagedExe`) reaches the supervisor as a failed start, which retries on its backoff; the instance keeps serving. |
| @@ -6713,7 +7279,7 @@ git commit -m "zdd plan 3 task 12: agent restart mid-swap and the never-healthy |
| - Create: `internal/activator/swap_sim_test.go`, `internal/activator/testdata/swapsim/mutate.sh` |
| **Interfaces:** |
| -- Consumes: everything above, plus the `// swapsim:M1`…`M7` markers placed in Tasks 2, 3, 4 and 6 (M7 is in Task 2's `stopMember`). |
| +- Consumes: everything above, plus the `// swapsim:M1`…`M8` markers placed in Tasks 2, 3, 4 and 6 (M7 is in Task 2's `stopMember`, M8 in Task 2's `stop()`). |
| - Produces: `TestSwapInvariants` (seeds: `SWAP_SIM_SEEDS`, default 100; replay one with `SWAP_SIM_SEED=<n>`); `mutate.sh`. |
| The invariants (spec §12), checked on every event: |
| @@ -7281,7 +7847,8 @@ M3|activator.go|swapsim:M3|s/e\.pin != nil/false/ |
| M4|opset.go|swapsim:M4|s/proven := a\.stopMember\(e, m\)/proven := true; go a.stopMember(e, m)/ |
| M5|swapadmit.go|swapsim:M5|s/ \|\| e\.state != actAwake// |
| M6|activator.go|swapsim:M6|s/if a\.gens \{/if false {/ |
| -M7|opset.go|swapsim:M7|s/return a\.genChildEmpty\(e, m\)/return true/' |
| +M7|opset.go|swapsim:M7|s/return a\.genChildEmpty\(rt, m\)/return true || a.genChildEmpty(rt, m)/ |
| +M8|activator.go|swapsim:M8|s/if !a\.genChildEmpty\(/if false \&\& !a.genChildEmpty(/' |
| fail=0 |
| while IFS='|' read -r name file marker sub; do |
| src="$SCRATCH/tree/$A/$file" |
| @@ -7317,7 +7884,7 @@ Then run `chmod +x internal/activator/testdata/swapsim/mutate.sh`. |
| - [ ] **Step 6: Run the mutation checks** |
| Run: `uptime && mkdir -p /tmp/k && nice -n 19 internal/activator/testdata/swapsim/mutate.sh 300` |
| -Expected: seven `CAUGHT` lines and exit status 0. Each mutant has a deterministic witness: |
| +Expected: eight `CAUGHT` lines and exit status 0. Each mutant has a deterministic witness: |
| - M1: `TestSwapWitnessBarrierJoinsBeforeTheLease`, `TestSwapWitnessAgentRestartMidSwap` |
| - M2: `TestSwapWitnessAgentRestartMidSwap` |
| - M3: `TestSwapWitnessHoldExecsTheStagedBuild` |
| @@ -7325,6 +7892,7 @@ Expected: seven `CAUGHT` lines and exit status 0. Each mutant has a deterministi |
| - M5: `TestSwapWitnessNoAdmissionOntoAStoppingEntry` |
| - M6: `TestSwapWitnessShutdownJoinsAWakingEntry` |
| - M7: `TestSwapWitnessMemberStopProvesTheGenerationChildEmpty` |
| +- M8: `TestSwapWitnessStopProvesTheCurrentGenerationEmpty` |
| The sweep count is reported separately. Record both in the branch description. |
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 5ab966238..38ba026a0 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 |
| @@ -2665,6 +2665,9 @@ type fakeAct struct { |
| maxFlight int |
| replaced []replaceCall |
| worker func(a activator.Attempt, fallback *release.StagedIdentity) activator.SwapResult |
| + // mode, when set, is the effective mode Swap reports (SwapResult.Mode): |
| + // plan 3 runs an overlap attempt on a foreign-layout row as a hold. |
| + mode activator.SwapMode |
| } |
| func newFakeAct() *fakeAct { |
| @@ -2722,7 +2725,7 @@ func (f *fakeAct) Swap(_ context.Context, a activator.Attempt, newest func(strin |
| f.inFlight-- |
| f.calls = append(f.calls, swapCall{a: a, outcome: out}) |
| if out != activator.SwapDone { |
| - return activator.SwapResult{Outcome: out, Reason: "health check failed"} |
| + return activator.SwapResult{Outcome: out, Reason: "health check failed", Mode: f.mode} |
| } |
| f.n++ |
| g := activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand} |
| @@ -2778,17 +2781,24 @@ type fakeRoutes struct { |
| mu sync.Mutex |
| rows map[string]registry.Route |
| // getHook, when set, runs at the start of every Get, with no lock held: |
| - // a test's window into the moment the scheduler reads a row. |
| + // a test's window into the moment the scheduler reads a row. getErr, |
| + // when set and answering an error, fails that Get. |
| getHook func(host string) |
| + getErr func(host string) error |
| } |
| func (f *fakeRoutes) Get(host string) (*registry.Route, error) { |
| f.mu.Lock() |
| - hook := f.getHook |
| + hook, fail := f.getHook, f.getErr |
| f.mu.Unlock() |
| if hook != nil { |
| hook(host) |
| } |
| + if fail != nil { |
| + if err := fail(host); err != nil { |
| + return nil, err |
| + } |
| + } |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| r, ok := f.rows[host] |
| @@ -3005,6 +3015,7 @@ Create `internal/rollout/manager_test.go`: |
| package rollout |
| import ( |
| + "errors" |
| "sync/atomic" |
| "testing" |
| "time" |
| @@ -3092,13 +3103,14 @@ func TestAFailedCanaryHaltsTheCohort(t *testing.T) { |
| } |
| } |
| -// Spec §8.1: the failure transition stops the cohort in the same step that |
| -// frees the failed attempt's slot. The hook runs the scheduler's admission |
| -// at the first moment done() lets go of the queue (its registry read). If |
| -// finish() freed the slot and the canary there but the halt waited for the |
| -// failure record, that admission starts a second attempt onto the failed |
| -// build, and an attempt already past spawn passes its newest re-check. |
| -func TestAFailureHaltsTheCohortInTheStepThatFreesItsSlot(t *testing.T) { |
| +// Spec §8.1: the failure transition (the record and the halt) happens in |
| +// the same step that frees the failed attempt's slot. The hook looks at the |
| +// queue at every moment done() reads a row after the failure: whenever the |
| +// failed attempt's slot is free, the record must already be in the book |
| +// (which wake plans read) and admission must start nothing onto the build. |
| +// Halting in that step but writing the record after the row read let a |
| +// wake select the failed build in between. |
| +func TestAFailureIsRecordedAndHaltedInTheStepThatFreesItsSlot(t *testing.T) { |
| h := newHarness(t, "a", "b") |
| h.stage("v2") |
| var failed atomic.Bool |
| @@ -3109,19 +3121,30 @@ func TestAFailureHaltsTheCohortInTheStepThatFreesItsSlot(t *testing.T) { |
| failed.Store(true) |
| return activator.SwapFailed |
| } |
| - var leaked atomic.Int32 |
| + var bad atomic.Int32 |
| h.rows.mu.Lock() |
| h.rows.getHook = func(host string) { |
| - if !failed.CompareAndSwap(true, false) { |
| + if !failed.Load() { |
| return |
| } |
| s := h.m.schedulerOf(host) |
| s.mu.Lock() |
| - ds := s.q.admit(currentView(s.q), h.clock()) |
| + hs := s.q.hosts[host] |
| + free := hs != nil && hs.running == nil |
| + var ds []dispatch |
| + if free { |
| + ds = s.q.admit(currentView(s.q), h.clock()) |
| + } |
| s.mu.Unlock() |
| + if !free { |
| + return |
| + } |
| + if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok { |
| + bad.Add(1) |
| + } |
| for _, d := range ds { |
| if d.ID.Build == "v2" { |
| - leaked.Add(1) |
| + bad.Add(1) |
| } |
| } |
| } |
| @@ -3129,11 +3152,40 @@ func TestAFailureHaltsTheCohortInTheStepThatFreesItsSlot(t *testing.T) { |
| h.install("v2", "a", "b") |
| h.pass() |
| waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok }) |
| - if n := leaked.Load(); n != 0 { |
| - t.Fatalf("%d attempt(s) onto v2 admitted between the failure and the halt", n) |
| + h.pass() // reads the rows again, with the slot free |
| + if n := bad.Load(); n != 0 { |
| + t.Fatalf("%d moment(s) with the slot free and the record missing or a v2 attempt admissible", n) |
| } |
| } |
| +// A row that cannot be read at the failure still gets its record: the |
| +// failure is the build's, not the row's (spec §8.1). Skipping the write on |
| +// a failed registry read left the failed build selectable by every wake. |
| +func TestAFailureIsRecordedWhenItsRowIsUnreadable(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + var failed atomic.Bool |
| + h.act.outcome = func(activator.Attempt) activator.SwapOutcome { |
| + failed.Store(true) |
| + return activator.SwapFailed |
| + } |
| + h.rows.mu.Lock() |
| + h.rows.getErr = func(string) error { |
| + if failed.Load() { |
| + return errors.New("registry: database is locked") |
| + } |
| + return nil |
| + } |
| + h.rows.mu.Unlock() |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "a failure record despite the unreadable row", func() bool { |
| + _, ok := h.book.Lookup("bab", "blog", "v2") |
| + return ok |
| + }) |
| +} |
| + |
| + |
| // Review focus 4. |
| func TestStagingRefusalWritesNoFailureRecord(t *testing.T) { |
| h := newHarness(t, "a") |
| @@ -4143,15 +4195,47 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) { |
| // swap mode, "" when no Swap ran. |
| func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.SwapMode) { |
| m := s.m |
| + if res.Mode != "" { |
| + // The mode the attempt actually ran in: an overlap attempt on a |
| + // foreign-layout row runs as a hold (spec §2.9), and a failed hold |
| + // is held down (§6.4, §8.3). |
| + mode = res.Mode |
| + } |
| + rt, rtErr := m.d.Routes.Get(d.ID.Host) |
| + if rtErr != nil { |
| + rt = nil |
| + } |
| + // A failure's record is built before the lock (the pointer read behind |
| + // failedUnder may be slow) and written in the same critical section that |
| + // frees the slot and halts the cohort (spec §8.1): no admission, no |
| + // newest re-check and no wake plan (which reads the book) can see the |
| + // slot free before the record. An unreadable row still records, with the |
| + // failure's own time standing in for the pointer's. |
| + var rec *FailureRecord |
| + if res.Outcome == activator.SwapFailed || res.Outcome == activator.SwapDown { |
| + r := FailureRecord{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Host: d.ID.Host, |
| + Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, PromotedAt: m.d.Now(), At: m.d.Now()} |
| + if rt != nil { |
| + r.PromotedAt = m.failedUnder(*rt, d.ID.Build) |
| + } |
| + if res.Outcome == activator.SwapDown && mode == activator.ModeHold { |
| + // A hold build executed against this instance's database: |
| + // nothing falls back on it automatically (spec §6.4, §8.3). |
| + r.HeldDown = []string{d.ID.Host} |
| + } |
| + rec = &r |
| + } |
| s.mu.Lock() |
| transition := s.q.finish(d, res.Outcome) |
| - if transition { |
| - // The cohort-failure transition's queue half (spec §8.1 step 2), in |
| - // the critical section that freed the slot and the canary: no |
| - // admission and no newest re-check can run between the failure and |
| - // the halt. cohortFailed below writes the record; its own halt of |
| - // the same build is then a no-op for admission. |
| - s.q.halt(d.ID.Build, res.Reason) |
| + var ( |
| + added bool |
| + superseded []string |
| + recErr error |
| + ) |
| + if transition && rec != nil { |
| + // The whole cohort-failure transition (spec §8.1 steps 1 and 2): the |
| + // record (in memory at once, on disk as it can) and the halt. |
| + added, superseded, recErr = failCohort(m.d.Book, s.q, *rec) |
| } |
| serving := "" |
| if h := s.q.hosts[d.ID.Host]; h != nil { |
| @@ -4159,10 +4243,18 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| } |
| s.mu.Unlock() |
| defer s.poke() |
| - rt, err := m.d.Routes.Get(d.ID.Host) |
| - if err != nil || rt == nil { |
| + if recErr != nil { |
| + m.d.Logf("rollout: write failure record %s/%s %s: %v (held in memory)", rec.Account, rec.App, rec.Build, recErr) |
| + } |
| + if len(superseded) > 0 { |
| + m.d.Logf("rollout: %s/%s build %s failed: superseding attempts past spawn on %v", rec.Account, rec.App, rec.Build, superseded) |
| + } |
| + if rt == nil { |
| return |
| } |
| + if added { |
| + m.emit(*rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, rec.Reason)) |
| + } |
| switch res.Outcome { |
| case activator.SwapDone: |
| m.served(d.ID.Host, res.Gen) |
| @@ -4174,17 +4266,6 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| m.mu.Unlock() |
| m.emit(*rt, EvSwapped, "info", textSwapped(d.ID.Build, res.Drain)) |
| case activator.SwapFailed, activator.SwapDown: |
| - if transition { |
| - rec := FailureRecord{Account: s.key.acct, App: s.key.app, Build: d.ID.Build, Host: d.ID.Host, |
| - Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, |
| - PromotedAt: m.failedUnder(*rt, d.ID.Build), At: m.d.Now()} |
| - if res.Outcome == activator.SwapDown && mode == activator.ModeHold { |
| - // A hold build executed against this instance's database: |
| - // nothing falls back on it automatically (spec §6.4, §8.3). |
| - rec.HeldDown = []string{d.ID.Host} |
| - } |
| - m.cohortFailed(s, *rt, rec) |
| - } |
| if res.Outcome == activator.SwapDown { |
| m.emit(*rt, EvSwapDown, "error", textSwapDown(d.ID.Build)) |
| } else { |
| @@ -4236,6 +4317,8 @@ git commit -m "zdd-5 task 8: the manager and the per-app scheduler" |
| // genKiller is plan 3 Task 2's (it calls KillGen after every member's drain); this task uses it and declares none |
| func (a *Service) claimedPlanResult(e *actEntry, w *wokenInstance) *wokenInstance |
| func plannedWorkerReady(ctx context.Context, p Proc) error // spec §6.5 readiness for a planned socketless wake; awaitWorkerSettle stays for flag-off wakes |
| + func (a *Service) armFirstWatchLocked(e *actEntry, w *wokenInstance) // registers a first wake's observation in wake()'s awake transition |
| + // actEntry gains firstWatchCtx context.Context. A planned worker's first wake emits GenSettled at once (its readiness was the settle); no second observation |
| // startPinned gains a last parameter: healthBy time.Time (zero: no cap). Plan 3's StartStaged returns the |
| // process at once, so this deadline is the only health bound, and a failed B comes back with its Proc (never nil) |
| // for spawnPlanned to record and kill (spec §8.2). |
| @@ -4263,6 +4346,7 @@ import ( |
| "strconv" |
| "strings" |
| "sync" |
| + "sync/atomic" |
| "testing" |
| "time" |
| @@ -4788,6 +4872,70 @@ func TestAPlannedOneShotWorkerThatExitsCleanlyIsReady(t *testing.T) { |
| } |
| } |
| +// Spec §6.4, §8.1: a first wake's observation is registered in the awake |
| +// transition, before the generation is reachable. Here the build crashes, |
| +// and the sweep starts stopping it, in the moment between the wake becoming |
| +// ready and its observer running. Registered late (in afterWake), Current |
| +// exposed the build as stable meanwhile, and the observer, finding the |
| +// entry stopping, reported GenCancelled: a real failure of the build lost. |
| +func TestAFirstWakeIsObservedFromTheMomentItIsReady(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + var visible atomic.Bool |
| + a.OnGen = func(ev GenEvent) { |
| + if ev.Kind == GenHealthy { |
| + if _, ok := a.Current("one.example.com"); ok { |
| + visible.Store(true) |
| + } |
| + b.procOf(ev.Gen.N).crash() |
| + a.sweepOnce() // the sweep sees the exit before the observer runs |
| + } |
| + evs.add(ev) |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" { |
| + t.Fatalf("failure event = %+v", ev) |
| + } |
| + if slices.Contains(evs.kinds(), GenCancelled) { |
| + t.Fatalf("events %v: a crash was reported as a cancelled observation", evs.kinds()) |
| + } |
| + if visible.Load() { |
| + t.Fatal("Current exposed the first wake as stable before its observation was registered") |
| + } |
| +} |
| + |
| +// Spec §6.5, §8.6: a socketless first wake is settled once its readiness |
| +// window is over. A crash after it is ordinary crash recovery, not a failed |
| +// release. A second, HTTP-style observation after the readiness window |
| +// recorded a worker that crashed at 10 s plus a moment as a failed build. |
| +func TestAPlannedWorkerFirstWakeSettlesAtItsReadiness(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatal(err) |
| + } |
| + evs.wait(t, GenSettled) |
| + b.lastWorker().crash() // after its readiness window |
| + time.Sleep(400 * time.Millisecond) |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a crash after readiness was recorded against the build", k) |
| + } |
| +} |
| + |
| // The flag off: no Planner, today's wake, no events. |
| func TestWithoutAPlannerTheWakeIsTodays(t *testing.T) { |
| a, b, _, _ := testActivator(t, time.Hour) |
| @@ -4859,7 +5007,7 @@ func TestBounceForNamesTheCause(t *testing.T) { |
| - [ ] **Step 2: Run them to see them fail** |
| -Run: `mkdir -p /tmp/k && TMPDIR=/tmp/k go test ./internal/activator/ -run 'TestFirstWake|TestAHoldFirstWake|TestAnUnprovenKill|TestAnExitInside|TestAnIntentionalStop|TestAFirstWakeThat|TestARefusedPlan|TestAClaimedFirstWake|TestAPlannedFirstWakeOnTheRealExecBackend|TestAPlannedOneShotWorker|TestWithoutAPlanner|TestDriftFeeds|TestBounceFor' -count=1` |
| +Run: `mkdir -p /tmp/k && TMPDIR=/tmp/k go test ./internal/activator/ -run 'TestFirstWake|TestAHoldFirstWake|TestAnUnprovenKill|TestAnExitInside|TestAnIntentionalStop|TestAFirstWakeThat|TestARefusedPlan|TestAClaimedFirstWake|TestAPlannedFirstWakeOnTheRealExecBackend|TestAPlannedOneShotWorker|TestAFirstWakeIsObserved|TestAPlannedWorkerFirstWake|TestWithoutAPlanner|TestDriftFeeds|TestBounceFor' -count=1` |
| Expected: FAIL, `undefined: WakePlan`. |
| - [ ] **Step 3: Write the plan types and helpers** |
| @@ -5119,12 +5267,32 @@ func (a *Service) afterWake(e *actEntry, w *wokenInstance, err error) { |
| a.emitGen(GenEvent{Kind: GenFellBack, Route: e.route, Gen: w.planned, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now}) |
| default: |
| a.emitGen(GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: plan.FirstWake, At: now}) |
| - if plan.FirstWake { |
| + switch { |
| + case plan.FirstWake && e.worker: |
| + // A socketless first wake's readiness window (plannedWorkerReady, |
| + // spec §6.5) was its settle: it is settled now. A second window |
| + // would fail a build that was already ready, where §8.6 calls for |
| + // ordinary crash recovery. |
| + a.emitGen(GenEvent{Kind: GenSettled, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: true, At: now}) |
| + case plan.FirstWake: |
| a.watchFirstWake(e, w.planned, plan) |
| } |
| } |
| } |
| +// armFirstWatchLocked registers a first wake's observation in the awake |
| +// transition, before e.ready closes (spec §6.4): Current hides the |
| +// generation from the instant it is reachable, and an exit or a stop |
| +// before the observer goroutine runs is classified by it, never lost. |
| +// The caller holds a.mu. |
| +func (a *Service) armFirstWatchLocked(e *actEntry, w *wokenInstance) { |
| + if a.Planner == nil || a.OnGen == nil || e.worker || !e.plan.planned() || !e.plan.FirstWake || w.fellBack || w.failed != nil { |
| + return |
| + } |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + e.firstWatch, e.firstWatchCtx = cancel, ctx |
| +} |
| + |
| // watchFirstWake is a first wake's 10 s observation (spec §6.4), on plan 3's |
| // observeSettle. The waiting requests do not sit through it. An intentional |
| // stop cancels it (beginStopLocked) and records nothing; an unexpected exit |
| @@ -5137,24 +5305,28 @@ func (a *Service) watchFirstWake(e *actEntry, g Gen, plan WakePlan) { |
| // as settling and records nothing (spec §6.4, §9.1). |
| a.emitGen(GenEvent{Kind: GenCancelled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()}) |
| } |
| + // The observation was registered in the awake transition |
| + // (armFirstWatchLocked), so a stop that began meanwhile has already |
| + // cancelled it, or, for a crash, marked unexpectedExit: observeSettle and |
| + // the classification below decide either way. Nothing registered means |
| + // no first-wake observation for this wake. |
| a.mu.Lock() |
| - if a.entries[e.host] != e || e.state != actAwake { |
| - rt := e.route |
| - a.mu.Unlock() |
| + ctx, rt, proc := e.firstWatchCtx, e.route, e.proc |
| + a.mu.Unlock() |
| + if ctx == nil { |
| cancelled(rt) |
| return |
| } |
| - ctx, cancel := context.WithCancel(context.Background()) |
| - e.firstWatch = cancel |
| - proc := e.proc |
| - a.mu.Unlock() |
| go func() { |
| - defer cancel() |
| r := observeSettle(ctx, proc, firstWakeSettle) |
| a.mu.Lock() |
| - e.firstWatch = nil |
| + cancel := e.firstWatch |
| + e.firstWatch, e.firstWatchCtx = nil, nil |
| rt, unexpected := e.route, e.unexpectedExit |
| a.mu.Unlock() |
| + if cancel != nil { |
| + cancel() |
| + } |
| exited := r == settleExited || (r == settleCancelled && unexpected) |
| switch { |
| case r == settleOK, exited && e.worker && !procCrashed(proc): |
| @@ -5231,6 +5403,10 @@ Add to `actEntry`: |
| wakeStart time.Time |
| unexpectedExit bool |
| firstWatch context.CancelFunc |
| + // firstWatchCtx is the first-wake observation's context, registered with |
| + // firstWatch in the awake transition (armFirstWatchLocked) and read by |
| + // watchFirstWake's goroutine. Written under a.mu. |
| + firstWatchCtx context.Context |
| ``` |
| Add to `wokenInstance` (import `internal/release` in `activator.go` if plan 4 has not already): |
| @@ -5278,7 +5454,16 @@ Only the `actWaking` arm changes; the `actStopping` arm keeps `wakeWaitBound`. |
| - [ ] **Step 6: Plan the wake** |
| -In `wake`, in the first locked block, after `e.wakeCancel = cancel`, add `e.wakeStart = time.Now()`. After `close(e.ready)` at the end of `wake`, add: |
| +In `wake`, in the first locked block, after `e.wakeCancel = cancel`, add `e.wakeStart = time.Now()`. In its success branch (the locked block that sets `e.state = actAwake`), directly after `e.staleSince = time.Time{}` and before the fence check, add: |
| + |
| +```go |
| + // A first wake's observation is registered before e.ready closes, |
| + // in this critical section (spec §6.4), not after the generation is |
| + // already reachable. |
| + a.armFirstWatchLocked(e, w) |
| +``` |
| + |
| +After `close(e.ready)` at the end of `wake`, add: |
| ```go |
| a.afterWake(e, w, err) |
| @@ -5306,12 +5491,12 @@ In step 4, plan 3 Task 6 left the instance spawn as: |
| ```go |
| } else if e.pin != nil { // swapsim:M3 |
| var g Gen |
| - proc, spawnVer, g, err = a.startPinned(ctx, rt, *e.pin) |
| + proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin) |
| w.gen = g |
| } else { |
| ``` |
| -Insert the planned branch before it and pass the new parameter, keeping plan 3's marker on its line. `w.gen` is what plan 1's `wake()` copies into `e.gen`: |
| +Insert the planned branch before it, keeping plan 3's marker on its line. `w.gen` is what plan 1's `wake()` copies into `e.gen`. (Step 7 gives `startPinned` its deadline parameter; `startPinnedChecked` passes `time.Time{}` to it.) |
| ```go |
| } else if plan.planned() { |
| @@ -5320,7 +5505,7 @@ Insert the planned branch before it and pass the new parameter, keeping plan 3's |
| w.gen, w.planned = g, g |
| } else if e.pin != nil { // swapsim:M3 |
| var g Gen |
| - proc, spawnVer, g, err = a.startPinned(ctx, rt, *e.pin, time.Time{}) |
| + proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin) |
| w.gen = g |
| } else { |
| ``` |
| @@ -5473,7 +5658,8 @@ git commit -m "zdd-5 task 9: planned wakes, the first-wake budget and fallback, |
| type wakeBook interface { Blocks(acct, app, build, host string, served bool) bool; HeldDown(acct, app, build, host string) bool } |
| type wakeInputs struct { Route registry.Route; OptDir string; Policy AppPolicy; Book wakeBook; Demand func(version, stagedPath string) (string, string, error) } |
| func decideWake(in wakeInputs) (activator.WakePlan, error) |
| - func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePlan |
| + func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePlan // refuses while the host's attempt is running |
| + func (m *Manager) attemptRunning(host string) bool |
| func (m *Manager) OnGen(ev activator.GenEvent) |
| ``` |
| @@ -5652,13 +5838,39 @@ func TestASleepersConfigNeedsNoAttempt(t *testing.T) { |
| t.Fatalf("attempts %+v: a sleeper got a config attempt", calls) |
| } |
| } |
| + |
| +// Spec §2.9 (rev 17), §6.4, §8.3: an overlap attempt on a foreign-layout row |
| +// runs as a hold, and the activator says so (SwapResult.Mode). Its failure |
| +// is a hold build's failure: held down, so no wake boots the good build in |
| +// its place. And while an attempt runs for an instance, no wake of it is |
| +// planned, so a hold that went down is recorded before anything wakes it. |
| +// Recording under the dispatched ModeOverlap left it un-held, and the next |
| +// wake booted the good build against the database the hold build ran on. |
| +func TestAForeignRowsFailedHoldIsHeldDown(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + h.act.outcome = func(activator.Attempt) activator.SwapOutcome { return activator.SwapDown } |
| + h.act.mode = activator.ModeHold // the effective mode, as plan 3's Swap reports it |
| + gate := h.act.hold("a") |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "the attempt to start", func() bool { _, started, _ := h.act.snapshot(); return len(started) == 1 }) |
| + if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" { |
| + t.Fatalf("a wake was planned while the instance's attempt ran: %+v", p) |
| + } |
| + close(gate) |
| + waitFor(t, "the hold-down record", func() bool { return h.book.HeldDown("bab", "blog", "v2", "a") }) |
| + if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" || p.Boot.StagedPath != "" { |
| + t.Fatalf("after the hold went down the wake planned %+v, want refused (held down)", p) |
| + } |
| +} |
| ``` |
| Add `"time"` and `"github.com/carlosframework/platform/internal/release"` to that file's imports. |
| - [ ] **Step 2: Run them to see them fail** |
| -Run: `go test ./internal/rollout/ -run 'TestDecideWake|TestPlanBoots|TestAFellBack|TestOnlyASettled|TestAFailedHold|TestAWakeActivates|TestASleepersConfig' -count=1` |
| +Run: `go test ./internal/rollout/ -run 'TestDecideWake|TestPlanBoots|TestAFellBack|TestOnlyASettled|TestAFailedHold|TestAWakeActivates|TestASleepersConfig|TestAForeignRowsFailedHold' -count=1` |
| Expected: FAIL, `undefined: decideWake`. |
| - [ ] **Step 3: Write the planner and the outcome handler** |
| @@ -5749,6 +5961,13 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePla |
| if !Managed(rt) { |
| return activator.WakePlan{} |
| } |
| + if m.attemptRunning(rt.Host) { |
| + // The instance's attempt has not reported yet (spec §6.4, §8.1): a |
| + // hold that went down, or a swap whose entry stopped under it, is |
| + // recorded by done() before anything wakes the instance, so no wake |
| + // can boot the good build ahead of a hold-down record. |
| + return activator.WakePlan{Refuse: "an attempt is still running for this instance"} |
| + } |
| acct, app := release.RouteAccount(rt), rt.App |
| in := wakeInputs{Route: rt, OptDir: m.d.OptDir, Policy: m.policies.get(ctx, acct, app), Book: m.d.Book, |
| Demand: func(ver, path string) (string, string, error) { |
| @@ -5772,6 +5991,19 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePla |
| return plan |
| } |
| +// attemptRunning reports whether host has an attempt in flight: admitted |
| +// and not yet finished by done(). |
| +func (m *Manager) attemptRunning(host string) bool { |
| + s := m.schedulerOf(host) |
| + if s == nil { |
| + return false |
| + } |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + h := s.q.hosts[host] |
| + return h != nil && h.running != nil |
| +} |
| + |
| // OnGen is the activator's OnGen: a planned wake's outcome (spec §6.4, §8.2, |
| // §8.5). A healthy wake records the serving build and activates the config |
| // it started with; a first wake becomes the good build only once it has |
| @@ -6900,8 +7132,18 @@ Before a spawn: |
| } |
| } |
| ``` |
| -- `doClaimWake`: directly after the planning block Task 9 added there, before `policy := startPolicy{…}` (claimed wakes never reach `doWake`'s planning block): the same block, returning `nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err)`. |
| -- plan 3's `swapOverlap`: directly before its `g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen)` line (plan 3 Task 10; the successor is already in the operation set, so the file names it): |
| +- `doClaimWake`: directly after the planning block Task 9 added there, before `policy := startPolicy{…}` (claimed wakes never reach `doWake`'s planning block). Task 9's `plan` is local to that block, so this reads what the block stored on the entry: |
| + ```go |
| + a.mu.Lock() |
| + planned, pinned := e.plan.planned(), e.pin != nil |
| + a.mu.Unlock() |
| + if planned || pinned { |
| + if err := a.persistGenerations(); err != nil { |
| + return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err) |
| + } |
| + } |
| + ``` |
| +- plan 3's `swapOverlap`: directly before its `// Step 3a: the §7.4 checks again, at the exec` block (plan 3 Task 4), so the checks there run after the publication and immediately before `g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen)` (plan 3 Task 10; the successor is already in the operation set, so the file names it): |
| ```go |
| if err := a.persistGenerations(); err != nil { |
| a.dropMember(e, m) // restores the raised parent (leaveSet) |
| @@ -6999,7 +7241,7 @@ git commit -m "zdd-5 task 12: the generations file; GoodVersion and generations |
| - 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 RolloutReport struct { Hosts map[string]HostRollout; Apps map[string]AppRollout; Failures []RolloutFailure; Unavailable string } |
| 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 } |
| @@ -7162,6 +7404,10 @@ 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"` |
| + // Unavailable, when set, is why CARLOS_SWAP=overlap is on but the |
| + // scheduler could not arm; every managed start is refused meanwhile |
| + // (Task 14, refuseManaged). The rest of the report is then empty. |
| + Unavailable string `json:"unavailable,omitempty"` |
| } |
| // HostRollout is one managed instance. Attempt and Reason are the closed |
| @@ -7394,8 +7640,11 @@ git commit -m "zdd-5 task 13: the rollout report in box status" |
| 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 |
| + func rolloutStatusProbe() *status.RolloutReport // with the flag on and the scheduler unarmed: only Unavailable |
| var statusNudge chan struct{} // cap 1 |
| + var rolloutDown atomic.Pointer[string] |
| + func rolloutUnavailable() string |
| + func refuseManaged(svc *activator.Service, reason string) // flag on, required state unavailable: managed wakes refused, bounce legs stand down |
| ``` |
| - [ ] **Step 1: Write the failing tests** |
| @@ -7467,12 +7716,49 @@ func TestTheFlagOnArmsTheScheduler(t *testing.T) { |
| } |
| } |
| -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") |
| +// Spec §6.4, §8.3, §8.4: with the flag on and the scheduler's state |
| +// unavailable, managed starts are refused and the reason reported. Falling |
| +// back to today's live-link wakes and legacy convergence (a nil Planner) |
| +// would exec the failed installed build the unreadable records hold down. |
| +func TestUnavailableStateRefusesManagedStarts(t *testing.T) { |
| + _, pub, err := release.GenerateKey() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, tc := range map[string]struct { |
| + env map[string]string |
| + store func(context.Context) (release.Store, error) |
| + }{ |
| + "no deploy store": {map[string]string{"CARLOS_SWAP": "overlap", "CARLOS_RELEASE_PUBKEY": pub}, |
| + func(context.Context) (release.Store, error) { return nil, errors.New("no deployment store") }}, |
| + "no public key": {map[string]string{"CARLOS_SWAP": "overlap"}, |
| + func(context.Context) (release.Store, error) { return release.NewDirStore(t.TempDir()), nil }}, |
| + } { |
| + t.Run(name, func(t *testing.T) { |
| + svc, reg, dd := rolloutFixture(t) |
| + t.Cleanup(func() { rolloutDown.Store(nil) }) |
| + m := armRollout(context.Background(), func(k string) string { return tc.env[k] }, svc, reg, dd, nil, tc.store) |
| + if m != nil || armedRollout() != nil { |
| + t.Fatal("armed without its required state") |
| + } |
| + if svc.Planner == nil { |
| + t.Fatal("no Planner: the box fell back to today's live-link wakes") |
| + } |
| + managed := registry.Route{Host: "a.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Hibernate: true, Version: "v2"} |
| + if p := svc.Planner(context.Background(), managed); p.Refuse == "" { |
| + t.Fatalf("managed wake planned %+v, want refused", p) |
| + } |
| + unit := registry.Route{Host: "u.example.com", Kind: "instance", App: "blog", Account: "bab", Status: "ready", Unit: "blog.service"} |
| + if p := svc.Planner(context.Background(), unit); p.Refuse != "" || p.Boot.StagedPath != "" { |
| + t.Fatalf("an unmanaged wake was changed: %+v", p) |
| + } |
| + if r := rolloutStatusProbe(); r == nil || r.Unavailable == "" { |
| + t.Fatalf("status = %+v, want the reason reported", r) |
| + } |
| + if adoptBounceFn(svc) != nil { |
| + t.Fatal("the adopt bounce still stops instances whose next wake is refused") |
| + } |
| + }) |
| } |
| } |
| @@ -7494,7 +7780,7 @@ func TestRestartsFeedTheSchedulerWithTheirIdentity(t *testing.T) { |
| - [ ] **Step 2: Run them to see them fail** |
| -Run: `TMPDIR=/tmp/k go test ./internal/host/ -run 'TestTheFlagOff|TestTheFlagOn|TestAnUnusableStore|TestRestartsFeed' -count=1` |
| +Run: `TMPDIR=/tmp/k go test ./internal/host/ -run 'TestTheFlagOff|TestTheFlagOn|TestUnavailableState|TestRestartsFeed' -count=1` |
| Expected: FAIL, `undefined: armRollout`. |
| - [ ] **Step 3: Write the arming** |
| @@ -7529,12 +7815,48 @@ func armedRollout() *rollout.Manager { return rolloutMgr.Load() } |
| // five-minute tick. Never sent to with the flag off. |
| var statusNudge = make(chan struct{}, 1) |
| +// rolloutDown is why CARLOS_SWAP=overlap is set but the scheduler could not |
| +// arm; nil when it armed, or when the flag is off. |
| +var rolloutDown atomic.Pointer[string] |
| + |
| +// rolloutUnavailable is rolloutDown's reason, "" when there is none. |
| +func rolloutUnavailable() string { |
| + if p := rolloutDown.Load(); p != nil { |
| + return *p |
| + } |
| + return "" |
| +} |
| + |
| +// refuseManaged is the flag on with the scheduler's required state |
| +// unavailable (spec §6.4, §8.3, §8.4). The legacy paths are NOT a fallback: |
| +// a wake from the live link could exec a build the unreadable failure book |
| +// holds down, including a hold build after its migration failed. So every |
| +// managed wake is refused before the lease, with the reason, and an |
| +// unmanaged one (a named unit, a unit-backed row) wakes as today. The |
| +// bounce-based legs stand down too (fastConvergeLoop, adoptBounceFn, |
| +// restartRequestLeg, syncExecSerialized): a stop whose next wake is refused |
| +// would only take an instance down. The reason is published in the box's |
| +// status (rolloutStatusProbe) until the agent restarts with the state |
| +// readable. |
| +func refuseManaged(svc *activator.Service, reason string) { |
| + log.Printf("agent: %s=overlap but %s: refusing every managed start until the agent restarts with it readable", rollout.SwapEnv, reason) |
| + r := reason |
| + rolloutDown.Store(&r) |
| + svc.Planner = func(_ context.Context, rt registry.Route) activator.WakePlan { |
| + if !rollout.Managed(rt) { |
| + return activator.WakePlan{} |
| + } |
| + return activator.WakePlan{Refuse: "overlap swaps are on but " + reason} |
| + } |
| +} |
| + |
| // 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). |
| +// public key or an unreadable failure book, it returns nil and refuses every |
| +// managed start (refuseManaged): each of those is a fact the scheduler |
| +// cannot do without (the canonical demand, the retry signal, the records), |
| +// and today's paths would ignore the failure 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) { |
| @@ -7542,17 +7864,17 @@ func armRollout(ctx context.Context, getenv func(string) string, svc *activator. |
| } |
| 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) |
| + refuseManaged(svc, "the deploy store is unavailable ("+err.Error()+")") |
| 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) |
| + refuseManaged(svc, err.Error()) |
| 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) |
| + refuseManaged(svc, "the failure book is unreadable ("+err.Error()+")") |
| return nil |
| } |
| platform := runtime.GOOS + "-" + runtime.GOARCH |
| @@ -7636,10 +7958,14 @@ func feedRestarts(routes []registry.Route, moved [][2]string, scoped []scopedRes |
| return n |
| } |
| -// rolloutStatusProbe feeds status.Probes.Rollout: nil with the flag off. |
| +// rolloutStatusProbe feeds status.Probes.Rollout: nil with the flag off, |
| +// and only the reason when the flag is on but the scheduler could not arm. |
| func rolloutStatusProbe() *status.RolloutReport { |
| m := armedRollout() |
| if m == nil { |
| + if why := rolloutUnavailable(); why != "" { |
| + return &status.RolloutReport{Unavailable: why} |
| + } |
| return nil |
| } |
| r := m.Status() |
| @@ -7689,6 +8015,9 @@ with: |
| ```go |
| if m := armedRollout(); m != nil { |
| go runConvergeLoop(rolloutLegFor(m, reg), adoptedConvergeEvery, nil) |
| + } else if rolloutUnavailable() != "" { |
| + // The flag is on but the scheduler could not arm: legacy convergence |
| + // would stop instances whose next wake is refused (refuseManaged). |
| } else if leg := convergeLegFor(svc, reg); leg != nil { |
| go runConvergeLoop(leg, adoptedConvergeEvery, nil) |
| } |
| @@ -7700,6 +8029,9 @@ In `adoptBounceFn`, before `return svc.BounceAdopted`, add: |
| if m := armedRollout(); m != nil { |
| return m.Adopted |
| } |
| + if rolloutUnavailable() != "" { |
| + return nil // no adopt bounce while managed starts are refused |
| + } |
| ``` |
| In `syncExecSerialized`, before its final `return appconfig.SyncExec(...)`, add: |
| @@ -7715,6 +8047,12 @@ In `syncExecSerialized`, before its final `return appconfig.SyncExec(...)`, add: |
| } |
| return appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, dd, os.Getenv("PLATFORM_APPS_DOMAIN"), cb, opts...) |
| } |
| + if rolloutUnavailable() != "" { |
| + // Delivered and acknowledged as today, but no restart: the wake |
| + // after it would be refused (refuseManaged). |
| + cb := appconfig.Callbacks{Restart: func(string) error { return nil }, Ack: 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 |
| @@ -7729,13 +8067,13 @@ with |
| ```go |
| if m := armedRollout(); m != nil { |
| feedRestarts(routes, moved, scoped, mem.gen, m.Restart) |
| - } else { |
| + } else if rolloutUnavailable() == "" { |
| 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) }`. |
| +and wrap the later `bounceScopedRoutes(routes, scoped, bounce)` the same way: `if armedRollout() == nil && rolloutUnavailable() == "" { bounceScopedRoutes(routes, scoped, bounce) }`. (With the flag on and the scheduler unarmed, an exec restart is not performed; the box's status names why.) |
| In `statusTick`'s probe block, after `p.ReleaseRefusals = releaseRefusalProbe()`, add `p.Rollout = rolloutStatusProbe()`. In `statusLoop`, replace `time.Sleep(delay)` with: |