diff --git a/__tests__/fixtures/refresh-server.cjs b/__tests__/fixtures/refresh-server.cjs index a9eb75ebe..00feeb482 100644 --- a/__tests__/fixtures/refresh-server.cjs +++ b/__tests__/fixtures/refresh-server.cjs @@ -12,6 +12,8 @@ let initialized = false; let initializeParams; let hostQuestion; process.stderr.write(`FIXTURE_PID=${process.pid}\n`); +// Arms the real liveness watchdog, as `serve --mcp` does. +if (options.liveness) require(options.liveness).installMainThreadWatchdog(); const lines = readline.createInterface({ input: process.stdin }); lines.on("line", (line) => { const message = JSON.parse(line); @@ -44,6 +46,7 @@ lines.on("line", (line) => { } else if (message.method === "tools/call") { const name = message.params.name; if (name === "crash") process.exit(9); + if (name === "wedge") for (;;) {} // Dies under the first call only, like a daemon restarting mid-request. if (name === "exit-once") { const marker = path.join(dist, "exit-once.marker"); diff --git a/__tests__/liveness-watchdog.test.ts b/__tests__/liveness-watchdog.test.ts index 9bc6983ec..cbd9144f0 100644 --- a/__tests__/liveness-watchdog.test.ts +++ b/__tests__/liveness-watchdog.test.ts @@ -1,12 +1,14 @@ -import { describe, it, expect, beforeAll } from 'vitest'; +import { describe, it, expect, beforeAll, afterEach, vi } from 'vitest'; import { spawn } from 'child_process'; import * as fs from 'fs'; +import * as os from 'os'; import * as path from 'path'; import { parseWatchdogTimeoutMs, deriveCheckIntervalMs, installMainThreadWatchdog, DEFAULT_WATCHDOG_TIMEOUT_MS, + LivenessMonitor, } from '../src/mcp/liveness-watchdog'; describe('config parsing', () => { @@ -211,3 +213,51 @@ describe('liveness watchdog (spawned, real watchdog process)', () => { expect(code).toBe(3); }, 12000); }); + +describe('LivenessMonitor (the refresh launcher side)', () => { + afterEach(() => vi.useRealTimers()); + + function armed(progressPaths: string[] = []) { + vi.useFakeTimers(); + const kills: string[] = []; + const monitor = new LivenessMonitor((notice) => kills.push(notice)); + monitor.receive(JSON.stringify({ timeoutMs: 1000, capMs: 3000, progressPaths })); + return { monitor, kills }; + } + + it('kills after a silent timeout and not while heartbeats arrive', () => { + const { monitor, kills } = armed(); + for (let i = 0; i < 5; i++) { vi.advanceTimersByTime(900); monitor.receive(''); } + expect(kills).toEqual([]); + vi.advanceTimersByTime(1000); + expect(kills).toHaveLength(1); + expect(kills[0]).toContain('Main thread unresponsive for ~1s'); + }); + + it('does not kill once disarmed', () => { + const { monitor, kills } = armed(); + monitor.receive('disarm'); + vi.advanceTimersByTime(5000); + expect(kills).toEqual([]); + }); + + it('defers while the watched file advances, up to the hard cap', () => { + const dir = fs.mkdtempSync(path.join(os.tmpdir(), 'cg-monitor-')); + const db = path.join(dir, 'db'); + fs.writeFileSync(db, 'a'); + try { + const { kills } = armed([db]); + let size = 1; + const grow = () => fs.writeFileSync(db, 'a'.repeat(++size)); + grow(); vi.advanceTimersByTime(1000); + expect(kills).toEqual([]); + grow(); vi.advanceTimersByTime(1000); + expect(kills).toEqual([]); + grow(); vi.advanceTimersByTime(1000); + expect(kills).toHaveLength(1); + expect(kills[0]).toContain('hard cap 3s reached'); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); +}); diff --git a/__tests__/refresh-launcher.test.ts b/__tests__/refresh-launcher.test.ts index b434f6b97..07c739aad 100644 --- a/__tests__/refresh-launcher.test.ts +++ b/__tests__/refresh-launcher.test.ts @@ -414,6 +414,29 @@ describe("isolated MCP refresh launcher", () => { expect(server.messages.filter((message) => message.id === 1)).toHaveLength(1); }); + it.skipIf(process.platform === "win32")( + "kills a wedged child itself, without a watchdog process, and recovers on a fresh child", + async () => { + const server = await start({ + options: { liveness: resolve(__dirname, "../dist/mcp/liveness-watchdog.js") }, + env: { CODEGRAPH_WATCHDOG_TIMEOUT_MS: "600" }, + }); + const first = await server.call(1); + const pid = first.result.structuredContent.pid; + // The heartbeat goes to the launcher, so the server starts no child. + if (process.platform === "linux") { + expect(readFileSync(`/proc/${pid}/task/${pid}/children`, "utf8").trim()).toBe(""); + } + // Both the child and its replay wedge, so the call is reported. + const wedged = await server.call(2, "wedge"); + expect(wedged.error?.message).toContain("not replayed again"); + expect(server.stderr()).toContain("Main thread unresponsive"); + const recovered = await server.call(3); + expect(recovered.result.structuredContent.pid).not.toBe(pid); + }, + 20000, + ); + it("closes both children and queued requests when the host disconnects during preflight", async () => { const server = await start(); server.deploy(B, { hang: true }); diff --git a/src/mcp/liveness-watchdog.ts b/src/mcp/liveness-watchdog.ts index e0f2e2e2d..c6d8515f0 100644 --- a/src/mcp/liveness-watchdog.ts +++ b/src/mcp/liveness-watchdog.ts @@ -181,6 +181,165 @@ export interface WatchdogOptions { progressPaths?: string[]; } +/** + * The refresh launcher sets this to its own pid and opens a pipe on + * {@link LAUNCHER_LIVENESS_FD} for each server it starts. That server sends its + * heartbeat there instead of starting a watchdog process, and the launcher (a + * separate process, so the wedge cannot reach it either) kills it on silence and + * starts a fresh one. A process with any other parent that inherits the variable + * (a daemon, a CLI command) starts its own watchdog process as before. + */ +export const LAUNCHER_LIVENESS_ENV = 'CODEGRAPH_LAUNCHER_LIVENESS'; +export const LAUNCHER_LIVENESS_FD = 3; + +/** + * Lines on the launcher pipe: this object as JSON arms the watch, an empty line + * is a heartbeat, and `disarm` ends the watch. + */ +interface LivenessArm { + timeoutMs: number; + capMs: number; + progressPaths: string[]; +} + +/** Size + mtime of each watched file; a change is forward disk progress. */ +function progressSnapshot(paths: string[]): string { + let s = ''; + for (const p of paths) { + try { const st = fs.statSync(p); s += st.size + ':' + st.mtimeMs + ';'; } catch { s += 'x;'; } + } + return s; +} + +/** + * The launcher's side of the watch: the same timeout, disk-progress deferral and + * hard cap as the watchdog process ({@link CHILD_SOURCE}), driven by lines from + * the server's pipe. `kill` gets the notice to log and must end the server. + */ +export class LivenessMonitor { + private armed: LivenessArm | null = null; + private timer: NodeJS.Timeout | null = null; + private lastSnap = ''; + private lastSnapAt = 0; + private silentSince: number | null = null; + + constructor(private readonly kill: (notice: string) => void) {} + + receive(line: string): void { + if (line === '') return this.beat(); + if (line === 'disarm') return this.disarm(); + let arm: Partial; + try { arm = JSON.parse(line) as Partial; } catch { return; } + if (!(Number(arm.timeoutMs) > 0) || !(Number(arm.capMs) > 0)) return; + const progressPaths = Array.isArray(arm.progressPaths) ? arm.progressPaths.filter((p) => typeof p === 'string') : []; + this.armed = { timeoutMs: Math.min(Number(arm.timeoutMs), MAX_TIMER_DELAY_MS), capMs: Number(arm.capMs), progressPaths }; + this.lastSnap = progressSnapshot(progressPaths); + this.lastSnapAt = Date.now(); + this.silentSince = null; + this.schedule(); + } + + disarm(): void { + this.armed = null; + if (this.timer) clearTimeout(this.timer); + this.timer = null; + } + + private beat(): void { + if (!this.armed) return; + this.silentSince = null; + // Keep the baseline fresh while healthy (throttled: a stat per second). + if (this.armed.progressPaths.length && Date.now() - this.lastSnapAt >= 1000) { + this.lastSnap = progressSnapshot(this.armed.progressPaths); + this.lastSnapAt = Date.now(); + } + this.schedule(); + } + + private schedule(): void { + if (this.timer) clearTimeout(this.timer); + this.timer = setTimeout(() => this.onTimeout(), this.armed!.timeoutMs); + this.timer.unref?.(); + } + + private onTimeout(): void { + const armed = this.armed; + if (!armed) return; + const notice = (extra: string): string => + `Main thread unresponsive for ~${Math.round(armed.timeoutMs / 1000)}s${extra}; killing the wedged process so a fresh one can start (#850). Disable with CODEGRAPH_NO_WATCHDOG=1.`; + if (!armed.progressPaths.length) { + this.disarm(); + return this.kill(notice('')); + } + const now = Date.now(); + if (this.silentSince === null) this.silentSince = now - armed.timeoutMs; + const cur = progressSnapshot(armed.progressPaths); + if (cur !== this.lastSnap && now - this.silentSince < armed.capMs) { + // Blocked, but the DB files advance: a long store on slow storage. + this.lastSnap = cur; + this.schedule(); + return; + } + this.disarm(); + this.kill(notice(cur !== this.lastSnap ? ` despite ongoing disk activity (hard cap ${Math.round(armed.capMs / 1000)}s reached)` : '')); + } +} + +/** + * Send heartbeats to the refresh launcher that started this process, or return + * null when there is none. Writes go through `fs.write` on the descriptor: a + * `net.Socket` opened on it delivers nothing under Bun 1.4. + */ +function launcherWatchdog(arm: LivenessArm, checkMs: number): WatchdogHandle | null { + if (process.platform === 'win32' || process.env[LAUNCHER_LIVENESS_ENV] !== String(process.ppid)) return null; + try { + const st = fs.fstatSync(LAUNCHER_LIVENESS_FD); + if (!st.isSocket() && !st.isFIFO()) return null; + } catch { + return null; + } + // One write at a time keeps the lines in order. A heartbeat is dropped while + // another line is still being written; that line already shows the loop turns. + const queue: string[] = []; + let writing = false; + let broken = false; + const send = (line: string, droppable = false): void => { + if (broken || (droppable && (writing || queue.length))) return; + queue.push(line + '\n'); + if (!writing) flush(); + }; + const flush = (): void => { + const next = queue.shift(); + if (next === undefined) { writing = false; return; } + writing = true; + fs.write(LAUNCHER_LIVENESS_FD, next, (err) => { + // The launcher can no longer watch this process (or is gone). Exit so a + // launcher that is still running replaces it, rather than leave a wedge unwatched. + if (err && err.code !== 'EAGAIN') { + broken = true; + queue.length = 0; + writing = false; + process.kill(process.pid, 'SIGKILL'); + return; + } + flush(); + }); + }; + send(JSON.stringify(arm)); + const heartbeat = setInterval(() => send('', true), checkMs); + heartbeat.unref(); + debug(`armed through launcher ${process.ppid}: timeoutMs=${arm.timeoutMs} checkMs=${checkMs} progressPaths=${arm.progressPaths.length}`); + let stopped = false; + return { + stop(): void { + if (stopped) return; + stopped = true; + clearInterval(heartbeat); + send('disarm'); + }, + }; +} + /** * Install the main-thread liveness watchdog for a long-lived process. Returns a * handle to stop it, or `null` when disabled or when the child can't be spawned @@ -195,6 +354,9 @@ export function installMainThreadWatchdog(options: WatchdogOptions = {}): Watchd const capMs = timeoutMs * PROGRESS_CAP_MULTIPLIER; const progressPaths = options.progressPaths ?? []; + const viaLauncher = launcherWatchdog({ timeoutMs, capMs, progressPaths }, checkMs); + if (viaLauncher) return viaLauncher; + let child: ChildProcess; try { // No execArgv inheritance (unlike Worker), so the child carries none of our diff --git a/src/mcp/refresh-launcher.ts b/src/mcp/refresh-launcher.ts index 24a2bab07..6c8e65a0b 100644 --- a/src/mcp/refresh-launcher.ts +++ b/src/mcp/refresh-launcher.ts @@ -14,6 +14,8 @@ import { parseHostPpid, parsePpidPollMs, supervisionLostReason } from "./ppid-wa import { HOST_PPID_ENV } from "../extraction/node-runtime-flags"; import { WRITER_LOCK_DEFER_ENV } from "./writer-lock"; import { armStartupHandshakeTimeout } from "./startup-handshake"; +import { LAUNCHER_LIVENESS_ENV, LAUNCHER_LIVENESS_FD, LivenessMonitor } from "./liveness-watchdog"; +import type { Readable } from "stream"; type Id = string | number | null; type Message = { @@ -68,10 +70,12 @@ class Backend { ...args, ], { - stdio: ["pipe", "pipe", "pipe"], + // fd 3 carries the child's liveness heartbeat (see LAUNCHER_LIVENESS_ENV). + stdio: ["pipe", "pipe", "pipe", "pipe"], windowsHide: true, env: { ...process.env, + ...(process.platform === "win32" ? {} : { [LAUNCHER_LIVENESS_ENV]: String(process.pid) }), [HOST_PPID_ENV]: String(parseHostPpid(process.env[HOST_PPID_ENV]) ?? EARLY_PPID), // This child may start while the one it replaces still serves, and // holds the writer lock. It succeeds that child rather than @@ -84,6 +88,22 @@ class Backend { this.child.once("error", (error) => this.fail(error)); this.child.stdin.on("error", (error) => this.fail(error)); this.child.once("close", () => this.fail(new Error("CodeGraph child exited"))); + // Kill a child whose main thread stops turning; the close above then + // replays its calls on a fresh child. + const liveness = this.child.stdio[LAUNCHER_LIVENESS_FD] as Readable | null | undefined; + if (liveness) { + const monitor = new LivenessMonitor((notice) => { + process.stderr.write(`[${new Date().toISOString()}] [CodeGraph refresh] Child ${this.child.pid}: ${notice}\n`); + this.child.kill("SIGKILL"); + }); + // A broken channel leaves the child unwatched; replace it. + liveness.on("error", () => { + monitor.disarm(); + this.child.kill("SIGKILL"); + }); + createInterface({ input: liveness }).on("line", (line) => monitor.receive(line)); + this.child.once("close", () => monitor.disarm()); + } const lines = createInterface({ input: this.child.stdout }); lines.on("line", (line) => { const message = parse(line);