amadan

carlos / platform Public

zdd plan 3 final fix: retained worker processes keep their build pinned until proven gone (F12, T11 shed-path test)

A failed worker wake retained in unprovenWakes carries its staged
identity (unprovenWake.build, from w.spawnIdentity), and a shed respawn
child carries its build on its operation-set member (genMember.build),
set before the pending record is dropped. The one serialized
generations writer (generationsSnapshotLocked) lists both while they
are retained, and reapUnprovenWake and leaveSet publish again once the
termination proof holds and the owner is gone.

Tests: a retained failed worker wake keeps v1 listed until its host
cgroup is proven empty; a retained shed respawn (generations on, lease
lost after the exec, its host cgroup unproven) keeps v2 listed beside
the serving v1 until the reaper proves it.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 2255d11a10e834d7c02259c2bb52e3135d4a318a parent 1f5de20
5 files changed, +187 −3
  • internal/activator/activator.go +6 −1
  • internal/activator/generations_file.go +13 −1
  • internal/activator/maintenance.go +4 −1
  • internal/activator/opset.go +18 −0
  • internal/activator/worker_retained_pin_test.go +146 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 8125f6d4e..9b1d362a2 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -2511,8 +2511,13 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// the member's process: dropMember stops it and proves its
// host cgroup empty, or keeps it in the set for the reaper.
if m != nil {
+ // The child's build moves to the member before its
+ // pending record is dropped below: if dropMember cannot
+ // prove the child gone, the member is retained, and the
+ // generations file keeps naming its build until the
+ // reaper proves it (final review F12).
a.mu.Lock()
- m.proc = proc
+ m.proc, m.build = proc, ident
a.mu.Unlock()
shed = true
a.dropMember(e, m)
diff --git a/internal/activator/generations_file.go b/internal/activator/generations_file.go
index eb5afd56e..b2166363e 100644
--- a/internal/activator/generations_file.go
+++ b/internal/activator/generations_file.go
@@ -69,7 +69,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 every pending build, each
+// 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
// 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
@@ -101,6 +103,16 @@ func (a *Service) generationsSnapshotLocked() release.Generations {
if e.worker {
add(host, e.workerServing)
}
+ // A shed respawn child retained in the operation set until its
+ // termination is proven (final review F12).
+ for _, m := range e.ops.members {
+ add(host, m.build)
+ }
+ }
+ // A failed worker wake retained until its termination is proven: its
+ // entry is gone, but its process may still run the build (F12).
+ for host, u := range a.unprovenWakes {
+ add(host, u.build)
}
for _, pb := range a.genPending {
add(pb.host, pb.id)
diff --git a/internal/activator/maintenance.go b/internal/activator/maintenance.go
index 949a0b415..3b6eb2e4a 100644
--- a/internal/activator/maintenance.go
+++ b/internal/activator/maintenance.go
@@ -288,7 +288,10 @@ func (a *Service) cleanupFailedWake(ctx context.Context, rt registry.Route, w *w
if a.unprovenWakes == nil {
a.unprovenWakes = make(map[string]*unprovenWake)
}
- a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, proc: w.proc, lease: lease, rep: w.rep}
+ // A worker's staged identity moves with it (final review F12): the
+ // failed wake's entry is about to go, and the generations file must
+ // keep naming the build its retained process may still run.
+ a.unprovenWakes[rt.Host] = &unprovenWake{rt: rt, gen: w.gen, proc: w.proc, lease: lease, rep: w.rep, build: w.spawnIdentity}
a.mu.Unlock()
return
}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index f1944be8b..6dfe5d00b 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -15,6 +15,7 @@ import (
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
)
type genRole int
@@ -52,6 +53,11 @@ type genMember struct {
unproven bool
reaping bool
retriedAt time.Time
+ // build is the worker build a shed respawn child exec'd (final review
+ // F12). The generations file lists it while the member is in the set,
+ // since a retained child may still run those bytes; leaveSet publishes
+ // again once it has left.
+ build release.StagedIdentity
}
// straySidecar is a legacy sidecar whose termination is unproven, with the
@@ -310,6 +316,10 @@ func (a *Service) leaveSet(e *actEntry, m *genMember) {
}
a.mu.Unlock()
close(m.done)
+ if m.build != (release.StagedIdentity{}) {
+ // Its child is proven gone: its build is no longer referenced.
+ _ = a.persistGenerations()
+ }
}
// restoreParent puts rt's parent back at tier, and reports whether it did.
@@ -487,6 +497,10 @@ type unprovenWake struct {
lease Lease
rep Rep
busy bool
+ // build is a failed worker wake's staged identity (final review F12):
+ // the generations file keeps listing it until the proof holds, since
+ // the retained process may still run those bytes.
+ build release.StagedIdentity
}
// errWakeUnproven refuses a wake while the host's last failed wake is not
@@ -581,6 +595,10 @@ func (a *Service) reapUnprovenWake(host string, u *unprovenWake) {
}
u.busy = false
a.mu.Unlock()
+ if u.build != (release.StagedIdentity{}) {
+ // Proven gone: its build is no longer referenced.
+ _ = a.persistGenerations()
+ }
}
// RetiredControl reports whether control is the control socket of a
diff --git a/internal/activator/worker_retained_pin_test.go b/internal/activator/worker_retained_pin_test.go
new file mode 100644
index 000000000..426450e13
--- /dev/null
+++ b/internal/activator/worker_retained_pin_test.go
@@ -0,0 +1,146 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "slices"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// stagedHostKill is stagedFake plus KillHost, failing while bad: a staged
+// worker backend whose host-cgroup proof the test controls.
+type stagedHostKill struct {
+ *stagedFake
+ hmu sync.Mutex
+ bad bool
+ kills int
+}
+
+func (h *stagedHostKill) KillHost(context.Context, registry.Route) error {
+ h.hmu.Lock()
+ defer h.hmu.Unlock()
+ h.kills++
+ if h.bad {
+ return errors.New("the host cgroup still holds a process")
+ }
+ return nil
+}
+
+func (h *stagedHostKill) setBad(bad bool) {
+ h.hmu.Lock()
+ h.bad = bad
+ h.hmu.Unlock()
+}
+
+func (h *stagedHostKill) killCount() int {
+ h.hmu.Lock()
+ defer h.hmu.Unlock()
+ return h.kills
+}
+
+// generationsHost is the versions the generations file lists for host,
+// read after a publication of a's state as it is now (persistGenerations is
+// serialized, so nothing older can land after it). The caller makes sure no
+// other writer is under way.
+func generationsHost(t *testing.T, a *Service, path, host string) []string {
+ t.Helper()
+ if err := a.persistGenerations(); err != nil {
+ t.Fatalf("persistGenerations: %v", err)
+ }
+ g, err := release.ReadGenerations(path)
+ if err != nil {
+ t.Fatalf("ReadGenerations: %v", err)
+ }
+ if g == nil {
+ return nil
+ }
+ return g.Hosts[host]
+}
+
+// Final review F12: a failed worker wake that is retained, because its
+// termination is not proven, keeps its build in the generations file
+// (root's pin set) until the proof holds. The process may still run those
+// bytes, so root must not restage or sweep them.
+func TestARetainedFailedWorkerWakeKeepsItsBuildPinned(t *testing.T) {
+ old := sidecarSettle
+ sidecarSettle = 20 * time.Millisecond
+ t.Cleanup(func() { sidecarSettle = old })
+ a, b, _, _, _ := stagedActivator(t)
+ a.gens = true
+ hk := &stagedHostKill{stagedFake: b, bad: true}
+ a.backend = hk
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ host := "bot8.example.com"
+ b.mu.Lock()
+ b.workerDeadOnStart = true // the wake fails its settle
+ b.mu.Unlock()
+ if err := a.WakeWorker(testWorkerRoute(host)); err == nil {
+ t.Fatal("a worker dead on start woke")
+ }
+ a.mu.Lock()
+ _, retained := a.unprovenWakes[host]
+ a.mu.Unlock()
+ if !retained {
+ t.Fatal("the failed wake was not retained over its unproven host cgroup")
+ }
+ if got := generationsHost(t, a, path, host); !slices.Equal(got, []string{"v1"}) {
+ t.Fatalf("generations file lists %v for %s while its failed wake is retained; want [v1]", got, host)
+ }
+ hk.setBad(false)
+ a.sweepOnce()
+ waitFor(t, "the retained wake's proof and its unpin", func() bool {
+ g, err := release.ReadGenerations(path)
+ return err == nil && g != nil && len(g.Hosts[host]) == 0
+ })
+}
+
+// Final review F12, T11: a respawn whose child is shed (here: its lease
+// is lost after the exec) and cannot be proven gone keeps that child's
+// build pinned while the member is retained, and drops it once the reaper
+// proves the child gone. The serving build stays pinned throughout.
+func TestARetainedShedRespawnKeepsItsBuildPinned(t *testing.T) {
+ a, b, _, _, ver := stagedActivator(t)
+ a.gens = true
+ hk := &stagedHostKill{stagedFake: b}
+ a.backend = hk
+ path := release.GenerationsPath(t.TempDir())
+ a.GenerationsFile = path
+ rt := alwaysOnWorkerRoute("bot9.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ rep := b.lastRep()
+ ver.Store("v2") // the respawn (from the live link) execs v2
+ b.mu.Lock()
+ b.workerStartHook = func(registry.Route) {
+ // After the exec: the lease fails, and the host cgroup will not
+ // prove the shed child gone.
+ rep.setErr(errors.New("renewal failed"))
+ hk.setBad(true)
+ }
+ 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()
+ <-done // the respawn goroutine has finished, its shed and publication included
+ if _, members := a.genStateForTest(rt.Host); len(members) != 1 {
+ t.Fatalf("members = %v: want the shed child retained", members)
+ }
+ if got := generationsHost(t, a, path, rt.Host); !slices.Equal(got, []string{"v1", "v2"}) {
+ t.Fatalf("generations file lists %v while the shed v2 child is retained; want [v1 v2]", got)
+ }
+ hk.setBad(false)
+ a.sweepOnce()
+ waitFor(t, "the shed child's proof and its unpin", func() bool {
+ g, err := release.ReadGenerations(path)
+ return err == nil && g != nil && !slices.Contains(g.Hosts[rt.Host], "v2")
+ })
+}