From cdfde85595454a90a6eb9fb98a55b82bb2c75693 Mon Sep 17 00:00:00 2001 From: redshift <213178690+1ftredsh@users.noreply.github.com> Date: Thu, 30 Apr 2026 13:36:16 +0530 Subject: [PATCH] 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. --- src/daemon/wallet/cocod-client.ts | 28 ++++-- src/start-daemon.ts | 82 +++++++++++------- src/utils/daemon-client.ts | 32 ++----- src/utils/process-lock.ts | 136 ++++++++++++++++++++++++++++++ 4 files changed, 216 insertions(+), 62 deletions(-) create mode 100644 src/utils/process-lock.ts diff --git a/src/daemon/wallet/cocod-client.ts b/src/daemon/wallet/cocod-client.ts index 665d599..1d85568 100644 --- a/src/daemon/wallet/cocod-client.ts +++ b/src/daemon/wallet/cocod-client.ts @@ -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 { 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; }); } diff --git a/src/start-daemon.ts b/src/start-daemon.ts index e9eebb1..4c51b00 100644 --- a/src/start-daemon.ts +++ b/src/start-daemon.ts @@ -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 { + 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 { 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 { + 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), + }, + ); +} diff --git a/src/utils/daemon-client.ts b/src/utils/daemon-client.ts index 666301b..ba3fda2 100644 --- a/src/utils/daemon-client.ts +++ b/src/utils/daemon-client.ts @@ -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 { - // 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 { diff --git a/src/utils/process-lock.ts b/src/utils/process-lock.ts new file mode 100644 index 0000000..798f1cf --- /dev/null +++ b/src/utils/process-lock.ts @@ -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 { + 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 { + try { + const raw = await readFile(`${lockDir}/owner.json`, "utf8"); + const parsed = JSON.parse(raw) as Partial; + 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 { + 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> { + 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( + lockDir: string, + fn: () => Promise, + options: CrossProcessLockOptions = {}, +): Promise { + const release = await acquireCrossProcessLock(lockDir, options); + try { + return await fn(); + } finally { + await release(); + } +}