diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index b3e8864f9d..855c39da59 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -18,6 +18,7 @@ import { writeTerminalWorkerRecord, type TerminalWorkerRecord, } from "./terminal-worker-record.js"; +import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js"; import { resolveTerminalWorkerSocketPath } from "./terminal-worker-socket-path.js"; import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV, @@ -214,13 +215,17 @@ function runWorker(argv: string[]): void { replayController.send(message); } - function persistRecord(): void { + const recordPublisher = createTerminalWorkerRecordPublisher((state) => { + writeTerminalWorkerRecord(args.paseoHome, state); + }); + + async function persistRecord(): Promise { if (!record || exiting) return; record = { ...record, terminals: record.terminals.filter((entry) => sessionsById.has(entry.id)), }; - writeTerminalWorkerRecord(args.paseoHome, record); + await recordPublisher.publish(record); } function removeOwnFiles(): void { @@ -233,6 +238,7 @@ function runWorker(argv: string[]): void { function exitWorker(code: number): void { if (exiting) return; exiting = true; + recordPublisher.stop(); removeOwnFiles(); server?.close(); process.exit(code); @@ -349,7 +355,9 @@ function runWorker(argv: string[]): void { outputCoalescer.flush(); clearTerminalSubscriptions(session.id); sessionsById.delete(session.id); - persistRecord(); + void persistRecord().catch((error) => { + if (!exiting) console.error("Cannot publish terminal exit:", error); + }); if (exitsHeldDuringCreation.has(session.id)) { exitsHeldDuringCreation.set(session.id, info); return; @@ -424,7 +432,7 @@ function runWorker(argv: string[]): void { { id: session.id, pid, startedAt: inspection.snapshot.startedAt }, ], }; - persistRecord(); + await persistRecord(); } function enqueueCreateTerminalRequest(message: TerminalCreateRequest): Promise { diff --git a/packages/server/src/terminal/terminal-worker-record-publisher.test.ts b/packages/server/src/terminal/terminal-worker-record-publisher.test.ts new file mode 100644 index 0000000000..dba50b9efe --- /dev/null +++ b/packages/server/src/terminal/terminal-worker-record-publisher.test.ts @@ -0,0 +1,70 @@ +import { describe, expect, it } from "vitest"; +import { setTimeout as delay } from "node:timers/promises"; +import { createTerminalWorkerRecordPublisher } from "./terminal-worker-record-publisher.js"; +import type { TerminalWorkerRecord } from "./terminal-worker-record.js"; + +const record: TerminalWorkerRecord = { + recordId: "worker", + pid: 123, + bootId: null, + createdAt: "now", + identity: { startedAt: "now", commandLine: "worker" }, + protocolVersion: 1, + paseoVersion: "test", + socketPath: "socket", + secret: "secret", + terminals: [{ id: "terminal", pid: 124, startedAt: "now" }], +}; + +describe("live worker record publication", () => { + it.each(["EPERM", "EBUSY", "EACCES"])( + "publishes the latest state after %s contention clears", + async (code) => { + let locked = true; + const written: TerminalWorkerRecord[] = []; + const publisher = createTerminalWorkerRecordPublisher((state) => { + if (locked) throw Object.assign(new Error("reader holds record"), { code }); + written.push(state); + }); + try { + const creation = publisher.publish(record); + const exited = { ...record, terminals: [] }; + const exit = publisher.publish(exited); + expect(written).toEqual([]); + locked = false; + await Promise.all([creation, exit]); + expect(written).toEqual([exited]); + } finally { + publisher.stop(); + } + }, + ); + + it("does not recreate a retired record when a reader releases it", async () => { + let locked = true; + const written: TerminalWorkerRecord[] = []; + const publisher = createTerminalWorkerRecordPublisher((state) => { + if (locked) + throw Object.assign(new Error("reader holds record"), { + code: "EPERM", + }); + written.push(state); + }); + const pending = publisher.publish(record); + const rejected = expect(pending).rejects.toThrow("stopped"); + publisher.stop(); + locked = false; + await rejected; + await delay(50); + expect(written).toEqual([]); + await expect(publisher.publish(record)).rejects.toThrow("stopped"); + }); + + it("reports permanent storage failures without retrying", async () => { + const publisher = createTerminalWorkerRecordPublisher(() => { + throw Object.assign(new Error("disk full"), { code: "ENOSPC" }); + }); + await expect(publisher.publish(record)).rejects.toThrow("disk full"); + publisher.stop(); + }); +}); diff --git a/packages/server/src/terminal/terminal-worker-record-publisher.ts b/packages/server/src/terminal/terminal-worker-record-publisher.ts new file mode 100644 index 0000000000..415c785102 --- /dev/null +++ b/packages/server/src/terminal/terminal-worker-record-publisher.ts @@ -0,0 +1,62 @@ +import type { TerminalWorkerRecord } from "./terminal-worker-record.js"; + +const TRANSIENT_ERRORS = new Set(["EPERM", "EBUSY", "EACCES"]); + +/** Serializes live worker publication, retaining the latest state across reader contention. + * Creation awaits publication; exit notifications need not wait for a reader to close. + * Stop before retiring the record so a pending retry cannot recreate it. + */ +export function createTerminalWorkerRecordPublisher(write: (record: TerminalWorkerRecord) => void) { + let stopped = false; + let latest: TerminalWorkerRecord; + let timer: ReturnType | undefined; + let pending: + | { + promise: Promise; + resolve: () => void; + reject: (error: unknown) => void; + } + | undefined; + + function attempt(): void { + timer = undefined; + const publication = pending; + if (!publication) return; + try { + write(latest); + publication.resolve(); + pending = undefined; + } catch (error) { + if (TRANSIENT_ERRORS.has((error as NodeJS.ErrnoException).code ?? "")) { + timer = setTimeout(attempt, 25); + } else { + publication.reject(error); + pending = undefined; + } + } + } + + return { + publish(record: TerminalWorkerRecord): Promise { + if (stopped) return Promise.reject(new Error("Worker record publisher stopped")); + latest = record; + if (pending) return pending.promise; + let resolve!: () => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + pending = { promise, resolve, reject }; + attempt(); + return promise; + }, + stop(): void { + stopped = true; + clearTimeout(timer); + timer = undefined; + pending?.reject(new Error("Worker record publisher stopped")); + pending = undefined; + }, + }; +} diff --git a/packages/server/src/terminal/worker-terminal-manager.test.ts b/packages/server/src/terminal/worker-terminal-manager.test.ts index ad066327a2..ffc9b18a45 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -16,7 +16,9 @@ import { afterEach, expect, it } from "vitest"; import { isPlatform } from "../test-utils/platform.js"; import { EventEmitter } from "node:events"; import { + closeSync, existsSync, + openSync, mkdirSync, mkdtempSync, readdirSync, @@ -58,6 +60,7 @@ import { listTerminalWorkerRecords, readTerminalWorkerRecord, terminalWorkerDirectory, + terminalWorkerRecordPath, type TerminalWorkerRecord, } from "./terminal-worker-record.js"; import { TERMINAL_WORKER_PROTOCOL_OVERRIDE_ENV } from "./terminal-worker-protocol.js"; @@ -807,52 +810,81 @@ async function recordedTerminalPids(paseoHome: string): Promise { - const paseoHome = createTemporaryHome(); - const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-")); - temporaryDirs.push(cwd); - const first = await startRealWorkerManager({ paseoHome }); - manager = first; - const ending = await first.createTerminal({ - workspaceId: "ws-unfinished-kill", - cwd, - ...nodeTerminalCommand("setInterval(() => {}, 1000);"), - }); - const remaining = await first.createTerminal({ - workspaceId: "ws-unfinished-kill", - cwd, - ...nodeTerminalCommand("setInterval(() => {}, 1000);"), - }); - const pids = await recordedTerminalPids(paseoHome); - const endingPid = pids.get(ending.id)!; - const remainingPid = pids.get(remaining.id)!; +it.each([false, true])( + "never adopts a live worker's terminals after a kill cleanup it could not finish, even once the worker rewrites its record (held reader: %s)", + async (holdReader) => { + const paseoHome = createTemporaryHome(); + const cwd = mkdtempSync(join(tmpdir(), "worker-terminal-manager-unfinished-kill-")); + temporaryDirs.push(cwd); + const first = await startRealWorkerManager({ paseoHome }); + manager = first; + const exitRequestPath = join(cwd, "exit-requested"); + const exitObservedPath = join(cwd, "exit-observed"); + const ending = await first.createTerminal({ + workspaceId: "ws-unfinished-kill", + cwd, + ...nodeTerminalCommand(` + const { existsSync, writeFileSync } = require("node:fs"); + setInterval(() => { + if (existsSync(${JSON.stringify(exitRequestPath)})) { + writeFileSync(${JSON.stringify(exitObservedPath)}, ""); + process.exit(0); + } + }, 25); + `), + }); + const remaining = await first.createTerminal({ + workspaceId: "ws-unfinished-kill", + cwd, + ...nodeTerminalCommand("setInterval(() => {}, 1000);"), + }); + const pids = await recordedTerminalPids(paseoHome); + const endingPid = pids.get(ending.id)!; + const remainingPid = pids.get(remaining.id)!; + const [record] = await workerRecords(paseoHome); - // A kill cleanup whose process table read fails signals nothing and leaves the worker live. - const inspector = createTerminalWorkerProcessInspector(); - await expect( - reapTerminalWorkers({ - paseoHome, - inspector: { ...inspector, listProcesses: () => Promise.reject(new Error("ps timed out")) }, - }), - ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); - await first.detach(); - expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); + // A kill cleanup whose process table read fails signals nothing and leaves the worker live. + const inspector = createTerminalWorkerProcessInspector(); + await expect( + reapTerminalWorkers({ + paseoHome, + inspector: { + ...inspector, + listProcesses: () => Promise.reject(new Error("ps timed out")), + }, + }), + ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); + expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); - // One terminal ends on its own: the live worker rewrites its record from memory. - process.kill(endingPid, "SIGKILL"); - await waitForCondition( - async () => !(await recordedTerminalPids(paseoHome)).has(ending.id), - 10000, - ); - expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); + // Keep an ordinary reader open until the worker delivers the terminal exit. + // Windows cannot replace the record while this handle is open. The exit must + // still reach the controller, and publication must recover after release. + const recordFd = holdReader + ? openSync(terminalWorkerRecordPath(paseoHome, record!.pid), "r") + : null; + try { + writeFileSync(exitRequestPath, ""); + await waitForCondition(() => first.getTerminal(ending.id) === undefined, 10000); + expect(existsSync(exitObservedPath)).toBe(true); + expect(isProcessAlive(endingPid)).toBe(false); + } finally { + if (recordFd !== null) closeSync(recordFd); + } + await waitForCondition( + async () => !(await recordedTerminalPids(paseoHome)).has(ending.id), + 10000, + ); + expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); - manager = await startRealWorkerManager({ paseoHome }); + await first.detach(); + manager = await startRealWorkerManager({ paseoHome }); - expect(manager.getTerminal(remaining.id)).toBeUndefined(); - expect(await manager.getTerminals(cwd)).toEqual([]); - await waitForCondition(() => !isProcessAlive(remainingPid), 10000); - expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]); -}); + expect(manager.getTerminal(remaining.id)).toBeUndefined(); + expect(await manager.getTerminals(cwd)).toEqual([]); + await waitForCondition(() => !isProcessAlive(remainingPid), 10000); + expect(await readUnconfirmedTerminalCleanup(paseoHome)).toEqual([]); + }, +); // The worker's own `ps`, delayed: a program that exits while the worker reads its birth time // lands its exit in the middle of creation. Only the worker launched while this PATH is set sees