amadan
plans: Astra P3 fixed (4 partials completed, 6 new) with one helper per concern: attempt gate, outcome classifier, termination proof
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
4 files changed,
+943
−156
docs/superpowers/plans/2026-10-07-zero-downtime-deploys-0-index.md+51 −3docs/superpowers/plans/2026-10-07-zero-downtime-deploys-1-generations.md+123 −33docs/superpowers/plans/2026-10-07-zero-downtime-deploys-3-swap.md+379 −74docs/superpowers/plans/2026-10-07-zero-downtime-deploys-5-scheduler.md+390 −46
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 ad0dfb9d8..2658f6094 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 |
| @@ -361,6 +361,46 @@ findings. |
| planner refuses a wake while the host's attempt is running. |
| - **8** (plan 5 Task 12): the claimed-wake pin publication reads `e.plan`. |
| +## Fixes for Astra's P3 review of the plans (2026-10-07) |
| + |
| +`CODEX-FINDINGS-ZDD-PLANS-P3.md`: four partial P2 fixes and six new |
| +findings. Where several paths re-implemented one concern, they now share |
| +one helper: |
| + |
| +- **The attempt gate** (`attemptCurrentLocked`, plan 3 Task 3) answers "is |
| + this attempt still current" at every exec point and completion path. |
| +- **The outcome classifier** (`classifyStep`, plan 3 Task 4) separates a |
| + lifecycle cancellation and a stale attempt from a failure of the build. |
| +- **The termination proof** (`genTerminated`, plan 3 Task 2) is asked by |
| + every path that would give up a generation's lease or set place. |
| + |
| +Per finding: |
| + |
| +- **P1 #1, completed** (plan 3 Task 2): `cleanupFailedWake` keeps the lease |
| + when the leader itself did not exit, and the retry re-stops it. |
| +- **P1 #6, completed** (plan 5 Tasks 5, 8): the failing pointer is read |
| + before the attempt runs (`dispatch.FailedUnder`), so no blocking read |
| + sits between a failure and its record. |
| +- **P2 #3, completed** (plan 3 Task 7): every sidecar a start leaves behind |
| + is either installed for the joined stop to prove, or discarded |
| + synchronously before the join closes (`discardSidecar`, |
| + `sidecarJoinBound`). |
| +- **P2 #4, completed** (plan 3 Task 10): a claims-on successor re-runs the |
| + gate after `Read`/`Prepare` and inside the work at the exec |
| + (`spawnClaimedGen`'s `before`). |
| +- **1** (plan 5 Task 8): `done` publishes before it frees the slot, and |
| + every serving- and good-build write is acknowledged and serialised. |
| +- **2** (plan 5 Tasks 10, 14): a cold managed host is installed before its |
| + wake is planned; no managed wake runs through an empty plan. |
| +- **3** (plan 5 Task 9): `spawnPlanned`/`spawnPlannedWorker` use the |
| + classifier; a cancelled wake records nothing and falls back to nothing. |
| +- **4** (plan 5 Task 8): `observe` always updates the desired build and its |
| + revision; only serving state and admission wait for a stable entry. |
| +- **5** (plan 5 Task 12): `generationsLocked` pins `unprovenWakes`. |
| +- **6** (plan 1 Task 5): the body is held from the transport until the |
| + request is on a connection to the chosen generation; `dialGen` rejects a |
| + connection whose generation changed before it completed. |
| + |
| ## Cross-plan interface ledger |
| Every type, function, method or field that one plan defines and another |
| @@ -389,7 +429,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` (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` | |
| +| `GenPools` internals: `genPool{tr, gen}`, `dialGen` (dials only its own generation's socket and rejects a new connection unless that generation is still current, rev 17 §7.2), `genMovedError`, `heldBody` (the transport cannot read or close the body before the request is on a connection, so a switch retry never needs `GetBody`; `replayable` is gone) | 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 | |
| @@ -430,9 +470,13 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | `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 | |
| -| `func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error`; `errSwapNotAwake`, `errSwapBusy`, `errSwapKind`, `errSwapStopped`; `func abandon(err error) SwapResult` | P3 T3, T4 | P5 T1 L172, T11 L5830 (`ReplaceWorker`) | |
| +| `func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error` (now `attemptCurrentLocked(e, at, newest, false)`); `errSwapNotAwake`, `errSwapBusy`, `errSwapKind`, `errSwapStopped`; `func abandon(err error) SwapResult` | P3 T3, T4 | P5 T1 L172, T11 (`ReplaceWorker`, its `attemptGate`) | |
| +| **The attempt gate**: `func (a *Service) attemptCurrentLocked(e *actEntry, at Attempt, newest func(string) uint64, waking bool) error`; `memberCurrentLocked(e, m, mctx, at, newest)`. Every exec point and completion path asks it: overlap Step 3a, the switch, `spawnSuccessor`'s claims `before` hook, a hold's `pinCheck` and result, `ReplaceWorker` | P3 T3 | P3 T4, T6, T10; P5 T11 (through `swapChecksLocked`) | |
| +| **The outcome classifier**: `type stepOutcome`; `stepOK`, `stepCancelled`, `stepStale`, `stepFailed`; `func classifyStep(lifecycle context.Context, err error) stepOutcome`; `stepResult` | P3 T4 | P3 T4, T6; P5 T9 (`spawnPlanned`, `spawnPlannedWorker`: only `stepFailed` records a failure or falls back) | |
| +| **The termination proof**: `func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool` | P3 T2 | every lease- or set-giving path in P3 (member stop, reaper, `stop()`, `reapCurrentGen`, `cleanupFailedWake`, `reapUnprovenWakesLocked`) | |
| +| `func (a *Service) discardSidecar(sc Proc)`; `func sidecarJoinBound(gens bool) time.Duration` | P3 T7 | P3 only | |
| | `actEntry.ops opSet` (`members map[uint64]*genMember`, `genMember.gen`); `addMemberLocked`; `leaveSet` | P3 T2 | P5 T1, T12 L6170, L6366, L6419 | |
| -| `swapOverlap`'s `g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen)` | P3 T10 | P5 T12 (persist point; a failed write abandons the attempt there) | |
| +| `swapOverlap`'s `// Step 3a` gate block, before `g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen, gate)` (`spawnSuccessor` and `startClaimedSuccessor` gain `gate func() error`) | P3 T4, T10 | P5 T12 (persist point, before Step 3a; a failed write abandons the attempt there) | |
| | `Service.genHealth func(ctx context.Context, g Gen) error` (test seam) | P3 T4 | P5 T1 L186, T9 L4344 | |
| | `func observeSettle(ctx context.Context, p Proc, window time.Duration) settleResult`; `settleOK`, `settleExited`, `settleCancelled` | P3 T5 | P5 T1 L174, T9 L4859, T11 L5883 | |
| | `actEntry.pin *Gen`; `func (a *Service) startPinned(ctx, rt, pin Gen) (Proc, string, Gen, error)` (plan 5 Task 9 adds `healthBy time.Time`). Its health check is the only one (StartStaged waits for none), and a failed build comes back with its `Proc` | P3 T6 | P5 T1 L173, T9 | |
| @@ -469,6 +513,10 @@ many line numbers in plans 1, 3 and 5; find each use by its quoted name. |
| | `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 | |
| +| `Deps.Materialize func(ctx, host string) error` (`Plan` installs a cold managed host first; `decideWake` never returns an empty plan for a managed host); `rolloutMaterialize` (`internal/host`) | P5 T8, T10, T14 | P5 only | |
| +| `dispatch.FailedUnder time.Time` (the failing pointer's PromotedAt, read before the attempt runs) | P5 T5, T8 | P5 only | |
| +| `Manager.served`, `publishServing`, `acknowledged`, `writeGood`/`writeGoodLocked`, `Manager.goodMu` (every serving- and good-build write is for the generation the activator serves, serialised; `done` publishes before it frees the slot) | P5 T8 | P5 T10 (`OnGen`) | |
| +| `generationsLocked` includes `unprovenWakes` | P5 T12 | consumes P3 T2's `unprovenWakes` | |
| | `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 | |
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 684f5c7c5..601279e82 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 |
| @@ -2230,7 +2230,8 @@ git commit -m "zdd plan 1 task 4: the canonical socket path is a symlink to the |
| ```go |
| var errNoGeneration error // wraps syscall.ENOENT |
| func dialUnixRetrying(ctx context.Context, d *net.Dialer, resolve func() (string, bool)) (net.Conn, error) |
| - type genMovedError struct{ to GenInfo } // a pool's own generation is gone and another is current |
| + type genMovedError struct{ to GenInfo } // a pool's own generation stopped being current before its dial completed |
| + type heldBody struct{ … } // the request body, held from the transport until the request is on a connection |
| func dialGen(ctx context.Context, d *net.Dialer, src GenSource, host string, own GenInfo) (net.Conn, error) |
| type genPool struct{ tr *http.Transport; gen GenInfo } |
| type GenPools struct{ … } |
| @@ -2244,7 +2245,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 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. |
| + **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). Every new connection is checked against `ServingGen` once connected (spec §7.2, rev 17): if own is no longer current, the connection is closed and the dial gives up with `genMovedError`. That covers a switch before the dial completes and a foreign row's shared canonical path alike. `RoundTrip` then sends the request again on the current generation's pool. The body is behind `heldBody`, which the transport can neither read nor close before it is writing on a connection, so the generation is settled before a byte of the body is consumed, and a streaming POST with no `GetBody` survives a switch (P3 review finding 6). `replayable` is gone. |
| - [ ] **Step 1: Write the failing tests** |
| @@ -2495,7 +2496,10 @@ func TestAppSocketProxyFollowsTheCurrentGeneration(t *testing.T) { |
| // AppSocket, the request reached B while it was stamped and tagged as A. |
| func TestAppSocketProxyAttributesTheConnectionsGeneration(t *testing.T) { |
| const host = "inst.example.com" |
| - t.Run("the retiring generation still accepts", func(t *testing.T) { |
| + // Rev 17: a switch that lands before the dial completes sends the request |
| + // to the new current generation, even though the retiring one still |
| + // accepts. Its response is attributed to the generation that served it. |
| + t.Run("a switch before the dial completes", func(t *testing.T) { |
| dir := genSockDir(t) |
| a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", "")) |
| b := startGenApp(t, filepath.Join(dir, "b.sock"), body("B", "")) |
| @@ -2510,13 +2514,55 @@ func TestAppSocketProxyAttributesTheConnectionsGeneration(t *testing.T) { |
| rec := httptest.NewRecorder() |
| p.ServeHTTP(rec, r) |
| _, n, _ := servedGen(r) |
| - if rec.Body.String() != "A" || rec.Header().Get("X-Carlos-Version") != "v1" || n != 1 { |
| - t.Fatalf("served %q, version %q, generation %d: want A, v1, 1 (the pool's own generation)", |
| + 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 (retried on the current generation)", |
| rec.Body.String(), rec.Header().Get("X-Carlos-Version"), n) |
| } |
| - if got := b.opened.Load(); got != 0 { |
| - t.Fatalf("generation 2 got %d connection(s) through generation 1's pool", got) |
| + if got := b.opened.Load(); got < 1 { |
| + t.Fatal("generation 2 was never reached") |
| + } |
| + _ = a |
| + }) |
| + // A proxied POST streams its body: no GetBody, unknown length. A switch |
| + // before the dial completes must not turn it into a 502. The generation |
| + // is settled before a byte of the body is read, so the body goes, whole, |
| + // to the current generation. Retrying only replayable requests failed it. |
| + t.Run("a streaming POST across a switch", func(t *testing.T) { |
| + dir := genSockDir(t) |
| + a := startGenApp(t, filepath.Join(dir, "a.sock"), body("A", "")) |
| + echo := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { |
| + b, _ := io.ReadAll(r.Body) |
| + io.WriteString(w, "B:"+string(b)) |
| + }) |
| + b := startGenApp(t, filepath.Join(dir, "b.sock"), echo) |
| + src := newFakeGens(true) |
| + src.switchTo(host, GenInfo{Socket: a.sock, N: 1, Build: "v1"}) |
| + var once sync.Once |
| + src.onServing = func() { |
| + once.Do(func() { |
| + a.srv.Close() // generation 1 has gone: its socket file with it |
| + src.switchTo(host, GenInfo{Socket: b.sock, N: 2, Build: "v2"}) |
| + }) |
| + } |
| + pr, pw := io.Pipe() |
| + go func() { |
| + io.WriteString(pw, "hello, ") |
| + io.WriteString(pw, "world") |
| + pw.Close() |
| + }() |
| + req := httptest.NewRequest("POST", "http://inst.example.com/form", pr) |
| + req.ContentLength = -1 |
| + if req.GetBody != nil { |
| + t.Fatal("fixture: the request must have no GetBody") |
| + } |
| + r, _ := withOutcome(req) |
| + rec := httptest.NewRecorder() |
| + appSocketProxy(host, src, NewGenPools()).ServeHTTP(rec, r) |
| + if rec.Code != http.StatusOK || rec.Body.String() != "B:hello, world" || rec.Header().Get("X-Carlos-Version") != "v2" { |
| + t.Fatalf("status %d, body %q, version %q: want 200, the whole body at generation 2, v2", |
| + rec.Code, rec.Body.String(), rec.Header().Get("X-Carlos-Version")) |
| } |
| + _ = b |
| }) |
| t.Run("the sampled generation is gone", func(t *testing.T) { |
| dir := genSockDir(t) |
| @@ -2683,6 +2729,7 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "io" |
| "net" |
| "net/http" |
| "net/http/httputil" |
| @@ -2766,21 +2813,18 @@ 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 { |
| - // 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. |
| + // Spec §7.2 (rev 17): if the generation stops being current |
| + // before the dial completes, the request is retried against the |
| + // new current generation. So every new connection is checked |
| + // once connected: own must still be current. This also covers a |
| + // foreign row, whose generations share one canonical path, so a |
| + // connection on it could have reached the replacement. A |
| + // connection already made (an idle one in own's pool) is not |
| + // re-checked: the request that chose own's pool sampled own. |
| 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} |
| @@ -2897,30 +2941,44 @@ type genRoundTripper struct { |
| const genRetries = 2 |
| func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { |
| + // The generation is chosen, and any switch retried, before a byte of the |
| + // body is consumed (spec §7.2, rev 17). The transport gets the body |
| + // behind heldBody: it reads it only once it is writing the request on a |
| + // connection dialGen has proven to be the chosen generation's, and a |
| + // Close it makes after a failed dial is held. So a dial that finds the |
| + // generation moved leaves the body untouched, and the request goes again |
| + // on the current generation's pool, a streaming POST with no GetBody |
| + // included. |
| + var hb *heldBody |
| + if req.Body != nil && req.Body != http.NoBody { |
| + hb = &heldBody{rc: req.Body} |
| + req.Body = hb |
| + } |
| for attempt := 0; ; attempt++ { |
| g, ok := t.src.ServingGen(t.host) |
| if !ok { |
| // No current generation: the dial waits for one inside its retry |
| // window. Nothing is pooled and nothing is stamped or cached, |
| // because which generation answers is not known here. |
| - return genTransportFor(t.host, t.src, nil).RoundTrip(req) |
| + resp, err := genTransportFor(t.host, t.src, nil).RoundTrip(req) |
| + if err != nil && hb != nil { |
| + hb.release() |
| + } |
| + return resp, err |
| } |
| gp := t.pools.pool(t.host, g, t.src) |
| resp, err := gp.tr.RoundTrip(req) |
| var moved *genMovedError |
| - if errors.As(err, &moved) && attempt < genRetries && replayable(req) { |
| - // The generation this request chose has gone and another is |
| - // current. Nothing was sent: send it again, on the current one. |
| - if req.GetBody != nil { |
| - body, berr := req.GetBody() |
| - if berr != nil { |
| - return nil, err |
| - } |
| - req.Body = body |
| - } |
| + if errors.As(err, &moved) && attempt < genRetries && (hb == nil || !hb.wasRead()) { |
| + // The generation this request chose stopped being current before |
| + // its dial completed. Nothing was sent and nothing of the body |
| + // was read: send it again, on the current one. |
| continue |
| } |
| if err != nil { |
| + if hb != nil { |
| + hb.release() |
| + } |
| return nil, err |
| } |
| // X-Carlos-Version names the build of the generation that served the |
| @@ -2934,12 +2992,44 @@ func (t *genRoundTripper) RoundTrip(req *http.Request) (*http.Response, error) { |
| } |
| } |
| -// replayable reports whether req can be sent again after a dial that never |
| -// connected: it has no body, or a way to make the body again. |
| -func replayable(req *http.Request) bool { |
| - return req.Body == nil || req.Body == http.NoBody || req.GetBody != nil |
| +// heldBody is a request body the transport cannot consume or close before |
| +// the request is on a connection. The transport reads a body only while it |
| +// writes the request on a dialled connection, and closes it on any error, |
| +// including a dial that never connected. Until the first Read, that Close |
| +// is held, so the same body can go again on another generation's pool. |
| +// After it, Close passes through as usual. |
| +type heldBody struct { |
| + rc io.ReadCloser |
| + mu sync.Mutex |
| + read bool |
| } |
| +func (b *heldBody) Read(p []byte) (int, error) { |
| + b.mu.Lock() |
| + b.read = true |
| + b.mu.Unlock() |
| + return b.rc.Read(p) |
| +} |
| + |
| +func (b *heldBody) Close() error { |
| + b.mu.Lock() |
| + read := b.read |
| + b.mu.Unlock() |
| + if !read { |
| + return nil // a failed dial's close: the body may still be sent |
| + } |
| + return b.rc.Close() |
| +} |
| + |
| +func (b *heldBody) wasRead() bool { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + return b.read |
| +} |
| + |
| +// release closes the body for good: the request has finished with an error. |
| +func (b *heldBody) release() { _ = b.rc.Close() } |
| + |
| // appSocketProxy is unixProxy for an activator-managed route with |
| // generations on: the same Rewrite, Host preservation, upgrade pass-through |
| // and error handler, over genRoundTripper instead of one fixed socket. |
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 8ed44204f..3d14ddad8 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 |
| @@ -1019,6 +1019,7 @@ git commit -m "zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged |
| 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(rt registry.Route, m *genMember) bool |
| + func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool // THE termination proof: leader exited and child empty; every lease- or set-giving path asks it |
| 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 |
| @@ -1551,13 +1552,26 @@ func (a *Service) stopMember(e *actEntry, m *genMember) bool { |
| if err != nil { |
| log.Printf("activator: stop generation %d: %v", m.gen.N, err) |
| } |
| - if !m.proc.Exited() { |
| - return false |
| - } |
| a.mu.Lock() |
| rt := e.route |
| a.mu.Unlock() |
| - return a.genChildEmpty(rt, m) // swapsim:M7 |
| + return a.genTerminated(rt, m.gen, m.proc) // swapsim:M7 |
| +} |
| + |
| +// genTerminated is the ONE termination proof for a generation (spec §7.3, |
| +// §7.4): its leader has exited AND its child is proven empty. Every path |
| +// that would give up a generation's lease or its place in the operation |
| +// set asks it: a member's stop, the reaper, the current generation's stop, |
| +// a pending stop's finish, a failed wake and its retained retry. A nil |
| +// process never ran. |
| +func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool { |
| + if p == nil { |
| + return true |
| + } |
| + if !p.Exited() { |
| + return false |
| + } |
| + return a.genChildEmpty(rt, &genMember{gen: g, proc: p}) |
| } |
| // genChildEmpty proves m's generation child empty, after its leader has |
| @@ -1677,7 +1691,7 @@ func (a *Service) reapExited(e *actEntry, m *genMember) { |
| a.mu.Lock() |
| rt := e.route |
| a.mu.Unlock() |
| - if a.genChildEmpty(rt, m) { |
| + if a.genTerminated(rt, m.gen, m.proc) { |
| a.leaveSet(e, m) |
| return |
| } |
| @@ -1699,7 +1713,7 @@ 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}) { |
| + if !a.genTerminated(rt, g, p) { |
| a.mu.Lock() |
| e.genReaping = false |
| a.mu.Unlock() |
| @@ -1741,7 +1755,13 @@ func (a *Service) reapUnprovenWakesLocked() { |
| } |
| u.busy = true |
| go func(host string, u *unprovenWake) { |
| - proven := a.genChildEmpty(u.rt, &genMember{gen: u.gen, proc: u.proc}) |
| + if u.proc != nil && !u.proc.Exited() { |
| + // The leader itself did not stop: stop it again, bounded. |
| + ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| + _ = u.proc.Stop(ctx) |
| + cancel() |
| + } |
| + proven := a.genTerminated(u.rt, u.gen, u.proc) |
| if proven { |
| ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| if u.rep != nil { |
| @@ -1828,7 +1848,7 @@ At the end of `stop()`, replace `a.finishStop(e, outcome)` (line 2268) with: |
| a.mu.Lock() |
| rt, g := e.route, e.gen |
| a.mu.Unlock() |
| - if !a.genChildEmpty(rt, &genMember{gen: g, proc: e.proc}) { // swapsim:M8 |
| + if !a.genTerminated(rt, g, 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 |
| @@ -1897,11 +1917,13 @@ with |
| 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. |
| + if a.gens && !a.genTerminated(rt, w.gen, w.proc) { |
| + // The failed generation is not proven terminated (spec §7.4): |
| + // its leader did not exit (ErrStillRunning, a lost stop) or its |
| + // child is not proven empty. Ownership and the lease are kept, |
| + // the sweep retries the stop and 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() |
| @@ -2000,7 +2022,9 @@ git commit -m "zdd plan 3 task 2: the entry's operation set, the barrier in stop |
| var errSwapNotAwake, errSwapLeaseLost, errSwapFenced, errSwapShutdown, errSwapSuperseded error |
| func (a *Service) selectWork(ctx context.Context, host string) (hostwrite.Work, error) |
| func (a *Service) admit(ctx context.Context, work hostwrite.Work, fn func() error) error |
| - func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error |
| + func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error // = attemptCurrentLocked(e, at, newest, false) |
| + func (a *Service) attemptCurrentLocked(e *actEntry, at Attempt, newest func(string) uint64, waking bool) error // THE attempt gate |
| + func (a *Service) memberCurrentLocked(e *actEntry, m *genMember, mctx context.Context, at Attempt, newest func(string) uint64) error |
| // test helpers reused by Tasks 4 and 6: swapWork, swapSelector, errClaimChanged |
| ``` |
| @@ -2231,10 +2255,24 @@ func (a *Service) admit(ctx context.Context, work hostwrite.Work, fn func() erro |
| // newest. The agent's own shutdown counts as a stop in progress. newest is |
| // called with a.mu held. The caller holds a.mu. |
| func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) uint64) error { |
| + return a.attemptCurrentLocked(e, at, newest, false) |
| +} |
| + |
| +// attemptCurrentLocked is THE attempt gate (spec §6.3, §7.4): every exec |
| +// point and every completion path of every attempt asks it, through |
| +// swapChecksLocked, memberCurrentLocked, a hold's pinCheck, a claims-on |
| +// spawn's before hook and a socketless respawn's attemptGate. waking is a |
| +// hold's pinned wake, whose entry is still waking and holds no lease yet: |
| +// the entry must be the host's, and the agent, the fences and the revision |
| +// are checked. Otherwise the entry must be awake with its lease. The caller |
| +// holds a.mu. |
| +func (a *Service) attemptCurrentLocked(e *actEntry, at Attempt, newest func(string) uint64, waking bool) error { |
| switch { |
| - case e == nil || a.entries[at.Host] != e || e.state != actAwake: // swapsim:M5 |
| + case e == nil || a.entries[at.Host] != e: |
| + return errSwapNotAwake |
| + case !waking && e.state != actAwake: // swapsim:M5 |
| return errSwapNotAwake |
| - case e.rep == nil || e.rep.Err() != nil: |
| + case !waking && (e.rep == nil || e.rep.Err() != nil): |
| return errSwapLeaseLost |
| case a.fences[at.Host] != 0: |
| return errSwapFenced |
| @@ -2245,6 +2283,20 @@ func (a *Service) swapChecksLocked(e *actEntry, at Attempt, newest func(string) |
| } |
| return nil |
| } |
| + |
| +// memberCurrentLocked is the gate for an attempt with a member in the |
| +// operation set (an overlap's successor): the attempt is current, and its |
| +// member is still in the set and not cancelled by the barrier. The caller |
| +// holds a.mu. |
| +func (a *Service) memberCurrentLocked(e *actEntry, m *genMember, mctx context.Context, at Attempt, newest func(string) uint64) error { |
| + if err := a.attemptCurrentLocked(e, at, newest, false); err != nil { |
| + return err |
| + } |
| + if e.ops.members[m.gen.N] != m || mctx.Err() != nil { |
| + return errSwapStopped |
| + } |
| + return nil |
| +} |
| ``` |
| - [ ] **Step 4: Run the tests to verify they pass** |
| @@ -2275,6 +2327,9 @@ git commit -m "zdd plan 3 task 3: swap admission through hostwrite, claims off a |
| func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, newest func(string) uint64) SwapResult |
| func (a *Service) attemptCtx(ctx context.Context) (context.Context, context.CancelFunc) |
| func (a *Service) genHealthy(ctx context.Context, g Gen) error |
| + type stepOutcome int; const stepOK, stepCancelled, stepStale, stepFailed stepOutcome |
| + func classifyStep(lifecycle context.Context, err error) stepOutcome // THE outcome classifier: only stepFailed is a failure of the build |
| + func stepResult(lifecycle context.Context, err error) (SwapResult, bool) |
| // Service gains: genHealth func(ctx context.Context, g Gen) error (test seam) |
| var errSwapOff, errSwapBusy, errSwapKind, errSwapStopped, errSuccessorExited, errSwapMode error |
| // test fixture: fakeSwapBackend, genServingProc (named apart from plan 1 Task 4's servingProc test type, which is in the same package), swapFixture, newSwapFixture, setCurrentGen, (*Service).genStateForTest, indexPrefix |
| @@ -2335,6 +2390,9 @@ type fakeSwapBackend struct { |
| // onRaise, when set, runs inside RaiseGenParent: a test's window between |
| // an attempt's admission and its exec. |
| onRaise func() |
| + // genStuck makes every generation spawned from then on a stuck |
| + // fakeProc: its Stop answers ErrStillRunning until exit(). |
| + genStuck bool |
| } |
| func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen Gen) (Gen, Proc, error) { |
| @@ -2358,7 +2416,10 @@ func (g *fakeSwapBackend) StartStaged(ctx context.Context, _ registry.Route, gen |
| gen.Socket = filepath.Join(g.genSocketDir, fmt.Sprintf("g%d.sock", gen.N)) |
| } |
| gen.Control = gen.Socket + ".control" |
| - p := &fakeProc{label: fmt.Sprintf("gen:%d", gen.N), rec: g.fakeBackend} |
| + g.gmu.Lock() |
| + stuck := g.genStuck |
| + g.gmu.Unlock() |
| + p := &fakeProc{label: fmt.Sprintf("gen:%d", gen.N), rec: g.fakeBackend, stuck: stuck} |
| var proc Proc = p |
| if g.genServe != nil { |
| proc = &genServingProc{fakeProc: p, stop: g.genServe(gen)} |
| @@ -2785,6 +2846,38 @@ func TestSwapWitnessNoAdmissionOntoAStoppingEntry(t *testing.T) { |
| } |
| } |
| +// The one outcome classifier (spec §6.4, §7.4): only a failure of the build |
| +// is stepFailed. A cancelled lifecycle, an intentional stop and a refusal |
| +// by the attempt gate record nothing; a health deadline that expires is the |
| +// build's. |
| +func TestClassifyStep(t *testing.T) { |
| + live, ended := context.Background(), func() context.Context { |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + cancel() |
| + return ctx |
| + }() |
| + for _, tc := range []struct { |
| + name string |
| + ctx context.Context |
| + err error |
| + want stepOutcome |
| + }{ |
| + {"ok", live, nil, stepOK}, |
| + {"the lifecycle ended", ended, errors.New("dial: connection refused"), stepCancelled}, |
| + {"cancelled", live, fmt.Errorf("health: %w", context.Canceled), stepCancelled}, |
| + {"fenced", live, fmt.Errorf("wake: %w", ErrFenced), stepCancelled}, |
| + {"the barrier", live, errSwapStopped, stepCancelled}, |
| + {"superseded", live, fmt.Errorf("pin: %w", errSwapSuperseded), stepStale}, |
| + {"lease lost", live, errSwapLeaseLost, stepStale}, |
| + {"a health deadline", live, fmt.Errorf("health: %w", context.DeadlineExceeded), stepFailed}, |
| + {"HTTP 500", live, errors.New("GET / answered 500"), stepFailed}, |
| + } { |
| + if got := classifyStep(tc.ctx, tc.err); got != tc.want { |
| + t.Errorf("%s: %d, want %d", tc.name, got, tc.want) |
| + } |
| + } |
| +} |
| + |
| // 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 |
| @@ -2897,6 +2990,51 @@ var ( |
| func abandon(err error) SwapResult { return SwapResult{Outcome: SwapAbandoned, Reason: err.Error()} } |
| func failSwap(err error) SwapResult { return SwapResult{Outcome: SwapFailed, Reason: err.Error()} } |
| +// stepOutcome is how one step of an attempt (a spawn, a health check, a |
| +// readiness or settle window, a pinned wake) ended. |
| +type stepOutcome int |
| + |
| +const ( |
| + stepOK stepOutcome = iota |
| + stepCancelled // the attempt's lifecycle ended it: its context, the barrier, a fence, an intentional stop |
| + stepStale // the attempt gate refused it: a newer attempt, a halted cohort, a lost lease, shutdown |
| + stepFailed // the build failed: spawn, health or readiness |
| +) |
| + |
| +// classifyStep is THE outcome classifier (spec §6.4, §7.4): every path |
| +// that might record a build failure asks it first, the swaps here and plan |
| +// 5's planned wakes alike. Only stepFailed is a failure of the build; a |
| +// cancelled or stale step records nothing and falls back to nothing. |
| +// lifecycle is the attempt's own context (nil when the caller has none). A |
| +// health deadline that expires is the build's failure (DeadlineExceeded); |
| +// the lifecycle's own end, or a cancellation, is not. |
| +func classifyStep(lifecycle context.Context, err error) stepOutcome { |
| + switch { |
| + case err == nil: |
| + return stepOK |
| + case lifecycle != nil && lifecycle.Err() != nil, |
| + errors.Is(err, context.Canceled), errors.Is(err, errSwapStopped), errors.Is(err, ErrFenced): |
| + return stepCancelled |
| + case errors.Is(err, errSwapSuperseded), errors.Is(err, errSwapNotAwake), errors.Is(err, errSwapLeaseLost), |
| + errors.Is(err, errSwapFenced), errors.Is(err, errSwapShutdown): |
| + return stepStale |
| + } |
| + return stepFailed |
| +} |
| + |
| +// stepResult is classifyStep for a swap step: a cancelled or stale step |
| +// abandons the attempt (over is true); a failure of the build is the |
| +// caller's to report (over is false). |
| +func stepResult(lifecycle context.Context, err error) (SwapResult, bool) { |
| + switch classifyStep(lifecycle, err) { |
| + case stepCancelled: |
| + return abandon(errSwapStopped), true |
| + case stepStale: |
| + return abandon(err), true |
| + } |
| + return SwapResult{}, false |
| +} |
| + |
| // Swap runs one attempt on an awake entry (spec §7.3–§7.5, §7.8). It |
| // blocks until the attempt is done, failed or abandoned. For an overlap |
| // that includes the retiring generation's drain. For a hold it lasts until |
| @@ -3039,17 +3177,12 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| // 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 { |
| + gate := 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 { |
| + return a.memberCurrentLocked(e, m, mctx, at, newest) |
| + } |
| + if err := a.admit(ctx, work, gate); err != nil { |
| a.dropMember(e, m) // restores the raised parent (leaveSet) |
| return abandon(err) |
| } |
| @@ -3069,8 +3202,8 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| a.mu.Unlock() |
| if spawnErr != nil { |
| a.dropMember(e, m) |
| - if mctx.Err() != nil { |
| - return abandon(errSwapStopped) |
| + if res, over := stepResult(mctx, spawnErr); over { |
| + return res |
| } |
| return failSwap(fmt.Errorf("spawn: %w", spawnErr)) |
| } |
| @@ -3078,8 +3211,8 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| // Step 4: the health check, through the generation's own handle. |
| if err := a.genHealthy(mctx, m.gen); err != nil { |
| a.dropMember(e, m) |
| - if mctx.Err() != nil { |
| - return abandon(errSwapStopped) |
| + if res, over := stepResult(mctx, err); over { |
| + return res |
| } |
| return failSwap(fmt.Errorf("health: %w", err)) |
| } |
| @@ -3091,12 +3224,9 @@ func (a *Service) swapOverlap(ctx context.Context, gb SwapBackend, at Attempt, n |
| err = a.admit(ctx, work, func() error { |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| - if err := a.swapChecksLocked(e, at, newest); err != nil { |
| + if err := a.memberCurrentLocked(e, m, mctx, at, newest); err != nil { |
| return err |
| } |
| - if e.ops.members[m.gen.N] != m || mctx.Err() != nil { |
| - return errSwapStopped |
| - } |
| if m.proc == nil || m.proc.Exited() { |
| return errSuccessorExited |
| } |
| @@ -3715,6 +3845,29 @@ func TestSwapHoldFailedWakeKeepsTheLeaseUntilItsChildIsEmpty(t *testing.T) { |
| waitFor(t, "the failed wake's lease given back", func() bool { return f.b.releaseCount() == before+2 }) |
| } |
| +// The same when the failed wake's leader itself never stops: an unproven |
| +// stop (ErrStillRunning) is no termination, so the lease is kept until the |
| +// sweep's retried stop and proof hold. Retaining only when the leader had |
| +// exited released the lease over a live process. |
| +func TestSwapHoldFailedWakeWhoseLeaderNeverStopsKeepsTheLease(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.b.gmu.Lock() |
| + f.b.genStuck = true |
| + f.b.gmu.Unlock() |
| + 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: the failed wake's lease was given back over a live leader", before, got) |
| + } |
| + _, p := f.b.lastGen() |
| + p.exit() |
| + 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. |
| @@ -3950,19 +4103,19 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) |
| // 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. |
| + // attempt replaces it. The same gate as every other exec point. |
| a.mu.Lock() |
| - cerr := a.swapChecksLocked(hw.next, at, newest) |
| + cerr := a.attemptCurrentLocked(hw.next, at, newest, false) |
| a.mu.Unlock() |
| if cerr != nil { |
| return abandon(cerr) |
| } |
| 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 |
| + case classifyStep(nil, werr) == stepStale: |
| + // The pinned wake's own gate 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): |
| + case classifyStep(nil, werr) == stepCancelled: |
| // An intentional stop ended the pinned wake (spec §6.4). |
| return abandon(errSwapStopped) |
| } |
| @@ -3973,31 +4126,17 @@ func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) |
| // has finished and under the same lock. If the check passes, it installs |
| // the pinned waking entry. The caller holds a.mu and starts the wake. |
| func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) { |
| - switch { |
| - case a.shuttingDown: |
| - return nil, errSwapShutdown |
| - case a.fences[e.host] != 0: |
| - return nil, errSwapFenced |
| - case h.newest(e.host) != h.at.SchedRev: |
| - return nil, errSwapSuperseded |
| - } |
| 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 |
| - } |
| + // The same gate here, at the stop's end, and again at the pinned wake's |
| + // exec (spec §6.3, §7.4): the lease and the restore come between. |
| + next.pinCheck = func() error { return a.attemptCurrentLocked(next, h.at, h.newest, true) } |
| a.entries[e.host] = next |
| + if err := next.pinCheck(); err != nil { |
| + delete(a.entries, e.host) |
| + return nil, err |
| + } |
| return next, nil |
| } |
| @@ -4303,6 +4442,8 @@ git commit -m "zdd plan 3 task 6: the hold swap, a pinned wake installed by fini |
| func (b *SplitBackend) StartSidecarGen(ctx context.Context, rt registry.Route, g Gen) (Proc, error) |
| func (a *Service) startSidecar(ctx context.Context, rt registry.Route, g Gen) (Proc, error) |
| func (a *Service) replaceSidecar(e *actEntry) // joins an in-flight supervised start, then replaces |
| + func (a *Service) discardSidecar(sc Proc) // flag off: today's async stop; on: synchronous, before the join closes |
| + func sidecarJoinBound(gens bool) time.Duration // stop()'s wait on an in-flight sidecar start |
| ``` |
| 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. |
| @@ -4552,6 +4693,42 @@ func TestSwapSidecarNeverRunsBesideTheRetiredGenerationsSidecar(t *testing.T) { |
| } |
| } |
| +// A terminal stop that begins while a supervised sidecar start is in flight |
| +// joins that start, and the sidecar it produced, before the lease is |
| +// released (spec §7.4). Discarding it asynchronously let the supervisor |
| +// close its join at once, so the stop released the lease while the new |
| +// sidecar was still running. |
| +func TestSwapStopJoinsTheSidecarAStartLeftBehind(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's start blocks |
| + waitFor(t, "the supervisor's start", func() bool { return len(f.b.sidecarGenPaths()) == 2 }) |
| + rep := f.b.lastRep() |
| + f.a.Bounce(f.host) |
| + time.Sleep(50 * time.Millisecond) |
| + close(startGate) // the start lands on a stopping entry; its sidecar's Stop blocks |
| + time.Sleep(100 * time.Millisecond) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while the sidecar a start left behind still ran") |
| + } |
| + close(stopGate) |
| + waitFor(t, "the lease release", rep.isClosed) |
| + f.b.mu.Lock() |
| + started := f.b.sidecars[len(f.b.sidecars)-1] |
| + f.b.mu.Unlock() |
| + if !started.isStopped() { |
| + t.Fatal("the sidecar the start left behind was never stopped") |
| + } |
| +} |
| + |
| // 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 |
| @@ -4815,16 +4992,20 @@ func (a *Service) replaceSidecar(e *actEntry) { |
| ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| sc, err := a.startSidecar(ctx, rt, g) |
| cancel() |
| + var discard Proc |
| + defer func() { a.discardSidecar(discard) }() // after the unlock, before done closes |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| if a.entries[e.host] != e { |
| - stopDiscardedSidecar(sc) |
| + discard = sc |
| return |
| } |
| e.sidecarStarting = false |
| e.sidecarStartDone = nil |
| if e.state != actAwake { |
| - stopDiscardedSidecar(sc) |
| + // A terminal stop is joined on this replacement (done). It stops |
| + // e.sidecar after the join and proves it: install sc for it. |
| + e.sidecar, e.sidecarGen = sc, g.N |
| return |
| } |
| e.sidecar, e.sidecarStarted, e.sidecarBackoff, e.sidecarGen = sc, a.now(), 0, g.N |
| @@ -4880,6 +5061,78 @@ In the same goroutine, directly after the existing `if e0.state != actAwake { |
| 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`. |
| +**No discarded sidecar outlives the join** (spec §7.4, P3 review partial P2 #3). The supervisor goroutine's two other completion branches (`if a.entries[host] != e0 { stopDiscardedSidecar(sc); return }` and `if e0.state != actAwake { stopDiscardedSidecar(sc); return }`) discard asynchronously and then close `sidecarStartDone`, so a terminal stop that joined on it can release the lease while the discarded sidecar still runs. With generations on they change. Directly after the goroutine's `defer close(done)`, and before its `a.mu.Lock()` / `defer a.mu.Unlock()`, add: |
| + |
| +```go |
| + var discard Proc |
| + defer func() { a.discardSidecar(discard) }() // after the unlock, before done closes |
| +``` |
| + |
| +Replace the first branch's `stopDiscardedSidecar(sc)` with `discard = sc`. Replace the second branch with: |
| + |
| +```go |
| + if e0.state != actAwake { |
| + if a.gens { |
| + // stop() is joined on this goroutine (done). It |
| + // stops e.sidecar after that join, synchronously, |
| + // and records whether it exited: install sc so it |
| + // is that stop's to prove. |
| + e0.sidecar, e0.sidecarGen = sc, gen.N |
| + return |
| + } |
| + discard = sc |
| + return |
| + } |
| +``` |
| + |
| +Add to `activator.go`, beside `stopDiscardedSidecar`: |
| + |
| +```go |
| +// discardSidecar is how a sidecar start that lost its entry is shed. With |
| +// generations off it is today's stopDiscardedSidecar, fire-and-forget. With |
| +// them on it is synchronous and bounded, and the caller runs it before it |
| +// closes the join (sidecarStartDone) a terminal stop waits on, so the stop |
| +// never releases the lease while it runs (spec §7.4). An unproven stop is |
| +// logged; it can only follow a stop that gave up its join. |
| +func (a *Service) discardSidecar(sc Proc) { |
| + if sc == nil { |
| + return |
| + } |
| + if !a.gens { |
| + stopDiscardedSidecar(sc) |
| + return |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + defer cancel() |
| + if err := sc.Stop(ctx); err != nil || !sc.Exited() { |
| + log.Printf("activator: a discarded SIDECAR did not prove its exit: %v", err) |
| + } |
| +} |
| +``` |
| + |
| +`replaceSidecar` (above) already sheds through `discardSidecar` and installs onto a stopping entry. |
| + |
| +In `stop()`, the join on a supervised start is bounded by `sidecarStopTimeout` today. With generations on it must cover the whole start and the synchronous discard, so replace `case <-time.After(sidecarStopTimeout):` with: |
| + |
| +```go |
| + case <-time.After(sidecarJoinBound(a.gens)): |
| +``` |
| + |
| +and add: |
| + |
| +```go |
| +// sidecarJoinBound is how long stop() waits for an in-flight sidecar start: |
| +// today's sidecarStopTimeout, or, with generations on, the start's own |
| +// bound (DeadlineControl) plus a synchronous discard, so the join covers |
| +// every process that start can leave behind. |
| +func sidecarJoinBound(gens bool) time.Duration { |
| + if !gens { |
| + return sidecarStopTimeout |
| + } |
| + return instancectl.DeadlineControl + sidecarStopTimeout |
| +} |
| +``` |
| + |
| 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 |
| @@ -6115,7 +6368,7 @@ In `swapOverlap` (`swap.go`), in the `if spawnErr != nil` block, before `return |
| } |
| ``` |
| -In `swapHold` (`swaphold.go`), add this case to the `switch` that ends the function, after its `ErrFenced` case: |
| +In `swapHold` (`swaphold.go`), add this case to the `switch` that ends the function, after its `case classifyStep(nil, werr) == stepCancelled:` case: |
| ```go |
| case errors.Is(werr, instancectl.ErrStagedExe): |
| @@ -6179,8 +6432,8 @@ Claims-on wakes spawn through the claim runtime's provider (`ClaimStart.Build`), |
| func (a *Service) entryParent(e *actEntry, s ClaimStart) (*cgroup.Prepared, error) |
| func (a *Service) closeEntryParent(e *actEntry) |
| func (a *Service) setParent(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, cur cgroup.Limit, succ *cgroup.Limit) error // captured tiers only |
| - func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error) |
| - func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error) |
| + func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen, gate func() error) (Gen, Proc, error) |
| + func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen, gate func() error) (Gen, Proc, error) // gate after Prepare, and as spawnClaimedGen's before hook |
| // Service gains: claimSpawn func(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error) (test seam) |
| ``` |
| @@ -6352,13 +6605,49 @@ func TestStartAdmittedFallsBackToTheGoodBuildOnce(t *testing.T) { |
| t.Fatalf("ready deadlines %v: B's window ends at HealthBy, the good build's does not", readyBy) |
| } |
| } |
| + |
| +// Spec §6.3, §7.4: a claims-on successor's Read and Prepare come after the |
| +// attempt's last check at Step 3a and may take time. The attempt gate runs |
| +// again after them (and inside the work at the exec, as spawnClaimedGen's |
| +// before hook). A newer attempt that lands during Prepare abandons the |
| +// successor as superseded before anything is prepared. Without that, the |
| +// superseded build went on to the strict parent and the exec. |
| +func TestClaimsOnSuccessorReChecksAfterPrepare(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.cfg.HostWrites = hostwrite.Config{Enabled: true, Selector: &swapSelector{work: &swapWork{}}} |
| + obs := hostclaim.Observed{Host: f.host, ETag: "one", Claim: &hostclaim.Claim{ |
| + Holder: &hostclaim.Holder{Location: "instances/" + f.host + "/inc", Session: 1, Generation: "one"}, |
| + Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}, |
| + }} |
| + work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn") |
| + f.a.cfg.Claims = &ClaimRuntime{Box: "box", Pins: true, Generations: true, |
| + Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil }, |
| + Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) { |
| + f.rev.Add(1) // a newer attempt lands while the runtime prepares |
| + return ClaimStart{Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Version: "v2", Gen: p.Gen, |
| + Restore: func(context.Context) error { return nil }, |
| + Materialize: func(context.Context) error { return nil }, Build: func() *exec.Cmd { return nil }, |
| + Ready: func(context.Context, *exec.Cmd) error { return nil }, Reserve: func(context.Context, int) error { return nil }, |
| + Failed: func(context.Context, int) error { return nil }, Healthy: func(context.Context, int, *exec.Cmd) error { return nil }, |
| + Replicate: func(context.Context, string) (Rep, error) { return nil, nil }, |
| + Publish: func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) { return obs, nil }, |
| + Retain: func(Proc, Rep, error) {}}, nil |
| + }} |
| + 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 after Prepare", res) |
| + } |
| + if got := f.appSocket(t); got != f.incumbent.Socket { |
| + t.Fatalf("AppSocket = %q, want the incumbent", got) |
| + } |
| +} |
| ``` |
| Add `"sync"` and `"github.com/carlosframework/platform/internal/cgroup"` to the file's imports. |
| - [ ] **Step 2: Run the tests to verify they fail** |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestClaimsOnSwapNeedsGenerations|TestStartAdmittedRefusesAGenerationMismatch|TestClaimGenAt|TestStartAdmittedFallsBack' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestClaimsOnSwapNeedsGenerations|TestStartAdmittedRefusesAGenerationMismatch|TestClaimGenAt|TestStartAdmittedFallsBack|TestClaimsOnSuccessorReChecks' -count=1` |
| Expected: build failure: `undefined: errClaimsNoGenerations`, `unknown field Gen in struct literal of type ClaimStart`, `a.claimGenAt undefined`, `unknown field Fallback in struct literal of type StartPolicy`. |
| - [ ] **Step 3: Extend the claim types and `startAdmitted`** |
| @@ -6708,14 +6997,23 @@ func (a *Service) setParent(ctx context.Context, gb SwapBackend, e *actEntry, rt |
| // backend's staged spawn. Claims on, the claim runtime builds the start and |
| // spawnClaimedGen runs it inside the selected work, which validates the |
| // claim and stamp at exec. |
| -func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error) { |
| +// |
| +// gate is the attempt gate (memberCurrentLocked, taking a.mu itself). It |
| +// runs immediately before every exec: claims off, right before |
| +// StartStaged; claims on, after the runtime's Read and Prepare (which may |
| +// take time) and again inside the selected work at the exec, as |
| +// spawnClaimedGen's before hook. |
| +func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen, gate func() error) (Gen, Proc, error) { |
| if hostwrite.Legacy(work) { |
| + if err := gate(); err != nil { |
| + return g, nil, err |
| + } |
| return gb.StartStaged(ctx, rt, g) |
| } |
| if a.claimSpawn != nil { |
| return a.claimSpawn(ctx, e, rt, work, g) |
| } |
| - return a.startClaimedSuccessor(ctx, e, rt, work, g) |
| + return a.startClaimedSuccessor(ctx, e, rt, work, g, gate) |
| } |
| // startClaimedSuccessor builds a claims-on successor: the runtime's Prepare |
| @@ -6724,7 +7022,7 @@ func (a *Service) spawnSuccessor(ctx context.Context, gb SwapBackend, e *actEntr |
| // Restore purges data, and it is never called here). The lease and the |
| // replication stay the agent's (spec §10): a successor is not a new |
| // admission. |
| -func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error) { |
| +func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen, gate func() error) (Gen, Proc, error) { |
| r := a.cfg.Claims |
| if r == nil || !r.Generations || !r.Pins || r.Read == nil || r.Prepare == nil { |
| return g, nil, errClaimsNoGenerations |
| @@ -6741,6 +7039,11 @@ func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt reg |
| if !s.valid() || s.Host != e.host || s.Version != g.Build || s.Gen.N != want.N || s.Gen.Socket != want.Socket { |
| return g, nil, hostwrite.ErrUnavailable |
| } |
| + // Read and Prepare took time: the attempt gate again before anything is |
| + // prepared, and once more inside the work at the exec (before, below). |
| + if err := gate(); err != nil { |
| + return g, nil, err |
| + } |
| parent, err := a.entryParent(e, s) |
| if err != nil { |
| return g, nil, err |
| @@ -6749,7 +7052,7 @@ func (a *Service) startClaimedSuccessor(ctx context.Context, e *actEntry, rt reg |
| if g.Tier != (cgroup.Limit{}) { |
| tier = g.Tier // the successor's own artifact's tier, captured at admission (Task 1) |
| } |
| - cmd, child, err := spawnClaimedGen(ctx, parent, g.N, tier, work, s.Identity, s.Build, nil) |
| + cmd, child, err := spawnClaimedGen(ctx, parent, g.N, tier, work, s.Identity, s.Build, gate) |
| p := newClaimProcess(cmd, child) |
| if err != nil { |
| return g, p, err |
| @@ -6786,9 +7089,11 @@ Then make two replacements. |
| `g, proc, spawnErr := gb.StartStaged(mctx, rt, m.gen)` becomes: |
| ```go |
| - g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen) |
| + g, proc, spawnErr := a.spawnSuccessor(mctx, gb, e, rt, work, m.gen, gate) |
| ``` |
| +(`gate` is Task 4's Step 3a closure over `memberCurrentLocked`.) |
| + |
| In `swapHold`, the claims check becomes: |
| ```go |
| @@ -7845,10 +8150,10 @@ MUTANTS='M1|activator.go|swapsim:M1|s/a\.joinOps\(e\)/true/ |
| M2|swap.go|swapsim:M2|s/a\.addMemberLocked\(e, retiring\)/_ = retiring/ |
| 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// |
| +M5|swapadmit.go|swapsim:M5|s/!waking \&\& e\.state != actAwake/false/ |
| M6|activator.go|swapsim:M6|s/if a\.gens \{/if false {/ |
| -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(/' |
| +M7|opset.go|swapsim:M7|s/return a\.genTerminated\(rt, m\.gen, m\.proc\)/return true || a.genTerminated(rt, m.gen, m.proc)/ |
| +M8|activator.go|swapsim:M8|s/if !a\.genTerminated\(/if false \&\& !a.genTerminated(/' |
| fail=0 |
| while IFS='|' read -r name file marker sub; do |
| src="$SCRATCH/tree/$A/$file" |
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 38ba026a0..4b22c3a10 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 |
| @@ -1588,6 +1588,11 @@ type dispatch struct { |
| Canary bool |
| Class attemptClass |
| Cohort string |
| + // FailedUnder is the PromotedAt of the pointer the attempt's build was |
| + // promoted under, read by the scheduler (Task 8) before the attempt |
| + // runs, so a failure is published with no blocking read in between |
| + // (spec §8.1, §8.4). Zero: the failure's own time stands in. |
| + FailedUnder time.Time |
| } |
| // hostSlot is one instance's desired state. The attempt to run is derived |
| @@ -2785,6 +2790,9 @@ type fakeRoutes struct { |
| // when set and answering an error, fails that Get. |
| getHook func(host string) |
| getErr func(host string) error |
| + // setGoodHook, when set, runs at the start of every SetGoodBuild, with |
| + // no lock held. |
| + setGoodHook func(host string) |
| } |
| func (f *fakeRoutes) Get(host string) (*registry.Route, error) { |
| @@ -2820,6 +2828,12 @@ func (f *fakeRoutes) List() ([]registry.Route, error) { |
| } |
| func (f *fakeRoutes) SetGoodBuild(host string, expect, next registry.GoodBuild) error { |
| + f.mu.Lock() |
| + hook := f.setGoodHook |
| + f.mu.Unlock() |
| + if hook != nil { |
| + hook(host) |
| + } |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| r := f.rows[host] |
| @@ -2895,6 +2909,8 @@ type harness struct { |
| failing atomic.Bool |
| touched atomic.Int32 |
| events *eventLog |
| + // pointerHook, when set, runs at the start of every pointer read. |
| + pointerHook func() |
| } |
| func newHarness(t *testing.T, hosts ...string) *harness { |
| @@ -2928,6 +2944,12 @@ func (h *harness) pointer(channel, ver string, promoted time.Time) { |
| } |
| func (h *harness) pointerFor(_ context.Context, _, _, channel string) (*release.ChannelPointer, error) { |
| + h.mu.Lock() |
| + hook := h.pointerHook |
| + h.mu.Unlock() |
| + if hook != nil { |
| + hook() |
| + } |
| h.mu.Lock() |
| defer h.mu.Unlock() |
| p, ok := h.pointers[channel] |
| @@ -3158,6 +3180,92 @@ func TestAFailureIsRecordedAndHaltedInTheStepThatFreesItsSlot(t *testing.T) { |
| } |
| } |
| +// Spec §8.5: a successful attempt keeps its host's slot until its state |
| +// (serving and good build) is published, so no newer attempt can be |
| +// admitted and complete first and then be overwritten by this one. Freeing |
| +// the slot before the write let that happen. |
| +func TestASuccessKeepsItsSlotUntilItIsPublished(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + var freed atomic.Int32 |
| + h.rows.mu.Lock() |
| + h.rows.setGoodHook = func(host string) { |
| + s := h.m.schedulerOf(host) |
| + s.mu.Lock() |
| + if hs := s.q.hosts[host]; hs != nil && hs.running == nil { |
| + freed.Add(1) |
| + } |
| + s.mu.Unlock() |
| + } |
| + h.rows.mu.Unlock() |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "the good build moved to v2", func() bool { return h.rows.get("a").GoodVersion == "v2" }) |
| + if n := freed.Load(); n != 0 { |
| + t.Fatalf("%d good-build write(s) made after the attempt's slot was freed", n) |
| + } |
| +} |
| + |
| +// A failure is published with no blocking read between the failure and the |
| +// record (spec §8.1): the pointer the build failed under is read before the |
| +// attempt runs. Reading it in done(), before the record, left every sibling's |
| +// wake free to plan the failed build for as long as that read took. |
| +func TestNoPointerReadSitsBetweenAFailureAndItsRecord(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + var failed, late atomic.Bool |
| + h.act.outcome = func(activator.Attempt) activator.SwapOutcome { |
| + failed.Store(true) |
| + return activator.SwapFailed |
| + } |
| + h.mu.Lock() |
| + h.pointerHook = func() { |
| + if failed.Load() { |
| + if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok { |
| + late.Store(true) |
| + } |
| + } |
| + } |
| + h.mu.Unlock() |
| + h.install("v2", "a") |
| + h.pass() |
| + waitFor(t, "a failure record", func() bool { _, ok := h.book.Lookup("bab", "blog", "v2"); return ok }) |
| + if late.Load() { |
| + t.Fatal("a pointer read ran after the failure and before its record was published") |
| + } |
| +} |
| + |
| +// Spec §6.3: a build installed while the entry is unstable (a hold |
| +// draining, a wake running) still supersedes the attempt in flight, so its |
| +// gates refuse the older build. Skipping wantBuild for such a host left the |
| +// revision on the older build, and a hold's pinned wake exec'd it. Nothing |
| +// is admitted onto the host until it is stable again. |
| +func TestAnInstallDuringAnUnstableEntryStillSupersedes(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.pass() |
| + s := h.m.schedulerOf("a") |
| + h.act.mu.Lock() |
| + delete(h.act.gens, "a") |
| + h.act.owned["a"] = true // stopping or waking: a process, no current generation |
| + h.act.mu.Unlock() |
| + s.mu.Lock() |
| + before := s.q.newest("a") |
| + s.mu.Unlock() |
| + h.stage("v2") |
| + h.install("v2", "a") |
| + h.pass() |
| + s.mu.Lock() |
| + after := s.q.newest("a") |
| + s.mu.Unlock() |
| + if after == before { |
| + t.Fatal("an install during an unstable entry left the attempt revision unchanged") |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if calls, _, _ := h.act.snapshot(); len(calls) != 0 { |
| + t.Fatalf("attempts %+v admitted onto an unstable entry", calls) |
| + } |
| +} |
| + |
| // 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. |
| @@ -3518,8 +3626,12 @@ type Deps struct { |
| Events activator.EventSink |
| // Touch asks root for a stage pass (the stage relay, spec §7.7). |
| Touch func() |
| - Now func() time.Time |
| - Logf func(format string, args ...any) |
| + // Materialize installs and stamps a cold managed host's build (the exec |
| + // backend's Materialize hook) so a wake is planned against it (Task 10, |
| + // spec §6.4, §8.4). Nil: a cold managed host's wake is refused. |
| + Materialize func(ctx context.Context, host string) error |
| + Now func() time.Time |
| + Logf func(format string, args ...any) |
| } |
| // pointerReadEvery bounds how often an app with a standing failure record |
| @@ -3544,6 +3656,11 @@ type Manager struct { |
| cfg map[string]*configTrack |
| retried map[string]string // host -> build a retry cleared: its next attempt's cause is retry |
| pointerRead map[appKey]time.Time |
| + |
| + // goodMu serialises every serving- and good-build write with its |
| + // acknowledgement check (served, publishServing, writeGood), so an older |
| + // generation's write can never land after a newer one's (spec §8.5). |
| + goodMu sync.Mutex |
| } |
| // NewManager builds the manager. Its schedulers run until ctx ends. |
| @@ -3667,10 +3784,16 @@ func (m *Manager) observe(routes []registry.Route) []appKey { |
| s.mu.Lock() |
| s.q.concurrency = policy.Concurrency |
| for _, x := range list { |
| - if !x.known { |
| - continue |
| + if x.known { |
| + // Serving state only from a stable entry. The desired build |
| + // and its revision are updated either way (spec §6.3): a |
| + // build root installs while a hold drains or a wake runs |
| + // supersedes the attempt's, so its gates (pinCheck, the |
| + // result check) refuse the older build. Admission still |
| + // waits for a stable entry (admit's view leaves a waking or |
| + // stopping host out). |
| + s.q.observe(x.rt.Host, x.serving, x.rt.GoodVersion, now) |
| } |
| - s.q.observe(x.rt.Host, x.serving, x.rt.GoodVersion, now) |
| cause := activator.CauseRelease |
| if retried[x.rt.Host] == x.rt.Version { |
| cause = activator.CauseRetry |
| @@ -3964,21 +4087,74 @@ func (m *Manager) resolve(ctx context.Context, d dispatch, rt registry.Route) (a |
| } |
| // served records a generation that became current after its checks: the |
| -// serving build, and the good build (spec §8.5). |
| +// serving build, and the good build (spec §8.5). Both only while g is the |
| +// generation the activator serves (acknowledged). |
| func (m *Manager) served(host string, g activator.Gen) { |
| if g.Build == "" { |
| return |
| } |
| + m.goodMu.Lock() |
| + defer m.goodMu.Unlock() |
| + if !m.acknowledged(host, g) { |
| + m.d.Logf("rollout: %s: generation %d (%s) is no longer served; not recording it", host, g.N, g.Build) |
| + return |
| + } |
| + if err := m.d.Routes.SetServingVersion(host, g.Build); err != nil { |
| + m.d.Logf("rollout: %s: record serving build %s: %v", host, g.Build, err) |
| + } |
| + m.writeGoodLocked(host, g) |
| +} |
| + |
| +// publishServing records the serving build alone (a healthy first wake, a |
| +// fallback), on the same terms as served. |
| +func (m *Manager) publishServing(host string, g activator.Gen) { |
| + if g.Build == "" { |
| + return |
| + } |
| + m.goodMu.Lock() |
| + defer m.goodMu.Unlock() |
| + if !m.acknowledged(host, g) { |
| + return |
| + } |
| if err := m.d.Routes.SetServingVersion(host, g.Build); err != nil { |
| m.d.Logf("rollout: %s: record serving build %s: %v", host, g.Build, err) |
| } |
| - m.writeGood(host, g) |
| } |
| -// writeGood replaces host's good build with g's identity, conditionally on |
| -// the row still holding what was just read. One retry covers a concurrent |
| -// writer; a second loss is logged and the next serve tries again. |
| +// acknowledged reports whether g is the generation the activator serves |
| +// for host right now (spec §8.5: the good build is written expecting the |
| +// current one). A late completion or event of an older generation is not, |
| +// so it can never overwrite what a newer one wrote. Generation numbers |
| +// decide when both sides have one; a worker's (no number) is compared by |
| +// its staged artifact. |
| +func (m *Manager) acknowledged(host string, g activator.Gen) bool { |
| + cur, ok := m.d.Act.Current(host) |
| + if !ok { |
| + return false |
| + } |
| + if cur.N != 0 && g.N != 0 { |
| + return cur.N == g.N |
| + } |
| + return cur.StagedPath == g.StagedPath && cur.Digest == g.Digest |
| +} |
| + |
| +// writeGood is writeGoodLocked for a caller outside served: a settled first |
| +// wake, a healthy planned wake. Same terms: g must be acknowledged. |
| func (m *Manager) writeGood(host string, g activator.Gen) { |
| + m.goodMu.Lock() |
| + defer m.goodMu.Unlock() |
| + if !m.acknowledged(host, g) { |
| + return |
| + } |
| + m.writeGoodLocked(host, g) |
| +} |
| + |
| +// writeGoodLocked replaces host's good build with g's identity, |
| +// conditionally on the row still holding what was just read. goodMu |
| +// serialises every writer and each has just proven g current, so a retry |
| +// only covers a write from outside this manager; a second loss is logged |
| +// and the next serve tries again. The caller holds m.goodMu. |
| +func (m *Manager) writeGoodLocked(host string, g activator.Gen) { |
| if g.Build == "" || g.Digest == "" { |
| return |
| } |
| @@ -4187,6 +4363,11 @@ func (s *scheduler) execute(ctx context.Context, d dispatch) { |
| s.done(d, activator.SwapResult{Outcome: activator.SwapAbandoned, Reason: err.Error()}, "") |
| return |
| } |
| + // The failing pointer's PromotedAt, read now, while nothing is at stake: |
| + // done() publishes a failure without a blocking read (spec §8.1). |
| + if d.Class == classAdopt { |
| + d.FailedUnder = m.failedUnder(*rt, d.ID.Build) |
| + } |
| m.emit(*rt, EvSwapping, "info", textSwapping(a.Build, a.Cause)) |
| s.done(d, m.d.Act.Swap(ctx, a, s.newest), a.Mode) |
| } |
| @@ -4201,22 +4382,18 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| // 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. |
| + // A failure is published with no blocking read before it (spec §8.1): |
| + // the failing pointer's PromotedAt was read before the attempt ran |
| + // (d.FailedUnder), and the record is written in the same critical |
| + // section that frees the slot and halts the cohort. No admission, no |
| + // newest re-check and no wake plan (which reads the book) sees the slot |
| + // free before the record. |
| 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) |
| + Cause: string(d.ID.Cause), Mode: string(mode), Reason: res.Reason, PromotedAt: d.FailedUnder, At: m.d.Now()} |
| + if r.PromotedAt.IsZero() { |
| + r.PromotedAt = r.At |
| } |
| if res.Outcome == activator.SwapDown && mode == activator.ModeHold { |
| // A hold build executed against this instance's database: |
| @@ -4225,6 +4402,14 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| } |
| rec = &r |
| } |
| + if res.Outcome == activator.SwapDone { |
| + // Published BEFORE the slot is freed (spec §8.5): while the attempt |
| + // still holds its host's slot no newer attempt can be admitted and |
| + // complete first, so this write can never land on top of a newer |
| + // one. writeGood also requires the generation to be the one the |
| + // activator serves (acknowledged). |
| + m.served(d.ID.Host, res.Gen) |
| + } |
| s.mu.Lock() |
| transition := s.q.finish(d, res.Outcome) |
| var ( |
| @@ -4249,7 +4434,8 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| 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 { |
| + rt, err := m.d.Routes.Get(d.ID.Host) |
| + if err != nil || rt == nil { |
| return |
| } |
| if added { |
| @@ -4257,7 +4443,6 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| } |
| switch res.Outcome { |
| case activator.SwapDone: |
| - m.served(d.ID.Host, res.Gen) |
| m.configActivated(d.ID.Host, d.ID.ConfigRev) |
| m.mu.Lock() |
| if m.retried[d.ID.Host] == d.ID.Build { |
| @@ -4872,6 +5057,33 @@ func TestAPlannedOneShotWorkerThatExitsCleanlyIsReady(t *testing.T) { |
| } |
| } |
| +// Spec §6.4, §7.4: a wake cancelled by its lifecycle (here Shutdown, as |
| +// maintenance or a stop would) is no failure of the build: no failure |
| +// event, no fallback, so no halted cohort and no hold-down. spawnPlanned |
| +// that marked every error with a process as a failure recorded one. |
| +func TestACancelledFirstWakeRecordsNothing(t *testing.T) { |
| + 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}) |
| + entered := make(chan struct{}) |
| + var once sync.Once |
| + a.genHealth = func(ctx context.Context, _ Gen) error { |
| + once.Do(func() { close(entered) }) |
| + <-ctx.Done() |
| + return ctx.Err() |
| + } |
| + go a.beginRequest(context.Background(), testRoute("one.example.com")) //nolint:errcheck |
| + <-entered |
| + a.Shutdown(context.Background()) |
| + time.Sleep(100 * time.Millisecond) |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a cancelled wake was recorded against the build", k) |
| + } |
| + if gens := b.started(); len(gens) != 1 { |
| + t.Fatalf("spawned %+v: a cancelled wake fell back", gens) |
| + } |
| +} |
| + |
| // 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 |
| @@ -5007,7 +5219,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|TestAFirstWakeIsObserved|TestAPlannedWorkerFirstWake|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|TestACancelledFirstWake|TestWithoutAPlanner|TestDriftFeeds|TestBounceFor' -count=1` |
| Expected: FAIL, `undefined: WakePlan`. |
| - [ ] **Step 3: Write the plan types and helpers** |
| @@ -5171,6 +5383,13 @@ func (a *Service) spawnPlanned(ctx context.Context, e *actEntry, rt registry.Rou |
| if err == nil || proc == nil { |
| return proc, ver, g, err |
| } |
| + if classifyStep(ctx, err) != stepFailed { |
| + // The wake's lifecycle ended it (maintenance, Shutdown, a stop) or |
| + // its attempt went stale: no failure of the build (spec §6.4, §7.4). |
| + // Nothing is recorded and nothing falls back; doWake's cleanup |
| + // stops the process and proves it gone. |
| + return proc, ver, g, err |
| + } |
| if plan.FirstWake { |
| w.failed, w.failReason = &plan.Boot, err.Error() |
| } |
| @@ -5204,6 +5423,11 @@ func (a *Service) spawnPlannedWorker(ctx context.Context, e *actEntry, rt regist |
| if err == nil || proc == nil { |
| return proc, id, err |
| } |
| + if classifyStep(ctx, err) != stepFailed { |
| + // A cancelled readiness window is the wake's lifecycle, not the |
| + // build (spec §6.4, §6.5): nothing recorded, nothing falls back. |
| + return proc, id, err |
| + } |
| if plan.FirstWake { |
| w.failed, w.failReason = &plan.Boot, err.Error() |
| } |
| @@ -5658,7 +5882,9 @@ 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 // refuses while the host's attempt is running |
| + func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePlan // refuses while the host's attempt is running; installs a cold managed host first (Deps.Materialize), never an empty plan |
| + func (m *Manager) served(host string, g activator.Gen); func (m *Manager) publishServing(host string, g activator.Gen); func (m *Manager) acknowledged(host string, g activator.Gen) bool |
| + // every serving- and good-build write is acknowledged (g is what the activator serves) and serialised by Manager.goodMu; done() publishes before it frees the slot |
| func (m *Manager) attemptRunning(host string) bool |
| func (m *Manager) OnGen(ev activator.GenEvent) |
| ``` |
| @@ -5697,7 +5923,7 @@ func TestDecideWake(t *testing.T) { |
| down := &FailureBook{} |
| down.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", HeldDown: []string{"a"}, At: t0}) |
| - if p, _ := decideWake(inputs(wakeRoute("", ""), &FailureBook{}, overlap)); p.Boot.StagedPath != "" { |
| + if p, _ := decideWake(inputs(wakeRoute("", ""), &FailureBook{}, overlap)); p.Boot.StagedPath != "" || p.Refuse == "" { |
| t.Fatal("nothing stamped: today's wake, unplanned") |
| } |
| p, _ := decideWake(inputs(wakeRoute("v1", "v1"), &FailureBook{}, overlap)) |
| @@ -5752,8 +5978,10 @@ func TestAFellBackWakeRecordsTheFailureAndKeepsTheGoodBuild(t *testing.T) { |
| h.install("v2", "a") |
| h.pass() |
| rt := h.rows.get("a") |
| + h.act.serve("a", "v1") // the good build the fallback left serving |
| + served, _ := h.act.Current("a") |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenFellBack, Route: rt, FirstWake: true, Mode: activator.ModeOverlap, |
| - Gen: activator.Gen{Build: "v1", Digest: "sha-v1"}, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"}) |
| + Gen: served, Failed: release.StagedIdentity{Version: "v2"}, Reason: "health check failed"}) |
| if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok { |
| t.Fatal("no failure record") |
| } |
| @@ -5771,14 +5999,34 @@ func TestOnlyASettledFirstWakeBecomesTheGoodBuild(t *testing.T) { |
| h.install("v2", "a") |
| h.pass() |
| rt := h.rows.get("a") |
| - g := activator.Gen{Build: "v2", Digest: "sha-v2"} |
| + h.act.serve("a", "v2") // the first wake's generation, which the activator serves |
| + g, _ := h.act.Current("a") |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenHealthy, Route: rt, Gen: g, FirstWake: true}) |
| if h.rows.get("a").GoodVersion != "v1" { |
| t.Fatal("a first wake became the good build before its 10 s (spec §8.5)") |
| } |
| h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: rt, Gen: g, FirstWake: true}) |
| - if h.rows.get("a").GoodVersion != "v2" { |
| - t.Fatal("the settled first wake did not become the good build") |
| + if r := h.rows.get("a"); r.GoodVersion != "v2" || r.ServingVersion != "v2" { |
| + t.Fatalf("row = %+v: the settled first wake must be the good and the serving build", r) |
| + } |
| +} |
| + |
| +// Spec §8.5: a good build is written expecting the current one. A late |
| +// success of an older generation (a completion delayed past a newer swap, a |
| +// stale wake event) is not the generation the activator serves, so it never |
| +// overwrites the newer good or serving build. Writing whatever arrived last |
| +// regressed both to the older build. |
| +func TestAStaleSuccessNeverOverwritesANewerBuild(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.act.serve("a", "v2") |
| + older, _ := h.act.Current("a") |
| + h.act.serve("a", "v3") // a newer generation took over |
| + newer, _ := h.act.Current("a") |
| + h.m.served("a", newer) |
| + h.m.served("a", older) // the older one's completion arrives late |
| + h.m.OnGen(activator.GenEvent{Kind: activator.GenSettled, Route: h.rows.get("a"), Gen: older, FirstWake: true}) |
| + if r := h.rows.get("a"); r.GoodVersion != "v3" || r.ServingVersion != "v3" { |
| + t.Fatalf("row = %+v: an older generation overwrote the newer good and serving build", r) |
| } |
| } |
| @@ -5839,6 +6087,45 @@ func TestASleepersConfigNeedsNoAttempt(t *testing.T) { |
| } |
| } |
| +// Spec §6.4, §8.4: a cold managed host (nothing installed yet) has its build |
| +// installed and stamped before the wake is planned, so the plan boots that |
| +// identity, as a first wake, and respects its failure records. An empty |
| +// plan sent the wake down the live-link path, which materialized the |
| +// channel head and ran it unplanned, a held build included. |
| +func TestAColdManagedHostIsInstalledBeforeItIsPlanned(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + cold := func() { |
| + r := h.rows.get("a") |
| + r.Version = "" |
| + h.rows.put(r) |
| + } |
| + cold() |
| + installs := 0 |
| + h.m.d.Materialize = func(_ context.Context, host string) error { |
| + installs++ |
| + r := h.rows.get(host) |
| + r.Version = "v2" |
| + h.rows.put(r) |
| + return nil |
| + } |
| + if p := h.m.Plan(context.Background(), h.rows.get("a")); installs != 1 || p.Boot.Version != "v2" || !p.FirstWake { |
| + t.Fatalf("installs %d, plan %+v: want v2 installed, then planned as a first wake", installs, p) |
| + } |
| + if _, err := h.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "a", At: t0}); err != nil { |
| + t.Fatal(err) |
| + } |
| + cold() |
| + if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Boot.Version == "v2" { |
| + t.Fatalf("plan %+v: a cold host was booted onto a build its app holds", p) |
| + } |
| + h.m.d.Materialize = nil |
| + cold() |
| + if p := h.m.Plan(context.Background(), h.rows.get("a")); p.Refuse == "" || p.Boot.StagedPath != "" { |
| + t.Fatalf("plan %+v: with nothing installed and no way to install, the wake must be refused", p) |
| + } |
| +} |
| + |
| // 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 |
| @@ -5870,7 +6157,7 @@ Add `"time"` and `"github.com/carlosframework/platform/internal/release"` to tha |
| - [ ] **Step 2: Run them to see them fail** |
| -Run: `go test ./internal/rollout/ -run 'TestDecideWake|TestPlanBoots|TestAFellBack|TestOnlyASettled|TestAFailedHold|TestAWakeActivates|TestASleepersConfig|TestAForeignRowsFailedHold' -count=1` |
| +Run: `go test ./internal/rollout/ -run 'TestDecideWake|TestPlanBoots|TestAFellBack|TestOnlyASettled|TestAStaleSuccess|TestAFailedHold|TestAWakeActivates|TestASleepersConfig|TestAForeignRowsFailedHold|TestAColdManagedHost' -count=1` |
| Expected: FAIL, `undefined: decideWake`. |
| - [ ] **Step 3: Write the planner and the outcome handler** |
| @@ -5918,7 +6205,9 @@ func decideWake(in wakeInputs) (activator.WakePlan, error) { |
| acct, app, host := release.RouteAccount(rt), rt.App, rt.Host |
| installed := rt.Version |
| if installed == "" { |
| - return activator.WakePlan{}, nil |
| + // Plan installs a cold host first; a plan for a host with nothing |
| + // installed refuses rather than let the live link run unplanned. |
| + return activator.WakePlan{Refuse: "no build is installed for this instance yet"}, nil |
| } |
| hasGood := rt.GoodVersion != "" && rt.GoodDigest != "" |
| good := release.StagedIdentity{Version: rt.GoodVersion, StagedPath: release.VersionPath(in.OptDir, acct, rt.GoodVersion, app), |
| @@ -5968,6 +6257,24 @@ func (m *Manager) Plan(ctx context.Context, rt registry.Route) activator.WakePla |
| // can boot the good build ahead of a hold-down record. |
| return activator.WakePlan{Refuse: "an attempt is still running for this instance"} |
| } |
| + if rt.Version == "" { |
| + // A cold managed host (spec §6.4, §8.4): its build is installed and |
| + // stamped first, and the wake is planned against that installed |
| + // identity, with its failure records and its first-wake canary. |
| + // Nothing managed ever runs through an empty plan: the live-link |
| + // wake would materialize the channel head and exec it unplanned. |
| + if m.d.Materialize == nil { |
| + return activator.WakePlan{Refuse: "no build is installed for this instance yet"} |
| + } |
| + if err := m.d.Materialize(ctx, rt.Host); err != nil { |
| + return activator.WakePlan{Refuse: "its build could not be installed: " + err.Error()} |
| + } |
| + fresh, err := m.d.Routes.Get(rt.Host) |
| + if err != nil || fresh == nil || fresh.Version == "" { |
| + return activator.WakePlan{Refuse: "no build is installed for this instance yet"} |
| + } |
| + rt = *fresh |
| + } |
| 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) { |
| @@ -6013,19 +6320,16 @@ func (m *Manager) OnGen(ev activator.GenEvent) { |
| rt := ev.Route |
| switch ev.Kind { |
| case activator.GenHealthy: |
| - if err := m.d.Routes.SetServingVersion(rt.Host, ev.Gen.Build); err != nil { |
| - m.d.Logf("rollout: %s: record serving build: %v", rt.Host, err) |
| - } |
| if !ev.FirstWake { |
| - m.writeGood(rt.Host, ev.Gen) |
| + m.served(rt.Host, ev.Gen) |
| } |
| + // A first wake is published when it settles: until then the |
| + // activator hides it (Current), so it is not yet acknowledged. |
| m.wakeConfigActive(rt.Host) |
| case activator.GenSettled: |
| - m.writeGood(rt.Host, ev.Gen) |
| + m.served(rt.Host, ev.Gen) |
| case activator.GenFellBack: |
| - if err := m.d.Routes.SetServingVersion(rt.Host, ev.Gen.Build); err != nil { |
| - m.d.Logf("rollout: %s: record serving build: %v", rt.Host, err) |
| - } |
| + m.publishServing(rt.Host, ev.Gen) |
| m.wakeConfigActive(rt.Host) |
| m.buildFailed(rt, ev) |
| m.emit(rt, EvFallbackServed, "warn", textFallback(ev.Failed.Version, ev.Gen.Build)) |
| @@ -6900,6 +7204,26 @@ func TestTheGenerationsFileNamesThePlannedBuildBeforeItSpawns(t *testing.T) { |
| t.Fatal("the spawning build and its fallback were not pinned before the spawn") |
| } |
| +// Spec §4.2, §8.5: a failed wake whose generation is not proven terminated |
| +// keeps its lease and its process may still run (plan 3, unprovenWakes), |
| +// though its entry is gone. Its build stays in the generations file until |
| +// the proof holds. Snapshotting only the entries dropped its pin, and root |
| +// could sweep or restage the artifact under the process. |
| +func TestARetainedFailedWakeStaysPinned(t *testing.T) { |
| + a, _, _ := planActivator(t, WakePlan{}) |
| + a.GenerationsFile = filepath.Join(t.TempDir(), "zdd", "generations.json") |
| + a.mu.Lock() |
| + a.unprovenWakes = map[string]*unprovenWake{"one.example.com": {gen: Gen{N: 4, Build: "v2"}}} |
| + a.mu.Unlock() |
| + if err := a.persistGenerations(); err != nil { |
| + t.Fatal(err) |
| + } |
| + g, err := release.ReadGenerations(a.GenerationsFile) |
| + if err != nil || g == nil || !slices.Equal(g.Hosts["one.example.com"], []string{"v2"}) { |
| + t.Fatalf("generations %+v, %v: want the retained failed wake's v2 pinned", g, err) |
| + } |
| +} |
| + |
| // unwritableGenerationsFile is a path whose parent is a regular file, so |
| // every write fails. |
| func unwritableGenerationsFile(t *testing.T) string { |
| @@ -7018,7 +7342,7 @@ func TestAnUnreadableGenerationsFileSkipsTheSweep(t *testing.T) { |
| - [ ] **Step 2: Run them to see them fail** |
| -Run: `TMPDIR=/tmp/k go test ./internal/activator/ ./internal/release/ ./internal/host/ -run 'TestTheGenerationsFile|TestAnUnpublishedBuild|TestBoxPinsNamesTheGoodBuild|TestSweepKeeps|TestAnUnreadableGenerations' -count=1` |
| +Run: `TMPDIR=/tmp/k go test ./internal/activator/ ./internal/release/ ./internal/host/ -run 'TestTheGenerationsFile|TestAnUnpublishedBuild|TestARetainedFailedWake|TestBoxPinsNamesTheGoodBuild|TestSweepKeeps|TestAnUnreadableGenerations' -count=1` |
| Expected: FAIL, `a.GenerationsFile undefined`, `undefined: errGenerationsUnpublished`, `undefined: sweepStagedVersionsWith`, and the `BoxPins` assertion. |
| - [ ] **Step 3: Write the generations writer** |
| @@ -7078,6 +7402,18 @@ func (a *Service) generationsLocked() release.Generations { |
| slices.Sort(vers) |
| g.Hosts[host] = vers |
| } |
| + // A failed wake whose generation is not proven terminated (plan 3 Task |
| + // 2, unprovenWakes) has no entry any more, but its process may still |
| + // run: its build stays pinned until the sweep proves it gone (spec §4.2, |
| + // §8.5). |
| + for host, u := range a.unprovenWakes { |
| + if u.gen.Build == "" || slices.Contains(g.Hosts[host], u.gen.Build) { |
| + continue |
| + } |
| + vers := append(g.Hosts[host], u.gen.Build) |
| + slices.Sort(vers) |
| + g.Hosts[host] = vers |
| + } |
| return g |
| } |
| @@ -7143,7 +7479,7 @@ Before a spawn: |
| } |
| } |
| ``` |
| -- 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): |
| +- 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, gate)` (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) |
| @@ -7643,6 +7979,7 @@ git commit -m "zdd-5 task 13: the rollout report in box status" |
| 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] |
| + var rolloutMaterialize func(ctx context.Context, host string) error // eb.Materialize, handed to Deps.Materialize |
| func rolloutUnavailable() string |
| func refuseManaged(svc *activator.Service, reason string) // flag on, required state unavailable: managed wakes refused, bounce legs stand down |
| ``` |
| @@ -7806,6 +8143,11 @@ import ( |
| var rolloutMgr atomic.Pointer[rollout.Manager] |
| +// rolloutMaterialize is the exec backend's Materialize hook, handed to the |
| +// scheduler so a cold managed host is installed before its wake is planned |
| +// (Task 10). runAgent sets it before armRollout; nil refuses such a wake. |
| +var rolloutMaterialize func(ctx context.Context, host string) error |
| + |
| // armedRollout is the box's rollout manager, nil unless CARLOS_SWAP=overlap |
| // armed one. Every leg that would bounce an exec child asks it first. |
| func armedRollout() *rollout.Manager { return rolloutMgr.Load() } |
| @@ -7889,8 +8231,9 @@ func armRollout(ctx context.Context, getenv func(string) string, svc *activator. |
| Pointer: func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error) { |
| return release.Channel(ctx, store, pub, acct, app, channel) |
| }, |
| - Events: sink, |
| - Touch: func() { touchStageMarker(dd) }, |
| + Events: sink, |
| + Materialize: rolloutMaterialize, // the exec backend's hook, set by runAgent (Step 4) |
| + Touch: func() { touchStageMarker(dd) }, |
| OnChange: func() { |
| select { |
| case statusNudge <- struct{}{}: |
| @@ -7987,6 +8330,7 @@ In `runAgent`, directly before `svc = activator.New(*cfg, &activator.SplitBacken |
| swapOn := rollout.Enabled(os.Getenv) |
| activator.SetGenerations(swapOn) |
| eb.GenContainment = swapOn |
| + rolloutMaterialize = eb.Materialize |
| ``` |
| After `svc.Logs = logSink` (line 393, still before `startActivator(...)`), add: |