amadan
zdd-4 second fix wave: a converge by recycle spends the sweep's converge budget and defers to the scheduler (N2, R17)
The sweep's per-sweep converge budget moves from a sweepOnce local (cycled) to Service.convergeBudget, reset to maxConvergeCyclesPerSweep at the top of each sweep and read and decremented under a.mu. The drift check spends it as before, and recycleWorkerLocked's R13 branch now spends it too: when the budget is spent, the recycle uses the captured build, the drift keeps its age, and the converge happens on a later sweep. When CoalesceRecycle is set (plan 5's scheduler is present), a recycle never converges: a false answer recycles the captured build and keeps the drift age, since the scheduler's admission owns build changes (spec §7.9). Tests: two workers with overdue drift crash together; one sweep converges at most maxConvergeCyclesPerSweep of them, the other recycles v1 with its drift age kept and converges on the next sweep. With CoalesceRecycle returning false, recycles past ConvergeGrace keep the captured v1 and the drift age.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+124
−8
internal/activator/activator.go+12 −4internal/activator/worker_identity_test.go+93 −0internal/activator/workerbuild.go+19 −4
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 2b425dd2e..360075633 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -658,6 +658,12 @@ type Service struct { |
| // new may be spawned against a DB whose lease is about to be released. |
| shuttingDown bool |
| + // convergeBudget is how many build-convergence cycles this sweep may |
| + // still start: reset to maxConvergeCyclesPerSweep at the top of every |
| + // sweepOnce, and spent by the sweep's drift check and by a recycle that |
| + // converges (recycleWorkerLocked), so the two share one pace. Under a.mu. |
| + convergeBudget int |
| + |
| fences map[string]uint64 |
| tokenPreparations map[string]chan struct{} |
| stopHistory map[string]*stopObservation |
| @@ -2481,7 +2487,9 @@ func (a *Service) sweepOnce() { |
| // a separate call at all: a pendingStop entry is in actStopping, not |
| // actAwake, so the idle pass would never reach it. |
| a.finishPendingStops() |
| - cycled := 0 // build-convergence cycles started this tick (rate-limited) |
| + // Build-convergence cycles this tick may start (rate-limited), shared |
| + // with the recycles below that converge instead (recycleWorkerLocked). |
| + a.convergeBudget = maxConvergeCyclesPerSweep |
| for _, e := range a.entries { |
| if a.fences[e.host] != 0 { |
| continue |
| @@ -2627,7 +2635,7 @@ func (a *Service) sweepOnce() { |
| cur, readErr := a.liveVer(e.host) |
| cycle, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, ConvergeGrace) |
| e.staleSince = next |
| - if cycle && cycled < maxConvergeCyclesPerSweep { |
| + if cycle && a.convergeBudget > 0 { |
| // An always-on worker converges as one MOVE, not as a cycle: the |
| // cycle's second half is "something wakes it back up", and a |
| // worker with no request path and no schedule has nothing that |
| @@ -2639,12 +2647,12 @@ func (a *Service) sweepOnce() { |
| if e.worker && !e.route.Hibernate { |
| log.Printf("activator: worker %s serving %s but live symlink is now %s — respawning onto the new build", e.host, e.spawnVersion, cur) |
| a.convergeWorkerLocked(e) |
| - cycled++ |
| + a.convergeBudget-- |
| continue |
| } |
| log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur) |
| a.beginStopLocked(e) |
| - cycled++ |
| + a.convergeBudget-- |
| continue |
| } |
| // Sidecar supervision — Restart=always, sweep-driven, no supervisor |
diff --git a/internal/activator/worker_identity_test.go b/internal/activator/worker_identity_test.go| index d1a7f6296..bb48fc049 100644 |
| --- a/internal/activator/worker_identity_test.go |
| +++ b/internal/activator/worker_identity_test.go |
| @@ -338,6 +338,99 @@ func TestWorkerMaxRunRecyclesConvergeOnceTheGraceHasPassed(t *testing.T) { |
| } |
| } |
| +func (a *Service) entryStaleSince(host string) time.Time { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + if e := a.entries[host]; e != nil { |
| + return e.staleSince |
| + } |
| + return time.Time{} |
| +} |
| + |
| +// Ruling R17 (N2): a converge by recycle draws on the sweep's converge |
| +// budget. Two workers whose drift is overdue crash together; one sweep |
| +// converges at most maxConvergeCyclesPerSweep of them, the other recycles its |
| +// captured build and keeps its drift age, and converges on a later sweep. |
| +func TestWorkerRecycleConvergesShareTheSweepsConvergeBudget(t *testing.T) { |
| + a, b, clk, _, ver := stagedActivator(t) |
| + hosts := []string{"bot1.x.internal", "bot2.x.internal"} |
| + for _, h := range hosts { |
| + if err := a.WakeWorker(alwaysOnWorkerRoute(h)); err != nil { |
| + t.Fatalf("WakeWorker %s: %v", h, err) |
| + } |
| + } |
| + ver.Store("v2") |
| + a.sweepOnce() // the drift is first seen here, by the sweep's own check |
| + seen := a.entryStaleSince(hosts[0]) |
| + if seen.IsZero() || !a.entryStaleSince(hosts[1]).Equal(seen) { |
| + t.Fatalf("drift clocks = %v, %v; want both started by the sweep", seen, a.entryStaleSince(hosts[1])) |
| + } |
| + clk.advance(ConvergeGrace + time.Minute) |
| + b.mu.Lock() |
| + workers := append([]*fakeProc(nil), b.workers...) |
| + b.mu.Unlock() |
| + for _, w := range workers { |
| + w.kill() |
| + } |
| + a.sweepOnce() |
| + waitFor(t, "both crash restarts", func() bool { |
| + return b.workerStarts() == 4 && !a.entryRespawning(hosts[0]) && !a.entryRespawning(hosts[1]) |
| + }) |
| + var converged, recycled []string |
| + for _, h := range hosts { |
| + switch a.entrySpawnVersion(h) { |
| + case "v2": |
| + converged = append(converged, h) |
| + case "v1": |
| + recycled = append(recycled, h) |
| + } |
| + } |
| + if len(converged) != maxConvergeCyclesPerSweep || len(recycled) != len(hosts)-maxConvergeCyclesPerSweep { |
| + t.Fatalf("in one sweep converged %v and recycled %v; want at most %d converge", converged, recycled, maxConvergeCyclesPerSweep) |
| + } |
| + if got := a.entryStaleSince(recycled[0]); !got.Equal(seen) { |
| + t.Fatalf("drift clock of the worker that recycled = %v; want its age kept (%v)", got, seen) |
| + } |
| + a.sweepOnce() |
| + waitFor(t, "the later converge", func() bool { |
| + return a.entrySpawnVersion(recycled[0]) == "v2" && !a.entryRespawning(recycled[0]) |
| + }) |
| +} |
| + |
| +// Ruling R17 (N2): with plan 5's scheduler present (CoalesceRecycle set), a |
| +// recycle never converges, even past ConvergeGrace: plan 5's admission owns |
| +// build changes. A false answer recycles the captured build, and the drift |
| +// keeps its age. |
| +func TestWorkerRecycleNeverConvergesWhenTheSchedulerIsPresent(t *testing.T) { |
| + a, b, clk, _, ver := stagedActivator(t) |
| + host := "bot.x.internal" |
| + a.CoalesceRecycle = func(string) bool { return false } |
| + if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| + t.Fatalf("WakeWorker: %v", err) |
| + } |
| + ver.Store("v2") |
| + b.lastWorker().kill() |
| + a.sweepOnce() // the crash restart sees the drift first |
| + waitForRestart(t, a, b.fakeBackend, host, 2) |
| + seen := a.entryStaleSince(host) |
| + if seen.IsZero() { |
| + t.Fatal("the crash restart did not start the drift clock") |
| + } |
| + b.lastWorker().kill() |
| + clk.advance(ConvergeGrace + time.Minute) |
| + a.sweepOnce() |
| + waitForRestart(t, a, b.fakeBackend, host, 3) |
| + calls := b.calls() |
| + for i := 1; i < len(calls); i++ { |
| + if calls[i] == nil || *calls[i] != fakeIdentity("v1") { |
| + t.Fatalf("recycle %d asked for %v; with the scheduler present a recycle keeps the captured v1", i, calls[i]) |
| + } |
| + } |
| + if got := a.entryStaleSince(host); !got.Equal(seen) { |
| + t.Fatalf("drift clock = %v after the recycle; want its age kept (%v)", got, seen) |
| + } |
| +} |
| + |
| // Ruling R13: an exited worker skips the sweep's drift check, so the crash |
| // restart measures the drift itself, and a crash loop converges once the |
| // grace has passed. |
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go| index cd3ae341b..8487df4c1 100644 |
| --- a/internal/activator/workerbuild.go |
| +++ b/internal/activator/workerbuild.go |
| @@ -51,9 +51,17 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st |
| // before the check fired. So the recycle reads live/<host> as the sweep does |
| // (ConvergeDecision over a.liveVer and spawnVersion), starting the drift |
| // clock if this is the first sight of it, and once the drift has lasted |
| -// ConvergeGrace it converges instead (onto nil). That replaces one spawn |
| -// with one spawn, so it adds no load, and it does not draw on the sweep's |
| -// per-sweep converge budget, which is local to sweepOnce. |
| +// ConvergeGrace it converges instead (onto nil). |
| +// |
| +// That converge is paced like the sweep's own (ruling R17): it spends the |
| +// sweep's converge budget (a.convergeBudget, reset each sweep), so a box |
| +// whose workers' recycles fall due together still moves at most |
| +// maxConvergeCyclesPerSweep of them per sweep. And it happens only while no |
| +// scheduler is present: when CoalesceRecycle is set (plan 5), that |
| +// scheduler's admission owns every build change, so a recycle it did not |
| +// coalesce never converges. Either way, a converge that is not admitted |
| +// recycles the captured build, and the drift keeps its age for a later |
| +// sweep. |
| func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) { |
| if hook := a.CoalesceRecycle; hook != nil && hook(e.host) { |
| log.Printf("activator: worker %s %s; a queued release attempt makes the spawn instead of a recycle", e.host, why) |
| @@ -64,7 +72,14 @@ func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) { |
| cur, readErr := a.liveVer(e.host) |
| converge, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, a.now(), ConvergeGrace) |
| e.staleSince = next |
| - if converge { |
| + switch { |
| + case !converge: |
| + case a.CoalesceRecycle != nil: |
| + log.Printf("activator: worker %s %s; live/%s names %s, but the release scheduler owns build changes, so it recycles its captured build %s", e.host, why, e.host, cur, e.spawnVersion) |
| + case a.convergeBudget <= 0: |
| + log.Printf("activator: worker %s %s; live/%s names %s, but this sweep's converge budget is spent, so it recycles its captured build %s and converges on a later sweep", e.host, why, e.host, cur, e.spawnVersion) |
| + default: |
| + a.convergeBudget-- |
| log.Printf("activator: worker %s %s; it serves %s but live/%s has named %s for at least %s, so it converges onto the new build instead of recycling", e.host, why, e.spawnVersion, e.host, cur, ConvergeGrace) |
| a.respawnWorkerLocked(e, stopOld, "converging onto a new build", nil) |
| return |