carlos / platform Public

Close the respawn's shed inside its done signal

Paul Campbell pushed by paul@keymail.dev 5fb391ede2f4208b1b1396cf9c76e8c206e7bbc8 parent f185a40
2 files changed, +84 −7
  • internal/activator/activator.go +21 −6
  • internal/activator/activator_test.go +63 −1
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 729589261..10cf5ad9a 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -976,10 +976,12 @@ func (a *Service) Bounce(host string) bool {
// socket. beginRequest, BeginSidecarRequest and ServeHTTP are untouched.
// newWorkerEntry builds the waking entry for a standalone sidecar: the route's
-// own ceilings (0 = platform default / no ceiling) and a freshly minted
-// per-wake token. Every field it sets is written BEFORE the entry is published
-// to a.entries and the wake goroutine starts, so the wake path may read them
-// without a.mu; nothing rewrites them afterwards.
+// own ceilings (0 = platform default / no ceiling) and the channel its settle
+// window waits on. The wake's token is NOT minted here — it is minted inside
+// doWake, under the lease and immediately before the spawn, and published onto
+// the entry there. Every field this does set is written BEFORE the entry is
+// published to a.entries and the wake goroutine starts, so the wake path may
+// read them without a.mu; nothing rewrites them afterwards.
func (a *Service) newWorkerEntry(rt registry.Route) *actEntry {
return &actEntry{
host: rt.Host,
@@ -1250,6 +1252,15 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
e.respawnDone = make(chan struct{})
done, rt, host := e.respawnDone, e.route, e.host
go func() {
+ // respawnDone closes LAST, once this goroutine is finished with every
+ // child it created — including one it had to SHED. That ordering is
+ // the invariant stop() leans on: when respawnDone is observed closed,
+ // no child spawned by this respawn is still running, so the final
+ // checkpoint and lease release below it cannot land while a live
+ // process is still pointed at the same database. Closing it before the
+ // shed (as this first did) only narrowed that race; it did not close
+ // it.
+ defer close(done)
var (
proc Proc
spawnVer string
@@ -1283,7 +1294,6 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) {
a.mu.Lock()
e.workerRespawning = false
- close(done)
live := a.entries[host] == e && e.state == actAwake && !a.shuttingDown
switch {
case !wanted || !live:
@@ -1443,11 +1453,16 @@ func (a *Service) stop(e *actEntry) {
respawning := e.workerRespawning
a.mu.Unlock()
if respawning && waitRespawn != nil {
+ // The bound covers the respawn goroutine's OWN worst case end to end —
+ // retiring the old child, the exec, and shedding a raced one — so this
+ // wait can never give up on a respawn that is still making progress.
+ bound := time.NewTimer(actStopTimeout + workerSpawnTimeout + sidecarStopTimeout)
select {
case <-waitRespawn:
- case <-time.After(workerSpawnTimeout + actStopTimeout):
+ case <-bound.C:
log.Printf("activator: %s respawn did not finish before its stop — proceeding without it", e.host)
}
+ bound.Stop() // the common case closes in milliseconds; don't leave a 2m timer standing
}
// Co-stop, sidecar FIRST — before the next-due probe and the instance's
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index 30652589e..0dc9ddc83 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -212,6 +212,10 @@ type fakeBackend struct {
// yet installed) rather than hoping to hit it.
workerStartDelay time.Duration
workerEnterN int // StartWorker calls ENTERED (before the delay)
+ // workerStopGate, when set, is installed on every worker child spawned
+ // from then on: its Stop blocks until the gate closes, so a test can hold
+ // a child mid-drain and watch what happens around it.
+ workerStopGate chan struct{}
// order interleaves backend calls and proc stops (start:instance,
// start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering
// assertions.
@@ -355,7 +359,7 @@ func (b *fakeBackend) StartWorker(ctx context.Context, rt registry.Route, env ma
if b.workerErr != nil {
return nil, "", b.workerErr
}
- p := &fakeProc{label: "worker", rec: b, dead: b.workerDeadOnStart}
+ p := &fakeProc{label: "worker", rec: b, dead: b.workerDeadOnStart, stopGate: b.workerStopGate}
b.workers = append(b.workers, p)
// Same symlink model as Start: execVerFn, when wired, is the live/<host>
// target this exec would have read — so a convergence test can move the
@@ -2840,6 +2844,64 @@ func TestShutdownStopsARespawnFromOutlivingTheLease(t *testing.T) {
}
}
+// THE invariant this task has been circling: when respawnDone is observed
+// closed, no child spawned by that respawn is still running.
+//
+// The case that proves it is the raced one — a stop lands after the respawn
+// has committed to spawning, so the child it produces must be SHED. If
+// respawnDone closed before that shed finished, stop() would fall through and
+// reach rep.Close — final sync plus lease release — with a live process still
+// pointed at the same database. Narrowing that window is not closing it: the
+// shed has to complete first.
+func TestRespawnShedsARacedChildBeforeTheLeaseIsReleased(t *testing.T) {
+ a, b, clk, ver := shippingActivator(t, time.Hour)
+ if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil {
+ t.Fatalf("WakeWorker: %v", err)
+ }
+ rep := b.lastRep()
+ // The gate applies to children spawned from now on — i.e. the respawn's,
+ // the one that will have to be shed. Its drain is held open so the
+ // ordering is observable rather than a microsecond wide.
+ gate := make(chan struct{})
+ b.mu.Lock()
+ b.workerStartDelay = 200 * time.Millisecond
+ b.workerStopGate = gate
+ b.mu.Unlock()
+
+ ver.Store("v2")
+ a.sweepOnce()
+ clk.advance(ConvergeGrace + time.Minute)
+ a.sweepOnce()
+ waitFor(t, "the respawn's exec to begin", func() bool { return b.workerEnters() == 2 })
+
+ // The stop lands mid-exec: past the pre-spawn check, so the child is
+ // already coming and can only be shed.
+ if !a.Bounce("bot.x.internal") {
+ t.Fatal("Bounce found no awake worker to stop")
+ }
+ waitFor(t, "the raced child to be spawned", func() bool { return b.workerStarts() == 2 })
+
+ // It is now draining, held by the gate. The lease must still be held.
+ deadline := time.Now().Add(300 * time.Millisecond)
+ for time.Now().Before(deadline) {
+ if rep.isClosed() {
+ t.Fatal("the lease was released while the respawn's raced child was still draining — a live process is pointed at a database we no longer hold")
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+
+ close(gate) // let the shed finish
+ waitFor(t, "the stop to complete", func() bool { return a.entryCount() == 0 })
+ if n := rep.closeCount(); n != 1 {
+ t.Fatalf("lease closes = %d, want exactly 1", n)
+ }
+ for i, p := range b.workers {
+ if !p.isStopped() {
+ t.Fatalf("worker child %d survived the stop", i)
+ }
+ }
+}
+
// Shutdown CANCELS an in-flight worker spawn rather than waiting it out. The
// wait is right everywhere else, but a shutdown answers to a deadline it does
// not set: systemd's TimeoutStopSec defaults to 90s and workerSpawnTimeout