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(); + } +}