amadan
activator: Shutdown aborts a waking entry and gives its lease back with swaps off (#292)
`systemctl restart carlos-edge` stops every carlos-instance@ unit before the agent gets its SIGTERM (BindsTo= and After=carlos-edge.service). On annie-welding at 13:18Z the old agent saw bunkhouse's child gone, recycled the entry (lease released, "hibernated"), and a request woke it again at 13:18:33, taking a fresh lease. That wake's start is root's systemctl start, whose job systemd queued behind the edge restart, so it was still in flight when SIGTERM arrived. With swaps off, Shutdown neither cancelled nor waited for a waking entry, and wake() only honoured shuttingDown with swaps on, so the agent exited with the wake holding its lease. The queued job ran after the new edge started and failed 209/STDOUT; the new agent refused bunkhouse for the lease's whole 15 minutes. Shutdown now cancels and joins waking entries whatever CARLOS_SWAP says, as it already did with swaps on: a cancelled wake gives its lease back under releaseLease's own budget, and one that finishes is stopped by wake(). Once Shutdown has begun, begin refuses (it no longer re-wakes a host for a request queued on a stop), BeginSidecarRequest and the sweep stop recycling exited children as unexpected, and wake() cancels any new wake before its lease. Lease TTLs are unchanged; swaps-on paths are unchanged. TestShutdownLeavesAWakingEntryWithSwapOff pinned the old rule; it becomes TestShutdownJoinsAWakingEntryWithSwapOff and joins M6's witnesses (the marker moved to the join, and the mutant is still CAUGHT).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
5 files changed,
+279
−30
internal/activator/activator.go+51 −14internal/activator/activator_test.go+17 −3internal/activator/opset_test.go+27 −12internal/activator/shutdown_lease_test.go+183 −0internal/activator/testdata/swapsim/mutate.sh+1 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index a3d9866a0..932aa804f 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -1010,6 +1010,10 @@ const configRefusedRetryAfter = 30 |
| var errWakeTimeout = errors.New("timed out waiting for instance wake") |
| var errStartupStopped = errors.New("activator stopped during startup") |
| +// errShuttingDown refuses a request once Shutdown has begun (#292): nothing |
| +// wakes, and nothing is recycled, after the agent has started going away. |
| +var errShuttingDown = errors.New("activator: the agent is shutting down") |
| + |
| // Service is the activator (Eleven's activatorSvc). It structurally satisfies |
| // the edge package's Activator interface — the edge never imports this |
| // package: the dependency points activator→edge only (backend_exec.go takes |
| @@ -1060,6 +1064,9 @@ type Service struct { |
| // shuttingDown is set by Shutdown before it stops anything, and read by |
| // the supervised-respawn goroutines: the agent is going away, so nothing |
| // new may be spawned against a DB whose lease is about to be released. |
| + // begin, WakeWorker, wake and the sweep read it too, with swaps on or |
| + // off (#292): once it is set nothing wakes, and an exited child is |
| + // Shutdown's to stop, never recycled as an unexpected exit. |
| shuttingDown bool |
| // convergeBudget is how many build-convergence cycles this sweep may |
| @@ -1343,6 +1350,17 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel |
| a.mu.Unlock() |
| return nil, ErrFenced |
| } |
| + // Once Shutdown has begun, a request wakes nothing and recycles |
| + // nothing (#292), WakeWorker's rule. Checked on every pass of the |
| + // loop, not only at entry: a request queued on a stop or a wake |
| + // before the shutdown began would otherwise come round and wake the |
| + // host afresh, taking a lease Shutdown never sees. An exited child |
| + // found here is left for Shutdown's own stop, not read as the |
| + // build's failure. |
| + if a.shuttingDown { |
| + a.mu.Unlock() |
| + return nil, errShuttingDown |
| + } |
| e := a.entries[rt.Host] |
| if e == nil { |
| // Asleep → kick off ONE wake; this request and any latecomers all |
| @@ -1486,7 +1504,9 @@ func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) er |
| return ErrFenced |
| } |
| e := a.entries[rt.Host] |
| - if e == nil { |
| + if e == nil || a.shuttingDown { |
| + // Shutting down: the stop is Shutdown's, and an exited child is |
| + // not recycled here (#292). |
| a.mu.Unlock() |
| return errSidecarAsleep |
| } |
| @@ -1603,7 +1623,10 @@ func (a *Service) wake(e *actEntry) { |
| // Real time, not a.now: it anchors a first wake's real health deadline |
| // (spawnPlanned, doClaimWake). |
| e.wakeStart = time.Now() |
| - if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| + // A wake that starts after Shutdown began is cancelled before its |
| + // lease, whatever CARLOS_SWAP says (#292): with swaps off it used to run |
| + // on, and take a lease the exiting agent never gave back. |
| + if a.fences[e.host] != 0 || a.shuttingDown { |
| cancel() |
| } |
| a.mu.Unlock() |
| @@ -1755,7 +1778,7 @@ func (a *Service) wake(e *actEntry) { |
| // already reachable. The fence and bounce checks below come after |
| // it, so a stop they begin cancels it. |
| a.armFirstWatchLocked(e, w) |
| - if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| + if a.fences[e.host] != 0 || a.shuttingDown { |
| a.beginStopLocked(e) |
| } else if e.bounceAfterWake { |
| // A config delivery or `carlos restart` asked for while this wake |
| @@ -4262,6 +4285,14 @@ func (a *Service) sweepOnce() { |
| earliest = nil |
| } |
| a.mu.Lock() |
| + if a.shuttingDown { |
| + // Shutdown has begun and is about to stop this sweeper: from here it |
| + // owns every entry (#292). A child found exited now is its stop's, |
| + // not an unexpected exit to recycle, and nothing is due-woken or |
| + // started. Shutdown runs finishPendingStops itself. |
| + a.mu.Unlock() |
| + return |
| + } |
| // Finish stops that were left pending because proc.Stop returned |
| // ErrStillRunning (design §6.3, codex R2-5): termination wasn't proven |
| // at the time, so stop() kept the lease and returned without forgetting |
| @@ -4889,18 +4920,24 @@ func (a *Service) Shutdown(ctx context.Context) { |
| case actStopping: |
| waits = append(waits, e.stopped) |
| case actWaking: |
| - // With swaps on, Shutdown covers waking entries too |
| - // (zero-downtime spec §7.4). The wake is cancelled, and once it |
| - // resolves, wake() itself stops the entry through the barrier. |
| - // With swaps off, today's rule holds: a mid-flight wake resolves |
| - // on its own, its instance is orphaned only if it wins the race |
| - // with process exit, and the lease TTL still bounds the damage. |
| - if a.gens { // swapsim:M6 |
| - if e.wakeCancel != nil { |
| - e.wakeCancel() |
| - } |
| - waking = append(waking, e) |
| + // Shutdown covers waking entries too (zero-downtime spec §7.4), |
| + // whatever CARLOS_SWAP says. The wake is cancelled, and once it |
| + // resolves, wake() itself stops the entry (through the barrier |
| + // with swaps on), or the failed wake has given its lease back |
| + // under a budget of its own (releaseLease). |
| + // |
| + // Swaps off used to leave a waking entry to resolve on its own, |
| + // with the lease's TTL to bound the damage. That is #292: an |
| + // edge restart stops every carlos-instance@ unit before the |
| + // agent's SIGTERM, the agent recycles one and a request wakes it |
| + // again, and that wake's start is a systemctl start job queued |
| + // behind the edge restart. The agent exited with the wake still |
| + // holding its lease, and the next agent refused the host for the |
| + // lease's whole 15 minutes. |
| + if e.wakeCancel != nil { |
| + e.wakeCancel() |
| } |
| + waking = append(waking, e) // swapsim:M6 |
| } |
| } |
| a.mu.Unlock() |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index 154e307e6..93998a83d 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -306,8 +306,14 @@ type fakeBackend struct { |
| seedRoutes []registry.Route |
| startErr error |
| startGate chan struct{} // when non-nil, Start() blocks until closed |
| - reps []*fakeRep |
| - procs []*fakeProc |
| + // startQueued, when non-nil, blocks Start() until it is closed or the |
| + // wake's context ends, and a start whose context ended returns no |
| + // process. It models a nested start whose systemctl start job systemd |
| + // has queued behind an edge restart: the instance-control client gives |
| + // up when its context does (#292). |
| + startQueued chan struct{} |
| + reps []*fakeRep |
| + procs []*fakeProc |
| // syncIntervals records, per wake, the replication-class tick doWake |
| // threads into Replicate() — the ladder's seam assertion. |
| syncIntervals []time.Duration |
| @@ -470,11 +476,12 @@ func (b *fakeBackend) Seed(_ context.Context, rt registry.Route) error { |
| return b.seedErr |
| } |
| -func (b *fakeBackend) Start(_ context.Context, rt registry.Route) (Proc, string, error) { |
| +func (b *fakeBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) { |
| b.mu.Lock() |
| b.starts++ |
| b.order = append(b.order, "start:instance") |
| gate := b.startGate |
| + queued := b.startQueued |
| fn := b.execVerFn |
| startErr := b.startErr |
| proc := &fakeProc{label: "instance", rec: b, stuck: b.procStuck} |
| @@ -487,6 +494,13 @@ func (b *fakeBackend) Start(_ context.Context, rt registry.Route) (Proc, string, |
| if gate != nil { |
| <-gate |
| } |
| + if queued != nil { |
| + select { |
| + case <-queued: |
| + case <-ctx.Done(): |
| + return nil, "", ctx.Err() |
| + } |
| + } |
| ver := "" |
| if fn != nil { |
| ver = fn() |
diff --git a/internal/activator/opset_test.go b/internal/activator/opset_test.go| index 59b8b5af9..e19d1f95e 100644 |
| --- a/internal/activator/opset_test.go |
| +++ b/internal/activator/opset_test.go |
| @@ -594,8 +594,12 @@ func TestAFencedWakeReleasesUnderALiveContext(t *testing.T) { |
| } |
| } |
| -// Flag off: Shutdown keeps today's rule and does not wait for a waking entry. |
| -func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) { |
| +// Flag off: Shutdown joins a waking entry exactly as it does with swaps on |
| +// (#292). It used to leave one to resolve on its own, and the agent then |
| +// exited with the wake still holding its lease. Here the start ignores its |
| +// cancelled context, so Shutdown must wait it out, stop what it woke and |
| +// give the lease back under a live context. |
| +func TestShutdownJoinsAWakingEntryWithSwapOff(t *testing.T) { |
| a, b, _, _ := testActivator(t, time.Hour) |
| gate := make(chan struct{}) |
| b.startGate = gate |
| @@ -610,18 +614,29 @@ func TestShutdownLeavesAWakingEntryWithSwapOff(t *testing.T) { |
| go func() { a.Shutdown(context.Background()); close(done) }() |
| select { |
| case <-done: |
| - case <-time.After(time.Second): |
| - t.Fatal("Shutdown waited for a waking entry with swaps off") |
| + t.Fatal("Shutdown returned while a wake was still spawning, with swaps off") |
| + case <-time.After(100 * time.Millisecond): |
| } |
| close(gate) |
| - waitFor(t, "the wake to finish", func() bool { |
| - a.mu.Lock() |
| - defer a.mu.Unlock() |
| - e := a.entries[host] |
| - return e != nil && e.state == actAwake |
| - }) |
| - a.Bounce(host) |
| - waitFor(t, "cleanup", func() bool { return a.entryCount() == 0 }) |
| + select { |
| + case <-done: |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("Shutdown never returned") |
| + } |
| + if !b.lastProc().isStopped() { |
| + t.Fatal("the woken child outlived Shutdown") |
| + } |
| + if got := b.releaseCount(); got != 1 { |
| + t.Fatalf("lease releases = %d, want 1", got) |
| + } |
| + for _, err := range b.releaseCtxErrors() { |
| + if err != nil { |
| + t.Fatalf("the lease was released under a dead context (%v)", err) |
| + } |
| + } |
| + if n := a.entryCount(); n != 0 { |
| + t.Fatalf("Shutdown left %d entr(ies) behind", n) |
| + } |
| } |
| // execGenBackend wakes through a real ExecBackend's StartGen (a placed flat |
diff --git a/internal/activator/shutdown_lease_test.go b/internal/activator/shutdown_lease_test.go| new file mode 100644 |
| index 000000000..e319ad7a2 |
| --- /dev/null |
| +++ b/internal/activator/shutdown_lease_test.go |
| @@ -0,0 +1,183 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "testing" |
| + "time" |
| +) |
| + |
| +// Discussion #292, the production sequence with swaps off. `systemctl |
| +// restart carlos-edge` stops every carlos-instance@ unit first |
| +// (BindsTo= and After=carlos-edge.service), while the old agent is still |
| +// running and has not had its SIGTERM. The agent sees the child gone, |
| +// recycles the entry and gives its lease back, and a request queued on |
| +// the stop wakes the host again, taking a fresh lease. That wake's start |
| +// is a systemctl start job systemd queues behind the edge restart, so it |
| +// is still in flight when the SIGTERM arrives. Shutdown has to abort it |
| +// and give its lease back before the process exits; otherwise the next |
| +// agent refuses the host for the lease's whole TTL. |
| +func TestShutdownAbortsAWakeAndReleasesItsLease(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + host := "restart.example.com" |
| + arrangeAwake(t, a, testRoute(host)) |
| + |
| + // systemd stops the bound unit, well before the agent's SIGTERM. |
| + b.lastProc().kill() |
| + queued := make(chan struct{}) |
| + b.mu.Lock() |
| + b.startQueued = queued |
| + b.mu.Unlock() |
| + t.Cleanup(func() { close(queued) }) |
| + |
| + reqErr := make(chan error, 1) |
| + go func() { |
| + release, err := a.beginRequest(context.Background(), testRoute(host)) |
| + if err == nil { |
| + release() |
| + } |
| + reqErr <- err |
| + }() |
| + waitFor(t, "the re-wake to reach its queued start", func() bool { return b.startCount() == 2 }) |
| + if acq, _, _ := b.counts(); acq != 2 { |
| + t.Fatalf("acquires = %d, want 2 (the first wake and the re-wake)", acq) |
| + } |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + if ctx.Err() != nil { |
| + t.Fatal("Shutdown ran out of time instead of aborting the wake") |
| + } |
| + |
| + // Every lease this agent took is given back by the time Shutdown |
| + // returns: the first wake's through its replica's Close, the |
| + // re-wake's through Release. |
| + acq, _, _ := b.counts() |
| + if got := b.releaseCount(); got != acq { |
| + t.Fatalf("lease releases = %d after Shutdown, want %d (one per acquire)", got, acq) |
| + } |
| + for _, err := range b.releaseCtxErrors() { |
| + if err != nil { |
| + t.Fatalf("a lease was released under a dead context (%v)", err) |
| + } |
| + } |
| + if n := a.entryCount(); n != 0 { |
| + t.Fatalf("Shutdown left %d entr(ies) behind", n) |
| + } |
| + select { |
| + case err := <-reqErr: |
| + if err == nil { |
| + t.Fatal("the request was served by a wake Shutdown aborted") |
| + } |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the queued request never returned") |
| + } |
| +} |
| + |
| +// Once Shutdown has begun, nothing wakes. A request queued on a stop must |
| +// not wake the host afresh when that stop finishes after Shutdown began: |
| +// the wake would take a lease once Shutdown had stopped looking. Startup |
| +// is closed, as it is on a running agent, and the request is already past |
| +// the startup gate (queued on a Bounce's stop) when the shutdown begins. |
| +func TestNoWakeOnceShutdownHasBegun(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + started := make(chan struct{}) |
| + close(started) |
| + a.Startup = started |
| + host := "queued.example.com" |
| + arrangeAwake(t, a, testRoute(host)) |
| + stopGate := make(chan struct{}) |
| + b.lastProc().setStopGate(stopGate) |
| + if !a.Bounce(host) { |
| + t.Fatal("Bounce(awake host) = false") |
| + } |
| + waitFor(t, "the Bounce's stop", func() bool { return b.lastProc().stopEntered() == 1 }) |
| + |
| + reqErr := make(chan error, 1) |
| + go func() { |
| + release, err := a.beginRequest(context.Background(), testRoute(host)) |
| + if err == nil { |
| + release() |
| + } |
| + reqErr <- err |
| + }() |
| + // The request is queued on the stop before the shutdown begins. |
| + time.Sleep(50 * time.Millisecond) |
| + |
| + done := make(chan struct{}) |
| + go func() { |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + close(done) |
| + }() |
| + waitFor(t, "Shutdown to begin", func() bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return a.shuttingDown |
| + }) |
| + close(stopGate) |
| + <-done |
| + |
| + select { |
| + case err := <-reqErr: |
| + if err == nil { |
| + t.Fatal("a request woke the host after Shutdown had begun") |
| + } |
| + case <-time.After(5 * time.Second): |
| + t.Fatal("the queued request never returned") |
| + } |
| + if acq, _, starts := b.counts(); acq != 1 || starts != 1 { |
| + t.Fatalf("acquires, starts = %d, %d; want 1, 1 (nothing woken after Shutdown began)", acq, starts) |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if n := a.entryCount(); n != 0 { |
| + t.Fatalf("%d entr(ies) live after Shutdown", n) |
| + } |
| +} |
| + |
| +// An exit seen once Shutdown has begun is the shutdown's, not the |
| +// build's: neither the sweep nor a request reads it as unexpected or |
| +// recycles it. Shutdown's own stop takes the entry down. |
| +func TestAnExitAfterShutdownBeganIsNotUnexpected(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + host := "exit.example.com" |
| + arrangeAwake(t, a, testRoute(host)) |
| + b.lastProc().kill() |
| + // The moment between Shutdown announcing itself and taking its |
| + // snapshot of the entries. |
| + a.mu.Lock() |
| + a.shuttingDown = true |
| + a.mu.Unlock() |
| + |
| + a.sweepOnce() |
| + if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| + t.Fatal("a request was served once Shutdown had begun") |
| + } |
| + a.mu.Lock() |
| + e := a.entries[host] |
| + state, unexpected := actAwake, false |
| + if e != nil { |
| + state, unexpected = e.state, e.unexpectedExit |
| + } |
| + a.mu.Unlock() |
| + if e == nil || state != actAwake { |
| + t.Fatalf("the exit was recycled after Shutdown began (entry %v, state %v)", e != nil, state) |
| + } |
| + if unexpected { |
| + t.Fatal("an exit after Shutdown began was read as unexpected") |
| + } |
| + if acq, _, _ := b.counts(); acq != 1 { |
| + t.Fatalf("acquires = %d, want 1", acq) |
| + } |
| + |
| + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) |
| + defer cancel() |
| + a.Shutdown(ctx) |
| + if n := a.entryCount(); n != 0 { |
| + t.Fatalf("Shutdown left %d entr(ies) behind", n) |
| + } |
| + if got := b.releaseCount(); got != 1 { |
| + t.Fatalf("lease releases = %d, want 1", got) |
| + } |
| +} |
diff --git a/internal/activator/testdata/swapsim/mutate.sh b/internal/activator/testdata/swapsim/mutate.sh| index 5011d2e10..0e56bbb8e 100755 |
| --- a/internal/activator/testdata/swapsim/mutate.sh |
| +++ b/internal/activator/testdata/swapsim/mutate.sh |
| @@ -76,7 +76,7 @@ M2|swap.go|swapsim:M2|TestSwapWitnessAgentRestartMidSwap|s/a\.addMemberLocked\(e |
| M3|activator.go|swapsim:M3|TestSwapWitnessHoldExecsTheStagedBuild|s/e\.pin != nil/false/ |
| M4|opset.go|swapsim:M4|TestSwapWitnessBarrierJoinsBeforeTheLease,TestSwapWitnessAgentRestartMidSwap|s/proven := a\.stopMember\(e, m\)/proven := true; go a.stopMember(e, m)/ |
| M5|swapadmit.go|swapsim:M5|TestSwapWitnessNoAdmissionOntoAStoppingEntry|s/!waking \&\& e\.state != actAwake/false/ |
| -M6|activator.go|swapsim:M6|TestSwapWitnessShutdownJoinsAWakingEntry|s/if a\.gens \{/if false {/ |
| +M6|activator.go|swapsim:M6|TestSwapWitnessShutdownJoinsAWakingEntry,TestShutdownJoinsAWakingEntryWithSwapOff|s/waking = append\(waking, e\)/_ = waking/ |
| M7|opset.go|swapsim:M7|TestSwapWitnessMemberStopProvesTheGenerationChildEmpty|s/return a\.genTerminated\(rt, m\.gen, m\.proc\)/return true || a.genTerminated(rt, m.gen, m.proc)/ |
| M8|activator.go|swapsim:M8|TestSwapWitnessStopProvesTheCurrentGenerationEmpty|s/if a\.gens \&\& !a\.releasable\(e\)/if false \&\& !a.releasable(e)/ |
| M9|activator.go|swapsim:M9|TestSwapWitnessUnprovenWakeKeepsTheLease|s/w\.genUnproven = genStartUnproven\(err\)/w.genUnproven = false/' |