amadan

carlos / platform Public

zdd plan 3 final fix 2: a shed respawn child is kept until its exit is proven (R2-1)

With generations off, a respawn child refused at its installation is now
held on the entry (e.shed) with its build pinned, registered before its
Stop. Neither the pin nor a terminal stop's lease release happens before
the child is proven exited: stop() keeps the stop pending while a respawn
is still in flight past its bounded wait or its shed child is unproven,
finishPendingStops finishes it only once respawnSettledLocked holds, and
reapShedLocked re-stops and proves the child on the sweep. No respawn
execs beside a kept shed child.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 895fc8602272e7308e4bf48c91a5285d5dc724e9 parent 37ffe34
4 files changed, +248 −8
  • internal/activator/activator.go +133 −5
  • internal/activator/activator_test.go +4 −1
  • internal/activator/generations_file.go +8 −2
  • internal/activator/worker_respawn_exec_test.go +103 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index a0951db29..5046de730 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -545,6 +545,23 @@ type actEntry struct {
// not release the lease while a spawn against the same DB is in flight.
workerRespawning bool
respawnDone chan struct{}
+ // shed is a respawn child, refused at its installation with generations
+ // off, whose exit is not proven yet (final review R2-1). With
+ // generations on the operation set holds such a child as a member. The
+ // entry keeps it, and its build pinned, until the sweep proves it
+ // exited (reapShedLocked): a terminal stop keeps the lease until then,
+ // and no respawn execs beside it.
+ shed *shedChild
+}
+
+// shedChild is a respawn child that was exec'd and then refused at its
+// installation, kept until its exit is proven (final review R2-1). Every
+// field is written under a.mu. reaping marks a re-stop or proof in flight.
+type shedChild struct {
+ proc Proc
+ build release.StagedIdentity
+ reaping bool
+ retriedAt time.Time
}
// Sidecar restart pacing — Restart=always semantics without a supervisor: the
@@ -2344,6 +2361,11 @@ func (a *Service) convergeWorkerLocked(e *actEntry) {
// minutes, not the entry forever.
const workerSpawnTimeout = 2 * time.Minute
+// respawnJoinBound bounds a terminal stop's wait on a supervised respawn: the
+// respawn goroutine's own worst case end to end (retiring the old child, the
+// exec, and shedding a raced one). A var so tests can shrink it.
+var respawnJoinBound = actStopTimeout + workerSpawnTimeout + sidecarStopTimeout
+
// respawnWorkerLocked replaces a worker's child on the SAME entry, keeping its
// lease and replication: only the process is new. The caller holds a.mu.
// stopOld is the child to retire first (nil when it is already dead).
@@ -2450,6 +2472,13 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
if a.gens {
oldGen.Worker = true
retired = a.genTerminated(rt, oldGen, stopOld)
+ } else {
+ // Generations off, an earlier respawn's shed child that is not
+ // proven gone may still run beside this exec (final review
+ // R2-1): nothing execs until the sweep has proven it exited.
+ a.mu.Lock()
+ retired = e.shed == nil
+ a.mu.Unlock()
}
// Still wanted? Asked HERE, before anything is spawned: past this
@@ -2508,7 +2537,7 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
switch {
case owned && !retired && live:
a.mu.Unlock()
- log.Printf("activator: worker %s NOT respawned (%s): the old generation's termination is not proven (its host cgroup may still hold a process, or the proof could not run), so nothing execs beside it; the sweep retries", host, why)
+ log.Printf("activator: worker %s NOT respawned (%s): an earlier child's termination is not proven (its host cgroup or a shed respawn child may still hold a process, or the proof could not run), so nothing execs beside it; the sweep retries", host, why)
case !wanted || !live:
a.mu.Unlock()
reason := ownErr
@@ -2533,9 +2562,31 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
shed = true
a.dropMember(e, m)
} else {
+ // Generations off (final review R2-1): the entry takes
+ // the child, and its build moves from the pending record
+ // to it in the same critical section, BEFORE the Stop.
+ // A terminal stop that gives up waiting on this respawn
+ // still finds it, and keeps the lease; the sweep reaps it
+ // once its exit is proven (reapShedLocked).
+ s := &shedChild{proc: proc, build: ident}
+ a.mu.Lock()
+ e.shed, s.retriedAt = s, a.now()
+ a.dropPendingGenerationLocked(pend)
+ a.mu.Unlock()
+ pend = 0
shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
- _ = proc.Stop(shedCtx)
+ serr := proc.Stop(shedCtx)
shedCancel()
+ a.mu.Lock()
+ gone := proc.Exited()
+ if gone && e.shed == s {
+ e.shed = nil
+ }
+ a.mu.Unlock()
+ if !gone {
+ log.Printf("activator: worker %s: the shed respawn child's exit is not proven (%v); the entry keeps it and its build until the sweep proves it", host, serr)
+ }
+ _ = a.persistGenerations()
}
}
if pend != 0 {
@@ -2745,11 +2796,11 @@ func (a *Service) stop(e *actEntry) {
// 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)
+ bound := time.NewTimer(respawnJoinBound)
select {
case <-waitRespawn:
case <-bound.C:
- log.Printf("activator: %s respawn did not finish before its stop — proceeding without it", e.host)
+ log.Printf("activator: %s respawn did not finish before its stop; stopping the rest, the lease is kept until it has", e.host)
}
bound.Stop() // the common case closes in milliseconds; don't leave a 2m timer standing
}
@@ -2902,6 +2953,21 @@ func (a *Service) stop(e *actEntry) {
a.mu.Unlock()
return
}
+ a.mu.Lock()
+ settled := a.respawnSettledLocked(e)
+ if !settled {
+ // A respawn still running past the bounded wait above, or the
+ // child it shed and could not prove gone (final review R2-1): the
+ // lease stays held, as for ErrStillRunning, and finishPendingStops
+ // finishes this stop once the respawn is done and its child proven
+ // exited.
+ a.keepPendingLocked(e, outcome)
+ }
+ a.mu.Unlock()
+ if !settled {
+ log.Printf("activator: stop %s: a respawn or the child it shed is not proven finished; keeping the lease until it is", e.host)
+ return
+ }
if a.gens && !a.releasable(e) { // swapsim:M8
// THE retain-until-proven rule (spec §7.4): unproven, the stop stays
// pending with the lease held, exactly as for ErrStillRunning, and
@@ -2957,10 +3023,13 @@ func (a *Service) finishPendingStops() {
// Members first (zero-downtime spec §7.4): a generation whose stop
// could not be proven is finished here once it exits.
a.reapMembersLocked(e)
+ // And a shed respawn child kept with generations off (final review
+ // R2-1).
+ a.reapShedLocked(e)
// A parent restore that failed when the last member left keeps its
// obligation and is retried here (final review F11).
a.retryParentRestoreLocked(e)
- if !e.pendingStop || len(e.ops.members) > 0 {
+ if !e.pendingStop || len(e.ops.members) > 0 || !a.respawnSettledLocked(e) {
continue
}
if a.gens && (e.proc == nil || e.proc.Exited()) {
@@ -3005,6 +3074,65 @@ func (a *Service) finishPendingStops() {
}
}
+// respawnDoneLocked reports whether e has no supervised respawn in flight.
+// The caller holds a.mu.
+func respawnDoneLocked(e *actEntry) bool {
+ if e.respawnDone == nil {
+ return true
+ }
+ select {
+ case <-e.respawnDone:
+ return true
+ default:
+ return false
+ }
+}
+
+// respawnSettledLocked reports whether e's supervised respawns are done
+// with every child they made: none is in flight, and no child one shed is
+// kept unproven (final review R2-1). A terminal stop finishes, and gives
+// up the lease, only then. The caller holds a.mu.
+func (a *Service) respawnSettledLocked(e *actEntry) bool {
+ return e.shed == nil && respawnDoneLocked(e)
+}
+
+// reapShedLocked finishes a shed respawn child kept with generations off
+// (final review R2-1), once the respawn that shed it is done. An exited
+// child is forgotten and its build unpinned. One still running is stopped
+// again, bounded, every actStopTimeout, the cadence finishPendingStops
+// keeps for the main process. The caller holds a.mu.
+func (a *Service) reapShedLocked(e *actEntry) {
+ s := e.shed
+ if s == nil || s.reaping || !respawnDoneLocked(e) {
+ return
+ }
+ if !s.proc.Exited() && a.now().Sub(s.retriedAt) < actStopTimeout {
+ return
+ }
+ s.reaping, s.retriedAt = true, a.now()
+ go func() {
+ if !s.proc.Exited() {
+ ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ err := s.proc.Stop(ctx)
+ cancel()
+ if err != nil && !errors.Is(err, ErrStillRunning) {
+ log.Printf("activator: %s: retry stop of the shed respawn child: %v", e.host, err)
+ }
+ }
+ a.mu.Lock()
+ s.reaping = false
+ gone := s.proc.Exited()
+ if gone && e.shed == s {
+ e.shed = nil
+ }
+ a.mu.Unlock()
+ if gone {
+ log.Printf("activator: %s: the shed respawn child has exited", e.host)
+ _ = a.persistGenerations()
+ }
+ }()
+}
+
// retryPendingStop re-issues proc.Stop for a pendingStop entry that is
// still running, off a.mu like finishPendingStop's own call. Unlike that
// one, a return here proves nothing by itself. Done closing is still what
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index f6ea62021..5fa25f5fa 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -348,6 +348,9 @@ type fakeBackend struct {
// 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{}
+ // workerStuck makes every worker child spawned from then on a stuck
+ // fakeProc: its Stop answers ErrStillRunning until exit().
+ workerStuck bool
// workerStartHook, when set, runs INSIDE StartWorker (off b.mu), so a test
// can model what the child itself does while its wake is still resolving —
// notably posting its next due time over the agent socket during the
@@ -558,7 +561,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, stopGate: b.workerStopGate}
+ p := &fakeProc{label: "worker", rec: b, dead: b.workerDeadOnStart, stopGate: b.workerStopGate, stuck: b.workerStuck}
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
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go
index b2166363e..9029ca3e3 100644
--- a/internal/activator/generations_file.go
+++ b/internal/activator/generations_file.go
@@ -70,8 +70,9 @@ func (a *Service) dropPendingGenerationLocked(k pendingGeneration) {
// generationsSnapshotLocked is what the generations file should say now: for
// every worker entry (waking, awake, mid-respawn or stopping, since its child
// may still run) its captured serving build and the build of any shed child
-// its operation set retains, the build of every retained failed worker wake
-// (unprovenWakes), and every pending build, each
+// its operation set (or, with generations off, the entry) retains, the build
+// of every retained failed worker wake (unprovenWakes), and every pending
+// build, each
// listed under its host by version when the version is a valid version name,
// and by staged path (made absolute) when it has one. A build with neither
// names nothing root could restage or sweep. Each host's lists are sorted and
@@ -108,6 +109,11 @@ func (a *Service) generationsSnapshotLocked() release.Generations {
for _, m := range e.ops.members {
add(host, m.build)
}
+ // With generations off, the entry keeps such a child itself
+ // (final review R2-1).
+ if e.shed != nil {
+ add(host, e.shed.build)
+ }
}
// A failed worker wake retained until its termination is proven: its
// entry is gone, but its process may still run the build (F12).
diff --git a/internal/activator/worker_respawn_exec_test.go b/internal/activator/worker_respawn_exec_test.go
index 0ea9bf295..f170a9f33 100644
--- a/internal/activator/worker_respawn_exec_test.go
+++ b/internal/activator/worker_respawn_exec_test.go
@@ -2,11 +2,13 @@ package activator
import (
"errors"
+ "slices"
"sync"
"testing"
"time"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
// Final review F1: a worker respawn asks its ownership again at the exec
@@ -103,3 +105,104 @@ func TestWorkerRespawnRechecksOwnershipAtTheExecAndTheInstall(t *testing.T) {
}
}
}
+
+// entryPendingStop reports whether host's entry is a stop held pending with
+// its lease kept.
+func (a *Service) entryPendingStop(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[host]
+ return e != nil && e.pendingStop
+}
+
+func (a *Service) entryPresent(host string) bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ return a.entries[host] != nil
+}
+
+// Final review R2-1: with generations off, a respawn child that exec'd and
+// was then refused at its installation (here: the lease failed after the
+// exec) stays the entry's to clean up until its exit is proven. Its build
+// stays pinned, and the terminal stop that the lost lease begins keeps the
+// lease and the entry until the child has exited, whether the shed Stop
+// failed to prove it or is still blocked when the stop's bounded wait on
+// the respawn runs out.
+func TestAShedRespawnChildIsRetainedUntilItsExitIsProven(t *testing.T) {
+ for _, blocked := range []bool{false, true} {
+ name := map[bool]string{false: "the shed Stop fails", true: "the shed Stop blocks past the stop's respawn wait"}[blocked]
+ t.Run(name, func(t *testing.T) {
+ if blocked {
+ old := respawnJoinBound
+ respawnJoinBound = 100 * time.Millisecond
+ t.Cleanup(func() { respawnJoinBound = old })
+ }
+ a, b, _, _, ver := stagedActivator(t)
+ a.gens = false
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ rt := alwaysOnWorkerRoute("bot10.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ old := b.lastWorker()
+ rep := b.lastRep()
+ ver.Store("v2") // the respawn (from the live link) execs v2
+ gate := make(chan struct{})
+ b.mu.Lock()
+ if blocked {
+ b.workerStopGate = gate
+ } else {
+ b.workerStuck = true
+ }
+ b.workerStartHook = func(registry.Route) { rep.setErr(errors.New("renewal failed")) }
+ b.mu.Unlock()
+ a.mu.Lock()
+ e := a.entries[rt.Host]
+ a.respawnWorkerLocked(e, nil, "test: shed after the exec", nil)
+ done := e.respawnDone
+ a.mu.Unlock()
+ var fresh *fakeProc
+ waitFor(t, "the respawn's exec", func() bool { fresh = b.lastWorker(); return fresh != old })
+ if blocked {
+ waitFor(t, "the shed Stop to block", func() bool { return fresh.stopEntered() > 0 })
+ } else {
+ <-done
+ }
+ if fresh.Exited() {
+ t.Fatal("fixture: the shed child should still be running")
+ }
+ if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") {
+ t.Errorf("generations file lists %v while the shed v2 child may still run; want v2 kept", got)
+ }
+ // The lost lease begins the terminal stop.
+ a.sweepOnce()
+ waitFor(t, "the stop to finish or be held", func() bool {
+ return rep.isClosed() || !a.entryPresent(rt.Host) || a.entryPendingStop(rt.Host)
+ })
+ if rep.isClosed() || !a.entryPresent(rt.Host) {
+ t.Fatalf("the stop released the lease (closed %v) and forgot the entry (present %v) while the shed child still runs", rep.isClosed(), a.entryPresent(rt.Host))
+ }
+ if !old.isStopped() {
+ t.Fatal("the stop did not stop the installed child")
+ }
+ if got := generationsHost(t, a, path, rt.Host); !slices.Contains(got, "v2") {
+ t.Fatalf("generations file lists %v during the held stop; want v2 kept", got)
+ }
+ // Proof: the shed child exits. The stop then finishes, and the
+ // build is unpinned.
+ if blocked {
+ close(gate)
+ } else {
+ fresh.exit()
+ }
+ waitFor(t, "the stop to finish once the shed child is proven gone", func() bool {
+ a.sweepOnce()
+ return rep.isClosed() && !a.entryPresent(rt.Host)
+ })
+ if got := generationsHost(t, a, path, rt.Host); slices.Contains(got, "v2") {
+ t.Fatalf("generations file lists %v after the shed child's exit; want v2 dropped", got)
+ }
+ })
+ }
+}