Daemon hardening epic

This commit is contained in:
Mohamed Boudra
2026-05-05 18:21:10 +07:00
parent 1571f002c7
commit 0cb9da17cd
27 changed files with 794 additions and 698 deletions
+4 -1
View File
@@ -26,7 +26,10 @@ const supervisorArgs = [
const supervisor = spawn(process.execPath, supervisorArgs, {
stdio: "inherit",
env: process.env,
env: {
...process.env,
PASEO_LOG_FORMAT: process.env.PASEO_LOG_FORMAT ?? "pretty",
},
});
function exitCodeForSignal(signal: NodeJS.Signals): number {
@@ -15,6 +15,17 @@ describe("supervision parity", () => {
expect(daemonRunnerCalls + devRunnerCalls).toBe(1);
});
test("supervisor worker implementation is not server/index", () => {
const daemonRunner = readFileSync(
new URL("./supervisor-entrypoint.ts", import.meta.url),
"utf8",
);
expect(daemonRunner).toContain("daemon-worker");
expect(daemonRunner).not.toContain("server/server/index.js");
expect(daemonRunner).not.toContain("src/server/index.ts");
});
test("dev runner waits asynchronously for supervisor shutdown", () => {
const devRunner = readFileSync(new URL("./dev-runner.ts", import.meta.url), "utf8");
@@ -8,9 +8,14 @@ import {
updatePidLock,
} from "../src/server/pid-lock.js";
import { resolvePaseoHome } from "../src/server/paseo-home.js";
import { loadPersistedConfig } from "../src/server/persisted-config.js";
import { runSupervisor } from "./supervisor.js";
import { applySherpaLoaderEnv } from "../src/server/speech/providers/local/sherpa/sherpa-runtime-env.js";
const DEFAULT_DAEMON_LOG_FILENAME = "daemon.log";
const DEFAULT_LOG_ROTATE_SIZE = "10m";
const DEFAULT_LOG_ROTATE_MAX_FILES = 2;
interface DaemonRunnerConfig {
devMode: boolean;
workerArgs: string[];
@@ -33,10 +38,10 @@ function parseConfig(argv: string[]): DaemonRunnerConfig {
function resolveWorkerEntry(): string {
const candidates = [
fileURLToPath(new URL("../server/server/index.js", import.meta.url)),
fileURLToPath(new URL("../dist/server/server/index.js", import.meta.url)),
fileURLToPath(new URL("../src/server/index.ts", import.meta.url)),
fileURLToPath(new URL("../../src/server/index.ts", import.meta.url)),
fileURLToPath(new URL("../server/server/daemon-worker.js", import.meta.url)),
fileURLToPath(new URL("../dist/server/server/daemon-worker.js", import.meta.url)),
fileURLToPath(new URL("../src/server/daemon-worker.ts", import.meta.url)),
fileURLToPath(new URL("../../src/server/daemon-worker.ts", import.meta.url)),
];
for (const candidate of candidates) {
@@ -49,7 +54,7 @@ function resolveWorkerEntry(): string {
}
function resolveDevWorkerEntry(): string {
const candidate = fileURLToPath(new URL("../src/server/index.ts", import.meta.url));
const candidate = fileURLToPath(new URL("../src/server/daemon-worker.ts", import.meta.url));
if (!existsSync(candidate)) {
throw new Error(`Dev worker entry not found: ${candidate}`);
}
@@ -72,6 +77,28 @@ function resolvePackagedNodeEntrypointRunnerPath(currentScriptPath: string): str
return existsSync(runnerPath) ? runnerPath : null;
}
function resolveSupervisorLogFile(
paseoHome: string,
persistedConfig: ReturnType<typeof loadPersistedConfig>,
) {
const configuredFile = persistedConfig.log?.file;
const configuredPath = configuredFile?.path;
let logPath = path.join(paseoHome, DEFAULT_DAEMON_LOG_FILENAME);
if (configuredPath) {
logPath = path.isAbsolute(configuredPath)
? configuredPath
: path.resolve(paseoHome, configuredPath);
}
return {
path: logPath,
rotate: {
maxSize: configuredFile?.rotate?.maxSize ?? DEFAULT_LOG_ROTATE_SIZE,
maxFiles: configuredFile?.rotate?.maxFiles ?? DEFAULT_LOG_ROTATE_MAX_FILES,
},
};
}
async function main(): Promise<void> {
const config = parseConfig(process.argv.slice(2));
const workerEntry = config.devMode ? resolveDevWorkerEntry() : resolveWorkerEntry();
@@ -85,6 +112,8 @@ async function main(): Promise<void> {
applySherpaLoaderEnv(workerEnv);
const paseoHome = resolvePaseoHome(workerEnv);
const persistedConfig = loadPersistedConfig(paseoHome);
const supervisorLogFile = resolveSupervisorLogFile(paseoHome, persistedConfig);
try {
await acquirePidLock(paseoHome, null, {
@@ -135,6 +164,7 @@ async function main(): Promise<void> {
})
: undefined,
restartOnCrash: config.devMode,
logFile: supervisorLogFile,
onWorkerReady: async ({ listen }) => {
await updatePidLock(paseoHome, { listen }, { ownerPid: process.pid });
},
@@ -0,0 +1,132 @@
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";
const repoRoot = path.resolve(fileURLToPath(new URL("../../..", import.meta.url)));
const supervisorPath = fileURLToPath(new URL("./supervisor.ts", import.meta.url));
async function runSupervisorFixture(options: {
workerSource: string;
restartOnCrash?: boolean;
}): Promise<{
code: number | null;
signal: NodeJS.Signals | null;
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 child = spawn(process.execPath, ["--import", "tsx", runnerPath], {
cwd: repoRoot,
env: { ...process.env },
stdio: ["ignore", "pipe", "pipe"],
});
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"));
}, 10000);
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, log, stdout, stderr };
}
describe("supervisor durable logging", () => {
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 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");
});
});
+97 -6
View File
@@ -1,4 +1,15 @@
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";
interface SupervisorLogFileOptions {
path: string;
rotate: {
maxSize: string;
maxFiles: number;
};
}
type WorkerLifecycleMessage =
| {
@@ -28,6 +39,7 @@ interface SupervisorOptions {
onWorkerReady?: (message: { listen: string }) => Promise<void> | void;
restartOnCrash?: boolean;
onSupervisorExit?: () => Promise<void> | void;
logFile?: SupervisorLogFileOptions;
}
function describeExit(code: number | null, signal: NodeJS.Signals | null): string {
@@ -59,6 +71,31 @@ function parseLifecycleMessage(msg: unknown): WorkerLifecycleMessage | null {
return null;
}
function toRotatingFileStreamSize(size: string): string {
const trimmed = size.trim();
const match = trimmed.match(/^(\d+)\s*([bBkKmMgG])?$/);
if (!match) {
return trimmed;
}
const value = match[1];
const unit = (match[2] ?? "M").toUpperCase();
return `${value}${unit}`;
}
function createSupervisorLogStream(options: SupervisorLogFileOptions | undefined) {
if (!options) {
return null;
}
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,
});
}
export function runSupervisor(options: SupervisorOptions): void {
const restartOnCrash = options.restartOnCrash ?? false;
const workerArgs = options.workerArgs ?? process.argv.slice(2);
@@ -70,11 +107,39 @@ export function runSupervisor(options: SupervisorOptions): void {
let restarting = false;
let shuttingDown = false;
let exiting = false;
const logStream = createSupervisorLogStream(options.logFile);
const writeDurableChunk = (chunk: string | Buffer): void => {
logStream?.write(chunk);
};
const writeLifecycleLog = (message: string, fields: Record<string, unknown> = {}): void => {
writeDurableChunk(
`${JSON.stringify({
level: "info",
time: new Date().toISOString(),
pid: process.pid,
name: options.name,
msg: message,
...fields,
})}\n`,
);
};
const log = (message: string): void => {
process.stderr.write(`[${options.name}] ${message}\n`);
writeLifecycleLog(message);
};
const closeLogStream = (): Promise<void> =>
new Promise((resolve) => {
if (!logStream) {
resolve();
return;
}
logStream.end(resolve);
});
const exitSupervisor = (code: number): void => {
if (exiting) {
return;
@@ -85,6 +150,7 @@ export function runSupervisor(options: SupervisorOptions): void {
const message = error instanceof Error ? error.message : String(error);
log(`Supervisor exit cleanup failed: ${message}`);
})
.then(closeLogStream)
.finally(() => {
process.exit(code);
});
@@ -103,19 +169,30 @@ export function runSupervisor(options: SupervisorOptions): void {
}
const spawnSpec = resolveWorkerSpawnSpec?.(workerEntry) ?? null;
writeLifecycleLog("Spawning worker", { workerEntry });
if (spawnSpec) {
child = spawn(spawnSpec.command, spawnSpec.args, {
stdio: ["inherit", "inherit", "inherit", "ipc"],
stdio: ["inherit", "pipe", "pipe", "ipc"],
env: spawnSpec.env ?? workerEnv,
});
} else {
child = fork(workerEntry, workerArgs, {
stdio: "inherit",
stdio: ["inherit", "pipe", "pipe", "ipc"],
env: workerEnv,
execArgv: workerExecArgv,
});
}
child.stdout?.on("data", (chunk: Buffer) => {
process.stdout.write(chunk);
writeDurableChunk(chunk);
});
child.stderr?.on("data", (chunk: Buffer) => {
process.stderr.write(chunk);
writeDurableChunk(chunk);
});
child.on("message", (msg: unknown) => {
const lifecycleMessage = parseLifecycleMessage(msg);
if (!lifecycleMessage) {
@@ -123,6 +200,7 @@ export function runSupervisor(options: SupervisorOptions): void {
}
if (lifecycleMessage.type === "paseo:ready") {
writeLifecycleLog("Worker ready", { listen: lifecycleMessage.listen });
Promise.resolve(options.onWorkerReady?.({ listen: lifecycleMessage.listen })).catch(
(error) => {
const message = error instanceof Error ? error.message : String(error);
@@ -133,15 +211,21 @@ export function runSupervisor(options: SupervisorOptions): void {
}
if (lifecycleMessage.type === "paseo:shutdown") {
writeLifecycleLog("Worker requested shutdown");
requestShutdown("Shutdown requested by worker");
return;
}
writeLifecycleLog(
"Worker requested restart",
lifecycleMessage.reason ? { reason: lifecycleMessage.reason } : {},
);
requestRestart("Restart requested by worker");
});
child.on("exit", (code, signal) => {
child.on("close", (code, signal) => {
const exitDescriptor = describeExit(code, signal);
writeLifecycleLog("Worker exited", { code, signal, exit: exitDescriptor });
if (shuttingDown) {
log(`Worker exited (${exitDescriptor}). Supervisor shutting down.`);
@@ -151,17 +235,21 @@ export function runSupervisor(options: SupervisorOptions): void {
const crashed =
restartOnCrash &&
((code !== 0 && code !== null) || (signal !== null && signal === "SIGKILL"));
((code !== 0 && code !== null) || (signal !== null && signal !== "SIGTERM"));
if (restarting || crashed) {
restarting = false;
log(`Worker exited (${exitDescriptor}). Restarting worker...`);
log(
crashed
? `Worker crashed (${exitDescriptor}). Restarting worker...`
: `Worker exited (${exitDescriptor}). Restarting worker...`,
);
spawnWorker();
return;
}
log(`Worker exited (${exitDescriptor}). Supervisor exiting.`);
exitSupervisor(typeof code === "number" ? code : 0);
exitSupervisor(typeof code === "number" ? code : 1);
});
};
@@ -170,6 +258,7 @@ export function runSupervisor(options: SupervisorOptions): void {
return;
}
restarting = true;
writeLifecycleLog("Restart requested", { reason });
log(`${reason}. Stopping worker for restart...`);
child.kill("SIGTERM");
};
@@ -180,6 +269,7 @@ export function runSupervisor(options: SupervisorOptions): void {
}
shuttingDown = true;
restarting = false;
writeLifecycleLog("Supervisor shutdown requested", { reason });
log(`${reason}. Stopping worker...`);
if (!child) {
exitSupervisor(0);
@@ -196,5 +286,6 @@ export function runSupervisor(options: SupervisorOptions): void {
process.on("SIGTERM", () => forwardSignal("SIGTERM"));
process.stdout.write(`[${options.name}] ${options.startupMessage}\n`);
writeLifecycleLog(options.startupMessage);
spawnWorker();
}