amadan

carlos / platform Public

zdd-4 task 8: workers exec a staged build by path and report what they exec'd

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude b880fbb7bc408a38a10f229e8787c0d54b5e88a4 parent fdec1bf
4 files changed, +261 −19
  • internal/activator/backend_exec.go +45 −19
  • internal/activator/backend_unit.go +12 −0
  • internal/activator/worker_staged_test.go +181 −0
  • internal/activator/workerbuild.go +23 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 4ae1ea1e7..199d59d79 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -1719,30 +1719,56 @@ func (b *ExecBackend) startSidecarChild(_ context.Context, rt registry.Route) (P
return p, nil
}
-// StartWorker execs the host's live symlink with the route's declared argv
-// (rt.Args, a JSON array — a sidecar's `["sidecar","run"]` today, any argv a
-// future worker kind declares tomorrow) under the instance's own env layers,
-// CARLOS_*-clean like every other child this backend spawns. Like
-// StartSidecar, deliberately NO health wait: a worker binds nothing by
-// definition — no socket, no port — so there is nothing waitHealthyCtx could
-// ever dial. Unlike StartSidecar, the caller supplies extra env (the
-// per-wake sidecar token) that must WIN over the tenant's own <host>.env —
-// applied last, after instanceEnv's own merge, for exactly that reason.
+// StartWorker is StartWorkerStaged onto whatever live/<host> names now. Its
+// string result is that build's version, as it always was.
func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra map[string]string) (Proc, string, error) {
+ p, id, err := b.StartWorkerStaged(ctx, rt, extra, nil)
+ return p, id.Version, err
+}
+
+// StartWorkerStaged execs a worker with the route's declared argv (rt.Args, a
+// JSON array: a sidecar's `["sidecar","run"]` today) under the instance's own
+// env layers, CARLOS_*-clean like every other child this backend spawns, and
+// with NO health wait: a worker binds nothing, so there is nothing
+// waitHealthyCtx could dial. extra (the per-wake sidecar token) is applied
+// last and wins over the tenant's own <host>.env.
+//
+// What it execs is a staged build BY PATH, never live/<host> (zero-downtime
+// deploys spec §7.9):
+// - onto == nil (a wake, or a converge onto a new build): the live link is
+// ensured, read ONCE, and the file it names is hashed and exec'd;
+// - onto != nil (a crash or max_run recycle): onto.StagedPath is checked
+// against onto.Digest (release.VerifyStaged) and exec'd. A copy that has
+// changed is refused with release.ErrStagedCopyChanged and nothing runs.
+//
+// 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.
+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 {
- return nil, "", fmt.Errorf("worker %s: no runnable argv in args %q", rt.Host, rt.Args)
+ return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: no runnable argv in args %q", rt.Host, rt.Args)
}
- if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
- return nil, "", err
+ var id release.StagedIdentity
+ if onto == nil {
+ if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
+ return nil, release.StagedIdentity{}, err
+ }
+ got, err := release.ResolveLive(b.OptDir, rt.Host)
+ if err != nil {
+ return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err)
+ }
+ id = got
+ } else {
+ if err := release.VerifyStaged(*onto); err != nil {
+ return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: not recycled onto %s: %w", rt.Host, onto.Version, err)
+ }
+ id = *onto
}
- // Read immediately before the spawn, same reasoning as Start: the
- // version this call returns must name the build actually exec'd.
- spawnVer, _ := b.LiveVersion(rt.Host)
fileEnv, err := b.instanceEnv(rt)
if err != nil {
- return nil, "", err
+ return nil, release.StagedIdentity{}, err
}
for k, v := range extra {
fileEnv = withVar(fileEnv, k, v)
@@ -1754,13 +1780,13 @@ func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra
// ships an uncapped process — the exact failure spawnInCgroup's own
// comment names.
cmd, settle, err := b.spawnInCgroup(rt, func() *exec.Cmd {
- cmd := exec.Command(b.liveLink(rt.Host), argv...)
+ cmd := exec.Command(id.StagedPath, argv...)
cmd.Env = childEnv(fileEnv)
b.teeStd(cmd, rt)
return cmd
})
if err != nil {
- return nil, "", err
+ return nil, release.StagedIdentity{}, err
}
p := &childProc{cmd: cmd, done: make(chan struct{})}
go func() {
@@ -1782,7 +1808,7 @@ func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra
b.cleanupHost(rt.Host)
close(p.done)
}()
- return p, spawnVer, nil
+ return p, id, nil
}
// startSidecarUnit is StartSidecar's nested body: a sidecar has no health
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go
index 137f63de0..765bb60c4 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -342,6 +342,18 @@ func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env m
return b.pick(rt).StartWorker(ctx, rt, env)
}
+// StartWorkerStaged forwards spec §7.9's staged spawn to the backend the
+// route picks. One without the capability (UnitBackend, which refuses
+// workers anyway) gets today's StartWorker, and its identity is the version
+// alone, which no recycle can be verified against.
+func (b *SplitBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
+ if s, ok := b.pick(rt).(stagedWorkerStarter); ok {
+ return s.StartWorkerStaged(ctx, rt, env, onto)
+ }
+ p, ver, err := b.pick(rt).StartWorker(ctx, rt, env)
+ return p, release.StagedIdentity{Version: ver}, err
+}
+
// LiveVersion forwards to the exec side's reader when it has one — both
// backings share the live/<host> symlink layout, so one reader serves the
// convergence sweep for every host. (New picks this up via the optional
diff --git a/internal/activator/worker_staged_test.go b/internal/activator/worker_staged_test.go
new file mode 100644
index 000000000..3c30a5e3f
--- /dev/null
+++ b/internal/activator/worker_staged_test.go
@@ -0,0 +1,181 @@
+package activator
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// stagedFake is fakeBackend with spec §7.9's capability. It records which
+// build each worker spawn asked for (nil = "what the live link names") and
+// reports a synthetic identity, so a test can tell builds apart without real
+// files.
+type stagedFake struct {
+ *fakeBackend
+ smu sync.Mutex
+ onto []*release.StagedIdentity
+}
+
+func fakeIdentity(ver string) release.StagedIdentity {
+ return release.StagedIdentity{Version: ver, StagedPath: "/opt/carlos/versions/" + ver + "/app", Digest: "sha-of-" + ver}
+}
+
+func (b *stagedFake) StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) {
+ b.smu.Lock()
+ if onto != nil {
+ c := *onto
+ b.onto = append(b.onto, &c)
+ } else {
+ b.onto = append(b.onto, nil)
+ }
+ b.smu.Unlock()
+ p, ver, err := b.fakeBackend.StartWorker(ctx, rt, env)
+ if err != nil {
+ return p, release.StagedIdentity{}, err
+ }
+ if onto != nil {
+ return p, *onto, nil
+ }
+ return p, fakeIdentity(ver), nil
+}
+
+func (b *stagedFake) calls() []*release.StagedIdentity {
+ b.smu.Lock()
+ defer b.smu.Unlock()
+ return append([]*release.StagedIdentity(nil), b.onto...)
+}
+
+// stageArgv0Worker stages a worker build at <OptDir>/versions/<ver>/app that
+// writes its own $0 to out and then sleeps, and points live/<host> at it. The
+// version is in the script, so each build hashes differently.
+func stageArgv0Worker(t *testing.T, b *ExecBackend, host, ver, out string) string {
+ t.Helper()
+ vdir := filepath.Join(b.OptDir, "versions", ver)
+ if err := os.MkdirAll(vdir, 0o755); err != nil {
+ t.Fatal(err)
+ }
+ bin := filepath.Join(vdir, "app")
+ script := "#!/bin/sh\n# build " + ver + "\nprintf '%s\\n' \"$0\" > '" + out + ".tmp' && mv '" + out + ".tmp' '" + out + "'\nexec sleep 30\n"
+ if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Join(b.OptDir, "live"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ os.Remove(b.liveLink(host))
+ if err := os.Symlink(bin, b.liveLink(host)); err != nil {
+ t.Fatal(err)
+ }
+ return bin
+}
+
+func workerRoute(host string) registry.Route {
+ return registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`}
+}
+
+// Spec §7.9: the serving identity is the build captured at exec, so the exec
+// must be of the file the identity names, never of live/<host>.
+func TestExecBackendWorkerExecsTheStagedPathItResolved(t *testing.T) {
+ b := testExecBackend(t)
+ host := "bot.x.internal"
+ out := filepath.Join(t.TempDir(), "argv0")
+ bin := stageArgv0Worker(t, b, host, "v1", out)
+ p, id, err := b.StartWorkerStaged(context.Background(), workerRoute(host), nil, nil)
+ if err != nil {
+ t.Fatalf("StartWorkerStaged: %v", err)
+ }
+ defer p.Stop(context.Background())
+ if got := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); got != bin {
+ t.Fatalf("$0 = %q; want the staged path %q, never live/%s", got, bin, host)
+ }
+ sum, err := release.FileSHA256(bin)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if want := (release.StagedIdentity{Version: "v1", StagedPath: bin, Digest: sum}); id != want {
+ t.Fatalf("identity = %+v, want %+v", id, want)
+ }
+}
+
+// Spec §7.9: a recycle execs the captured build even after root has
+// repointed the link.
+func TestExecBackendWorkerRecycleExecsTheCapturedBuildAfterARepoint(t *testing.T) {
+ b := testExecBackend(t)
+ host := "bot.x.internal"
+ out1 := filepath.Join(t.TempDir(), "argv0-v1")
+ bin1 := stageArgv0Worker(t, b, host, "v1", out1)
+ p1, id1, err := b.StartWorkerStaged(context.Background(), workerRoute(host), nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitForFile(t, out1, 5*time.Second)
+ p1.Stop(context.Background())
+ os.Remove(out1)
+
+ stageArgv0Worker(t, b, host, "v2", filepath.Join(t.TempDir(), "argv0-v2")) // root installs v2
+
+ p2, id2, err := b.StartWorkerStaged(context.Background(), workerRoute(host), nil, &id1)
+ if err != nil {
+ t.Fatalf("recycle onto v1: %v", err)
+ }
+ defer p2.Stop(context.Background())
+ if got := strings.TrimSpace(waitForFile(t, out1, 5*time.Second)); got != bin1 {
+ t.Fatalf("recycle exec'd %q, want the captured %q", got, bin1)
+ }
+ if id2 != id1 {
+ t.Fatalf("recycle reported %+v, want the captured %+v", id2, id1)
+ }
+}
+
+// Spec §4.2: a recovery whose staged copy has changed is refused, and nothing
+// runs.
+func TestExecBackendWorkerRecycleRefusesAChangedStagedCopy(t *testing.T) {
+ b := testExecBackend(t)
+ host := "bot.x.internal"
+ out := filepath.Join(t.TempDir(), "argv0")
+ bin := stageArgv0Worker(t, b, host, "v1", out)
+ p1, id1, err := b.StartWorkerStaged(context.Background(), workerRoute(host), nil, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ waitForFile(t, out, 5*time.Second)
+ p1.Stop(context.Background())
+ os.Remove(out)
+ if err := os.WriteFile(bin, []byte("#!/bin/sh\n# tampered, and longer than the original\nexec sleep 30\n"), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ p2, _, err := b.StartWorkerStaged(context.Background(), workerRoute(host), nil, &id1)
+ if !errors.Is(err, release.ErrStagedCopyChanged) {
+ t.Fatalf("recycle onto changed bytes = %v, want ErrStagedCopyChanged", err)
+ }
+ if p2 != nil {
+ p2.Stop(context.Background())
+ t.Fatal("a refused recycle still spawned a child")
+ }
+}
+
+func TestSplitBackendForwardsStagedWorkerStarts(t *testing.T) {
+ exec := &stagedFake{fakeBackend: &fakeBackend{workerSpawnVer: "v1"}}
+ sb := &SplitBackend{Exec: exec, Unit: &fakeBackend{workerSpawnVer: "u1"}}
+ rt := testWorkerRoute("bot.x.internal")
+ onto := fakeIdentity("v0")
+ if _, id, err := sb.StartWorkerStaged(context.Background(), rt, nil, &onto); err != nil || id != onto {
+ t.Fatalf("exec route: identity %+v, %v; want %+v forwarded", id, err, onto)
+ }
+ if calls := exec.calls(); len(calls) != 1 || calls[0] == nil || *calls[0] != onto {
+ t.Fatalf("exec backend saw %v, want one call onto v0", calls)
+ }
+ unit := rt
+ unit.Backing = registry.BackingUnit
+ if _, id, err := sb.StartWorkerStaged(context.Background(), unit, nil, &onto); err != nil || id != (release.StagedIdentity{Version: "u1"}) {
+ t.Fatalf("a backend without the capability: %+v, %v; want today's StartWorker answer", id, err)
+ }
+}
diff --git a/internal/activator/workerbuild.go b/internal/activator/workerbuild.go
new file mode 100644
index 000000000..cef329906
--- /dev/null
+++ b/internal/activator/workerbuild.go
@@ -0,0 +1,23 @@
+package activator
+
+import (
+ "context"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// stagedWorkerStarter is the optional Backend capability behind spec §7.9:
+// spawn a worker from a NAMED staged build (onto != nil, a recycle) or from
+// 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.
+type stagedWorkerStarter interface {
+ StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error)
+}
+
+var (
+ _ stagedWorkerStarter = (*ExecBackend)(nil)
+ _ stagedWorkerStarter = (*SplitBackend)(nil)
+)