| index 3d14ddad8..b5758edec 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 |
| @@ -999,7 +999,7 @@ git commit -m "zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged |
| - `finishPendingStops`, lines 2301–2328; |
| - `Shutdown`, lines 2921–2936; |
| - `doWake`'s four `cleanupFailedWake` calls and its head (the unproven-wake refusal). |
| -- Modify: `internal/activator/maintenance.go` (`cleanupFailedWake`) |
| +- Modify: `internal/activator/maintenance.go` (`cleanupFailedWake`); `internal/activator/claimwake.go` (`claimProcess.provenStopped`) |
| - Test: `internal/activator/opset_test.go` |
| |
| **Interfaces:** |
| @@ -1020,6 +1020,7 @@ git commit -m "zdd plan 3 task 1: swap types, the SwapBackend seam, flat staged |
| 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 (p *claimProcess) provenStopped(ctx context.Context) bool // claims-on: its own Stop completed (KillAndEmpty, exit, settle, close) |
| 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 |
| @@ -1046,6 +1047,7 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "os/exec" |
| "slices" |
| "sync" |
| "testing" |
| @@ -1299,6 +1301,60 @@ func TestMemberStopWithoutContainmentAcceptsTheLeaderExit(t *testing.T) { |
| } |
| } |
| |
| +// failingContainment is a claims-on child whose KillAndEmpty fails until |
| +// heal is called. |
| +type failingContainment struct { |
| + mu sync.Mutex |
| + bad bool |
| +} |
| + |
| +func (c *failingContainment) KillAndEmpty(context.Context) error { |
| + c.mu.Lock() |
| + defer c.mu.Unlock() |
| + if c.bad { |
| + return errors.New("cgroup.kill: the child is not empty") |
| + } |
| + return nil |
| +} |
| +func (c *failingContainment) Close() error { return nil } |
| +func (c *failingContainment) heal() { |
| + c.mu.Lock() |
| + c.bad = false |
| + c.mu.Unlock() |
| +} |
| + |
| +// A claims-on generation is terminated only when its own Stop completed |
| +// (spec §7.4): KillAndEmpty on its child included. Here the leader has |
| +// exited but KillAndEmpty fails, so the member stays in the set, unproven, |
| +// until the child is empty. Treating any claimProcess as proven accepted |
| +// the leader's exit alone. |
| +func TestAClaimsOnMemberNeedsItsChildEmptied(t *testing.T) { |
| + x := awakeForBarrier(t, true) |
| + cmd := exec.Command("true") |
| + if err := cmd.Run(); err != nil { |
| + t.Skipf("no true(1): %v", err) |
| + } |
| + done := make(chan struct{}) |
| + close(done) // the leader has exited |
| + cont := &failingContainment{bad: true} |
| + p := &claimProcess{cmd: cmd, prepared: cont, done: done} |
| + m := &genMember{gen: Gen{N: 11, Build: "v2"}, role: roleRetiring, proc: p, done: make(chan struct{})} |
| + x.a.mu.Lock() |
| + e := x.a.entries[x.host] |
| + x.a.addMemberLocked(e, m) |
| + x.a.mu.Unlock() |
| + if x.a.dropMember(e, m) { |
| + t.Fatal("a claims-on generation whose KillAndEmpty failed was taken as terminated") |
| + } |
| + cont.heal() |
| + x.a.sweepOnce() |
| + select { |
| + case <-m.done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the member never left once its child was empty") |
| + } |
| +} |
| + |
| // The current generation is held to the same proof (spec §7.4): a terminal |
| // stop whose main process has exited keeps the lease while that |
| // generation's child is not proven empty, and finishes once it is. A stop |
| @@ -1404,7 +1460,7 @@ func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) { |
| |
| - [ ] **Step 2: Run the tests to verify they fail** |
| |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestAClaimsOnMember|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1` |
| Expected: build failure: `undefined: genRole`, `undefined: genMember`, `x.a.entries[x.host].ops undefined`. |
| |
| - [ ] **Step 3: Write `opset.go`** |
| @@ -1586,11 +1642,14 @@ func (a *Service) genTerminated(rt registry.Route, g Gen, p Proc) bool { |
| // is the proof (plan 2, "stop evidence on the parent"). Any other error is |
| // an unproven termination. |
| func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool { |
| - if _, ok := m.proc.(*claimProcess); ok { |
| - // A claims-on generation (Task 10): claimProcess.Stop already ran |
| - // KillAndEmpty on its own child and succeeded, or Stop failed and |
| - // stopMember never got here. |
| - return true |
| + if cp, ok := m.proc.(*claimProcess); ok { |
| + // A claims-on generation (Task 10): its own Stop is the proof, and |
| + // only a Stop that completed (KillAndEmpty, the exit, the settle and |
| + // the close) counts. A leader that exited after KillAndEmpty failed |
| + // is not terminated. |
| + ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| + defer cancel() |
| + return cp.provenStopped(ctx) |
| } |
| k, ok := a.backend.(genKiller) |
| if !ok || m.gen.N == 0 { |
| @@ -1861,6 +1920,23 @@ At the end of `stop()`, replace `a.finishStop(e, outcome)` (line 2268) with: |
| a.finishStop(e, outcome) |
| ``` |
| |
| +Add to `claimwake.go`, beside `claimProcess.Stop`: |
| + |
| +```go |
| +// provenStopped is a claims-on generation's termination proof (zero-downtime |
| +// spec §7.4): its Stop has completed, which takes KillAndEmpty on its own |
| +// child, the exit, the settle and the close. Stop is idempotent once |
| +// closed, so this retries a Stop that failed before. |
| +func (p *claimProcess) provenStopped(ctx context.Context) bool { |
| + if err := p.Stop(ctx); err != nil { |
| + return false |
| + } |
| + p.mu.Lock() |
| + defer p.mu.Unlock() |
| + return p.closed |
| +} |
| +``` |
| + |
| Add to `actEntry`, beside `ops`: `genReaping bool // reapCurrentGen is running for this pending stop; written under a.mu`. Add to `Service`: `unprovenWakes map[string]*unprovenWake // failed wakes whose generation is not proven stopped; under a.mu` (made lazily). |
| |
| Replace the head of `finishPendingStops`'s loop (lines 2302–2305): |
| @@ -1992,7 +2068,7 @@ In `Shutdown`, replace lines 2921–2936 (from `var waits []chan struct{}` throu |
| |
| - [ ] **Step 5: Run the tests to verify they pass** |
| |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1 -v 2>&1 | tail -20` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestSwapWitnessBarrier|TestBarrierKeeps|TestMemberStop|TestSwapWitnessMemberStop|TestSwapWitnessStopProves|TestAClaimsOnMember|TestSwapWitnessShutdownJoins|TestShutdownLeaves' -count=1 -v 2>&1 | tail -20` |
| Expected: every subtest `--- PASS`, then `ok`. |
| |
| - [ ] **Step 6: Run the whole package (the flag-off paths are unchanged)** |
| @@ -2003,7 +2079,7 @@ Expected: `ok`. Every existing test passes without edits. |
| - [ ] **Step 7: Commit** |
| |
| ```bash |
| -git add internal/activator/opset.go internal/activator/activator.go internal/activator/maintenance.go internal/activator/opset_test.go |
| +git add internal/activator/opset.go internal/activator/activator.go internal/activator/maintenance.go internal/activator/claimwake.go internal/activator/opset_test.go |
| git commit -m "zdd plan 3 task 2: the entry's operation set, the barrier in stop(), Shutdown joins waking entries" |
| ``` |
| |
| @@ -4375,19 +4451,20 @@ In `doClaimWake`, replace line 108, `policy := startPolicy{MaxStarts: 5, Window: |
| |
| ```go |
| policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin} |
| - if e.pin != nil { |
| - // A claims-on hold's pinned wake: its attempt's checks again before |
| - // anything is admitted or exec'd (spec §6.3, §7.4), as |
| + if e.pin != nil && e.pinCheck != nil { |
| + // A claims-on hold's pinned wake: the attempt gate before anything |
| + // is admitted, and (policy.Gate, Task 10) again inside the work at |
| + // every exec, after the admission and Restore (spec §6.3, §7.4), as |
| // startPinnedChecked does for a claims-off one. |
| - a.mu.Lock() |
| - var perr error |
| - if e.pinCheck != nil { |
| - perr = e.pinCheck() |
| + gate := func() error { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return e.pinCheck() |
| } |
| - a.mu.Unlock() |
| - if perr != nil { |
| - return nil, perr |
| + if err := gate(); err != nil { |
| + return nil, err |
| } |
| + _ = gate // Task 10 hands it to startAdmitted as policy.Gate |
| } |
| ``` |
| |
| @@ -4442,8 +4519,10 @@ 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 |
| + func (a *Service) discardSidecar(host string, sc Proc) // flag off: today's async stop; on: synchronous, before the join closes; unproven: tracked |
| + func (a *Service) trackStraySidecar(host string, sc Proc) |
| + func (a *Service) reapStraySidecarsLocked() |
| + // Service gains straySidecars map[string][]Proc; with generations on, stop() joins an in-flight sidecar start with no timeout |
| ``` |
| 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. |
| @@ -4496,6 +4575,7 @@ import ( |
| "os" |
| "path/filepath" |
| "slices" |
| + "strings" |
| "testing" |
| "time" |
| |
| @@ -4729,6 +4809,28 @@ func TestSwapStopJoinsTheSidecarAStartLeftBehind(t *testing.T) { |
| } |
| } |
| |
| +// Spec §7.4: a discarded sidecar whose stop is unproven is never forgotten. |
| +// It stays tracked, the host takes no new wake beside it, and the sweep |
| +// stops it again until it exits. Logging it and moving on let the next wake |
| +// run a second sidecar beside a live one. |
| +func TestAStraySidecarIsNeverForgotten(t *testing.T) { |
| + f := newSwapFixture(t) |
| + f.a.Bounce(f.host) |
| + waitFor(t, "the instance asleep", func() bool { return f.a.entryCount() == 0 }) |
| + stray := &fakeProc{label: "sidecar", rec: f.b.fakeBackend, stuck: true} |
| + f.a.discardSidecar(f.host, stray) |
| + if _, err := f.a.beginRequest(context.Background(), testRoute(f.host)); err == nil || !strings.Contains(err.Error(), errWakeUnproven.Error()) { |
| + t.Fatalf("a wake beside an unproven sidecar = %v, want refused", err) |
| + } |
| + stray.exit() |
| + f.a.sweepOnce() |
| + release, err := f.a.beginRequest(context.Background(), testRoute(f.host)) |
| + if err != nil { |
| + t.Fatalf("the wake once the stray exited: %v", err) |
| + } |
| + release() |
| +} |
| + |
| // 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 |
| @@ -4816,7 +4918,7 @@ var _ SidecarGenStarter = (*SplitBackend)(nil) |
| |
| - [ ] **Step 2: Run the tests to verify they fail** |
| |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestLegacySidecar|TestSwapSidecar|TestExecBackendStartSidecarGen|TestStartSidecarGenKeeps' -count=1` (`TestSwapSidecar` covers `TestSwapSidecarNeverRunsBesideTheRetiredGenerationsSidecar`) |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestLegacySidecar|TestSwapSidecar|TestSwapStopJoins|TestAStraySidecar|TestExecBackendStartSidecarGen|TestStartSidecarGenKeeps' -count=1` (`TestSwapSidecar` covers `TestSwapSidecarNeverRunsBesideTheRetiredGenerationsSidecar`) |
| Expected: build failure: `undefined: SidecarGenStarter`, `b.StartSidecarGen undefined`, `undefined: errNestedSidecarGen`. |
| |
| - [ ] **Step 3: Write the backend side** |
| @@ -4993,7 +5095,7 @@ func (a *Service) replaceSidecar(e *actEntry) { |
| sc, err := a.startSidecar(ctx, rt, g) |
| cancel() |
| var discard Proc |
| - defer func() { a.discardSidecar(discard) }() // after the unlock, before done closes |
| + defer func() { a.discardSidecar(e.host, discard) }() // after the unlock, before done closes |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| if a.entries[e.host] != e { |
| @@ -5065,7 +5167,7 @@ and in the same goroutine's two success installs (`e0.sidecar = sc` in the error |
| |
| ```go |
| var discard Proc |
| - defer func() { a.discardSidecar(discard) }() // after the unlock, before done closes |
| + defer func() { a.discardSidecar(host, discard) }() // after the unlock, before done closes |
| ``` |
| |
| Replace the first branch's `stopDiscardedSidecar(sc)` with `discard = sc`. Replace the second branch with: |
| @@ -5091,10 +5193,9 @@ Add to `activator.go`, beside `stopDiscardedSidecar`: |
| // 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) { |
| +// closes the join (sidecarStartDone) a terminal stop waits on (spec §7.4). |
| +// A sidecar whose exit it cannot prove stays tracked (trackStraySidecar). |
| +func (a *Service) discardSidecar(host string, sc Proc) { |
| if sc == nil { |
| return |
| } |
| @@ -5105,32 +5206,83 @@ func (a *Service) discardSidecar(sc Proc) { |
| 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) |
| + log.Printf("activator: %s: a discarded SIDECAR did not prove its exit (%v); keeping it tracked", host, err) |
| + a.trackStraySidecar(host, sc) |
| + } |
| +} |
| + |
| +// trackStraySidecar keeps a legacy sidecar whose stop is unproven: it is |
| +// never forgotten (spec §7.4). The host takes no new wake while one is |
| +// tracked (errWakeUnproven), and the sweep (reapStraySidecarsLocked) stops |
| +// it again until it exits. |
| +func (a *Service) trackStraySidecar(host string, sc Proc) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + if a.straySidecars == nil { |
| + a.straySidecars = make(map[string][]Proc) |
| + } |
| + a.straySidecars[host] = append(a.straySidecars[host], sc) |
| +} |
| + |
| +// reapStraySidecarsLocked retries each stray sidecar's stop and forgets it |
| +// once it has exited. The caller holds a.mu. |
| +func (a *Service) reapStraySidecarsLocked() { |
| + for host, procs := range a.straySidecars { |
| + var keep []Proc |
| + for _, sc := range procs { |
| + if sc.Exited() { |
| + continue |
| + } |
| + keep = append(keep, sc) |
| + go func(sc Proc) { |
| + ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + _ = sc.Stop(ctx) |
| + cancel() |
| + }(sc) |
| + } |
| + if len(keep) == 0 { |
| + delete(a.straySidecars, host) |
| + } else { |
| + a.straySidecars[host] = keep |
| + } |
| } |
| } |
| ``` |
| |
| -`replaceSidecar` (above) already sheds through `discardSidecar` and installs onto a stopping entry. |
| +Add to `Service`: `straySidecars map[string][]Proc // legacy sidecars whose stop is unproven, per host; under a.mu`. Call `a.reapStraySidecarsLocked()` beside Task 2's `a.reapUnprovenWakesLocked()` at the top of `finishPendingStops`. In `doWake`, extend Task 2's unproven-wake refusal so a tracked stray also refuses: `_, unproven := a.unprovenWakes[host]; unproven = unproven || len(a.straySidecars[host]) > 0`. |
| |
| -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: |
| +In `stop()`'s co-stop, with generations on, a sidecar whose stop is unproven is tracked as well. After main's `outcome.sidecarStop, outcome.sidecarExited = err == nil, e.sidecar.Exited()`, add: |
| |
| ```go |
| - case <-time.After(sidecarJoinBound(a.gens)): |
| + if a.gens && !outcome.sidecarExited { |
| + a.trackStraySidecar(e.host, e.sidecar) |
| + } |
| ``` |
| |
| -and add: |
| +`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, and a stop whose join times out goes on to release the lease. With generations on the join has no timeout of its own (P4 review, partial P2 #3): the start goroutine is bounded by its own `DeadlineControl` and its synchronous discard by `sidecarStopTimeout`, and whatever it leaves unproven stays tracked, so the stop never completes past a sidecar it did not prove. Replace |
| |
| ```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 |
| + if startDone != nil { |
| + select { |
| + case <-startDone: |
| + case <-time.After(sidecarStopTimeout): |
| + } |
| + } |
| +``` |
| +with |
| +```go |
| + if startDone != nil { |
| + if a.gens { |
| + <-startDone // the start, and every process it left, is accounted for |
| + } else { |
| + select { |
| + case <-startDone: |
| + case <-time.After(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: |
| @@ -6422,6 +6574,8 @@ Claims-on wakes spawn through the claim runtime's provider (`ClaimStart.Build`), |
| ```go |
| // StartPolicy gains Gen Gen; ClaimStart gains Gen Gen; ClaimRuntime gains Generations bool |
| // StartPolicy gains Fallback *Gen; HealthBy time.Time (a claimed first wake's fallback and health cap, spec §8.2; plan 5 sets them) |
| + // StartPolicy gains Gate func() error: the attempt gate, run in startAdmitted's beforeStart at every exec (a claims-on hold's pinCheck) |
| + // startAdmitted records pinFailed and falls back only when classifyStep says stepFailed |
| // actEntry gains pinFailed *Gen; pinFailReason string (the pinned build a claimed start saw fail before it was ready) |
| // Service gains: claimWakeSpawn func(ctx context.Context, s ClaimStart, g Gen, before func() error) (*exec.Cmd, *cgroup.Prepared, error) (test seam) |
| // opSet gains parent *cgroup.Prepared, parentTier cgroup.Limit |
| @@ -6606,6 +6760,97 @@ func TestStartAdmittedFallsBackToTheGoodBuildOnce(t *testing.T) { |
| } |
| } |
| |
| +// claimedStartFor is a ClaimStart for obs with every hook a no-op, version v. |
| +func claimedStartFor(obs hostclaim.Observed, work hostwrite.Work, v string) ClaimStart { |
| + return ClaimStart{Host: obs.Host, Work: work, Identity: hostwrite.Identity{Config: "c", Release: "r"}, Version: v, |
| + 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) {}} |
| +} |
| + |
| +func claimedObs(host string) hostclaim.Observed { |
| + return hostclaim.Observed{Host: host, ETag: "one", Claim: &hostclaim.Claim{ |
| + Holder: &hostclaim.Holder{Location: "instances/" + host + "/inc", Session: 1, Generation: "one"}, |
| + Op: hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}, |
| + }} |
| +} |
| + |
| +// Spec §6.3, §7.4: a claims-on hold's pinned wake runs its attempt gate |
| +// inside the work at the exec, after the admission and Restore. Here the |
| +// attempt goes stale during Restore: nothing execs. Checked only before the |
| +// admission, the superseded hold build ran. |
| +func TestAClaimsOnHoldRunsTheGateAtTheExec(t *testing.T) { |
| + obs := claimedObs("x.example") |
| + work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn") |
| + var stale, spawned bool |
| + a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", |
| + Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil }, |
| + Prepare: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) { |
| + s := claimedStartFor(obs, work, "v2") |
| + s.Restore = func(context.Context) error { stale = true; return nil } // a newer attempt lands |
| + return s, nil |
| + }}}} |
| + a.claimWakeSpawn = func(_ context.Context, _ ClaimStart, _ Gen, before func() error) (*exec.Cmd, *cgroup.Prepared, error) { |
| + if err := before(); err != nil { |
| + return nil, nil, err |
| + } |
| + spawned = true |
| + return nil, nil, errors.New("test: stop after the exec") |
| + } |
| + policy := startPolicy{MaxStarts: 1, Window: time.Second, Pin: &Gen{Build: "v2"}, |
| + Gate: func() error { |
| + if stale { |
| + return errSwapSuperseded |
| + } |
| + return nil |
| + }} |
| + err := a.startAdmitted(context.Background(), &actEntry{host: obs.Host}, obs, policy) |
| + if spawned { |
| + t.Fatal("the hold build was exec'd after its attempt went stale during Restore") |
| + } |
| + if !errors.Is(err, errSwapSuperseded) { |
| + t.Fatalf("startAdmitted = %v, want errSwapSuperseded", err) |
| + } |
| +} |
| + |
| +// Spec §6.4, §7.4: a claimed first wake that its lifecycle cancels (here |
| +// mid health window) is no failure of the build: no pinFailed, no fallback. |
| +// startAdmitted recorded every not-ready pinned start as a failed build. |
| +func TestAClaimedWakeCancelledByItsLifecycleRecordsNoFailure(t *testing.T) { |
| + obs := claimedObs("x.example") |
| + work, _ := (hostwrite.Config{}).Select(context.Background(), obs.Host, "spawn") |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + defer cancel() |
| + prepares := 0 |
| + a := &Service{cfg: Config{HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", |
| + Read: func(context.Context, string) (hostclaim.Observed, error) { return obs, nil }, |
| + Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) { |
| + prepares++ |
| + s := claimedStartFor(obs, work, p.Pin.Build) |
| + s.Ready = func(rctx context.Context, _ *exec.Cmd) error { cancel(); return rctx.Err() } // Shutdown, maintenance |
| + return s, nil |
| + }}}} |
| + a.claimWakeSpawn = func(_ context.Context, _ ClaimStart, _ Gen, before func() error) (*exec.Cmd, *cgroup.Prepared, error) { |
| + if err := before(); err != nil { |
| + return nil, nil, err |
| + } |
| + return exec.Command("true"), nil, nil |
| + } |
| + e := &actEntry{host: obs.Host} |
| + policy := startPolicy{MaxStarts: 2, Window: time.Second, Pin: &Gen{Build: "v2"}, Fallback: &Gen{Build: "v1"}} |
| + _ = a.startAdmitted(ctx, e, obs, policy) |
| + if e.pinFailed != nil { |
| + t.Fatalf("pinFailed = %+v: a cancelled wake was recorded as a failed build", e.pinFailed) |
| + } |
| + if prepares != 1 { |
| + t.Fatalf("Prepare ran %d times: a cancelled wake fell back", prepares) |
| + } |
| +} |
| + |
| // 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 |
| @@ -6647,7 +6892,7 @@ Add `"sync"` and `"github.com/carlosframework/platform/internal/cgroup"` to the |
| |
| - [ ] **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|TestClaimsOnSuccessorReChecks' -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestClaimsOnSwapNeedsGenerations|TestStartAdmittedRefusesAGenerationMismatch|TestClaimGenAt|TestStartAdmittedFallsBack|TestClaimsOnSuccessorReChecks|TestAClaimsOnHoldRunsTheGate|TestAClaimedWakeCancelled' -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`** |
| @@ -6669,6 +6914,29 @@ Expected: build failure: `undefined: errClaimsNoGenerations`, `unknown field Gen |
| // wake's health check ends at the earlier of Window and HealthBy (spec |
| // §8.2: the waiters' deadline minus 25 s). Fallback's start is not capped. |
| HealthBy time.Time |
| + // Gate, when set, is the attempt gate (attemptCurrentLocked, taking a.mu |
| + // itself) for a claims-on hold's pinned wake. startAdmitted runs it |
| + // inside the selected work immediately before every exec, after the |
| + // admission and Restore (spec §6.3, §7.4). |
| + Gate func() error |
| +``` |
| + |
| +In `startAdmitted`'s attempt loop, replace main's |
| + |
| +```go |
| + beforeStart := func() error { return budget.start(ctx, s.OpenProbe) } |
| +``` |
| +with |
| +```go |
| + beforeStart := func() error { |
| + if policy.Gate != nil { |
| + // The attempt gate at the exec (Task 6's pinCheck). |
| + if err := policy.Gate(); err != nil { |
| + return err |
| + } |
| + } |
| + return budget.start(ctx, s.OpenProbe) |
| + } |
| ``` |
| |
| Add to `actEntry` (`activator.go`, after Task 6's `holdNext`): |
| @@ -6709,7 +6977,7 @@ Add to `ClaimRuntime`: |
| Generations bool |
| ``` |
| |
| -In `doClaimWake`, after the `policy :=` line Task 6 wrote, add: |
| +In `doClaimWake`'s pinned block (Task 6), replace `_ = gate // Task 10 hands it to startAdmitted as policy.Gate` with `policy.Gate = gate`. After the block, add: |
| |
| ```go |
| if a.gens && r.Generations { |
| @@ -6790,7 +7058,12 @@ with: |
| return errors.Join(startErr, err) |
| } |
| last = startErr |
| - if policy.Pin != nil && !ready && cmd != nil { |
| + // THE outcome classifier (Task 4): a start the wake's lifecycle ended |
| + // (maintenance, Shutdown, a stop) or the attempt gate refused is no |
| + // failure of the build, records nothing and falls back to nothing |
| + // (spec §6.4, §7.4). |
| + buildFailed := classifyStep(ctx, startErr) == stepFailed |
| + if policy.Pin != nil && !ready && cmd != nil && buildFailed { |
| // The pinned build ran and failed before it was ready. Recorded |
| // for plan 5's events (a failed first wake, spec §8.2); a later |
| // ready start of the same pin clears it below. |
| @@ -6801,7 +7074,7 @@ with: |
| } |
| a.mu.Unlock() |
| } |
| - if policy.Fallback != nil && !ready && cmd != nil && ctx.Err() == nil && attempt+1 < policy.MaxStarts { |
| + if policy.Fallback != nil && !ready && cmd != nil && buildFailed && attempt+1 < policy.MaxStarts { |
| // A first wake onto a build this instance has never served |
| // (zero-downtime spec §8.2). p.Stop above killed the failed |
| // child outright. The good build gets what remains, with no |
| @@ -7171,12 +7444,21 @@ git commit -m "zdd plan 3 task 10: claims-on spawns get generation sockets, hand |
| Plan 1 gives workers (native sidecar members) no `Gen`. With generations on, this task gives every worker child a numbered handle and makes a supervised respawn a member of the operation set. Then the barrier and plan 5's generations file see it like any other generation. Workers still bind nothing (`Socket` stays empty) and keep stop-then-start (§7.9). Recycling onto the captured serving build is plan 4's. |
| |
| **Files:** |
| -- Modify: `internal/activator/activator.go`: `wake()`'s awake transition; `respawnWorkerLocked`, lines 1868–1976 |
| +- Modify: `internal/activator/activator.go`: `wake()`'s awake transition; `respawnWorkerLocked`, lines 1868–1976; `internal/activator/opset.go` (`genChildEmpty`, `hostKiller`); `internal/activator/genbackend.go` (`KillHost`); `internal/activator/generations.go` (`Gen.Worker`) |
| - Test: `internal/activator/worker_gen_test.go` |
| |
| **Interfaces:** |
| -- Consumes: Task 2's set, C1 and C3. |
| -- Produces: no new names. A worker's `e.gen` is `Gen{N, Build}` with generations on, and a respawn in flight is a `roleSpawning` member. |
| +- Consumes: Task 2's set, C1 and C3; `cgroup.Manager.OpenCleanup(host)` and `(*Cleanup).KillAndEmpty` (on `main`). |
| +- Produces: |
| + ```go |
| + // Gen gains Worker bool: a socketless generation, which runs in its host's own cgroup (plan 2 D4), not in g<n> |
| + type hostKiller interface{ KillHost(ctx context.Context, rt registry.Route) error } |
| + func (b *ExecBackend) KillHost(ctx context.Context, rt registry.Route) error |
| + func (b *SplitBackend) KillHost(ctx context.Context, rt registry.Route) error |
| + ``` |
| + A worker's `e.gen` is `Gen{N, Build, Worker: true}` with generations on, and a respawn in flight is a `roleSpawning` member. |
| + |
| + **A worker's termination proof is its host cgroup** (spec §7.4, §7.9; P4 review finding 1). Workers keep one cgroup per host (plan 2 D4), so `KillGen` on a `g<n>` they never ran in proves nothing, and `childProc.Stop` signals only the leader. `genChildEmpty` therefore proves a `Worker` generation by `KillHost`: `KillAndEmpty` on the host cgroup, which on a worker host holds that worker alone. Every worker path goes through `genTerminated`: a terminal stop, a respawn retiring the old child before it spawns the new, a respawn shedding a child it raced into existence, and plan 5's `killFailed` before a fallback. When the proof fails, ownership is kept: the stop stays pending with the lease, a respawn spawns nothing, and a shed child stays in the operation set for the reaper. |
| |
| - [ ] **Step 1: Write the failing test** |
| |
| @@ -7186,8 +7468,13 @@ Create `internal/activator/worker_gen_test.go`: |
| package activator |
| |
| import ( |
| + "context" |
| + "errors" |
| + "sync" |
| "testing" |
| "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| |
| func TestWorkerGenerations(t *testing.T) { |
| @@ -7218,17 +7505,96 @@ func TestWorkerGenerations(t *testing.T) { |
| waitFor(t, "the respawn to finish", func() bool { return b.workerStarts() == 2 }) |
| waitFor(t, "the member to leave", func() bool { _, m := a.genStateForTest(rt.Host); return len(m) == 0 }) |
| if on { |
| - if second, _ := a.genStateForTest(rt.Host); second.N <= first.N { |
| - t.Fatalf("respawned worker generation %d, want above %d", second.N, first.N) |
| + if second, _ := a.genStateForTest(rt.Host); second.N <= first.N || !second.Worker { |
| + t.Fatalf("respawned worker generation %+v, want a Worker above %d", second, first.N) |
| } |
| } |
| } |
| } |
| + |
| +// hostKillBackend is fakeBackend plus KillHost, failing while bad. |
| +type hostKillBackend struct { |
| + *fakeBackend |
| + hmu sync.Mutex |
| + bad bool |
| + kills int |
| +} |
| + |
| +func (h *hostKillBackend) KillHost(context.Context, registry.Route) error { |
| + h.hmu.Lock() |
| + defer h.hmu.Unlock() |
| + h.kills++ |
| + if h.bad { |
| + return errors.New("the host cgroup still holds a process") |
| + } |
| + return nil |
| +} |
| + |
| +func (h *hostKillBackend) heal() { |
| + h.hmu.Lock() |
| + h.bad = false |
| + h.hmu.Unlock() |
| +} |
| + |
| +// Spec §7.4, §7.9: a worker runs in its host cgroup (plan 2 D4), so that is |
| +// what proves it terminated, not a g<n> it never ran in (whose absence |
| +// KillGen counts as empty). While the host cgroup still holds a process |
| +// (a descendant the worker forked), a terminal stop keeps the lease and a |
| +// respawn execs nothing beside it. |
| +func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) { |
| + t.Run("a terminal stop keeps the lease", func(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + a.gens = true |
| + hk := &hostKillBackend{fakeBackend: b, bad: true} |
| + a.backend = hk |
| + rt := testWorkerRoute("bot.example.com") |
| + if err := a.WakeWorker(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + rep := b.lastRep() |
| + a.Bounce(rt.Host) |
| + waitFor(t, "the stop to pend", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[rt.Host] |
| + return e != nil && e.pendingStop |
| + }) |
| + time.Sleep(50 * time.Millisecond) |
| + if rep.isClosed() { |
| + t.Fatal("the lease was released while the worker's host cgroup still held a process") |
| + } |
| + hk.heal() |
| + a.sweepOnce() |
| + waitFor(t, "the lease release", rep.isClosed) |
| + }) |
| + t.Run("a respawn execs nothing beside the old child", func(t *testing.T) { |
| + old := sidecarSettle |
| + sidecarSettle = 20 * time.Millisecond |
| + t.Cleanup(func() { sidecarSettle = old }) |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + a.gens = true |
| + hk := &hostKillBackend{fakeBackend: b} |
| + a.backend = hk |
| + rt := testWorkerRoute("bot2.example.com") |
| + if err := a.WakeWorker(rt); err != nil { |
| + t.Fatal(err) |
| + } |
| + hk.hmu.Lock() |
| + hk.bad = true |
| + hk.hmu.Unlock() |
| + b.lastWorker().crash() |
| + a.sweepOnce() // the respawn retires the crashed child, and its proof fails |
| + waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) }) |
| + if got := b.workerStarts(); got != 1 { |
| + t.Fatalf("worker starts = %d: a respawn exec'd beside an unproven old child", got) |
| + } |
| + }) |
| +} |
| ``` |
| |
| - [ ] **Step 2: Run the test to verify it fails** |
| |
| -Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run TestWorkerGenerations -count=1` |
| +Run: `TMPDIR=/tmp/k nice -n 19 go test ./internal/activator -run 'TestWorkerGenerations|TestWorkerTermination' -count=1` |
| Expected: FAIL: `gens true: worker generation {N:0 ...}`. |
| |
| - [ ] **Step 3: Write the code** |
| @@ -7239,11 +7605,92 @@ In `wake()`, after plan 1's `if e.gen.N == 0 && !e.worker { ... }` block, add: |
| if a.gens && e.worker && e.gen.N == 0 { |
| // A worker binds nothing, but it is still a generation: numbered, |
| // with the build it runs, so the barrier and the generations file |
| - // see it (zero-downtime plan 3, Task 11). |
| - e.gen = Gen{N: nextGen(), Build: w.spawnVersion} |
| + // see it (zero-downtime plan 3, Task 11). Worker: its termination |
| + // is proven on its host cgroup. |
| + e.gen = Gen{N: nextGen(), Build: w.spawnVersion, Worker: true} |
| } |
| ``` |
| |
| +Add the field to plan 1's `Gen`, after Task 1's `Tier`: |
| + |
| +```go |
| + // Worker marks a socketless generation (a worker or native sidecar |
| + // member). It runs in its host's own cgroup, not in a g<n> child (plan 2 |
| + // D4), so its termination is proven there (genChildEmpty, KillHost). |
| + Worker bool |
| +``` |
| + |
| +In `opset.go`'s `genChildEmpty`, insert directly after its `claimProcess` branch: |
| + |
| +```go |
| + if m.gen.Worker { |
| + // A worker runs in its host cgroup (plan 2 D4): that cgroup, not a |
| + // g<n> it never ran in, is what must be empty (spec §7.4, §7.9). |
| + hk, ok := a.backend.(hostKiller) |
| + if !ok { |
| + return true // a backend with no host cgroup: the leader's exit is all there is |
| + } |
| + ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl) |
| + defer cancel() |
| + if err := hk.KillHost(ctx, rt); err != nil { |
| + log.Printf("activator: %s: worker generation %d's host cgroup is not proven empty: %v", rt.Host, m.gen.N, err) |
| + return false |
| + } |
| + return true |
| + } |
| +``` |
| + |
| +and add to `opset.go`: |
| + |
| +```go |
| +// hostKiller is the exec backend's host-cgroup kill: cgroup.kill on host's |
| +// own cgroup, then wait until it is empty. It proves a worker generation |
| +// terminated (plan 2 D4 keeps workers there). |
| +type hostKiller interface { |
| + KillHost(ctx context.Context, rt registry.Route) error |
| +} |
| +``` |
| + |
| +Add to `genbackend.go`: |
| + |
| +```go |
| +// KillHost kills whatever is left in rt's own host cgroup and returns once |
| +// it is empty: a worker's termination proof (zero-downtime spec §7.4, §7.9). |
| +// A nested or unit-backed worker is a systemd unit, whose stop already |
| +// takes its whole cgroup, so there is nothing more to prove here. An absent |
| +// cgroup is empty. No cgroup delegation at all is an unproven kill. |
| +func (b *ExecBackend) KillHost(ctx context.Context, rt registry.Route) error { |
| + if p := b.pathsFor(rt); p.layout == LayoutNested { |
| + return nil |
| + } |
| + if b.Cgroups == nil { |
| + return cgroup.ErrNoDelegation |
| + } |
| + c, err := b.Cgroups.OpenCleanup(rt.Host) |
| + if errors.Is(err, cgroup.ErrNoCgroup) { |
| + return nil |
| + } |
| + if err != nil { |
| + return err |
| + } |
| + if err := c.KillAndEmpty(ctx); err != nil { |
| + _ = c.Close() |
| + return err |
| + } |
| + return c.Close() |
| +} |
| + |
| +func (b *SplitBackend) KillHost(ctx context.Context, rt registry.Route) error { |
| + if rt.Backing == registry.BackingUnit { |
| + return nil // systemd owns the unit tenant's tree |
| + } |
| + if k, ok := b.Exec.(hostKiller); ok { |
| + return k.KillHost(ctx, rt) |
| + } |
| + return nil |
| +} |
| +``` |
| + |
| In `respawnWorkerLocked`, replace its first three lines: |
| |
| ```go |
| @@ -7262,8 +7709,9 @@ with |
| // Task 11). |
| spawnParent, cancelSpawn := context.WithCancel(a.lifeCtx) |
| var m *genMember |
| + oldGen := e.gen // the retiring child's generation, for its proof |
| if a.gens { |
| - m = &genMember{gen: Gen{N: nextGen()}, role: roleSpawning, cancel: cancelSpawn, done: make(chan struct{})} |
| + m = &genMember{gen: Gen{N: nextGen(), Worker: true}, role: roleSpawning, cancel: cancelSpawn, done: make(chan struct{})} |
| a.addMemberLocked(e, m) |
| } |
| ``` |
| @@ -7272,11 +7720,63 @@ Inside the goroutine, after `defer close(done)`, add: |
| |
| ```go |
| defer cancelSpawn() |
| + shed := false // a shed child went to dropMember, which leaves the set itself |
| if m != nil { |
| - defer a.leaveSet(e, m) // runs before close(done): defers are LIFO |
| + defer func() { // runs before close(done): defers are LIFO |
| + if !shed { |
| + a.leaveSet(e, m) |
| + } |
| + }() |
| } |
| ``` |
| |
| +Directly after the retire block (`if stopOld != nil { … }`) and before the "Still wanted?" `a.mu.Lock()`, add: |
| + |
| +```go |
| + // The retired child must be proven gone, its host cgroup included, |
| + // before anything new execs beside its descendants (zero-downtime |
| + // spec §7.4, §7.9). Unproven: nothing spawns; the entry keeps its |
| + // lease, and the sweep's crash path stops it through genTerminated. |
| + retired := true |
| + if a.gens && stopOld != nil { |
| + retired = a.genTerminated(rt, oldGen, stopOld) |
| + } |
| +``` |
| + |
| +and directly after that block's `a.mu.Unlock()`, add `wanted = wanted && retired`. |
| + |
| +In the `case !wanted || !live:` arm, replace main's shed of `proc` |
| + |
| +```go |
| + if proc != nil { |
| + // Spawned into a window that closed under us: shed it rather |
| + // than leave a child nothing owns. |
| + shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + _ = proc.Stop(shedCtx) |
| + shedCancel() |
| + } |
| +``` |
| +with |
| +```go |
| + if proc != nil { |
| + // Spawned into a window that closed under us: shed it rather |
| + // than leave a child nothing owns. With generations on it is |
| + // the member's process: dropMember stops it and proves its |
| + // host cgroup empty, or keeps it in the set for the reaper. |
| + if m != nil { |
| + a.mu.Lock() |
| + m.proc = proc |
| + a.mu.Unlock() |
| + shed = true |
| + a.dropMember(e, m) |
| + } else { |
| + shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + _ = proc.Stop(shedCtx) |
| + shedCancel() |
| + } |
| + } |
| +``` |
| + |
| Then replace the spawn context |
| |
| ```go |
| @@ -7291,7 +7791,7 @@ In the success arm (`default:`), after the line that sets `e.spawnVersion` (`e.s |
| |
| ```go |
| if m != nil { |
| - e.gen = Gen{N: m.gen.N, Build: e.spawnVersion} |
| + e.gen = Gen{N: m.gen.N, Build: e.spawnVersion, Worker: true} |
| } |
| ``` |
| |
| @@ -7303,7 +7803,7 @@ Expected: `ok`. The existing worker tests are the flag-off proof. |
| - [ ] **Step 5: Commit** |
| |
| ```bash |
| -git add internal/activator/activator.go internal/activator/worker_gen_test.go |
| +git add internal/activator/activator.go internal/activator/opset.go internal/activator/genbackend.go internal/activator/generations.go internal/activator/worker_gen_test.go |
| git commit -m "zdd plan 3 task 11: worker children are numbered generations; a respawn joins the operation set" |
| ``` |
| |
amadan