amadan
zdd-4 final review: a recycle refused for a changed copy drops the captured build (F1, R12)
When the backend refuses a recycle onto the captured build with release.ErrStagedCopyChanged, the failed-spawn arm reports it as before and, in the same critical section, clears the entry's captured build. The crash restart's next attempt then takes recycleWorkerLocked's existing no-captured-build path and execs what live/<host> names, instead of retrying bytes that can never verify. Any other spawn error keeps the captured build.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
3 files changed,
+101
−1
internal/activator/activator.go+15 −1internal/activator/worker_identity_test.go+75 −0internal/activator/worker_staged_test.go+11 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index dd00aac30..c3b54bde7 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -1852,7 +1852,8 @@ const workerSpawnTimeout = 2 * time.Minute |
| // (recycleWorkerLocked), or nil for a converge, which execs what live/<host> |
| // names now. A recycle onto a staged copy whose bytes have changed is refused |
| // by the backend (release.ErrStagedCopyChanged) and lands in the failed-spawn |
| -// arm below, which leaves the crash restart to retry with its backoff. |
| +// arm below, which drops the captured build so the crash restart's next |
| +// attempt, with its backoff, execs what live/<host> names. |
| // |
| // Everything expensive happens OFF a.mu, on its own goroutine. That is not a |
| // stylistic preference: ExecBackend.StartWorker can materialize a build from |
| @@ -1962,9 +1963,22 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont |
| a.mu.Unlock() |
| log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused) |
| case err != nil: |
| + // A recycle refused because the captured copy changed or vanished |
| + // (ruling R12) drops the captured build: retrying those bytes can |
| + // never succeed, so the crash restart's next attempt takes |
| + // recycleWorkerLocked's no-captured-build path and execs what |
| + // live/<host> names, as a failed wake's next wake does. Any other |
| + // failure may be transient and keeps the captured build. |
| + dropped := onto != nil && errors.Is(err, release.ErrStagedCopyChanged) |
| + if dropped { |
| + e.workerServing = release.StagedIdentity{} |
| + } |
| a.mu.Unlock() |
| log.Printf("activator: WORKER %s failed to respawn (%s): %v — the sweep's crash restart takes it from here", host, why, err) |
| emitEvent(a.Logs, rt, "error", "sidecar could not be restarted — retrying") |
| + if dropped { |
| + log.Printf("activator: worker %s dropped its captured build %s because the staged copy no longer matches the digest it was exec'd with; its next restart execs what live/%s names", host, onto.Version, host) |
| + } |
| default: |
| e.proc = proc |
| e.workerToken = tok |
diff --git a/internal/activator/worker_identity_test.go b/internal/activator/worker_identity_test.go| index 0771af5db..da987c191 100644 |
| --- a/internal/activator/worker_identity_test.go |
| +++ b/internal/activator/worker_identity_test.go |
| @@ -1,6 +1,8 @@ |
| package activator |
| import ( |
| + "errors" |
| + "fmt" |
| "path/filepath" |
| "sync/atomic" |
| "testing" |
| @@ -208,3 +210,76 @@ func TestWorkerReportingAnAppChosenVersionStillRecyclesItsCapturedBuild(t *testi |
| t.Fatalf("WorkerReportedVersion = %q after a recycle; the new child has not reported", got) |
| } |
| } |
| + |
| +func (b *stagedFake) failNextStagedSpawn(err error) { |
| + b.smu.Lock() |
| + defer b.smu.Unlock() |
| + b.failNextStaged = err |
| +} |
| + |
| +// waitForSpawnCalls waits until the backend has been asked for n spawns and |
| +// the respawn that asked has finished, failed or not. |
| +func waitForSpawnCalls(t *testing.T, a *Service, b *stagedFake, host string, n int) { |
| + t.Helper() |
| + waitFor(t, fmt.Sprintf("spawn call #%d of %s", n, host), func() bool { |
| + return len(b.calls()) == n && !a.entryRespawning(host) |
| + }) |
| +} |
| + |
| +// Ruling R12: a recycle the backend refuses because the captured copy |
| +// changed drops the captured build, so the crash restart's next attempt |
| +// execs what live/<host> names, as a failed wake's next wake would. |
| +func TestWorkerRecycleRefusedForAChangedCopyDropsTheCapturedBuild(t *testing.T) { |
| + a, b, clk, _, _ := stagedActivator(t) |
| + host := "bot.x.internal" |
| + if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| + t.Fatalf("WakeWorker: %v", err) |
| + } |
| + b.failNextStagedSpawn(fmt.Errorf("worker %s: %w", host, release.ErrStagedCopyChanged)) |
| + b.lastWorker().kill() |
| + a.sweepOnce() |
| + waitForSpawnCalls(t, a, b, host, 2) |
| + if calls := b.calls(); calls[1] == nil || *calls[1] != fakeIdentity("v1") { |
| + t.Fatalf("crash recycle asked for %v; want the captured v1 build", calls) |
| + } |
| + if got := a.entryServing(host); got != (release.StagedIdentity{}) { |
| + t.Fatalf("serving after a refused recycle = %+v; want the captured build dropped", got) |
| + } |
| + clk.advance(sidecarRestartMax + time.Second) |
| + a.sweepOnce() |
| + waitForSpawnCalls(t, a, b, host, 3) |
| + if calls := b.calls(); calls[2] != nil { |
| + t.Fatalf("restart after the refusal asked for %v; want the live link (nil)", calls[2]) |
| + } |
| + if b.workerStarts() != 2 || b.lastWorker().Exited() { |
| + t.Fatalf("workerStarts = %d; want the worker running again from the live link", b.workerStarts()) |
| + } |
| + if got := a.entryServing(host); got != fakeIdentity("v1") { |
| + t.Fatalf("serving after the respawn = %+v, want what live/<host> names", got) |
| + } |
| +} |
| + |
| +// Ruling R12: any other spawn failure keeps the captured build. |
| +func TestWorkerRecycleFailingForAnotherReasonKeepsTheCapturedBuild(t *testing.T) { |
| + a, b, clk, _, _ := stagedActivator(t) |
| + host := "bot.x.internal" |
| + if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil { |
| + t.Fatalf("WakeWorker: %v", err) |
| + } |
| + b.failNextStagedSpawn(errors.New("fork/exec: resource temporarily unavailable")) |
| + b.lastWorker().kill() |
| + a.sweepOnce() |
| + waitForSpawnCalls(t, a, b, host, 2) |
| + if got := a.entryServing(host); got != fakeIdentity("v1") { |
| + t.Fatalf("serving after a failed recycle = %+v; want the captured v1 kept", got) |
| + } |
| + clk.advance(sidecarRestartMax + time.Second) |
| + a.sweepOnce() |
| + waitForSpawnCalls(t, a, b, host, 3) |
| + if calls := b.calls(); calls[2] == nil || *calls[2] != fakeIdentity("v1") { |
| + t.Fatalf("restart after the failure asked for %v; want the captured v1 again", calls[2]) |
| + } |
| + if b.workerStarts() != 2 { |
| + t.Fatalf("workerStarts = %d; want the worker running again", b.workerStarts()) |
| + } |
| +} |
diff --git a/internal/activator/worker_staged_test.go b/internal/activator/worker_staged_test.go| index 3c30a5e3f..f6e857434 100644 |
| --- a/internal/activator/worker_staged_test.go |
| +++ b/internal/activator/worker_staged_test.go |
| @@ -22,6 +22,10 @@ type stagedFake struct { |
| *fakeBackend |
| smu sync.Mutex |
| onto []*release.StagedIdentity |
| + // failNextStaged, when set, is what the next spawn onto a named build |
| + // (onto != nil) returns instead of a child, once: the backend refusing a |
| + // recycle. |
| + failNextStaged error |
| } |
| func fakeIdentity(ver string) release.StagedIdentity { |
| @@ -36,7 +40,14 @@ func (b *stagedFake) StartWorkerStaged(ctx context.Context, rt registry.Route, e |
| } else { |
| b.onto = append(b.onto, nil) |
| } |
| + fail := b.failNextStaged |
| + if onto != nil { |
| + b.failNextStaged = nil |
| + } |
| b.smu.Unlock() |
| + if onto != nil && fail != nil { |
| + return nil, release.StagedIdentity{}, fail |
| + } |
| p, ver, err := b.fakeBackend.StartWorker(ctx, rt, env) |
| if err != nil { |
| return p, release.StagedIdentity{}, err |