carlos / platform Public

zdd-4 third fix wave: a worker's build is published by version and path before its exec, or it is not spawned (F1(b)(c)(d), R19 items 2, 3)

startWorker passes a before-exec hook on the spawn's context
(withBeforeExec); ExecBackend.StartWorkerStaged runs it after resolving
the identity (ResolveLive or onto) and before spawnInCgroup. The hook
records the build as a pending generation and publishes the file; a
failed publication returns ErrBuildNotPublished and nothing is exec'd,
so the existing failed-spawn handling applies. The captured identity
replaces the pending one under a.mu when the spawn installs; a failed
spawn or a shed child drops it. StartWorkerStaged's signature and the
stagedWorkerStarter interface are unchanged; the private startWorker
also returns the pending record.

The snapshot lists every serving and pending build under its host by
version (when valid) and by absolute staged path (when present), sorted
and once each, so an identity with an empty Version (a chained live
link) is published by path.

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

claude 2 weeks ago 85711f5ed5944c072ae4f4f8c6207b8f63f63793 parent c5ff366

6 files changed, +462 −37

internal/activator/activator.go +30 −7
internal/activator/backend_exec.go +11 −0
internal/activator/generations.go +102 −23
internal/activator/worker_generations_test.go +170 −0
internal/activator/worker_staged_test.go +77 −0
internal/activator/workerbuild.go +72 −7

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

index 360075633..6d6d5e699 100644

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

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

@@ -715,6 +715,11 @@ type Service struct {

genMu sync.Mutex

genWritten bool

genDirty bool

+ // genPending holds the builds spawns are about to exec, published

+ // before each exec (generations.go, ruling R19 item 3); genPendingSeq

+ // names them. Both under a.mu.

+ genPending map[pendingGeneration]pendingBuild

+ genPendingSeq uint64

}

// New wires the state machine to its side-effect backend and due-time store.

@@ -1310,16 +1315,24 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro

// The generations file exists before any worker runs (ruling R17).

a.publishGenerationsBeforeSpawn()

- var ident release.StagedIdentity

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

+ var (

+ ident release.StagedIdentity

+ pend pendingGeneration

+ )

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

spawnVer, w.spawnIdentity = ident.Version, ident

if err == nil {

- // Published while the entry is still waking: the child is

- // running these bytes from this moment, settle window or not,

- // so root must already count them as referenced. The awake

- // transition stores the same identity again.

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

+ // pending build, in one critical section, so no snapshot sees

+ // neither: the child is running these bytes from this moment,

+ // settle window or not, so root must keep counting them as

+ // referenced. The awake transition stores the same identity

+ // again.

a.mu.Lock()

e.workerServing = ident

+ a.dropPendingGenerationLocked(pend)

a.mu.Unlock()

_ = a.persistGenerations()

}

@@ -1939,6 +1952,7 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

var (

proc Proc

ident release.StagedIdentity

+ pend pendingGeneration // the build published before the exec

tok string

err error

// refused is the config gate's answer for this respawn (amadan

@@ -1973,7 +1987,7 @@ func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, ont

if refused == nil {

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

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

- proc, ident, 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)

}

cancel()

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

_ = proc.Stop(shedCtx)

shedCancel()

}

+ if pend != 0 {

+ // Its published build stays listed until the child is shed.

+ a.mu.Lock()

+ a.dropPendingGenerationLocked(pend)

+ a.mu.Unlock()

+ _ = a.persistGenerations()

+ }

case refused != nil:

// Stand down: the old child is already gone, and nothing may

// start in its place. A normal stop checkpoints and releases the

@@ -2018,6 +2039,8 @@ 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

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

// recycle onto the same captured build serves exactly what it

// served before, so a drift it was serving keeps its age, and a

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

index 199d59d79..cd746240b 100644

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

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

@@ -1744,6 +1744,10 @@ func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra

// The identity returned is the build exec'd, the agent's serving identity for

// this child. What the child later reports over the agent socket is display

