diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 987da5080..739a3c1d7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -402,4 +402,4 @@ jobs: - name: Install dependencies run: bun install --frozen-lockfile - name: Run process and terminal regressions - run: bun test tests/daemon-command.test.ts tests/daemon-orphan.test.ts tests/host-exec-lifetime.test.ts tests/host-exec-tty.test.ts tests/shell-observation.test.ts + run: bun test tests/daemon-command.test.ts tests/daemon-orphan.test.ts tests/host-exec-lifetime.test.ts tests/host-exec-tty.test.ts tests/shell-observation.test.ts tests/shared-host-lifecycle.test.ts diff --git a/docs/lifecycle.md b/docs/lifecycle.md index 4d0bc15d4..e9761669a 100644 --- a/docs/lifecycle.md +++ b/docs/lifecycle.md @@ -46,6 +46,11 @@ still be reconnecting. When the wrong listener would matter, pair `onConflict: " `lifecycle.up.before` hook that verifies the listener's identity (for example, a request to a known endpoint) and fails when it does not match. +The [shared host helper lifecycle contract](./reference/shared-host-lifecycle.md) defines the +identity-verified alternative: adoption only after the listener names the expected helper and upstream, +exact PID-plus-start-time claims, serialized starters, and release that never signals a foreign +process. It is a library contract today; `singleton` does not use it yet. + ## Config ```json diff --git a/docs/reference.md b/docs/reference.md index ee490977b..ae65b4fcf 100644 --- a/docs/reference.md +++ b/docs/reference.md @@ -12,6 +12,7 @@ - [Environment model](./env.md) - [Lifecycle & startup](./lifecycle.md) +- [Shared host helper lifecycle](./reference/shared-host-lifecycle.md) — identity-verified adoption, exact process claims and release rules for helpers shared across branches - [Sessions](./sessions.md) - [Agent-first setup](./guides/agent-first-setup.md) diff --git a/docs/reference/shared-host-lifecycle.md b/docs/reference/shared-host-lifecycle.md new file mode 100644 index 000000000..e07b83b03 --- /dev/null +++ b/docs/reference/shared-host-lifecycle.md @@ -0,0 +1,254 @@ +# Shared host helper lifecycle contract + +A shared host helper is a long-lived local process that several branches or clients +use through one fixed loopback port: an issuer bridge, a local API proxy, a database +port-forward or an opt-in tunnel. This contract defines how Hack starts, adopts, +reuses and releases such a helper without killing a process it does not own. + +Source: [`src/lib/shared-host-lifecycle.ts`](../../src/lib/shared-host-lifecycle.ts). +Fixtures: [`tests/shared-host-lifecycle.test.ts`](../../tests/shared-host-lifecycle.test.ts). + +Status: this is a library contract with portable process fixtures. No configuration +key or CLI command uses it yet. Consumers (WU18 shared helpers, bounded external +claims and opt-in tunnels) adopt it in their own changes. The existing +`lifecycle.singleton` port check in [Lifecycle & startup](../lifecycle.md) is +unchanged. It adopts any listener on the configured ports; this contract adopts +only a listener that proves its identity. + +## States + +| Axis | Values | Meaning | +| --- | --- | --- | +| Ownership | `owned` | Hack started the helper and holds an exact process claim. | +| | `adopted` | Someone else started it; its identity endpoint matched. Hack never signals it. | +| | `disabled` | The helper is off. Hack takes no lock, writes no claim and starts no child. | +| Phase | `starting` | Owned child spawned and claimed; readiness not yet proven. | +| | `ready` | The identity endpoint answered for the expected helper and upstream. | +| | `failed` | One closed failure reason (below). Terminal for this call. | + +A disabled public tunnel is the `disabled` case: it never spawns a child. + +## Identity and readiness + +The helper serves a JSON identity endpoint on `127.0.0.1:`: + +```json +{ "helper": "issuer-bridge", "upstream": "https://issuer.branch-a.test", "pid": 4242 } +``` + +- `helper` must equal the expected helper kind. Otherwise the listener is foreign. +- `upstream` must equal the upstream this branch requires. Callers derive it per + branch. A shared helper serving another upstream is refused, never retargeted. +- `pid` is optional. When present for an owned helper, it must equal the claimed PID. + A different process answering for the port is foreign. + +Readiness is this identity answer. A live process, an open port or a sleeping +placeholder is never readiness. Each probe has its own timeout and reads at most +4 KiB; a larger answer is not an identity. + +## Persisted claim + +The claim is `/.claim.json`, owner-only (0600), at most 8 KiB, +replaced atomically, and parsed strictly (exact keys, version 1): + +```json +{"version":1,"helper":"issuer-bridge","port":4100,"ownership":"owned","phase":"ready","pid":4242,"birth":"Mon Oct 19 10:00:00 2026","holders":["branch-a","branch-b"]} +{"version":1,"helper":"issuer-bridge","port":4100,"ownership":"adopted","holders":["branch-a"]} +``` + +The lock is `/.lock`. Every branch or client that uses one helper +port must pass the same `stateDir` and `name`: the state directory belongs to the +helper, not to each caller. The claim and lock are the only coordination between +clients, and Hack cannot see a claim in another directory. A client with a +different `stateDir` or `name` takes a different lock and finds no claim. If the +running owned helper answers its identity check, that client adopts it as if +someone else had started it. The owner's last release then stops the helper while +the other client still holds it, and the two clients' starts are not serialized. +Derive both values from the helper and its port, never from the branch or +checkout. + +- Ownership is the pair `pid` + `birth`. `birth` is the process start time from + `ps lstart`, always read with `TZ=UTC0` and `LC_ALL=C`. `ps` prints `lstart` in + local time, so without a fixed zone two clients with different `TZ` values, or a + host whose zone changes, would see different births for the same live process. +- The claim is live only when the census shows that exact pair. A missing PID, or + the same PID with a different birth time (PID reuse), makes the claim stale. A + stale claim grants nothing. Hack discards it and never signals that PID. +- A zombie (a process that has exited but that its parent has not reaped yet, + `ps` state `Z`) counts as exited. It runs no code and holds no listener, and its + PID cannot be reused until it is reaped. So a stop that leaves the helper a zombie + is `stopped`, and a claim whose helper is a zombie is stale. Otherwise a helper + that crashed under a suspended starter would block every restart until that + starter resumed. +- One exception: if a stale owned claim's PID still answers the identity endpoint + as the expected helper, ownership is uncertain. Hack keeps the claim and fails + with `claim-invalid`. It neither adopts the helper (which would drop the owned + record and leak the helper on the last release) nor signals it. +- `holders` names the branches or clients that use the helper. Holders carry over + when a stale owned helper is restarted. +- A malformed, oversized, unreadable, symlinked or group/world-accessible claim, or + one for a different port, fails closed with `claim-invalid`. Hack does not start, + adopt or stop anything until an operator inspects it. +- A census that did not observe the calling process (for example `ps` failed) fails + closed with `census-unavailable`. An empty census never reads as "every claimed + process is gone". + +## Start (`ensureSharedHostHelper`) + +1. `enabled: false` returns `disabled` immediately. +2. Take the helper's exclusive lock (`/.lock`, the projects-registry + lock protocol: hard-link publication, dead-owner reclamation by `ESRCH` only). + Competing starters wait in turn (`lock-waiting`, then `lock-acquired`). One + deadline bounds the lock wait, the port probe, the spawn and readiness. +3. A live owned claim is reused. The starter waits for readiness and adds its holder. + It does not spawn a second helper. + - A `ready` helper may serve other holders, so if readiness fails the starter + leaves it running and returns the failure. + - A `starting` claim seen under the lock has lost its starter, because a starter + holds the lock until its helper is ready or stopped. If that helper still + cannot become ready (`deadline`, `child-exited`, `foreign-listener`, + `wrong-upstream`), the starter stops it through its birth-verified tree and + removes the claim. `cancelled` and `census-unavailable` leave it recorded. +4. Otherwise, if the deadline has already passed, return `deadline`. Then probe the + port: + - identity matches: write an `adopted` claim and return `ready/adopted`; + - wrong upstream: `wrong-upstream`; another helper, a non-identity answer or a + listener that stays silent for the full probe timeout: `foreign-listener`. + Either way the listener keeps running and no child is spawned; + - a probe whose timeout was shortened by the deadline and then timed out proves + nothing about the listener: `deadline`; + - connection failed (refused, reset or unreachable): confirm with a loopback + bind probe. An occupied port is `foreign-listener`; a free port is spawned. + The refused-connection error code differs by platform, so the error itself + never proves absence. +5. Spawn through an exec gate in a new session: + `/bin/sh -c 'IFS= read -r hack_gate || exit 0; [ "$hack_gate" = open ] || exit 0; exec "$@" `. +- PID reuse within one `lstart` second. Birth times have one-second resolution, the + same assumption as the existing lifecycle process-tree helpers. +- Linux `lstart` drift. Linux `ps` computes `lstart` from the boot time, which can + move when the wall clock is changed, so a live helper's birth can change. Start + then fails closed if the helper reports its PID in the identity answer, but a + release would read the claim as stale and clear it without stopping the helper. + macOS records the start time directly and is not affected. +- Wiring into `hack.config.json`, `hack up`/`down` or the daemon, and migrating + existing `lifecycle.singleton` configurations. +- Native (macOS VM) and container-runtime effects. These fixtures qualify the + process and claim protocol only. diff --git a/src/lib/private-file.ts b/src/lib/private-file.ts index 3776ef235..6101e353b 100644 --- a/src/lib/private-file.ts +++ b/src/lib/private-file.ts @@ -278,7 +278,10 @@ async function observeReplaceableTarget(opts: { * unreadable file is never mistaken for an empty one and overwritten. */ export async function readReplaceablePrivateFile( - opts: ReplaceablePrivateFileOptions + opts: ReplaceablePrivateFileOptions & { + /** Largest accepted file size in bytes; a larger file throws `refusal()` before it is read. */ + readonly maxBytes?: number; + } ): Promise { const directories = await captureRealDirectories(opts); const observed = await observeReplaceableTarget(opts); @@ -303,11 +306,19 @@ export async function readReplaceablePrivateFile( if ( !opened.isFile() || opened.dev !== observed.dev || - opened.ino !== observed.ino + opened.ino !== observed.ino || + (opts.maxBytes !== undefined && opened.size > opts.maxBytes) ) { throw opts.refusal(); } const text = await file.readFile({ encoding: "utf8" }); + if ( + opts.maxBytes !== undefined && + Buffer.byteLength(text, "utf8") > opts.maxBytes + ) { + // The file grew in place after the size check. + throw opts.refusal(); + } await recheckRealDirectories({ directories, refusal: opts.refusal }); return { text, mode: opened.mode & 0o7777 }; } finally { diff --git a/src/lib/process-tree.ts b/src/lib/process-tree.ts index c354fc612..d51ba0fb7 100644 --- a/src/lib/process-tree.ts +++ b/src/lib/process-tree.ts @@ -2,14 +2,31 @@ export type ProcessIdentityRow = { readonly pid: number; readonly parentPid: number; readonly birth: string; + /** + * The process has exited but its parent has not reaped it yet (`ps` state + * `Z`). A zombie runs no code and holds no listener; its PID cannot be + * reused until it is reaped. + */ + readonly zombie: boolean; }; -const IDENTITY_ROW = /^\s*(\d+)\s+(\d+)\s+(.+?)\s*$/; +const IDENTITY_ROW = /^\s*(\d+)\s+(\d+)\s+(\S+)\s+(.+?)\s*$/; -/** Capture only process identity and lineage; never command arguments or environment. */ +/** + * Capture only process identity and lineage; never command arguments or environment. + * + * `birth` is the `lstart` start time rendered in UTC with the C locale. `ps` + * prints `lstart` in local time, so without a fixed zone two clients with + * different `TZ` values, or one host whose zone changes, would see different + * births for the same live process. Births may be persisted and compared across + * processes, so the zone is pinned here (POSIX `UTC0` needs no zoneinfo). + * + * Zombie rows are kept and flagged, not dropped: each caller decides whether + * an exited, unreaped process still counts for its purpose. + */ export async function readProcessIdentities(): Promise { try { - const proc = Bun.spawn(["ps", "-A", "-o", "pid=,ppid=,lstart="], { - env: { ...process.env, LC_ALL: "C" }, + const proc = Bun.spawn(["ps", "-A", "-o", "pid=,ppid=,stat=,lstart="], { + env: { ...process.env, LC_ALL: "C", TZ: "UTC0" }, stdin: "ignore", stdout: "pipe", stderr: "ignore", @@ -26,7 +43,8 @@ export async function readProcessIdentities(): Promise { { pid: Number(match[1]), parentPid: Number(match[2]), - birth: (match[3] ?? "").replace(/\s+/g, " "), + birth: (match[4] ?? "").replace(/\s+/g, " "), + zombie: (match[3] ?? "").startsWith("Z"), }, ] : []; diff --git a/src/lib/shared-host-lifecycle.ts b/src/lib/shared-host-lifecycle.ts new file mode 100644 index 000000000..20a47d7e6 --- /dev/null +++ b/src/lib/shared-host-lifecycle.ts @@ -0,0 +1,1246 @@ +import { mkdir, unlink } from "node:fs/promises"; +import { join } from "node:path"; +import { setTimeout as delay } from "node:timers/promises"; +import { isRecord } from "./guards.ts"; +import { + readReplaceablePrivateFile, + replacePrivateFileAtomic, +} from "./private-file.ts"; +import { + type ProcessIdentityRow, + readProcessIdentities, + selectProcessTree, + signalVerifiedProcessTree, +} from "./process-tree.ts"; +import { inspectListeningTcpPorts } from "./project-lifecycle-singleton.ts"; +import { withProjectsRegistryLock } from "./projects-registry-lock.ts"; + +/** + * Ownership-safe lifecycle for one shared host helper: a long-lived local + * process (proxy, tunnel, issuer bridge) that several branches or clients use + * through one fixed loopback port. + * + * Contract (see docs/reference/shared-host-lifecycle.md): + * - Ownership is `owned` (Hack started it and holds an exact process claim), + * `adopted` (an identity-verified helper someone else started) or `disabled` + * (no lock, no claim, no child). Phases are `starting`, `ready` and `failed`. + * - Starters for one helper are serialized by an exclusive lock next to its + * claim; only one of them may spawn. Every client of one helper port must + * therefore use the same `stateDir` and `name` (see + * {@link SharedHostHelperSpec.stateDir}). + * - A listener is only ever adopted after its identity endpoint names the + * expected helper and upstream. Anything else on the port is refused and + * left running: this module never signals a process it cannot prove it owns. + * - Ownership is a persisted PID plus process birth time (`ps lstart`, read in + * UTC so clients with different time zones agree). A claim whose PID is + * gone, an unreaped zombie, or now belongs to a different process (PID + * reuse) is stale and is discarded without signalling anything. A stale + * claim whose PID still answers as the helper is uncertain: it is kept, + * never adopted. + * - A child is spawned behind an exec gate: it waits on stdin until its claim + * is durable. If the starter dies first, the gate reads EOF and exits, so a + * spawned helper is never left running without a claim. + * - Readiness is the identity endpoint answering for the expected helper, not + * process liveness. Retries back off and are bounded by one deadline that + * also covers the lock wait; cancellation stops the owned child and removes + * its claim. A `starting` claim found under the lock has lost its starter; + * if that helper still cannot become ready, it is stopped. + * - Release removes one holder. Only when no holder remains is an owned helper + * stopped, through its birth-verified process tree. Adopted helpers are + * never signalled. + * + * Diagnostics are closed: every failure is one of {@link SharedHostFailure} + * with a fixed message that never includes paths, argv, env or response + * bodies. + */ + +export type SharedHostOwnership = "owned" | "adopted" | "disabled"; +export type SharedHostPhase = "starting" | "ready" | "failed"; + +export type SharedHostFailure = + | "foreign-listener" + | "wrong-upstream" + | "claim-invalid" + | "state-unavailable" + | "census-unavailable" + | "lock-unavailable" + | "port-probe-failed" + | "spawn-failed" + | "child-exited" + | "deadline" + | "cancelled"; + +const FAILURE_MESSAGES: Readonly> = { + "foreign-listener": + "The shared helper port is held by a process that is not the expected helper; it was left running.", + "wrong-upstream": + "The shared helper on this port serves a different upstream; it was left running.", + "claim-invalid": + "The persisted shared helper claim is unreadable, uncertain or belongs to another configuration; inspect and release it before retrying.", + "state-unavailable": + "The shared helper state directory could not be read or written.", + "census-unavailable": + "The process census is unavailable, so helper ownership cannot be verified.", + "lock-unavailable": + "Another starter holds the shared helper lock beyond the deadline.", + "port-probe-failed": "The shared helper port could not be inspected.", + "spawn-failed": "The shared helper could not be started.", + "child-exited": "The shared helper exited before it became ready.", + deadline: "The shared helper did not become ready before the deadline.", + cancelled: "Shared helper startup was cancelled.", +}; + +/** Fixed, value-free message for a closed failure reason. */ +export function sharedHostFailureMessage(reason: SharedHostFailure): string { + return FAILURE_MESSAGES[reason]; +} + +/** Identity a helper must report at its identity endpoint. */ +export type SharedHostIdentity = { + /** Stable helper kind, for example `issuer-bridge`. */ + readonly helper: string; + /** The upstream this branch requires the helper to serve. */ + readonly upstream: string; +}; + +export type SharedHostHelperSpec = { + /** + * Claim and lock name; `[A-Za-z0-9][A-Za-z0-9_.-]{0,127}`. Every client of + * one helper port must use the same name (see {@link stateDir}). + */ + readonly name: string; + /** Disabled helpers take no lock, write no claim and start no child. */ + readonly enabled: boolean; + /** + * Private (owner-only) directory that holds this helper's claim and lock. + * + * It is shared, not per caller: every branch or client that uses the helper + * on `port` must pass the same `stateDir` and `name`. The claim and lock are + * the only coordination between clients, and nothing here can see a claim in + * another directory. A client with a different `stateDir` or `name` takes a + * different lock and finds no claim. If the running owned helper answers its + * identity check, that client adopts it, so the owner's last release stops a + * helper this client still holds, and their starts are not serialized. + * Derive both values from the helper and its port, never from the branch. + */ + readonly stateDir: string; + /** Loopback port the helper serves. */ + readonly port: number; + /** Absolute path of the JSON identity endpoint, for example `/__hack/identity`. */ + readonly identityPath: string; + readonly expected: SharedHostIdentity; + /** Helper argv; executed directly, never through a shell string. */ + readonly command: readonly string[]; + readonly cwd: string; + /** Complete child environment. The runner environment is never inherited. */ + readonly env: Readonly>; + /** One bound for lock wait, spawn and readiness. */ + readonly deadlineMs: number; + readonly backoff: { readonly initialMs: number; readonly maxMs: number }; + /** Per-request bound for one identity probe. */ + readonly probeTimeoutMs: number; +}; + +export type SharedHostEnsureResult = + | { readonly state: "disabled"; readonly ownership: "disabled" } + | { + readonly state: "ready"; + readonly ownership: "owned" | "adopted"; + readonly endpoint: string; + /** Present for owned helpers only. */ + readonly pid?: number; + readonly spawned: boolean; + } + | { + readonly state: "failed"; + readonly reason: SharedHostFailure; + readonly message: string; + }; + +/** Closed progress events. The observer is awaited (a test and UI seam). */ +export type SharedHostEvent = + | "lock-waiting" + | "lock-acquired" + | "claim-stale" + | "claim-reused" + | "adopted" + | "spawned" + | "claimed" + | "gate-opened" + | "probe-retry" + | "ready" + | "stopped-after-failure"; + +export type SharedHostReleaseResult = + | { + readonly state: + | "absent" + | "retained" + | "released-adopted" + | "stale-cleared" + | "stopped" + | "stop-unconfirmed"; + } + | { + readonly state: "failed"; + readonly reason: ReleaseFailure; + readonly message: string; + }; + +type ReleaseFailure = Extract< + SharedHostFailure, + | "claim-invalid" + | "state-unavailable" + | "census-unavailable" + | "lock-unavailable" + | "cancelled" +>; + +type OwnedClaim = { + readonly version: 1; + readonly helper: string; + readonly port: number; + readonly ownership: "owned"; + readonly phase: "starting" | "ready"; + readonly pid: number; + readonly birth: string; + readonly holders: readonly string[]; +}; +type AdoptedClaim = { + readonly version: 1; + readonly helper: string; + readonly port: number; + readonly ownership: "adopted"; + readonly holders: readonly string[]; +}; +type Claim = OwnedClaim | AdoptedClaim; + +type ProbeObservation = + /** The connection failed (refused, reset or unreachable). */ + | { readonly kind: "no-answer" } + | { readonly kind: "unidentified" } + /** The listener did not answer within this probe's own timeout. */ + | { readonly kind: "timeout" } + | { readonly kind: "foreign" } + | { readonly kind: "wrong-upstream" } + | { readonly kind: "match"; readonly pid: number | null } + | { readonly kind: "cancelled" }; + +const NAME = /^[A-Za-z0-9][A-Za-z0-9_.-]{0,127}$/; +const MAX_CLAIM_BYTES = 8192; +const MAX_IDENTITY_BYTES = 4096; +const MAX_HOLDERS = 256; +const STOP_POLL_MS = 50; +/** + * Exec gate: the helper argv runs only after the starter writes `open`. EOF + * (starter lost before its claim was durable) exits without running it. `exec` + * keeps the PID and birth time recorded in the claim. + */ +const GATE_SCRIPT = + 'IFS= read -r hack_gate || exit 0; [ "$hack_gate" = open ] || exit 0; exec "$@" (run: () => Promise): Promise { + try { + return await run(); + } catch (error: unknown) { + throw new LockedRunError(error); + } +} + +function failed(reason: SharedHostFailure): SharedHostEnsureResult { + return { state: "failed", reason, message: FAILURE_MESSAGES[reason] }; +} + +function releaseFailed(reason: ReleaseFailure): SharedHostReleaseResult { + return { state: "failed", reason, message: FAILURE_MESSAGES[reason] }; +} + +function assertSpec(spec: SharedHostHelperSpec): void { + const valid = + NAME.test(spec.name) && + Number.isInteger(spec.port) && + spec.port > 0 && + spec.port <= 65_535 && + spec.identityPath.startsWith("/") && + spec.command.length > 0 && + spec.deadlineMs > 0 && + spec.probeTimeoutMs > 0 && + spec.backoff.initialMs > 0 && + spec.backoff.maxMs >= spec.backoff.initialMs; + if (!valid) { + throw new Error("Invalid shared host helper specification"); + } +} + +function assertHolder(holder: string): void { + if (!NAME.test(holder)) { + throw new Error("Invalid shared host helper holder"); + } +} + +function claimPath(spec: Pick) { + return join(spec.stateDir, `${spec.name}.claim.json`); +} + +function lockPath(spec: Pick) { + return join(spec.stateDir, `${spec.name}.lock`); +} + +function endpointFor(port: number): string { + return `http://127.0.0.1:${port}`; +} + +function exactKeys(value: Record, keys: readonly string[]) { + const actual = Object.keys(value).sort(); + const wanted = [...keys].sort(); + return ( + actual.length === wanted.length && + actual.every((key, index) => key === wanted[index]) + ); +} + +function parseHolders(value: unknown): readonly string[] | null { + if ( + !Array.isArray(value) || + value.length > MAX_HOLDERS || + !value.every((entry) => typeof entry === "string" && NAME.test(entry)) || + new Set(value).size !== value.length + ) { + return null; + } + return value as string[]; +} + +/** Strict v1 claim parser; any other shape is an uncertain receipt. */ +export function parseSharedHostClaim(text: string): Claim | null { + let value: unknown; + try { + value = JSON.parse(text); + } catch { + return null; + } + if (!isRecord(value) || value.version !== 1) { + return null; + } + const holders = parseHolders(value.holders); + const port = value.port; + if ( + holders === null || + typeof value.helper !== "string" || + !NAME.test(value.helper) || + typeof port !== "number" || + !Number.isInteger(port) || + port <= 0 || + port > 65_535 + ) { + return null; + } + if (value.ownership === "adopted") { + return exactKeys(value, [ + "version", + "helper", + "port", + "ownership", + "holders", + ]) + ? { + version: 1, + helper: value.helper, + port, + ownership: "adopted", + holders, + } + : null; + } + const pid = value.pid; + if ( + value.ownership !== "owned" || + !exactKeys(value, [ + "version", + "helper", + "port", + "ownership", + "phase", + "pid", + "birth", + "holders", + ]) || + (value.phase !== "starting" && value.phase !== "ready") || + typeof pid !== "number" || + !Number.isSafeInteger(pid) || + pid <= 1 || + typeof value.birth !== "string" || + value.birth.length === 0 || + value.birth.length > 64 + ) { + return null; + } + return { + version: 1, + helper: value.helper, + port, + ownership: "owned", + phase: value.phase, + pid, + birth: value.birth, + holders, + }; +} + +type ClaimRead = + | { readonly kind: "absent" } + | { readonly kind: "invalid" } + | { readonly kind: "claim"; readonly claim: Claim }; + +async function readClaim( + spec: Pick +): Promise { + let file: Awaited>; + try { + file = await readReplaceablePrivateFile({ + path: claimPath(spec), + directories: [spec.stateDir], + refusal: () => new ClaimRefusal(), + maxBytes: MAX_CLAIM_BYTES, + }); + } catch { + return { kind: "invalid" }; + } + if (file === null) { + return { kind: "absent" }; + } + // Only an owner-only receipt is trusted; group/world access makes it uncertain. + if ((file.mode & 0o077) !== 0) { + return { kind: "invalid" }; + } + const claim = parseSharedHostClaim(file.text); + if (claim === null || claim.helper !== spec.name) { + return { kind: "invalid" }; + } + return { kind: "claim", claim }; +} + +async function writeClaim( + spec: Pick, + claim: Claim +): Promise { + await replacePrivateFileAtomic({ + path: claimPath(spec), + directories: [spec.stateDir], + refusal: () => new ClaimRefusal(), + text: `${JSON.stringify(claim)}\n`, + }); +} + +async function removeClaim( + spec: Pick +): Promise { + try { + await unlink(claimPath(spec)); + } catch (error: unknown) { + if (!(isRecord(error) && error.code === "ENOENT")) { + throw error; + } + } +} + +/** + * The process census and signal boundary. Production callers omit it and get + * {@link systemProcesses}; tests replace it to simulate states real processes + * cannot produce portably, such as a process that stays in the census after + * SIGKILL. + */ +export type SharedHostProcesses = { + /** Every process row, including zombies; `[]` when `ps` failed. */ + readonly census: () => Promise; + /** Signal the members of a captured tree that still carry their birth. */ + readonly signalTree: ( + tree: readonly ProcessIdentityRow[], + signal: NodeJS.Signals + ) => Promise; +}; + +const systemProcesses: SharedHostProcesses = { + census: readProcessIdentities, + signalTree: signalVerifiedProcessTree, +}; + +/** + * A census is usable only when it observed this process; an empty or failed + * `ps` must never be read as "every claimed process is gone". + * + * Zombies are left out: an exited process that its parent has not reaped yet + * runs no code, holds no listener or open file, and cannot be stopped any + * further. Its PID cannot be reused until it is reaped, so treating it as + * exited can never confuse it with another process. Counting it as live would + * report a finished stop as unconfirmed, and would make a `ready` claim whose + * helper crashed under a suspended (for example Ctrl-Z) starter look live, so + * no other client could restart that helper until the starter resumed. + */ +async function readCensus( + processes: SharedHostProcesses = systemProcesses +): Promise { + const rows = await processes.census(); + if (!rows.some((row) => row.pid === process.pid)) { + return null; + } + return rows.filter((row) => !row.zombie); +} + +function isLive( + census: readonly ProcessIdentityRow[], + claim: { readonly pid: number; readonly birth: string } +): boolean { + return census.some( + (row) => row.pid === claim.pid && row.birth === claim.birth + ); +} + +async function readBounded(response: Response): Promise { + const reader = response.body?.getReader(); + if (!reader) { + return ""; + } + const chunks: Uint8Array[] = []; + let size = 0; + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + size += value.byteLength; + if (size > MAX_IDENTITY_BYTES) { + await reader.cancel().catch(() => undefined); + return null; + } + chunks.push(value); + } + } catch { + return null; + } + try { + return new TextDecoder("utf-8", { fatal: true }).decode( + Buffer.concat(chunks) + ); + } catch { + return null; + } +} + +/** Classify the listener on the helper port by its identity endpoint. */ +export async function probeSharedHostIdentity(opts: { + readonly port: number; + readonly identityPath: string; + readonly expected: SharedHostIdentity; + readonly timeoutMs: number; + readonly signal?: AbortSignal; +}): Promise { + const timeout = AbortSignal.timeout(opts.timeoutMs); + const signals = [timeout]; + if (opts.signal) { + signals.push(opts.signal); + } + let response: Response; + try { + response = await fetch(`${endpointFor(opts.port)}${opts.identityPath}`, { + redirect: "manual", + signal: AbortSignal.any(signals), + }); + } catch { + if (opts.signal?.aborted) { + return { kind: "cancelled" }; + } + if (timeout.aborted) { + return { kind: "timeout" }; + } + // The refused-connection error code differs by platform and runtime, so a + // transport error alone never proves absence: the caller confirms it with + // a loopback bind probe before spawning. + return { kind: "no-answer" }; + } + const text = await readBounded(response); + if (opts.signal?.aborted) { + return { kind: "cancelled" }; + } + if (text === null && timeout.aborted) { + return { kind: "timeout" }; + } + if (response.status !== 200 || text === null) { + return { kind: "unidentified" }; + } + return classifyIdentity(text, opts.expected); +} + +function classifyIdentity( + text: string, + expected: SharedHostIdentity +): ProbeObservation { + let body: unknown; + try { + body = JSON.parse(text); + } catch { + return { kind: "unidentified" }; + } + if ( + !isRecord(body) || + typeof body.helper !== "string" || + typeof body.upstream !== "string" + ) { + return { kind: "unidentified" }; + } + if (body.helper !== expected.helper) { + return { kind: "foreign" }; + } + if (body.upstream !== expected.upstream) { + return { kind: "wrong-upstream" }; + } + const pid = + typeof body.pid === "number" && Number.isSafeInteger(body.pid) + ? body.pid + : null; + return { kind: "match", pid }; +} + +/** + * Stop a birth-verified process tree: TERM, bounded grace, then KILL for the + * members that still carry their captured birth time. Nothing is signalled + * when the root no longer matches its claim. A member that has exited counts + * as stopped even before its parent reaps it (see {@link readCensus}). + */ +export async function stopVerifiedSharedHostProcess(opts: { + readonly pid: number; + readonly birth: string; + readonly graceMs: number; + readonly processes?: SharedHostProcesses; +}): Promise<"stopped" | "absent" | "unconfirmed" | "census-unavailable"> { + const processes = opts.processes ?? systemProcesses; + const census = await readCensus(processes); + if (census === null) { + return "census-unavailable"; + } + if (!isLive(census, opts)) { + return "absent"; + } + const tree = selectProcessTree(census, opts.pid); + const remaining = async (): Promise => { + const live = await readCensus(processes); + if (live === null) { + return null; + } + return tree.filter((row) => isLive(live, row)); + }; + await processes.signalTree(tree, "SIGTERM"); + const termDeadline = performance.now() + opts.graceMs; + let left = await remaining(); + while (left !== null && left.length > 0 && performance.now() < termDeadline) { + await delay(STOP_POLL_MS); + left = await remaining(); + } + if (left !== null && left.length === 0) { + return "stopped"; + } + await processes.signalTree(left ?? tree, "SIGKILL"); + const killDeadline = performance.now() + 1000; + left = await remaining(); + while (left !== null && left.length > 0 && performance.now() < killDeadline) { + await delay(STOP_POLL_MS); + left = await remaining(); + } + return left !== null && left.length === 0 ? "stopped" : "unconfirmed"; +} + +function withHolder(holders: readonly string[], holder: string): string[] { + return holders.includes(holder) ? [...holders] : [...holders, holder]; +} + +type EnsureContext = { + readonly spec: SharedHostHelperSpec; + readonly holder: string; + readonly signal?: AbortSignal; + readonly observe: (event: SharedHostEvent) => Promise; + readonly deadline: number; +}; + +function remainingMs(context: EnsureContext): number { + return context.deadline - performance.now(); +} + +type ReadinessOutcome = + | { readonly kind: "ready" } + | { + readonly kind: "failed"; + readonly reason: Extract< + SharedHostFailure, + | "foreign-listener" + | "wrong-upstream" + | "child-exited" + | "deadline" + | "cancelled" + | "census-unavailable" + >; + }; + +/** + * Probe until the identity endpoint answers for the expected helper and, when + * it reports a PID, for the owned process. Liveness alone is never readiness. + */ +async function awaitReadiness( + context: EnsureContext, + owned: { readonly pid: number; readonly birth: string } +): Promise { + const { spec } = context; + let wait = spec.backoff.initialMs; + while (true) { + if (context.signal?.aborted) { + return { kind: "failed", reason: "cancelled" }; + } + const census = await readCensus(); + if (census === null) { + return { kind: "failed", reason: "census-unavailable" }; + } + if (!isLive(census, owned)) { + return { kind: "failed", reason: "child-exited" }; + } + const left = remainingMs(context); + if (left <= 0) { + return { kind: "failed", reason: "deadline" }; + } + const probe = await probeSharedHostIdentity({ + port: spec.port, + identityPath: spec.identityPath, + expected: spec.expected, + timeoutMs: Math.min(spec.probeTimeoutMs, left), + signal: context.signal, + }); + switch (probe.kind) { + case "cancelled": + return { kind: "failed", reason: "cancelled" }; + case "wrong-upstream": + return { kind: "failed", reason: "wrong-upstream" }; + case "foreign": + return { kind: "failed", reason: "foreign-listener" }; + case "match": + if (probe.pid === null || probe.pid === owned.pid) { + return { kind: "ready" }; + } + return { kind: "failed", reason: "foreign-listener" }; + default: + break; + } + const sleep = Math.min(wait, remainingMs(context)); + if (sleep <= 0) { + return { kind: "failed", reason: "deadline" }; + } + await context.observe("probe-retry"); + try { + await delay(sleep, undefined, { signal: context.signal }); + } catch { + return { kind: "failed", reason: "cancelled" }; + } + wait = Math.min(wait * 2, spec.backoff.maxMs); + } +} + +type OwnedIdentity = { readonly pid: number; readonly birth: string }; + +/** + * Readiness failures after which an owned helper that never reached `ready` is + * stopped. `cancelled` and `census-unavailable` leave it recorded instead: a + * cancelled caller or a blind census is not evidence that the start failed. + */ +const ABANDONED_START_FAILURES: ReadonlySet = new Set([ + "child-exited", + "deadline", + "foreign-listener", + "wrong-upstream", +]); + +/** + * Stop an owned helper whose start failed and forget its claim. When the stop + * cannot be confirmed the claim is kept, so the process stays recorded. + */ +async function stopFailedStart(opts: { + readonly context: EnsureContext; + readonly owned: OwnedIdentity; + readonly notify: boolean; +}): Promise { + const stopped = await stopVerifiedSharedHostProcess({ + ...opts.owned, + graceMs: 1000, + }); + if (stopped === "stopped" || stopped === "absent") { + await removeClaim(opts.context.spec); + } + if (opts.notify) { + await opts.context.observe("stopped-after-failure"); + } +} + +/** + * Reuse a live owned helper. A `ready` helper may serve other holders, so a + * failed readiness check never stops it. A `starting` claim seen under the + * lock has no live starter (a starter holds the lock until its helper is + * ready or stopped), so a helper that still cannot become ready is an + * abandoned start and is stopped through its birth-verified tree. + */ +async function reuseOwned( + context: EnsureContext, + claim: OwnedClaim +): Promise { + const { spec, holder } = context; + await context.observe("claim-reused"); + const outcome = await awaitReadiness(context, claim); + if (outcome.kind === "failed") { + if ( + claim.phase === "starting" && + ABANDONED_START_FAILURES.has(outcome.reason) + ) { + await stopFailedStart({ context, owned: claim, notify: true }); + } + return failed(outcome.reason); + } + await writeClaim(spec, { + ...claim, + phase: "ready", + holders: withHolder(claim.holders, holder), + }); + await context.observe("ready"); + return { + state: "ready", + ownership: "owned", + endpoint: endpointFor(spec.port), + pid: claim.pid, + spawned: false, + }; +} + +async function ensureLocked( + context: EnsureContext +): Promise { + const { spec } = context; + await context.observe("lock-acquired"); + const census = await readCensus(); + if (census === null) { + return failed("census-unavailable"); + } + const read = await readClaim(spec); + if (read.kind === "invalid") { + return failed("claim-invalid"); + } + let carried: readonly string[] = []; + let staleOwned: OwnedClaim | null = null; + if (read.kind === "claim") { + const claim = read.claim; + if (claim.port !== spec.port) { + return failed("claim-invalid"); + } + if (claim.ownership === "owned" && isLive(census, claim)) { + return await reuseOwned(context, claim); + } + if (claim.ownership === "owned") { + // Dead or reused PID: the claim no longer proves ownership of anything. + await context.observe("claim-stale"); + staleOwned = claim; + } + carried = claim.holders; + } + const settled = await adoptOrRefuse(context, { carried, staleOwned }); + if (settled !== null) { + return settled; + } + try { + const occupied = await inspectListeningTcpPorts({ ports: [spec.port] }); + if (occupied.length > 0) { + return failed("foreign-listener"); + } + } catch { + return failed("port-probe-failed"); + } + if (read.kind === "claim") { + await removeClaim(spec); + } + return await spawnOwned(context, carried); +} + +/** + * Probe the helper port when no live owned claim exists. Returns the final + * result (adopted, refused, deadline or cancelled), or null when nothing + * answers and the caller may spawn. + */ +async function adoptOrRefuse( + context: EnsureContext, + prior: { + readonly carried: readonly string[]; + readonly staleOwned: OwnedClaim | null; + } +): Promise { + const { spec, holder } = context; + const { carried, staleOwned } = prior; + // The lock wait shares the deadline; never start a probe after it passed. + const left = remainingMs(context); + if (left <= 0) { + return failed("deadline"); + } + const cutByDeadline = left < spec.probeTimeoutMs; + const existing = await probeSharedHostIdentity({ + port: spec.port, + identityPath: spec.identityPath, + expected: spec.expected, + timeoutMs: cutByDeadline ? left : spec.probeTimeoutMs, + signal: context.signal, + }); + switch (existing.kind) { + case "cancelled": + return failed("cancelled"); + case "timeout": + // A probe cut short by the deadline proves nothing about the listener; + // one that used its full timeout met a listener that does not answer. + return failed(cutByDeadline ? "deadline" : "foreign-listener"); + case "unidentified": + case "foreign": + return failed("foreign-listener"); + case "wrong-upstream": + return failed("wrong-upstream"); + case "match": + if (staleOwned !== null && existing.pid === staleOwned.pid) { + // The claimed PID answers as the helper, but its birth no longer + // matches: ownership is uncertain. Adopting it would drop the owned + // record (and a later release would leak the helper); stopping it + // could signal a stranger. Keep the claim and fail closed. + return failed("claim-invalid"); + } + await writeClaim(spec, { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "adopted", + holders: withHolder(carried, holder), + }); + await context.observe("adopted"); + return { + state: "ready", + ownership: "adopted", + endpoint: endpointFor(spec.port), + spawned: false, + }; + default: + return null; + } +} + +async function spawnOwned( + context: EnsureContext, + carried: readonly string[] +): Promise { + const { spec, holder } = context; + let child: Bun.Subprocess<"pipe", "ignore", "ignore">; + try { + child = Bun.spawn( + ["/bin/sh", "-c", GATE_SCRIPT, "hack-shared-host-gate", ...spec.command], + { + cwd: spec.cwd, + env: { ...spec.env }, + stdin: "pipe", + stdout: "ignore", + stderr: "ignore", + // A new session keeps terminal signals for the starter away from a + // helper that other branches share. + detached: true, + } + ); + } catch { + return failed("spawn-failed"); + } + child.unref(); + const closeGate = () => { + try { + child.stdin.end(); + } catch { + // Already closed; the gate treats EOF as "do not run". + } + }; + let owned: OwnedIdentity | null = null; + let gateOpened = false; + try { + await context.observe("spawned"); + const census = await readCensus(); + const row = census?.find((entry) => entry.pid === child.pid); + if (!row) { + closeGate(); + return failed(census === null ? "census-unavailable" : "spawn-failed"); + } + const identity: OwnedIdentity = { pid: row.pid, birth: row.birth }; + const holders = withHolder(carried, holder); + await writeClaim(spec, { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "starting", + pid: identity.pid, + birth: identity.birth, + holders, + }); + owned = identity; + await context.observe("claimed"); + if (context.signal?.aborted) { + closeGate(); + await removeClaim(spec); + return failed("cancelled"); + } + gateOpened = true; + try { + child.stdin.write("open\n"); + await child.stdin.flush(); + } catch { + // The gate died before it could run the helper. + closeGate(); + await stopFailedStart({ context, owned: identity, notify: false }); + return failed("spawn-failed"); + } + closeGate(); + await context.observe("gate-opened"); + const outcome = await awaitReadiness(context, identity); + if (outcome.kind === "failed") { + await stopFailedStart({ context, owned: identity, notify: true }); + return failed(outcome.reason); + } + await writeClaim(spec, { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: identity.pid, + birth: identity.birth, + holders, + }); + await context.observe("ready"); + return { + state: "ready", + ownership: "owned", + endpoint: endpointFor(spec.port), + pid: identity.pid, + spawned: true, + }; + } catch (error: unknown) { + // An observer or a state write failed. Never leave a gate waiting while + // this starter lives, nor a running helper the caller was not handed. + closeGate(); + if (owned !== null) { + const recorded = owned; + await (gateOpened + ? stopFailedStart({ context, owned: recorded, notify: false }) + : removeClaim(spec) + ).catch(() => undefined); + } + throw error; + } +} + +async function prepareStateDir(stateDir: string): Promise { + await mkdir(stateDir, { recursive: true, mode: 0o700 }); +} + +function isAbort(error: unknown, signal?: AbortSignal): boolean { + return ( + signal?.aborted === true || + (error instanceof Error && error.name === "AbortError") + ); +} + +function hasErrorCode(error: unknown): boolean { + return isRecord(error) && typeof error.code === "string"; +} + +/** + * Map an error to a closed reason. A caller's observer error is rethrown + * unchanged; errors raised while the lock is held are never `lock-unavailable`. + */ +function classifyFailure( + error: unknown, + signal: AbortSignal | undefined +): ReleaseFailure { + if (error instanceof ObserverError) { + throw error.original; + } + if (error instanceof LockedRunError) { + const inner = error.original; + if (inner instanceof ObserverError) { + throw inner.original; + } + if (inner instanceof ClaimRefusal) { + return "claim-invalid"; + } + return isAbort(inner, signal) ? "cancelled" : "state-unavailable"; + } + if (isAbort(error, signal)) { + return "cancelled"; + } + // The lock times out with a plain error; a filesystem error carries a code. + return hasErrorCode(error) ? "state-unavailable" : "lock-unavailable"; +} + +/** + * Make the shared helper ready for `holder`, adopting an identity-verified + * helper, reusing the owned one, or starting it under an exact claim. + * + * An error thrown by `observe` propagates unchanged after cleanup: a gate that + * has not opened is closed and its claim removed, and a helper started by this + * call is stopped. A holder recorded before the error stays recorded. + */ +export async function ensureSharedHostHelper(opts: { + readonly spec: SharedHostHelperSpec; + readonly holder: string; + readonly signal?: AbortSignal; + readonly observe?: (event: SharedHostEvent) => void | Promise; +}): Promise { + assertSpec(opts.spec); + assertHolder(opts.holder); + if (!opts.spec.enabled) { + return { state: "disabled", ownership: "disabled" }; + } + if (opts.signal?.aborted) { + return failed("cancelled"); + } + const observer = opts.observe; + const context: EnsureContext = { + spec: opts.spec, + holder: opts.holder, + signal: opts.signal, + observe: async (event) => { + try { + await observer?.(event); + } catch (error: unknown) { + throw new ObserverError(error); + } + }, + deadline: performance.now() + opts.spec.deadlineMs, + }; + try { + await prepareStateDir(opts.spec.stateDir); + } catch { + return failed("state-unavailable"); + } + try { + await context.observe("lock-waiting"); + return await withProjectsRegistryLock({ + lockPath: lockPath(opts.spec), + label: "shared host helper", + timeoutMs: opts.spec.deadlineMs, + signal: opts.signal, + run: () => runLocked(() => ensureLocked(context)), + }); + } catch (error: unknown) { + return failed(classifyFailure(error, opts.signal)); + } +} + +/** + * Drop `holder`. The last holder of an owned helper stops it through its + * birth-verified tree; adopted helpers and stale claims are never signalled. + * Releasing a holder that is not recorded changes nothing and reports + * `retained` while other holders remain. + */ +export async function releaseSharedHostHelper(opts: { + readonly spec: Pick; + readonly holder: string; + readonly graceMs?: number; + readonly timeoutMs?: number; + readonly signal?: AbortSignal; + /** Test seam for the census and signals; production callers omit it. */ + readonly processes?: SharedHostProcesses; +}): Promise { + if (!NAME.test(opts.spec.name)) { + throw new Error("Invalid shared host helper specification"); + } + assertHolder(opts.holder); + try { + await prepareStateDir(opts.spec.stateDir); + } catch { + return releaseFailed("state-unavailable"); + } + try { + return await withProjectsRegistryLock({ + lockPath: lockPath(opts.spec), + label: "shared host helper", + timeoutMs: opts.timeoutMs ?? 10_000, + signal: opts.signal, + run: () => runLocked(() => releaseLocked(opts)), + }); + } catch (error: unknown) { + return releaseFailed(classifyFailure(error, opts.signal)); + } +} + +async function releaseLocked(opts: { + readonly spec: Pick; + readonly holder: string; + readonly graceMs?: number; + readonly processes?: SharedHostProcesses; +}): Promise { + const read = await readClaim(opts.spec); + if (read.kind === "absent") { + return { state: "absent" }; + } + if (read.kind === "invalid") { + return releaseFailed("claim-invalid"); + } + const claim = read.claim; + const holders = claim.holders.filter((entry) => entry !== opts.holder); + if (holders.length > 0) { + await writeClaim(opts.spec, { ...claim, holders }); + return { state: "retained" }; + } + if (claim.ownership === "adopted") { + await removeClaim(opts.spec); + return { state: "released-adopted" }; + } + const stopped = await stopVerifiedSharedHostProcess({ + pid: claim.pid, + birth: claim.birth, + graceMs: opts.graceMs ?? 2000, + processes: opts.processes, + }); + switch (stopped) { + case "census-unavailable": + return releaseFailed("census-unavailable"); + case "unconfirmed": + await writeClaim(opts.spec, { ...claim, holders }); + return { state: "stop-unconfirmed" }; + case "absent": + await removeClaim(opts.spec); + return { state: "stale-cleared" }; + default: + await removeClaim(opts.spec); + return { state: "stopped" }; + } +} diff --git a/tests/fixtures/shared-host-helper.ts b/tests/fixtures/shared-host-helper.ts new file mode 100644 index 000000000..5a60a5ca4 --- /dev/null +++ b/tests/fixtures/shared-host-helper.ts @@ -0,0 +1,111 @@ +/** + * Portable shared-host-helper fixture process for + * tests/shared-host-lifecycle.test.ts. + * + * argv: + * - `identity`: HTTP on 127.0.0.1:; `/__hack/identity` reports + * {helper, upstream, pid}; `/` answers `served:`; + * `/__fixture/env` lists this process's environment variable names. + * - `stubborn`: like `identity`, but ignores SIGTERM (only SIGKILL stops it). + * - `oversized`: like `identity`, but pads the identity answer past 4 KiB. + * - `exit`: exits immediately (a helper that dies during startup). + * - `tcp`: a plain TCP listener that never answers (a foreign port owner). + * - `reset`: a TCP listener that drops every connection at once. + * - `placeholder`: stays alive and never listens (a sleeping placeholder). + * - `parent`: spawns a `placeholder` grandchild, prints its PID, then serves + * like `identity` (a helper with its own child process). + * + * Every mode exits once its private HOME (the test's scratch root) is gone, + * so no fixture outlives its test even when a failing run skips cleanup. + */ +import { existsSync } from "node:fs"; + +const [mode, portText, helper, upstream] = process.argv.slice(2); +const port = Number(portText); +const home = process.env.HOME; +setInterval(() => { + if (!(home && existsSync(home))) { + process.exit(0); + } +}, 250); + +function serveIdentity(opts: { readonly padding?: number } = {}): void { + Bun.serve({ + hostname: "127.0.0.1", + port, + fetch(request) { + const url = new URL(request.url); + if (url.pathname === "/__hack/identity") { + return Response.json({ + helper, + upstream, + pid: process.pid, + ...(opts.padding ? { padding: "x".repeat(opts.padding) } : {}), + }); + } + if (url.pathname === "/__fixture/env") { + return Response.json(Object.keys(process.env).sort()); + } + return new Response(`served:${upstream}`); + }, + }); +} + +switch (mode) { + case "identity": + serveIdentity(); + break; + case "stubborn": + process.on("SIGTERM", () => { + // Ignored on purpose: the stop path must escalate to SIGKILL. + }); + serveIdentity(); + break; + case "oversized": + serveIdentity({ padding: 8192 }); + break; + case "exit": + process.exit(3); + break; + case "tcp": + Bun.listen({ + hostname: "127.0.0.1", + port, + socket: { + data() { + // Accepts connections and never answers. + }, + }, + }); + break; + case "reset": + Bun.listen({ + hostname: "127.0.0.1", + port, + socket: { + open(socket) { + // Accepts and drops every connection without answering. + socket.terminate(); + }, + data() { + // Never reached. + }, + }, + }); + break; + case "placeholder": + setInterval(() => undefined, 60_000); + break; + case "parent": { + const child = Bun.spawn( + [process.execPath, import.meta.path, "placeholder", "0", "-", "-"], + { stdin: "ignore", stdout: "ignore", stderr: "ignore" } + ); + console.log(`grandchild ${child.pid}`); + serveIdentity(); + break; + } + default: + process.exit(2); +} +console.log("listening"); diff --git a/tests/fixtures/shared-host-starter.ts b/tests/fixtures/shared-host-starter.ts new file mode 100644 index 000000000..9b0ee3226 --- /dev/null +++ b/tests/fixtures/shared-host-starter.ts @@ -0,0 +1,63 @@ +/** + * Independent starter client for tests/shared-host-lifecycle.test.ts. + * + * argv: [options-json] + * options: { pauseAt?: event, resume?: "never" | "stdin", linger?: boolean } + * + * Prints `event ` for each lifecycle event and `result ` at the + * end. With `pauseAt` it stops after printing that event: forever by default + * (so the test can kill this parent at a precise point, possibly while it + * holds the helper lock), or until a line arrives on stdin with + * `resume: "stdin"`. With `linger` it stays alive after printing the result. + */ +import { + ensureSharedHostHelper, + type SharedHostEvent, + type SharedHostHelperSpec, +} from "../../src/lib/shared-host-lifecycle.ts"; + +type StarterOptions = { + readonly pauseAt?: SharedHostEvent; + readonly resume?: "never" | "stdin"; + readonly linger?: boolean; +}; + +const [specText, holder, optionsText] = process.argv.slice(2); +const spec = JSON.parse(specText ?? "{}") as SharedHostHelperSpec; +const options = JSON.parse(optionsText ?? "{}") as StarterOptions; + +async function waitForStdinLine(): Promise { + const reader = Bun.stdin.stream().getReader(); + const decoder = new TextDecoder(); + try { + while (true) { + const { done, value } = await reader.read(); + if (done || decoder.decode(value, { stream: true }).includes("\n")) { + return; + } + } + } finally { + reader.releaseLock(); + } +} + +const result = await ensureSharedHostHelper({ + spec, + holder: holder ?? "", + observe: async (event) => { + console.log(`event ${event}`); + if (event !== options.pauseAt) { + return; + } + if (options.resume === "stdin") { + await waitForStdinLine(); + return; + } + // An interval keeps the event loop alive until the test kills us. + await new Promise(() => setInterval(() => undefined, 60_000)); + }, +}); +console.log(`result ${JSON.stringify(result)}`); +if (options.linger) { + setInterval(() => undefined, 60_000); +} diff --git a/tests/shared-host-lifecycle.test.ts b/tests/shared-host-lifecycle.test.ts new file mode 100644 index 000000000..f6b684783 --- /dev/null +++ b/tests/shared-host-lifecycle.test.ts @@ -0,0 +1,1576 @@ +import { afterEach, expect, test } from "bun:test"; +import { + chmod, + mkdir, + mkdtemp, + readdir, + readFile, + rm, + writeFile, +} from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join, resolve } from "node:path"; +import { + readProcessIdentities, + selectProcessTree, + signalVerifiedProcessTree, +} from "../src/lib/process-tree.ts"; +import { withProjectsRegistryLock } from "../src/lib/projects-registry-lock.ts"; +import { + ensureSharedHostHelper, + parseSharedHostClaim, + releaseSharedHostHelper, + type SharedHostEvent, + type SharedHostHelperSpec, + type SharedHostProcesses, +} from "../src/lib/shared-host-lifecycle.ts"; +import { restoreEnv } from "./helpers/env.ts"; + +/** + * Portable process fixtures for the shared host helper contract + * (docs/reference/shared-host-lifecycle.md). Every fixture is a local Bun + * process on a loopback port: no Docker, no real HOME, no privileged effect. + * Foreign processes are started and stopped by the test through their own + * handles; the library under test may only signal the helpers it claimed. + */ + +const HELPER = resolve(import.meta.dir, "fixtures/shared-host-helper.ts"); +const STARTER = resolve(import.meta.dir, "fixtures/shared-host-starter.ts"); +const TIMEOUT = 30_000; + +type Foreign = Bun.Subprocess<"ignore", "pipe", "ignore">; +const foreign: Foreign[] = []; +const starters: Bun.Subprocess<"pipe", "pipe", "ignore">[] = []; +const claimedHelpers: { pid: number; birth: string }[] = []; +const roots: string[] = []; + +afterEach(async () => { + for (const proc of [...starters.splice(0), ...foreign.splice(0)]) { + proc.kill("SIGKILL"); + await proc.exited; + } + // Reap every helper the library claimed, through the shared birth-verified + // tree primitives rather than the code under test, so a failing or mutated + // run cannot leak fixture processes onto a shared host. + const owned = [...claimedHelpers.splice(0)]; + for (const root of roots) { + const text = await readFile( + join(root, "state", "issuer-bridge.claim.json"), + "utf8" + ).catch(() => null); + const claim = text === null ? null : parseSharedHostClaim(text); + if (claim?.ownership === "owned") { + owned.push({ pid: claim.pid, birth: claim.birth }); + } + } + const census = await readProcessIdentities(); + for (const identity of owned) { + if ( + census.some( + (row) => row.pid === identity.pid && row.birth === identity.birth + ) + ) { + await signalVerifiedProcessTree( + selectProcessTree(census, identity.pid), + "SIGKILL" + ); + } + } + for (const root of roots.splice(0)) { + await rm(root, { recursive: true, force: true }); + } +}); + +function childEnv(home: string): Record { + return { PATH: process.env.PATH ?? "/usr/bin:/bin", HOME: home }; +} + +async function scratch(): Promise { + const root = await mkdtemp(join(tmpdir(), "hack-shared-host-")); + roots.push(root); + return root; +} + +function freePort(): number { + const probe = Bun.listen({ + hostname: "127.0.0.1", + port: 0, + socket: { + data() { + // Bind probe only. + }, + }, + }); + const port = probe.port; + probe.stop(true); + return port; +} + +function specFor(opts: { + readonly root: string; + readonly port: number; + readonly upstream?: string; + readonly mode?: + | "identity" + | "placeholder" + | "parent" + | "stubborn" + | "oversized" + | "exit"; + readonly backoff?: SharedHostHelperSpec["backoff"]; + readonly probeTimeoutMs?: number; + readonly deadlineMs?: number; +}): SharedHostHelperSpec { + const upstream = opts.upstream ?? "https://issuer.branch-a.test"; + return { + name: "issuer-bridge", + enabled: true, + stateDir: join(opts.root, "state"), + port: opts.port, + identityPath: "/__hack/identity", + expected: { helper: "issuer-bridge", upstream }, + command: [ + process.execPath, + HELPER, + opts.mode ?? "identity", + String(opts.port), + "issuer-bridge", + upstream, + ], + cwd: opts.root, + env: childEnv(opts.root), + deadlineMs: opts.deadlineMs ?? 10_000, + backoff: opts.backoff ?? { initialMs: 25, maxMs: 200 }, + probeTimeoutMs: opts.probeTimeoutMs ?? 500, + }; +} + +/** + * Reads fixture stdout continuously, so `lines` is current even while the + * test is not waiting (a negative check can inspect what was printed). + */ +function lineReader(stream: ReadableStream) { + const reader = stream.getReader(); + const decoder = new TextDecoder(); + const lines: string[] = []; + let buffer = ""; + let closed = false; + const pump = (async () => { + try { + while (true) { + const { done, value } = await reader.read(); + if (done) { + break; + } + buffer += decoder.decode(value, { stream: true }); + const parts = buffer.split("\n"); + buffer = parts.pop() ?? ""; + lines.push(...parts); + } + } catch { + // The fixture was killed; keep what was read. + } finally { + closed = true; + } + })(); + return { + lines, + async waitFor(match: (line: string) => boolean): Promise { + const deadline = performance.now() + 15_000; + while (performance.now() < deadline) { + const found = lines.find(match); + if (found !== undefined) { + return found; + } + if (closed) { + break; + } + await Promise.race([pump, Bun.sleep(10)]); + } + const found = lines.find(match); + if (found === undefined) { + throw new Error(`fixture line not observed; saw ${lines.join(" | ")}`); + } + return found; + }, + }; +} + +async function startForeign(opts: { + readonly mode: "identity" | "tcp" | "reset" | "placeholder" | "oversized"; + readonly port: number; + readonly helper?: string; + readonly upstream?: string; + readonly home: string; +}): Promise { + const proc = Bun.spawn( + [ + process.execPath, + HELPER, + opts.mode, + String(opts.port), + opts.helper ?? "issuer-bridge", + opts.upstream ?? "https://issuer.branch-a.test", + ], + { + env: childEnv(opts.home), + stdin: "ignore", + stdout: "pipe", + stderr: "ignore", + } + ); + foreign.push(proc); + await lineReader(proc.stdout).waitFor((line) => line === "listening"); + return proc; +} + +function startStarter(opts: { + readonly spec: SharedHostHelperSpec; + readonly holder: string; + readonly pauseAt?: SharedHostEvent; + /** Resume a paused starter with {@link resumeStarter} instead of never. */ + readonly resumable?: boolean; + /** Keep the starter process alive after it prints its result. */ + readonly linger?: boolean; + /** The starter client's own time zone. */ + readonly tz?: string; +}) { + const proc = Bun.spawn( + [ + process.execPath, + STARTER, + JSON.stringify(opts.spec), + opts.holder, + JSON.stringify({ + pauseAt: opts.pauseAt, + resume: opts.resumable ? "stdin" : "never", + linger: opts.linger === true, + }), + ], + { + env: { + ...childEnv(opts.spec.cwd), + ...(opts.tz ? { TZ: opts.tz } : {}), + }, + stdin: "pipe", + stdout: "pipe", + stderr: "ignore", + } + ); + starters.push(proc); + return { proc, output: lineReader(proc.stdout) }; +} + +async function resumeStarter(starter: ReturnType) { + starter.proc.stdin.write("resume\n"); + await starter.proc.stdin.flush(); +} + +async function starterResult(starter: ReturnType) { + const line = await starter.output.waitFor((entry) => + entry.startsWith("result ") + ); + return JSON.parse(line.slice("result ".length)) as Record; +} + +async function identityOf(pid: number) { + const row = (await readProcessIdentities()).find( + (entry) => entry.pid === pid + ); + return row ? { pid: row.pid, birth: row.birth } : null; +} + +async function isAlive(identity: { pid: number; birth: string }) { + return (await readProcessIdentities()).some( + (row) => row.pid === identity.pid && row.birth === identity.birth + ); +} + +async function waitGone(identity: { pid: number; birth: string }) { + const deadline = performance.now() + 5000; + while (performance.now() < deadline) { + if (!(await isAlive(identity))) { + return true; + } + await Bun.sleep(50); + } + return false; +} + +async function readClaim(spec: SharedHostHelperSpec) { + const text = await readFile( + join(spec.stateDir, `${spec.name}.claim.json`), + "utf8" + ).catch(() => null); + return text === null ? null : parseSharedHostClaim(text); +} + +async function trackClaimedHelper(spec: SharedHostHelperSpec) { + const claim = await readClaim(spec); + if (claim?.ownership !== "owned") { + throw new Error("expected an owned claim"); + } + const identity = { pid: claim.pid, birth: claim.birth }; + claimedHelpers.push(identity); + return identity; +} + +function recorder() { + const events: SharedHostEvent[] = []; + return { + events, + observe: (event: SharedHostEvent) => { + events.push(event); + }, + }; +} + +test( + "a disabled helper takes no lock, writes no claim and starts no child", + async () => { + const root = await scratch(); + const spec = { ...specFor({ root, port: freePort() }), enabled: false }; + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + expect(result).toEqual({ state: "disabled", ownership: "disabled" }); + expect(seen.events).toEqual([]); + expect(await readdir(root)).toEqual([]); + }, + TIMEOUT +); + +test( + "an owned helper becomes ready by identity, serves its branch upstream and is stopped by its last holder", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + const owned = await trackClaimedHelper(spec); + expect(result).toMatchObject({ + state: "ready", + ownership: "owned", + spawned: true, + pid: owned.pid, + endpoint: `http://127.0.0.1:${spec.port}`, + }); + expect(seen.events).toContain("claimed"); + expect(seen.events.indexOf("claimed")).toBeLessThan( + seen.events.indexOf("gate-opened") + ); + expect(await (await fetch(`http://127.0.0.1:${spec.port}/`)).text()).toBe( + "served:https://issuer.branch-a.test" + ); + expect(await readClaim(spec)).toMatchObject({ + phase: "ready", + holders: ["branch-a"], + }); + + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "stopped" } + ); + expect(await isAlive(owned)).toBe(false); + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +); + +test( + "a healthy shared helper started elsewhere is adopted and never signalled on release", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const existing = await startForeign({ + mode: "identity", + port: spec.port, + home: root, + }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + expect(result).toEqual({ + state: "ready", + ownership: "adopted", + endpoint: `http://127.0.0.1:${spec.port}`, + spawned: false, + }); + expect(seen.events).not.toContain("spawned"); + expect(await (await fetch(`http://127.0.0.1:${spec.port}/`)).text()).toBe( + "served:https://issuer.branch-a.test" + ); + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "released-adopted" } + ); + await Bun.sleep(200); + expect(existing.exitCode).toBeNull(); + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +); + +test( + "a helper serving the wrong upstream is refused and left running", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const existing = await startForeign({ + mode: "identity", + port: spec.port, + upstream: "https://issuer.production.example", + home: root, + }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + expect(result).toMatchObject({ state: "failed", reason: "wrong-upstream" }); + expect(JSON.stringify(result)).not.toContain("production"); + expect(seen.events).not.toContain("spawned"); + expect(await readClaim(spec)).toBeNull(); + await Bun.sleep(200); + expect(existing.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "foreign listeners on the helper port are refused and preserved", + async () => { + for (const fixture of [ + { mode: "tcp" as const }, + { mode: "reset" as const }, + { mode: "identity" as const, helper: "unrelated-dev-server" }, + ]) { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), deadlineMs: 3000 }); + const existing = await startForeign({ + ...fixture, + port: spec.port, + home: root, + }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + expect(result).toMatchObject({ + state: "failed", + reason: "foreign-listener", + }); + expect(seen.events).not.toContain("spawned"); + await Bun.sleep(200); + expect(existing.exitCode).toBeNull(); + } + }, + TIMEOUT +); + +test( + "independent concurrent starters are serialized onto one owned helper; sibling holders and branches are preserved", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + // Barrier: the first starter holds the lock until it is resumed, and the + // second is already contending for it, so both runs overlap for certain. + const first = startStarter({ + spec, + holder: "branch-a", + pauseAt: "lock-acquired", + resumable: true, + }); + await first.output.waitFor((line) => line === "event lock-acquired"); + const second = startStarter({ spec, holder: "branch-b" }); + await second.output.waitFor((line) => line === "event lock-waiting"); + await Bun.sleep(300); + expect(second.output.lines).not.toContain("event lock-acquired"); + await resumeStarter(first); + const results = await Promise.all([ + starterResult(first), + starterResult(second), + ]); + const owned = await trackClaimedHelper(spec); + expect(results.map((entry) => entry.state)).toEqual(["ready", "ready"]); + expect(results.map((entry) => entry.ownership)).toEqual(["owned", "owned"]); + expect(results.map((entry) => entry.pid)).toEqual([owned.pid, owned.pid]); + expect(results.filter((entry) => entry.spawned === true)).toHaveLength(1); + const spawnedEvents = [first, second].filter((starter) => + starter.output.lines.includes("event spawned") + ); + expect(spawnedEvents).toHaveLength(1); + expect([...((await readClaim(spec))?.holders ?? [])].sort()).toEqual([ + "branch-a", + "branch-b", + ]); + + // A sibling branch with a different upstream cannot take over the helper. + const sibling = await ensureSharedHostHelper({ + spec: specFor({ + root, + port: spec.port, + upstream: "https://issuer.branch-c.test", + }), + holder: "branch-c", + }); + expect(sibling).toMatchObject({ + state: "failed", + reason: "wrong-upstream", + }); + + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "retained" } + ); + expect(await isAlive(owned)).toBe(true); + expect(await (await fetch(`http://127.0.0.1:${spec.port}/`)).text()).toBe( + "served:https://issuer.branch-a.test" + ); + expect(await releaseSharedHostHelper({ spec, holder: "branch-b" })).toEqual( + { state: "stopped" } + ); + expect(await isAlive(owned)).toBe(false); + }, + TIMEOUT +); + +test( + "a stale claim whose PID was reused is discarded without signalling the live process", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const bystander = await startForeign({ + mode: "placeholder", + port: 0, + home: root, + }); + const reused = await identityOf(bystander.pid); + expect(reused).not.toBeNull(); + const staleClaim = { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: bystander.pid, + birth: "Thu Jan 1 00:00:00 1970", + holders: ["branch-old"], + }; + await mkdir(spec.stateDir, { recursive: true, mode: 0o700 }); + const claimFile = join(spec.stateDir, `${spec.name}.claim.json`); + await writeFile(claimFile, JSON.stringify(staleClaim), { mode: 0o600 }); + + // Release by the last holder must not signal the reused PID. + expect( + await releaseSharedHostHelper({ spec, holder: "branch-old" }) + ).toEqual({ state: "stale-cleared" }); + await Bun.sleep(200); + expect(bystander.exitCode).toBeNull(); + + // A starter treats the same stale claim as no ownership and starts fresh. + await writeFile(claimFile, JSON.stringify(staleClaim), { mode: 0o600 }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + const owned = await trackClaimedHelper(spec); + expect(seen.events).toContain("claim-stale"); + expect(result).toMatchObject({ state: "ready", ownership: "owned" }); + expect(owned.pid).not.toBe(bystander.pid); + expect((await readClaim(spec))?.holders).toEqual([ + "branch-old", + "branch-a", + ]); + await Bun.sleep(200); + expect(bystander.exitCode).toBeNull(); + expect(await isAlive(reused ?? owned)).toBe(true); + }, + TIMEOUT +); + +test( + "a crashed owned child is restarted under a fresh claim by the next starter", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const first = await ensureSharedHostHelper({ spec, holder: "branch-a" }); + const crashed = await trackClaimedHelper(spec); + expect(first).toMatchObject({ state: "ready", spawned: true }); + process.kill(crashed.pid, "SIGKILL"); + expect(await waitGone(crashed)).toBe(true); + + const seen = recorder(); + const second = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + const restarted = await trackClaimedHelper(spec); + expect(seen.events).toContain("claim-stale"); + expect(second).toMatchObject({ + state: "ready", + ownership: "owned", + spawned: true, + pid: restarted.pid, + }); + expect(restarted).not.toEqual(crashed); + expect((await readClaim(spec))?.holders).toEqual(["branch-a", "branch-b"]); + expect(await (await fetch(`http://127.0.0.1:${spec.port}/`)).text()).toBe( + "served:https://issuer.branch-a.test" + ); + }, + TIMEOUT +); + +test( + "cancellation during bounded readiness retries stops the owned placeholder and keeps foreign processes", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), mode: "placeholder" }); + const bystander = await startForeign({ + mode: "tcp", + port: freePort(), + home: root, + }); + const controller = new AbortController(); + let owned: { pid: number; birth: string } | null = null; + let retries = 0; + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + signal: controller.signal, + observe: async (event) => { + if (event === "claimed") { + owned = await trackClaimedHelper(spec); + } + if (event === "probe-retry") { + retries += 1; + if (retries === 3) { + controller.abort(); + } + } + }, + }); + expect(result).toMatchObject({ state: "failed", reason: "cancelled" }); + expect(retries).toBe(3); + expect(owned).not.toBeNull(); + expect(await isAlive(owned ?? { pid: 0, birth: "" })).toBe(false); + expect(await readClaim(spec)).toBeNull(); + expect(bystander.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a live placeholder is never readiness: the deadline bounds retries and stops the owned child", + async () => { + const root = await scratch(); + const spec = specFor({ + root, + port: freePort(), + mode: "placeholder", + deadlineMs: 800, + }); + let owned: { pid: number; birth: string } | null = null; + const started = performance.now(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: async (event) => { + if (event === "claimed") { + owned = await trackClaimedHelper(spec); + } + }, + }); + expect(result).toMatchObject({ state: "failed", reason: "deadline" }); + expect(performance.now() - started).toBeLessThan(5000); + expect(await isAlive(owned ?? { pid: 0, birth: "" })).toBe(false); + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +); + +test( + "a reused ready claim is not ready when a different process answers its identity, and neither process is signalled", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), deadlineMs: 2000 }); + const claimedProcess = await startForeign({ + mode: "placeholder", + port: 0, + home: root, + }); + const impostor = await startForeign({ + mode: "identity", + port: spec.port, + home: root, + }); + const identity = await identityOf(claimedProcess.pid); + await mkdir(spec.stateDir, { recursive: true, mode: 0o700 }); + await writeFile( + join(spec.stateDir, `${spec.name}.claim.json`), + JSON.stringify({ + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: claimedProcess.pid, + birth: identity?.birth ?? "", + holders: ["branch-a"], + }), + { mode: 0o600 } + ); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + expect(seen.events).toContain("claim-reused"); + expect(result).toMatchObject({ + state: "failed", + reason: "foreign-listener", + }); + expect((await readClaim(spec))?.holders).toEqual(["branch-a"]); + await Bun.sleep(200); + expect(claimedProcess.exitCode).toBeNull(); + expect(impostor.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a starter lost before its claim opens the gate leaves no helper running", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const starter = startStarter({ + spec, + holder: "branch-a", + pauseAt: "claimed", + }); + await starter.output.waitFor((line) => line === "event claimed"); + const gated = await trackClaimedHelper(spec); + starter.proc.kill("SIGKILL"); + await starter.proc.exited; + expect(await waitGone(gated)).toBe(true); + const port = await fetch(`http://127.0.0.1:${spec.port}/`).catch( + () => null + ); + expect(port).toBeNull(); + + const seen = recorder(); + const next = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + await trackClaimedHelper(spec); + expect(seen.events).toContain("claim-stale"); + expect(next).toMatchObject({ state: "ready", spawned: true }); + }, + TIMEOUT +); + +test( + "a starter lost after readiness leaves a claimed helper that another client reuses and releases", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const starter = startStarter({ + spec, + holder: "branch-a", + pauseAt: "ready", + }); + await starter.output.waitFor((line) => line === "event ready"); + const owned = await trackClaimedHelper(spec); + starter.proc.kill("SIGKILL"); + await starter.proc.exited; + await Bun.sleep(200); + expect(await isAlive(owned)).toBe(true); + + const seen = recorder(); + const reused = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + expect(seen.events).toContain("claim-reused"); + expect(reused).toMatchObject({ + state: "ready", + ownership: "owned", + spawned: false, + pid: owned.pid, + }); + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "retained" } + ); + expect(await releaseSharedHostHelper({ spec, holder: "branch-b" })).toEqual( + { state: "stopped" } + ); + expect(await isAlive(owned)).toBe(false); + }, + TIMEOUT +); + +test( + "release stops the owned helper's own descendants and nothing else", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), mode: "parent" }); + const bystander = await startForeign({ + mode: "placeholder", + port: 0, + home: root, + }); + const result = await ensureSharedHostHelper({ spec, holder: "branch-a" }); + expect(result).toMatchObject({ state: "ready", ownership: "owned" }); + const owned = await trackClaimedHelper(spec); + const grandchildRow = (await readProcessIdentities()).find( + (row) => row.parentPid === owned.pid + ); + expect(grandchildRow).toBeDefined(); + const grandchild = { + pid: grandchildRow?.pid ?? 0, + birth: grandchildRow?.birth ?? "", + }; + claimedHelpers.push(grandchild); + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "stopped" } + ); + expect(await isAlive(owned)).toBe(false); + expect(await waitGone(grandchild)).toBe(true); + expect(bystander.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "an unreadable claim fails closed for start and release", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + await mkdir(spec.stateDir, { recursive: true, mode: 0o700 }); + await writeFile( + join(spec.stateDir, `${spec.name}.claim.json`), + '{"version":1,"helper":"issuer-bridge","ownership":"owned"}', + { mode: 0o600 } + ); + const seen = recorder(); + expect( + await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }) + ).toMatchObject({ state: "failed", reason: "claim-invalid" }); + expect(seen.events).not.toContain("spawned"); + expect( + await releaseSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "failed", reason: "claim-invalid" }); + }, + TIMEOUT +); + +test( + "an unavailable process census fails closed: no spawn, no signal, claim kept", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const bystander = await startForeign({ + mode: "placeholder", + port: 0, + home: root, + }); + const identity = await identityOf(bystander.pid); + await mkdir(spec.stateDir, { recursive: true, mode: 0o700 }); + const claimFile = join(spec.stateDir, `${spec.name}.claim.json`); + const claim = JSON.stringify({ + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: bystander.pid, + birth: identity?.birth ?? "", + holders: ["branch-a"], + }); + await writeFile(claimFile, claim, { mode: 0o600 }); + const bin = join(root, "bin"); + await mkdir(bin); + await writeFile(join(bin, "ps"), "#!/bin/sh\nexit 1\n", { mode: 0o755 }); + const savedPath = process.env.PATH; + process.env.PATH = `${bin}:${savedPath ?? "/usr/bin:/bin"}`; + try { + const seen = recorder(); + expect( + await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }) + ).toMatchObject({ state: "failed", reason: "census-unavailable" }); + expect(seen.events).not.toContain("spawned"); + expect( + await releaseSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "failed", reason: "census-unavailable" }); + } finally { + restoreEnv("PATH", savedPath); + } + expect(await readFile(claimFile, "utf8")).toBe(claim); + await Bun.sleep(200); + expect(bystander.exitCode).toBeNull(); + }, + TIMEOUT +); + +test("claim parser accepts only the exact v1 shapes", () => { + const owned = { + version: 1, + helper: "issuer-bridge", + port: 4100, + ownership: "owned", + phase: "ready", + pid: 4242, + birth: "Mon Oct 19 10:00:00 2026", + holders: ["branch-a"], + }; + const adopted = { + version: 1, + helper: "issuer-bridge", + port: 4100, + ownership: "adopted", + holders: ["branch-a"], + }; + expect(parseSharedHostClaim(JSON.stringify(owned))).toEqual(owned as never); + expect(parseSharedHostClaim(JSON.stringify(adopted))).toEqual( + adopted as never + ); + for (const invalid of [ + { ...owned, pid: 1 }, + { ...owned, birth: "" }, + { ...owned, phase: "failed" }, + { ...owned, extra: true }, + { ...owned, holders: ["branch-a", "branch-a"] }, + { ...owned, holders: ["../escape"] }, + { ...adopted, pid: 4242 }, + { ...adopted, version: 2 }, + { ...adopted, port: 70_000 }, + ]) { + expect(parseSharedHostClaim(JSON.stringify(invalid))).toBeNull(); + } + expect(parseSharedHostClaim("not json")).toBeNull(); +}); + +async function writeClaimFile( + spec: SharedHostHelperSpec, + value: unknown, + opts: { readonly mode?: number; readonly padding?: number } = {} +) { + await mkdir(spec.stateDir, { recursive: true, mode: 0o700 }); + const path = join(spec.stateDir, `${spec.name}.claim.json`); + const text = `${JSON.stringify(value)}${" ".repeat(opts.padding ?? 0)}`; + await writeFile(path, text, { mode: 0o600 }); + await chmod(path, opts.mode ?? 0o600); + return { path, text }; +} + +async function processGroupOf(pid: number): Promise { + const proc = Bun.spawn(["ps", "-o", "pgid=", "-p", String(pid)], { + env: { PATH: process.env.PATH ?? "/usr/bin:/bin", LC_ALL: "C" }, + stdin: "ignore", + stdout: "pipe", + stderr: "ignore", + }); + const text = await new Response(proc.stdout).text(); + await proc.exited; + return Number(text.trim()); +} + +test( + "clients in different time zones agree on an owned helper's birth: it is reused, stays owned and is stopped by its last holder", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + // POSIX zone strings need no zoneinfo, so the zones differ on minimal hosts too. + const first = startStarter({ spec, holder: "branch-a", tz: "UTC0" }); + expect(await starterResult(first)).toMatchObject({ + state: "ready", + ownership: "owned", + spawned: true, + }); + const owned = await trackClaimedHelper(spec); + const second = startStarter({ spec, holder: "branch-b", tz: "HST10" }); + expect(await starterResult(second)).toMatchObject({ + state: "ready", + ownership: "owned", + spawned: false, + pid: owned.pid, + }); + expect(second.output.lines).toContain("event claim-reused"); + expect(second.output.lines).not.toContain("event claim-stale"); + expect(second.output.lines).not.toContain("event adopted"); + expect(await readClaim(spec)).toMatchObject({ + ownership: "owned", + pid: owned.pid, + birth: owned.birth, + holders: ["branch-a", "branch-b"], + }); + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "retained" } + ); + expect(await releaseSharedHostHelper({ spec, holder: "branch-b" })).toEqual( + { state: "stopped" } + ); + expect(await isAlive(owned)).toBe(false); + }, + TIMEOUT +); + +test( + "a starter lost after opening the gate leaves a starting helper that the next starter stops when it never becomes ready", + async () => { + const root = await scratch(); + const spec = specFor({ + root, + port: freePort(), + mode: "placeholder", + deadlineMs: 1500, + }); + const starter = startStarter({ + spec, + holder: "branch-a", + pauseAt: "gate-opened", + }); + await starter.output.waitFor((line) => line === "event gate-opened"); + const abandoned = await trackClaimedHelper(spec); + starter.proc.kill("SIGKILL"); + await starter.proc.exited; + expect(await readClaim(spec)).toMatchObject({ phase: "starting" }); + expect(await isAlive(abandoned)).toBe(true); + + const seen = recorder(); + const next = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + expect(next).toMatchObject({ state: "failed", reason: "deadline" }); + expect(seen.events).toContain("claim-reused"); + expect(seen.events).toContain("stopped-after-failure"); + expect(await waitGone(abandoned)).toBe(true); + expect(await readClaim(spec)).toBeNull(); + + // The helper port is free again for a fresh, working start. + const fresh = await ensureSharedHostHelper({ + spec: specFor({ root, port: spec.port }), + holder: "branch-c", + }); + await trackClaimedHelper(spec); + expect(fresh).toMatchObject({ state: "ready", spawned: true }); + }, + TIMEOUT +); + +test( + "a lock wait that uses up the deadline reports deadline, never a foreign listener, and adopts nothing", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), deadlineMs: 600 }); + const healthy = await startForeign({ + mode: "identity", + port: spec.port, + home: root, + }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: async (event) => { + seen.observe(event); + if (event === "lock-acquired") { + // Stands in for lock contention that consumed the whole deadline. + await Bun.sleep(700); + } + }, + }); + expect(result).toMatchObject({ state: "failed", reason: "deadline" }); + expect(seen.events).not.toContain("adopted"); + expect(await readClaim(spec)).toBeNull(); + expect(healthy.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a port probe cut short by the deadline reports deadline; a full-length silent probe is a foreign listener", + async () => { + const root = await scratch(); + const port = freePort(); + const silent = await startForeign({ mode: "tcp", port, home: root }); + const cut = await ensureSharedHostHelper({ + spec: specFor({ root, port, deadlineMs: 600, probeTimeoutMs: 5000 }), + holder: "branch-a", + }); + expect(cut).toMatchObject({ state: "failed", reason: "deadline" }); + const full = await ensureSharedHostHelper({ + spec: specFor({ root, port, deadlineMs: 5000, probeTimeoutMs: 300 }), + holder: "branch-a", + }); + expect(full).toMatchObject({ state: "failed", reason: "foreign-listener" }); + expect(silent.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a stale owned claim whose PID still answers as the helper is kept, not adopted, and nothing is signalled", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const answering = await startForeign({ + mode: "identity", + port: spec.port, + home: root, + }); + const { path, text } = await writeClaimFile(spec, { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: answering.pid, + birth: "Thu Jan 1 00:00:00 1970", + holders: ["branch-a"], + }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }); + expect(result).toMatchObject({ state: "failed", reason: "claim-invalid" }); + expect(seen.events).toContain("claim-stale"); + expect(seen.events).not.toContain("adopted"); + expect(await readFile(path, "utf8")).toBe(text); + await Bun.sleep(200); + expect(answering.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a claim for another port, a group-readable claim and an oversized claim fail closed", + async () => { + const root = await scratch(); + const bystander = await startForeign({ + mode: "placeholder", + port: 0, + home: root, + }); + const identity = await identityOf(bystander.pid); + const cases = [ + { label: "other port", portOffset: 1, ownership: "owned" }, + { label: "group-readable", mode: 0o640, ownership: "owned" }, + { label: "oversized", padding: 9000, ownership: "adopted" }, + ] as const; + for (const entry of cases) { + const spec = specFor({ root: await scratch(), port: freePort() }); + const portOffset = "portOffset" in entry ? entry.portOffset : 0; + const claim = + entry.ownership === "owned" + ? { + version: 1, + helper: spec.name, + port: spec.port + portOffset, + ownership: "owned", + phase: "ready", + pid: bystander.pid, + birth: identity?.birth ?? "", + holders: ["branch-a"], + } + : { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "adopted", + holders: ["branch-a"], + }; + const { path, text } = await writeClaimFile(spec, claim, { + mode: "mode" in entry ? entry.mode : undefined, + padding: "padding" in entry ? entry.padding : undefined, + }); + const seen = recorder(); + expect({ + label: entry.label, + result: await ensureSharedHostHelper({ + spec, + holder: "branch-b", + observe: seen.observe, + }), + }).toMatchObject({ + label: entry.label, + result: { state: "failed", reason: "claim-invalid" }, + }); + expect(seen.events).not.toContain("spawned"); + expect(seen.events).not.toContain("claim-reused"); + if (portOffset === 0) { + expect( + await releaseSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "failed", reason: "claim-invalid" }); + } + expect(await readFile(path, "utf8")).toBe(text); + } + await Bun.sleep(200); + expect(bystander.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a helper that exits during startup reports child-exited and leaves no claim", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), mode: "exit" }); + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }); + expect(result).toMatchObject({ state: "failed", reason: "child-exited" }); + expect(seen.events).toContain("gate-opened"); + expect(seen.events).toContain("stopped-after-failure"); + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +); + +test( + "cancellation after the claim and before the gate never runs the helper", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const controller = new AbortController(); + let gated: { pid: number; birth: string } | null = null; + const seen = recorder(); + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + signal: controller.signal, + observe: async (event) => { + seen.observe(event); + if (event === "claimed") { + gated = await trackClaimedHelper(spec); + controller.abort(); + } + }, + }); + expect(result).toMatchObject({ state: "failed", reason: "cancelled" }); + expect(seen.events).not.toContain("gate-opened"); + expect(await waitGone(gated ?? { pid: 0, birth: "" })).toBe(true); + expect(await readClaim(spec)).toBeNull(); + expect( + await fetch(`http://127.0.0.1:${spec.port}/`).catch(() => null) + ).toBeNull(); + }, + TIMEOUT +); + +test( + "a helper that ignores SIGTERM is stopped by SIGKILL after the grace period", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort(), mode: "stubborn" }); + expect( + await ensureSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "ready", ownership: "owned" }); + const owned = await trackClaimedHelper(spec); + expect( + await releaseSharedHostHelper({ spec, holder: "branch-a", graceMs: 200 }) + ).toEqual({ state: "stopped" }); + expect(await isAlive(owned)).toBe(false); + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +); + +test( + "an owned helper runs in its own session with only the caller's environment", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const saved = process.env.HACK_SHARED_HOST_RUNNER_ONLY; + process.env.HACK_SHARED_HOST_RUNNER_ONLY = "runner-value"; + try { + expect( + await ensureSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "ready", ownership: "owned" }); + } finally { + restoreEnv("HACK_SHARED_HOST_RUNNER_ONLY", saved); + } + const owned = await trackClaimedHelper(spec); + const keys = (await ( + await fetch(`http://127.0.0.1:${spec.port}/__fixture/env`) + ).json()) as string[]; + expect(keys).toContain("HOME"); + expect(keys).not.toContain("HACK_SHARED_HOST_RUNNER_ONLY"); + const helperGroup = await processGroupOf(owned.pid); + expect(helperGroup).toBe(owned.pid); + expect(helperGroup).not.toBe(await processGroupOf(process.pid)); + }, + TIMEOUT +); + +test( + "cancellation interrupts the readiness backoff sleep", + async () => { + const root = await scratch(); + const spec = specFor({ + root, + port: freePort(), + mode: "placeholder", + deadlineMs: 20_000, + backoff: { initialMs: 5000, maxMs: 5000 }, + }); + const controller = new AbortController(); + let abortedAt = 0; + let owned: { pid: number; birth: string } | null = null; + const result = await ensureSharedHostHelper({ + spec, + holder: "branch-a", + signal: controller.signal, + observe: async (event) => { + if (event === "claimed") { + owned = await trackClaimedHelper(spec); + } + if (event === "probe-retry") { + setTimeout(() => { + abortedAt = performance.now(); + controller.abort(); + }, 50); + } + }, + }); + expect(result).toMatchObject({ state: "failed", reason: "cancelled" }); + expect(abortedAt).toBeGreaterThan(0); + expect(performance.now() - abortedAt).toBeLessThan(2500); + expect(await isAlive(owned ?? { pid: 0, birth: "" })).toBe(false); + }, + TIMEOUT +); + +test( + "release waits for the helper lock and fails closed when it is held past the timeout", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + expect( + await ensureSharedHostHelper({ spec, holder: "branch-a" }) + ).toMatchObject({ state: "ready", ownership: "owned" }); + const owned = await trackClaimedHelper(spec); + const claimFile = join(spec.stateDir, `${spec.name}.claim.json`); + const before = await readFile(claimFile, "utf8"); + const blocked = await withProjectsRegistryLock({ + lockPath: join(spec.stateDir, `${spec.name}.lock`), + label: "test holder", + run: () => + releaseSharedHostHelper({ spec, holder: "branch-a", timeoutMs: 300 }), + }); + expect(blocked).toMatchObject({ + state: "failed", + reason: "lock-unavailable", + }); + expect(await isAlive(owned)).toBe(true); + expect(await readFile(claimFile, "utf8")).toBe(before); + expect(await releaseSharedHostHelper({ spec, holder: "branch-a" })).toEqual( + { state: "stopped" } + ); + }, + TIMEOUT +); + +test( + "an identity answer larger than 4 KiB is not identity: the listener is refused and preserved", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const oversized = await startForeign({ + mode: "oversized", + port: spec.port, + home: root, + }); + const seen = recorder(); + expect( + await ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: seen.observe, + }) + ).toMatchObject({ state: "failed", reason: "foreign-listener" }); + expect(seen.events).not.toContain("adopted"); + expect(await readClaim(spec)).toBeNull(); + await Bun.sleep(200); + expect(oversized.exitCode).toBeNull(); + }, + TIMEOUT +); + +test( + "a stop that cannot be confirmed keeps the claim so the process stays recorded", + async () => { + // No real process can portably survive SIGKILL, so the census and signal + // boundary is simulated: the claimed helper and its child stay in every + // census whatever is sent to them. + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const birth = "Thu Jan 1 00:00:00 1970"; + const helper = { pid: 4242, parentPid: 1, birth, zombie: false }; + const child = { pid: 4243, parentPid: 4242, birth, zombie: false }; + const self = { + pid: process.pid, + parentPid: process.ppid, + birth, + zombie: false, + }; + await writeClaimFile(spec, { + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: helper.pid, + birth, + holders: ["branch-a"], + }); + const signals: [NodeJS.Signals, number[]][] = []; + const processes: SharedHostProcesses = { + census: async () => [self, helper, child], + signalTree: async (tree, signal) => { + signals.push([signal, tree.map((row) => row.pid)]); + }, + }; + expect( + await releaseSharedHostHelper({ + spec, + holder: "branch-a", + graceMs: 100, + processes, + }) + ).toEqual({ state: "stop-unconfirmed" }); + // TERM, then KILL, each to the whole captured tree and nothing else. + expect(signals).toEqual([ + ["SIGTERM", [helper.pid, child.pid]], + ["SIGKILL", [helper.pid, child.pid]], + ]); + // The claim keeps the exact process recorded, now with no holders. + expect(await readClaim(spec)).toEqual({ + version: 1, + helper: spec.name, + port: spec.port, + ownership: "owned", + phase: "ready", + pid: helper.pid, + birth, + holders: [], + }); + }, + TIMEOUT +); + +test( + "an exited helper its parent has not reaped yet is stopped, not unconfirmed", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + // The starter stays alive and is then frozen, so the helper it spawned + // cannot be reaped: once it exits on SIGTERM it is a zombie in the census. + const starter = startStarter({ spec, holder: "branch-a", linger: true }); + expect(await starterResult(starter)).toMatchObject({ + state: "ready", + ownership: "owned", + }); + const owned = await trackClaimedHelper(spec); + // process.kill, not Subprocess.kill: Bun 1.4.2 on macOS delivers SIGCONT + // for Subprocess.kill("SIGSTOP"), which would leave the starter running. + process.kill(starter.proc.pid, "SIGSTOP"); + try { + expect( + await releaseSharedHostHelper({ + spec, + holder: "branch-a", + graceMs: 1000, + }) + ).toEqual({ state: "stopped" }); + expect(await readClaim(spec)).toBeNull(); + // The helper really is an unreaped zombie, so this run exercised the + // zombie path rather than an ordinary reaped exit. + const row = (await readProcessIdentities()).find( + (entry) => entry.pid === owned.pid && entry.birth === owned.birth + ); + expect(row).toMatchObject({ parentPid: starter.proc.pid, zombie: true }); + } finally { + process.kill(starter.proc.pid, "SIGCONT"); + } + expect(await waitGone(owned)).toBe(true); + }, + TIMEOUT +); + +test( + "an observer error is rethrown after the waiting gate is closed and its claim removed", + async () => { + const root = await scratch(); + const spec = specFor({ root, port: freePort() }); + const children = async () => + (await readProcessIdentities()) + .filter((row) => row.parentPid === process.pid) + .map((row) => ({ pid: row.pid, birth: row.birth })); + const before = new Set((await children()).map((row) => row.pid)); + let gates: { pid: number; birth: string }[] = []; + const failure = new Error("observer failed on purpose"); + await expect( + ensureSharedHostHelper({ + spec, + holder: "branch-a", + observe: async (event) => { + if (event === "spawned") { + // Each census sees its own short-lived `ps` child; the gate is + // the new child present in two consecutive censuses. + const first = await children(); + const second = await children(); + gates = first.filter( + (row) => + !before.has(row.pid) && + second.some( + (entry) => entry.pid === row.pid && entry.birth === row.birth + ) + ); + claimedHelpers.push(...gates); + throw failure; + } + }, + }) + ).rejects.toBe(failure); + expect(gates.length).toBeGreaterThan(0); + for (const gate of gates) { + expect(await waitGone(gate)).toBe(true); + } + expect(await readClaim(spec)).toBeNull(); + }, + TIMEOUT +);