diff --git a/src/frontdoor/commands.ts b/src/frontdoor/commands.ts index b0f38e0c..f7406671 100644 --- a/src/frontdoor/commands.ts +++ b/src/frontdoor/commands.ts @@ -12,7 +12,7 @@ import type { LayoutPaths } from "../paths"; import { realSessionFs } from "../agents/realFs"; import { collectSessionList, showSession } from "../agents/sessions"; import { listIdentities } from "../identityStore"; -import { realFarmFs, realFsPort, realHeadroomSocketTrust, realIsProcessRunning, realCredentialPort, realOwnExecutablePath, spawnDetachedSupervisor, spawnSelfDetached } from "../realPorts"; +import { realFarmFs, realFsPort, realHeadroomSocketTrust, realIsProcessRunning, realCredentialPort, realOwnExecutablePath, selfInvocation, spawnDetachedSupervisor, spawnSelfDetached } from "../realPorts"; import { checkForUpdate } from "../update/commands"; import { identityExists, listAnchoredIdentities } from "../identityStore"; import { readJobs } from "../jobStore"; @@ -63,18 +63,27 @@ import { answerRcControlRequest, authenticateRcSessionMcpServer, endRcSession, g import { RC_STREAM_BACKOFF_MS, createRcEventFanout, createRcStreamHub, type RcStreamHub } from "./rcStream"; import type { RoutedRequest } from "./route"; import { createFrontDoorServer, listenFrontDoor } from "./server"; -import { DOOR_STALL_BOUND_MS, DOOR_WATCHDOG_CHECK_MS, startThreadWatchdog } from "./watchdog"; +import { DOOR_AWAIT_PID_ENV, DOOR_RESPAWN_MIN_SERVING_MS, DOOR_STALL_BOUND_MS, DOOR_WATCHDOG_CHECK_MS, startThreadWatchdog } from "./watchdog"; import { readFrontDoorListeners, realFrontDoorListenerProbes } from "./listenerReadings"; import { collectFrontDoorStatus, formatFrontDoorListeners, formatFrontDoorStatus, headroomSocketTarget } from "./status"; import { liveSessionTokens, writeFrontDoorSession } from "./state"; import { declareThisProcessTheFrontDoor } from "./selfIdentity"; -import { runFrontDoorSupervisor, watchedInstallEntry, type FrontDoorSupervisorPorts, type OwnExecutableStat } from "./supervisor"; +import { awaitPidExit, runFrontDoorSupervisor, watchedInstallEntry, type FrontDoorSupervisorPorts, type OwnExecutableStat } from "./supervisor"; import { createAccountReader } from "../usage/account"; import { createUsageMiddleware } from "../usage/middleware"; import { withQuotaRefresh } from "../usage/quotaRefresh"; import { createRealQuotaRefresher } from "../usage/realQuotaRefresher"; import { createUsageStore } from "../usage/store"; +/** How a door's watchdog starts a replacement: this very executable running the supervisor subcommand, pinned to this door's state root exactly as `spawnDetachedSupervisor` pins it. Every environment value is a string, as the watchdog thread's `spawn` requires. */ +function respawnInvocation(paths: LayoutPaths): { readonly command: string; readonly args: readonly string[]; readonly env: Readonly> } { + const env: Record = {}; + for (const [name, value] of Object.entries({ ...process.env, AGENT_SHIM_HOME: paths.root })) { + env[name] = value; + } + return { ...selfInvocation(["__frontdoor-supervisor"]), env }; +} + function appendLog(paths: LayoutPaths, line: string): void { fs.mkdirSync(paths.logsDir, { recursive: true }); fs.appendFileSync(paths.frontdoorLogPath, `${new Date().toISOString()} ${line}\n`); @@ -531,7 +540,16 @@ function realFrontDoorSupervisorPorts(paths: LayoutPaths): FrontDoorSupervisorPo }, closeRcStreamHub: rcHub.close, startWatchdog: () => - startThreadWatchdog({ stateFile: paths.frontdoorStateFile, ownPid: process.pid, boundMs: DOOR_STALL_BOUND_MS, checkEveryMs: DOOR_WATCHDOG_CHECK_MS, action: "kill-process" }), + startThreadWatchdog({ + stateFile: paths.frontdoorStateFile, + ownPid: process.pid, + boundMs: DOOR_STALL_BOUND_MS, + checkEveryMs: DOOR_WATCHDOG_CHECK_MS, + action: "kill-process", + logFile: paths.frontdoorLogPath, + respawn: respawnInvocation(paths), + respawnAfterServingMs: DOOR_RESPAWN_MIN_SERVING_MS, + }), // The launch lifecycle publisher over the same backbone the typed API serves: the tick's registry facts go in here, and the door's own register, prune and end moments come out as source-tagged events. observeLaunchRegistry: createLaunchEventPublisher(doorEvents, () => Date.now()).observe, doorHealth, @@ -597,15 +615,28 @@ export function registerFrontDoorCommand(program: Command, deps: CommandDeps): v .allowUnknownOption() .action(async () => { const idleShutdownMinutes = readGlobalConfig(paths)?.frontdoor?.idleShutdownMinutes ?? FRONTDOOR_DEFAULT_IDLE_SHUTDOWN_MINUTES; - // The listeners are servers inside this process, so an orderly exit takes them with it: there is no child left behind to stop. - process.on("SIGTERM", () => { - deps.exit(0); - }); - process.on("SIGINT", () => { - deps.exit(0); + const supervisorPorts = realFrontDoorSupervisorPorts(paths); + const ownPid = process.pid; + // Every way this process can end writes its reason to the log first, because the log is the only record that outlives it: a door that exits silently looks, from outside, like one that was never there. + process.on("exit", (code) => { + supervisorPorts.log(`frontdoor ${String(ownPid)}: exiting with code ${String(code)}`); }); + // The listeners are servers inside this process, so an orderly exit takes them with it: there is no child left behind to stop. + for (const signal of ["SIGTERM", "SIGINT"] as const) { + process.on(signal, () => { + supervisorPorts.log(`frontdoor ${String(ownPid)}: received ${signal}; exiting`); + deps.exit(0); + }); + } + // A door started by its watchdog's stall handling names the stalled door's pid and waits for it to be gone before binding (see `awaitPidExit`). + const awaited = Number(process.env[DOOR_AWAIT_PID_ENV]); + // Not inherited by anything this door starts. + Reflect.deleteProperty(process.env, DOOR_AWAIT_PID_ENV); + if (Number.isInteger(awaited) && awaited > 0) { + await awaitPidExit(awaited, DOOR_STALL_BOUND_MS, supervisorPorts); + } // Declared before anything is served, so a launch the door plans for itself never asks `ps` or probes its own held event loop whether it is alive. declareThisProcessTheFrontDoor(); - deps.exit(await runFrontDoorSupervisor(idleShutdownMinutes, realFrontDoorSupervisorPorts(paths))); + deps.exit(await runFrontDoorSupervisor(idleShutdownMinutes, supervisorPorts)); }); } diff --git a/src/frontdoor/supervisor.test.ts b/src/frontdoor/supervisor.test.ts index 9cc9b610..0482888a 100644 --- a/src/frontdoor/supervisor.test.ts +++ b/src/frontdoor/supervisor.test.ts @@ -8,7 +8,7 @@ import { createDoorEventHub } from "./eventHub"; import { createDoorHealthPublisher } from "./doorHealthEvents"; import { createLaunchEventPublisher } from "./launchEvents"; import { FRONT_DOOR_PROTOCOL, listFrontDoorDirectRecords, listFrontDoorSessions, readFrontDoorState, removeFrontDoorSession, writeFrontDoorSession, writeFrontDoorState } from "./state"; -import { FRONTDOOR_POLL_MS, FRONTDOOR_SUPERVISOR_STILL_RUNNING, runFrontDoorSupervisor, watchedInstallEntry, type FrontDoorSupervisorPorts, type OwnExecutableStat } from "./supervisor"; +import { awaitPidExit, FRONTDOOR_POLL_MS, FRONTDOOR_SUPERVISOR_STILL_RUNNING, runFrontDoorSupervisor, watchedInstallEntry, type FrontDoorSupervisorPorts, type OwnExecutableStat } from "./supervisor"; const paths = buildLayoutPaths("/home/testuser/.agent-shim"); @@ -200,6 +200,37 @@ function makeWorld(options: { readonly ownPid?: number; readonly fs?: HeadroomFs /** The existence check that always answers yes: the sibling rule applies wherever the candidate path would be. */ const SIBLING_EXISTS = (): boolean => true; +describe("awaitPidExit", () => { + const PREDECESSOR_PID = 4242; + const WAIT_BOUND_MS = 2 * FRONTDOOR_POLL_MS; + + /** A clock the fake sleep advances, so the wait's bound is exercised without real time. */ + function fakeClock(runningUntil: number) { + let now = 0; + return { + now: () => now, + sleep: async (ms: number) => { + now += ms; + await Promise.resolve(); + }, + isRunning: () => now < runningUntil, + }; + } + + it("returns as soon as the pid is gone, so the replacement binds the freed sticky ports", async () => { + const clock = fakeClock(FRONTDOOR_POLL_MS); + await awaitPidExit(PREDECESSOR_PID, WAIT_BOUND_MS, clock); + expect(clock.now()).toBeGreaterThanOrEqual(FRONTDOOR_POLL_MS); + expect(clock.now()).toBeLessThan(WAIT_BOUND_MS); + }); + + it("gives up at the bound for a pid that never exits, so a wedged predecessor cannot keep the door down", async () => { + const clock = fakeClock(Number.POSITIVE_INFINITY); + await awaitPidExit(PREDECESSOR_PID, WAIT_BOUND_MS, clock); + expect(clock.now()).toBeGreaterThanOrEqual(WAIT_BOUND_MS); + }); +}); + describe("watchedInstallEntry", () => { it("watches the resolved entry itself when it is already named agent-shim", () => { expect(watchedInstallEntry("/home/testuser/.local/bin/agent-shim", SIBLING_EXISTS)).toBe("/home/testuser/.local/bin/agent-shim"); diff --git a/src/frontdoor/supervisor.ts b/src/frontdoor/supervisor.ts index 8bd4352d..8d73f1cf 100644 --- a/src/frontdoor/supervisor.ts +++ b/src/frontdoor/supervisor.ts @@ -85,6 +85,19 @@ export interface FrontDoorSupervisorPorts { readonly log: (line: string) => void; } +/** How often `awaitPidExit` looks: short, because the stalled door is ended immediately after the replacement is started. */ +const AWAIT_PID_POLL_MS = 50; + +/** + * Waits until `pid` is no longer running, or `boundMs` has passed. A door its watchdog replaced is started before the stalled one is ended, so it must not bind until the stalled one has released the sticky ports: a bind that found them held would take other ports and strand every session still pointing at the old ones. Gives up at the bound rather than blocking forever on a pid that never exits, and then lets the bind fall back as it always does. + */ +export async function awaitPidExit(pid: number, boundMs: number, ports: Pick): Promise { + const deadline = ports.now() + boundMs; + while (ports.isRunning(pid) && ports.now() < deadline) { + await ports.sleep(AWAIT_PID_POLL_MS); + } +} + /** Options for `runFrontDoorSupervisor`. */ export interface RunFrontDoorSupervisorOptions { /** Bounds the loop's iterations so a test can observe steady state; such a run returns `FRONTDOOR_SUPERVISOR_STILL_RUNNING`. */ diff --git a/src/frontdoor/watchdog.test.ts b/src/frontdoor/watchdog.test.ts index 77f7fb8a..e91c263b 100644 --- a/src/frontdoor/watchdog.test.ts +++ b/src/frontdoor/watchdog.test.ts @@ -2,11 +2,11 @@ import fs from "node:fs"; import os from "node:os"; import path from "node:path"; -import { afterEach, beforeEach, describe, expect, it } from "vitest"; +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { FRONTDOOR_PROBE_ATTEMPTS } from "./ensure"; import { PROBE_TIMEOUT_MS } from "./probe"; -import { DOOR_STALL_BOUND_MS, DOOR_WATCHDOG_CHECK_MS, startThreadWatchdog, type DoorWatchdog } from "./watchdog"; +import { DOOR_AWAIT_PID_ENV, DOOR_STALL_BOUND_MS, DOOR_WATCHDOG_CHECK_MS, startThreadWatchdog, type DoorWatchdog, type ThreadWatchdogParams } from "./watchdog"; /** The check cadence the thread tests run at: far shorter than the production one so a stall is seen in a few checks. */ const TEST_CHECK_MS = 20; @@ -56,6 +56,7 @@ describe("startThreadWatchdog", () => { boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", + respawnAfterServingMs: 0, onReport: (message) => { if (timer !== undefined) clearInterval(timer); resolve(message); @@ -86,7 +87,7 @@ describe("startThreadWatchdog", () => { it("acts while the main thread is blocked, which is the case it exists for", async () => { const reported = new Promise((resolve) => { - watchdog = startThreadWatchdog({ stateFile, ownPid: process.pid, boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", onReport: (message) => { + watchdog = startThreadWatchdog({ stateFile, ownPid: process.pid, boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", respawnAfterServingMs: 0, onReport: (message) => { resolve(message); } }); }); @@ -99,7 +100,7 @@ describe("startThreadWatchdog", () => { it("does nothing once stopped, so a slow orderly retirement is never taken for a stall", async () => { const reported: string[] = []; - watchdog = startThreadWatchdog({ stateFile, ownPid: process.pid, boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", onReport: (message) => { + watchdog = startThreadWatchdog({ stateFile, ownPid: process.pid, boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", respawnAfterServingMs: 0, onReport: (message) => { reported.push(message); } }); watchdog.stop(); @@ -109,4 +110,42 @@ describe("startThreadWatchdog", () => { expect(reported).toEqual([]); expect(JSON.parse(fs.readFileSync(stateFile, "utf8"))).not.toHaveProperty("lastError"); }); + + describe("on a stall", () => { + const MARKER = "respawned"; + + /** Waits for the thread's report, then for the log to hold `needle`, which the thread writes before it reports. */ + async function stallWith(extra: { readonly respawnAfterServingMs: number; readonly respawn?: ThreadWatchdogParams["respawn"] }): Promise { + const logFile = path.join(directory, "frontdoor.log"); + await new Promise((resolve) => { + watchdog = startThreadWatchdog({ stateFile, ownPid: process.pid, boundMs: TEST_BOUND_MS, checkEveryMs: TEST_CHECK_MS, action: "report", logFile, ...extra, onReport: () => { + resolve(); + } }); + }); + return fs.readFileSync(logFile, "utf8"); + } + + it("writes the stall to the door's log, the one record that outlives the process", async () => { + const log = await stallWith({ respawnAfterServingMs: 0 }); + expect(log).toMatch(new RegExp(`frontdoor ${String(process.pid)}: the front door's event loop did not run`)); + }); + + it("starts a replacement naming the stalled pid when the door had served long enough", async () => { + const markerFile = path.join(directory, MARKER); + const respawn = { command: process.execPath, args: ["-e", `require("node:fs").writeFileSync(${JSON.stringify(markerFile)}, process.env.${DOOR_AWAIT_PID_ENV})`], env: { PATH: process.env.PATH ?? "" } }; + const log = await stallWith({ respawnAfterServingMs: 0, respawn }); + expect(log).toMatch(/started replacement door \d+/); + await vi.waitFor(() => { + expect(fs.readFileSync(markerFile, "utf8")).toBe(String(process.pid)); + }); + }); + + it("leaves the next launch to start one when the door stalled before it had served long enough", async () => { + const markerFile = path.join(directory, MARKER); + const respawn = { command: process.execPath, args: ["-e", `require("node:fs").writeFileSync(${JSON.stringify(markerFile)}, "x")`], env: { PATH: process.env.PATH ?? "" } }; + const log = await stallWith({ respawnAfterServingMs: Number.MAX_SAFE_INTEGER, respawn }); + expect(log).toMatch(/less than the \d+ms it must have served to be replaced/); + expect(fs.existsSync(markerFile)).toBe(false); + }); + }); }); diff --git a/src/frontdoor/watchdog.ts b/src/frontdoor/watchdog.ts index 26f0c757..2878749e 100644 --- a/src/frontdoor/watchdog.ts +++ b/src/frontdoor/watchdog.ts @@ -1,5 +1,6 @@ import { Worker } from "node:worker_threads"; +import { FRONTDOOR_DEFAULT_IDLE_SHUTDOWN_MINUTES } from "../config/schema"; import { FRONTDOOR_PROBE_ATTEMPTS } from "./ensure"; import { PROBE_TIMEOUT_MS } from "./probe"; @@ -8,6 +9,16 @@ import { PROBE_TIMEOUT_MS } from "./probe"; */ export const DOOR_STALL_BOUND_MS = 2 * FRONTDOOR_PROBE_ATTEMPTS * PROBE_TIMEOUT_MS; +const MS_PER_MINUTE = 60_000; + +/** + * How long a door must have served before its watchdog replaces it on a stall: the idle shutdown window a door already stays up through. A door that stalled sooner than that after binding has not shown it can serve, so a stall that recurs at once is treated as deterministic and left to the next launch's ensure rather than respawned in a loop. + */ +export const DOOR_RESPAWN_MIN_SERVING_MS = FRONTDOOR_DEFAULT_IDLE_SHUTDOWN_MINUTES * MS_PER_MINUTE; + +/** The environment variable naming the pid a respawned door waits to see gone before it binds, so the sticky ports are free when it asks for them. */ +export const DOOR_AWAIT_PID_ENV = "AGENT_SHIM_FRONTDOOR_AWAIT_PID"; + /** How many times the thread looks at the beat counter within one probe deadline, so a stall is noticed within a small fraction of the bound. */ const CHECKS_PER_PROBE_DEADLINE = 2; @@ -26,12 +37,13 @@ export interface DoorWatchdog { type StallAction = "kill-process" | "report"; /** - * The watchdog thread: its own event loop, which a blocked main loop cannot stall. It compares a shared beat counter against its previous reading on a fixed cadence, and when the counter has not moved for `boundMs` it records the stall in the state file (the existing `lastError` field, so no reader of the file sees a new key) and ends the process. A thread that was itself descheduled for longer than its cadence (the machine was suspended or starved) restarts its clock instead of judging, since the main loop stalled by the same pause is not a hung one. + * The watchdog thread: its own event loop, which a blocked main loop cannot stall. It compares a shared beat counter against its previous reading on a fixed cadence, and when the counter has not moved for `boundMs` it records the stall in the state file (the existing `lastError` field, so no reader of the file sees a new key) and in the door's log (the next generation's first state write replaces `lastError`, and a SIGKILL runs no handler, so the log line is the only record that outlives the replacement), starts a replacement door when the door had served long enough to have proved it can, and ends the process. A thread that was itself descheduled for longer than its cadence (the machine was suspended or starved) restarts its clock instead of judging, since the main loop stalled by the same pause is not a hung one. */ const WATCHDOG_SOURCE = ` const { workerData, parentPort } = require("node:worker_threads"); const fs = require("node:fs"); -const { shared, boundMs, checkEveryMs, stateFile, ownPid, action } = workerData; +const { shared, boundMs, checkEveryMs, stateFile, ownPid, action, logFile, respawn, respawnAfterServingMs, awaitPidEnv } = workerData; +const startedAt = Date.now(); const counter = new Int32Array(shared); let lastValue = Atomics.load(counter, 0); let lastChangeAt = Date.now(); @@ -48,6 +60,31 @@ const record = (message) => { // The state file is a courtesy to whoever looks next; the process still has to end. } }; +const log = (line) => { + if (logFile === undefined) return; + try { + fs.appendFileSync(logFile, new Date().toISOString() + " frontdoor " + String(ownPid) + ": " + line + "\\n"); + } catch (error) { + // Nowhere left to report a failed log write to; the process still has to end. + } +}; +const spawnReplacement = () => { + if (respawn === undefined || logFile === undefined) return; + const servedMs = Date.now() - startedAt; + if (servedMs < respawnAfterServingMs) { + log("the door served for " + String(servedMs) + "ms before stalling, less than the " + String(respawnAfterServingMs) + "ms it must have served to be replaced; leaving the next launch to start one"); + return; + } + try { + const out = fs.openSync(logFile, "a"); + const child = require("node:child_process").spawn(respawn.command, respawn.args, { detached: true, stdio: ["ignore", out, out], env: Object.assign({}, respawn.env, { [awaitPidEnv]: String(ownPid) }) }); + child.unref(); + fs.closeSync(out); + log("started replacement door " + String(child.pid) + ", which binds once this process is gone"); + } catch (error) { + log("could not start a replacement door: " + (error instanceof Error ? error.message : String(error))); + } +}; const timer = setInterval(() => { if (stopped) return; const now = Date.now(); @@ -62,6 +99,8 @@ const timer = setInterval(() => { if (now - lastChangeAt >= boundMs) { const message = "the front door's event loop did not run for " + String(now - lastChangeAt) + "ms; its watchdog ended the process so the next launch starts a fresh door on the same ports"; record(message); + log(message); + spawnReplacement(); stopped = true; clearInterval(timer); if (action === "kill-process") { @@ -87,6 +126,12 @@ export interface ThreadWatchdogParams { readonly boundMs: number; readonly checkEveryMs: number; readonly action: StallAction; + /** The door's log file, which the thread appends the stall to before ending the process. Omitted by a host with no log of its own. */ + readonly logFile?: string; + /** How long the door must have served before a stall is answered with a replacement (`DOOR_RESPAWN_MIN_SERVING_MS` in production). */ + readonly respawnAfterServingMs: number; + /** See `respawnAfterServingMs`: how to start a replacement door on a stall: the invocation and environment of a door, to which the thread adds `DOOR_AWAIT_PID_ENV`. Omitted by a host that does not respawn. */ + readonly respawn?: { readonly command: string; readonly args: readonly string[]; readonly env: Readonly> }; /** Called on the main thread with the recorded message when `action` is `report` and the stall was seen. */ readonly onReport?: (message: string) => void; } @@ -97,7 +142,7 @@ export function startThreadWatchdog(params: Readonly): Doo const counter = new Int32Array(shared); const worker = new Worker(WATCHDOG_SOURCE, { eval: true, - workerData: { shared, boundMs: params.boundMs, checkEveryMs: params.checkEveryMs, stateFile: params.stateFile, ownPid: params.ownPid, action: params.action }, + workerData: { shared, boundMs: params.boundMs, checkEveryMs: params.checkEveryMs, stateFile: params.stateFile, ownPid: params.ownPid, action: params.action, logFile: params.logFile, respawn: params.respawn, respawnAfterServingMs: params.respawnAfterServingMs, awaitPidEnv: DOOR_AWAIT_PID_ENV }, }); worker.unref(); worker.on("message", (message: unknown) => {