// metadata only.

+//

+// Between resolving that identity and the exec it runs ctx's before-exec hook

+// (runBeforeExec), through which the activator publishes the build first; a

+// hook error means nothing runs (ruling R19 item 3).

func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, extra map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {

var argv []string

if err := json.Unmarshal([]byte(rt.Args), &argv); err != nil || len(argv) == 0 {

@@ -1774,6 +1778,13 @@ func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route,

fileEnv = withVar(fileEnv, k, v)

}

+ // The identity is resolved and nothing has run: the activator publishes

+ // the build now, before the exec, and a failure means no child (ruling

+ // R19 item 3).

+ if err := runBeforeExec(ctx, id); err != nil {

+ return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err)

+ }

// Through spawnInCgroup, not Cgroups.Start directly: that seam is what

// applies the route's resource tier, and a worker is a long-lived child

// of a tenant's app like any other. Bypassing it compiles and silently

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

index 090048d0e..eb5afd56e 100644

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

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

@@ -2,45 +2,123 @@ package activator

import (

"log"

+ "path/filepath"

+ "sort"

"github.com/carlosframework/platform/internal/release"

)

-// Worker pin publication (zero-downtime deploys spec §4.2, §8.5; ruling R17).

+// Worker pin publication (zero-downtime deploys spec §4.2, §8.5; rulings

+// R17, R19).

// A worker recycles onto the build it was exec'd from, its captured serving

// identity, long after live/<host> and the registry row have moved on. Root's

-// staging must never overwrite that staged copy, and root is a separate

-// process that cannot see this one's memory, so the activator publishes it:

-// the generations file (release.WriteGenerations, read by root through

-// release.GenerationPins) lists, for every worker entry it holds with a

-// captured serving build, that build's version under the worker's host. It

-// does so whatever CARLOS_SWAP says; this flag-independent writer is the one

-// plan 5 extends.

+// staging and root's version sweep must never overwrite or remove that staged

+// copy, and root is a separate process that cannot see this one's memory, so

+// the activator publishes it: the generations file (release.WriteGenerations,

+// read by root through release.GenerationPins and release.GenerationVersions)

+// lists, for every worker entry it holds with a captured serving build and

+// for every build a spawn is about to exec, that build under the worker's

+// host, by version (when it is a valid version name) and by its staged path

+// (when it has one). It does so whatever CARLOS_SWAP says; this

+// flag-independent writer is the one plan 5 extends.

// The file is written before the first worker spawn (so it exists, even

-// empty, before any worker runs), after every worker spawn or respawn

-// captures an identity, and after every worker stop removes one. A write

-// that fails is logged and marks the publication dirty, and the sweep

-// retries it. It never blocks a spawn.

+// empty, before any worker runs), before every staged spawn's exec, naming

+// the build about to run (startWorker's before-exec hook, ruling R19 item 3),

+// after every spawn or respawn installs its captured identity in place of

+// that pending one, and after every worker stop removes one. A write that

+// fails is logged and marks the publication dirty, and the sweep retries it.

+// A failed write before an exec refuses that spawn; elsewhere it blocks

+// nothing, because root reads a publication that did not finish as an

+// incomplete pin set (release.ReadGenerations) and then neither restages nor

+// sweeps.

+// pendingGeneration names one build a spawn is about to exec, recorded by

+// startWorker's before-exec hook until the spawn installs or fails. 0 is

+// none.

+type pendingGeneration uint64

