diff --git a/packages/server/src/terminal/terminal-worker-process.ts b/packages/server/src/terminal/terminal-worker-process.ts index 576993da30..c2e784e27a 100644 --- a/packages/server/src/terminal/terminal-worker-process.ts +++ b/packages/server/src/terminal/terminal-worker-process.ts @@ -22,6 +22,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, @@ -217,19 +218,17 @@ function runWorker(argv: string[]): void { controller?.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)), }; - console.error("CI terminal record write attempted"); - try { - writeTerminalWorkerRecord(args.paseoHome, record); - } catch (error) { - console.error("CI terminal record write failed:", (error as NodeJS.ErrnoException).code); - throw error; - } + await recordPublisher.publish(record); } function removeOwnFiles(): void { @@ -242,6 +241,7 @@ function runWorker(argv: string[]): void { function exitWorker(code: number): void { if (exiting) return; exiting = true; + recordPublisher.stop(); removeOwnFiles(); server?.close(); process.exit(code); @@ -370,7 +370,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; @@ -445,7 +447,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 65eb079839..2d72d6cb11 100644 --- a/packages/server/src/terminal/worker-terminal-manager.test.ts +++ b/packages/server/src/terminal/worker-terminal-manager.test.ts @@ -45,7 +45,6 @@ import { listTerminalWorkerRecords, readTerminalWorkerRecord, terminalWorkerDirectory, - terminalWorkerLogPath, terminalWorkerRecordPath, type TerminalWorkerRecord, } from "./terminal-worker-record.js"; @@ -807,45 +806,35 @@ it.each([false, true])( await expect( reapTerminalWorkers({ paseoHome, - inspector: { ...inspector, listProcesses: () => Promise.reject(new Error("ps timed out")) }, + inspector: { + ...inspector, + listProcesses: () => Promise.reject(new Error("ps timed out")), + }, }), ).rejects.toBeInstanceOf(TerminalCleanupNotConfirmedError); - await first.detach(); expect((await readUnconfirmedTerminalCleanup(paseoHome)).length).toBeGreaterThan(0); - // Diagnostic: keep the same read handle the poller uses open across the exit write. + // 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; - const attemptsBeforeExit = - readFileSync(terminalWorkerLogPath(paseoHome), "utf8").match( - /CI terminal record write attempted/g, - )?.length ?? 0; try { writeFileSync(exitRequestPath, ""); - await waitForCondition( - async () => !(await recordedTerminalPids(paseoHome)).has(ending.id), - 10000, - ); - } catch (error) { - const workerLog = readFileSync(terminalWorkerLogPath(paseoHome), "utf8"); - console.error("CI terminal exit diagnostics", { - holdReader, - attemptsAfterExit: - (workerLog.match(/CI terminal record write attempted/g)?.length ?? 0) - - attemptsBeforeExit, - uncaughtExceptionLogged: workerLog.includes("Terminal worker uncaught exception"), - exitObserved: existsSync(exitObservedPath), - endingAlive: isProcessAlive(endingPid), - remainingAlive: isProcessAlive(remainingPid), - recordWriteErrors: workerLog.match(/CI terminal record write failed: \w+/g) ?? [], - }); - throw error; + 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); + await first.detach(); manager = await startRealWorkerManager({ paseoHome }); expect(manager.getTerminal(remaining.id)).toBeUndefined();