fix: prevent race condition spawning multiple cocod and routstrd instances

When multiple routstrd processes (or CLI auto-starts) bootstrap simultaneously,
each process found cocod/routstrd unreachable, spawned its own copy, and
contended for the same database socket. The paired ~200ms burst timestamps
in the logs were the symptom.

Fixes applied:

- Add cross-process startup lock (process-lock.ts)
  - Uses atomic mkdir as the lock primitive.
  - Stores PID + UUID token; verifies ownership on release to avoid
    removing a lock taken by a new process after the original crashed.
  - Removes stale locks (unreachable PID or lock older than staleAfterMs).
  - Exports withCrossProcessLock() for simple RAII-style usage.

- Fix cocod startup race (cocod-client.ts)
  - Wrap ensureDaemonRunning() in a socket-derived lock path.
  - Re-ping inside the lock before spawning; if another process just started
    cocod, the current process detects it and connects instead of spawning.
  - Switch from 'cocod daemon' to 'cocod init' for cleaner daemonization.
  - Allow exit code 0 from 'cocod init' (normal daemon exit without error).

- Consolidate routstrd daemon startup (daemon-client.ts)
  - startDaemonProcess() now delegates to startDaemon() from start-daemon.ts,
    sharing the same lock and health-check logic.

- Add cross-process lock to routstrd daemon startup (start-daemon.ts)
  - Performs pre-lock isDaemonHealthy() check to skip lock entirely when
    already running.
  - Performs re-check inside the lock before spawning to avoid duplicate
    daemon processes.
