carlos / platform Public

zdd plan 3 final fix: a worker respawn asks ownership at its exec and before its install (F1, F5/T11 log lines)

startWorker takes a gate, the caller's ownership check at the exec: it
is composed into the before-exec hook after the build's publication, or
asked immediately before StartWorker on a backend without the staged
capability. The respawn's gate is its context plus ownedLocked, so a
lease lost, a fence set or the entry stopped during the config
admission, the materialization or the publication execs nothing. The
install check is ownedLocked again (it covers the old liveness check
and adds the lease and fences), so ownership lost after the exec sheds
the child. A gate refusal is never a member's failure event.

The respawn's log lines name the actual refusal (the ownedLocked error)
instead of "stopped mid-respawn", and the unproven-retirement line has
no em dash and no longer asserts that a process remains.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

claude 2 weeks ago fc14e7fd3c7c99789719aec4f833930cb51cdaa7 parent 967812f

4 files changed, +170 −7

internal/activator/activator.go +39 −6
internal/activator/activator_test.go +10 −0
internal/activator/worker_respawn_exec_test.go +105 −0
internal/activator/workerbuild.go +16 −1

diff --git a/internal/activator/activator.go b/internal/activator/activator.go

index 91d583394..590d3d96f 100644

--- a/internal/activator/activator.go

+++ b/internal/activator/activator.go

@@ -1612,7 +1612,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro

)

// startWorker publishes the build before its exec (ruling R19

// item 3), and refuses the spawn when it cannot.

- proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, token), nil)

+ proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, token), nil, nil)

spawnVer, w.spawnIdentity = ident.Version, ident

if err == nil {

// Installed while the entry is still waking, in place of the

@@ -2448,8 +2448,9 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

// and not shutting down, and adds the lease still held and no

// fence.

a.mu.Lock()

- owned := a.ownedLocked(e, false) == nil

+ ownErr := a.ownedLocked(e, false)

a.mu.Unlock()

+ owned := ownErr == nil

wanted := owned && retired

if wanted {

@@ -2457,13 +2458,27 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

// 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(spawnParent, workerSpawnTimeout)

+ // The ownership check again at the exec itself (final review

+ // F1): the config admission, the materialization and the

+ // build's publication all come between the check above and the

+ // exec, and the lease can be lost, a fence set or the entry

+ // stopped in that time. startWorker composes it into the

+ // before-exec hook, after the publication.

+ gate := func() error {

+ if err := ctx.Err(); err != nil {

+ return err

+ }

+ a.mu.Lock()

+ defer a.mu.Unlock()

+ return a.ownedLocked(e, false)

+ }

if g, ok := a.backend.(configAdmitter); ok {

refused = g.AdmitConfig(ctx, rt)

}

if refused == nil {

// No registry write (spec §7.9): see doWake.

if tok, err = newWorkerToken(); err == nil {

- proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok), onto)

+ proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok), onto, gate)

}

cancel()

@@ -2471,14 +2486,25 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

a.mu.Lock()

e.workerRespawning = false

- live := a.entries[host] == e && e.state == actAwake && !a.shuttingDown

+ // And once more before the child is installed: ownership lost after

+ // the exec sheds the child (final review F1). ownedLocked covers what

+ // the old liveness check did (the entry is the host's, awake, not

+ // shutting down) and adds the lease and the fences.

+ if owned {

+ ownErr = a.ownedLocked(e, false)

+ }

+ live := ownErr == nil

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)

+ 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)

case !wanted || !live:

a.mu.Unlock()

- log.Printf("activator: worker %s stopped mid-respawn (%s) — abandoning it", host, why)

+ reason := ownErr

+ if reason == nil {

+ reason = errors.New("the old generation's termination is not proven")

+ }

+ log.Printf("activator: worker %s NOT respawned (%s): %v; abandoning it", host, why, reason)

