carlos / platform Public

Sidecar isolation Task 6 (3/6): StartWorkerStaged dispatches on layout; the nested arm runs the worker unit (D5)

A fenced or conflicting worker row is refused with the reason before
anything is resolved. A nested row, after the shared first half, runs
startWorkerUnit: the claim refusal, nestedGate, prepareWorker, a
launch.worker.json with argv [live link, args...] and the composed env
(CARLOS_SIDECAR_DB and the per-host CARLOS_AGENT_SOCKET on top of the
tenant's layers), the ctx's exec gate immediately before the start request,
then the control service's start for the worker role always naming the
staged build the first half resolved. Root refusals that ran nothing stay
plain; every other start error, and an unread invocation, is marked both
legacyStartUncertain and startUncertain (D19, R2-1). Flat and foreign rows
run today's child.

Tests run through a real instancectl.Server on a unix socket whose
Systemctl is a fake systemd that execs the launch file as the trampoline
would, with UnitFacts and CgroupEvents read from its children.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev 8af9818325f4a3a1fc53c3e68ed10c33d8ea76e8 parent dcd04fb
2 files changed, +651 −0
  • internal/activator/backend_exec.go +93 −0
  • internal/activator/worker_nested_test.go +558 −0
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 7e5c1f3b4..175fb79e7 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -3056,11 +3056,24 @@ func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra
// A worker on a replica row passes replicaKeyGate as a seed does, with no
// control socket, before that hook: its home's key, verified, or nothing
// runs (instance-key plan ruling 26). A worker never mints a key.
+//
+// It dispatches on the row's layout (sidecar isolation D5), as Start and
+// StartSidecar do. A fenced or conflicting row is refused first, naming the
+// reason, before anything is resolved or published. The first half above
+// (argv, build, LiveExecGate, env layers, the replica gate, runBeforeExec)
+// is shared; then a nested row runs the worker role's unit through the root
+// control service (startWorkerUnit), and a flat or foreign row runs the
+// child below.
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, release.StagedIdentity{}, fmt.Errorf("worker %s: no runnable argv in args %q", rt.Host, rt.Args)
}
+ paths := b.pathsFor(rt)
+ switch paths.layout {
+ case LayoutFenced, LayoutConflict:
+ return nil, release.StagedIdentity{}, fmt.Errorf("%s: refusing to start worker: %s", rt.Host, paths.reason)
+ }
var id release.StagedIdentity
if onto == nil {
if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
@@ -3118,6 +3131,12 @@ func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route,
return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err)
}
+ if paths.layout == LayoutNested {
+ // A nested worker row runs as its own unit (sidecar isolation D5):
+ // never a flat child of the agent, whatever the row looks like.
+ return b.startWorkerUnit(ctx, rt, id, argv, fileEnv)
+ }
+
// 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
@@ -3164,6 +3183,80 @@ func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route,
return p, id, nil
}
+// startWorkerUnit is StartWorkerStaged's nested arm (sidecar isolation D5):
+// the claim refusal, nestedGate, prepareWorker (the directories, the env
+// layers, no admin token, the log listener and the worker's own agent
+// listener, D3), a launch.worker.json whose argv is the live link followed
+// by the manifest's args and whose env is the composed one the flat arm
+// would exec with (the tenant's layers, the agent-owned variables on top),
+// the ctx's exec gate immediately before the start request, and then the
+// control service's start for the worker role ALWAYS naming the staged
+// build the first half resolved: the worker contract is a build by path,
+// never the live link (zero-downtime spec §7.9), and root binds exactly the
+// file the agent hashed. No health wait: a worker binds nothing.
+//
+// A start the control service refused before it laid out or started
+// anything (instancectl.ErrStagedExe), or that was never sent (the channel
+// cannot name a build, errNoStagedExeControl), ran nothing. Every other
+// outcome short of the gate's own refusal (launchRefused) is marked both
+// legacyStartUncertain and startUncertain (D19, R2-1): root may have
+// started the unit, a start job systemd queued outlives this call's
+// cancellation, and an invocation root could not read may still have run.
+// The failed wake's cleanup then stops the unit through the fenced /stop
+// and proves it gone (RoleGone) before the lease goes, and a replacement
+// treats the attempt's build as having possibly run. A failed start closes
+// the listeners prepareWorker bound.
+func (b *ExecBackend) startWorkerUnit(ctx context.Context, rt registry.Route, id release.StagedIdentity, args []string, fileEnv map[string]string) (Proc, release.StagedIdentity, error) {
+ none := release.StagedIdentity{}
+ if b.Cfg.HostWrites.Enabled {
+ return nil, none, ErrClaimIsolationUnavailable
+ }
+ if err := b.nestedGate(rt); err != nil {
+ return nil, none, err
+ }
+ p, logs, agent, _, _, err := b.prepareWorker(rt)
+ if err != nil {
+ return nil, none, err
+ }
+ fail := func(err error) (Proc, release.StagedIdentity, error) {
+ logs.Close()
+ agent.Close()
+ return nil, none, err
+ }
+ argv := append([]string{b.liveLink(rt.Host)}, args...)
+ if err := b.writeLaunch(p, instancectl.RoleWorker, rt.Host, argv, childEnv(fileEnv)); err != nil {
+ return fail(err)
+ }
+ // The caller's exec gate (and the wake's pending record), asked
+ // immediately before root is asked, as the nested instance start asks
+ // it and as the flat worker's cgroup hook asks it: the config read, the
+ // listeners and the launch file above all took time. A refusal starts
+ // nothing, and is known not to have (launchRefused).
+ if gate := b.launchGate(ctx, rt.Host); gate != nil {
+ if err := gate(); err != nil {
+ return fail(launchRefused{err})
+ }
+ }
+ began := time.Now()
+ unit := unitName(instancectl.RoleWorker, rt.Host, 0)
+ inv, err := b.startRole(ctx, rt, instancectl.RoleWorker, 0, stagedExeOf(id), cgroup.Limit{})
+ switch {
+ case errors.Is(err, errNoStagedExeControl), errors.Is(err, instancectl.ErrStagedExe):
+ // Nothing was sent, or root refused before it laid out or started
+ // anything: proven not run.
+ return fail(fmt.Errorf("%s: start %s: %w", rt.Host, unit, err))
+ case err != nil:
+ return fail(markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start %s: %w", rt.Host, unit, err))))
+ case inv == "":
+ // Root started the unit and read no invocation: it had exited
+ // already (errStartedThenExited). The unit ran; its stop and proof
+ // are still the cleanup's, as for any uncertain worker start.
+ return fail(markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: %w", rt.Host, exitedUnobserved{unit: unit}))))
+ }
+ log.Printf("activator: %s: %s launched in %s on %s", rt.Host, unit, time.Since(began).Round(time.Millisecond), id.Version)
+ return b.newWorkerUnitProc(rt, p, inv, logs, agent), id, nil
+}
+
// startSidecarUnit is StartSidecar's nested body: a sidecar has no health
// wait either way, and no control listener of its own. Sidecar env is a
// single layer, read straight from its own file, never merged with the
diff --git a/internal/activator/worker_nested_test.go b/internal/activator/worker_nested_test.go
index f31990362..be1694240 100644
--- a/internal/activator/worker_nested_test.go
+++ b/internal/activator/worker_nested_test.go
@@ -3,16 +3,30 @@
package activator
import (
+ "context"
"encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "net"
"os"
+ "os/exec"
"path/filepath"
"slices"
+ "strconv"
+ "strings"
"sync"
"sync/atomic"
+ "syscall"
"testing"
"time"
+ "github.com/carlosframework/platform/internal/agentsock"
"github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/version"
)
// Sidecar isolation D4: every worker, flat or nested, is told where its
@@ -225,3 +239,547 @@ func TestRequestedStopNeverReachesProcCrashed(t *testing.T) {
t.Fatalf("worker starts = %d, want 1: a requested stop was restarted as a crash", b.workerStarts())
}
}
+
+// ---- the production-wired harness --------------------------------------
+//
+// A real instancectl.Server on a unix socket, its Systemctl seam played by
+// fakeSystemd, which does what the trampoline would: on `start` it reads
+// launch.<role>.json from the instance directory and execs the build root
+// laid out at <ExeRoot>/<iid>/app with the file's argv and env, its output
+// on the role's log socket; on `stop` it signals the child. The agent side
+// is a real ExecBackend whose Control is the real client, with UnitFacts
+// and CgroupEvents read from the fake's children.
+
+// fakeSystemd is the real control server's systemd.
+type fakeSystemd struct {
+ t *testing.T
+ mu sync.Mutex
+ dataDir, runDir, exeRoot string
+ units map[string]*fakeSvc
+ calls []string
+ nextInv int
+ // holdStart, when set, holds every `start` until it is closed or the
+ // call's context ends, as exec.CommandContext would end it (D19: a
+ // handler held past its client's cancellation).
+ holdStart chan struct{}
+ // starts counts `start` calls that reached their exec.
+ starts int
+}
+
+// fakeSvc is one unit the fake systemd has started.
+type fakeSvc struct {
+ cmd *exec.Cmd
+ inv string
+ active string // active, inactive, failed
+ result string // success, exit-code
+ status string // the exit record's status
+ done chan struct{}
+}
+
+func newFakeSystemd(t *testing.T, dataDir, runDir, exeRoot string) *fakeSystemd {
+ return &fakeSystemd{t: t, dataDir: dataDir, runDir: runDir, exeRoot: exeRoot, units: map[string]*fakeSvc{}}
+}
+
+func (s *fakeSystemd) joined() string {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return strings.Join(s.calls, "\n")
+}
+
+func (s *fakeSystemd) startCount() int {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.starts
+}
+
+// unitInstance is the unit's role and unescaped instance id.
+func unitInstance(unit string) (instancectl.Role, string, bool) {
+ role, esc, ok := instancectl.ParseUnitName(unit)
+ if !ok {
+ return "", "", false
+ }
+ iid, ok := instancectl.Unescape(esc)
+ return role, iid, ok
+}
+
+func (s *fakeSystemd) run(ctx context.Context, args ...string) (string, error) {
+ s.mu.Lock()
+ s.calls = append(s.calls, strings.Join(args, " "))
+ hold := s.holdStart
+ s.mu.Unlock()
+ if len(args) == 0 {
+ return "", errors.New("systemctl: no verb")
+ }
+ switch args[0] {
+ case "set-property", "reset-failed", "list-jobs", "daemon-reload":
+ if args[0] == "reset-failed" && len(args) > 1 {
+ s.mu.Lock()
+ if svc := s.units[args[1]]; svc != nil && svc.active != "active" {
+ delete(s.units, args[1])
+ }
+ s.mu.Unlock()
+ }
+ return "", nil
+ case "list-units":
+ return "", nil
+ case "show":
+ unit := args[len(args)-1]
+ s.mu.Lock()
+ svc := s.units[unit]
+ s.mu.Unlock()
+ inv := ""
+ if svc != nil {
+ inv = svc.inv
+ }
+ if slices.Contains(args, "--value") {
+ return inv + "\n", nil
+ }
+ active := "inactive"
+ if svc != nil {
+ active = svc.active
+ }
+ return "ActiveState=" + active + "\nInvocationID=" + inv + "\n", nil
+ case "start":
+ if hold != nil {
+ select {
+ case <-hold:
+ case <-ctx.Done():
+ return "", ctx.Err()
+ }
+ }
+ if ctx.Err() != nil {
+ return "", ctx.Err()
+ }
+ return "", s.start(args[1])
+ case "stop":
+ return "", s.stop(args[1])
+ }
+ return "", fmt.Errorf("systemctl %v: not played by this fake", args)
+}
+
+// start is the trampoline's stub: exec the laid-out build with the launch
+// file's argv and env, its output on the role's log socket.
+func (s *fakeSystemd) start(unit string) error {
+ role, iid, ok := unitInstance(unit)
+ if !ok {
+ return fmt.Errorf("start %s: not an instance unit", unit)
+ }
+ raw, err := os.ReadFile(filepath.Join(s.dataDir, "instances", iid, instancectl.LaunchFileName(role)))
+ if err != nil {
+ return fmt.Errorf("start %s: %v", unit, err)
+ }
+ var launch instancectl.Launch
+ if err := json.Unmarshal(raw, &launch); err != nil || len(launch.Argv) == 0 {
+ return fmt.Errorf("start %s: launch file: %v", unit, err)
+ }
+ // The namespace binds the exe root's app over the live link; the fake
+ // execs the file that link names, so what runs is what root laid out.
+ exe, err := filepath.EvalSymlinks(filepath.Join(s.exeRoot, iid, "app"))
+ if err != nil {
+ return fmt.Errorf("start %s: exe root: %v", unit, err)
+ }
+ cmd := exec.Command(exe, launch.Argv[1:]...)
+ cmd.Env = launch.Env
+ var out io.Writer = io.Discard
+ var conn net.Conn
+ if c, err := net.Dial("unix", filepath.Join(s.runDir, "logs", iid+"."+string(role)+".sock")); err == nil {
+ conn, out = c, c
+ }
+ cmd.Stdout, cmd.Stderr = out, out
+ if err := cmd.Start(); err != nil {
+ if conn != nil {
+ conn.Close()
+ }
+ return fmt.Errorf("start %s: %v", unit, err)
+ }
+ s.mu.Lock()
+ s.nextInv++
+ s.starts++
+ svc := &fakeSvc{cmd: cmd, inv: fmt.Sprintf("inv%d", s.nextInv), active: "active", done: make(chan struct{})}
+ s.units[unit] = svc
+ s.mu.Unlock()
+ go func() {
+ werr := cmd.Wait()
+ if conn != nil {
+ conn.Close()
+ }
+ status := 0
+ if ps := cmd.ProcessState; ps != nil {
+ if ws, ok := ps.Sys().(syscall.WaitStatus); ok && ws.Signaled() {
+ // The trampoline reports a tenant's signal death as
+ // 128+signal (initExitCode).
+ status = 128 + int(ws.Signal())
+ } else {
+ status = ps.ExitCode()
+ }
+ } else if werr != nil {
+ status = 1
+ }
+ s.mu.Lock()
+ svc.status = strconv.Itoa(status)
+ if status == 0 {
+ svc.active, svc.result = "inactive", "success"
+ } else {
+ svc.active, svc.result = "failed", "exit-code"
+ }
+ s.mu.Unlock()
+ // ExecStopPost's advisory record.
+ rec, _ := json.Marshal(instancectl.ExitRecord{Invocation: svc.inv, Result: svc.result, Code: "exited", Status: svc.status, At: time.Now()})
+ _ = os.WriteFile(filepath.Join(s.runDir, "instances", iid, instancectl.ExitRecordName(role, svc.inv)), rec, 0o600)
+ close(svc.done)
+ }()
+ return nil
+}
+
+func (s *fakeSystemd) stop(unit string) error {
+ s.mu.Lock()
+ svc := s.units[unit]
+ s.mu.Unlock()
+ if svc == nil {
+ return nil
+ }
+ select {
+ case <-svc.done:
+ return nil
+ default:
+ }
+ _ = svc.cmd.Process.Signal(syscall.SIGTERM)
+ select {
+ case <-svc.done:
+ case <-time.After(5 * time.Second):
+ _ = svc.cmd.Process.Kill()
+ <-svc.done
+ }
+ return nil
+}
+
+// facts is the backend's UnitFacts: what `systemctl show` would say.
+func (s *fakeSystemd) facts(_ context.Context, unit string) (string, string, string, error) {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ svc := s.units[unit]
+ if svc == nil {
+ return "inactive", "", "success", nil
+ }
+ return svc.active, svc.inv, svc.result, nil
+}
+
+// events is the backend's CgroupEvents: populated while the child runs.
+func (s *fakeSystemd) events(_, unit string) (bool, bool, error) {
+ s.mu.Lock()
+ svc := s.units[unit]
+ s.mu.Unlock()
+ if svc == nil {
+ return false, false, nil
+ }
+ select {
+ case <-svc.done:
+ return false, false, nil
+ default:
+ return true, true, nil
+ }
+}
+
+// unitOf is the fake's record of unit, or nil.
+func (s *fakeSystemd) unitOf(unit string) *fakeSvc {
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ return s.units[unit]
+}
+
+// workerHarness is the harness above for one nested worker row.
+type workerHarness struct {
+ t *testing.T
+ b *ExecBackend
+ srv *instancectl.Server
+ sd *fakeSystemd
+ rt registry.Route
+ host string
+ root string
+ sink *workerSink
+}
+
+const harnessHost = "bot.shop.abc123.sidecar.internal"
+
+// newWorkerHarness wires the harness. workers answers the agent socket (the
+// Service in the end-to-end tests, a pinWorkers otherwise).
+func newWorkerHarness(t *testing.T, workers agentsock.Workers) *workerHarness {
+ t.Helper()
+ root, err := os.MkdirTemp("", "wh")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(root) })
+ b := testExecBackend(t)
+ b.DataDir = filepath.Join(root, "data")
+ b.OptDir = filepath.Join(root, "opt")
+ b.RunDirOverride = filepath.Join(root, "run")
+ b.ObservePoll = 20 * time.Millisecond
+ exeRoot := filepath.Join(root, "exe")
+ sd := newFakeSystemd(t, b.DataDir, b.RunDirOverride, exeRoot)
+ b.UnitFacts, b.CgroupEvents = sd.facts, sd.events
+ sink := &workerSink{}
+ b.Logs = sink
+ if workers == nil {
+ workers = &pinWorkers{tokens: map[string]string{}}
+ }
+ b.AgentSock = agentsock.New(workers)
+
+ srv := &instancectl.Server{
+ Systemctl: sd.run,
+ PeerOK: func(net.Conn) error { return nil },
+ CgroupRoot: filepath.Join(root, "cgroup"),
+ ExeRoot: exeRoot,
+ OptDir: b.OptDir,
+ HostApp: func(string) (string, string, error) { return "", "app", nil },
+ Logf: t.Logf,
+ Build: version.Build,
+ }
+ sock := filepath.Join(root, "c.sock")
+ ln, err := net.Listen("unix", sock)
+ if err != nil {
+ t.Skipf("cannot bind a unix socket here (%v); run with TMPDIR=/tmp/g", err)
+ }
+ ctx, cancel := context.WithCancel(context.Background())
+ done := make(chan struct{})
+ go func() { defer close(done); _ = srv.Serve(ctx, ln) }()
+ t.Cleanup(func() { cancel(); ln.Close(); <-done })
+ b.Control = instancectl.NewClient(sock)
+
+ rt := registry.Route{Host: harnessHost, Kind: "sidecar", Args: `["work"]`, Hibernate: true, Status: "ready",
+ DBPath: provision.NestedDBPath(b.DataDir, harnessHost)}
+ if l, why := b.Layout(rt); l != LayoutNested {
+ t.Fatalf("harness worker row classifies as %s (%s), want nested", l, why)
+ }
+ return &workerHarness{t: t, b: b, srv: srv, sd: sd, rt: rt, host: harnessHost, root: root, sink: sink}
+}
+
+// stage writes build ver of the worker's app as script at the version
+// store's path for it, points live/<host> at it, and returns its identity.
+func (h *workerHarness) stage(ver, script string) release.StagedIdentity {
+ h.t.Helper()
+ bin := release.VersionPath(h.b.OptDir, "", ver, "app")
+ if err := os.MkdirAll(filepath.Dir(bin), 0o755); err != nil {
+ h.t.Fatal(err)
+ }
+ if err := os.WriteFile(bin, []byte(script), 0o755); err != nil {
+ h.t.Fatal(err)
+ }
+ if err := os.MkdirAll(filepath.Join(h.b.OptDir, "live"), 0o755); err != nil {
+ h.t.Fatal(err)
+ }
+ os.Remove(h.b.liveLink(h.host))
+ if err := os.Symlink(bin, h.b.liveLink(h.host)); err != nil {
+ h.t.Fatal(err)
+ }
+ sum, err := release.FileSHA256(bin)
+ if err != nil {
+ h.t.Fatal(err)
+ }
+ return release.StagedIdentity{Version: ver, StagedPath: bin, Digest: sum}
+}
+
+// workerUnit is the worker unit's name.
+func (h *workerHarness) workerUnit() string { return instancectl.UnitName(instancectl.RoleWorker, h.host) }
+
+// instanceDir is the row's instance directory.
+func (h *workerHarness) instanceDir() string { return provision.InstanceDir(h.b.DataDir, h.host) }
+
+// writeTenantEnv writes the tenant's own <host>.env in the instance dir.
+func (h *workerHarness) writeTenantEnv(content string) {
+ h.t.Helper()
+ if err := os.MkdirAll(h.instanceDir(), 0o700); err != nil {
+ h.t.Fatal(err)
+ }
+ if err := os.WriteFile(filepath.Join(h.instanceDir(), h.host+".env"), []byte(content), 0o600); err != nil {
+ h.t.Fatal(err)
+ }
+}
+
+// readLaunch reads launch.worker.json.
+func (h *workerHarness) readLaunch() instancectl.Launch {
+ h.t.Helper()
+ raw, err := os.ReadFile(filepath.Join(h.instanceDir(), instancectl.LaunchFileName(instancectl.RoleWorker)))
+ if err != nil {
+ h.t.Fatalf("launch.worker.json: %v", err)
+ }
+ var l instancectl.Launch
+ if err := json.Unmarshal(raw, &l); err != nil {
+ h.t.Fatalf("launch.worker.json: %v", err)
+ }
+ return l
+}
+
+// argv0Script is a build that records its own path and then waits.
+func argv0Script(ver, out string) string {
+ return "#!/bin/sh\n# build " + ver + "\nprintf '%s\\n' \"$0\" > '" + out + ".tmp' && mv '" + out + ".tmp' '" + out + "'\nexec sleep 30\n"
+}
+
+func stopProc(p Proc) {
+ if p == nil {
+ return
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ _ = p.Stop(ctx)
+}
+
+// ---- D5: StartWorkerStaged dispatches on layout ------------------------
+
+// D5: the nested arm names the staged build the first half resolved to
+// root, never the live link, and root binds exactly that file; the proc is
+// the worker role's unit.
+func TestStartWorkerStagedNestedNamesStagedBuild(t *testing.T) {
+ h := newWorkerHarness(t, nil)
+ out := filepath.Join(h.root, "argv0")
+ id := h.stage("v1", argv0Script("v1", out))
+ p, got, err := h.b.StartWorkerStaged(context.Background(), h.rt, map[string]string{"CARLOS_SIDECAR_TOKEN": "tok"}, nil)
+ if err != nil {
+ t.Fatalf("StartWorkerStaged: %v\ncalls:\n%s", err, h.sd.joined())
+ }
+ defer stopProc(p)
+ up, ok := p.(*unitProc)
+ if !ok || up.role != instancectl.RoleWorker || up.unit != h.workerUnit() {
+ t.Fatalf("proc = %T %+v, want the worker unit's unitProc", p, p)
+ }
+ if got != id {
+ t.Fatalf("identity %+v, want the resolved staged build %+v", got, id)
+ }
+ // Root laid out the exe root's app as the staged file, not the link.
+ target, err := os.Readlink(filepath.Join(h.root, "exe", provision.InstanceID(h.host), "app"))
+ if err != nil || target != id.StagedPath {
+ t.Fatalf("root bound %q (%v), want the staged build %q", target, err, id.StagedPath)
+ }
+ if ran := strings.TrimSpace(waitForFile(t, out, 5*time.Second)); ran != id.StagedPath {
+ t.Fatalf("the unit ran %q, want the staged build %q", ran, id.StagedPath)
+ }
+ if calls := h.sd.joined(); !strings.Contains(calls, "start "+h.workerUnit()) || strings.Contains(calls, "carlos-instance@") {
+ t.Fatalf("systemd calls:\n%s\nwant the worker unit started and no run unit", calls)
+ }
+ if up.Exited() || up.Crashed() {
+ t.Fatal("a running unit reads as exited")
+ }
+}
+
+// D5: the ctx's exec gate is asked immediately before the start request, as
+// the nested instance start asks it: after the launch file is written, and
+// a refusal starts nothing, is launchRefused (not uncertain) and leaves no
+// listener bound.
+func TestStartWorkerStagedNestedAsksExecGateBeforeStart(t *testing.T) {
+ h := newWorkerHarness(t, nil)
+ h.stage("v1", argv0Script("v1", filepath.Join(h.root, "argv0")))
+ refused := errors.New("the lease is gone")
+ var startsAtGate int
+ var launchAtGate bool
+ gate := func() error {
+ startsAtGate = h.sd.startCount()
+ _, err := os.Stat(filepath.Join(h.instanceDir(), instancectl.LaunchFileName(instancectl.RoleWorker)))
+ launchAtGate = err == nil
+ return refused
+ }
+ published := 0
+ ctx := withBeforeExec(withExecGate(context.Background(), gate), func(release.StagedIdentity) error { published++; return nil })
+ p, _, err := h.b.StartWorkerStaged(ctx, h.rt, nil, nil)
+ if p != nil || !errors.Is(err, refused) {
+ t.Fatalf("StartWorkerStaged = %v, %v; want the gate's refusal and no proc", p, err)
+ }
+ if !launchNotStarted(err) || legacyStartMayRun(err) || startMayHaveRun(nil, err) {
+ t.Fatalf("a gate refusal must read as launchRefused and nothing else: %v", err)
+ }
+ if published != 1 {
+ t.Fatalf("published %d, want the build published once before the gate", published)
+ }
+ if startsAtGate != 0 || h.sd.startCount() != 0 || strings.Contains(h.sd.joined(), "start ") {
+ t.Fatalf("root was asked to start before or after the gate refused; calls:\n%s", h.sd.joined())
+ }
+ if !launchAtGate {
+ t.Fatal("the gate was asked before the launch file was written; it is asked immediately before the start request")
+ }
+ // Nothing left bound: the agent listener is closed, so a dial fails.
+ time.Sleep(50 * time.Millisecond)
+ if code := heartbeatOver(filepath.Join(h.b.pathsFor(h.rt).runDir, agentSocketName), h.host, "tok"); code != 0 {
+ t.Fatalf("the per-host agent listener still answers (%d) after a refused start", code)
+ }
+}
+
+// D4, D5: launch.worker.json carries the manifest's argv behind the live
+// link and the composed env: the tenant's layers with the agent-owned
+// variables on top (CARLOS_SIDECAR_DB names the row's database, and the
+// per-host CARLOS_AGENT_SOCKET names the listener in the run directory),
+// so a tenant's own CARLOS_SIDECAR_DB never wins.
+func TestLaunchWorkerJSONCarriesArgvAndEnv(t *testing.T) {
+ h := newWorkerHarness(t, nil)
+ h.stage("v1", argv0Script("v1", filepath.Join(h.root, "argv0")))
+ h.writeTenantEnv("CARLOS_SIDECAR_DB=/evil.db\nTENANT_VAR=kept\n")
+ extra := workerEnv(h.b.agentSocketFor(h.rt), h.host, "tok", h.rt.DBPath)
+ p, _, err := h.b.StartWorkerStaged(context.Background(), h.rt, extra, nil)
+ if err != nil {
+ t.Fatalf("StartWorkerStaged: %v", err)
+ }
+ defer stopProc(p)
+ l := h.readLaunch()
+ if l.Host != h.host || !slices.Equal(l.Argv, []string{h.b.liveLink(h.host), "work"}) {
+ t.Fatalf("launch argv %v (host %q), want [%s work]", l.Argv, l.Host, h.b.liveLink(h.host))
+ }
+ env := map[string]string{}
+ for _, kv := range l.Env {
+ k, v, _ := strings.Cut(kv, "=")
+ env[k] = v
+ }
+ wantSock := filepath.Join(h.b.pathsFor(h.rt).runDir, agentSocketName)
+ if env["CARLOS_SIDECAR_DB"] != h.rt.DBPath {
+ t.Fatalf("CARLOS_SIDECAR_DB = %q, want %q (agent-owned, over the tenant's /evil.db)", env["CARLOS_SIDECAR_DB"], h.rt.DBPath)
+ }
+ if env["CARLOS_AGENT_SOCKET"] != wantSock {
+ t.Fatalf("CARLOS_AGENT_SOCKET = %q, want the per-host listener %q", env["CARLOS_AGENT_SOCKET"], wantSock)
+ }
+ if env["CARLOS_SIDECAR_TOKEN"] != "tok" || env["CARLOS_SIDECAR_HOST"] != h.host || env["TENANT_VAR"] != "kept" {
+ t.Fatalf("env %v, want the token, the host and the tenant's own variable", env)
+ }
+ // The same file is what the fake systemd exec'd, so the child saw it.
+ if _, err := os.Stat(wantSock); err != nil {
+ t.Fatalf("per-host agent socket: %v", err)
+ }
+}
+
+// D5: a fenced or conflicting worker row is refused with the reason, before
+// anything is published, prepared or asked of root.
+func TestStartWorkerStagedFencedAndConflictRefuse(t *testing.T) {
+ for _, tc := range []struct {
+ name string
+ facts func(h *workerHarness) LayoutInputs
+ reason string
+ }{
+ {"fenced", func(h *workerHarness) LayoutInputs {
+ return LayoutInputs{Journal: func(host string) bool { return host == h.host }}
+ }, "layout migration in flight"},
+ {"conflict", func(h *workerHarness) LayoutInputs {
+ return LayoutInputs{Peers: func() ([]registry.Route, error) {
+ return []registry.Route{h.rt, {Host: "other.example.com", Kind: "instance", Hibernate: true,
+ DBPath: provision.NestedDBPath(h.b.DataDir, "other.example.com"),
+ Socket: filepath.Join(h.b.RunDirOverride, "instances", provision.InstanceID("other.example.com"), "app.sock")}}, nil
+ }, SameID: func(string, string) bool { return true }}
+ }, "collides with other.example.com"},
+ } {
+ t.Run(tc.name, func(t *testing.T) {
+ h := newWorkerHarness(t, nil)
+ h.stage("v1", argv0Script("v1", filepath.Join(h.root, "argv0")))
+ h.b.LayoutFacts = tc.facts(h)
+ if l, _ := h.b.Layout(h.rt); l != LayoutFenced && l != LayoutConflict {
+ t.Fatalf("fixture classifies as %s", l)
+ }
+ published := 0
+ ctx := withBeforeExec(context.Background(), func(release.StagedIdentity) error { published++; return nil })
+ p, _, err := h.b.StartWorkerStaged(ctx, h.rt, nil, nil)
+ if p != nil || err == nil || !strings.Contains(err.Error(), tc.reason) {
+ t.Fatalf("StartWorkerStaged = %v, %v; want a refusal naming %q", p, err, tc.reason)
+ }
+ if legacyStartMayRun(err) || startMayHaveRun(nil, err) {
+ t.Fatalf("a refusal reads as uncertain: %v", err)
+ }
+ if published != 0 || h.sd.joined() != "" {
+ t.Fatalf("published %d, systemd calls %q; want nothing", published, h.sd.joined())
+ }
+ })
+ }
+}