amadan
zdd-readiness: clean one-shot replacements record good, WakeWorker waits a fallback out, halts are announced
A worker replacement whose build exits cleanly as a one-shot now carries SwapResult.CleanExit, and done() records its build good without the acknowledged check (the attempt still holds the host's slot, so no newer generation can have completed), under the attempt's book snapshot. WakeWorker extends its wait once to arrival + firstWakeBudget on a first wake, as request waits do. buildFailed with no scheduler emits the cohort-halted event when Record reports the failure newly Recorded, never for PreRetry.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
10 files changed,
+251
−4
internal/activator/activator.go+16 −0internal/activator/swap.go+8 −0internal/activator/workerswap.go+4 −1internal/activator/workerswap_test.go+80 −1internal/rollout/fake_test.go+19 −0internal/rollout/manager.go+20 −0internal/rollout/scheduler.go+9 −1internal/rollout/wakeplan.go+12 −1internal/rollout/wakeplan_test.go+54 −0internal/rollout/worker_test.go+29 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 587d25e16..99a59f6c6 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -2499,6 +2499,8 @@ func (a *Service) awaitWorkerSettle(ctx context.Context, e *actEntry, proc Proc) |
| func (a *Service) WakeWorker(rt registry.Route) error { |
| deadline := time.NewTimer(wakeWaitBound) |
| defer deadline.Stop() |
| + arrival := time.Now() |
| + extended := false |
| for { |
| a.mu.Lock() |
| // Refused once a shutdown has announced itself. Shutdown sets the flag |
| @@ -2565,6 +2567,20 @@ func (a *Service) WakeWorker(rt registry.Route) error { |
| // rule, for the same reason). |
| return e.err |
| case <-deadline.C: |
| + // A first wake onto a new build gives its waiters arrival + |
| + // firstWakeBudget (spec §8.2), a worker's caller as much as |
| + // a request: B's health window, B's kill and the good |
| + // build's boot. Without it a fallback that succeeds after |
| + // the ordinary bound was reported as a timeout (Astra R3 |
| + // flag triage T10). |
| + a.mu.Lock() |
| + fw := e.firstWake |
| + a.mu.Unlock() |
| + if fw && !extended { |
| + extended = true |
| + deadline.Reset(time.Until(arrival.Add(firstWakeBudget))) |
| + continue |
| + } |
| return errWakeTimeout |
| } |
diff --git a/internal/activator/swap.go b/internal/activator/swap.go| index 5c47e68bc..2fd96e99f 100644 |
| --- a/internal/activator/swap.go |
| +++ b/internal/activator/swap.go |
| @@ -92,6 +92,14 @@ type SwapResult struct { |
| // it; every other result leaves it false, and each of those returns |
| // before anything of the build could start. |
| Executed bool |
| + // CleanExit is set by ReplaceWorker on a SwapDone whose replacement's |
| + // build ran to a clean exit inside the settle window: a one-shot's |
| + // success (spec §6.5). It has no generation the activator will go on |
| + // serving, so nothing acknowledges it afterwards; the scheduler records |
| + // it as the good build from this result (spec §8.5, Astra R3 flag |
| + // triage T9). A replacement still running at the end of the window is |
| + // SwapDone with CleanExit false. |
| + CleanExit bool |
| } |
| var ( |
diff --git a/internal/activator/workerswap.go b/internal/activator/workerswap.go| index f48a59180..c67fb0242 100644 |
| --- a/internal/activator/workerswap.go |
| +++ b/internal/activator/workerswap.go |
| @@ -189,7 +189,10 @@ func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *relea |
| } |
| g := a.workerResultGen(e, onto) |
| a.mu.Unlock() |
| - return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain} |
| + // A clean exit inside the window is a one-shot's success (spec |
| + // §6.5): nothing keeps running to be acknowledged later, so the |
| + // result says so and the scheduler records the build good from it. |
| + return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain, CleanExit: r == settleExited} |
| } |
| if r == settleCancelled && !unexpected { |
| // An intentional stop cancels the observation and records nothing |
diff --git a/internal/activator/workerswap_test.go b/internal/activator/workerswap_test.go| index ae76a37b2..a2cbad45d 100644 |
| --- a/internal/activator/workerswap_test.go |
| +++ b/internal/activator/workerswap_test.go |
| @@ -197,9 +197,88 @@ func TestReplaceWorkerOneShotCleanExitIsReady(t *testing.T) { |
| waitForWorkerStarts(t, b, 2) |
| b.lastWorker().exit() |
| }() |
| - if res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one); res.Outcome != SwapDone { |
| + res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| + if res.Outcome != SwapDone { |
| t.Fatalf("ReplaceWorker = %+v, want done", res) |
| } |
| + if !res.CleanExit { |
| + t.Fatalf("ReplaceWorker = %+v: a one-shot that exited cleanly must say so, so the scheduler records its build good (spec 6.5, 8.5)", res) |
| + } |
| +} |
| + |
| +// Spec §6.5, §8.5: only a build that exited cleanly is a one-shot's success. |
| +// One still running after the settle window is up and serving (done, but |
| +// the scheduler records it good when the activator acknowledges it, not |
| +// here), and a crash is a failure. |
| +func TestReplaceWorkerRunningChildIsNotACleanExit(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 100*time.Millisecond) |
| + a, _, _, _, _ := stagedActivator(t) |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatal(err) |
| + } |
| + res := a.ReplaceWorker(context.Background(), workerAttempt(CauseRelease, ModeOverlap, fakeIdentity("v2")), nil, one) |
| + if res.Outcome != SwapDone || res.CleanExit { |
| + t.Fatalf("ReplaceWorker = %+v, want done and not a clean exit", res) |
| + } |
| +} |
| + |
| +// Astra R3 flag triage T10: a first-wake worker whose build fails and whose |
| +// fallback boots only after the ordinary wake bound is a success for |
| +// WakeWorker, as it is for a request: the wait extends once to arrival + |
| +// firstWakeBudget (spec 8.2). Without the extension the caller was told |
| +// "timed out" for a wake that went on to succeed. A wake that is not a |
| +// first wake still gives up at the ordinary bound. |
| +func TestWakeWorkerWaitsOutAFirstWakeFallback(t *testing.T) { |
| + old := wakeWaitBound |
| + wakeWaitBound = 100 * time.Millisecond |
| + t.Cleanup(func() { wakeWaitBound = old }) |
| + for _, tc := range []struct { |
| + name string |
| + firstWake bool |
| + wantErr bool |
| + }{ |
| + {"first wake", true, false}, |
| + {"not a first wake", false, true}, |
| + } { |
| + t.Run(tc.name, func(t *testing.T) { |
| + shrinkFirstWake(t, 900*time.Millisecond, 200*time.Millisecond, 50*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + a.backend = &plannedHostKill{stagedFake: b} // the failed child's host cgroup proves empty |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + var starts atomic.Int32 |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + if !tc.firstWake { |
| + // An ordinary wake of the good build; its start is slow. |
| + return WakePlan{Boot: good, Mode: ModeOverlap} |
| + } |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + b.mu.Lock() |
| + b.workerStartHook = func(registry.Route) { |
| + n := starts.Add(1) |
| + switch { |
| + case tc.firstWake && n == 1: |
| + // The failing build dies inside its settle window. |
| + time.AfterFunc(20*time.Millisecond, func() { b.lastWorker().crash() }) |
| + case tc.firstWake && n == 2, !tc.firstWake && n == 1: |
| + time.Sleep(250 * time.Millisecond) // the good build boots slowly |
| + } |
| + } |
| + b.mu.Unlock() |
| + err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) |
| + if tc.wantErr { |
| + if !errors.Is(err, errWakeTimeout) { |
| + t.Fatalf("WakeWorker = %v, want errWakeTimeout at the ordinary bound", err) |
| + } |
| + return |
| + } |
| + if err != nil { |
| + t.Fatalf("WakeWorker = %v: a first wake's successful fallback was reported as a failure", err) |
| + } |
| + }) |
| + } |
| } |
| // Spec §6.3, §7.4, §8.1: the attempt's checks run again at the actual |
diff --git a/internal/rollout/fake_test.go b/internal/rollout/fake_test.go| index 62b93f534..e41b5c6f4 100644 |
| --- a/internal/rollout/fake_test.go |
| +++ b/internal/rollout/fake_test.go |
| @@ -166,6 +166,12 @@ func (f *fakeAct) ReplaceWorker(_ context.Context, a activator.Attempt, fallback |
| f.mu.Lock() |
| defer f.mu.Unlock() |
| switch { |
| + case res.Outcome == activator.SwapDone && res.CleanExit: |
| + // A one-shot's child exited: the result names its generation, but |
| + // the activator serves nothing for the host afterwards. |
| + f.n++ |
| + res.Gen = activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand} |
| + delete(f.gens, a.Host) |
| case res.Outcome == activator.SwapDone: |
| f.n++ |
| res.Gen = activator.Gen{N: f.n, Build: a.Build, StagedPath: a.StagedPath, Digest: a.Digest, Demand: a.Demand} |
| @@ -307,6 +313,19 @@ func (e *eventLog) has(substr string) bool { |
| return false |
| } |
| +// count is how many logged lines contain substr. |
| +func (e *eventLog) count(substr string) int { |
| + e.mu.Lock() |
| + defer e.mu.Unlock() |
| + n := 0 |
| + for _, l := range e.lines { |
| + if strings.Contains(l, substr) { |
| + n++ |
| + } |
| + } |
| + return n |
| +} |
| + |
| // harness is a manager over the fakes, for app "blog" of account "bab", |
| // every host awake on v1, its good build. |
| type harness struct { |
diff --git a/internal/rollout/manager.go b/internal/rollout/manager.go| index 44c9fa18f..9cc32eed3 100644 |
| --- a/internal/rollout/manager.go |
| +++ b/internal/rollout/manager.go |
| @@ -1318,6 +1318,26 @@ func (m *Manager) served(host string, g activator.Gen, snap uint64) { |
| m.writeGoodLocked(host, g, snap) |
| } |
| +// servedOneShot records the good build of a replacement whose build exited |
| +// cleanly as a one-shot (spec §6.5, §8.5; Astra R3 flag triage T9). Nothing |
| +// of it keeps running, so the activator has no current generation for the |
| +// host to acknowledge, and served would drop the write. It skips only the |
| +// acknowledged check: done() calls it before the attempt's slot is freed, |
| +// so no newer attempt for the host can have been admitted, let alone |
| +// completed, and the result is the host's newest outcome. Everything else |
| +// is served's: goodMu, the generation's identity, and the book snapshot of |
| +// the execution's start (snap), so only a hold-ran marker older than the |
| +// attempt is cleared. A one-shot has no serving build to publish, and a |
| +// worker's is the activator's alone (setServingLocked). |
| +func (m *Manager) servedOneShot(host string, g activator.Gen, snap uint64) { |
| + if g.Build == "" { |
| + return |
| + } |
| + m.goodMu.Lock() |
| + defer m.goodMu.Unlock() |
| + m.writeGoodLocked(host, g, snap) |
| +} |
| + |
| // setServingLocked writes an instance's serving build. A worker's is the |
| // activator's alone (Task 11, publishWorkerServing, ordered by incarnation), |
| // so two writers never race on one row. The caller holds m.goodMu. |
diff --git a/internal/rollout/scheduler.go b/internal/rollout/scheduler.go| index f4c8d7904..746ccbfa3 100644 |
| --- a/internal/rollout/scheduler.go |
| +++ b/internal/rollout/scheduler.go |
| @@ -289,7 +289,15 @@ func (s *scheduler) done(d dispatch, res activator.SwapResult, mode activator.Sw |
| // one. writeGood also requires the generation to be the one the |
| // activator serves (acknowledged), and clears only a hold-ran marker |
| // set before this attempt started (d.BookSeq, Astra R3 N6). |
| - m.served(d.ID.Host, res.Gen, d.BookSeq) |
| + if res.CleanExit { |
| + // A one-shot replacement that exited cleanly (spec §6.5): no |
| + // generation is left for the activator to acknowledge, so the |
| + // acknowledged check is skipped, and only that check (see |
| + // servedOneShot for why that is safe under the held slot). |
| + m.servedOneShot(d.ID.Host, res.Gen, d.BookSeq) |
| + } else { |
| + m.served(d.ID.Host, res.Gen, d.BookSeq) |
| + } |
| } |
| now := m.d.Now() |
| s.mu.Lock() |
diff --git a/internal/rollout/wakeplan.go b/internal/rollout/wakeplan.go| index 422256910..a4eb34fc7 100644 |
| --- a/internal/rollout/wakeplan.go |
| +++ b/internal/rollout/wakeplan.go |
| @@ -604,9 +604,20 @@ func (m *Manager) buildFailed(rt registry.Route, ev activator.GenEvent, started |
| } |
| s := m.schedulerOf(rt.Host) |
| if s == nil { |
| - if _, err := m.d.Book.Record(rec); err != nil { |
| + // No scheduler for the host's app: the record still halts the |
| + // cohort, and the halt is still announced, on the same condition |
| + // cohortFailed announces it (a newly recorded failure, never a |
| + // pre-retry one or one merged into a record that stands). Without |
| + // it a member whose first wake failed here saw the fallback event |
| + // and never the halt (Astra R3 flag triage T11). |
| + res, err := m.d.Book.Record(rec) |
| + if err != nil { |
| m.d.Logf("rollout: write failure record %s: %v (held in memory)", rec.Build, err) |
| } |
| + if res == Recorded { |
| + m.d.Logf("rollout: %s/%s build %s failed on %s: %s", rec.Account, rec.App, rec.Build, rec.Host, rec.Reason) |
| + m.emit(rt, EvCohortHalted, "warn", textCohortHalted(rec.Build, m.d.Box, reasonWord(rec.Reason))) |
| + } |
| return |
| } |
| m.cohortFailed(s, rt, rec) |
diff --git a/internal/rollout/wakeplan_test.go b/internal/rollout/wakeplan_test.go| index 26bec7f50..dd44c694b 100644 |
| --- a/internal/rollout/wakeplan_test.go |
| +++ b/internal/rollout/wakeplan_test.go |
| @@ -117,6 +117,60 @@ func TestAFellBackWakeRecordsTheFailureAndKeepsTheGoodBuild(t *testing.T) { |
| } |
| } |
| +// Spec §8.1, §9.4 (Astra R3 flag triage T11): a first wake that fails on a |
| +// host whose app has no scheduler yet still halts the cohort and says so, |
| +// once, when the record is new. The same failure delivered again (the record |
| +// stands) is not announced twice, and a failure of an execution that began |
| +// before a retry of the build (a pre-retry record) is never announced as a |
| +// halt. |
| +func TestAFirstWakeFailureWithNoSchedulerAnnouncesTheHalt(t *testing.T) { |
| + h := newHarness(t, "a") |
| + h.stage("v2") |
| + h.install("v2", "a") // no pass: the manager has no scheduler for the host |
| + if h.m.schedulerOf("a") != nil { |
| + t.Fatal("the harness already has a scheduler for the host") |
| + } |
| + halted := textCohortHalted("v2", h.m.d.Box, status.ReasonHealth) |
| + ev := activator.GenEvent{Kind: activator.GenFailed, Route: h.rows.get("a"), FirstWake: true, Mode: activator.ModeOverlap, |
| + Failed: release.StagedIdentity{Version: "v2"}, Reason: "GET / answered 500"} |
| + h.m.OnGen(ev) |
| + if _, ok := h.book.Lookup("bab", "blog", "v2"); !ok { |
| + t.Fatal("no failure record") |
| + } |
| + if n := h.events.count(halted); n != 1 { |
| + t.Fatalf("%d cohort-halted event(s) after a new record, want 1", n) |
| + } |
| + h.m.OnGen(ev) |
| + if n := h.events.count(halted); n != 1 { |
| + t.Fatalf("%d cohort-halted event(s) after the same failure again, want still 1", n) |
| + } |
| + |
| + // A wake planned before a retry of its build fails stale: PreRetry. |
| + p := newHarness(t, "a") |
| + p.stage("v2") |
| + p.install("v2", "a") |
| + p.wakeHost("a") |
| + plan := p.m.Plan(context.Background(), p.rows.get("a")) |
| + if !plan.FirstWake || plan.Boot.Version != "v2" { |
| + t.Fatalf("plan %+v: want a first wake of v2", plan) |
| + } |
| + if p.m.schedulerOf("a") != nil { |
| + t.Fatal("the harness has a scheduler for the host") |
| + } |
| + p.book.Record(FailureRecord{Account: "bab", App: "blog", Build: "v2", Host: "s", PromotedAt: t0, At: t0}) |
| + if ok, err := p.book.RetryHost("bab", "blog", "v2", "a"); !ok || err != nil { |
| + t.Fatalf("RetryHost = %v, %v: cleared nothing", ok, err) |
| + } |
| + p.m.OnGen(activator.GenEvent{Kind: activator.GenFailed, Route: p.rows.get("a"), Failed: plan.Boot, Mode: plan.Mode, |
| + FirstWake: true, Reason: "health check failed", Token: plan.Token, ConfigRev: plan.ConfigRev}) |
| + if n := p.events.count(halted); n != 0 { |
| + t.Fatalf("%d cohort-halted event(s) for a failure that began before a retry", n) |
| + } |
| + if p.book.Blocks("bab", "blog", "v2", "a", false) { |
| + t.Fatal("the stale failure halted the retried host") |
| + } |
| +} |
| + |
| // Spec §8.1: a fallback publishes the failure hold before the serving build |
| // and the active config. Writing the serving build first showed the good |
| // build serving with no hold on the failed one, a window in which a wake or |
diff --git a/internal/rollout/worker_test.go b/internal/rollout/worker_test.go| index e0b74372e..6bcd09d2d 100644 |
| --- a/internal/rollout/worker_test.go |
| +++ b/internal/rollout/worker_test.go |
| @@ -41,6 +41,35 @@ func TestAWorkerReleaseReplacesOnTheSameEntry(t *testing.T) { |
| } |
| } |
| +// Spec §6.5, §8.5 (Astra R3 flag triage T9): a replacement whose build exits |
| +// cleanly as a one-shot leaves nothing for the activator to serve, so the |
| +// acknowledged check would drop its good write. done() records the build |
| +// good from the result, and clears a hold-ran marker older than the attempt. |
| +// A replacement that is not a clean exit still needs the activator to serve |
| +// it. |
| +func TestACleanOneShotReplacementRecordsItsBuildGood(t *testing.T) { |
| + h := workerHarness(t, "bot") |
| + h.stage("v2") |
| + h.book.MarkHoldRan("bot", HoldRan{Account: "bab", App: "blog", Build: "v9"}) // set before the attempt starts |
| + h.act.mu.Lock() |
| + h.act.worker = func(activator.Attempt, *release.StagedIdentity) activator.SwapResult { |
| + return activator.SwapResult{Outcome: activator.SwapDone, CleanExit: true} |
| + } |
| + h.act.mu.Unlock() |
| + h.install("v2", "bot") |
| + h.pass() |
| + waitFor(t, "the good build to move to v2", func() bool { return h.rows.get("bot").GoodVersion == "v2" }) |
| + if r := h.rows.get("bot"); r.GoodDigest != "sha-v2" { |
| + t.Fatalf("row = %+v, want v2's digest as the good build's", r) |
| + } |
| + if _, ok := h.act.Current("bot"); ok { |
| + t.Fatal("the fake still serves the one-shot: the test does not model a parked worker") |
| + } |
| + if _, held := h.book.HoldRanOn("bot"); held { |
| + t.Fatal("the clean one-shot passed, but the older hold-ran marker stands") |
| + } |
| +} |
| + |
| // Astra R3 N5: a hold build that ran on a worker without passing (a |
| // superseded hold, its marker set by done()) may have migrated its |
| // database, so the next scheduled replacement carries no fallback onto the |