缩短网关崩溃后的锁接管时间

Author: javazys

docs/architecture.md

@@ -48,7 +48,7 @@ Dashboard ─────┘       ↑                 ↑
 Gateway 启动时按以下顺序工作：
 
 1. 初始化数据库，自动执行 `drizzle/` migrations，开启外键并运行 `PRAGMA foreign_key_check`；有孤立记录时直接拒绝启动。
-2. 用 `BEGIN IMMEDIATE` 更新 `gateway_lock`。锁每 10 秒心跳，30 秒未更新才允许新实例接管；此时 `ready_at` 仍为空。
+2. 用 `BEGIN IMMEDIATE` 更新 `gateway_lock`。锁每 10 秒心跳；记录的 PID 已不存在时可立即接管，否则要等心跳 30 秒未更新；此时 `ready_at` 仍为空。
 3. 加载并校验配置；非法配置直接失败，不静默回退。
 4. 启动 Worker、Watchdog、Scheduler，最后按配置启动内嵌 Dashboard；全部成功后才写入 `ready_at`。
 5. 收到 `SIGINT` 或 `SIGTERM` 时先停止接单，等待 `worker.shutdownGracePeriodMs`；宽限期内完成的任务正常落库，只有剩余子进程才会终止、重置为 `pending`，对应执行记录标为失败。

src/core/process-control.ts

@@ -10,6 +10,16 @@ function isSafePid(pid: number): boolean {
     return Number.isInteger(pid) && pid > 1 && pid !== process.pid;
 }
 
+export function isProcessAlive(pid: number): boolean {
+    if (!Number.isInteger(pid) || pid <= 0) return false;
+    try {
+        process.kill(pid, 0);
+        return true;
+    } catch (error) {
+        return error instanceof Error && 'code' in error && error.code === 'EPERM';
+    }
+}
+
 function inspectUnixProcess(pid: number): ProcessInfo | null {
     const result = spawnSync('ps', ['-o', 'pgid=', '-o', 'command=', '-p', String(pid)], {
         encoding: 'utf8',

src/gateway/index.ts

@@ -7,12 +7,13 @@ import { closeDb } from '@core/db';
 import { TaskService } from '@core/services/task.service';
 import { TaskRunService } from '@core/services/task-run.service';
 import { initializeGatewayHealth, resetGatewayHealth } from './health';
+import { isProcessAlive } from '@core/process-control';
 
 // gateway_lock.heartbeat_at / acquired_at 单位：毫秒（Date.now()）
 // 超过此阈值未心跳则视为锁持有者已死亡
 const STALE_THRESHOLD_MS = 30_000;
 
-function acquireLock(): boolean {
+export function acquireLock(): boolean {
     const now = Date.now();
     const pid = process.pid;
 
@@ -26,7 +27,8 @@ function acquireLock(): boolean {
         } | undefined;
 
         if (existing) {
-            if (now - existing.heartbeat_at < STALE_THRESHOLD_MS) {
+            const lockHolderAlive = existing.pid !== pid && isProcessAlive(existing.pid);
+            if (now - existing.heartbeat_at < STALE_THRESHOLD_MS && lockHolderAlive) {
                 sqlite.exec('ROLLBACK');
                 console.error(JSON.stringify({
                     ts: new Date().toISOString(),
@@ -58,7 +60,7 @@ function acquireLock(): boolean {
     }
 }
 
-function releaseLock() {
+export function releaseLock() {
     try {
         sqlite.exec('DELETE FROM gateway_lock WHERE pid = ?', [process.pid]);
     } catch {}

tests/gateway-lock.test.ts

@@ -0,0 +1,21 @@
+import { describe, expect, test } from 'bun:test';
+import { setupTestDb } from './helpers/mock-db';
+
+describe('Gateway 单实例锁', () => {
+    test('新鲜锁记录的 PID 已不存在时立即接管', async () => {
+        setupTestDb();
+        const { sqlite } = await import('../src/core/db');
+        const missingPid = 2_147_483_647;
+        sqlite.exec(
+            'INSERT INTO gateway_lock (id, pid, acquired_at, heartbeat_at, ready_at) VALUES (1, ?, ?, ?, ?)',
+            [missingPid, Date.now(), Date.now(), Date.now()],
+        );
+        const { acquireLock, releaseLock } = await import('../src/gateway/index');
+
+        expect(acquireLock()).toBe(true);
+        expect(sqlite.prepare('SELECT pid FROM gateway_lock WHERE id = 1').get()).toEqual({
+            pid: process.pid,
+        });
+        releaseLock();
+    });
+});

tests/helpers/mock-db.ts

@@ -96,6 +96,7 @@ export function setupTestDb() {
 
     mock.module('../../src/core/db', () => ({
         db: testDb,
+        sqlite,
         schema,
         getDb: () => testDb,
         getSqlite: () => sqlite,

tests/process-control.test.ts

@@ -3,7 +3,7 @@ import { chmodSync, existsSync, mkdtempSync, readFileSync, rmSync, writeFileSync
 import { basename, join } from 'path';
 import { tmpdir } from 'os';
 import { spawn, type ChildProcess } from 'child_process';
-import { signalRecordedProcessTree, signalSpawnedProcessTree } from '../src/core/process-control';
+import { isProcessAlive, signalRecordedProcessTree, signalSpawnedProcessTree } from '../src/core/process-control';
 
 const dirs: string[] = [];
 const children: ChildProcess[] = [];
@@ -57,6 +57,11 @@ setInterval(() => {}, 1000);
 }
 
 describe('进程树终止', () => {
+    test('能区分当前进程与不存在的 PID', () => {
+        expect(isProcessAlive(process.pid)).toBe(true);
+        expect(isProcessAlive(2_147_483_647)).toBe(false);
+    });
+
     test('命令不匹配时拒绝终止记录的 PID', async () => {
         const tree = createProcessTree();
         await waitForFile(tree.childPidFile);