amadan

carlos / platform Public

activator: the nested spawn path: prepare, launch file, unitProc and the exit observer (plan 3, task 4)

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Paul Campbell pushed by paul@keymail.dev 48d084c419b129f1f08023b1df0be7d706b1c276 parent 637a13f
6 files changed, +1051 −58
  • internal/activator/backend_exec.go +372 −51
  • internal/activator/backend_unit.go +10 −5
  • internal/activator/paths.go +14 −2
  • internal/activator/prepare.go +162 −0
  • internal/activator/unitproc.go +216 −0
  • internal/activator/unitproc_test.go +277 −0

CI passed — run details

diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index 06c7a4573..7313baace 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -23,6 +23,7 @@ import (
"github.com/carlosframework/platform/internal/applog"
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -110,6 +111,36 @@ type ExecBackend struct {
// Layout supplies the box facts the classification of §8.1 needs beyond
// the row: journals and peers. Zero value: never fenced, never conflict.
Layout LayoutInputs
+ // Control is the root instance-control channel a nested wake talks to
+ // (design §5.3). Nil refuses every nested wake with "instance control
+ // channel not configured" rather than dialing nothing — a flat or
+ // foreign host never needs it, and a nil Control does not touch them.
+ Control Control
+ // UnitState answers `systemctl show` for the observer (design §6.5).
+ // Nil: systemctlUnitState, the real systemctl binary.
+ UnitState UnitState
+ // CgroupEvents reads a unit's cgroup.events for the observer and for a
+ // seed's cgroup-empty wait. Nil: cgroupEventsUnder("/sys/fs/cgroup"),
+ // or CgroupRoot when set.
+ CgroupEvents CgroupEvents
+ // CgroupRoot overrides where cgroup v2 is mounted, for the default
+ // CgroupEvents reader. Nil (empty): /sys/fs/cgroup.
+ CgroupRoot string
+ // NestedOK gates a nested wake on the box facts a unit test can't see:
+ // the isolation mounts, the control service's own /ready. Nil: no
+ // gate (every unit test and any box that predates the check). Wired in
+ // production by Task 6.
+ NestedOK func() error
+ // ObservePoll is how often the exit observer (unitproc.go) and a
+ // seed's cgroup-empty wait re-check unit state. 0: one second. A field
+ // so tests can shrink the wait without a package-global var shared
+ // across every test in the suite.
+ ObservePoll time.Duration
+ // RunDirOverride is test-only: when set, layoutInputs() classifies
+ // against it instead of provision.SocketDir and pathsFor resolves a
+ // nested run dir and log socket beneath it, so a test can exercise the
+ // nested layout without writing to /run/carlos.
+ RunDirOverride string
}
// execLogLineMax caps one teed log line (spec §5): longer is truncated with
@@ -328,21 +359,61 @@ var restoreReplica = lease.Restore
// if your own env file sets them, those two names being the agent's to
// author. A non-zero exit fails the wake:
// a half-seeded instance must not serve.
+// Seed dispatches by layout, the same way Start does: a nested row seeds
+// through its own launch file and unit, a fenced or conflicting row
+// refuses, and a flat or foreign row seeds through today's exec child.
func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
- host := rt.Host
+ p := b.pathsFor(rt)
+ switch p.layout {
+ case LayoutNested:
+ return b.seedUnit(ctx, rt)
+ case LayoutFenced, LayoutConflict:
+ return fmt.Errorf("%s: refusing to seed: %s", rt.Host, p.reason)
+ }
+ return b.seedChild(ctx, rt)
+}
+
+// seedArgv parses rt.Seed's JSON argv array and appends the --db flag, the
+// composition both spawn paths need identically (design §6.2) — an empty
+// declared seed (an empty array, len 0) returns a nil argv, which both
+// callers treat as "nothing to run" (doWake only calls Seed for a
+// non-empty rt.Seed; this stays defensive rather than assuming that).
+func seedArgv(rt registry.Route, dbPath string) ([]string, error) {
var seedArgs []string
if err := json.Unmarshal([]byte(rt.Seed), &seedArgs); err != nil {
// Never a silent skip: a mangled seed column must fail the wake
// loudly, not boot an empty instance that looks healthy.
- return fmt.Errorf("parse seed argv for %s (%q): %w", host, rt.Seed, err)
+ return nil, fmt.Errorf("parse seed argv for %s (%q): %w", rt.Host, rt.Seed, err)
}
if len(seedArgs) == 0 {
+ return nil, nil
+ }
+ return append(seedArgs, "--db", dbPath), nil
+}
+
+// removeSeedDB deletes a failed seed's own half-built database and its
+// WAL/SHM siblings — the cleanup both spawn paths run when, and only when,
+// seedOwnsDB says the database is the seed's own work (seedCreatedTheDB).
+// Shared so the two paths cannot drift on which files "the database" means.
+func removeSeedDB(dbPath string) {
+ for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
+ _ = os.Remove(p)
+ }
+}
+
+func (b *ExecBackend) seedChild(ctx context.Context, rt registry.Route) error {
+ host := rt.Host
+ paths := b.pathsFor(rt)
+ argv, err := seedArgv(rt, paths.dbPath)
+ if err != nil {
+ return err
+ }
+ if argv == nil {
return nil // defensive: doWake only calls us for a non-empty seed
}
if err := b.ensureLiveLink(ctx, host); err != nil {
return err
}
- paths := b.pathsFor(rt)
// Mint before the env is read, exactly as Start does — a seed is the FIRST
// thing a fresh instance runs, so this is usually where the token comes
// into existence, and instanceEnv's contract is that a seed sees the same
@@ -364,7 +435,6 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
ctx, cancel := context.WithTimeout(ctx, seedTimeout)
defer cancel()
- argv := append(seedArgs, "--db", dbPath)
var stderr capBuffer
log.Printf("activator: seeding %s: %s %s", host, b.liveLink(host), strings.Join(argv, " "))
// The seed shares the instance's carlos-exec@<host> cgroup: first-boot
@@ -432,9 +502,7 @@ func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error {
// database that wasn't there when we started is ours to remove, and
// under the lease nothing else can have written it.
if seedOwnsDB {
- for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} {
- _ = os.Remove(p)
- }
+ removeSeedDB(dbPath)
}
if tail := strings.TrimSpace(stderr.String()); tail != "" {
return fmt.Errorf("seeding %s: %w: %s", host, err, tail)
@@ -495,6 +563,106 @@ func (c *capBuffer) String() string {
return string(c.b)
}
+// seedUnit is Seed's nested body: the same contract as seedChild — the
+// merged env, the CARLOS_ADMIN_TOKEN mint, the cleanup on a failure this
+// call itself caused — run through the root instance-control channel
+// instead of a direct exec (design §6.2). Argv and env come from the same
+// seedArgv and withoutVars calls seedChild uses, so the two cannot drift.
+func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route) error {
+ host := rt.Host
+ if b.Control == nil {
+ return fmt.Errorf("%s: instance control channel not configured", host)
+ }
+ if err := b.ensureLiveLink(ctx, host); err != nil {
+ return err
+ }
+ p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSeed)
+ if err != nil {
+ return err
+ }
+ defer logs.Close()
+ dbPath := p.dbPath
+ seedArgs, err := seedArgv(rt, dbPath)
+ if err != nil {
+ return err
+ }
+ if seedArgs == nil {
+ return nil // defensive: doWake only calls us for a non-empty seed
+ }
+ argv := append([]string{b.liveLink(host)}, seedArgs...)
+ env := childEnv(withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE"))
+ if err := b.writeLaunch(p, instancectl.RoleSeed, host, argv, env); err != nil {
+ return err
+ }
+ // Whether the database existed BEFORE the seed ran decides whether a
+ // failed seed may clean up after itself below — seedChild's rule,
+ // unchanged for the unit path.
+ _, statErr := os.Stat(dbPath)
+ seedOwnsDB := seedCreatedTheDB(statErr)
+
+ sctx, cancel := context.WithTimeout(ctx, seedTimeout)
+ defer cancel()
+ log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " "))
+ res, err := b.Control.Seed(sctx, host, b.tier(rt).MemoryMax)
+ if err != nil {
+ if seedOwnsDB {
+ removeSeedDB(dbPath)
+ }
+ return fmt.Errorf("seed for %s: %w", host, err)
+ }
+ // §6.2: the result the control service reports is only trustworthy once
+ // the seed's cgroup is provably empty — systemd reporting the unit
+ // inactive is not the same fact as no process from it still running and
+ // possibly still writing the database this result is about to judge.
+ if err := b.waitSeedCgroupEmpty(ctx, rt); err != nil {
+ log.Printf("activator: %s: seed cgroup did not empty: %v", host, err)
+ }
+ if res.Result != "success" {
+ if seedOwnsDB {
+ removeSeedDB(dbPath)
+ }
+ return fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result)
+ }
+ log.Printf("activator: seeded %s", host)
+ return nil
+}
+
+// waitSeedCgroupEmpty polls the seed unit's cgroup.events until it reports
+// empty or actStopTimeout elapses, honouring ctx and b.ObservePoll like the
+// exit observer (unitproc.go). A timeout or a read error is returned to the
+// caller as a log line, not a failure: it is the deciding result's
+// confidence that is at stake here, not the wake itself.
+func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route) error {
+ events := b.CgroupEvents
+ if events == nil {
+ events = cgroupEventsUnder(b.cgroupRoot())
+ }
+ poll := b.ObservePoll
+ if poll <= 0 {
+ poll = time.Second
+ }
+ slice := instancectl.SliceName(rt.Host)
+ unit := instancectl.UnitName(instancectl.RoleSeed, rt.Host)
+ deadline := time.Now().Add(actStopTimeout)
+ for {
+ exists, populated, err := events(slice, unit)
+ if err != nil {
+ return err
+ }
+ if !exists || !populated {
+ return nil
+ }
+ if time.Now().After(deadline) {
+ return fmt.Errorf("processes remained after %s", actStopTimeout)
+ }
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case <-time.After(poll):
+ }
+ }
+}
+
// NextDue asks the (still-running) instance for its earliest pending scheduled
// delivery over its socket. The endpoint is admin-gated like every instance
// data endpoint, so present the instance's own admin token — read from the
@@ -551,7 +719,24 @@ func (b *ExecBackend) ensureLiveLink(ctx context.Context, host string) error {
return fmt.Errorf("no live binary for %s (%s missing and no Materialize hook)", host, bin)
}
+// Start dispatches a wake by layout (design §8.1): a nested row starts as a
+// unit through the root instance-control channel, a fenced or conflicting
+// row refuses outright, and a flat or foreign row runs today's exec child.
func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ p := b.pathsFor(rt)
+ switch p.layout {
+ case LayoutNested:
+ return b.startUnit(ctx, rt)
+ case LayoutFenced, LayoutConflict:
+ return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason)
+ }
+ return b.startChild(ctx, rt)
+}
+
+// startChild is Start's exec-child body (flat and foreign layouts):
+// exec of the versioned binary directly, exactly as before nested layouts
+// existed.
+func (b *ExecBackend) startChild(ctx context.Context, rt registry.Route) (Proc, string, error) {
host := rt.Host
paths := b.pathsFor(rt)
dbPath := paths.dbPath
@@ -650,22 +835,7 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// the rollout ordering harmless. It does not replace the socket channel;
// both are live for exec-backed instances, and converging them is deferred
// (spec §10.3).
- childVars := fileEnv
- if ctl != nil {
- childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", controlSocket)
- } else {
- childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET")
- }
- if b.Cache != nil {
- childVars = withVar(childVars, "CARLOS_PURGE", "header")
- } else {
- childVars = withoutVars(childVars, "CARLOS_PURGE")
- }
- if b.Region != "" {
- childVars = withVar(childVars, "CARLOS_REGION", b.Region)
- } else {
- childVars = withoutVars(childVars, "CARLOS_REGION")
- }
+ childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil)
// Record the build we're about to exec — read the symlink immediately before
// the start so it names the version the kernel will resolve. Reading it here,
// rather than back in wake() after the health wait, closes a real window: a
@@ -736,35 +906,136 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
_ = p.Stop(stopCtx)
return nil, "", err
}
- // The instance-socket mode is the PLATFORM's, not the app's (issue #103):
- // normalize to 0660 so a deployment's enrolled cross-user consumers (the
- // carlos-sockets group — group-inherited from the setgid /run/carlos,
- // see host.tmpfilesConf) can dial, no matter what the child's umask or
- // its own post-bind chmod produced — titogo tightened its socket to 0600
- // and stranded the sibling router that proxies to it. After the health
- // wait, so the app's own bind/chmod has already happened and this wins.
- //
- // The two guards make "only a socket THIS backend's child bound" true
- // rather than merely intended. waitHealthy tolerates a squatter (its own
- // contract — it only dials), so a child that lost the bind race with a
- // stray supervisor (EADDRINUSE, exits) can still see the health wait
- // succeed off the SQUATTER's answers; a dead child means the socket is
- // not ours to widen (that route is the #17 external-socket advisory's
- // territory). The Lstat keeps the chmod from ever following a
- // non-socket — a symlink planted at the route's path must not become a
- // mode change somewhere else. On a box with no consumers group the
- // socket's group is just `carlos`, so the wider mode grants nobody
- // anything. All three branches log and never fail the wake: a box with
- // no consumers must not lose a healthy wake to a normalization it
- // doesn't need.
- if p.Exited() {
+ b.normalizeSocketMode(rt, p)
+ return p, spawnVer, nil
+}
+
+// agentOwnedVars composes the child's agent-owned environment variables —
+// the control socket, the purge channel, the region — on top of fileEnv,
+// exactly as startChild did inline before the unit spawn path needed the
+// same composition. Extracted so the two paths cannot drift (design §7.2):
+// haveControl is whether this call's control listener actually bound
+// (childVars must follow the OUTCOME of the bind, not the intention — see
+// the long comment this replaced, still true here).
+func (b *ExecBackend) agentOwnedVars(rt registry.Route, fileEnv map[string]string, haveControl bool) map[string]string {
+ childVars := fileEnv
+ if haveControl {
+ childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", rt.Socket+".control")
+ } else {
+ childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET")
+ }
+ if b.Cache != nil {
+ childVars = withVar(childVars, "CARLOS_PURGE", "header")
+ } else {
+ childVars = withoutVars(childVars, "CARLOS_PURGE")
+ }
+ if b.Region != "" {
+ childVars = withVar(childVars, "CARLOS_REGION", b.Region)
+ } else {
+ childVars = withoutVars(childVars, "CARLOS_REGION")
+ }
+ return childVars
+}
+
+// normalizeSocketMode is the PLATFORM's 0660 on a spawned instance's socket
+// (issue #103), not the app's — extracted from startChild so the unit path
+// (whose own bind is the trampoline's, not this process's exec) applies the
+// identical rule rather than a second copy that can drift. So a deployment's
+// enrolled cross-user consumers (the carlos-sockets group — group-inherited
+// from the setgid /run/carlos, see host.tmpfilesConf) can dial, no matter
+// what the child's umask or its own post-bind chmod produced — titogo
+// tightened its socket to 0600 and stranded the sibling router that proxies
+// to it. Called after the health wait, so the app's own bind/chmod has
+// already happened and this wins.
+//
+// The two guards make "only a socket THIS backend's child bound" true
+// rather than merely intended. waitHealthy tolerates a squatter (its own
+// contract — it only dials), so a child that lost the bind race with a
+// stray supervisor (EADDRINUSE, exits) can still see the health wait
+// succeed off the SQUATTER's answers; a dead child means the socket is
+// not ours to widen (that route is the #17 external-socket advisory's
+// territory). The Lstat keeps the chmod from ever following a
+// non-socket — a symlink planted at the route's path must not become a
+// mode change somewhere else. On a box with no consumers group the
+// socket's group is just `carlos`, so the wider mode grants nobody
+// anything. All three branches log and never fail the wake: a box with
+// no consumers must not lose a healthy wake to a normalization it
+// doesn't need.
+func (b *ExecBackend) normalizeSocketMode(rt registry.Route, proc Proc) {
+ host := rt.Host
+ if proc.Exited() {
log.Printf("activator: %s: child exited during the health wait (external process serving its socket?) — not normalizing the socket mode", host)
} else if fi, statErr := os.Lstat(rt.Socket); statErr != nil || fi.Mode()&os.ModeSocket == 0 {
log.Printf("activator: %s: %s is not a socket we can normalize (stat: %v)", host, rt.Socket, statErr)
} else if err := os.Chmod(rt.Socket, 0o660); err != nil {
log.Printf("activator: normalize socket mode for %s: %v (enrolled consumers would get permission denied)", host, err)
}
- return p, spawnVer, nil
+}
+
+// startUnit is Start's nested body (design §6): the same admin token, env
+// composition and health gate as startChild, but the process itself is
+// started by the root instance-control channel from a launch file rather
+// than exec'd directly — the agent has no privilege to enter the tenant's
+// namespace, only the control service does.
+func (b *ExecBackend) startUnit(ctx context.Context, rt registry.Route) (Proc, string, error) {
+ if b.Control == nil {
+ return nil, "", fmt.Errorf("%s: instance control channel not configured", rt.Host)
+ }
+ if b.NestedOK != nil {
+ if err := b.NestedOK(); err != nil {
+ return nil, "", fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err)
+ }
+ }
+ if err := b.ensureLiveLink(ctx, rt.Host); err != nil {
+ return nil, "", err
+ }
+ p, ctl, logs, fileEnv, err := b.prepare(rt, instancectl.RoleRun)
+ if err != nil {
+ return nil, "", err
+ }
+ fail := func(err error) (Proc, string, error) {
+ if ctl != nil {
+ ctl.Close()
+ }
+ logs.Close()
+ return nil, "", err
+ }
+ childVars := b.agentOwnedVars(rt, fileEnv, ctl != nil)
+ argv := []string{b.liveLink(rt.Host), "--socket", rt.Socket, "--db", p.dbPath}
+ if err := b.writeLaunch(p, instancectl.RoleRun, rt.Host, argv, childEnv(childVars)); err != nil {
+ return fail(err)
+ }
+ // Read the symlink immediately before the start, same fail-safe
+ // direction as startChild: a concurrent repoint at worst records the
+ // older version (one extra convergence cycle), never the newer.
+ spawnVer, _ := b.LiveVersion(rt.Host)
+ inv, err := b.Control.Start(ctx, rt.Host, instancectl.RoleRun, b.tier(rt).MemoryMax)
+ if err != nil {
+ return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err))
+ }
+ if inv == "" {
+ // The server read the invocation after the start and found none:
+ // the unit already exited (plan 2 final review). Nothing is
+ // running, so there is nothing to stop; the wake failed.
+ return fail(fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host)))
+ }
+ log.Printf("activator: %s: started %s memory.max=%d", rt.Host, instancectl.UnitName(instancectl.RoleRun, rt.Host), b.tier(rt).MemoryMax)
+ up := b.newUnitProc(rt, instancectl.RoleRun, p, inv, ctl, logs)
+ if b.Cache != nil {
+ // A fresh process for this host is exactly the moment its cached
+ // content becomes suspect, same rule as startChild.
+ if n := b.Cache.PurgeHost(rt.Host); n > 0 {
+ log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host)
+ }
+ }
+ if err := waitHealthyCtx(ctx, rt.Socket, spawnHealthTimeout); err != nil {
+ stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
+ defer cancel()
+ _ = up.Stop(stopCtx)
+ return nil, "", err
+ }
+ b.normalizeSocketMode(rt, up)
+ return up, spawnVer, nil
}
// instanceEnv composes the host's two EnvironmentFile layers, both optional
@@ -779,11 +1050,20 @@ func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
// environment" includes the same secret.)
func (b *ExecBackend) instanceEnv(rt registry.Route) (map[string]string, error) {
paths := b.pathsFor(rt)
- cfgEnv, err := readEnvFile(paths.configEnv)
+ // A nested host's files live under a tenant-writable directory (design
+ // §3.2): read them without following a symlink. A flat host's files sit
+ // outside any tenant's reach, so the plain read stays exactly as it was.
+ read := readEnvFile
+ if paths.layout == LayoutNested {
+ read = func(path string) (map[string]string, error) {
+ return readEnvFileBeneath(filepath.Dir(path), filepath.Base(path))
+ }
+ }
+ cfgEnv, err := read(paths.configEnv)
if err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("read config env file: %w", err)
}
- ownEnv, err := readEnvFile(paths.envFile)
+ ownEnv, err := read(paths.envFile)
if err != nil && !os.IsNotExist(err) {
return nil, fmt.Errorf("read env file: %w", err)
}
@@ -888,13 +1168,25 @@ func (b *ExecBackend) HasSidecar(rt registry.Route) bool {
return err == nil
}
-// StartSidecar execs the host's live symlink with `sidecar run` under the
-// sidecar's EnvironmentFile, CARLOS_*-clean like the instance spawn. No
+// StartSidecar dispatches by layout, the same way Start does.
+func (b *ExecBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) {
+ p := b.pathsFor(rt)
+ switch p.layout {
+ case LayoutNested:
+ return b.startSidecarUnit(ctx, rt)
+ case LayoutFenced, LayoutConflict:
+ return nil, fmt.Errorf("%s: refusing to start sidecar: %s", rt.Host, p.reason)
+ }
+ return b.startSidecarChild(ctx, rt)
+}
+
+// startSidecarChild execs the host's live symlink with `sidecar run` under
+// the sidecar's EnvironmentFile, CARLOS_*-clean like the instance spawn. No
// health wait — the sidecar is a pure outbound client with no socket to
// probe; its own connect/backoff loop absorbs a slow instance. The live
// symlink exists by now: the instance spawn checked it. The ctx is unused (an
// exec doesn't block); it's here for the backend seam's symmetry.
-func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc, error) {
+func (b *ExecBackend) startSidecarChild(_ context.Context, rt registry.Route) (Proc, error) {
fileEnv, err := readEnvFile(b.pathsFor(rt).sidecarEnv)
if err != nil {
return nil, fmt.Errorf("read sidecar env: %w", err)
@@ -923,6 +1215,35 @@ func (b *ExecBackend) StartSidecar(_ context.Context, rt registry.Route) (Proc,
return p, 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
+// instance's two-layer config.
+func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error) {
+ if b.Control == nil {
+ return nil, fmt.Errorf("%s: instance control channel not configured", rt.Host)
+ }
+ p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSidecar)
+ if err != nil {
+ return nil, err
+ }
+ argv := []string{b.liveLink(rt.Host), "sidecar", "run"}
+ if err := b.writeLaunch(p, instancectl.RoleSidecar, rt.Host, argv, childEnv(fileEnv)); err != nil {
+ logs.Close()
+ return nil, err
+ }
+ inv, err := b.Control.Start(ctx, rt.Host, instancectl.RoleSidecar, b.tier(rt).MemoryMax)
+ if err != nil {
+ logs.Close()
+ return nil, fmt.Errorf("%s: start sidecar unit: %w", rt.Host, err)
+ }
+ if inv == "" {
+ logs.Close()
+ return nil, fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleSidecar, rt.Host))
+ }
+ return b.newUnitProc(rt, instancectl.RoleSidecar, p, inv, nil, logs), 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..4969187c8 100644
--- a/internal/activator/backend_unit.go
+++ b/internal/activator/backend_unit.go
@@ -155,7 +155,7 @@ func (b *UnitBackend) Start(ctx context.Context, rt registry.Route) (Proc, strin
_ = b.systemctl(stopCtx, "stop", unit)
return nil, "", err
}
- return &unitProc{b: b, unit: unit}, spawnVer, nil
+ return &svcUnitProc{b: b, unit: unit}, spawnVer, nil
}
func (b *UnitBackend) waitHealthy(ctx context.Context, rt registry.Route) error {
@@ -187,8 +187,13 @@ func (b *UnitBackend) StartSidecar(context.Context, registry.Route) (Proc, error
return nil, errors.New("unit-backed tenants have no agent-managed sidecars")
}
-// unitProc adapts the unit to the Proc seam.
-type unitProc struct {
+// svcUnitProc adapts a UnitBackend-managed systemd unit (carlos-app@<host>.service,
+// a whole-app tenant whose process manager stays systemd) to the Proc seam.
+// Named apart from the isolated-exec unitProc in unitproc.go, which is a
+// different thing entirely: this one wraps a tenant unit the operator
+// authored, that one is a nested exec instance the activator itself spawns
+// through the root instance-control channel.
+type svcUnitProc struct {
b *UnitBackend
unit string
@@ -198,7 +203,7 @@ type unitProc struct {
// Stop is `systemctl stop` — systemd runs the unit's own SIGTERM drain and
// the ctx bounds how long we wait for the job.
-func (p *unitProc) Stop(ctx context.Context) error {
+func (p *svcUnitProc) Stop(ctx context.Context) error {
err := p.b.systemctl(ctx, "stop", p.unit)
p.mu.Lock()
p.stopped = true
@@ -214,7 +219,7 @@ func (p *unitProc) Stop(ctx context.Context) error {
// wedged past its start rate limit looks awake until its idle window stops
// it (the edge's dials 502 meanwhile) — operator territory, exactly as it
// is for a non-hibernating unit today.
-func (p *unitProc) Exited() bool {
+func (p *svcUnitProc) Exited() bool {
p.mu.Lock()
defer p.mu.Unlock()
return p.stopped
diff --git a/internal/activator/paths.go b/internal/activator/paths.go
index e1fe4a40e..00d207288 100644
--- a/internal/activator/paths.go
+++ b/internal/activator/paths.go
@@ -28,7 +28,13 @@ func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
p := instancePaths{layout: l, reason: reason, dataDir: b.DataDir, runDir: provision.SocketDir}
if l == LayoutNested {
p.dataDir = provision.InstanceDir(b.DataDir, rt.Host)
- p.runDir = provision.InstanceRunDir(rt.Host)
+ if b.RunDirOverride != "" {
+ // Test-only: a test points the nested run dir at a temp dir
+ // rather than /run/carlos, exactly as it points DataDir at one.
+ p.runDir = filepath.Join(b.RunDirOverride, "instances", provision.InstanceID(rt.Host))
+ } else {
+ p.runDir = provision.InstanceRunDir(rt.Host)
+ }
}
p.dbPath = rt.DBPath
if p.dbPath == "" {
@@ -40,9 +46,15 @@ func (b *ExecBackend) pathsFor(rt registry.Route) instancePaths {
return p
}
-// layoutInputs is the backend's LayoutInputs with DataDir filled in.
+// layoutInputs is the backend's LayoutInputs with DataDir filled in. A
+// non-empty RunDirOverride also steers the socket dir Classify tests
+// against, so a route built against the override dir still classifies as
+// nested through Task 2's sockets() helper.
func (b *ExecBackend) layoutInputs() LayoutInputs {
in := b.Layout
in.DataDir = b.DataDir
+ if b.RunDirOverride != "" {
+ in.SocketDir = b.RunDirOverride
+ }
return in
}
diff --git a/internal/activator/prepare.go b/internal/activator/prepare.go
new file mode 100644
index 000000000..5322e0653
--- /dev/null
+++ b/internal/activator/prepare.go
@@ -0,0 +1,162 @@
+// prepare.go is everything a nested role needs before its unit starts
+// (isolated-exec design §6.1): the launch file's directories, the merged
+// environment, the control listener (run only) and the role's log
+// listener. It is the nested counterpart of what startChild/seedChild do
+// inline for a flat host.
+package activator
+
+import (
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/schedule"
+)
+
+// prepare is everything a nested role needs before its unit starts
+// (design §6.1): the two directories, the admin token and merged env (for
+// run and seed), the control listener (run only), the role's log listener,
+// and stale records cleared. It returns the listeners so the caller can
+// hand them to the proc or close them on a failed start.
+func (b *ExecBackend) prepare(rt registry.Route, role instancectl.Role) (instancePaths, *edge.ControlListener, *logListener, map[string]string, error) {
+ p := b.pathsFor(rt)
+ if err := os.MkdirAll(p.dataDir, 0o700); err != nil {
+ return p, nil, nil, nil, err
+ }
+ if err := os.MkdirAll(p.runDir, 0o750|os.ModeSetgid); err != nil {
+ return p, nil, nil, nil, err
+ }
+ if err := os.MkdirAll(filepath.Dir(b.logSocket(rt.Host, role)), 0o700); err != nil {
+ return p, nil, nil, nil, err
+ }
+ var env map[string]string
+ if role == instancectl.RoleRun || role == instancectl.RoleSeed {
+ if err := EnsureAdminToken(p.envFile); err != nil {
+ return p, nil, nil, nil, fmt.Errorf("admin token for %s: %w", rt.Host, err)
+ }
+ var err error
+ env, err = b.instanceEnv(rt)
+ if err != nil {
+ return p, nil, nil, nil, err
+ }
+ } else {
+ var err error
+ env, err = readEnvFileBeneath(p.dataDir, filepath.Base(p.sidecarEnv))
+ if err != nil {
+ return p, nil, nil, nil, err
+ }
+ }
+ // Stale exit records from earlier invocations of this role: a start
+ // that reads one would mis-word the very first exit line this
+ // invocation eventually gets, attributing a previous run's status to
+ // this one.
+ if entries, err := os.ReadDir(p.runDir); err == nil {
+ for _, e := range entries {
+ if strings.HasPrefix(e.Name(), "exit."+string(role)+".") {
+ _ = os.Remove(filepath.Join(p.runDir, e.Name()))
+ }
+ }
+ }
+ var ctl *edge.ControlListener
+ if role == instancectl.RoleRun {
+ // No pre-emptive removal of a stale rt.Socket here (deviation from
+ // an earlier draft of this method): startChild does not do this
+ // for a flat host either — a leftover socket file fails the
+ // SUBSEQUENT bind with "address already in use", which
+ // memberSafeWakeReason already recognises as the stale-socket wake
+ // trap (#129) and reports as a diagnosis rather than a symptom.
+ // Removing it here would silently paper over exactly the case #129
+ // exists to surface: two invocations of the same host actually
+ // live at once. The two spawn paths stay consistent on this by
+ // construction rather than by a second copy of the same rule.
+ ctl = b.listenControl(rt, p.envFile)
+ }
+ logs, err := listenLogs(b.logSocket(rt.Host, role), b.logEmitter(rt))
+ if err != nil {
+ if ctl != nil {
+ ctl.Close()
+ }
+ return p, nil, nil, nil, fmt.Errorf("log listener for %s %s: %w", rt.Host, role, err)
+ }
+ return p, ctl, logs, env, nil
+}
+
+// listenControl binds the control socket exactly as startChild does for an
+// exec child (the tokenReader and schedule mount are unchanged, aside from
+// reading beneath the instance directory rather than following a plain
+// path — design §3.2); a bind failure is logged and the instance runs
+// without its control verbs, as today.
+func (b *ExecBackend) listenControl(rt registry.Route, envPath string) *edge.ControlListener {
+ tokenReader := func() (string, error) {
+ env, err := readEnvFileBeneath(filepath.Dir(envPath), filepath.Base(envPath))
+ if err != nil {
+ return "", err
+ }
+ return env["CARLOS_ADMIN_TOKEN"], nil
+ }
+ var mount func(*http.ServeMux)
+ if b.Timers != nil {
+ mount = func(mux *http.ServeMux) { schedule.Mount(mux, rt.Host, b.Timers, tokenReader, time.Now) }
+ }
+ ctl, err := edge.ListenControl(rt.Socket+".control", rt.Host, b.Cache, mount)
+ if err != nil {
+ log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err)
+ return nil
+ }
+ return ctl
+}
+
+// logSocket is the role's log socket path, honouring the test override of
+// the run directory (RunDirOverride).
+func (b *ExecBackend) logSocket(host string, role instancectl.Role) string {
+ if b.RunDirOverride != "" {
+ return filepath.Join(b.RunDirOverride, "logs", provision.InstanceID(host)+"."+string(role)+".sock")
+ }
+ return provision.LogSocket(host, string(role))
+}
+
+// logEmitter is the same sink teeStd feeds, or nil when Logs is unset.
+func (b *ExecBackend) logEmitter(rt registry.Route) func(string) {
+ if b.Logs == nil {
+ return nil
+ }
+ return func(line string) { b.Logs.App(release.RouteAccount(rt), rt.App, rt.Host, line) }
+}
+
+// writeLaunch writes launch.<role>.json atomically into the instance
+// directory (design §6.2). The env is childEnv's deduplicated list; the
+// trampoline execs it as is.
+func (b *ExecBackend) writeLaunch(p instancePaths, role instancectl.Role, host string, argv, env []string) error {
+ raw, err := json.MarshalIndent(instancectl.Launch{Schema: instancectl.LaunchSchema, Host: host, Argv: argv, Env: env}, "", " ")
+ if err != nil {
+ return err
+ }
+ return instancefs.WriteAtomic(p.dataDir, instancectl.LaunchFileName(role), append(raw, '\n'), 0o600)
+}
+
+// readEnvFileBeneath is readEnvFile through instancefs: a nested host's env
+// files are read without following a symlink (design §3.2). A missing file
+// is an empty env, as readEnvFile treats it (the caller's own
+// os.IsNotExist check would otherwise have nothing to check against).
+func readEnvFileBeneath(dir, name string) (map[string]string, error) {
+ b, err := instancefs.ReadBeneathLimit(dir, name, 1<<20)
+ if errors.Is(err, os.ErrNotExist) {
+ return map[string]string{}, nil
+ }
+ if err != nil {
+ return nil, err
+ }
+ return parseEnvBytes(b), nil
+}
diff --git a/internal/activator/unitproc.go b/internal/activator/unitproc.go
new file mode 100644
index 000000000..05f81d613
--- /dev/null
+++ b/internal/activator/unitproc.go
@@ -0,0 +1,216 @@
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "sync/atomic"
+ "time"
+
+ "github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/instancefs"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// ErrStillRunning is Stop's answer when ctx expired before termination was
+// proven (design §6.3). The activator then keeps the lease and the entry
+// in stopping, and the sweeper retries; the lease is never released over a
+// writer that may still exist.
+var ErrStillRunning = errors.New("activator: stop unproven; processes remain")
+
+// unitProc is a nested host's Proc (design §6.3): the unit the control
+// service started, the listeners the agent holds for it, and the observer
+// that proves termination. Named apart from backend_unit.go's svcUnitProc,
+// a whole-app tenant's own systemd unit — this one is a nested exec
+// instance the activator spawns through the root instance-control channel.
+type unitProc struct {
+ b *ExecBackend
+ rt registry.Route
+ role instancectl.Role
+ unit, slice string
+ invocation string
+ runDir string
+ ctl *edge.ControlListener
+ logs *logListener
+ exited atomic.Bool
+ done chan struct{}
+ closeOnce atomic.Bool
+}
+
+func (b *ExecBackend) newUnitProc(rt registry.Route, role instancectl.Role, p instancePaths, inv string, ctl *edge.ControlListener, logs *logListener) *unitProc {
+ up := &unitProc{b: b, rt: rt, role: role, unit: instancectl.UnitName(role, rt.Host), slice: instancectl.SliceName(rt.Host),
+ invocation: inv, runDir: p.runDir, ctl: ctl, logs: logs, done: make(chan struct{})}
+ go b.observe(up)
+ return up
+}
+
+func (p *unitProc) Exited() bool { return p.exited.Load() }
+
+// Stop asks the control service to stop the unit and waits for termination
+// to be proven. The control listener and the log listener are closed only
+// after that, preserving childProc's ordering (signal first, close the
+// listener after the process is gone).
+func (p *unitProc) Stop(ctx context.Context) error {
+ if !p.Exited() {
+ sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineControl)
+ err := p.b.Control.Stop(sctx, p.rt.Host, p.role)
+ cancel()
+ if err != nil {
+ log.Printf("activator: stop %s %s: %v", p.rt.Host, p.role, err)
+ }
+ }
+ select {
+ case <-p.done:
+ p.closeListeners()
+ return nil
+ case <-ctx.Done():
+ return ErrStillRunning
+ }
+}
+
+func (p *unitProc) closeListeners() {
+ if p.closeOnce.Swap(true) {
+ return
+ }
+ if p.ctl != nil {
+ p.ctl.Close()
+ }
+ if p.logs != nil {
+ p.logs.Close()
+ }
+}
+
+// observe is the goroutine of design §6.5. It proves termination from two
+// facts the tenant cannot forge: systemd's unit state and cgroup emptiness.
+// The exit record decides the wording of the event line and nothing else.
+func (b *ExecBackend) observe(p *unitProc) {
+ poll := b.ObservePoll
+ if poll <= 0 {
+ poll = time.Second
+ }
+ state := b.UnitState
+ if state == nil {
+ state = systemctlUnitState
+ }
+ events := b.CgroupEvents
+ if events == nil {
+ events = cgroupEventsUnder(b.cgroupRoot())
+ }
+ foreignStopped := false
+ lastRemain := time.Time{}
+ for {
+ ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+ active, inv, err := state(ctx, p.unit)
+ cancel()
+ if err != nil {
+ log.Printf("activator: %s: unit state: %v", p.rt.Host, err)
+ }
+ terminal := false
+ foreign := false
+ switch {
+ case err != nil:
+ case inv != "" && inv != p.invocation:
+ foreign = true
+ case active == "inactive" || active == "failed":
+ terminal = true
+ }
+ if foreign && !foreignStopped {
+ log.Printf("activator: %s: unit restarted outside the activator (invocation %s, ours %s); stopping it", p.rt.Host, inv, p.invocation)
+ sctx, scancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
+ if err := b.Control.Stop(sctx, p.rt.Host, p.role); err != nil {
+ log.Printf("activator: %s: stop foreign invocation: %v", p.rt.Host, err)
+ }
+ scancel()
+ foreignStopped = true
+ }
+ if terminal || (foreign && foreignStopped) {
+ exists, populated, eerr := events(p.slice, p.unit)
+ if eerr != nil {
+ log.Printf("activator: %s: cgroup.events: %v", p.rt.Host, eerr)
+ }
+ if !exists || !populated {
+ if terminal {
+ emitEvent(b.Logs, p.rt, exitLevel(p), exitLine(p))
+ } else {
+ emitEvent(b.Logs, p.rt, "warn", "instance exited (status unknown: restarted outside the activator)")
+ }
+ p.exited.Store(true)
+ // Stop closes the listeners when it observes done; a crash
+ // nobody is stopping yet leaves them for the sweeper's
+ // stand-down, which calls Stop.
+ close(p.done)
+ return
+ }
+ if time.Since(lastRemain) > 2*time.Second {
+ log.Printf("activator: %s: processes remain after %s; not releasing", p.rt.Host, active)
+ lastRemain = time.Now()
+ }
+ }
+ time.Sleep(poll)
+ }
+}
+
+// exitLine reads the advisory record for this invocation and words the
+// event line as childProc does: a non-zero status is a warn line, a clean
+// exit is info, and no record is unknown. The record is tenant-writable
+// text (design §6.5, final review of plan 2), so every field that reaches
+// the line passes through recordToken first.
+func exitLine(p *unitProc) string {
+ rec, ok := p.readRecord()
+ switch {
+ case !ok:
+ return "instance exited (status unknown: no exit record)"
+ case rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0"):
+ return "instance exited (exit 0)"
+ case rec.Code == "killed" || rec.Code == "dumped":
+ return "instance exited: signal " + recordToken(rec.Status)
+ default:
+ return fmt.Sprintf("instance exited: exit %s (%s)", recordToken(rec.Status), recordToken(rec.Result))
+ }
+}
+
+// recordToken bounds a field of the exit record to what systemd itself
+// would write: at most 32 bytes of [A-Za-z0-9_-]. Anything else becomes
+// "?" so a tenant cannot put arbitrary text or control characters into
+// the platform's event stream.
+func recordToken(s string) string {
+ if len(s) == 0 || len(s) > 32 {
+ return "?"
+ }
+ for _, c := range s {
+ if !(c >= 'a' && c <= 'z' || c >= 'A' && c <= 'Z' || c >= '0' && c <= '9' || c == '_' || c == '-') {
+ return "?"
+ }
+ }
+ return s
+}
+
+func exitLevel(p *unitProc) string {
+ if rec, ok := p.readRecord(); ok && (rec.Result == "success" || (rec.Code == "exited" && rec.Status == "0")) {
+ return "info"
+ }
+ return "warn"
+}
+
+func (p *unitProc) readRecord() (instancectl.ExitRecord, bool) {
+ raw, err := instancefs.ReadBeneathLimit(p.runDir, instancectl.ExitRecordName(p.role, p.invocation), 4096)
+ if err != nil {
+ return instancectl.ExitRecord{}, false
+ }
+ var rec instancectl.ExitRecord
+ if err := json.Unmarshal(raw, &rec); err != nil {
+ return instancectl.ExitRecord{}, false
+ }
+ return rec, true
+}
+
+// cgroupRoot is CgroupRoot with the nil default applied.
+func (b *ExecBackend) cgroupRoot() string {
+ if b.CgroupRoot != "" {
+ return b.CgroupRoot
+ }
+ return "/sys/fs/cgroup"
+}
diff --git a/internal/activator/unitproc_test.go b/internal/activator/unitproc_test.go
new file mode 100644
index 000000000..755bcb0e0
--- /dev/null
+++ b/internal/activator/unitproc_test.go
@@ -0,0 +1,277 @@
+package activator
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "net"
+ "net/http"
+ "os"
+ "path/filepath"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/instancectl"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// fakeControl records verbs and lets a test drive the unit's state.
+type fakeControl struct {
+ mu sync.Mutex
+ calls []string
+ inv string
+ seed instancectl.SeedResponse
+ startErr error
+ stopHook func(host string, role instancectl.Role)
+ readyErr error
+}
+
+func (f *fakeControl) rec(s string) { f.mu.Lock(); f.calls = append(f.calls, s); f.mu.Unlock() }
+func (f *fakeControl) Start(_ context.Context, host string, role instancectl.Role, max int64) (string, error) {
+ f.rec("start " + host + " " + string(role))
+ return f.inv, f.startErr
+}
+func (f *fakeControl) Seed(_ context.Context, host string, max int64) (instancectl.SeedResponse, error) {
+ f.rec("seed " + host)
+ return f.seed, nil
+}
+func (f *fakeControl) Stop(_ context.Context, host string, role instancectl.Role) error {
+ f.rec("stop " + host + " " + string(role))
+ if f.stopHook != nil {
+ f.stopHook(host, role)
+ }
+ return nil
+}
+func (f *fakeControl) Reset(_ context.Context, host string, role instancectl.Role) error {
+ f.rec("reset")
+ return nil
+}
+func (f *fakeControl) Sweep(context.Context) (instancectl.SweepResponse, error) {
+ f.rec("sweep")
+ return instancectl.SweepResponse{}, nil
+}
+func (f *fakeControl) Ready(context.Context) error { f.rec("ready"); return f.readyErr }
+func (f *fakeControl) joined() string {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ return strings.Join(f.calls, "\n")
+}
+
+// fakeUnit is the unit's observable state: what `systemctl show` and
+// cgroup.events would say.
+type fakeUnit struct {
+ mu sync.Mutex
+ active string
+ inv string
+ exists bool
+ populated bool
+}
+
+func (u *fakeUnit) set(active, inv string, exists, populated bool) {
+ u.mu.Lock()
+ u.active, u.inv, u.exists, u.populated = active, inv, exists, populated
+ u.mu.Unlock()
+}
+func (u *fakeUnit) state(context.Context, string) (string, string, error) {
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ return u.active, u.inv, nil
+}
+func (u *fakeUnit) events(string, string) (bool, bool, error) {
+ u.mu.Lock()
+ defer u.mu.Unlock()
+ return u.exists, u.populated, nil
+}
+
+// nestedBackend returns an ExecBackend wired for a nested host with fakes,
+// and the route. The data and run dirs are temp dirs standing in for
+// /data/instances/<iid> and /run/carlos/instances/<iid>; provision's
+// helpers are not used for the temp paths, so the route is built to
+// classify as nested by construction through the Layout seam.
+func nestedBackend(t *testing.T, ctl *fakeControl, unit *fakeUnit) (*ExecBackend, registry.Route) {
+ t.Helper()
+ root, err := os.MkdirTemp("", "nb")
+ if err != nil {
+ t.Fatal(err)
+ }
+ t.Cleanup(func() { os.RemoveAll(root) })
+ host := "h.example.com"
+ b := testExecBackend(t)
+ b.DataDir = filepath.Join(root, "data")
+ b.Control = ctl
+ b.UnitState = unit.state
+ b.CgroupEvents = unit.events
+ b.ObservePoll = 20 * time.Millisecond
+ b.RunDirOverride = filepath.Join(root, "run")
+ stageLive(t, b, host, "v1")
+ rt := registry.Route{Host: host, Kind: "instance", Hibernate: true,
+ DBPath: provision.NestedDBPath(b.DataDir, host), Socket: filepath.Join(b.RunDirOverride, "instances", host, "app.sock")}
+ return b, rt
+}
+
+func TestStartUnitWritesTheLaunchFileAndReturnsAUnitProc(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ // The fake control does not start anything, so health must be faked:
+ // bind a listener at rt.Socket that answers 200.
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, ver, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatalf("Start: %v", err)
+ }
+ _ = ver
+ up, ok := p.(*unitProc)
+ if !ok || up.invocation != "inv1" {
+ t.Fatalf("proc = %T %+v", p, p)
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.run.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l instancectl.Launch
+ if err := json.Unmarshal(raw, &l); err != nil {
+ t.Fatal(err)
+ }
+ if l.Schema != 1 || l.Host != rt.Host || len(l.Argv) != 5 || l.Argv[0] != filepath.Join(b.OptDir, "live", rt.Host) || l.Argv[2] != rt.Socket || l.Argv[4] != rt.DBPath {
+ t.Errorf("launch = %+v", l)
+ }
+ seenCtl := false
+ for _, kv := range l.Env {
+ if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") {
+ seenCtl = true
+ }
+ if strings.HasPrefix(kv, "CARLOS_ADMIN_TOKEN=") && len(kv) < 30 {
+ t.Errorf("token missing from env: %s", kv)
+ }
+ }
+ if !seenCtl {
+ t.Error("CARLOS_CONTROL_SOCKET not in the launch env")
+ }
+ if !strings.Contains(ctl.joined(), "start h.example.com run") {
+ t.Errorf("control calls: %s", ctl.joined())
+ }
+ if p.Exited() {
+ t.Error("exited before the unit went inactive")
+ }
+ // Termination: terminal state plus empty cgroup.
+ unit.set("inactive", "inv1", false, false)
+ waitFor(t, "the unit to exit", func() bool { return p.Exited() })
+ ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+ defer cancel()
+ if err := p.Stop(ctx); err != nil {
+ t.Errorf("Stop after exit: %v", err)
+ }
+}
+
+func TestUnitProcDoesNotExitOnTerminalStateWhileTheCgroupIsPopulated(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ unit.set("failed", "inv1", true, true) // a process survived SIGKILL
+ time.Sleep(150 * time.Millisecond)
+ if p.Exited() {
+ t.Fatal("exited while the cgroup is still populated (codex R2-5)")
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond)
+ defer cancel()
+ if err := p.Stop(ctx); !errors.Is(err, ErrStillRunning) {
+ t.Fatalf("Stop = %v, want ErrStillRunning", err)
+ }
+ unit.set("failed", "inv1", false, false)
+ waitFor(t, "the unit to exit", func() bool { return p.Exited() })
+}
+
+func TestUnitProcStopsAForeignInvocation(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ stopServing := serveOK(t, rt.Socket)
+ defer stopServing()
+ unit.set("active", "inv1", true, true)
+ p, _, err := b.Start(context.Background(), rt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ ctl.stopHook = func(string, instancectl.Role) { unit.set("inactive", "", false, false) }
+ unit.set("active", "inv2", true, true) // someone restarted it outside the activator
+ waitFor(t, "the unit to exit", func() bool { return p.Exited() })
+ if !strings.Contains(ctl.joined(), "stop h.example.com run") {
+ t.Errorf("foreign invocation was not stopped: %s", ctl.joined())
+ }
+}
+
+func TestStartUnitRefusesWhenTheBoxIsNotReadyOrFenced(t *testing.T) {
+ ctl := &fakeControl{inv: "inv1"}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ b.NestedOK = func() error { return errors.New("probe: pid1 fail") }
+ if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "pid1") {
+ t.Fatalf("not-ready wake = %v", err)
+ }
+ if strings.Contains(ctl.joined(), "start") {
+ t.Error("a start was issued on a box that is not ready")
+ }
+ b.NestedOK = nil
+ b.Layout.Journal = func(string) bool { return true }
+ if _, _, err := b.Start(context.Background(), rt); err == nil || !strings.Contains(err.Error(), "migration in flight") {
+ t.Fatalf("fenced wake = %v", err)
+ }
+}
+
+func TestSeedUnitCleansUpOnlyItsOwnDBOnFailure(t *testing.T) {
+ ctl := &fakeControl{seed: instancectl.SeedResponse{InvocationID: "s1", Result: "exit-code", Status: "3", Code: "exited"}}
+ unit := &fakeUnit{}
+ b, rt := nestedBackend(t, ctl, unit)
+ rt.Seed = `["seed","--fast"]`
+ unit.set("inactive", "", false, false)
+ err := b.Seed(context.Background(), rt)
+ if err == nil || !strings.Contains(err.Error(), "exit 3") {
+ t.Fatalf("Seed = %v", err)
+ }
+ raw, err := os.ReadFile(filepath.Join(provision.InstanceDir(b.DataDir, rt.Host), "launch.seed.json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ var l instancectl.Launch
+ _ = json.Unmarshal(raw, &l)
+ if len(l.Argv) != 5 || l.Argv[1] != "seed" || l.Argv[2] != "--fast" || l.Argv[3] != "--db" || l.Argv[4] != rt.DBPath {
+ t.Errorf("seed argv = %v", l.Argv)
+ }
+ for _, kv := range l.Env {
+ if strings.HasPrefix(kv, "CARLOS_CONTROL_SOCKET=") || strings.HasPrefix(kv, "CARLOS_PURGE=") {
+ t.Errorf("seed env carries %s", kv)
+ }
+ }
+ if _, err := os.Stat(rt.DBPath); !os.IsNotExist(err) {
+ t.Errorf("failed seed left its database: %v", err)
+ }
+}
+
+// serveOK binds a unix socket at path and answers 200 to every request, so
+// waitHealthyCtx passes without a real unit.
+func serveOK(t *testing.T, path string) func() {
+ t.Helper()
+ if err := os.MkdirAll(filepath.Dir(path), 0o750); err != nil {
+ t.Fatal(err)
+ }
+ ln, err := net.Listen("unix", path)
+ if err != nil {
+ t.Skipf("cannot bind here (%v); run with TMPDIR=/tmp/g outside the sandbox", err)
+ }
+ srv := &http.Server{Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(200) })}
+ go srv.Serve(ln) //nolint:errcheck
+ return func() { srv.Close(); ln.Close() }
+}