jb/paseo
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

Preserve daemon sessions when workers stall (#3263)

Browse Source 
Stop treating heartbeat silence as proof that a live daemon worker should be destroyed. Crash recovery, explicit lifecycle requests, and supervisor-loss cleanup remain unchanged.
This commit is contained in:
Mohamed Boudra
2026-08-12 10:58:58 +02:00
committed by GitHub
parent d012b6f071
commit 74f2158c45
6 changed files with 20 additions and 225 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/server/scripts/supervisor.logging.test.ts
+8 -122
View File
@@ -10,15 +10,9 @@ 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;
@@ -58,23 +52,10 @@ async function runSupervisorFixture(options: {
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");
@@ -208,46 +189,21 @@ describe("supervisor durable logging", () => {
expect(result.log).toContain('"workerPid":');
});
test("keeps a worker alive while heartbeats continue", async () => {
test("does not restart a worker based on heartbeat absence", 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,
timeoutMs: 20_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);
process.send?.({ type: "paseo:shutdown", reason: "silent_worker_test_complete" });
}, 16_000);
setInterval(() => {}, 1_000);
} else {
process.send?.({ type: "paseo:shutdown", reason: "unexpected_heartbeat_restart" });
process.send?.({ type: "paseo:shutdown", reason: "unexpected_silent_worker_restart" });
setInterval(() => {}, 1_000);
}
`,
@@ -255,44 +211,10 @@ describe("supervisor durable logging", () => {
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).toContain('"reason":"silent_worker_test_complete"');
expect(result.log).not.toContain('"reason":"unexpected_silent_worker_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,
);
}, 25_000);
test.skipIf(isPlatform("win32"))(
"forces shutdown when a worker ignores SIGTERM",
@@ -315,42 +237,6 @@ describe("supervisor durable logging", () => {
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 () => {
packages/server/scripts/supervisor.ts
-37
View File
@@ -5,7 +5,6 @@ import { createStream as createRotatingFileStream } from "rotating-file-stream";
import { signalProcessTree } from "../src/utils/tree-kill.js";
const WORKER_HEARTBEAT_INTERVAL_MS = 1_000;
const WORKER_HEARTBEAT_MISSED_CHECK_LIMIT = 15;
const WORKER_TERMINATION_GRACE_MS = 10_000;
interface SupervisorLogFileOptions {
@@ -34,10 +33,6 @@ interface SupervisorHeartbeatMessage {
type: "paseo:supervisor-heartbeat";
}
interface WorkerHeartbeatMessage {
type: "paseo:worker-heartbeat";
}
interface SupervisorOptions {
name: string;
startupMessage: string;
@@ -93,15 +88,6 @@ function parseLifecycleMessage(msg: unknown): WorkerLifecycleMessage | null {
return null;
}
function isWorkerHeartbeatMessage(msg: unknown): msg is WorkerHeartbeatMessage {
return (
typeof msg === "object" &&
msg !== null &&
"type" in msg &&
(msg as { type?: unknown }).type === "paseo:worker-heartbeat"
);
}
function toRotatingFileStreamSize(size: string): string {
const trimmed = size.trim();
const match = trimmed.match(/^(\d+)\s*([bBkKmMgG])?$/);
@@ -250,7 +236,6 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
}
const currentChild = child;
let missedWorkerHeartbeatChecks = 0;
const heartbeat = setInterval(() => {
const message: SupervisorHeartbeatMessage = { type: "paseo:supervisor-heartbeat" };
if (currentChild.connected) {
@@ -267,23 +252,6 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
}, WORKER_HEARTBEAT_INTERVAL_MS);
heartbeat.unref();
const workerWatchdog = setInterval(() => {
if (child !== currentChild || restarting || shuttingDown) {
return;
}
missedWorkerHeartbeatChecks += 1;
if (missedWorkerHeartbeatChecks < WORKER_HEARTBEAT_MISSED_CHECK_LIMIT) {
return;
}
writeLifecycleLog("Worker heartbeat timed out; restarting worker", {
missedHeartbeatChecks: missedWorkerHeartbeatChecks,
supervisorPid: process.pid,
workerPid: currentChild.pid ?? null,
});
requestRestart("worker_heartbeat_timeout");
}, WORKER_HEARTBEAT_INTERVAL_MS);
workerWatchdog.unref();
child.on("disconnect", () => {
writeLifecycleLog("Worker IPC channel disconnected");
});
@@ -299,10 +267,6 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
});
child.on("message", (msg: unknown) => {
if (isWorkerHeartbeatMessage(msg)) {
missedWorkerHeartbeatChecks = 0;
return;
}
const lifecycleMessage = parseLifecycleMessage(msg);
if (!lifecycleMessage) {
return;
@@ -333,7 +297,6 @@ export function runSupervisor(options: SupervisorOptions): SupervisorController
child.on("exit", (code, signal) => {
clearInterval(heartbeat);
clearInterval(workerWatchdog);
clearForceKillTimer();
const exitDescriptor = describeExit(code, signal);
writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor });