amadan
Add StartWorker: a spawn path for a child that binds nothing
5 files changed,
+176
−0
internal/activator/activator.go+8 −0internal/activator/activator_test.go+103 −0internal/activator/backend_exec.go+47 −0internal/activator/backend_unit.go+9 −0internal/activator/backend_unit_test.go+9 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 2ec94c275..086540b33 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -164,6 +164,14 @@ type Backend interface { |
| // pure outbound client with no socket, so there's nothing to health-check — |
| // its own connect/backoff loop recovers from a slow start. |
| StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) |
| + // StartWorker spawns a child that binds NOTHING: no socket, no port. No |
| + // health probe is possible or wanted — readiness is the caller's settle |
| + // window, or the child's own call over the agent socket. rt.Args (a JSON |
| + // argv array) names what to run; empty or unparseable is refused rather |
| + // than exec'd bare. env is merged over the route's own env layers and |
| + // wins conflicts — it carries the per-wake sidecar token, which the |
| + // tenant's own env file must never be able to override. |
| + StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) |
| } |
| // DueStore persists the due-time index (Eleven's actDueStore) — the registry |
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go| index 62bad5c92..af752ae12 100644 |
| --- a/internal/activator/activator_test.go |
| +++ b/internal/activator/activator_test.go |
| @@ -170,6 +170,14 @@ type fakeBackend struct { |
| sidecarErr error |
| sidecarStarts int |
| sidecars []*fakeProc |
| + // Worker spawn (StartWorker) — mirrors the sidecar fields above for the |
| + // same reason: workerStaged... no, a worker has no HasSidecar-style probe |
| + // (Task 6's territory); workerErr injects spawn failures, workerEnvs |
| + // records what env each call was handed so a caller-side test can assert |
| + // the per-wake token actually reached the backend. |
| + workerErr error |
| + workers []*fakeProc |
| + workerEnvs []map[string]string |
| // order interleaves backend calls and proc stops (start:instance, |
| // start:sidecar, nextDue, stop:sidecar, stop:instance) for the ordering |
| // assertions. |
| @@ -288,6 +296,19 @@ func (b *fakeBackend) setSidecarErr(err error) { |
| b.sidecarErr = err |
| } |
| +func (b *fakeBackend) StartWorker(_ context.Context, _ registry.Route, env map[string]string) (Proc, string, error) { |
| + b.mu.Lock() |
| + defer b.mu.Unlock() |
| + b.order = append(b.order, "start:worker") |
| + b.workerEnvs = append(b.workerEnvs, env) |
| + if b.workerErr != nil { |
| + return nil, "", b.workerErr |
| + } |
| + p := &fakeProc{label: "worker", rec: b} |
| + b.workers = append(b.workers, p) |
| + return p, "", nil |
| +} |
| + |
| func (b *fakeBackend) setDue(v int64) { |
| b.mu.Lock() |
| defer b.mu.Unlock() |
| @@ -2332,6 +2353,88 @@ func TestExecBackendHasSidecar(t *testing.T) { |
| } |
| } |
| +// StartWorker spawns a child that binds nothing — no socket, no port — so it |
| +// must return the moment the exec succeeds, never waiting on waitHealthyCtx |
| +// (which would dial unix://"" every 150ms for the full spawnHealthTimeout, |
| +// 30s by default in this test, and still fail). The assertion is on elapsed |
| +// wall time against a child that sleeps well past any reasonable spawn |
| +// latency, specifically so a regression that reintroduces a health wait |
| +// fails this test instead of merely running slow. |
| +func TestExecBackendStartWorkerSpawnsWithoutAnyHealthProbe(t *testing.T) { |
| + b := testExecBackend(t) |
| + host := "bot.x.internal" |
| + stageSeedScript(t, b, host, "v1", "sleep 30\n") |
| + |
| + rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`, |
| + DBPath: filepath.Join(b.DataDir, "bot.db")} |
| + |
| + start := time.Now() |
| + p, _, err := b.StartWorker(context.Background(), rt, map[string]string{"CARLOS_SIDECAR_NAME": "bot"}) |
| + if err != nil { |
| + t.Fatalf("StartWorker: %v", err) |
| + } |
| + defer p.Stop(context.Background()) |
| + if d := time.Since(start); d > 5*time.Second { |
| + t.Fatalf("StartWorker took %v — it waited for a health probe that can never answer", d) |
| + } |
| + if p.Exited() { |
| + t.Fatal("worker exited immediately") |
| + } |
| +} |
| + |
| +// A route with no runnable argv has nothing to exec — refusing beats |
| +// spawning the bare binary with no arguments (which for most binaries is |
| +// either a no-op forever or an immediate usage-error crash loop). |
| +func TestExecBackendStartWorkerRefusesEmptyArgs(t *testing.T) { |
| + b := testExecBackend(t) |
| + if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "b.x.internal", Kind: "sidecar"}, nil); err == nil { |
| + t.Fatal("a worker with no argv has nothing to run — refuse rather than exec the bare binary") |
| + } |
| +} |
| + |
| +// Malformed (not valid JSON) argv is refused exactly like empty argv, not |
| +// silently coerced into "run nothing" or "run the bare binary". |
| +func TestExecBackendStartWorkerRefusesMalformedArgs(t *testing.T) { |
| + b := testExecBackend(t) |
| + if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "b.x.internal", Kind: "sidecar", Args: "not json"}, nil); err == nil { |
| + t.Fatal("malformed args was silently accepted") |
| + } |
| +} |
| + |
| +// The env argument is merged OVER the route's own env layers and wins every |
| +// conflict: it carries the per-wake sidecar token, which a tenant's own env |
| +// file must never be able to override. |
| +func TestExecBackendStartWorkerEnvWinsOverTenantFile(t *testing.T) { |
| + b := testExecBackend(t) |
| + host := "worker-env.x.internal" |
| + _, envPath := stageRecordingLiveFullEnv(t, b, host, "v1") |
| + |
| + ownEnv := "CARLOS_SIDECAR_TOKEN=tenant-forged\nSHARED=instance\n" |
| + if err := os.WriteFile(filepath.Join(b.DataDir, host+".env"), []byte(ownEnv), 0o600); err != nil { |
| + t.Fatal(err) |
| + } |
| + |
| + rt := registry.Route{Host: host, Kind: "sidecar", Args: `["sidecar","run"]`} |
| + p, _, err := b.StartWorker(context.Background(), rt, map[string]string{"CARLOS_SIDECAR_TOKEN": "agent-issued"}) |
| + if err != nil { |
| + t.Fatalf("StartWorker: %v", err) |
| + } |
| + defer p.Stop(context.Background()) |
| + |
| + child := map[string]string{} |
| + for _, line := range strings.Split(waitForFile(t, envPath, time.Second), "\n") { |
| + if k, v, ok := strings.Cut(line, "="); ok { |
| + child[k] = v |
| + } |
| + } |
| + if got := child["CARLOS_SIDECAR_TOKEN"]; got != "agent-issued" { |
| + t.Fatalf("CARLOS_SIDECAR_TOKEN = %q, want %q — the agent-issued token must win over the tenant's own env file", got, "agent-issued") |
| + } |
| + if got := child["SHARED"]; got != "instance" { |
| + t.Fatalf("SHARED = %q, want %q — the route's own env layers must still flow through", got, "instance") |
| + } |
| +} |
| + |
| // stageRecordingLive stages host's live binary as a shell script that dumps |
| // its argv and its CARLOS_CONTROL_SOCKET to two files and exits — the only way |
| // to see what the child was ACTUALLY handed, argv and environment both. |
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go| index e0679120e..39b24acc4 100644 |
| --- a/internal/activator/backend_exec.go |
| +++ b/internal/activator/backend_exec.go |
| @@ -772,6 +772,53 @@ func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc, |
| 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. |
| +func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra map[string]string) (Proc, string, 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) |
| + } |
| + if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| + return nil, "", err |
| + } |
| + // 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.Host) |
| + if err != nil { |
| + return nil, "", err |
| + } |
| + for k, v := range extra { |
| + fileEnv = withVar(fileEnv, k, v) |
| + } |
| + |
| + cmd, err := b.Cgroups.Start(rt.Host, func() *exec.Cmd { |
| + cmd := exec.Command(b.liveLink(rt.Host), argv...) |
| + cmd.Env = childEnv(fileEnv) |
| + b.teeStd(cmd, rt) |
| + return cmd |
| + }) |
| + if err != nil { |
| + return nil, "", err |
| + } |
| + p := &childProc{cmd: cmd, done: make(chan struct{})} |
| + go func() { |
| + p.err = cmd.Wait() |
| + b.Cgroups.CleanupHost(rt.Host) |
| + close(p.done) |
| + }() |
| + return p, spawnVer, nil |
| +} |
| + |
| // socketClient returns an HTTP client that dials an instance's unix socket |
| // (request URLs use the placeholder host http://carlos). |
| func socketClient(socket string) *http.Client { |
diff --git a/internal/activator/backend_unit.go b/internal/activator/backend_unit.go| index 992e5423f..021856d8d 100644 |
| --- a/internal/activator/backend_unit.go |
| +++ b/internal/activator/backend_unit.go |
| @@ -187,6 +187,12 @@ func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error |
| return nil, errors.New("unit-backed tenants have no agent-managed sidecars") |
| } |
| +// StartWorker: worker spawn is exec-child machinery exactly like sidecar |
| +// co-spawn — a unit tenant that wants a worker runs it inside its own unit. |
| +func (b *UnitBackend) StartWorker(context.Context, registry.Route, map[string]string) (Proc, string, error) { |
| + return nil, "", errors.New("unit-backed tenants have no agent-managed workers") |
| +} |
| + |
| // unitProc adapts the unit to the Proc seam. |
| type unitProc struct { |
| b *UnitBackend |
| @@ -304,6 +310,9 @@ func (b *SplitBackend) HasSidecar(rt registry.Route) bool { return b.pick(rt).Ha |
| func (b *SplitBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) { |
| return b.pick(rt).StartSidecar(ctx, rt) |
| } |
| +func (b *SplitBackend) StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error) { |
| + return b.pick(rt).StartWorker(ctx, rt, env) |
| +} |
| // 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 |
diff --git a/internal/activator/backend_unit_test.go b/internal/activator/backend_unit_test.go| index 89494ab29..5d87bc3f1 100644 |
| --- a/internal/activator/backend_unit_test.go |
| +++ b/internal/activator/backend_unit_test.go |
| @@ -302,6 +302,15 @@ func TestUnitBackendInertSeams(t *testing.T) { |
| } |
| } |
| +// Worker spawn is exec-child machinery, exactly like sidecar co-spawn: a |
| +// unit tenant that wants a worker runs it inside its own unit. |
| +func TestUnitBackendRefusesWorkers(t *testing.T) { |
| + var b UnitBackend |
| + if _, _, err := b.StartWorker(context.Background(), registry.Route{Host: "x"}, nil); err == nil { |
| + t.Fatal("unit-backed tenants have no agent-managed workers, exactly as they have no agent-managed sidecars") |
| + } |
| +} |
| + |
| // LiveVersion reads the same live/<host> symlink layout as ExecBackend, so |
| // build convergence cycles awake unit tenants onto shipped builds. |
| func TestUnitBackendLiveVersion(t *testing.T) { |