From 23a92db3003edd92381aae64d79c0c262579e85c Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 15:38:33 -0800 Subject: [PATCH 1/2] fix(hub): keep the local sidecar running through a hub restart A hub restart (SIGTERM, a fatal error, a failed start or a crash) now leaves the local sidecar running, and the next hub adopts it when the recorded pid is still the same process (start time and command), so it reconnects over Interchange's stock path and a turn in flight carries on. SIGINT, as when the dev script exits, still stops it. The e2e harness can restart the hub in place and hold a recorded model reply. --- apps/hub/src/app.ts | 5 +- .../hub/src/provisioning/workflow-services.ts | 6 +- apps/hub/src/server.ts | 24 +- apps/hub/src/sidecar/local-provisioner.ts | 106 +++++-- apps/hub/src/sidecar/manifest-store.ts | 29 +- apps/hub/src/sidecar/spawn.ts | 53 +++- docs/IMPLEMENTATION.md | 6 +- test/AGENTS.md | 5 +- test/e2e/hub-restart.test.ts | 271 ++++++++++++++++++ test/e2e/support/hub.ts | 42 ++- test/e2e/support/recorded-model.ts | 30 ++ 11 files changed, 513 insertions(+), 64 deletions(-) create mode 100644 test/e2e/hub-restart.test.ts diff --git a/apps/hub/src/app.ts b/apps/hub/src/app.ts index 7ec775cc..f60a0c60 100644 --- a/apps/hub/src/app.ts +++ b/apps/hub/src/app.ts @@ -667,7 +667,8 @@ export async function createHubRuntime( workflows.start(); for (const scheduler of projectSchedulers) scheduler.start(); }, - async close() { + /** `keepSidecar` leaves the local sidecar running for the next hub to adopt; a full stop takes it down. */ + async close(options?: { readonly keepSidecar?: boolean }) { modelLogins.stop(); for (const scheduler of projectSchedulers) scheduler.stop(); workflows.stop(); @@ -675,7 +676,7 @@ export async function createHubRuntime( liveApps.proxy.close(); agentActivity.dispose(); sidecarWebSocket.close(); - await workflows.drain(); + await workflows.drain(options); sessions.dispose(); authEmail?.close(); await handle.close(); diff --git a/apps/hub/src/provisioning/workflow-services.ts b/apps/hub/src/provisioning/workflow-services.ts index 55423c85..fc03aa37 100644 --- a/apps/hub/src/provisioning/workflow-services.ts +++ b/apps/hub/src/provisioning/workflow-services.ts @@ -49,6 +49,7 @@ import { createLocalSidecarManifestStore } from "../sidecar/manifest-store.ts"; import { createPlacementChooser } from "../sidecar/placement.ts"; import { restoreLocalSidecars } from "../sidecar/restore.ts"; import { + adoptLocalSidecar, childProcessEnv, createSpawnLocalSidecar, stopStaleLocalSidecar, @@ -131,6 +132,7 @@ export async function createWorkflowServices(options: { maxActiveDeployments, processEnv: childProcessEnv(environment), }), + adoptSidecar: adoptLocalSidecar, stopStaleSidecar: stopStaleLocalSidecar, manifests: createLocalSidecarManifestStore({ dataRoot, @@ -390,9 +392,9 @@ export async function createWorkflowServices(options: { for (const scheduler of schedulers) scheduler.stop(); for (const off of unsubscribers) off(); }, - async drain() { + async drain(options?: { readonly keepSidecar?: boolean }) { await Promise.allSettled(pending); - await local.shutdown(); + await local.shutdown(options); }, }; } diff --git a/apps/hub/src/server.ts b/apps/hub/src/server.ts index 42c8123e..9cdff64a 100644 --- a/apps/hub/src/server.ts +++ b/apps/hub/src/server.ts @@ -12,10 +12,11 @@ const errorMetadata = (error: unknown) => ({ if (import.meta.main) { await setupHubLogging(); const log = getLogger(["corbits-builder", "hub"]); - let close = async () => {}; + type Stop = { readonly keepSidecar: boolean }; + let close: (how: Stop) => Promise = async () => {}; let stopping: Promise | undefined; - const shutDown = async () => { - for (const step of [close, dispose]) { + const shutDown = async (how: Stop) => { + for (const step of [() => close(how), dispose]) { try { await step(); } catch (error) { @@ -24,11 +25,14 @@ if (import.meta.main) { } } }; - const stop = () => (stopping ??= shutDown()); + // A restart (SIGTERM from a supervisor, a fatal error or a failed start) leaves the local sidecar running for the next hub to + // adopt, so turns in flight carry on; SIGINT, as when the dev script exits, is a full stop that takes it down. + const stop = (how: Stop) => (stopping ??= shutDown(how)); + const restart = { keepSidecar: true }; const exitOnFatal = (message: string) => async (error: unknown) => { try { log.error(message, errorMetadata(error)); - await stop(); + await stop(restart); } finally { // Handles a failed shutdown left open would keep a broken hub alive instead of letting its supervisor restart it. process.exit(1); @@ -40,10 +44,10 @@ if (import.meta.main) { ); process.on("uncaughtException", exitOnFatal("Hub hit an uncaught exception")); process.once("SIGTERM", () => { - void stop(); + void stop(restart); }); process.once("SIGINT", () => { - void stop(); + void stop({ keepSidecar: false }); }); try { const runtime = await createHubRuntime(); @@ -55,9 +59,9 @@ if (import.meta.main) { fetch: runtime.app.fetch, websocket: runtime.websocket, }); - close = async () => { + close = async (how) => { await server.stop(true); - await runtime.close(); + await runtime.close(how); }; await runtime.start(); log.info( @@ -69,6 +73,6 @@ if (import.meta.main) { errorMetadata(error), ); process.exitCode = 1; - await stop(); + await stop(restart); } } diff --git a/apps/hub/src/sidecar/local-provisioner.ts b/apps/hub/src/sidecar/local-provisioner.ts index 981cd82f..4a02dbf5 100644 --- a/apps/hub/src/sidecar/local-provisioner.ts +++ b/apps/hub/src/sidecar/local-provisioner.ts @@ -24,6 +24,7 @@ import type { LocalSidecarHold, LocalSidecarManifest, LocalSidecarManifestStore, + LocalSidecarPid, } from "./manifest-store.ts"; const DEFAULT_STOP_TIMEOUT_MS = 25_000; @@ -39,8 +40,10 @@ export type LocalSidecarExit = { readonly signal: string | null; }; -interface LocalSidecarProcess { +export interface LocalSidecarProcess { readonly pid: number; + /** What `ps` reported the process started as, so a later hub can tell it from a process that reuses its pid. */ + readonly identity?: string; readonly exited: Promise; kill(signal: NodeJS.Signals): void; } @@ -53,6 +56,10 @@ export type SpawnLocalSidecar = (args: { type CreateLocalProcessSidecarProvisionerOpts = { readonly dataRoot: string; readonly spawnSidecar: SpawnLocalSidecar; + /** The sidecar a previous hub left running, when it is still the process recorded. */ + readonly adoptSidecar: ( + record: LocalSidecarPid, + ) => LocalSidecarProcess | undefined; /** Stops a sidecar still running at a recorded pid; resolves whether one was running there. */ readonly stopStaleSidecar: ( pid: number, @@ -82,7 +89,11 @@ type LocalSidecarRecords = { export interface LocalProcessSidecarProvisioner { readonly provisioner: SidecarProvisioner; restore(records: LocalSidecarRecords): Promise; - shutdown(): Promise; + /** + * Stops taking work. The sidecar is stopped too unless `keepSidecar`, which leaves it running for the next hub to + * adopt: it reconnects to that hub and its turns carry on. + */ + shutdown(options?: { readonly keepSidecar?: boolean }): Promise; } type ManagedProcess = { @@ -114,6 +125,7 @@ async function exitedWithin( export function createLocalProcessSidecarProvisioner({ dataRoot, spawnSidecar, + adoptSidecar, stopStaleSidecar, manifests, capacity, @@ -194,19 +206,29 @@ export function createLocalProcessSidecarProvisioner({ } } - // The recorded pid lets the next hub stop this process if this one dies without shutting it down. + function manage( + handle: LocalSidecarProcess, + sidecarId: string, + ): ManagedProcess { + const managed: ManagedProcess = { handle, exited: false }; + void trackExit(managed, sidecarId); + running = managed; + return managed; + } + + // The recorded pid and identity let the next hub adopt this process, or stop it when it is no longer this one. async function start(current: LocalSidecarManifest): Promise { const handle = spawnSidecar({ manifest: current, adapterManifest }); try { - await manifests.writePid(handle.pid); + await manifests.writePid({ + pid: handle.pid, + ...(handle.identity !== undefined && { identity: handle.identity }), + }); } catch (error) { handle.kill("SIGKILL"); throw error; } - const managed: ManagedProcess = { handle, exited: false }; - void trackExit(managed, current.sidecarId); - running = managed; - return managed; + return manage(handle, current.sidecarId); } /** The running sidecar, started under `request`'s identity when none was minted yet. */ @@ -446,25 +468,46 @@ export function createLocalProcessSidecarProvisioner({ await fs.rm(dataDir, { recursive: true, force: true }); } - // A copy whose hold a destroy released is removed only once the sidecar reconnects and the hub undeploys it, so the - // sidecar comes back for that cleanup too; forgetting it would leave the copy holding a slot for good. - async function stopStale(): Promise { - const pid = await manifests.readPid(); - if (pid === null) return; - if (await stopStaleSidecar(pid, stopTimeoutMs)) - log({ level: "warn", msg: "local_sidecar_stale_stopped", pid }); + // The sidecar a previous hub left running, still the process it recorded. A process at that pid that is not the + // recorded one is not ours and is left alone; only a bare pid from a hub before identities is stopped there, so + // two processes never share one identity and data dir. + async function reclaim(): Promise { + const record = await manifests.readPid(); + if (record === null) return undefined; + const adopted = adoptSidecar(record); + if (adopted !== undefined) return adopted; + if ( + record.identity === undefined && + (await stopStaleSidecar(record.pid, stopTimeoutMs)) + ) + log({ + level: "warn", + msg: "local_sidecar_stale_stopped", + pid: record.pid, + }); await manifests.removePid(); + return undefined; } + // A copy whose hold a destroy released is removed only once the sidecar reconnects and the hub undeploys it, so the + // sidecar comes back for that cleanup too; forgetting it would leave the copy holding a slot for good. async function restoreSidecar(records: LocalSidecarRecords): Promise { const { isLive, hasWork, reusable } = records; stillOpen = records.stillOpen; - await stopStale(); + const left = await reclaim(); + // A sidecar left running for an identity this hub does not keep has nothing to come back for. + const retire = async (): Promise => { + if (left === undefined) return; + await stopProcess({ handle: left, exited: false }); + log({ level: "info", msg: "local_sidecar_retired", pid: left.pid }); + }; await restoreFences(); const stored = await manifests.read(); const recorded = await manifests.readHolds(); if (stored === null) { + await retire(); if (recorded.length > 0) await forget(); + else if (left !== undefined) await manifests.removePid(); return; } const live: LocalSidecarHold[] = []; @@ -480,6 +523,7 @@ export function createLocalProcessSidecarProvisioner({ if (!(await reusable(stored.sidecarId))) { if (host === undefined) throw new Error("Local sidecar provisioner has not been attached"); + await retire(); const report = await host.sidecarStopped(stored.sidecarId, { endReconnectWindow: true, }); @@ -494,6 +538,7 @@ export function createLocalProcessSidecarProvisioner({ return; } if (live.length === 0 && !(await hasWork(stored.sidecarId))) { + await retire(); await forget(); log({ level: "info", @@ -506,6 +551,18 @@ export function createLocalProcessSidecarProvisioner({ manifest = stored; for (const hold of live) holds.set(hold.allocationId, hold); await manifests.writeHolds(live); + if (left !== undefined) { + manage(left, stored.sidecarId); + log({ + level: "info", + msg: "local_sidecar_adopted", + sidecarId: stored.sidecarId, + pid: left.pid, + holds: live.length, + dropped, + }); + return; + } const child = await start(stored); log({ level: "info", @@ -533,9 +590,18 @@ export function createLocalProcessSidecarProvisioner({ } } - // The manifest, holds and data dir survive shutdown so the next hub boot can respawn the sidecar. - async function stopAll(): Promise { + // The manifest, holds and data dir survive shutdown so the next hub boot can adopt or respawn the sidecar. + async function stopAll(keepSidecar: boolean): Promise { await queue.idle(); + if (keepSidecar) { + if (running !== null && !running.exited) + log({ + level: "info", + msg: "local_sidecar_left_running", + pid: running.handle.pid, + }); + return; + } if (running !== null) await stopProcess(running); running = null; await manifests.removePid(); @@ -572,8 +638,8 @@ export function createLocalProcessSidecarProvisioner({ return { provisioner, restore: restoreOnce, - shutdown() { - shutdownPromise ??= stopAll(); + shutdown({ keepSidecar = false } = {}) { + shutdownPromise ??= stopAll(keepSidecar); return shutdownPromise; }, }; diff --git a/apps/hub/src/sidecar/manifest-store.ts b/apps/hub/src/sidecar/manifest-store.ts index 1e82de18..c60edcab 100644 --- a/apps/hub/src/sidecar/manifest-store.ts +++ b/apps/hub/src/sidecar/manifest-store.ts @@ -34,6 +34,14 @@ const LocalSidecarFence = type({ export type LocalSidecarFence = typeof LocalSidecarFence.infer; const LocalSidecarFences = LocalSidecarFence.array(); +/** The sidecar process last started on a data root, and what `ps` reported it started as. */ +const LocalSidecarPid = type({ + pid: "number.integer > 0", + "identity?": "string > 0", +}); +export type LocalSidecarPid = typeof LocalSidecarPid.infer; +const StoredLocalSidecarPid = type("string.json.parse").to(LocalSidecarPid); + export interface LocalSidecarManifestStore { read(): Promise; write(manifest: LocalSidecarManifest): Promise; @@ -43,9 +51,9 @@ export interface LocalSidecarManifestStore { /** The stored fences, or null when the file could not be read and was moved aside. */ readFences(): Promise; writeFences(fences: readonly LocalSidecarFence[]): Promise; - /** The pid of the sidecar process last started on this data root, or null when none is recorded. */ - readPid(): Promise; - writePid(pid: number): Promise; + /** The sidecar process last started on this data root, or null when none is recorded. */ + readPid(): Promise; + writePid(record: LocalSidecarPid): Promise; removePid(): Promise; } @@ -179,13 +187,18 @@ export function createLocalSidecarManifestStore({ await writeAtomically(fencesFile, JSON.stringify(fences)); } - async function readPid(): Promise { - const pid = Number((await readOptional(pidFile))?.trim()); - return Number.isSafeInteger(pid) && pid > 0 ? pid : null; + // A bare pid is what hubs before identities wrote; it is still stopped, never adopted. + async function readPid(): Promise { + const text = (await readOptional(pidFile))?.trim(); + if (text === undefined) return null; + const stored = StoredLocalSidecarPid(text); + if (!(stored instanceof type.errors)) return stored; + const pid = Number(text); + return Number.isSafeInteger(pid) && pid > 0 ? { pid } : null; } - async function writePid(pid: number): Promise { - await writeAtomically(pidFile, String(pid)); + async function writePid(record: LocalSidecarPid): Promise { + await writeAtomically(pidFile, JSON.stringify(record)); } async function removePid(): Promise { diff --git a/apps/hub/src/sidecar/spawn.ts b/apps/hub/src/sidecar/spawn.ts index cbb6f0b9..9b7e98a0 100644 --- a/apps/hub/src/sidecar/spawn.ts +++ b/apps/hub/src/sidecar/spawn.ts @@ -1,8 +1,10 @@ import { hasCode } from "@intx/types"; import type { LocalSidecarExit, + LocalSidecarProcess, SpawnLocalSidecar, } from "./local-provisioner.ts"; +import type { LocalSidecarPid } from "./manifest-store.ts"; /** The variables a sidecar needs to find binaries and scratch space; nothing else from the hub env crosses over. */ type ChildProcessEnv = { PATH?: string; HOME?: string; TMPDIR?: string }; @@ -28,11 +30,20 @@ type CreateSpawnLocalSidecarOpts = { const SIDECAR_ENTRY = "apps/sidecar/src/index.ts"; const EXIT_POLL_MS = 100; -function commandOf(pid: number): string | undefined { - const ps = Bun.spawnSync(["ps", "-o", "command=", "-p", String(pid)], { - stderr: "ignore", - }); - return ps.exitCode === 0 ? ps.stdout.toString() : undefined; +// When the process at `pid` started, and its command. Together they tell one process from a later one that reuses +// its pid. +function processAt(pid: number): string | undefined { + const ps = Bun.spawnSync( + ["ps", "-o", "lstart=", "-o", "command=", "-p", String(pid)], + { env: { ...process.env, LC_ALL: "C" }, stderr: "ignore" }, + ); + return ps.exitCode === 0 ? ps.stdout.toString().trim() : undefined; +} + +/** The identity of the sidecar running at `pid`, or undefined when no sidecar runs there. */ +function localSidecarIdentity(pid: number): string | undefined { + const identity = processAt(pid); + return identity?.includes(SIDECAR_ENTRY) ? identity : undefined; } // Signals a process that is still running; false when it no longer exists. @@ -47,15 +58,15 @@ function sendSignal(pid: number, name: NodeJS.Signals | 0): boolean { } /** - * Stop the sidecar a hub that died without shutting down left running at `pid`, so two processes never share one - * identity and data dir. The process is no child of this hub, so its exit is polled. A pid that no longer runs the + * Stop the sidecar a hub that recorded no identity left running at `pid`, so two processes never share one identity + * and data dir. The process is no child of this hub, so its exit is polled. A pid that no longer runs the * sidecar entry is left alone. */ export async function stopStaleLocalSidecar( pid: number, timeoutMs: number, ): Promise { - if (!commandOf(pid)?.includes(SIDECAR_ENTRY)) return false; + if (localSidecarIdentity(pid) === undefined) return false; for (const name of ["SIGTERM", "SIGKILL"] as const) { if (!sendSignal(pid, name)) return true; for (const deadline = Date.now() + timeoutMs; Date.now() < deadline;) { @@ -66,6 +77,30 @@ export async function stopStaleLocalSidecar( throw new Error(`Stale local sidecar process ${String(pid)} did not exit`); } +/** + * The sidecar a previous hub left running, when the process at `record.pid` is still the one it started. The process + * is no child of this hub, so its exit is polled. + */ +export function adoptLocalSidecar( + record: LocalSidecarPid, +): LocalSidecarProcess | undefined { + const { pid, identity } = record; + if (identity === undefined || localSidecarIdentity(pid) !== identity) + return undefined; + const exited = (async (): Promise => { + while (sendSignal(pid, 0)) await Bun.sleep(EXIT_POLL_MS); + return { code: null, signal: null }; + })(); + return { + pid, + identity, + exited, + kill(signal) { + sendSignal(pid, signal); + }, + }; +} + async function exitOf(child: Bun.Subprocess): Promise { const code = await child.exited; const signal = child.signalCode; @@ -98,8 +133,10 @@ export function createSpawnLocalSidecar({ stdout: "inherit", stderr: "inherit", }); + const identity = localSidecarIdentity(child.pid); return { pid: child.pid, + ...(identity !== undefined && { identity }), exited: exitOf(child), kill(signal) { child.kill(signal); diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index fe5c84d7..d668c2e6 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -45,7 +45,11 @@ Agents run on sidecars, which the hub provisions - **Local process sidecars** are always available. The hub spawns them from the vendored sidecar app on the same machine, keeps their state under `HUB_DATA_DIR`, and restores them after a restart. One holds up to - `SIDECAR_LOCAL_MAX_ACTIVE` deployments. + `SIDECAR_LOCAL_MAX_ACTIVE` deployments. A hub restart (SIGTERM, a fatal + error, a failed start or a crash) leaves the sidecar running, and the next + hub adopts it when it is still the process it recorded, so turns in flight + carry on over Interchange's reconnect. SIGINT, as when the dev script exits, + stops it. - **Docker sidecars** are added when `SIDECAR_IMAGE` names an image (built from the `sidecar` target of [deploy/Dockerfile](https://github.com/corbitsdev/corbits-builder/blob/main/deploy/Dockerfile)), diff --git a/test/AGENTS.md b/test/AGENTS.md index 2e9fdbcf..a96d90a0 100644 --- a/test/AGENTS.md +++ b/test/AGENTS.md @@ -9,7 +9,8 @@ end-to-end tests: `bun run test` with no database. - **`test/e2e/`:** end-to-end tests against an isolated Postgres, the real hub and real sidecars (`custom-provider.test.ts`, `oauth-model-login.test.ts`, - `specialist-end-notices.test.ts` and `pipeline.test.ts`). + `specialist-end-notices.test.ts`, `hub-restart.test.ts` and + `pipeline.test.ts`). - **`test/e2e/support/`:** shared harness (`hub.ts`), the recorded OpenAI-compatible model server (`recorded-model.ts` and its unit tests in `recorded-model.test.ts`) and the fixture extractor (`extract.ts`). @@ -45,6 +46,6 @@ With the `builder-e2e` database running ([docs/DEV.md](../docs/DEV.md#tests)): ```sh bun run test:e2e bun --conditions=intx-src test test/e2e/support/recorded-model.test.ts -bun --env-file=.env.hub --conditions=intx-src test test/e2e/custom-provider.test.ts test/e2e/oauth-model-login.test.ts test/e2e/specialist-end-notices.test.ts +bun --env-file=.env.hub --conditions=intx-src test test/e2e/custom-provider.test.ts test/e2e/oauth-model-login.test.ts test/e2e/specialist-end-notices.test.ts test/e2e/hub-restart.test.ts bun --env-file=.env.hub --conditions=intx-src test test/e2e/pipeline.test.ts ``` diff --git a/test/e2e/hub-restart.test.ts b/test/e2e/hub-restart.test.ts new file mode 100644 index 00000000..4d652a76 --- /dev/null +++ b/test/e2e/hub-restart.test.ts @@ -0,0 +1,271 @@ +import { expect, test } from "bun:test"; +import { mkdtemp, rm } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { startTestHub, type TestHubHandle } from "./support/hub.ts"; +import { + formatOpenAISseResponse, + startRecordedModelServer, + writeScenarioFixture, + type RecordedModelServer, +} from "./support/recorded-model.ts"; + +const DEADLINE_MS = 180_000; +const COORDINATOR_TOOLS = [ + "read_artifact", + "write_artifact", + "request_approval", + "deploy_writer", +]; +const WRITER_TOOLS = ["write_artifact"]; + +async function poll( + recordedModel: RecordedModelServer, + description: string, + check: () => Promise, +): Promise { + const deadline = Date.now() + DEADLINE_MS; + while (Date.now() < deadline) { + const failure = recordedModel.firstFailure(); + if (failure) throw failure; + const found = await check(); + if (found !== undefined) return found; + await Bun.sleep(250); + } + throw new Error(`Timed out waiting for ${description}`); +} + +// A writer whose model reply is still in flight when the hub restarts. The sidecar outlives the hub, the new hub +// adopts it, and the writer's turn files its document as if nothing happened: no failed run, and no notice that would +// send the coordinator an exchange the recording does not have. +test( + "a specialist mid-turn finishes its turn across a hub restart", + async () => { + const fixtureDir = await mkdtemp(join(tmpdir(), "builder-e2e-restart-")); + let recordedModel: RecordedModelServer | undefined; + let hub: TestHubHandle | undefined; + try { + await writeScenarioFixture( + fixtureDir, + { scenario: "hub-restart-mid-turn", recordedFromCommit: "3038faa" }, + [ + { + runKey: "coordinator-1", + role: "coordinator", + stage: "documents", + sourceRunId: "run_coord_restart_1", + exchanges: [ + { + sse: formatOpenAISseResponse({ + toolCalls: [ + { + id: "coord_write_brief", + name: "write_artifact", + arguments: { + kind: "problem_brief", + title: "Brief", + content: "Brief", + }, + }, + ], + }), + expect: { + lastMessageRole: "user", + userTextContains: "Restart Pitch", + requiredTools: COORDINATOR_TOOLS, + toolResults: [], + }, + }, + { + sse: formatOpenAISseResponse({ + toolCalls: [ + { + id: "coord_approve_brief", + name: "request_approval", + arguments: { kinds: ["problem_brief"] }, + }, + ], + }), + expect: { + lastMessageRole: "tool", + requiredTools: COORDINATOR_TOOLS, + toolResults: [ + { + callId: "coord_write_brief", + toolName: "write_artifact", + isError: false, + }, + ], + }, + }, + { + sse: formatOpenAISseResponse({ + toolCalls: [ + { + id: "coord_deploy_constraints", + name: "deploy_writer", + arguments: { + kind: "constraints", + instruction: + "Write the constraints for the restart pitch from its brief.", + }, + }, + ], + }), + expect: { + lastMessageRole: "tool", + requiredTools: COORDINATOR_TOOLS, + toolResults: [ + { + callId: "coord_approve_brief", + toolName: "request_approval", + isError: false, + }, + ], + }, + }, + { + sse: formatOpenAISseResponse({ text: "Writer started." }), + expect: { + lastMessageRole: "tool", + requiredTools: COORDINATOR_TOOLS, + toolResults: [ + { + callId: "coord_deploy_constraints", + toolName: "deploy_writer", + isError: false, + keys: ["runId", "role", "status"], + }, + ], + }, + }, + { + sse: formatOpenAISseResponse({ + text: "The constraints are filed as version 1.", + }), + expect: { + lastMessageRole: "user", + userTextContains: "Constraints filed.", + requiredTools: COORDINATOR_TOOLS, + toolResults: [], + }, + }, + ], + }, + { + runKey: "writer-1", + role: "writer", + stage: "documents", + sourceRunId: "run_writer_restart_1", + exchanges: [ + { + sse: formatOpenAISseResponse({ + toolCalls: [ + { + id: "writer_write", + name: "write_artifact", + arguments: { + title: "Constraints", + content: "Constraints", + }, + }, + ], + }), + expect: { + lastMessageRole: "user", + userTextContains: '"artifact":"constraints"', + requiredTools: WRITER_TOOLS, + toolResults: [], + }, + }, + { + sse: formatOpenAISseResponse({ text: "Constraints filed." }), + expect: { + lastMessageRole: "tool", + requiredTools: WRITER_TOOLS, + toolResults: [ + { + callId: "writer_write", + toolName: "write_artifact", + isError: false, + keys: ["artifactId", "kind", "version", "sha256"], + }, + ], + }, + }, + ], + }, + ], + ); + const model = await startRecordedModelServer(fixtureDir); + recordedModel = model; + const writerTurn = model.hold("writer-1", 0); + let writing = false; + void writerTurn.reached.then(() => (writing = true)); + // The ask's time comes from Postgres's clock and the brief's from the hub's, so the ask waits until the brief + // is plainly older than it. + const ask = model.hold("coordinator-1", 1); + void ask.reached.then(async () => { + await Bun.sleep(1_000); + ask.release(); + }); + const started = await startTestHub(model); + hub = started; + const { client, workspaceTenantId } = started; + const projectId = "restart-pitch"; + await client.mutateProject(workspaceTenantId, projectId, { + mutationId: `mut-create-${projectId}`, + idempotencyKey: `idem-create-${projectId}`, + command: "create_project", + payload: { + title: "Restart Pitch", + profile: "recorded_replay", + problemStatement: "Restart Pitch", + }, + }); + const { tenantId } = await client.project(workspaceTenantId, projectId); + + const approval = await poll(model, "the brief approval", async () => { + const { data } = await client.nativeApprovals(tenantId); + return data.find( + (item) => + item.status === "pending" && + item.toolDefinition["name"] === "request_approval", + ); + }); + await client.decideNativeApproval( + tenantId, + approval.id, + "approve", + "once", + ); + + await poll(model, "the writer's turn", async () => writing || undefined); + await started.restart(); + writerTurn.release(); + + await poll(model, "the coordinator's reply to the writer", async () => { + const { turns } = await client.conversationEvents(tenantId, projectId); + return turns?.find( + (turn) => + turn.role === "assistant" && + turn.text.includes("filed as version 1"), + ); + }); + const { artifacts, failed } = await client.projectArtifacts( + tenantId, + projectId, + ); + expect(failed).toEqual([]); + expect( + artifacts.find((item) => item.kind === "constraints")?.version, + ).toBe(1); + model.assertClean(); + } finally { + await hub?.close(); + await recordedModel?.stop(); + await rm(fixtureDir, { recursive: true, force: true }); + } + }, + DEADLINE_MS * 2, +); diff --git a/test/e2e/support/hub.ts b/test/e2e/support/hub.ts index 24b5aab7..93d21a07 100644 --- a/test/e2e/support/hub.ts +++ b/test/e2e/support/hub.ts @@ -20,6 +20,8 @@ export type TestHubHandle = Readonly<{ runId: string, onFrame: (event: string, data: string) => void, ) => () => void; + /** Stops the hub as a supervisor restart does, leaving its sidecar running, and boots a new one in its place. */ + restart: () => Promise; close: () => Promise; }>; @@ -50,15 +52,24 @@ async function signUpSessionCookie( password: string, name: string, ): Promise { - const response = await fetch(new URL("/api/auth/sign-up/email", baseUrl), { - method: "POST", - redirect: "error", - headers: { - "Content-Type": "application/json", - Origin: baseUrl, - }, - body: JSON.stringify({ email, password, name }), - }); + const signUp = () => + fetch(new URL("/api/auth/sign-up/email", baseUrl), { + method: "POST", + redirect: "error", + headers: { + "Content-Type": "application/json", + Origin: baseUrl, + }, + body: JSON.stringify({ email, password, name }), + }); + let response = await signUp(); + // Hubs in one test process share the auth library's sign-up limit, so a hub booted right after another waits it out. + if (response.status === 429) { + await response.body?.cancel(); + const retryAfter = Number(response.headers.get("X-Retry-After") ?? "10"); + await Bun.sleep((retryAfter + 1) * 1_000); + response = await signUp(); + } if (!response.ok) { const body = await response.text(); throw new Error(`E2E sign-up failed (${response.status}): ${body}`); @@ -161,8 +172,7 @@ export async function startTestHub( } } - try { - await migrateDatabase(hubEnv); + async function boot(): Promise { runtime = await createHubRuntime(hubEnv); server = Bun.serve({ hostname: runtime.hostname, @@ -171,6 +181,11 @@ export async function startTestHub( websocket: runtime.websocket, }); runtime.start(); + } + + try { + await migrateDatabase(hubEnv); + await boot(); const cookie = await signUpSessionCookie( baseUrl, @@ -311,6 +326,11 @@ export async function startTestHub( client, fakeCredentialSecret, subscribeRunStream, + restart: async () => { + await server?.stop(true); + await runtime?.close({ keepSidecar: true }); + await boot(); + }, close: async () => { for (const controller of activeStreams) { controller.abort(); diff --git a/test/e2e/support/recorded-model.ts b/test/e2e/support/recorded-model.ts index 63805bd0..22bac98a 100644 --- a/test/e2e/support/recorded-model.ts +++ b/test/e2e/support/recorded-model.ts @@ -142,6 +142,11 @@ export type RecordedModelServer = Readonly<{ assertClean: () => void; firstFailure: () => Error | undefined; onActivity: (listener: () => void) => () => void; + /** Holds the response to one exchange of a run until released, so a test can act while that turn is in flight. */ + hold: ( + runKey: string, + exchange: number, + ) => Readonly<{ reached: Promise; release: () => void }>; stop: () => Promise; }>; @@ -305,6 +310,13 @@ export async function startRecordedModelServer( const cursors = new Map(runs.map((r) => [r.runKey, 0])); const bindings = new Map(); const listeners = new Set<() => void>(); + const holds = new Map< + string, + { + reached: PromiseWithResolvers; + released: PromiseWithResolvers; + } + >(); let failure: Error | undefined; function cursorFor(runKey: string): number { @@ -511,6 +523,8 @@ export async function startRecordedModelServer( const server = Bun.serve({ hostname: "127.0.0.1", port: 0, + // A held response must outlive Bun's 10-second idle default. + idleTimeout: 0, fetch: async (request) => { const url = new URL(request.url); if (request.method === "GET" && url.pathname.endsWith("/models")) { @@ -556,6 +570,11 @@ export async function startRecordedModelServer( }); cursors.set(run.runKey, cursor + 1); notify(); + const held = holds.get(`${run.runKey}#${cursor}`); + if (held) { + held.reached.resolve(); + await held.released.promise; + } return new Response(sse, { status: 200, headers: exchange.responseHeaders, @@ -583,6 +602,17 @@ export async function startRecordedModelServer( listeners.delete(listener); }; }, + hold: (runKey, exchange) => { + const held = { + reached: Promise.withResolvers(), + released: Promise.withResolvers(), + }; + holds.set(`${runKey}#${exchange}`, held); + return { + reached: held.reached.promise, + release: () => held.released.resolve(), + }; + }, assertClean: () => { if (failure) throw failure; const unconsumed: string[] = []; From c1c21fdb6b790d0edf0a6dd738f7bc0115b41337 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 15:59:16 -0800 Subject: [PATCH 2/2] fix(hub): adopt only an identical sidecar and bound one left running Tag the local sidecar and a new watchdog with the data dir and a fingerprint of their spawn inputs; adopt only an exact match, stop every other process on the data dir, replace an adopted sidecar that does not reconnect within a minute, and stop an orphan after ten minutes without a hub. Keep the graceful stop when the hub is process 1 or SIDECAR_LOCAL_OUTLIVES_HUB=false, release the child so a SIGTERM'd hub exits, and restart a real hub process in the e2e test. --- .env.hub.example | 3 + .../hub/src/provisioning/workflow-services.ts | 22 +- apps/hub/src/sidecar/local-provisioner.ts | 126 +++++--- apps/hub/src/sidecar/manifest-store.ts | 29 +- apps/hub/src/sidecar/restore.ts | 3 +- apps/hub/src/sidecar/spawn.ts | 282 +++++++++++++----- apps/hub/src/sidecar/watchdog.ts | 70 +++++ deploy/Dockerfile | 2 +- docs/IMPLEMENTATION.md | 10 +- test/e2e/hub-restart.test.ts | 12 +- test/e2e/support/hub.ts | 115 ++++--- 11 files changed, 482 insertions(+), 192 deletions(-) create mode 100644 apps/hub/src/sidecar/watchdog.ts diff --git a/.env.hub.example b/.env.hub.example index b840fac5..046a73d6 100644 --- a/.env.hub.example +++ b/.env.hub.example @@ -98,6 +98,9 @@ PRODUCT_HELLO_WORDING= # Agents run in sidecars. By default the hub starts them as local processes. # Most local sidecars running at once. # SIDECAR_LOCAL_MAX_ACTIVE=16 +# Whether the local sidecar keeps running through a hub restart for the next hub to adopt. Defaults to true unless +# the hub is a container's first process, where the container stops with it. +# SIDECAR_LOCAL_OUTLIVES_HUB=true # Longest wait, in milliseconds, for a project's coordinator to start. Raise it on a slow machine; it must stay under 300000. # COORDINATOR_START_TIMEOUT_MS=60000 # Optional Docker sidecars. deploy/compose.docker-sidecars.yml sets them for the hub diff --git a/apps/hub/src/provisioning/workflow-services.ts b/apps/hub/src/provisioning/workflow-services.ts index fc03aa37..2c4801dd 100644 --- a/apps/hub/src/provisioning/workflow-services.ts +++ b/apps/hub/src/provisioning/workflow-services.ts @@ -49,9 +49,8 @@ import { createLocalSidecarManifestStore } from "../sidecar/manifest-store.ts"; import { createPlacementChooser } from "../sidecar/placement.ts"; import { restoreLocalSidecars } from "../sidecar/restore.ts"; import { - adoptLocalSidecar, childProcessEnv, - createSpawnLocalSidecar, + createLocalSidecarLauncher, stopStaleLocalSidecar, } from "../sidecar/spawn.ts"; @@ -124,15 +123,19 @@ export async function createWorkflowServices(options: { const dataRoot = localSidecarDataRoot( environment["HUB_DATA_DIR"] ?? "data/hub", ); + // A container's runtime stops every process in it once the hub, its first process, exits, so a sidecar left running + // there is never adopted; the hub stops it gracefully instead. + const sidecarOutlivesHub = + (environment["SIDECAR_LOCAL_OUTLIVES_HUB"] ?? String(process.pid !== 1)) === + "true"; const local = createLocalProcessSidecarProvisioner({ dataRoot, - spawnSidecar: createSpawnLocalSidecar({ + ...createLocalSidecarLauncher({ interchangeDir: INTERCHANGE_DIR, credentialEncryptionKey, maxActiveDeployments, processEnv: childProcessEnv(environment), }), - adoptSidecar: adoptLocalSidecar, stopStaleSidecar: stopStaleLocalSidecar, manifests: createLocalSidecarManifestStore({ dataRoot, @@ -290,7 +293,12 @@ export async function createWorkflowServices(options: { retireProvisionerSidecar: allocationStore.retireProvisionerSidecar, isConnected: router.holdsAllocatedBinding, }); - await restoreLocalSidecars(db, local, adapterManifest); + await restoreLocalSidecars( + db, + local, + adapterManifest, + router.holdsAllocatedBinding, + ); const pending = new Set>(); const track = (run: () => Promise): Promise => { const task = Promise.resolve().then(run); @@ -394,7 +402,9 @@ export async function createWorkflowServices(options: { }, async drain(options?: { readonly keepSidecar?: boolean }) { await Promise.allSettled(pending); - await local.shutdown(options); + await local.shutdown({ + keepSidecar: sidecarOutlivesHub && options?.keepSidecar === true, + }); }, }; } diff --git a/apps/hub/src/sidecar/local-provisioner.ts b/apps/hub/src/sidecar/local-provisioner.ts index 4a02dbf5..99fa3cee 100644 --- a/apps/hub/src/sidecar/local-provisioner.ts +++ b/apps/hub/src/sidecar/local-provisioner.ts @@ -24,10 +24,10 @@ import type { LocalSidecarHold, LocalSidecarManifest, LocalSidecarManifestStore, - LocalSidecarPid, } from "./manifest-store.ts"; const DEFAULT_STOP_TIMEOUT_MS = 25_000; +const DEFAULT_ADOPTED_CONNECT_TIMEOUT_MS = 60_000; /** The one local sidecar's SIDECAR_DATA_DIR under the provisioner's data root. */ export function localSidecarDataDir(dataRoot: string): string { @@ -42,10 +42,10 @@ export type LocalSidecarExit = { export interface LocalSidecarProcess { readonly pid: number; - /** What `ps` reported the process started as, so a later hub can tell it from a process that reuses its pid. */ - readonly identity?: string; readonly exited: Promise; kill(signal: NodeJS.Signals): void; + /** Stops watching the process and leaves it running, so it no longer keeps the hub's process alive. */ + release(): void; } export type SpawnLocalSidecar = (args: { @@ -53,13 +53,20 @@ export type SpawnLocalSidecar = (args: { readonly adapterManifest: AdapterManifest; }) => LocalSidecarProcess; +/** + * Stops every sidecar process on `dataDir` except one spawned from exactly `spawnedFrom`, which it returns for the hub + * to adopt. + */ +export type ReclaimLocalSidecar = (args: { + readonly dataDir: string; + readonly spawnedFrom: Parameters[0] | undefined; + readonly timeoutMs: number; +}) => Promise; + type CreateLocalProcessSidecarProvisionerOpts = { readonly dataRoot: string; readonly spawnSidecar: SpawnLocalSidecar; - /** The sidecar a previous hub left running, when it is still the process recorded. */ - readonly adoptSidecar: ( - record: LocalSidecarPid, - ) => LocalSidecarProcess | undefined; + readonly reclaimSidecar: ReclaimLocalSidecar; /** Stops a sidecar still running at a recorded pid; resolves whether one was running there. */ readonly stopStaleSidecar: ( pid: number, @@ -70,6 +77,8 @@ type CreateLocalProcessSidecarProvisionerOpts = { /** Operator config, so a respawned sidecar loads the hub's adapters, not the ones it first started with. */ readonly adapterManifest: AdapterManifest; readonly stopTimeoutMs?: number; + /** How long an adopted sidecar has to reconnect before it is replaced. */ + readonly adoptedConnectTimeoutMs?: number; readonly log: (entry: Record) => void; }; @@ -84,6 +93,8 @@ type LocalSidecarRecords = { stillOpen(allocationIds: readonly string[]): Promise>; /** Whether the sidecar was minted with the hub's current adapter manifest and may take work again. */ reusable(sidecarId: string): Promise; + /** Whether the sidecar is connected to this hub for this allocation, at this generation. */ + connected(hold: LocalSidecarHold): boolean; }; export interface LocalProcessSidecarProvisioner { @@ -125,12 +136,13 @@ async function exitedWithin( export function createLocalProcessSidecarProvisioner({ dataRoot, spawnSidecar, - adoptSidecar, + reclaimSidecar, stopStaleSidecar, manifests, capacity, adapterManifest, stopTimeoutMs = DEFAULT_STOP_TIMEOUT_MS, + adoptedConnectTimeoutMs = DEFAULT_ADOPTED_CONNECT_TIMEOUT_MS, log, }: CreateLocalProcessSidecarProvisionerOpts): LocalProcessSidecarProvisioner { if (stopTimeoutMs <= 0) { @@ -146,6 +158,7 @@ export function createLocalProcessSidecarProvisioner({ let shutdownPromise: Promise | null = null; let stillOpen: LocalSidecarRecords["stillOpen"] | undefined; let host: SidecarProvisionerHost | undefined; + let adoptedWatch: ReturnType | undefined; // Ensure and destroy answer from the sidecar and holds restore brings back, so they wait for it; a failed restore // fails them rather than letting an ensure start a second sidecar. const restored = Promise.withResolvers(); @@ -216,19 +229,11 @@ export function createLocalProcessSidecarProvisioner({ return managed; } - // The recorded pid and identity let the next hub adopt this process, or stop it when it is no longer this one. async function start(current: LocalSidecarManifest): Promise { - const handle = spawnSidecar({ manifest: current, adapterManifest }); - try { - await manifests.writePid({ - pid: handle.pid, - ...(handle.identity !== undefined && { identity: handle.identity }), - }); - } catch (error) { - handle.kill("SIGKILL"); - throw error; - } - return manage(handle, current.sidecarId); + return manage( + spawnSidecar({ manifest: current, adapterManifest }), + current.sidecarId, + ); } /** The running sidecar, started under `request`'s identity when none was minted yet. */ @@ -468,46 +473,75 @@ export function createLocalProcessSidecarProvisioner({ await fs.rm(dataDir, { recursive: true, force: true }); } - // The sidecar a previous hub left running, still the process it recorded. A process at that pid that is not the - // recorded one is not ours and is left alone; only a bare pid from a hub before identities is stopped there, so - // two processes never share one identity and data dir. - async function reclaim(): Promise { - const record = await manifests.readPid(); - if (record === null) return undefined; - const adopted = adoptSidecar(record); - if (adopted !== undefined) return adopted; - if ( - record.identity === undefined && - (await stopStaleSidecar(record.pid, stopTimeoutMs)) - ) - log({ - level: "warn", - msg: "local_sidecar_stale_stopped", - pid: record.pid, - }); + // Hubs from before data-dir tagging recorded only a pid; the sidecar one left running there is stopped. + async function stopStale(): Promise { + const pid = await manifests.readPid(); + if (pid === null) return; + if (await stopStaleSidecar(pid, stopTimeoutMs)) + log({ level: "warn", msg: "local_sidecar_stale_stopped", pid }); await manifests.removePid(); - return undefined; + } + + // An adopted sidecar that has not reconnected for any hold it hosts by then is wedged, so it is replaced the way a + // hub restart replaced every sidecar before adoption. + function watchAdopted( + adopted: ManagedProcess, + current: LocalSidecarManifest, + connected: LocalSidecarRecords["connected"], + ): void { + adoptedWatch = setTimeout(() => { + serialize(undefined, async () => { + if (running !== adopted || adopted.exited) return; + if (holds.size === 0 || [...holds.values()].some(connected)) return; + log({ + level: "warn", + msg: "local_sidecar_adopted_unreachable", + sidecarId: current.sidecarId, + pid: adopted.handle.pid, + }); + await stopProcess(adopted); + const child = await start(current); + log({ + level: "info", + msg: "local_sidecar_respawned", + sidecarId: current.sidecarId, + pid: child.handle.pid, + }); + }).catch((error: unknown) => { + if (shutdownPromise === null) + log({ + level: "error", + msg: "local_sidecar_replace_failed", + sidecarId: current.sidecarId, + error: errorText(error), + }); + }); + }, adoptedConnectTimeoutMs); } // A copy whose hold a destroy released is removed only once the sidecar reconnects and the hub undeploys it, so the // sidecar comes back for that cleanup too; forgetting it would leave the copy holding a slot for good. async function restoreSidecar(records: LocalSidecarRecords): Promise { - const { isLive, hasWork, reusable } = records; + const { isLive, hasWork, reusable, connected } = records; stillOpen = records.stillOpen; - const left = await reclaim(); + await stopStale(); + await restoreFences(); + const stored = await manifests.read(); + const left = await reclaimSidecar({ + dataDir, + spawnedFrom: + stored === null ? undefined : { manifest: stored, adapterManifest }, + timeoutMs: stopTimeoutMs, + }); // A sidecar left running for an identity this hub does not keep has nothing to come back for. const retire = async (): Promise => { if (left === undefined) return; await stopProcess({ handle: left, exited: false }); log({ level: "info", msg: "local_sidecar_retired", pid: left.pid }); }; - await restoreFences(); - const stored = await manifests.read(); const recorded = await manifests.readHolds(); if (stored === null) { - await retire(); if (recorded.length > 0) await forget(); - else if (left !== undefined) await manifests.removePid(); return; } const live: LocalSidecarHold[] = []; @@ -552,7 +586,7 @@ export function createLocalProcessSidecarProvisioner({ for (const hold of live) holds.set(hold.allocationId, hold); await manifests.writeHolds(live); if (left !== undefined) { - manage(left, stored.sidecarId); + watchAdopted(manage(left, stored.sidecarId), stored, connected); log({ level: "info", msg: "local_sidecar_adopted", @@ -592,8 +626,10 @@ export function createLocalProcessSidecarProvisioner({ // The manifest, holds and data dir survive shutdown so the next hub boot can adopt or respawn the sidecar. async function stopAll(keepSidecar: boolean): Promise { + clearTimeout(adoptedWatch); await queue.idle(); if (keepSidecar) { + running?.handle.release(); if (running !== null && !running.exited) log({ level: "info", diff --git a/apps/hub/src/sidecar/manifest-store.ts b/apps/hub/src/sidecar/manifest-store.ts index c60edcab..7cfaf71e 100644 --- a/apps/hub/src/sidecar/manifest-store.ts +++ b/apps/hub/src/sidecar/manifest-store.ts @@ -34,14 +34,6 @@ const LocalSidecarFence = type({ export type LocalSidecarFence = typeof LocalSidecarFence.infer; const LocalSidecarFences = LocalSidecarFence.array(); -/** The sidecar process last started on a data root, and what `ps` reported it started as. */ -const LocalSidecarPid = type({ - pid: "number.integer > 0", - "identity?": "string > 0", -}); -export type LocalSidecarPid = typeof LocalSidecarPid.infer; -const StoredLocalSidecarPid = type("string.json.parse").to(LocalSidecarPid); - export interface LocalSidecarManifestStore { read(): Promise; write(manifest: LocalSidecarManifest): Promise; @@ -51,9 +43,8 @@ export interface LocalSidecarManifestStore { /** The stored fences, or null when the file could not be read and was moved aside. */ readFences(): Promise; writeFences(fences: readonly LocalSidecarFence[]): Promise; - /** The sidecar process last started on this data root, or null when none is recorded. */ - readPid(): Promise; - writePid(record: LocalSidecarPid): Promise; + /** The pid a hub from before data-dir tagging recorded for its sidecar, or null when none is recorded. */ + readPid(): Promise; removePid(): Promise; } @@ -187,18 +178,9 @@ export function createLocalSidecarManifestStore({ await writeAtomically(fencesFile, JSON.stringify(fences)); } - // A bare pid is what hubs before identities wrote; it is still stopped, never adopted. - async function readPid(): Promise { - const text = (await readOptional(pidFile))?.trim(); - if (text === undefined) return null; - const stored = StoredLocalSidecarPid(text); - if (!(stored instanceof type.errors)) return stored; - const pid = Number(text); - return Number.isSafeInteger(pid) && pid > 0 ? { pid } : null; - } - - async function writePid(record: LocalSidecarPid): Promise { - await writeAtomically(pidFile, JSON.stringify(record)); + async function readPid(): Promise { + const pid = Number((await readOptional(pidFile))?.trim()); + return Number.isSafeInteger(pid) && pid > 0 ? pid : null; } async function removePid(): Promise { @@ -214,7 +196,6 @@ export function createLocalSidecarManifestStore({ readFences, writeFences, readPid, - writePid, removePid, }; } diff --git a/apps/hub/src/sidecar/restore.ts b/apps/hub/src/sidecar/restore.ts index 1ef31096..86613dc7 100644 --- a/apps/hub/src/sidecar/restore.ts +++ b/apps/hub/src/sidecar/restore.ts @@ -12,6 +12,7 @@ export async function restoreLocalSidecars( db: DB["db"], local: LocalProcessSidecarProvisioner, adapterManifest: AdapterManifest, + connected: (hold: LocalSidecarHold) => boolean, ): Promise { const credentials = createSidecarCredentialResolver({ db }); const currentDigest = adapterManifestDigest(adapterManifest); @@ -67,5 +68,5 @@ export async function restoreLocalSidecars( ); return new Set([...allocations, ...probes].map((row) => row.id)); } - await local.restore({ isLive, hasWork, stillOpen, reusable }); + await local.restore({ isLive, hasWork, stillOpen, reusable, connected }); } diff --git a/apps/hub/src/sidecar/spawn.ts b/apps/hub/src/sidecar/spawn.ts index 9b7e98a0..30262213 100644 --- a/apps/hub/src/sidecar/spawn.ts +++ b/apps/hub/src/sidecar/spawn.ts @@ -1,10 +1,12 @@ +import { createHash } from "node:crypto"; +import { readFileSync } from "node:fs"; +import path from "node:path"; import { hasCode } from "@intx/types"; import type { LocalSidecarExit, - LocalSidecarProcess, + ReclaimLocalSidecar, SpawnLocalSidecar, } from "./local-provisioner.ts"; -import type { LocalSidecarPid } from "./manifest-store.ts"; /** The variables a sidecar needs to find binaries and scratch space; nothing else from the hub env crosses over. */ type ChildProcessEnv = { PATH?: string; HOME?: string; TMPDIR?: string }; @@ -20,7 +22,7 @@ export function childProcessEnv( }; } -type CreateSpawnLocalSidecarOpts = { +type CreateLocalSidecarLauncherOpts = { readonly interchangeDir: string; readonly credentialEncryptionKey: string; readonly maxActiveDeployments: number; @@ -28,24 +30,11 @@ type CreateSpawnLocalSidecarOpts = { }; const SIDECAR_ENTRY = "apps/sidecar/src/index.ts"; +const WATCHDOG_ENTRY = path.join(import.meta.dir, "watchdog.ts"); +// How long a sidecar left running waits for a hub to answer before it stops itself. +const ORPHAN_TIMEOUT_MS = 10 * 60_000; const EXIT_POLL_MS = 100; -// When the process at `pid` started, and its command. Together they tell one process from a later one that reuses -// its pid. -function processAt(pid: number): string | undefined { - const ps = Bun.spawnSync( - ["ps", "-o", "lstart=", "-o", "command=", "-p", String(pid)], - { env: { ...process.env, LC_ALL: "C" }, stderr: "ignore" }, - ); - return ps.exitCode === 0 ? ps.stdout.toString().trim() : undefined; -} - -/** The identity of the sidecar running at `pid`, or undefined when no sidecar runs there. */ -function localSidecarIdentity(pid: number): string | undefined { - const identity = processAt(pid); - return identity?.includes(SIDECAR_ENTRY) ? identity : undefined; -} - // Signals a process that is still running; false when it no longer exists. function sendSignal(pid: number, name: NodeJS.Signals | 0): boolean { try { @@ -57,48 +46,89 @@ function sendSignal(pid: number, name: NodeJS.Signals | 0): boolean { } } -/** - * Stop the sidecar a hub that recorded no identity left running at `pid`, so two processes never share one identity - * and data dir. The process is no child of this hub, so its exit is polled. A pid that no longer runs the - * sidecar entry is left alone. - */ -export async function stopStaleLocalSidecar( +type ProcessRow = { + readonly pid: number; + readonly ppid: number; + readonly command: string; +}; + +function processes(pid?: number): ProcessRow[] { + const ps = Bun.spawnSync( + [ + "ps", + "-ww", + "-o", + "pid=,ppid=,command=", + ...(pid === undefined ? ["-A"] : ["-p", String(pid)]), + ], + { stderr: "ignore" }, + ); + return ps.stdout + .toString() + .split("\n") + .flatMap((line) => { + const match = /^\s*(\d+)\s+(\d+)\s+(.*)$/.exec(line); + return match === null + ? [] + : [ + { + pid: Number(match[1]), + ppid: Number(match[2]), + command: match[3]!, + }, + ]; + }); +} + +function commandAt(pid: number): string | undefined { + return processes(pid)[0]?.command; +} + +function hasArgument(command: string, argument: string): boolean { + return `${command} `.includes(` ${argument} `); +} + +// Signals `pid` only while it still runs `command`, so a pid the system has since reused is never signalled. +function signalIfStill( + pid: number, + command: string, + name: NodeJS.Signals | 0, +): boolean { + return commandAt(pid) === command && sendSignal(pid, name); +} + +async function stopProcessAt( pid: number, + command: string, timeoutMs: number, -): Promise { - if (localSidecarIdentity(pid) === undefined) return false; +): Promise { for (const name of ["SIGTERM", "SIGKILL"] as const) { - if (!sendSignal(pid, name)) return true; + if (!signalIfStill(pid, command, name)) return; for (const deadline = Date.now() + timeoutMs; Date.now() < deadline;) { - if (!sendSignal(pid, 0)) return true; + if (!signalIfStill(pid, command, 0)) return; await Bun.sleep(EXIT_POLL_MS); } } - throw new Error(`Stale local sidecar process ${String(pid)} did not exit`); + throw new Error(`Local sidecar process ${String(pid)} did not exit`); } /** - * The sidecar a previous hub left running, when the process at `record.pid` is still the one it started. The process - * is no child of this hub, so its exit is polled. + * Stop the sidecar a hub from before data-dir tagging left running at `pid`, so two processes never share one + * identity and data dir. A pid that runs anything else, a tagged sidecar included, is left alone. */ -export function adoptLocalSidecar( - record: LocalSidecarPid, -): LocalSidecarProcess | undefined { - const { pid, identity } = record; - if (identity === undefined || localSidecarIdentity(pid) !== identity) - return undefined; - const exited = (async (): Promise => { - while (sendSignal(pid, 0)) await Bun.sleep(EXIT_POLL_MS); - return { code: null, signal: null }; - })(); - return { - pid, - identity, - exited, - kill(signal) { - sendSignal(pid, signal); - }, - }; +export async function stopStaleLocalSidecar( + pid: number, + timeoutMs: number, +): Promise { + const command = commandAt(pid); + if ( + command === undefined || + !command.includes(SIDECAR_ENTRY) || + command.includes("--data-dir=") + ) + return false; + await stopProcessAt(pid, command, timeoutMs); + return true; } async function exitOf(child: Bun.Subprocess): Promise { @@ -110,37 +140,149 @@ async function exitOf(child: Bun.Subprocess): Promise { }; } -export function createSpawnLocalSidecar({ +function recordedInterchangeRevision(interchangeDir: string): string { + try { + const manifest: unknown = JSON.parse( + readFileSync(`${interchangeDir}.manifest.json`, "utf8"), + ); + if ( + manifest !== null && + typeof manifest === "object" && + "revision" in manifest && + typeof manifest.revision === "string" + ) + return manifest.revision; + } catch (error) { + if (!(hasCode(error) && error.code === "ENOENT")) throw error; + } + return "unrecorded"; +} + +export function createLocalSidecarLauncher({ interchangeDir, credentialEncryptionKey, maxActiveDeployments, processEnv, -}: CreateSpawnLocalSidecarOpts): SpawnLocalSidecar { - return function spawnSidecar({ manifest, adapterManifest }) { - const child = Bun.spawn(["bun", "--conditions=intx-src", SIDECAR_ENTRY], { - cwd: interchangeDir, - env: { - ...processEnv, - HUB_WS_URL: manifest.hubWebSocketUrl, - SIDECAR_ID: manifest.sidecarId, - SIDECAR_TOKEN: manifest.token, - SIDECAR_DATA_DIR: manifest.dataDir, - SIDECAR_CREDENTIAL_ENCRYPTION_KEY: credentialEncryptionKey, - SIDECAR_ADAPTER_MANIFEST: JSON.stringify(adapterManifest), - SIDECAR_MAX_ACTIVE_DEPLOYMENTS: String(maxActiveDeployments), +}: CreateLocalSidecarLauncherOpts): { + readonly spawnSidecar: SpawnLocalSidecar; + readonly reclaimSidecar: ReclaimLocalSidecar; +} { + const interchangeRevision = recordedInterchangeRevision(interchangeDir); + const absolute = (dataDir: string) => path.resolve(interchangeDir, dataDir); + + // Everything the sidecar is spawned from; a sidecar spawned from anything else is not adopted. + const fingerprintOf: (...args: Parameters) => string = ({ + manifest, + adapterManifest, + }) => + createHash("sha256") + .update( + JSON.stringify([ + interchangeRevision, + manifest.hubWebSocketUrl, + manifest.sidecarId, + manifest.token, + absolute(manifest.dataDir), + credentialEncryptionKey, + maxActiveDeployments, + processEnv, + adapterManifest, + ORPHAN_TIMEOUT_MS, + ]), + ) + .digest("hex"); + + const spawnSidecar: SpawnLocalSidecar = (args) => { + const { manifest, adapterManifest } = args; + const child = Bun.spawn( + [ + "bun", + WATCHDOG_ENTRY, + `--data-dir=${absolute(manifest.dataDir)}`, + `--fingerprint=${fingerprintOf(args)}`, + `--orphan-timeout-ms=${String(ORPHAN_TIMEOUT_MS)}`, + ], + { + cwd: interchangeDir, + env: { + ...processEnv, + HUB_WS_URL: manifest.hubWebSocketUrl, + SIDECAR_ID: manifest.sidecarId, + SIDECAR_TOKEN: manifest.token, + SIDECAR_DATA_DIR: manifest.dataDir, + SIDECAR_CREDENTIAL_ENCRYPTION_KEY: credentialEncryptionKey, + SIDECAR_ADAPTER_MANIFEST: JSON.stringify(adapterManifest), + SIDECAR_MAX_ACTIVE_DEPLOYMENTS: String(maxActiveDeployments), + }, + stdin: "ignore", + stdout: "inherit", + stderr: "inherit", }, - stdin: "ignore", - stdout: "inherit", - stderr: "inherit", - }); - const identity = localSidecarIdentity(child.pid); + ); return { pid: child.pid, - ...(identity !== undefined && { identity }), exited: exitOf(child), kill(signal) { child.kill(signal); }, + release() { + child.unref(); + }, }; }; + + // Any process tagged with the data dir is ours, whoever started it: one spawned from exactly the inputs given is + // adopted, and every other one is stopped, so two sidecars never run on one data dir. + const reclaimSidecar: ReclaimLocalSidecar = async ({ + dataDir, + spawnedFrom, + timeoutMs, + }) => { + const tag = `--data-dir=${absolute(dataDir)}`; + const ours = processes().filter( + ({ command }) => + (command.includes(WATCHDOG_ENTRY) || command.includes(SIDECAR_ENTRY)) && + hasArgument(command, tag), + ); + const fingerprint = + spawnedFrom === undefined + ? undefined + : `--fingerprint=${fingerprintOf(spawnedFrom)}`; + const kept = + fingerprint === undefined + ? undefined + : ours.find( + ({ command }) => + command.includes(WATCHDOG_ENTRY) && + hasArgument(command, fingerprint), + ); + await Promise.all( + ours + .filter((row) => row !== kept && row.ppid !== kept?.pid) + .map(({ pid, command }) => stopProcessAt(pid, command, timeoutMs)), + ); + if (kept === undefined) return undefined; + const { pid, command } = kept; + let released = false; + const exited = (async (): Promise => { + while (signalIfStill(pid, command, 0)) { + // A released sidecar outlives this hub, so its exit is never seen here. + if (released) return new Promise(() => undefined); + await Bun.sleep(EXIT_POLL_MS); + } + return { code: null, signal: null }; + })(); + return { + pid, + exited, + kill(signal) { + signalIfStill(pid, command, signal); + }, + release() { + released = true; + }, + }; + }; + + return { spawnSidecar, reclaimSidecar }; } diff --git a/apps/hub/src/sidecar/watchdog.ts b/apps/hub/src/sidecar/watchdog.ts new file mode 100644 index 00000000..379168c9 --- /dev/null +++ b/apps/hub/src/sidecar/watchdog.ts @@ -0,0 +1,70 @@ +// Runs the hub's local sidecar and stops it once no hub has answered for `--orphan-timeout-ms`, so a sidecar a hub +// left running across its restart never outlives the hub for good. Both processes carry `--data-dir` on their command +// line, which is how a hub finds the sidecar on its data dir; the sidecar ignores its arguments. `--fingerprint` only +// names the inputs it was spawned from, so a hub adopts it only while they still match. +import { parseArgs } from "node:util"; + +const SIDECAR_ENTRY = "apps/sidecar/src/index.ts"; +const CHECK_INTERVAL_MS = 30_000; +const CHECK_TIMEOUT_MS = 5_000; +const STOP_GRACE_MS = 25_000; + +const { values } = parseArgs({ + options: { + "data-dir": { type: "string" }, + fingerprint: { type: "string" }, + "orphan-timeout-ms": { type: "string" }, + }, +}); +const dataDir = values["data-dir"]; +const orphanTimeoutMs = Number(values["orphan-timeout-ms"]); +const hubWebSocketUrl = process.env["HUB_WS_URL"]; +if ( + dataDir === undefined || + hubWebSocketUrl === undefined || + !(orphanTimeoutMs > 0) +) + throw new Error( + "Sidecar watchdog needs --data-dir, --orphan-timeout-ms and HUB_WS_URL", + ); +const hub = new URL(hubWebSocketUrl); +hub.protocol = hub.protocol === "wss:" ? "https:" : "http:"; + +const child = Bun.spawn( + ["bun", "--conditions=intx-src", SIDECAR_ENTRY, `--data-dir=${dataDir}`], + { stdin: "ignore", stdout: "inherit", stderr: "inherit" }, +); +for (const signal of ["SIGTERM", "SIGINT", "SIGHUP"] as const) + process.on(signal, () => child.kill(signal)); + +// Any answer from the hub's origin counts: the sidecar reconnects on its own once a hub listens there again. +async function hubAnswers(): Promise { + try { + const response = await fetch(hub.origin, { + signal: AbortSignal.timeout(CHECK_TIMEOUT_MS), + }); + await response.body?.cancel(); + return true; + } catch { + return false; + } +} + +let lastAnswered = Date.now(); +const watch = setInterval(async () => { + if (await hubAnswers()) { + lastAnswered = Date.now(); + return; + } + if (Date.now() - lastAnswered < orphanTimeoutMs) return; + clearInterval(watch); + console.error( + `Sidecar watchdog: no hub at ${hub.origin} for ${String(orphanTimeoutMs)} ms; stopping the sidecar`, + ); + child.kill("SIGTERM"); + setTimeout(() => child.kill("SIGKILL"), STOP_GRACE_MS).unref(); +}, CHECK_INTERVAL_MS); + +const code = await child.exited; +clearInterval(watch); +process.exit(child.signalCode === null ? code : 1); diff --git a/deploy/Dockerfile b/deploy/Dockerfile index 1cd736a4..78ea3ef9 100644 --- a/deploy/Dockerfile +++ b/deploy/Dockerfile @@ -84,7 +84,7 @@ COPY --from=deps --chown=bun:bun /app/workflows ./workflows COPY --from=deps --chown=bun:bun /app/vendor/interchange ./vendor/interchange RUN mkdir -p /app/data/hub && chown -R bun:bun /app/data USER bun -ENV HOST=0.0.0.0 PORT=8000 HUB_DATA_DIR=/app/data/hub HUB_WEB_DIR=/app/apps/web/dist +ENV HOST=0.0.0.0 PORT=8000 HUB_DATA_DIR=/app/data/hub HUB_WEB_DIR=/app/apps/web/dist SIDECAR_LOCAL_OUTLIVES_HUB=false EXPOSE 8000 # ChatGPT (1455) and xAI (1456) send a subscription login back to a fixed localhost port, where the # hub listens on loopback only. socat carries the published port from the container's address to it. diff --git a/docs/IMPLEMENTATION.md b/docs/IMPLEMENTATION.md index d668c2e6..cab79ba2 100644 --- a/docs/IMPLEMENTATION.md +++ b/docs/IMPLEMENTATION.md @@ -47,9 +47,13 @@ Agents run on sidecars, which the hub provisions `HUB_DATA_DIR`, and restores them after a restart. One holds up to `SIDECAR_LOCAL_MAX_ACTIVE` deployments. A hub restart (SIGTERM, a fatal error, a failed start or a crash) leaves the sidecar running, and the next - hub adopts it when it is still the process it recorded, so turns in flight - carry on over Interchange's reconnect. SIGINT, as when the dev script exits, - stops it. + hub adopts it when it was spawned from the same inputs, so turns in flight + carry on over Interchange's reconnect. Any other sidecar on the hub's data + dir is stopped, as is an adopted one that does not reconnect within a + minute. SIGINT, as when the dev script exits, stops it, and so does its + watchdog once no hub has answered for ten minutes. A hub that is a + container's first process, or that sets `SIDECAR_LOCAL_OUTLIVES_HUB=false`, + stops it on every shutdown. - **Docker sidecars** are added when `SIDECAR_IMAGE` names an image (built from the `sidecar` target of [deploy/Dockerfile](https://github.com/corbitsdev/corbits-builder/blob/main/deploy/Dockerfile)), diff --git a/test/e2e/hub-restart.test.ts b/test/e2e/hub-restart.test.ts index 4d652a76..435a3070 100644 --- a/test/e2e/hub-restart.test.ts +++ b/test/e2e/hub-restart.test.ts @@ -35,9 +35,9 @@ async function poll( throw new Error(`Timed out waiting for ${description}`); } -// A writer whose model reply is still in flight when the hub restarts. The sidecar outlives the hub, the new hub -// adopts it, and the writer's turn files its document as if nothing happened: no failed run, and no notice that would -// send the coordinator an exchange the recording does not have. +// A writer whose model reply is still in flight when the hub process gets SIGTERM and a new one starts. The sidecar +// outlives the hub, the new hub adopts it, and the writer's turn files its document as if nothing happened: no failed +// run, and no notice that would send the coordinator an exchange the recording does not have. test( "a specialist mid-turn finishes its turn across a hub restart", async () => { @@ -202,14 +202,14 @@ test( const writerTurn = model.hold("writer-1", 0); let writing = false; void writerTurn.reached.then(() => (writing = true)); - // The ask's time comes from Postgres's clock and the brief's from the hub's, so the ask waits until the brief - // is plainly older than it. + // The hub compares the ask's time, from Postgres's clock, with the brief's, from its own, so the ask waits until + // the brief is plainly older. Goes once that comparison uses one clock, which a separate fix does. const ask = model.hold("coordinator-1", 1); void ask.reached.then(async () => { await Bun.sleep(1_000); ask.release(); }); - const started = await startTestHub(model); + const started = await startTestHub(model, { process: true }); hub = started; const { client, workspaceTenantId } = started; const projectId = "restart-pitch"; diff --git a/test/e2e/support/hub.ts b/test/e2e/support/hub.ts index 93d21a07..568146f4 100644 --- a/test/e2e/support/hub.ts +++ b/test/e2e/support/hub.ts @@ -20,7 +20,9 @@ export type TestHubHandle = Readonly<{ runId: string, onFrame: (event: string, data: string) => void, ) => () => void; - /** Stops the hub as a supervisor restart does, leaving its sidecar running, and boots a new one in its place. */ + /** + * Sends a hub started with `process` SIGTERM, as a supervisor restarting it does, and starts another in its place. + */ restart: () => Promise; close: () => Promise; }>; @@ -52,24 +54,15 @@ async function signUpSessionCookie( password: string, name: string, ): Promise { - const signUp = () => - fetch(new URL("/api/auth/sign-up/email", baseUrl), { - method: "POST", - redirect: "error", - headers: { - "Content-Type": "application/json", - Origin: baseUrl, - }, - body: JSON.stringify({ email, password, name }), - }); - let response = await signUp(); - // Hubs in one test process share the auth library's sign-up limit, so a hub booted right after another waits it out. - if (response.status === 429) { - await response.body?.cancel(); - const retryAfter = Number(response.headers.get("X-Retry-After") ?? "10"); - await Bun.sleep((retryAfter + 1) * 1_000); - response = await signUp(); - } + const response = await fetch(new URL("/api/auth/sign-up/email", baseUrl), { + method: "POST", + redirect: "error", + headers: { + "Content-Type": "application/json", + Origin: baseUrl, + }, + body: JSON.stringify({ email, password, name }), + }); if (!response.ok) { const body = await response.text(); throw new Error(`E2E sign-up failed (${response.status}): ${body}`); @@ -85,8 +78,13 @@ async function signUpSessionCookie( return cookies.join("; "); } +const REPO_ROOT = join(import.meta.dir, "..", "..", ".."); +const HUB_READY_TIMEOUT_MS = 120_000; + +/** `process` runs the hub as its own `server.ts` process instead of inside the test, so it can be restarted for real. */ export async function startTestHub( recordedModel: RecordedModelServer, + options: { readonly process?: boolean } = {}, ): Promise { const dbServer = readTestDbServerConfig(); const dbName = `builder_e2e_${crypto.randomUUID().replaceAll("-", "").slice(0, 16)}`; @@ -153,6 +151,7 @@ export async function startTestHub( let runtime: Awaited> | undefined; let server: ReturnType | undefined; + let hubProcess: Bun.Subprocess | undefined; const activeStreams = new Set(); async function dropDatabase(): Promise { @@ -172,20 +171,65 @@ export async function startTestHub( } } - async function boot(): Promise { - runtime = await createHubRuntime(hubEnv); - server = Bun.serve({ - hostname: runtime.hostname, - port: runtime.port, - fetch: runtime.app.fetch, - websocket: runtime.websocket, - }); - runtime.start(); + // The hub serves only once its runtime is built, sidecars restored or adopted included. + async function spawnHub(): Promise { + const started = Bun.spawn( + [ + "bun", + "--no-env-file", + "--conditions=intx-src", + "apps/hub/src/server.ts", + ], + { + cwd: REPO_ROOT, + env: hubEnv, + stdin: "ignore", + stdout: "inherit", + stderr: "inherit", + }, + ); + hubProcess = started; + for (const deadline = Date.now() + HUB_READY_TIMEOUT_MS; ;) { + if (started.exitCode !== null) + throw new Error(`E2E hub exited with code ${String(started.exitCode)}`); + try { + await (await fetch(baseUrl)).body?.cancel(); + return; + } catch { + if (Date.now() > deadline) throw new Error("E2E hub did not start"); + await Bun.sleep(200); + } + } + } + + // SIGINT is the hub's full stop, sidecar included. + async function stopHub(signal: NodeJS.Signals): Promise { + if (!hubProcess) return; + hubProcess.kill(signal); + await hubProcess.exited; + hubProcess = undefined; + } + + async function shutDown(): Promise { + if (server) await server.stop(true); + if (runtime) await runtime.close(); + await stopHub("SIGINT"); } try { await migrateDatabase(hubEnv); - await boot(); + if (options.process) { + await spawnHub(); + } else { + runtime = await createHubRuntime(hubEnv); + server = Bun.serve({ + hostname: runtime.hostname, + port: runtime.port, + fetch: runtime.app.fetch, + websocket: runtime.websocket, + }); + runtime.start(); + } const cookie = await signUpSessionCookie( baseUrl, @@ -327,17 +371,17 @@ export async function startTestHub( fakeCredentialSecret, subscribeRunStream, restart: async () => { - await server?.stop(true); - await runtime?.close({ keepSidecar: true }); - await boot(); + if (!options.process) + throw new Error("Restarting the E2E hub needs a hub process"); + await stopHub("SIGTERM"); + await spawnHub(); }, close: async () => { for (const controller of activeStreams) { controller.abort(); } activeStreams.clear(); - if (server) await server.stop(true); - if (runtime) await runtime.close(); + await shutDown(); await rm(dataDir, { recursive: true, force: true }); await dropDatabase(); }, @@ -347,8 +391,7 @@ export async function startTestHub( controller.abort(); } activeStreams.clear(); - if (server) await server.stop(true); - if (runtime) await runtime.close(); + await shutDown(); await rm(dataDir, { recursive: true, force: true }); await dropDatabase(); throw error;