amadan

carlos / platform Public

zdd plan 3 task 11: worker children are numbered generations; a respawn joins the operation set

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude e757c085363979e582c9ee43f6d9f09240fb4a37 parent f089cff
4 files changed, +363 −7
  • internal/activator/activator.go +77 −7
  • internal/activator/genbackend.go +36 −0
  • internal/activator/opset.go +22 −0
  • internal/activator/worker_gen_test.go +228 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 39a404c18..f009447a3 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -1254,6 +1254,13 @@ func (a *Service) wake(e *actEntry) {
if e.gen.N == 0 && !e.worker {
e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
}
+ if a.gens && e.worker && e.gen.N == 0 {
+ // A worker woken some way that carried no number (doWake gives
+ // every worker spawn its own): numbered here, with the build it
+ // runs, so the barrier and the generations file see it. Worker:
+ // its termination is proven on its host cgroup (Task 11).
+ e.gen = Gen{N: nextGen(), Build: w.spawnVersion, Worker: true}
+ }
if e.gen.N != 0 {
a.noteCurrent(e.host, e.gen.N)
}
@@ -1592,6 +1599,14 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro
a.mu.Unlock()
_ = a.persistGenerations()
}
+ if a.gens {
+ // The wake's generation, whether or not the worker becomes
+ // ready (zero-downtime spec §7.4): a worker binds nothing, but
+ // it is numbered and carries the build it execed, and
+ // cleanupFailedWake proves its host cgroup before the lease
+ // goes back (Worker, genChildEmpty).
+ w.gen = Gen{N: nextGen(), Build: spawnVer, Worker: true}
+ }
} else if e.pin != nil { // swapsim:M3
var g Gen
proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin)
@@ -2333,6 +2348,17 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.workerRespawning = true
e.respawnDone = make(chan struct{})
done, rt, host := e.respawnDone, e.route, e.host
+ // With generations on, the respawn is a member of the operation set
+ // until it has finished with every child it made: a stop's barrier joins
+ // it, and cancelling it tears down the exec (zero-downtime plan 3,
+ // Task 11). Off, m stays nil and spawnParent is just lifeCtx's child.
+ spawnParent, cancelSpawn := context.WithCancel(a.lifeCtx)
+ var m *genMember
+ oldGen := e.gen // the retiring child's generation, for its proof
+ if a.gens {
+ m = &genMember{gen: Gen{N: nextGen(), Worker: true}, role: roleSpawning, cancel: cancelSpawn, done: make(chan struct{})}
+ a.addMemberLocked(e, m)
+ }
go func() {
// respawnDone closes LAST, once this goroutine is finished with every
// child it created — including one it had to SHED. That ordering is
@@ -2343,6 +2369,15 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
// shed (as this first did) only narrowed that race; it did not close
// it.
defer close(done)
+ defer cancelSpawn()
+ shed := false // a shed child went to dropMember, which leaves the set itself
+ if m != nil {
+ defer func() { // runs before close(done): defers are LIFO
+ if !shed {
+ a.leaveSet(e, m)
+ }
+ }()
+ }
var (
proc Proc
ident release.StagedIdentity
@@ -2364,17 +2399,36 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
cancel()
}
+ // The retired child must be proven gone, its host cgroup included,
+ // before anything new execs beside its descendants (zero-downtime
+ // spec §7.4, §7.9). Unproven: nothing spawns; the entry keeps its
+ // lease, and the sweep's crash path retires it again through
+ // genTerminated. stopOld may be nil (a crash restart, a restore):
+ // the old child is gone, but what it forked lives in the host
+ // cgroup, and genTerminated proves a Worker generation there with
+ // or without a handle.
+ retired := true
+ if a.gens {
+ oldGen.Worker = true
+ retired = a.genTerminated(rt, oldGen, stopOld)
+ }
+
// Still wanted? Asked HERE, before anything is spawned: past this
- // point a stop can only be raced, never prevented.
+ // point a stop can only be raced, never prevented. The gate's
+ // ownership check (spec §7.4): every exec asks it, whatever the
+ // switch says. It covers the entry still being the host's, awake
+ // and not shutting down, and adds the lease still held and no
+ // fence.
a.mu.Lock()
- wanted := a.entries[host] == e && e.state == actAwake && !a.shuttingDown
+ owned := a.ownedLocked(e, false) == nil
a.mu.Unlock()
+ wanted := owned && retired
if wanted {
// Off the service lifetime, so Shutdown cancels this exec instead
// of blocking on it. The timeout is the slow-materialize bound;
// the parent is the "are we still here at all" one.
- ctx, cancel := context.WithTimeout(a.lifeCtx, workerSpawnTimeout)
+ ctx, cancel := context.WithTimeout(spawnParent, workerSpawnTimeout)
if g, ok := a.backend.(configAdmitter); ok {
refused = g.AdmitConfig(ctx, rt)
}
@@ -2391,15 +2445,28 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.workerRespawning = false
live := a.entries[host] == e && e.state == actAwake && !a.shuttingDown
switch {
+ case owned && !retired && live:
+ a.mu.Unlock()
+ log.Printf("activator: worker %s NOT respawned (%s): the old generation is not proven terminated (its host cgroup still holds a process), so nothing execs beside it — the sweep retries", host, why)
case !wanted || !live:
a.mu.Unlock()
log.Printf("activator: worker %s stopped mid-respawn (%s) — abandoning it", host, why)
if proc != nil {
// Spawned into a window that closed under us: shed it rather
- // than leave a child nothing owns.
- shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
- _ = proc.Stop(shedCtx)
- shedCancel()
+ // than leave a child nothing owns. With generations on it is
+ // 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 {
+ a.mu.Lock()
+ m.proc = proc
+ a.mu.Unlock()
+ shed = true
+ a.dropMember(e, m)
+ } else {
+ shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
+ _ = proc.Stop(shedCtx)
+ shedCancel()
+ }
}
if pend != 0 {
// Its published build stays listed until the child is shed.
@@ -2433,6 +2500,9 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont
e.spawnVersion = ident.Version
e.workerServing = ident
e.workerVersion = "" // the new child has not reported yet
+ if m != nil {
+ e.gen = Gen{N: m.gen.N, Build: ident.Version, Worker: true}
+ }
// The captured build replaces the one published before the exec.
a.dropPendingGenerationLocked(pend)
// Only an exec from the live link starts a fresh drift clock. A
diff --git a/internal/activator/genbackend.go b/internal/activator/genbackend.go
index 96d4a55a3..4c2c6c237 100644
--- a/internal/activator/genbackend.go
+++ b/internal/activator/genbackend.go
@@ -358,3 +358,39 @@ func (b *SplitBackend) GenPaths(rt registry.Route, n uint64) (Gen, bool) {
}
return Gen{}, false
}
+
+// KillHost kills whatever is left in rt's own host cgroup and returns once
+// it is empty: a worker's termination proof (zero-downtime spec §7.4, §7.9).
+// It does not look at the row's layout: StartWorker spawns every worker
+// through spawnInCgroup into carlos-exec@<host>, whatever the layout says,
+// so a nested-looking worker is proven there too. An absent cgroup is
+// empty. No cgroup delegation at all is an unproven kill.
+func (b *ExecBackend) KillHost(ctx context.Context, rt registry.Route) error {
+ if b.Cgroups == nil {
+ return cgroup.ErrNoDelegation
+ }
+ c, err := b.Cgroups.OpenCleanup(rt.Host)
+ if errors.Is(err, cgroup.ErrNoCgroup) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ if err := c.KillAndEmpty(ctx); err != nil {
+ _ = c.Close()
+ return err
+ }
+ return c.Close()
+}
+
+// KillHost dispatches like Start: a unit-backed row's one process tree is
+// systemd's, so there is no host cgroup of ours to kill.
+func (b *SplitBackend) KillHost(ctx context.Context, rt registry.Route) error {
+ if rt.Backing == registry.BackingUnit {
+ return nil // systemd owns the unit tenant's tree
+ }
+ if k, ok := b.Exec.(hostKiller); ok {
+ return k.KillHost(ctx, rt)
+ }
+ return nil
+}
diff --git a/internal/activator/opset.go b/internal/activator/opset.go
index 7eda235a5..3688b2ac5 100644
--- a/internal/activator/opset.go
+++ b/internal/activator/opset.go
@@ -127,6 +127,13 @@ type genKiller interface {
KillGen(ctx context.Context, rt registry.Route, n uint64) error
}
+// hostKiller is the exec backend's host-cgroup kill: cgroup.kill on host's
+// own cgroup, then wait until it is empty. It proves a worker generation
+// terminated (plan 2 D4 keeps workers there).
+type hostKiller interface {
+ KillHost(ctx context.Context, rt registry.Route) error
+}
+
// stopMember stops m's process, bounded by actStopTimeout: SIGTERM, the
// generation's own graceful drain, then SIGKILL (spec §7.5 step 6). It
// reports whether termination is proven, which takes two things: the
@@ -222,6 +229,21 @@ func (a *Service) genChildEmpty(rt registry.Route, m *genMember) bool {
defer cancel()
return cp.provenStopped(ctx)
}
+ if m.gen.Worker {
+ // A worker runs in its host cgroup (plan 2 D4): that cgroup, not a
+ // g<n> it never ran in, is what must be empty (spec §7.4, §7.9).
+ hk, ok := a.backend.(hostKiller)
+ if !ok {
+ return true // a backend with no host cgroup: the leader's exit is all there is
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
+ defer cancel()
+ if err := hk.KillHost(ctx, rt); err != nil {
+ log.Printf("activator: %s: worker generation %d's host cgroup is not proven empty: %v", rt.Host, m.gen.N, err)
+ return false
+ }
+ return true
+ }
k, ok := a.backend.(genKiller)
if !ok || m.gen.N == 0 {
return true
diff --git a/internal/activator/worker_gen_test.go b/internal/activator/worker_gen_test.go
new file mode 100644
index 000000000..4cb4b9401
--- /dev/null
+++ b/internal/activator/worker_gen_test.go
@@ -0,0 +1,228 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "path/filepath"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+func TestWorkerGenerations(t *testing.T) {
+ for _, on := range []bool{true, false} {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = on
+ old := sidecarSettle
+ sidecarSettle = 20 * time.Millisecond
+ t.Cleanup(func() { sidecarSettle = old })
+ rt := testWorkerRoute("bot.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ first, members := a.genStateForTest(rt.Host)
+ if on != (first.N != 0) || first.Socket != "" || len(members) != 0 {
+ t.Fatalf("gens %v: worker generation %+v, members %v", on, first, members)
+ }
+ b.mu.Lock()
+ b.workerStartDelay = 200 * time.Millisecond // hold the respawn open
+ b.mu.Unlock()
+ b.lastWorker().crash()
+ a.sweepOnce() // a crashed hibernating worker respawns in place
+ waitFor(t, "the respawn to begin", func() bool { return b.workerEnters() == 2 })
+ _, members = a.genStateForTest(rt.Host)
+ if on != (len(members) == 1) {
+ t.Fatalf("gens %v: members during the respawn = %v", on, members)
+ }
+ waitFor(t, "the respawn to finish", func() bool { return b.workerStarts() == 2 })
+ waitFor(t, "the member to leave", func() bool { _, m := a.genStateForTest(rt.Host); return len(m) == 0 })
+ if on {
+ if second, _ := a.genStateForTest(rt.Host); second.N <= first.N || !second.Worker {
+ t.Fatalf("respawned worker generation %+v, want a Worker above %d", second, first.N)
+ }
+ }
+ }
+}
+
+// StartWorker spawns every worker into the flat host cgroup, whatever the
+// row's layout, so KillHost proves that cgroup for a nested-looking row too.
+// Here the cgroup is a fixture, not cgroup v2, so the proof is refused.
+// Answering nil for a nested layout skipped the proof for a worker that
+// runs in carlos-exec@<host> all the same.
+func TestKillHostProvesANestedLayoutWorkersHostCgroup(t *testing.T) {
+ host := "nw.example.com"
+ b, _ := delegatedBackend(t, host)
+ rd := t.TempDir()
+ b.RunDirOverride = rd
+ rt := registry.Route{Host: host, Kind: "sidecar", DBPath: provision.NestedDBPath(b.DataDir, host),
+ Socket: filepath.Join(rd, "instances", provision.InstanceID(host), "app.sock")}
+ if err := b.KillHost(context.Background(), rt); err == nil {
+ t.Fatal("KillHost answered a nested-layout worker's host cgroup as proven without proving it")
+ }
+}
+
+// hostKillBackend is fakeBackend plus KillHost, failing while bad.
+type hostKillBackend struct {
+ *fakeBackend
+ hmu sync.Mutex
+ bad bool
+ kills int
+}
+
+func (h *hostKillBackend) 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 *hostKillBackend) heal() {
+ h.hmu.Lock()
+ h.bad = false
+ h.hmu.Unlock()
+}
+
+// Spec §7.4, §7.9: a worker runs in its host cgroup (plan 2 D4), so that is
+// what proves it terminated, not a g<n> it never ran in (whose absence
+// KillGen counts as empty). While the host cgroup still holds a process
+// (a descendant the worker forked), a terminal stop keeps the lease and a
+// respawn execs nothing beside it.
+func TestWorkerTerminationProvesItsHostCgroup(t *testing.T) {
+ t.Run("a terminal stop keeps the lease", func(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = true
+ hk := &hostKillBackend{fakeBackend: b, bad: true}
+ a.backend = hk
+ rt := testWorkerRoute("bot.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ rep := b.lastRep()
+ a.Bounce(rt.Host)
+ waitFor(t, "the stop to pend", func() bool {
+ a.mu.Lock()
+ defer a.mu.Unlock()
+ e := a.entries[rt.Host]
+ return e != nil && e.pendingStop
+ })
+ time.Sleep(50 * time.Millisecond)
+ if rep.isClosed() {
+ t.Fatal("the lease was released while the worker's host cgroup still held a process")
+ }
+ hk.heal()
+ a.sweepOnce()
+ waitFor(t, "the lease release", rep.isClosed)
+ })
+ t.Run("a respawn with no old handle still proves the host cgroup", func(t *testing.T) {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = true
+ hk := &hostKillBackend{fakeBackend: b}
+ a.backend = hk
+ rt := testWorkerRoute("bot3.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ hk.hmu.Lock()
+ hk.bad = true
+ hk.hmu.Unlock()
+ a.mu.Lock()
+ // main's signature; once plan 4 Task 9 has merged, pass nil as its
+ // fourth (onto) argument.
+ a.respawnWorkerLocked(a.entries[rt.Host], nil, "test: a restart with no old handle", nil)
+ a.mu.Unlock()
+ waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
+ if got := b.workerStarts(); got != 1 {
+ t.Fatalf("worker starts = %d: a nil-stopOld respawn skipped the host cgroup proof", got)
+ }
+ })
+ t.Run("a failed worker wake keeps its lease", func(t *testing.T) {
+ old := sidecarSettle
+ sidecarSettle = 20 * time.Millisecond
+ t.Cleanup(func() { sidecarSettle = old })
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = true
+ hk := &hostKillBackend{fakeBackend: b, bad: true}
+ a.backend = hk
+ b.mu.Lock()
+ b.workerDeadOnStart = true // the wake fails its settle
+ b.mu.Unlock()
+ if err := a.WakeWorker(testWorkerRoute("bot4.example.com")); err == nil {
+ t.Fatal("a worker dead on start woke")
+ }
+ // The lease goes back by Release, or by closing the replica when the
+ // wake had one (cleanupFailedWake).
+ given := func() bool {
+ b.mu.Lock()
+ released, reps := b.releases, append([]*fakeRep(nil), b.reps...)
+ b.mu.Unlock()
+ if released > 0 {
+ return true
+ }
+ for _, r := range reps {
+ if r.isClosed() {
+ return true
+ }
+ }
+ return false
+ }
+ if given() {
+ t.Fatal("a failed worker wake gave its lease back over an unproven host cgroup")
+ }
+ hk.heal()
+ a.sweepOnce()
+ waitFor(t, "the lease given back", given)
+ })
+ t.Run("a respawn execs nothing beside the old child", func(t *testing.T) {
+ old := sidecarSettle
+ sidecarSettle = 20 * time.Millisecond
+ t.Cleanup(func() { sidecarSettle = old })
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = true
+ hk := &hostKillBackend{fakeBackend: b}
+ a.backend = hk
+ rt := testWorkerRoute("bot2.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ hk.hmu.Lock()
+ hk.bad = true
+ hk.hmu.Unlock()
+ b.lastWorker().crash()
+ a.sweepOnce() // the respawn retires the crashed child, and its proof fails
+ waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
+ if got := b.workerStarts(); got != 1 {
+ t.Fatalf("worker starts = %d: a respawn exec'd beside an unproven old child", got)
+ }
+ })
+}
+
+// Index P5/P6: every exec asks the gate's ownership check, whatever the
+// switch says. A respawn whose lease has been lost execs nothing, with
+// generations on or off.
+func TestWorkerRespawnAsksOwnershipAtItsExec(t *testing.T) {
+ for _, on := range []bool{true, false} {
+ a, b, _, _ := testActivator(t, time.Hour)
+ a.gens = on
+ rt := testWorkerRoute("bot5.example.com")
+ if err := a.WakeWorker(rt); err != nil {
+ t.Fatal(err)
+ }
+ b.lastRep().setErr(errors.New("renewal failed"))
+ a.mu.Lock()
+ a.respawnWorkerLocked(a.entries[rt.Host], nil, "test: the lease is lost", nil)
+ a.mu.Unlock()
+ waitFor(t, "the respawn to finish", func() bool { return !a.entryRespawning(rt.Host) })
+ if got := b.workerStarts(); got != 1 {
+ t.Fatalf("gens %v: worker starts = %d: a respawn exec'd without the lease", on, got)
+ }
+ if _, members := a.genStateForTest(rt.Host); len(members) != 0 {
+ t.Fatalf("gens %v: members after the refused respawn = %v", on, members)
+ }
+ }
+}