This commit is contained in:
redshift
2026-04-30 13:36:16 +05:30
parent 55dbb1a6b5
commit cdfde85595
4 changed files with 216 additions and 62 deletions
+22 -6
View File
@@ -1,6 +1,7 @@
import { existsSync } from "fs";
import { createHash } from "crypto";
import { logger } from "../../utils/logger";
import { withCrossProcessLock } from "../../utils/process-lock";
const DEFAULT_CONFIG_DIR =
process.env.COCOD_DIR || `${process.env.HOME || process.env.USERPROFILE || ""}/.cocod`;
@@ -139,6 +140,7 @@ export function createCocodClient(
): CocodClient {
const executable = resolveCocodExecutable(options.cocodPath);
const socketPath = options.socketPath || DEFAULT_SOCKET_PATH;
const startupLockPath = `${socketPath}.startup.lock`;
const fetchImpl = options.fetchImpl || (fetch as CocodFetch);
const pollIntervalMs = options.pollIntervalMs ?? 100;
const startupTimeoutMs = options.startupTimeoutMs ?? 5000;
@@ -228,7 +230,7 @@ export function createCocodClient(
async function startDaemon(): Promise<void> {
const env = { ...process.env, COCOD_SOCKET: socketPath };
const proc = spawnDaemon([executable, "daemon"], env);
const proc = spawnDaemon([executable, "init"], env);
const maxPolls = Math.ceil(startupTimeoutMs / pollIntervalMs);
let exitCode: number | null = null;
@@ -239,8 +241,8 @@ export function createCocodClient(
for (let i = 0; i < maxPolls; i++) {
await delay(pollIntervalMs);
if (exitCode !== null) {
throw new Error(`cocod daemon exited early with code ${exitCode}`);
if (exitCode !== null && exitCode !== 0) {
throw new Error(`cocod init exited early with code ${exitCode}`);
}
if (await pingInternal()) {
@@ -250,7 +252,7 @@ export function createCocodClient(
}
throw new Error(
`cocod daemon failed to start within ${Math.round(startupTimeoutMs / 1000)} seconds`,
`cocod failed to start within ${Math.round(startupTimeoutMs / 1000)} seconds`,
);
}
@@ -260,8 +262,22 @@ export function createCocodClient(
}
if (!startPromise) {
logger.debug(`Starting cocod daemon via ${executable}...`);
startPromise = startDaemon().finally(() => {
startPromise = withCrossProcessLock(
startupLockPath,
async () => {
if (await pingInternal()) {
return;
}
logger.debug(`Starting cocod daemon via ${executable} init...`);
await startDaemon();
},
{
acquireTimeoutMs: startupTimeoutMs + 30_000,
staleAfterMs: startupTimeoutMs + 30_000,
log: (message) => logger.debug(message),
},
).finally(() => {
startPromise = null;
});
}
+52 -30
View File
@@ -1,6 +1,9 @@
import { logger } from "./utils/logger";
import { existsSync } from "fs";
import { LOGS_DIR } from "./utils/config";
import { CONFIG_DIR, LOGS_DIR } from "./utils/config";
import { withCrossProcessLock } from "./utils/process-lock";
const DAEMON_STARTUP_LOCK_PATH = `${CONFIG_DIR}/routstrd-startup.lock`;
function getTodayLogFile(): string {
const now = new Date();
@@ -10,27 +13,32 @@ function getTodayLogFile(): string {
return `${LOGS_DIR}/${year}-${month}-${day}.log`;
}
export async function startDaemon(
options: { port?: string; provider?: string } = {},
async function isDaemonHealthy(port: string): Promise<boolean> {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), 2000);
try {
const existing = await fetch(`http://localhost:${port}/health`, {
signal: controller.signal,
});
return existing.ok;
} catch {
return false;
} finally {
clearTimeout(timeoutId);
}
}
async function startDaemonUnlocked(
options: { port?: string; provider?: string },
): Promise<void> {
const args: string[] = [];
const port = options.port || "8008";
const pollIntervalMs = 250;
const startupTimeoutMs = 10 * 60 * 1000;
try {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), 2000);
const existing = await fetch(`http://localhost:${port}/health`, {
signal: controller.signal,
});
clearTimeout(timeoutId);
if (existing.ok) {
logger.log(`Routstr daemon already running on http://localhost:${port}/v1`);
return;
}
} catch {
// Daemon is not running yet; continue with startup.
if (await isDaemonHealthy(port)) {
logger.log(`Routstr daemon already running on http://localhost:${port}/v1`);
return;
}
if (options.port) {
@@ -47,7 +55,7 @@ export async function startDaemon(
const daemonScript = new URL("./daemon/index.js", import.meta.url).pathname;
const todayLogFile = getTodayLogFile();
const shellCmd = `bun run "${daemonScript}" ${args.map(a => `'${a}'`).join(" ")} >> "${todayLogFile}" 2>&1`;
const shellCmd = `bun run "${daemonScript}" ${args.map((a) => `'${a}'`).join(" ")} >> "${todayLogFile}" 2>&1`;
const proc = Bun.spawn(["sh", "-c", shellCmd], {
stdout: "inherit",
@@ -73,19 +81,9 @@ export async function startDaemon(
);
}
try {
const controller = new AbortController();
const timeoutId = setTimeout(() => controller.abort(), 2000);
const res = await fetch(`http://localhost:${port}/health`, {
signal: controller.signal,
});
clearTimeout(timeoutId);
if (res.ok) {
logger.log(`Routstr daemon started (PID: ${proc.pid}).`);
return;
}
} catch {
// Not ready yet
if (await isDaemonHealthy(port)) {
logger.log(`Routstr daemon started (PID: ${proc.pid}).`);
return;
}
}
@@ -93,3 +91,27 @@ export async function startDaemon(
`Daemon failed to start within ${Math.round(startupTimeoutMs / 1000)} seconds. Check logs in ${LOGS_DIR}`,
);
}
export async function startDaemon(
options: { port?: string; provider?: string } = {},
): Promise<void> {
const port = options.port || "8008";
const startupTimeoutMs = 10 * 60 * 1000;
if (await isDaemonHealthy(port)) {
logger.log(`Routstr daemon already running on http://localhost:${port}/v1`);
return;
}
await withCrossProcessLock(
DAEMON_STARTUP_LOCK_PATH,
async () => {
await startDaemonUnlocked(options);
},
{
acquireTimeoutMs: startupTimeoutMs + 30_000,
staleAfterMs: startupTimeoutMs + 30_000,
log: (message) => logger.debug(message),
},
);
}
+6 -26
View File
@@ -1,8 +1,8 @@
import { existsSync } from "fs";
import { startDaemon } from "../start-daemon";
import {
CONFIG_FILE,
DEFAULT_CONFIG,
LOGS_DIR,
type RoutstrdConfig,
} from "./config";
import {
@@ -115,31 +115,11 @@ export function getNpubSuffix(config: RoutstrdConfig): string | null {
}
export async function startDaemonProcess(): Promise<void> {
// Ensure logs directory exists (logger handles date-based files)
if (!existsSync(LOGS_DIR)) {
await Bun.$`mkdir -p ${LOGS_DIR}`;
}
const proc = Bun.spawn(
["bun", "run", `${import.meta.dir}/../daemon/index.ts`],
{
stdout: "inherit",
stderr: "inherit",
stdin: "ignore",
detached: true,
},
);
proc.unref();
for (let i = 0; i < 50; i++) {
await new Promise((resolve) => setTimeout(resolve, 100));
if (await isDaemonRunning()) {
return;
}
}
throw new Error("Daemon failed to start within 5 seconds");
const config = await loadConfig();
await startDaemon({
port: String(config.port || 8008),
provider: config.provider || undefined,
});
}
export async function ensureDaemonRunning(): Promise<void> {
+136
View File
@@ -0,0 +1,136 @@
import { randomUUID } from "crypto";
import { mkdir, readFile, rm, stat, writeFile } from "fs/promises";
import { dirname } from "path";
export interface CrossProcessLockOptions {
/** How long to wait while another process holds the lock. */
acquireTimeoutMs?: number;
/** How often to retry acquiring the lock. */
retryIntervalMs?: number;
/** Treat locks older than this as stale even if their PID cannot be checked. */
staleAfterMs?: number;
/** Optional logger used when removing stale locks. */
log?: (message: string) => void;
}
interface LockOwner {
pid: number;
createdAt: number;
token?: string;
}
function delay(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
function isProcessRunning(pid: number): boolean {
if (!Number.isFinite(pid) || pid <= 0) {
return false;
}
try {
process.kill(pid, 0);
return true;
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
return code === "EPERM";
}
}
async function readLockOwner(lockDir: string): Promise<LockOwner | null> {
try {
const raw = await readFile(`${lockDir}/owner.json`, "utf8");
const parsed = JSON.parse(raw) as Partial<LockOwner>;
if (
typeof parsed.pid === "number" &&
typeof parsed.createdAt === "number"
) {
return {
pid: parsed.pid,
createdAt: parsed.createdAt,
token: typeof parsed.token === "string" ? parsed.token : undefined,
};
}
} catch {
// The lock may have been created but not fully written yet.
}
return null;
}
async function isLockStale(
lockDir: string,
staleAfterMs: number,
): Promise<boolean> {
const owner = await readLockOwner(lockDir);
if (owner) {
return !isProcessRunning(owner.pid) || Date.now() - owner.createdAt > staleAfterMs;
}
try {
const info = await stat(lockDir);
return Date.now() - info.mtimeMs > staleAfterMs;
} catch {
return false;
}
}
export async function acquireCrossProcessLock(
lockDir: string,
options: CrossProcessLockOptions = {},
): Promise<() => Promise<void>> {
const acquireTimeoutMs = options.acquireTimeoutMs ?? 120_000;
const retryIntervalMs = options.retryIntervalMs ?? 100;
const staleAfterMs = options.staleAfterMs ?? 120_000;
const deadline = Date.now() + acquireTimeoutMs;
await mkdir(dirname(lockDir), { recursive: true });
while (true) {
try {
await mkdir(lockDir);
const token = randomUUID();
const owner: LockOwner = { pid: process.pid, createdAt: Date.now(), token };
await writeFile(`${lockDir}/owner.json`, JSON.stringify(owner), "utf8");
let released = false;
return async () => {
if (released) return;
released = true;
const currentOwner = await readLockOwner(lockDir);
if (currentOwner?.token === token) {
await rm(lockDir, { recursive: true, force: true });
}
};
} catch (error) {
const code = (error as NodeJS.ErrnoException).code;
if (code !== "EEXIST") {
throw error;
}
if (await isLockStale(lockDir, staleAfterMs)) {
options.log?.(`Removing stale lock at ${lockDir}`);
await rm(lockDir, { recursive: true, force: true });
continue;
}
if (Date.now() >= deadline) {
throw new Error(`Timed out waiting to acquire lock ${lockDir}`);
}
await delay(retryIntervalMs);
}
}
}
export async function withCrossProcessLock<T>(
lockDir: string,
fn: () => Promise<T>,
options: CrossProcessLockOptions = {},
): Promise<T> {
const release = await acquireCrossProcessLock(lockDir, options);
try {
return await fn();
} finally {
await release();
}
}