The worker watchdog treated wall-clock time spent in host sleep as evidence that the worker was unresponsive. Count consecutive observed heartbeat misses instead so sleep does not destroy live terminal sessions while genuinely stalled workers still restart after the same active observation window.
409 lines
14 KiB
TypeScript
409 lines
14 KiB
TypeScript
import { mkdtemp, readFile, writeFile } from "node:fs/promises";
|
|
import { tmpdir } from "node:os";
|
|
import path from "node:path";
|
|
import { fileURLToPath, pathToFileURL } from "node:url";
|
|
import { spawn } from "node:child_process";
|
|
import { describe, expect, test } from "vitest";
|
|
import { isPlatform } from "../src/test-utils/platform.js";
|
|
import { resolveSupervisorLogFile } from "./supervisor-log-config.js";
|
|
|
|
const repoRoot = path.resolve(fileURLToPath(new URL("../../..", import.meta.url)));
|
|
const supervisorPath = fileURLToPath(new URL("./supervisor.ts", import.meta.url));
|
|
|
|
interface SupervisorSuspension {
|
|
afterMs: number;
|
|
durationMs: number;
|
|
}
|
|
|
|
async function runSupervisorFixture(options: {
|
|
workerSource: string;
|
|
restartOnCrash?: boolean;
|
|
suspension?: SupervisorSuspension;
|
|
timeoutMs?: number;
|
|
}): Promise<{
|
|
code: number | null;
|
|
signal: NodeJS.Signals | null;
|
|
elapsedMs: number;
|
|
log: string;
|
|
stdout: string;
|
|
stderr: string;
|
|
}> {
|
|
const tempDir = await mkdtemp(path.join(tmpdir(), "paseo-supervisor-log-"));
|
|
const logPath = path.join(tempDir, "daemon.log");
|
|
const workerPath = path.join(tempDir, "worker.mjs");
|
|
const runnerPath = path.join(tempDir, "runner.mjs");
|
|
|
|
await writeFile(workerPath, options.workerSource);
|
|
await writeFile(
|
|
runnerPath,
|
|
`
|
|
import { runSupervisor } from ${JSON.stringify(pathToFileURL(supervisorPath).href)};
|
|
|
|
runSupervisor({
|
|
name: "TestSupervisor",
|
|
startupMessage: "starting fixture",
|
|
resolveWorkerEntry: () => ${JSON.stringify(workerPath)},
|
|
workerArgs: [],
|
|
workerEnv: process.env,
|
|
workerExecArgv: [],
|
|
restartOnCrash: ${JSON.stringify(options.restartOnCrash ?? false)},
|
|
logFile: {
|
|
path: ${JSON.stringify(logPath)},
|
|
rotate: { maxSize: "1m", maxFiles: 2 },
|
|
},
|
|
});
|
|
`,
|
|
);
|
|
|
|
const startedAt = Date.now();
|
|
const child = spawn(process.execPath, ["--import", "tsx", runnerPath], {
|
|
cwd: repoRoot,
|
|
detached: options.suspension !== undefined,
|
|
env: { ...process.env },
|
|
stdio: ["ignore", "pipe", "pipe"],
|
|
});
|
|
|
|
if (options.suspension) {
|
|
const supervisorPid = child.pid;
|
|
if (supervisorPid === undefined) {
|
|
throw new Error("Supervisor fixture did not start");
|
|
}
|
|
const { afterMs, durationMs } = options.suspension;
|
|
setTimeout(() => {
|
|
process.kill(-supervisorPid, "SIGSTOP");
|
|
setTimeout(() => process.kill(-supervisorPid, "SIGCONT"), durationMs);
|
|
}, afterMs);
|
|
}
|
|
|
|
let stdout = "";
|
|
let stderr = "";
|
|
child.stdout.setEncoding("utf8");
|
|
child.stderr.setEncoding("utf8");
|
|
child.stdout.on("data", (chunk) => {
|
|
stdout += chunk;
|
|
});
|
|
child.stderr.on("data", (chunk) => {
|
|
stderr += chunk;
|
|
});
|
|
|
|
const { code, signal } = await new Promise<{
|
|
code: number | null;
|
|
signal: NodeJS.Signals | null;
|
|
}>((resolve, reject) => {
|
|
const timeout = setTimeout(() => {
|
|
child.kill("SIGKILL");
|
|
reject(new Error("supervisor fixture timed out"));
|
|
}, options.timeoutMs ?? 10_000);
|
|
|
|
child.on("error", (error) => {
|
|
clearTimeout(timeout);
|
|
reject(error);
|
|
});
|
|
child.on("close", (exitCode, exitSignal) => {
|
|
clearTimeout(timeout);
|
|
resolve({ code: exitCode, signal: exitSignal });
|
|
});
|
|
});
|
|
|
|
const log = await readFile(logPath, "utf8");
|
|
return { code, signal, elapsedMs: Date.now() - startedAt, log, stdout, stderr };
|
|
}
|
|
|
|
describe("supervisor durable logging", () => {
|
|
test("resolves rotation defaults", () => {
|
|
const paseoHome = path.join(path.sep, "tmp", "paseo-home");
|
|
const logFile = resolveSupervisorLogFile(paseoHome, {}, {});
|
|
|
|
expect(logFile).toEqual({
|
|
path: path.join(paseoHome, "daemon.log"),
|
|
rotate: { maxSize: "10m", maxFiles: 3 },
|
|
});
|
|
});
|
|
|
|
test("lets persisted rotation override env rotation defaults", () => {
|
|
const paseoHome = path.join(path.sep, "tmp", "paseo-home");
|
|
const logFile = resolveSupervisorLogFile(
|
|
paseoHome,
|
|
{
|
|
log: {
|
|
file: {
|
|
path: "logs/daemon.log",
|
|
rotate: { maxSize: "25m", maxFiles: 4 },
|
|
},
|
|
},
|
|
},
|
|
{
|
|
PASEO_LOG_ROTATE_SIZE: "200m",
|
|
PASEO_LOG_ROTATE_COUNT: "12",
|
|
},
|
|
);
|
|
|
|
expect(logFile).toEqual({
|
|
path: path.resolve(paseoHome, "logs", "daemon.log"),
|
|
rotate: { maxSize: "25m", maxFiles: 4 },
|
|
});
|
|
});
|
|
|
|
test("uses env rotation when persisted rotation is absent", () => {
|
|
const paseoHome = path.join(path.sep, "tmp", "paseo-home");
|
|
const logFile = resolveSupervisorLogFile(
|
|
paseoHome,
|
|
{},
|
|
{
|
|
PASEO_LOG_ROTATE_SIZE: "50m",
|
|
PASEO_LOG_ROTATE_COUNT: "8",
|
|
},
|
|
);
|
|
|
|
expect(logFile).toEqual({
|
|
path: path.join(paseoHome, "daemon.log"),
|
|
rotate: { maxSize: "50m", maxFiles: 8 },
|
|
});
|
|
});
|
|
|
|
test("writes supervised worker stdout and stderr to daemon.log", async () => {
|
|
const result = await runSupervisorFixture({
|
|
workerSource: `
|
|
process.stdout.write('{"level":30,"msg":"worker-json-stdout"}\\n');
|
|
process.stderr.write('{"level":50,"msg":"worker-json-stderr"}\\n');
|
|
process.exit(0);
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"worker-json-stdout"');
|
|
expect(result.log).toContain('"worker-json-stderr"');
|
|
expect(result.stdout).toContain('"worker-json-stdout"');
|
|
expect(result.stderr).toContain('"worker-json-stderr"');
|
|
});
|
|
|
|
test("preserves raw non-JSON stdout and stderr lines", async () => {
|
|
const result = await runSupervisorFixture({
|
|
workerSource: `
|
|
process.stdout.write('raw stdout line\\n');
|
|
process.stderr.write('raw stderr line\\n');
|
|
process.exit(0);
|
|
`,
|
|
});
|
|
|
|
expect(result.log).toContain("raw stdout line\n");
|
|
expect(result.log).toContain("raw stderr line\n");
|
|
});
|
|
|
|
test("logs the worker shutdown reason before signaling the worker", async () => {
|
|
const result = await runSupervisorFixture({
|
|
workerSource: `
|
|
process.send?.({ type: "paseo:shutdown", reason: "client_shutdown_rpc" });
|
|
setInterval(() => {}, 1000);
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"msg":"Worker requested shutdown"');
|
|
expect(result.log).toContain('"reason":"client_shutdown_rpc"');
|
|
expect(result.log).toContain('"msg":"Supervisor sending signal to worker"');
|
|
expect(result.log).toContain('"signal":"SIGTERM"');
|
|
expect(result.log).toContain('"workerPid":');
|
|
});
|
|
|
|
test("keeps a worker alive while heartbeats continue", async () => {
|
|
const result = await runSupervisorFixture({
|
|
timeoutMs: 10_000,
|
|
workerSource: `
|
|
setInterval(() => {
|
|
process.send?.({ type: "paseo:worker-heartbeat" });
|
|
}, 250);
|
|
setTimeout(() => {
|
|
process.send?.({ type: "paseo:shutdown", reason: "healthy_heartbeat_test_complete" });
|
|
}, 6000);
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"reason":"healthy_heartbeat_test_complete"');
|
|
expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"');
|
|
}, 10_000);
|
|
|
|
test("tolerates a transient seven-second heartbeat pause", async () => {
|
|
const result = await runSupervisorFixture({
|
|
timeoutMs: 10_000,
|
|
workerSource: `
|
|
import { existsSync, writeFileSync } from "node:fs";
|
|
|
|
const marker = process.argv[1] + ".started";
|
|
if (!existsSync(marker)) {
|
|
writeFileSync(marker, "started");
|
|
let heartbeatCount = 0;
|
|
const heartbeat = setInterval(() => {
|
|
process.send?.({ type: "paseo:worker-heartbeat" });
|
|
heartbeatCount += 1;
|
|
if (heartbeatCount === 3) clearInterval(heartbeat);
|
|
}, 100);
|
|
setTimeout(() => {
|
|
process.send?.({ type: "paseo:shutdown", reason: "heartbeat_pause_tolerated" });
|
|
}, 7_000);
|
|
setInterval(() => {}, 1_000);
|
|
} else {
|
|
process.send?.({ type: "paseo:shutdown", reason: "unexpected_heartbeat_restart" });
|
|
setInterval(() => {}, 1_000);
|
|
}
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"reason":"heartbeat_pause_tolerated"');
|
|
expect(result.log).not.toContain('"reason":"unexpected_heartbeat_restart"');
|
|
expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"');
|
|
}, 10_000);
|
|
|
|
test.skipIf(isPlatform("win32"))(
|
|
"keeps a healthy worker alive after the host resumes from sleep",
|
|
async () => {
|
|
const result = await runSupervisorFixture({
|
|
suspension: { afterMs: 1_000, durationMs: 16_000 },
|
|
timeoutMs: 25_000,
|
|
workerSource: `
|
|
import { existsSync, writeFileSync } from "node:fs";
|
|
|
|
const marker = process.argv[1] + ".started";
|
|
if (!existsSync(marker)) {
|
|
writeFileSync(marker, "started");
|
|
setInterval(() => {
|
|
process.send?.({ type: "paseo:worker-heartbeat" });
|
|
}, 250);
|
|
setTimeout(() => {
|
|
process.send?.({ type: "paseo:shutdown", reason: "sleep_resume_test_complete" });
|
|
}, 17_000);
|
|
} else {
|
|
process.send?.({ type: "paseo:shutdown", reason: "unexpected_sleep_restart" });
|
|
}
|
|
setInterval(() => {}, 1_000);
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"reason":"sleep_resume_test_complete"');
|
|
expect(result.log).not.toContain('"reason":"unexpected_sleep_restart"');
|
|
expect(result.log).not.toContain('"msg":"Worker heartbeat timed out; restarting worker"');
|
|
},
|
|
30_000,
|
|
);
|
|
|
|
test.skipIf(isPlatform("win32"))(
|
|
"forces shutdown when a worker ignores SIGTERM",
|
|
async () => {
|
|
const result = await runSupervisorFixture({
|
|
timeoutMs: 15_000,
|
|
workerSource: `
|
|
process.on("SIGTERM", () => {});
|
|
process.send?.({ type: "paseo:shutdown", reason: "stalled_worker_shutdown" });
|
|
setInterval(() => {}, 1_000);
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"reason":"stalled_worker_shutdown"');
|
|
expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"');
|
|
expect(result.log).toContain('"signal":"SIGKILL"');
|
|
},
|
|
20_000,
|
|
);
|
|
|
|
// POSIX-only: the watchdog uses SIGKILL after its graceful shutdown window.
|
|
test.skipIf(isPlatform("win32"))(
|
|
"restarts a worker that stops heartbeating",
|
|
async () => {
|
|
const result = await runSupervisorFixture({
|
|
timeoutMs: 35_000,
|
|
workerSource: `
|
|
import { existsSync, writeFileSync } from "node:fs";
|
|
|
|
const marker = process.argv[1] + ".started";
|
|
if (!existsSync(marker)) {
|
|
writeFileSync(marker, "started");
|
|
let heartbeatCount = 0;
|
|
const heartbeat = setInterval(() => {
|
|
process.send?.({ type: "paseo:worker-heartbeat" });
|
|
heartbeatCount += 1;
|
|
if (heartbeatCount === 3) clearInterval(heartbeat);
|
|
}, 100);
|
|
process.on("SIGTERM", () => {});
|
|
setInterval(() => {}, 1000);
|
|
} else {
|
|
process.send?.({ type: "paseo:shutdown", reason: "watchdog_test_complete" });
|
|
setInterval(() => {}, 1000);
|
|
}
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"msg":"Worker heartbeat timed out; restarting worker"');
|
|
expect(result.log).toContain('"msg":"Worker did not exit after SIGTERM; forcing SIGKILL"');
|
|
expect(result.log).toContain('"signal":"SIGKILL"');
|
|
},
|
|
40_000,
|
|
);
|
|
|
|
test.skipIf(isPlatform("win32"))(
|
|
"restarts after worker exit while a descendant retains the worker stdio",
|
|
async () => {
|
|
const result = await runSupervisorFixture({
|
|
timeoutMs: 7_000,
|
|
workerSource: `
|
|
import { spawn } from "node:child_process";
|
|
import { existsSync, writeFileSync } from "node:fs";
|
|
|
|
const marker = process.argv[1] + ".started";
|
|
if (!existsSync(marker)) {
|
|
writeFileSync(marker, "started");
|
|
const descendant = spawn(
|
|
process.execPath,
|
|
["-e", "process.on('SIGTERM', () => {}); setTimeout(() => process.exit(0), 4000)"],
|
|
{ detached: true, stdio: ["ignore", "inherit", "inherit"] },
|
|
);
|
|
descendant.unref();
|
|
process.on("SIGTERM", () => process.exit(0));
|
|
process.send?.({ type: "paseo:restart", reason: "stdio_descendant" });
|
|
setInterval(() => {}, 1000);
|
|
} else {
|
|
process.send?.({ type: "paseo:shutdown", reason: "stdio_restart_complete" });
|
|
setInterval(() => {}, 1000);
|
|
}
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(0);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.elapsedMs).toBeLessThan(2_500);
|
|
expect(result.log).toContain('"reason":"stdio_descendant"');
|
|
expect(result.log).toContain("Restarting worker");
|
|
},
|
|
7_000,
|
|
);
|
|
|
|
// POSIX-only: Windows reports the worker self-kill as an exit code, not SIGKILL.
|
|
test.skipIf(isPlatform("win32"))(
|
|
"logs worker signal exits even when the worker cannot log",
|
|
async () => {
|
|
const result = await runSupervisorFixture({
|
|
workerSource: `
|
|
process.kill(process.pid, "SIGKILL");
|
|
`,
|
|
});
|
|
|
|
expect(result.code).toBe(1);
|
|
expect(result.signal).toBeNull();
|
|
expect(result.log).toContain('"msg":"Worker exited"');
|
|
expect(result.log).toContain('"signal":"SIGKILL"');
|
|
expect(result.log).toContain("Supervisor exiting");
|
|
},
|
|
);
|
|
});
|