if proc != nil {

// Spawned into a window that closed under us: shed it rather

// than leave a child nothing owns. With generations on it is

@@ -2512,6 +2538,13 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

a.beginStopLocked(e)

a.mu.Unlock()

log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused)

+ case verdictOf(err) != gateProceed:

+ // The exec's own ownership check refused (startWorker's gate),

+ // and ownership came back before the install check (a fence

+ // lifted): nothing exec'd, and this is no failure of the build.

+ // The dead child stays, and the sweep's crash restart asks again.

+ a.mu.Unlock()

+ log.Printf("activator: worker %s NOT respawned (%s): refused at its exec: %v", host, why, err)

case err != nil:

// The captured build is kept whatever the failure, a refused

// recycle (release.ErrStagedCopyChanged) included: spec §4.2

diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go

index 42100f027..f396d12c8 100644

--- a/internal/activator/activator_test.go

+++ b/internal/activator/activator_test.go

@@ -263,6 +263,10 @@ type fakeBackend struct {

// admitErr is what AdmitConfig answers: the config gate a retained-lease

// respawn must ask again (amadan #213).

admitErr error

+ // admitHook, when set, runs at the start of AdmitConfig, off b.mu: a

+ // test's window inside a respawn's preparation, after its first

+ // ownership check and before its exec.

+ admitHook func()

// restoreFresh is what Restore reports: a first boot (local DB absent AND

// no replica). False by default, so every pre-seed test doubles as the

// "a restored instance is never seeded" case.

@@ -389,6 +393,12 @@ func (b *fakeBackend) Acquire(context.Context, registry.Route) (Lease, error) {

}

func (b *fakeBackend) AdmitConfig(context.Context, registry.Route) error {

+ b.mu.Lock()

+ hook := b.admitHook

+ b.mu.Unlock()

+ if hook != nil {

+ hook() // a respawn's preparation, held open by the test

+ }

b.mu.Lock()

defer b.mu.Unlock()

b.admits++

diff --git a/internal/activator/worker_respawn_exec_test.go b/internal/activator/worker_respawn_exec_test.go

new file mode 100644

index 000000000..0ea9bf295

--- /dev/null

+++ b/internal/activator/worker_respawn_exec_test.go

@@ -0,0 +1,105 @@

+package activator

+import (

+ "errors"

+ "sync"

+ "testing"

+ "time"

+ "github.com/carlosframework/platform/internal/registry"

+// Final review F1: a worker respawn asks its ownership again at the exec

+// (composed into the before-exec hook, after the build's publication, or

+// right before StartWorker on a backend without the staged capability),

+// and once more before it installs the child. A lease lost while the

+// respawn's preparation is under way (here, held open in the config

+// admission) execs nothing; one lost after the exec installs nothing, and

+// the child it made is shed. With generations on or off, on a backend with

+// or without the staged capability.

+func TestWorkerRespawnRechecksOwnershipAtTheExecAndTheInstall(t *testing.T) {

+ type rig struct {

+ a *Service

+ b *fakeBackend

+ staged bool

+ }

+ setup := func(t *testing.T, on, staged bool) rig {

+ a, b, _, _ := testActivator(t, time.Hour)

+ a.gens = on

+ if staged {

+ a.backend = &stagedFake{fakeBackend: b}

+ }

+ return rig{a, b, staged}

+ }

+ for _, on := range []bool{true, false} {

+ for _, staged := range []bool{true, false} {

+ name := map[bool]string{true: "gens on", false: "gens off"}[on] + map[bool]string{true: ", staged", false: ", plain"}[staged]

+ t.Run(name+": lost during the preparation", func(t *testing.T) {

+ r := setup(t, on, staged)

+ rt := testWorkerRoute("bot6.example.com")

+ if err := r.a.WakeWorker(rt); err != nil {

+ t.Fatal(err)

+ }

+ old := r.b.lastWorker()

+ rep := r.b.lastRep()

+ entered, release := make(chan struct{}), make(chan struct{})

+ var once sync.Once

+ r.b.mu.Lock()

+ r.b.admitHook = func() {

+ once.Do(func() { close(entered) })

+ <-release

+ }

+ r.b.mu.Unlock()

+ r.a.mu.Lock()

+ r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails mid-preparation", nil)

+ r.a.mu.Unlock()

+ <-entered // past the first ownership check, inside the preparation

+ rep.setErr(errors.New("renewal failed"))

+ close(release)

+ waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) })

+ if got := r.b.workerEnters(); got != 1 {

+ t.Fatalf("worker spawns begun = %d: the respawn exec'd after its lease was lost", got)

+ }

+ r.a.mu.Lock()

+ e := r.a.entries[rt.Host]

+ proc := e.proc

+ r.a.mu.Unlock()

+ if proc != Proc(old) {

+ t.Fatal("the respawn installed a child after its lease was lost")

+ }

+ if _, members := r.a.genStateForTest(rt.Host); len(members) != 0 {

+ t.Fatalf("members after the refused respawn = %v", members)

+ }

+ })

+ t.Run(name+": lost after the exec", func(t *testing.T) {

+ r := setup(t, on, staged)

+ rt := testWorkerRoute("bot7.example.com")

+ if err := r.a.WakeWorker(rt); err != nil {

+ t.Fatal(err)

+ }

+ old := r.b.lastWorker()

+ rep := r.b.lastRep()

+ r.b.mu.Lock()

+ r.b.workerStartHook = func(_ registry.Route) { rep.setErr(errors.New("renewal failed")) }

+ r.b.mu.Unlock()

+ r.a.mu.Lock()

+ r.a.respawnWorkerLocked(r.a.entries[rt.Host], nil, "test: the lease fails at the exec", nil)

+ r.a.mu.Unlock()

+ waitFor(t, "the respawn to finish", func() bool { return !r.a.entryRespawning(rt.Host) })

+ if got := r.b.workerStarts(); got != 2 {

+ t.Fatalf("worker starts = %d, want 2 (the wake and the respawn's exec)", got)

+ }

+ fresh := r.b.lastWorker()

+ r.a.mu.Lock()

+ proc := r.a.entries[rt.Host].proc

+ r.a.mu.Unlock()

+ if proc != Proc(old) || proc == Proc(fresh) {

+ t.Fatal("the respawn installed its child after its lease was lost")

+ }

+ if !fresh.isStopped() {

+ t.Fatal("the uninstalled child was left running")

+ }

+ })

+ }

diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go

index 4af988ed6..c17dee2fa 100644

--- a/internal/activator/workerbuild.go

+++ b/internal/activator/workerbuild.go

@@ -67,9 +67,21 @@ func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {

// dropped it; on success the caller drops it (dropPendingGenerationLocked)

// under a.mu in the same critical section that installs the captured

// identity, or after shedding the child, and then publishes again.

-func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, pendingGeneration, error) {

+//

+// gate, when set, is the caller's ownership check at the exec (final review

+// F1): composed into the before-exec hook after the publication, so it is

+// asked once the preparation (materialization, the publication) is done

+// and immediately before the exec, or, on a backend without the staged

+// capability, immediately before StartWorker. Its refusal execs nothing

+// and is returned as is.

+func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity, gate func() error) (Proc, release.StagedIdentity, pendingGeneration, error) {

s, ok := a.backend.(stagedWorkerStarter)

if !ok {

+ if gate != nil {

+ if err := gate(); err != nil {

+ return nil, release.StagedIdentity{}, 0, err

+ }

p, ver, err := a.backend.StartWorker(ctx, rt, env)

return p, release.StagedIdentity{Version: ver}, 0, err

}

@@ -85,6 +97,9 @@ func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[st

log.Printf("activator: worker %s not spawned: the build it would exec (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err)

return fmt.Errorf("%w: %v", ErrBuildNotPublished, err)

}

+ if gate != nil {

+ return gate()

+ }

return nil

}

p, id, err := s.StartWorkerStaged(withBeforeExec(ctx, hook), rt, env, onto)