diff --git a/packages/server/scripts/supervisor.logging.test.ts b/packages/server/scripts/supervisor.logging.test.ts index cbb8b51309..bdee5f6f7d 100644 --- a/packages/server/scripts/supervisor.logging.test.ts +++ b/packages/server/scripts/supervisor.logging.test.ts @@ -1,4 +1,4 @@ -import { mkdtemp, readFile, writeFile } from "node:fs/promises"; +import { mkdir, mkdtemp, readFile, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import path from "node:path"; import { fileURLToPath, pathToFileURL } from "node:url"; @@ -23,6 +23,10 @@ async function runSupervisorFixture(options: { workerSource: string; restartOnCrash?: boolean; timeoutMs?: number; + /** POSIX RLIMIT_FSIZE for the supervisor, in 512-byte blocks: writes past it fail like a full disk. */ + fileSizeLimitBlocks?: number; + /** Occupy the log path with a directory so the supervisor cannot open daemon.log. */ + blockLogPath?: boolean; }): Promise<{ code: number | null; signal: NodeJS.Signals | null; @@ -36,6 +40,9 @@ async function runSupervisorFixture(options: { const workerPath = path.join(tempDir, "worker.mjs"); const runnerPath = path.join(tempDir, "runner.mjs"); + if (options.blockLogPath) { + await mkdir(logPath); + } await writeFile( workerPath, `process.send?.({ type: "paseo:ready", listen: "fixture", serverId: "srv_fixture" });\n${options.workerSource}`, @@ -62,7 +69,18 @@ async function runSupervisorFixture(options: { ); const startedAt = Date.now(); - const child = spawn(process.execPath, ["--import", "tsx", runnerPath], { + const runnerArgs = [process.execPath, "--import", "tsx", runnerPath]; + const [command, ...args] = + options.fileSizeLimitBlocks === undefined + ? runnerArgs + : [ + "/bin/sh", + "-c", + `ulimit -f ${options.fileSizeLimitBlocks} && exec "$@"`, + "sh", + ...runnerArgs, + ]; + const child = spawn(command, args, { cwd: repoRoot, env: { ...process.env }, stdio: ["ignore", "pipe", "pipe"], @@ -348,4 +366,55 @@ describe("supervisor durable logging", () => { expect(result.log).toContain("Supervisor exiting"); }, ); + + // POSIX-only: RLIMIT_FSIZE is how the test makes daemon.log writes fail. + test.skipIf(isPlatform("win32"))( + "keeps supervising the worker when daemon.log can no longer be written", + async () => { + const result = await runSupervisorFixture({ + fileSizeLimitBlocks: 128, + workerSource: ` + process.on("message", (message) => { + if (message?.type === "paseo:graceful-shutdown") process.exit(0); + }); + const line = '{"level":30,"msg":"' + "x".repeat(1000) + '"}\\n'; + for (let i = 0; i < 200; i += 1) process.stdout.write(line); + setTimeout(() => { + process.send?.({ type: "paseo:shutdown", reason: "log_write_failure_probe" }); + }, 1000); + setInterval(() => {}, 1000); + `, + }); + + expect(result.stderr).not.toContain("Unhandled 'error' event"); + expect(result.code).toBe(0); + expect(result.signal).toBeNull(); + }, + ); + + test("resumes writing daemon.log once the log path is writable again", async () => { + const result = await runSupervisorFixture({ + blockLogPath: true, + workerSource: ` + import { rmdirSync } from "node:fs"; + + process.on("message", (message) => { + if (message?.type === "paseo:graceful-shutdown") process.exit(0); + }); + process.stdout.write("line while daemon.log is blocked\\n"); + setTimeout(() => { + rmdirSync(process.argv[1].replace(/worker\\.mjs$/, "daemon.log")); + process.stdout.write("line after daemon.log is writable\\n"); + setTimeout(() => { + process.send?.({ type: "paseo:shutdown", reason: "log_recovery_probe" }); + }, 200); + }, 500); + setInterval(() => {}, 1000); + `, + }); + + expect(result.code).toBe(0); + expect(result.log).toContain("line after daemon.log is writable\n"); + expect(result.log).toContain('"reason":"log_recovery_probe"'); + }); }); diff --git a/packages/server/scripts/supervisor.ts b/packages/server/scripts/supervisor.ts index 4798753668..1f8c1374c6 100644 --- a/packages/server/scripts/supervisor.ts +++ b/packages/server/scripts/supervisor.ts @@ -1,7 +1,9 @@ import { fork, spawn, type ChildProcess } from "child_process"; -import { mkdirSync } from "node:fs"; import path from "node:path"; -import { createStream as createRotatingFileStream } from "rotating-file-stream"; +import { + createStream as createRotatingFileStream, + type RotatingFileStream, +} from "rotating-file-stream"; import { signalProcessTree } from "../src/utils/tree-kill.js"; const WORKER_HEARTBEAT_INTERVAL_MS = 1_000; @@ -110,17 +112,58 @@ function toRotatingFileStreamSize(size: string): string { return `${value}${unit}`; } -function createSupervisorLogStream(options: SupervisorLogFileOptions | undefined) { - if (!options) { - return null; - } +// daemon.log is best effort: a full disk or a replaced log path must not take the +// supervisor, and with it the worker, down. A failed stream is dropped and a new one +// is opened on the next write, so logging resumes once the file is writable again. +function createDurableLog( + options: SupervisorLogFileOptions | undefined, + reportFailure: (message: string) => void, +) { + let stream: RotatingFileStream | null = null; + let failing = false; - mkdirSync(path.dirname(options.path), { recursive: true }); - return createRotatingFileStream(path.basename(options.path), { - path: path.dirname(options.path), - size: toRotatingFileStreamSize(options.rotate.maxSize), - maxFiles: options.rotate.maxFiles, - }); + const open = (logFile: SupervisorLogFileOptions): RotatingFileStream => { + const next = createRotatingFileStream(path.basename(logFile.path), { + path: path.dirname(logFile.path), + size: toRotatingFileStreamSize(logFile.rotate.maxSize), + maxFiles: logFile.rotate.maxFiles, + }); + next.on("error", (error) => { + if (stream === next) { + stream = null; + } + if (!failing) { + failing = true; + reportFailure( + `Cannot write ${logFile.path}, will retry on the next log line: ${error.message}`, + ); + } + }); + return next; + }; + + return { + write(chunk: string | Buffer): void { + if (!options) { + return; + } + stream ??= open(options); + stream.write(chunk, (error) => { + if (!error) { + failing = false; + } + }); + }, + close(): Promise { + return new Promise((resolve) => { + if (!stream) { + resolve(); + return; + } + stream.end(() => resolve()); + }); + }, + }; } export function runSupervisor(options: SupervisorOptions): SupervisorController { @@ -137,10 +180,12 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController let lifecycleFailed = false; let publication = Promise.resolve(); let forceKillTimer: NodeJS.Timeout | null = null; - const logStream = createSupervisorLogStream(options.logFile); + const durableLog = createDurableLog(options.logFile, (message) => { + process.stderr.write(`[${options.name}] ${message}\n`); + }); const writeDurableChunk = (chunk: string | Buffer): void => { - logStream?.write(chunk); + durableLog.write(chunk); }; const writeLifecycleLog = (message: string, fields: Record = {}): void => { @@ -161,14 +206,7 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController writeLifecycleLog(message); }; - const closeLogStream = (): Promise => - new Promise((resolve) => { - if (!logStream) { - resolve(); - return; - } - logStream.end(resolve); - }); + const closeLogStream = (): Promise => durableLog.close(); const exitSupervisor = (code: number): void => { if (exiting) {