amadan

carlos / platform Public

activator: stale sidecar restarts are discarded by entry identity; pending stops re-call Stop and finish under Shutdown (plan 3, task 5 fix round 1)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev c8d540da066e89711003fdb7284a63f93cde09df parent 66b33fc
2 files changed, +256 −88
  • internal/activator/activator.go +159 −84
  • internal/activator/activator_test.go +97 −4

CI passed — run details

diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index e76214b69..b88df25d1 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -309,23 +309,19 @@ type actEntry struct {
sidecarBackoff time.Duration // last restart delay (doubles, capped)
sidecarRetryAt time.Time // earliest next restart attempt
sidecarStarting bool // a restart's backend.StartSidecar call is in flight, off a.mu (design §6.7)
+ // sidecarStartDone is closed by the in-flight restart goroutine when it
+ // finishes (installed or discarded), non-nil exactly while
+ // sidecarStarting is true. stop() waits on it, bounded by
+ // sidecarStopTimeout, instead of polling sidecarStarting.
+ sidecarStartDone chan struct{}
// pendingStop marks a stop whose proc.Stop returned ErrStillRunning
// (design §6.3): termination is unproven, so stop() returned without
// releasing the lease or forgetting the entry, and the entry stays in
- // stopping. sweepOnce polls proc.Exited() for every pendingStop entry
- // and finishes the stop (finishStop) the moment it's proven.
+ // stopping. sweepOnce (and, during Shutdown, finishPendingStops) polls
+ // proc.Exited() for every pendingStop entry and finishes the stop
+ // (finishPendingStop) the moment it's proven.
pendingStop bool
-
- // wakeGen counts this host's wakes, bumped each time an entry becomes
- // awake (in wake()). A sidecar restart goroutine captures it before
- // dropping a.mu (design §6.7) and re-checks it on the far side of the
- // async StartSidecar call: a generation mismatch means the entry moved
- // on (stopped and possibly rewoken) while the call was in flight, so
- // the goroutine's result belongs to a wake that's no longer current and
- // must be discarded (stopping whatever it started) rather than
- // installed.
- wakeGen uint64
}
// Sidecar restart pacing — Restart=always semantics without a supervisor: the
@@ -338,11 +334,34 @@ const (
sidecarHealthyRun = time.Minute
)
-// sidecarStopTimeout bounds the sidecar's stop at sleep time. `sidecar run`
-// has no drain handler, so the SIGTERM is effectively immediate; the SIGKILL
-// fallback behind this timeout is insurance.
+// sidecarStopTimeout serves two related bounds. First, its name: the
+// sidecar's own stop at sleep time — `sidecar run` has no drain handler, so
+// the SIGTERM is effectively immediate, and the SIGKILL fallback behind this
+// timeout is insurance. Second (design §6.7): how long stop() waits on
+// sidecarStartDone for an in-flight sidecar restart to land before giving up
+// and proceeding on whatever e.sidecar currently is. The restart goroutine's
+// own entry-identity check makes that second wait purely a promptness
+// trade-off, never a correctness one — nothing is orphaned or misplaced
+// whether it fires in time or times out.
const sidecarStopTimeout = 5 * time.Second
+// stopDiscardedSidecar stops a sidecar process a restart attempt returned
+// but could not install — the wake it belonged to moved on (stopped, or
+// replaced by a fresh wake of the same host) while the call was in flight.
+// Bounded by sidecarStopTimeout so a wedged Stop can't leak the goroutine
+// forever; fire-and-forget, since nothing is waiting on this beyond the
+// process actually dying.
+func stopDiscardedSidecar(sc Proc) {
+ if sc == nil {
+ return
+ }
+ go func() {
+ ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
+ defer cancel()
+ sc.Stop(ctx) //nolint:errcheck
+ }()
+}
+
// Build convergence: how long an awake child may serve a symlink-stale binary
// before the sweep gracefully cycles it, and how many to cycle per tick.
// - ConvergeGrace comfortably exceeds one sweep interval (≤30s) and a
@@ -710,7 +729,6 @@ func (a *Service) wake(e *actEntry) {
delete(a.entries, e.host) // next request retries from scratch
} else {
e.state = actAwake
- e.wakeGen++ // this entry's sidecar-restart goroutines (design §6.7) key on this generation
e.rep, e.proc = w.rep, w.proc
e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted
e.sidecarUncertain = w.sidecarUncertain
@@ -1034,20 +1052,21 @@ func (a *Service) stop(e *actEntry) {
// a.mu (design §6.7) and may still be in flight when a stop begins. Wait,
// bounded by sidecarStopTimeout, for it to land — either it installs a
// fresh e.sidecar in time and the co-stop below stops that one directly,
- // or it doesn't and the restart goroutine's own generation check (it
+ // or it doesn't and the restart goroutine's own entry-identity check (it
// re-reads a.entries[host] on the far side of the call) finds this entry
// gone or stopping and stops whatever it started instead. Either way
// nothing is left orphaned; this wait only decides which of the two
- // stops it.
- deadline := time.Now().Add(sidecarStopTimeout)
- for {
- a.mu.Lock()
- starting := e.sidecarStarting
- a.mu.Unlock()
- if !starting || time.Now().After(deadline) {
- break
+ // stops it. sidecarStartDone is closed by the restart goroutine when it
+ // finishes — however it finishes — so this waits on that signal directly
+ // rather than polling sidecarStarting on a timer.
+ a.mu.Lock()
+ startDone := e.sidecarStartDone
+ a.mu.Unlock()
+ if startDone != nil {
+ select {
+ case <-startDone:
+ case <-time.After(sidecarStopTimeout):
}
- time.Sleep(20 * time.Millisecond)
}
// Co-stop, sidecar FIRST — before the next-due probe and the instance's
@@ -1124,12 +1143,56 @@ func (a *Service) stop(e *actEntry) {
a.finishStop(e, outcome)
}
-// finishStop is stop()'s tail: the final checkpoint, lease release, and
-// forgetting the entry. Split out (design §6.3, codex R2-5) so sweepOnce can
-// run it once a pendingStop entry's proc.Exited() finally proves the
-// termination that stop() itself couldn't — outcome carries forward whatever
-// stop() had already observed (the sidecar co-stop, the due-time probe)
-// before it hit ErrStillRunning and returned.
+// finishPendingStop finishes a stop that was left pending because proc.Stop
+// returned ErrStillRunning (design §6.3): now that proc.Exited() has proven
+// termination, spec §6.3 calls for retrying proc.Stop once more before
+// declaring the main process closed — "sweepOnce retries proc.Stop on its
+// cadence until done closes" — so outcome.mainStop reflects an actual
+// observation rather than an assumption baked in at the ErrStillRunning
+// return. A unitProc that has already exited takes its done branch and
+// returns nil immediately; a childProc's Stop is a no-op on an
+// already-exited process either way, so retrying costs nothing once the
+// process is well and truly gone. mainExited is unconditionally true — the
+// exit that made this entry eligible for this pass IS that proof — whatever
+// this retried Stop call itself reports.
+func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) {
+ ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ err := e.proc.Stop(ctx)
+ cancel()
+ outcome.mainStop, outcome.mainExited = err == nil, true
+ if err != nil {
+ log.Printf("activator: stop %s: %v", e.host, err)
+ }
+ a.finishStop(e, outcome)
+}
+
+// finishPendingStops scans for entries left pending because proc.Stop
+// returned ErrStillRunning (design §6.3, codex R2-5) and finishes every one
+// whose proc has since exited, via finishPendingStop. Called from
+// sweepOnce's own tick and, once more, from Shutdown — the sweeper is
+// stopped by the time Shutdown runs, so without this second call a process
+// that exits during shutdown would never get its final checkpoint. The
+// caller holds a.mu.
+func (a *Service) finishPendingStops() {
+ for _, e := range a.entries {
+ if e.pendingStop && e.proc != nil && e.proc.Exited() {
+ e.pendingStop = false
+ outcome := e.stopOutcome
+ if outcome == nil {
+ outcome = &stopObservation{}
+ }
+ go a.finishPendingStop(e, outcome)
+ }
+ }
+}
+
+// finishStop is stop()'s (and finishPendingStop's) tail: the final
+// checkpoint, lease release, and forgetting the entry. Split out (design
+// §6.3, codex R2-5) so a pendingStop entry's eventual finish shares it with
+// stop()'s own ordinary path — outcome carries forward whatever stop() had
+// already observed (the sidecar co-stop, the due-time probe) before it hit
+// ErrStillRunning and returned, plus whatever finishPendingStop's own
+// retried proc.Stop just added.
func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
// Close = final sync + lease release. After a lost lease the release part
// fails harmlessly (someone else owns the lock); the sync still preserves
@@ -1215,20 +1278,9 @@ func (a *Service) sweepOnce() {
// ErrStillRunning (design §6.3, codex R2-5): termination wasn't proven
// at the time, so stop() kept the lease and returned without forgetting
// the entry. Checked before the idle pass below for the same reason it's
- // a separate loop at all — a pendingStop entry is in actStopping, not
+ // a separate call at all — a pendingStop entry is in actStopping, not
// actAwake, so the idle pass would never reach it.
- for _, e := range a.entries {
- if e.pendingStop && e.proc != nil && e.proc.Exited() {
- e.pendingStop = false
- outcome := e.stopOutcome
- if outcome == nil {
- outcome = &stopObservation{}
- }
- // The exit that unblocked this loop IS the proof stop() lacked.
- outcome.mainStop, outcome.mainExited = true, true
- go a.finishStop(e, outcome)
- }
- }
+ a.finishPendingStops()
cycled := 0 // build-convergence cycles started this tick (rate-limited)
for _, e := range a.entries {
if a.fences[e.host] != 0 {
@@ -1323,65 +1375,73 @@ func (a *Service) sweepOnce() {
// R2-6): held under the lock, a stalled StartSidecar (a hung
// exec, a wedged control socket) would block every other host's
// wake and sweep behind it. sidecarStarting keeps a second
- // attempt from launching while one is already in flight;
- // wakeGen, re-checked on the far side, catches an entry that
- // stopped (and possibly rewoke under a new generation) while
- // the call was running, so a stale attempt's result is stopped
- // rather than installed onto a wake it doesn't belong to.
+ // attempt from launching while one is already in flight.
+ //
+ // The far side re-checks by ENTRY IDENTITY (e0, the *actEntry
+ // itself), not a generation counter (codex review round 1,
+ // Critical): a per-entry counter that starts at 0 and is bumped
+ // once when an entry becomes awake reads 1 for every entry's
+ // first wake, so it can never tell "this entry" apart from "a
+ // completely different entry — for the same host, after a stop
+ // and rewake — that also just had its first wake". A pointer
+ // comparison against a.entries[host] has no such collision: a
+ // fresh wake always allocates a brand new *actEntry (begin() and
+ // the due-wake path both do `&actEntry{...}`), so identity alone
+ // says whether this attempt's wake is still the current one.
if !e.sidecarStarting {
e.sidecarStarting = true
- gen := e.wakeGen
+ done := make(chan struct{})
+ e.sidecarStartDone = done
+ e0 := e
host := e.host
route := e.route
backoff := e.sidecarBackoff
go func() {
+ defer close(done)
ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
sc, err := a.backend.StartSidecar(ctx, route)
cancel()
a.mu.Lock()
defer a.mu.Unlock()
- e := a.entries[host]
- if e == nil || e.wakeGen != gen {
- // The entry is gone (stopped and forgotten) or this
- // host has moved on to a fresh wake under a new
- // generation — that entry's own sidecarStarting, if
- // any, belongs to a different attempt and is not
- // this goroutine's to touch. Just stop whatever
- // came back so it doesn't outlive the wake it
- // doesn't belong to.
- if sc != nil {
- go sc.Stop(context.Background()) //nolint:errcheck
- }
+ if a.entries[host] != e0 {
+ // This host's map slot no longer holds e0 at all —
+ // forgotten (stopped and deleted) or replaced by an
+ // entirely different entry from a fresh wake. e0 is
+ // unreachable from here on, so there's nothing left
+ // on it to update; just stop whatever came back so
+ // it doesn't outlive the wake it doesn't belong to,
+ // or clobber a fresh wake's own sidecar.
+ stopDiscardedSidecar(sc)
return
}
- // Still the same entry and generation: this attempt owns
- // its sidecarStarting flag regardless of what state the
- // entry is in now — a stop may have begun while the call
- // was in flight (stop() itself waits, bounded, for this
- // to clear; leaving it set here would make that wait run
- // the full sidecarStopTimeout for nothing).
- e.sidecarStarting = false
- if e.state != actAwake {
- // Stopped mid-call: don't install onto a wake that's
- // on its way out.
- if sc != nil {
- go sc.Stop(context.Background()) //nolint:errcheck
- }
+ // Still e0: this attempt owns its sidecarStarting flag
+ // and clears it (and the done channel already closing
+ // via the defer above) unconditionally, regardless of
+ // what state e0 is in now — a stop may have begun
+ // (flipping e0 to actStopping) while the call was in
+ // flight, and stop()'s own wait on sidecarStartDone
+ // must not run the full sidecarStopTimeout for nothing.
+ e0.sidecarStarting = false
+ e0.sidecarStartDone = nil
+ if e0.state != actAwake {
+ // Stopped mid-call, not yet forgotten: don't install
+ // onto a wake that's on its way out.
+ stopDiscardedSidecar(sc)
return
}
if err != nil {
- e.sidecarUncertain = true
+ e0.sidecarUncertain = true
if sc == nil {
log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", host, err, backoff)
} else {
log.Printf("activator: SIDECAR restart for %s failed: %v — retaining returned process", host, err)
}
- e.sidecar = sc
+ e0.sidecar = sc
return
}
log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", host, backoff)
- e.sidecar = sc
- e.sidecarStarted = a.now()
+ e0.sidecar = sc
+ e0.sidecarStarted = a.now()
}()
}
}
@@ -1497,6 +1557,12 @@ func (a *Service) Shutdown(ctx context.Context) {
<-a.sweepDone
}
a.mu.Lock()
+ // The sweeper is stopped as of the line above, so nothing else will ever
+ // poll a pendingStop entry (design §6.3) again on its own — run the same
+ // pass once here, and again on the ticker below while waiting, so a
+ // process that finally exits during shutdown still gets its final
+ // checkpoint instead of being abandoned mid-drain.
+ a.finishPendingStops()
var waits []chan struct{}
for _, e := range a.entries {
switch e.state {
@@ -1513,11 +1579,20 @@ func (a *Service) Shutdown(ctx context.Context) {
}
}
a.mu.Unlock()
+ poll := time.NewTicker(100 * time.Millisecond)
+ defer poll.Stop()
for _, w := range waits {
- select {
- case <-w:
- case <-ctx.Done():
- return
+ for done := false; !done; {
+ select {
+ case <-w:
+ done = true
+ case <-poll.C:
+ a.mu.Lock()
+ a.finishPendingStops()
+ a.mu.Unlock()
+ case <-ctx.Done():
+ return
+ }
}
}
}
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 443056eb0..a7045124d 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -212,6 +212,12 @@ type fakeBackend struct {
// sidecarGate, when non-nil, blocks StartSidecar() until closed — models
// a stalled restart call so a test can assert it doesn't hold a.mu.
sidecarGate chan struct{}
+ // sidecarStartEntered, when non-nil, is closed the first time
+ // StartSidecar is entered (before it waits on sidecarGate) — lets a test
+ // wait for a restart's call to actually begin rather than sleeping a
+ // guessed duration.
+ sidecarStartEntered chan struct{}
+ sidecarStartEnteredOnce sync.Once
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
@@ -311,7 +317,11 @@ func (b *fakeBackend) HasSidecar(registry.Route) bool {
func (b *fakeBackend) StartSidecar(context.Context, registry.Route) (Proc, error) {
b.mu.Lock()
gate := b.sidecarGate
+ entered := b.sidecarStartEntered
b.mu.Unlock()
+ if entered != nil {
+ b.sidecarStartEnteredOnce.Do(func() { close(entered) })
+ }
if gate != nil {
<-gate
}
@@ -4126,8 +4136,16 @@ func TestStopDoesNotReleaseTheLeaseOverAnUnprovenStop(t *testing.T) {
t.Fatal("entry forgotten before the stop was proven")
}
p.exit()
- a.sweepOnce()
- waitFor(t, "the lease to release once termination is proven", func() bool { return b.releaseCount() == 1 })
+ // Drive the sweep from inside the wait itself: finishing a pendingStop
+ // entry now retries proc.Stop (design §6.3) on the sweeper's own
+ // cadence, so a single sweepOnce() right after exit() is a race against
+ // exactly when the sweeper would next notice — polling sweepOnce()
+ // alongside the release count removes that race instead of hoping one
+ // call lands after exit() is visible.
+ waitFor(t, "the lease to release once termination is proven", func() bool {
+ a.sweepOnce()
+ return b.releaseCount() == 1
+ })
}
// The sidecar restart must not hold the activator mutex across the start
@@ -4143,13 +4161,19 @@ func TestSidecarRestartDoesNotHoldTheLockAcrossTheStart(t *testing.T) {
b.lastSidecar().exit() // the sweeper now wants a restart
gate := make(chan struct{})
+ entered := make(chan struct{})
+ b.mu.Lock()
b.sidecarGate = gate // StartSidecar blocks on this channel when non-nil
+ b.sidecarStartEntered = entered
+ b.mu.Unlock()
clk.advance(time.Minute)
- go a.sweepOnce() // blocks inside StartSidecar until the gate closes
- time.Sleep(50 * time.Millisecond)
+ a.sweepOnce() // launches the restart; it blocks inside StartSidecar on gate
+ <-entered // wait for the restart's call to actually begin, not a guessed sleep
// host b has no sidecar of its own — HasSidecar false keeps its wake's
// own co-spawn (unrelated to what's under test) off the same gate.
+ b.mu.Lock()
b.sidecarStaged = false
+ b.mu.Unlock()
done := make(chan struct{})
go func() {
// wake b.example.com
@@ -4190,3 +4214,72 @@ func TestDoWakeRefreshesTheRouteFromTheRegistry(t *testing.T) {
t.Fatalf("Start saw Socket %q, want the registry's %q", got, fresh.Socket)
}
}
+
+// A restart whose StartSidecar call is still stalled must never install
+// onto a DIFFERENT wake of the same host once it finally returns (codex
+// review round 1, Critical). The bug it regression-tests: the discard check
+// used to compare a per-entry generation counter that is always 1 on an
+// entry's first wake (entries are never reused, so nothing ever bumps it a
+// second time), which meant it could never tell "this entry" apart from "a
+// completely different entry — for the same host, after a stop and rewake —
+// that also just had its own first wake". A stale restart could install its
+// result straight over a fresh wake's own sidecar. The fix (this test's
+// assertion) compares entry identity: the *actEntry pointer captured when
+// the restart launched, not a counter on it.
+//
+// This test legitimately costs several seconds of wall time: stop() itself
+// waits on the in-flight restart, bounded by sidecarStopTimeout, before it
+// can finish taking the old entry down (that wait is a separate, already
+// covered property — see TestSidecarRestartDoesNotHoldTheLockAcrossTheStart
+// and stop()'s own doc comment) — the file has no seam to shrink
+// sidecarStopTimeout for tests, so this accepts the wall time rather than
+// adding one for a single regression test.
+func TestStaleSidecarRestartNeverInstallsOntoAFreshWake(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ b.sidecarStaged = true
+ host := "stale.example.com"
+
+ arrangeAwake(t, a, testRoute(host))
+ b.lastSidecar().exit() // the sweeper now wants a restart
+
+ gate := make(chan struct{})
+ entered := make(chan struct{})
+ b.mu.Lock()
+ b.sidecarGate = gate
+ b.sidecarStartEntered = entered
+ b.mu.Unlock()
+ a.sweepOnce() // launches the restart; it blocks inside StartSidecar on gate
+ <-entered
+
+ if !a.Bounce(host) {
+ t.Fatal("Bounce did not find the awake entry")
+ }
+ deadline := time.Now().Add(8 * time.Second)
+ for a.entryCount() != 0 && time.Now().Before(deadline) {
+ time.Sleep(20 * time.Millisecond)
+ }
+ if a.entryCount() != 0 {
+ t.Fatal("Bounce'd entry never finished stopping (the stalled restart wedged the stop past its bound)")
+ }
+
+ // A fresh wake of the same host gets its own sidecar — this is what the
+ // stale restart, still stalled, must not be allowed to overwrite.
+ b.mu.Lock()
+ b.sidecarGate = nil // the fresh wake's own co-spawn must not block on it too
+ b.mu.Unlock()
+ arrangeAwake(t, a, testRoute(host))
+ freshSidecar := b.lastSidecar()
+ wantStarts := b.sidecarStartCount() + 1 // + the stale call, about to land
+
+ close(gate) // let the stale StartSidecar call finally return
+ waitFor(t, "the stale restart to resolve", func() bool { return b.sidecarStartCount() == wantStarts })
+ stale := b.lastSidecar()
+ waitFor(t, "the stale sidecar to be stopped", func() bool { return stale.stopCalls() >= 1 })
+
+ a.mu.Lock()
+ installed := a.entries[host].sidecar
+ a.mu.Unlock()
+ if installed != freshSidecar {
+ t.Fatal("the stale restart installed over the fresh wake's own sidecar")
+ }
+}