+type pendingBuild struct {

+ host string

+ id release.StagedIdentity

+// addPendingGeneration records id as a build host's spawn is about to exec,

+// and returns its name. Never call it with a.mu held.

+func (a *Service) addPendingGeneration(host string, id release.StagedIdentity) pendingGeneration {

+ a.mu.Lock()

+ defer a.mu.Unlock()

+ if a.genPending == nil {

+ a.genPending = map[pendingGeneration]pendingBuild{}

+ }

+ a.genPendingSeq++

+ k := pendingGeneration(a.genPendingSeq)

+ a.genPending[k] = pendingBuild{host: host, id: id}

+ return k

+// dropPendingGenerationLocked forgets a pending build: its spawn failed, its

+// child was shed, or its captured identity has been installed on the entry.

+// 0 is a no-op. The caller holds a.mu.

+func (a *Service) dropPendingGenerationLocked(k pendingGeneration) {

+ if k != 0 {

+ delete(a.genPending, k)

+ }

// 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) whose captured serving build has a version in the version

-// store, that version. An identity with no version, or one that is not a

-// valid version name, came from a link outside the version store, which no

-// pass restages, so there is nothing to protect. The caller holds a.mu.

+// may still run) its captured serving build, 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

+// hold each entry once. The caller holds a.mu.

func (a *Service) generationsSnapshotLocked() release.Generations {

- hosts := map[string][]string{}

+ hosts, paths := map[string]map[string]bool{}, map[string]map[string]bool{}

+ add := func(host string, id release.StagedIdentity) {

+ if v := id.Version; v != "" && release.ValidVersion(v) {

+ if hosts[host] == nil {

+ hosts[host] = map[string]bool{}

+ }

+ hosts[host][v] = true

+ }

+ if p := id.StagedPath; p != "" {

+ // A relative path is published as the exec resolved it, from

+ // this process's working directory; if even that fails, the

+ // relative path stays and WriteGenerations refuses it, so the

+ // spawn does not happen.

+ if abs, err := filepath.Abs(p); err == nil {

+ p = abs

+ }

+ if paths[host] == nil {

+ paths[host] = map[string]bool{}

+ }

+ paths[host][p] = true

+ }

for host, e := range a.entries {

- if !e.worker {

- continue

+ if e.worker {

+ add(host, e.workerServing)

}

- if v := e.workerServing.Version; v != "" && release.ValidVersion(v) {

- hosts[host] = []string{v}

+ }

+ for _, pb := range a.genPending {

+ add(pb.host, pb.id)

+ }

+ return release.Generations{Hosts: sortedLists(hosts), Paths: sortedLists(paths)}

+func sortedLists(m map[string]map[string]bool) map[string][]string {

+ out := make(map[string][]string, len(m))

+ for k, set := range m {

+ l := make([]string, 0, len(set))

+ for v := range set {

+ l = append(l, v)

}

+ sort.Strings(l)

+ out[k] = l

}

- return release.Generations{Hosts: hosts}

+ return out

}

// persistGenerations writes the generations file from a snapshot of the

@@ -75,7 +153,8 @@ func (a *Service) persistGenerationsGenLocked() error {

// publishGenerationsBeforeSpawn makes sure the generations file has been

// written once before a worker is spawned. A failure is logged and left to

-// the sweep; the spawn goes ahead. Never call it with a.mu held.

+// the sweep; this write does not decide the spawn, the staged spawn's own

+// before-exec publication does (startWorker). Never call it with a.mu held.

func (a *Service) publishGenerationsBeforeSpawn() {

if a.GenerationsFile == "" {

return

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

index 131258718..e276c8355 100644

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

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

@@ -3,6 +3,8 @@ package activator

import (

"bytes"

"context"

+ "errors"

+ "fmt"

"os"

"path/filepath"

"reflect"

@@ -372,3 +374,171 @@ func TestWorkerCapturedBuildSurvivesARestageOfItsReusedName(t *testing.T) {

t.Fatal("the worker recycled onto A is not running")

}

+// readGenerationsFile is the whole generations file; it fails the test when

+// the file does not read.

+func readGenerationsFile(t *testing.T, path string) *release.Generations {

+ t.Helper()

+ g, err := release.ReadGenerations(path)

+ if err != nil {

+ t.Fatalf("ReadGenerations: %v", err)

+ }

+ if g == nil {

+ return &release.Generations{}

+ }

+ return g

+func describeGenerations(gs []*release.Generations) string {

+ var b strings.Builder

+ for i, g := range gs {

+ fmt.Fprintf(&b, "[%d] hosts %v paths %v ", i, g.Hosts, g.Paths)

+ }

+ return b.String()

+// Ruling R19 item 3 (Astra r3 F1(c)): the build a spawn is about to exec is

+// in the generations file BEFORE the exec, for a wake (from the live link)

+// and for a converge onto a newly installed build. Once the spawn has

+// installed, the captured build replaces the pending one.

+func TestWorkerSpawnPublishesItsBuildBeforeTheExec(t *testing.T) {

+ a, b, clk, _, ver := stagedActivator(t)

+ b.runBeforeExec = true

+ path := release.GenerationsPath(t.TempDir())

+ a.GenerationsFile = path

+ host := "bot.x.internal"

+ var atExec []*release.Generations

+ b.workerStartHook = func(registry.Route) {

+ g, err := release.ReadGenerations(path)

+ if err != nil || g == nil {

+ t.Errorf("at the exec the generations file = %+v, %v; want it written", g, err)

+ g = &release.Generations{}

+ }

+ atExec = append(atExec, g)

+ }

+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {

+ t.Fatalf("WakeWorker: %v", err)

+ }

+ if len(atExec) != 1 || !reflect.DeepEqual(atExec[0].Hosts[host], []string{"v1"}) || !reflect.DeepEqual(atExec[0].Paths[host], []string{fakeIdentity("v1").StagedPath}) {

+ t.Fatalf("at the wake's exec the file named %s; want %s: v1 by version and path", describeGenerations(atExec), host)

+ }

+ ver.Store("v2")

+ a.sweepOnce() // the drift clock starts

+ clk.advance(ConvergeGrace + time.Minute)

+ a.sweepOnce()

+ waitFor(t, "the converge onto v2", func() bool { return a.entrySpawnVersion(host) == "v2" && !a.entryRespawning(host) })

+ if len(atExec) != 2 || !reflect.DeepEqual(atExec[1].Hosts[host], []string{"v1", "v2"}) {

+ t.Fatalf("at the converge's exec the file named %s; want %s: the serving v1 and the pending v2", describeGenerations(atExec), host)

+ }

+ waitForGenerations(t, path, map[string][]string{host: {"v2"}})

+ if g := readGenerationsFile(t, path); !reflect.DeepEqual(g.Paths, map[string][]string{host: {fakeIdentity("v2").StagedPath}}) {

+ t.Fatalf("after the converge the file's paths = %v; want only v2's", g.Paths)

+ }

+// Ruling R19 items 3 and 4: when the build cannot be published first, the

+// spawn does not happen, for a wake and for a crash recycle. Once the file

+// can be written again, the sweep's retry publishes the set without the

+// refused build.

+func TestWorkerSpawnRefusedWhenItsBuildCannotBePublished(t *testing.T) {

+ a, b, _, _, _ := stagedActivator(t)

+ b.runBeforeExec = true

+ data := filepath.Join(t.TempDir(), "data")

+ if err := os.WriteFile(data, []byte("a file where the data dir should be"), 0o644); err != nil {

+ t.Fatal(err)

+ }

+ path := release.GenerationsPath(data)

+ a.GenerationsFile = path

+ if err := a.WakeWorker(alwaysOnWorkerRoute("first.x.internal")); err == nil || !errors.Is(err, ErrBuildNotPublished) {

+ t.Fatalf("WakeWorker with an unwritable generations file = %v; want ErrBuildNotPublished", err)

+ }

+ if n := b.workerStarts(); n != 0 {

+ t.Fatalf("workerStarts = %d; a build that could not be published must not be exec'd", n)

+ }

+ // The file becomes writable; a worker wakes. Then publication breaks

+ // again and the child crashes: the recycle is not spawned either.

+ if err := os.Remove(data); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.MkdirAll(data, 0o755); err != nil {

+ t.Fatal(err)

+ }

+ host := "bot.x.internal"

+ if err := a.WakeWorker(alwaysOnWorkerRoute(host)); err != nil {

+ t.Fatalf("WakeWorker: %v", err)

+ }

+ if n := b.workerStarts(); n != 1 {

+ t.Fatalf("workerStarts = %d after the wake; want 1", n)

+ }

+ zdd := filepath.Dir(path)

+ if err := os.RemoveAll(zdd); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.WriteFile(zdd, []byte("a file where the zdd dir should be"), 0o644); err != nil {

+ t.Fatal(err)

+ }

+ asked := len(b.calls())

+ b.lastWorker().kill()

+ a.sweepOnce()

+ waitFor(t, "the recycle attempt to finish", func() bool { return len(b.calls()) == asked+1 && !a.entryRespawning(host) })

+ if n := b.workerStarts(); n != 1 {

+ t.Fatalf("workerStarts = %d after the recycle; a build that could not be published must not be exec'd", n)

+ }

+ if err := os.Remove(zdd); err != nil {

+ t.Fatal(err)

+ }

+ a.sweepOnce()

+ waitForGenerations(t, path, map[string][]string{host: {"v1"}})

+// Ruling R19 item 2 (Astra r3 F1(b)), with a real ExecBackend: a worker

+// exec'd through a chained live link (live/<host> -> operator-current ->

+// versions/v1/app) has Version "", and is published by its staged path,

+// which root's pin set then holds without any registry row.

+func TestWorkerFromAChainedLinkIsPublishedByPath(t *testing.T) {

+ eb := testExecBackend(t)

+ fb := &fakeBackend{}

+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))

+ if err != nil {

+ t.Fatal(err)

+ }

+ a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, &realWorkerBackend{fakeBackend: fb, exec: eb}, reg)

+ a.liveVer = eb.LiveVersion

+ a.GenerationsFile = release.GenerationsPath(t.TempDir())

+ t.Cleanup(func() { a.Shutdown(context.Background()) })

+ host := "bot.example.com"

+ argv0 := filepath.Join(t.TempDir(), "argv0")

+ staged := stageArgv0Worker(t, eb, host, "v1", argv0)

+ current := filepath.Join(t.TempDir(), "operator-current")

+ if err := os.Symlink(staged, current); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.Remove(eb.liveLink(host)); err != nil {

+ t.Fatal(err)

+ }

+ if err := os.Symlink(current, eb.liveLink(host)); err != nil {

+ t.Fatal(err)

+ }

+ rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`, Status: "ready"}

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

+ t.Fatalf("WakeWorker: %v", err)

+ }

+ if got := strings.TrimSpace(waitForFile(t, argv0, 5*time.Second)); got != staged {

+ t.Fatalf("the wake exec'd %q; want %s", got, staged)

+ }

+ if id := a.entryServing(host); id.Version != "" || id.StagedPath != staged {

+ t.Fatalf("captured %+v; want no Version and the staged path %s", id, staged)

+ }

+ g := readGenerationsFile(t, a.GenerationsFile)

+ if len(g.Hosts) != 0 || !reflect.DeepEqual(g.Paths, map[string][]string{host: {staged}}) {

+ t.Fatalf("generations file = hosts %v, paths %v; want only %s by path", g.Hosts, g.Paths, staged)

+ }

+ pins, err := release.GenerationPins(a.GenerationsFile, func() ([]registry.Route, error) { return nil, nil }).StagedPins(context.Background(), eb.OptDir)

+ if err != nil || !pins.Has(staged) {

+ t.Fatalf("root's generation pins = %v, %v; want %s pinned with no row", pins, err, staged)

+ }

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

index f6e857434..fe0ca501d 100644

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

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

@@ -26,6 +26,11 @@ type stagedFake struct {

// (onto != nil) returns instead of a child, once: the backend refusing a

// recycle.

failNextStaged error

+ // runBeforeExec, when set, makes the fake keep ExecBackend's contract

+ // with the activator's before-exec hook (ruling R19 item 3): it runs the

+ // hook with the identity it is about to "exec", and a hook error means

+ // no child. Off by default, so the R17 tests keep the spawn they pinned.

+ runBeforeExec bool

}

func fakeIdentity(ver string) release.StagedIdentity {

@@ -48,6 +53,15 @@ func (b *stagedFake) StartWorkerStaged(ctx context.Context, rt registry.Route, e

if onto != nil && fail != nil {

return nil, release.StagedIdentity{}, fail

}

+ if b.runBeforeExec {

+ id := fakeIdentity(b.fakeBackend.liveVerNow())

+ if onto != nil {

+ id = *onto

+ }

+ if err := runBeforeExec(ctx, id); err != nil {

+ return nil, release.StagedIdentity{}, err

+ }

p, ver, err := b.fakeBackend.StartWorker(ctx, rt, env)

if err != nil {

return p, release.StagedIdentity{}, err

@@ -58,6 +72,17 @@ func (b *stagedFake) StartWorkerStaged(ctx context.Context, rt registry.Route, e

return p, fakeIdentity(ver), nil

}

+// liveVerNow is the version the fake's next exec from the live link would

+// report, as StartWorker computes it.

+func (b *fakeBackend) liveVerNow() string {

+ b.mu.Lock()

+ defer b.mu.Unlock()

+ if b.execVerFn != nil {

+ return b.execVerFn()

+ }

+ return b.workerSpawnVer

func (b *stagedFake) calls() []*release.StagedIdentity {

b.smu.Lock()

defer b.smu.Unlock()

@@ -190,3 +215,55 @@ func TestSplitBackendForwardsStagedWorkerStarts(t *testing.T) {

t.Fatalf("a backend without the capability: %+v, %v; want today's StartWorker answer", id, err)

}

+// Ruling R19 item 3: ExecBackend runs the activator's before-exec hook with

+// the identity it resolved, before the child exists, and returns that same

+// identity.

+func TestExecBackendWorkerRunsTheBeforeExecHookFirst(t *testing.T) {

+ b := testExecBackend(t)

+ host := "bot.x.internal"

+ out := filepath.Join(t.TempDir(), "argv0")

+ stageArgv0Worker(t, b, host, "v1", out)

+ var seen []release.StagedIdentity

+ ctx := withBeforeExec(context.Background(), func(id release.StagedIdentity) error {

+ if _, err := os.Stat(out); err == nil {

+ t.Error("the child ran before the before-exec hook")

+ }

+ seen = append(seen, id)

+ return nil

+ })

+ p, id, err := b.StartWorkerStaged(ctx, workerRoute(host), nil, nil)

+ if err != nil {

+ t.Fatalf("StartWorkerStaged: %v", err)

+ }

+ t.Cleanup(func() { p.Stop(context.Background()) })

+ if len(seen) != 1 || seen[0] != id {

+ t.Fatalf("hook saw %+v; want exactly the identity exec'd, %+v", seen, id)

+ }

+ waitForFile(t, out, 5*time.Second)

+// Ruling R19 item 3: a before-exec hook that fails means no child, for a

+// wake from the live link and for a recycle onto a captured build.

+func TestExecBackendWorkerDoesNotExecWhenTheHookFails(t *testing.T) {

+ b := testExecBackend(t)

+ host := "bot.x.internal"

+ out := filepath.Join(t.TempDir(), "argv0")

+ stageArgv0Worker(t, b, host, "v1", out)

+ id, err := release.ResolveLive(b.OptDir, host)

+ if err != nil {

+ t.Fatal(err)

+ }

+ refuse := errors.New("publication failed")

+ ctx := withBeforeExec(context.Background(), func(release.StagedIdentity) error { return refuse })

+ for name, onto := range map[string]*release.StagedIdentity{"wake": nil, "recycle": &id} {

+ p, _, err := b.StartWorkerStaged(ctx, workerRoute(host), nil, onto)

+ if !errors.Is(err, refuse) || p != nil {

+ t.Fatalf("%s: StartWorkerStaged = %v, %v; want no child and the hook's error", name, p, err)

+ }

+ time.Sleep(200 * time.Millisecond)

+ if _, err := os.Stat(out); err == nil {

+ t.Fatal("a child ran although the before-exec hook failed")

+ }

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

index 8487df4c1..4af988ed6 100644

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

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

@@ -2,6 +2,8 @@ package activator

import (

"context"

+ "errors"

+ "fmt"

"log"

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

@@ -13,7 +15,9 @@ import (

// whatever live/<host> names now (onto == nil, a wake or a converge), and say

// exactly which bytes were exec'd. ExecBackend and SplitBackend have it. A

// backend without it keeps today's StartWorker, and its workers have no

-// captured identity to recycle onto.

+// captured identity to recycle onto. An implementation runs ctx's

+// before-exec hook (runBeforeExec) once it has resolved the identity and

+// before it execs, and does not exec when the hook fails (ruling R19 item 3).

type stagedWorkerStarter interface {

StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error)

}

@@ -23,14 +27,75 @@ var (

_ stagedWorkerStarter = (*SplitBackend)(nil)

)

+// ErrBuildNotPublished is a worker spawn refused because the build it was

+// about to exec could not be written to the generations file first (ruling

+// R19 item 3): root could not see that build as referenced, so it is not

+// run. The failed-spawn handling applies, and the sweep retries the write.

+var ErrBuildNotPublished = errors.New("worker build not published before its exec")

+// beforeExecKey carries the activator's before-exec hook on a spawn's

+// context (withBeforeExec).

+type beforeExecKey struct{}

+// withBeforeExec returns ctx carrying f, the hook a staged worker spawn runs

+// between resolving the identity it will exec and the exec itself (ruling R19

+// item 3). It rides on the context so StartWorkerStaged's signature, and the

+// stagedWorkerStarter interface, stay as they are.

+func withBeforeExec(ctx context.Context, f func(release.StagedIdentity) error) context.Context {

+ return context.WithValue(ctx, beforeExecKey{}, f)

+// runBeforeExec runs ctx's before-exec hook, if any, with the identity about

+// to be exec'd. A backend that implements stagedWorkerStarter calls it after

+// resolving that identity and before the exec, and does not exec when it

+// returns an error.

+func runBeforeExec(ctx context.Context, id release.StagedIdentity) error {

+ if f, ok := ctx.Value(beforeExecKey{}).(func(release.StagedIdentity) error); ok && f != nil {

+ return f(id)

+ }

+ return nil

// startWorker spawns through the staged capability when the backend has it,

-// and through StartWorker otherwise.

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

- if s, ok := a.backend.(stagedWorkerStarter); ok {

- return s.StartWorkerStaged(ctx, rt, env, onto)

+// and through StartWorker otherwise. Never call it with a.mu held.

+//

+// Through the staged capability, the build about to be exec'd is published

+// BEFORE the exec (ruling R19 item 3): the backend's before-exec hook records

+// it as a pending generation and writes the generations file. A write that

+// fails refuses the spawn with ErrBuildNotPublished. The returned

+// pendingGeneration names that record. On an error startWorker has already

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

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

+ if !ok {

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

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

+ }

+ var pend pendingGeneration

+ hook := func(id release.StagedIdentity) error {

+ if pend != 0 {

+ a.mu.Lock()

+ a.dropPendingGenerationLocked(pend)

+ a.mu.Unlock()

+ }

+ pend = a.addPendingGeneration(rt.Host, id)

+ if err := a.persistGenerations(); err != nil {

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

+ }

+ return nil

+ }

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

+ if err != nil && pend != 0 {

+ a.mu.Lock()

+ a.dropPendingGenerationLocked(pend)

+ a.mu.Unlock()

+ _ = a.persistGenerations()

+ pend = 0

}

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

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

+ return p, id, pend, err

}

// recycleWorkerLocked replaces a worker's child for a reason that is not a