@mearl/daemon-core - npm Package Compare versions

Comparing version 2.9.5 to 2.9.6

dist/daemon.d.ts

@@ -13,3 +13,3 @@ import { type DaemonRecord } from './registry.js';
     scriptPath: string;
-    /** Path to this instance's JSON record file (written by the foreground child). */
+    /** Path to this instance's JSON record file. */
     configFile: string;
@@ -24,6 +24,8 @@ /** Path to this instance's log file (the detached child's stdout/stderr). */
     formatInfo?: (record: DaemonRecord) => string | null;
+    /** Record persisted immediately after spawn so concurrent starts can reuse the child. */
+    startingRecord?: (pid: number, startedAt: string) => DaemonRecord;
 }
 export interface Daemon {
     start(): Promise<void>;
-    stop(): Promise<void>;
+    stop(expectedPid?: number): Promise<void>;
     restart(): Promise<void>;
@@ -38,8 +40,7 @@ status(): void;
  *
- * The model: `start()` spawns the consumer's CLI detached with `--foreground`;
- * that child runs the actual service and, once ready, writes its own pid into
- * `configFile`. The parent polls for `record.pid === child.pid` to confirm the
- * child came up (vs. died or hung). Multi-instance subsystems create one manager
- * per instance, deriving `configFile`/`logFile` from the instance key.
+ * The model: `start()` atomically reserves startup, spawns the consumer's CLI
+ * detached with `--foreground`, and records the child pid as `starting`. The
+ * child replaces that marker when ready. Multi-instance subsystems create one
+ * manager per instance, deriving `configFile` / `logFile` from the instance key.
  */
 export declare function createDaemon(opts: CreateDaemonOptions): Daemon;

dist/daemon.js

 import { spawn } from 'node:child_process';
-import { openSync } from 'node:fs';
+import { closeSync, openSync } from 'node:fs';
 import { dirname } from 'node:path';
 import { isAlive } from './process.js';
-import { ensureDir, readRecord, removeRecord } from './registry.js';
+import { claimRecord, ensureDir, readRecord, removeInvalidRecord, removeRecordIfPid, writeRecord, } from './registry.js';
 import { rotateLogIfNeeded, tailFile } from './log.js';
@@ -13,11 +13,11 @@ function sleep(ms) {
  *
- * The model: `start()` spawns the consumer's CLI detached with `--foreground`;
- * that child runs the actual service and, once ready, writes its own pid into
- * `configFile`. The parent polls for `record.pid === child.pid` to confirm the
- * child came up (vs. died or hung). Multi-instance subsystems create one manager
- * per instance, deriving `configFile`/`logFile` from the instance key.
+ * The model: `start()` atomically reserves startup, spawns the consumer's CLI
+ * detached with `--foreground`, and records the child pid as `starting`. The
+ * child replaces that marker when ready. Multi-instance subsystems create one
+ * manager per instance, deriving `configFile` / `logFile` from the instance key.
  */
 export function createDaemon(opts) {
-    const { name, scriptPath, configFile, logFile, foregroundArgs = [], readyTimeoutMs = 15_000, formatInfo, } = opts;
+    const { name, scriptPath, configFile, logFile, foregroundArgs = [], readyTimeoutMs = 15_000, formatInfo, startingRecord, } = opts;
     const tag = `[${name}]`;
+    const startLockFile = `${configFile}.start.lock`;
     function running() {
@@ -35,19 +35,79 @@ const rec = readRecord(configFile);
         if (live) {
-            console.log(`${tag} Already running (pid ${live.pid})`);
+            const state = live.daemonState === 'starting' ? 'starting' : 'running';
+            console.log(`${tag} Already ${state} (pid ${live.pid})`);
             printInfo(live);
             return;
         }
-        // Clear any stale record left by a crashed run before spawning a new child.
-        removeRecord(configFile);
+        const existingLock = readRecord(startLockFile);
+        if (existingLock && isAlive(existingLock.pid)) {
+            console.log(`${tag} Start already in progress (pid ${existingLock.pid})`);
+            return;
+        }
+        if (existingLock)
+            removeRecordIfPid(startLockFile, existingLock.pid);
+        else
+            removeInvalidRecord(startLockFile);
+        const lockPid = process.pid;
+        if (!claimRecord(startLockFile, { pid: lockPid, startedAt: new Date().toISOString() })) {
+            const competingLock = readRecord(startLockFile);
+            if (competingLock && isAlive(competingLock.pid)) {
+                console.log(`${tag} Start already in progress (pid ${competingLock.pid})`);
+                return;
+            }
+            if (competingLock)
+                removeRecordIfPid(startLockFile, competingLock.pid);
+            else
+                removeInvalidRecord(startLockFile);
+            if (!claimRecord(startLockFile, { pid: lockPid, startedAt: new Date().toISOString() })) {
+                const currentLock = readRecord(startLockFile);
+                if (currentLock && isAlive(currentLock.pid)) {
+                    console.log(`${tag} Start already in progress (pid ${currentLock.pid})`);
+                    return;
+                }
+                console.error(`${tag} Failed to acquire startup lock: ${startLockFile}`);
+                process.exitCode = 1;
+                return;
+            }
+        }
+        // Clear any stale record left by a crashed run after taking the startup lock.
+        const stale = readRecord(configFile);
+        if (stale && !isAlive(stale.pid))
+            removeRecordIfPid(configFile, stale.pid);
+        else if (!stale)
+            removeInvalidRecord(configFile);
         ensureDir(dirname(configFile));
         ensureDir(dirname(logFile));
         rotateLogIfNeeded(logFile);
+        const startedAt = new Date().toISOString();
         const logFd = openSync(logFile, 'a');
-        const child = spawn(process.execPath, [scriptPath, '--foreground', ...foregroundArgs], {
-            detached: true,
-            stdio: ['ignore', logFd, logFd],
-            env: process.env,
-            windowsHide: true,
+        let child;
+        try {
+            child = spawn(process.execPath, [scriptPath, '--foreground', ...foregroundArgs], {
+                detached: true,
+                stdio: ['ignore', logFd, logFd],
+                env: process.env,
+                windowsHide: true,
+            });
+        }
+        catch (error) {
+            removeRecordIfPid(startLockFile, lockPid);
+            throw error;
+        }
+        finally {
+            closeSync(logFd);
+        }
+        child.unref();
+        const childPid = child.pid;
+        if (!childPid) {
+            removeRecordIfPid(startLockFile, lockPid);
+            console.error(`${tag} Failed to start: spawned process has no pid`);
+            process.exitCode = 1;
+            return;
+        }
+        writeRecord(startLockFile, { pid: childPid, startedAt });
+        claimRecord(configFile, {
+            ...(startingRecord?.(childPid, startedAt) ?? { pid: childPid, startedAt }),
+            daemonState: 'starting',
         });
-        child.unref();
+        removeRecordIfPid(startLockFile, childPid);
         // Wait until the child writes its own pid into the record (i.e. it is ready),
@@ -57,3 +117,4 @@ // or until it dies / we time out.
         while (Date.now() < deadline) {
-            if (!isAlive(child.pid)) {
+            if (!isAlive(childPid)) {
+                removeRecordIfPid(configFile, childPid);
                 console.error(`${tag} Failed to start. Recent log:`);
@@ -68,4 +129,4 @@ const log = tailFile(logFile, 20);
             const rec = readRecord(configFile);
-            if (rec && rec.pid === child.pid) {
-                console.log(`${tag} Started in background (pid ${child.pid})`);
+            if (rec && rec.pid === childPid && rec.daemonState !== 'starting') {
+                console.log(`${tag} Started in background (pid ${childPid})`);
                 console.log(`${tag} Logs: ${logFile}`);
@@ -80,7 +141,14 @@ printInfo(rec);
     }
-    async function stop() {
+    async function stop(expectedPid) {
         const rec = readRecord(configFile);
+        if (expectedPid !== undefined && rec?.pid !== expectedPid) {
+            console.log(`${tag} Registered pid ${expectedPid} is no longer the current instance`);
+            return;
+        }
         if (!rec || !isAlive(rec.pid)) {
             console.log(`${tag} Not running`);
-            removeRecord(configFile);
+            if (rec)
+                removeRecordIfPid(configFile, rec.pid);
+            else
+                removeInvalidRecord(configFile);
             return;
@@ -109,3 +177,3 @@ }
         }
-        removeRecord(configFile);
+        removeRecordIfPid(configFile, pid);
         console.log(`${tag} Stopped (pid ${pid})`);
@@ -123,3 +191,3 @@ }
         }
-        console.log(`${tag} Status: running`);
+        console.log(`${tag} Status: ${rec.daemonState === 'starting' ? 'starting' : 'running'}`);
         console.log(`  pid:        ${rec.pid}`);
@@ -126,0 +194,0 @@ if (rec.startedAt)

dist/registry.d.ts

@@ -10,8 +10,16 @@ /** Base shape every daemon record shares; subsystems extend it with their own fields. */
  * Atomically write a record: serialise to a temp file then rename over the
- * target, so a concurrent reader never observes a half-written file. The temp
- * name is pid-scoped so two writers don't clobber each other's temp file.
+ * target, so a concurrent reader never observes a half-written file. Temp file
+ * names include a random token so independent writes cannot collide.
  */
 export declare function writeRecord(file: string, record: DaemonRecord): void;
+/** Atomically create a record only when the target does not already exist. */
+export declare function claimRecord(file: string, record: DaemonRecord): boolean;
 export declare function readRecord<T extends DaemonRecord = DaemonRecord>(file: string): T | null;
 export declare function removeRecord(file: string): void;
+/** Remove a record only while it still belongs to the expected process. */
+export declare function removeRecordIfPid(file: string, pid: number): boolean;
+/** Replace a record only while it is still owned by the expected process. */
+export declare function writeRecordIfPid(file: string, pid: number, record: DaemonRecord): boolean;
+/** Remove an existing record whose contents are not a valid daemon record. */
+export declare function removeInvalidRecord(file: string): boolean;
 /** Absolute paths of every `*.json` record file in a directory. */
@@ -18,0 +26,0 @@ export declare function listRecordFiles(dir: string): string[];

dist/registry.js

@@ -1,4 +0,9 @@
-import { existsSync, mkdirSync, readFileSync, writeFileSync, renameSync, rmSync, readdirSync, } from 'node:fs';
+import { closeSync, existsSync, linkSync, openSync, statSync, mkdirSync, readFileSync, writeFileSync, renameSync, rmSync, readdirSync, } from 'node:fs';
 import { dirname, join } from 'node:path';
+import { randomUUID } from 'node:crypto';
 import { isAlive } from './process.js';
+const LOCK_WAIT_MS = 10;
+const LOCK_TIMEOUT_MS = 5_000;
+const OWNER_WRITE_GRACE_MS = 1_000;
+const lockWaitBuffer = new Int32Array(new SharedArrayBuffer(Int32Array.BYTES_PER_ELEMENT));
 export function ensureDir(dir) {
@@ -9,11 +14,22 @@ if (!existsSync(dir)) {
 }
-/**
- * Atomically write a record: serialise to a temp file then rename over the
- * target, so a concurrent reader never observes a half-written file. The temp
- * name is pid-scoped so two writers don't clobber each other's temp file.
- */
-export function writeRecord(file, record) {
-    ensureDir(dirname(file));
-    const tmp = `${file}.${process.pid}.tmp`;
+function pauseForLock() {
+    Atomics.wait(lockWaitBuffer, 0, 0, LOCK_WAIT_MS);
+}
+function readRecordUnlocked(file) {
     try {
+        if (!existsSync(file))
+            return null;
+        const parsed = JSON.parse(readFileSync(file, 'utf-8'));
+        return parsed && typeof parsed.pid === 'number' ? parsed : null;
+    }
+    catch {
+        return null;
+    }
+}
+function removeRecordUnlocked(file) {
+    rmSync(file, { force: true });
+}
+function writeRecordUnlocked(file, record) {
+    const tmp = `${file}.${process.pid}.${randomUUID()}.tmp`;
+    try {
         writeFileSync(tmp, JSON.stringify(record), { mode: 0o600 });
@@ -32,16 +48,113 @@ renameSync(tmp, file);
 }
-export function readRecord(file) {
+function removeStaleLock(lockDir) {
+    const ownerFile = join(lockDir, 'owner.json');
+    const owner = readRecordUnlocked(ownerFile);
+    if (owner) {
+        if (isAlive(owner.pid))
+            return false;
+    }
+    else {
+        try {
+            if (Date.now() - statSync(lockDir).mtimeMs < OWNER_WRITE_GRACE_MS)
+                return false;
+        }
+        catch {
+            return true;
+        }
+    }
     try {
-        if (!existsSync(file))
-            return null;
-        const parsed = JSON.parse(readFileSync(file, 'utf-8'));
-        return parsed && typeof parsed.pid === 'number' ? parsed : null;
+        rmSync(lockDir, { recursive: true, force: true });
+        return true;
     }
     catch {
-        return null;
+        return false;
     }
 }
+/** Serialize record ownership changes across processes sharing the registry. */
+function withRecordLock(file, operation) {
+    ensureDir(dirname(file));
+    const lockDir = `${file}.registry.lock`;
+    const ownerFile = join(lockDir, 'owner.json');
+    const token = randomUUID();
+    const deadline = Date.now() + LOCK_TIMEOUT_MS;
+    while (true) {
+        try {
+            mkdirSync(lockDir, { mode: 0o700 });
+        }
+        catch (error) {
+            if (error.code !== 'EEXIST')
+                throw error;
+            if (!removeStaleLock(lockDir)) {
+                if (Date.now() >= deadline) {
+                    throw new Error(`Timed out waiting for registry lock: ${file}`, { cause: error });
+                }
+                pauseForLock();
+            }
+            continue;
+        }
+        let ownerFd = null;
+        try {
+            ownerFd = openSync(ownerFile, 'wx', 0o600);
+            writeFileSync(ownerFd, JSON.stringify({ pid: process.pid, token }));
+        }
+        catch (error) {
+            if (ownerFd !== null)
+                closeSync(ownerFd);
+            ownerFd = null;
+            if (error.code === 'EEXIST') {
+                if (Date.now() >= deadline) {
+                    throw new Error(`Timed out acquiring registry lock: ${file}`, { cause: error });
+                }
+                pauseForLock();
+                continue;
+            }
+            throw error;
+        }
+        finally {
+            if (ownerFd !== null)
+                closeSync(ownerFd);
+        }
+        try {
+            return operation();
+        }
+        finally {
+            const owner = readRecordUnlocked(ownerFile);
+            if (owner?.token === token)
+                rmSync(lockDir, { recursive: true, force: true });
+        }
+    }
+}
+/**
+ * Atomically write a record: serialise to a temp file then rename over the
+ * target, so a concurrent reader never observes a half-written file. Temp file
+ * names include a random token so independent writes cannot collide.
+ */
+export function writeRecord(file, record) {
+    withRecordLock(file, () => writeRecordUnlocked(file, record));
+}
+/** Atomically create a record only when the target does not already exist. */
+export function claimRecord(file, record) {
+    return withRecordLock(file, () => {
+        const candidate = `${file}.${process.pid}.${randomUUID()}.claim`;
+        try {
+            writeFileSync(candidate, JSON.stringify(record), { mode: 0o600 });
+            linkSync(candidate, file);
+            return true;
+        }
+        catch (error) {
+            if (error.code === 'EEXIST')
+                return false;
+            throw error;
+        }
+        finally {
+            rmSync(candidate, { force: true });
+        }
+    });
+}
+export function readRecord(file) {
+    return readRecordUnlocked(file);
+}
 export function removeRecord(file) {
     try {
-        rmSync(file, { force: true });
+        withRecordLock(file, () => removeRecordUnlocked(file));
     }
@@ -52,2 +165,41 @@ catch {
 }
+/** Remove a record only while it still belongs to the expected process. */
+export function removeRecordIfPid(file, pid) {
+    try {
+        return withRecordLock(file, () => {
+            const record = readRecordUnlocked(file);
+            if (record?.pid !== pid)
+                return false;
+            removeRecordUnlocked(file);
+            return true;
+        });
+    }
+    catch {
+        return false;
+    }
+}
+/** Replace a record only while it is still owned by the expected process. */
+export function writeRecordIfPid(file, pid, record) {
+    return withRecordLock(file, () => {
+        const current = readRecordUnlocked(file);
+        if (current?.pid !== pid)
+            return false;
+        writeRecordUnlocked(file, record);
+        return true;
+    });
+}
+/** Remove an existing record whose contents are not a valid daemon record. */
+export function removeInvalidRecord(file) {
+    try {
+        return withRecordLock(file, () => {
+            if (!existsSync(file) || readRecordUnlocked(file))
+                return false;
+            removeRecordUnlocked(file);
+            return true;
+        });
+    }
+    catch {
+        return false;
+    }
+}
 /** Absolute paths of every `*.json` record file in a directory. */
@@ -76,4 +228,7 @@ export function listRecordFiles(dir) {
         }
+        else if (rec) {
+            removeRecordIfPid(file, rec.pid);
+        }
         else {
-            removeRecord(file);
+            removeInvalidRecord(file);
         }
@@ -80,0 +235,0 @@ }

package.json

 {
   "name": "@mearl/daemon-core",
-  "version": "2.9.5",
+  "version": "2.9.6",
   "description": "Shared Node daemon lifecycle primitives for Mearl — process liveness, JSON record registry, log rotation, and a detached daemon manager",
@@ -5,0 +5,0 @@ "type": "module",

README.md

@@ -37,5 +37,9 @@ # @mearl/daemon-core
 ensureDir(dir: string): void
-writeRecord(file: string, record: DaemonRecord): void   // 临时文件 + rename 原子写
+writeRecord(file: string, record: DaemonRecord): void   // 跨进程加锁，临时文件 + rename 原子写
+claimRecord(file: string, record: DaemonRecord): boolean // 仅在文件不存在时原子占用
 readRecord<T>(file: string): T | null
 removeRecord(file: string): void
+removeRecordIfPid(file: string, pid: number): boolean   // 仅删除指定进程仍拥有的记录
+writeRecordIfPid(file: string, pid: number, record: DaemonRecord): boolean
+removeInvalidRecord(file: string): boolean              // 清理无法解析的残留记录
 listRecordFiles(dir: string): string[]                  // 目录下所有 *.json 绝对路径
@@ -51,3 +55,6 @@ readLiveRecords<T>(dir: string): T[]                    // 读取存活记录，并顺带清理死记录
 
-围绕一对 `(configFile, logFile)` 构建单实例守护进程管理器。模型：`start()` 以 `--foreground` detached 方式重新拉起消费方自己的 CLI；子进程运行实际服务，就绪后把自己的 pid 写入 `configFile`，父进程轮询 `record.pid === child.pid` 确认其已起来（而非死亡或卡住）。多实例子系统按实例 key 派生 `configFile`/`logFile`，为每个实例各创建一个管理器。
+围绕一对 `(configFile, logFile)` 构建单实例守护进程管理器。`start()` 先原子占用启动锁，
+再以 `--foreground` detached 方式拉起消费方 CLI，并立即写入带子进程 pid 的 `starting`
+记录。子进程就绪后覆盖该记录，父进程据此确认启动完成。并发 `start()` 会复用正在启动或
+已经运行的子进程。多实例子系统按实例 key 派生各自的 `configFile` / `logFile`。
 
@@ -63,2 +70,3 @@ ```typescript
   formatInfo?: (record: DaemonRecord) => string | null;  // 起来后额外打印的信息
+  startingRecord?: (pid: number, startedAt: string) => DaemonRecord;
 }
@@ -68,3 +76,3 @@
   start(): Promise<void>;
-  stop(): Promise<void>;
+  stop(expectedPid?: number): Promise<void>; // 指定 PID 时仅停止仍匹配的实例
   restart(): Promise<void>;
@@ -71,0 +79,0 @@ status(): void;

New alerts

Filesystem access

Supply chain risk

Accesses the file system, and could potentially read sensitive data.

Fixed alerts

Filesystem access

Supply chain risk

Accesses the file system, and could potentially read sensitive data.

Improved metrics

Total package byte prevSize

29276

45.7%

Lines of code

623

59.34%

Number of lines in readme file

95

9.2%

No dependency changes