fix(jobs): 无 Redis 时用数据库租约保证「一个 tick 只跑一个副本」

apps/api/src/infra.ts

@@ -17,7 +17,8 @@ import { env } from './config/env.ts';
 import { useRedisState } from './state/index.ts';
 import { useRedisCache } from './cache/index.ts';
 import { configureOutboxEventBus } from './plugins/events.ts';
-import { initJobs } from './jobs/index.ts';
+import { DbJobLease, initJobs } from './jobs/index.ts';
+import type { LeaseDb } from './jobs/lease.ts';
 import { getPrisma } from './plugins/prisma.ts';
 import { resolveStorageDriver, type StorageExtensionRuntime } from './storage/provider-router.ts';
 
@@ -127,6 +128,8 @@ export async function initInfra(): Promise<void> {
   // start consuming. Kernel sweeps were registered synchronously in buildApp().
   await initJobs({
     redisUrl: getInfraConfig().redisUrl ?? null,
+    // 进程内后端靠数据库租约保证「一个 tick 只有一个副本执行」；BullMQ 下不用。
+    lease: new DbJobLease(getPrisma() as unknown as LeaseDb),
     logger: {
       warn: (message) => console.warn(message),
       info: (message) => console.info(message),

apps/api/src/jobs/index.ts

@@ -12,15 +12,23 @@ import type { JobContext } from '@stackpanel/sdk';
 import { redisConnectionOptions } from './redis-connection.ts';
 import { getJobRuntime, KERNEL_JOBS } from './kernel-jobs.ts';
 import { BullMqJobBackend, KERNEL_QUEUE_NAME } from './bullmq-backend.ts';
+import type { JobLease } from './lease.ts';
 import { MemoryJobBackend } from './memory-backend.ts';
 
 export { createKernelJobContext, getJobRuntime, KERNEL_JOBS, qualify } from './kernel-jobs.ts';
 export type { JobBackend, JobRuntime, RegisteredSchedule } from './types.ts';
+export { DbJobLease, leaseTtlMs, resolveLeaseHolder } from './lease.ts';
+export type { JobLease } from './lease.ts';
 export { KERNEL_QUEUE_NAME } from './bullmq-backend.ts';
 
 export interface InitJobsOptions {
   /** Configured Redis URL; null in development means the in-process fallback. */
   redisUrl: string | null;
+  /**
+   * 进程内后端的跨副本租约。BullMQ 下不需要（队列本身按 jobId 抢占）。
+   * 缺省时进程内后端退化为「每个副本各跑一份」。
+   */
+  lease?: JobLease;
   logger: { info: (m: string) => void; warn: (m: string) => void };
 }
 
@@ -49,9 +57,16 @@ export async function initJobs(options: InitJobsOptions): Promise<void> {
     // `ok|unreachable|disabled` check map has no room for it), so do not
     // remove this log line without adding that signal elsewhere first.
     options.logger.warn(
-      '[jobs] 未配置 Redis：任务使用进程内实现，cron 计划任务将被忽略、且仅单副本正确。生产环境请配置 REDIS_URL。',
+      options.lease
+        ? '[jobs] 未配置 Redis：任务使用进程内实现，cron 计划任务将被忽略；重复执行由数据库租约抑制。生产环境仍建议配置 REDIS_URL。'
+        : '[jobs] 未配置 Redis：任务使用进程内实现，cron 计划任务将被忽略、且仅单副本正确（无租约）。生产环境请配置 REDIS_URL。',
+    );
+    await runtime.useBackend(
+      new MemoryJobBackend({
+        logger: options.logger,
+        ...(options.lease ? { guard: options.lease } : {}),
+      }),
     );
-    await runtime.useBackend(new MemoryJobBackend({ logger: options.logger }));
   }
   await runtime.start();
 }

apps/api/src/jobs/lease.ts

@@ -0,0 +1,68 @@
+/**
+ * 跨副本任务租约 —— 无 Redis 时的单实例保证（对标 #12.5）。
+ *
+ * 为什么需要：`MemoryJobBackend` 每个副本各有自己的 `setInterval`，`N` 个副本
+ * 就是 `N` 份定时器，同一 tick 会执行 `N` 次。开发环境无所谓，集群部署下
+ * 「到期暂停」「回收站清理」这类动作被重复执行就是真问题（对标对象用配置表锁
+ * + 5 分钟陈旧期解决同一件事）。
+ *
+ * 做法：一个 tick 开始前先抢租约（`INSERT ... ON CONFLICT ... WHERE 已过期`），
+ * 抢到才执行。租约靠 `expiresAtMs` 自然过期，不依赖清理任务 —— 持有者被 kill
+ * 之后，最迟 `ttlMs` 就有别的副本接手，不需要「检测死节点」那套。
+ *
+ * 只给进程内后端用：BullMQ 本身按队列抢占，再套一层租约只会让同一 job 的
+ * 重试互相挡住。
+ */
+import { hostname } from 'node:os';
+import { randomUUID } from 'node:crypto';
+
+/** 一个 tick 的执行许可。 */
+export interface JobLease {
+  /** 抢 `${name}` 的租约；返回 `true` 表示本次由本副本执行。 */
+  tryAcquire(name: string, ttlMs: number): Promise<boolean>;
+}
+
+/** 租约只需要 `$queryRaw`，测试里可以给一个假实现。 */
+export interface LeaseDb {
+  $queryRaw<T = unknown>(query: TemplateStringsArray, ...values: unknown[]): Promise<T>;
+}
+
+/**
+ * 租约时长：等于任务间隔（下限 1s、上限 5 分钟）。
+ *
+ * 不能取「间隔的若干倍」：那会把有效频率压到几倍间隔以下 —— 抢到租约的副本
+ * 每轮都成功，其余副本在整个租约窗口内全部跳过。取一个间隔，语义是
+ * 「同一轮最多一个副本执行」，下一轮到谁抢到就谁跑。
+ */
+export function leaseTtlMs(everyMs: number): number {
+  return Math.max(1_000, Math.min(everyMs, 5 * 60_000));
+}
+
+/** 持有者标识：仅用于排查「上一次是谁在跑」，不参与判定。 */
+export function resolveLeaseHolder(): string {
+  return `${hostname()}:${process.pid}:${randomUUID().slice(0, 8)}`;
+}
+
+/** 基于数据库的租约。表结构见 `kernel_job_leases` 迁移。 */
+export class DbJobLease implements JobLease {
+  constructor(
+    private readonly db: LeaseDb,
+    private readonly holder: string = resolveLeaseHolder(),
+  ) {}
+
+  async tryAcquire(name: string, ttlMs: number): Promise<boolean> {
+    const now = Date.now();
+    // 单条语句完成「插入或抢占已过期租约」，不需要事务：
+    // 并发时 Postgres 行锁保证只有一个 UPDATE 命中 `expiresAtMs < now`。
+    const rows = await this.db.$queryRaw<Array<{ name: string }>>`
+      INSERT INTO "kernel_job_leases" ("name", "holder", "expiresAtMs")
+      VALUES (${name}, ${this.holder}, ${BigInt(now + ttlMs)})
+      ON CONFLICT ("name") DO UPDATE
+        SET "holder" = EXCLUDED."holder",
+            "expiresAtMs" = EXCLUDED."expiresAtMs"
+        WHERE "kernel_job_leases"."expiresAtMs" < ${BigInt(now)}
+      RETURNING "name"
+    `;
+    return rows.length > 0;
+  }
+}

apps/api/src/jobs/memory-backend.ts

@@ -1,12 +1,16 @@
 /**
  * In-process job backend for development and tests without Redis (ADR-0013
  * fallback). It preserves the *contract* — namespaced handlers, recurring jobs,
- * bounded retries — but not the cluster guarantee: timers run in whichever
- * process created them. Production requires Redis, so this is never the
- * multi-replica path.
+ * bounded retries — and, when a {@link JobLease} is supplied, the cluster
+ * guarantee too: every replica fires its own timer, but only the replica that
+ * wins the lease actually runs the tick (对标 #12.5).
+ *
+ * Without a lease (`guard` omitted) timers run in whichever process created
+ * them — fine for tests and single-replica development, not for a cluster.
  */
 import { randomUUID } from 'node:crypto';
 import type { JobHandler, JobOptions } from '@stackpanel/sdk';
+import { leaseTtlMs, type JobLease } from './lease.ts';
 import {
   DEFAULT_ATTEMPTS,
   DEFAULT_BACKOFF_MS,
@@ -16,6 +20,11 @@ import {
 
 export interface MemoryJobBackendOptions {
   logger: { warn: (m: string) => void };
+  /**
+   * 跨副本租约。给了就「一个 tick 只有一个副本执行」；不给则各副本各跑一份
+   * （测试与单副本开发用，集群下会重复执行）。
+   */
+  guard?: JobLease;
 }
 
 export class MemoryJobBackend implements JobBackend {
@@ -47,7 +56,7 @@ export class MemoryJobBackend implements JobBackend {
     }
     await this.unschedule(entry.fullName);
     const timer = setInterval(() => {
-      void this.run(entry.fullName, entry.payload, entry.options);
+      void this.tick(entry);
     }, entry.schedule.everyMs);
     timer.unref?.();
     this.schedules.set(entry.fullName, { timer, entry });
@@ -103,6 +112,30 @@ export class MemoryJobBackend implements JobBackend {
     this.timers.clear();
   }
 
+  /**
+   * 一个定时器 tick：先抢租约，抢到才执行。
+   *
+   * 租约出错按「没抢到」处理 —— 宁可这一轮不跑（下一轮会补），也不能因为
+   * 数据库抖一下就变成多副本重复执行。
+   */
+  private async tick(entry: RegisteredSchedule): Promise<void> {
+    const guard = this.options.guard;
+    if (guard) {
+      let acquired: boolean;
+      try {
+        const everyMs = 'everyMs' in entry.schedule ? entry.schedule.everyMs : 60_000;
+        acquired = await guard.tryAcquire(entry.fullName, leaseTtlMs(everyMs));
+      } catch (error) {
+        this.options.logger.warn(
+          `[jobs] 租约获取失败，跳过本轮 ${entry.fullName}：${String(error)}`,
+        );
+        return;
+      }
+      if (!acquired) return;
+    }
+    await this.run(entry.fullName, entry.payload, entry.options);
+  }
+
   /** Attempt `fullName` with bounded retries. Never throws to the timer. */
   private async run(
     fullName: string,

apps/api/tests/integration/job-lease.test.ts

@@ -0,0 +1,77 @@
+import { randomUUID } from 'node:crypto';
+import { afterAll, describe, expect, it } from 'vitest';
+import { closeRedis } from '@stackpanel/db';
+import { DbJobLease } from '../../src/jobs/lease.ts';
+import { getPrisma } from '../../src/plugins/prisma.ts';
+
+/**
+ * 对标 #12.5：跨副本单实例保证。
+ *
+ * 真实场景：两个副本同时到点，各自 `setInterval` 都触发了。谁执行由租约决定 ——
+ * 只有一个副本的 `INSERT ... ON CONFLICT ... WHERE 已过期` 能命中，另一个拿到空
+ * 结果集并跳过。租约过期后由下一轮接手，不需要死节点检测。
+ */
+const dbAvailable = await (async (): Promise<boolean> => {
+  try {
+    await getPrisma().$queryRaw`SELECT 1`;
+    return true;
+  } catch {
+    return false;
+  }
+})();
+
+const prefix = `test.lease.${randomUUID().slice(0, 8)}`;
+const created: string[] = [];
+
+function leaseJob(suffix: string): string {
+  // 每个用例必须有独立任务名：同名任务配独立租约本身就是「稳态」，会让后一个
+  // 用例被前一个用例的未过期租约挡住，测出来的失败是假的。
+  const name = `${prefix}.${suffix}`;
+  created.push(name);
+  return name;
+}
+
+afterAll(async () => {
+  if (created.length > 0) {
+    await getPrisma().$executeRaw`DELETE FROM "kernel_job_leases" WHERE "name" = ANY(${created})`;
+  }
+  await closeRedis();
+});
+
+describe.skipIf(!dbAvailable)('DbJobLease', () => {
+  it('两个副本抢同一个任务，只有一个抢到', async () => {
+    const name = leaseJob('contend');
+    const a = new DbJobLease(getPrisma(), 'replica-a');
+    const b = new DbJobLease(getPrisma(), 'replica-b');
+
+    expect(await a.tryAcquire(name, 5_000)).toBe(true);
+    expect(await b.tryAcquire(name, 5_000)).toBe(false);
+  });
+
+  it('租约过期后另一个副本可以接手（持有者崩溃不需要清理）', async () => {
+    const name = leaseJob('expire');
+    const a = new DbJobLease(getPrisma(), 'replica-a');
+    const b = new DbJobLease(getPrisma(), 'replica-b');
+
+    expect(await a.tryAcquire(name, 30)).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 60));
+    expect(await b.tryAcquire(name, 5_000)).toBe(true);
+  });
+
+  it('不同任务互不阻塞', async () => {
+    const first = leaseJob('independent-a');
+    const second = leaseJob('independent-b');
+    const lease = new DbJobLease(getPrisma(), 'replica-a');
+
+    expect(await lease.tryAcquire(first, 5_000)).toBe(true);
+    expect(await lease.tryAcquire(second, 5_000)).toBe(true);
+  });
+
+  it('同一副本连续抢同一任务：租约内第二次被自己挡住（不会自我放大频率）', async () => {
+    const name = leaseJob('self');
+    const lease = new DbJobLease(getPrisma(), 'replica-a');
+
+    expect(await lease.tryAcquire(name, 5_000)).toBe(true);
+    expect(await lease.tryAcquire(name, 5_000)).toBe(false);
+  });
+});

apps/api/tests/unit/job-lease-memory.test.ts

@@ -0,0 +1,91 @@
+import { describe, expect, it, vi } from 'vitest';
+import { leaseTtlMs } from '../../src/jobs/lease.ts';
+import { MemoryJobBackend } from '../../src/jobs/memory-backend.ts';
+
+/**
+ * 对标 #12.5：进程内后端在集群下会「每个副本各跑一份」。租约把「谁执行」这件事
+ * 从「每个定时器都跑」改成「抢到租约的那个跑」，其余副本安静跳过。
+ *
+ * 这里只锁行为契约；数据库侧的抢占语义在 `tests/integration/job-lease.test.ts`。
+ */
+function waitFor(predicate: () => boolean, timeoutMs = 2_000): Promise<void> {
+  return new Promise((resolve, reject) => {
+    const started = Date.now();
+    const check = (): void => {
+      if (predicate()) return resolve();
+      if (Date.now() - started > timeoutMs) return reject(new Error('waitFor timed out'));
+      setTimeout(check, 10);
+    };
+    check();
+  });
+}
+
+describe('MemoryJobBackend 租约', () => {
+  it('抢到租约才执行，没抢到静默跳过', async () => {
+    const ran: string[] = [];
+    const backend = new MemoryJobBackend({
+      logger: { warn: () => undefined },
+      guard: {
+        // 第一轮给，之后不给：模拟另一个副本先抢到。
+        tryAcquire: vi.fn(async () => ran.length === 0),
+      },
+    });
+    backend.handle('k.sweep', () => {
+      ran.push('tick');
+    });
+    await backend.schedule({ fullName: 'k.sweep', schedule: { everyMs: 20 }, payload: undefined });
+
+    await waitFor(() => ran.length >= 1);
+    await new Promise((resolve) => setTimeout(resolve, 120));
+    await backend.stop();
+
+    // 只跑了一次：后续 tick 全部被租约挡掉，不是「每 20ms 跑一次」。
+    expect(ran).toEqual(['tick']);
+  });
+
+  it('租约出错按「没抢到」处理：跳过本轮，不执行、不抛给定时器', async () => {
+    const ran: string[] = [];
+    const warned: string[] = [];
+    const backend = new MemoryJobBackend({
+      logger: { warn: (m) => warned.push(m) },
+      guard: {
+        tryAcquire: async () => {
+          throw new Error('connection reset');
+        },
+      },
+    });
+    backend.handle('k.sweep', () => {
+      ran.push('tick');
+    });
+    await backend.schedule({ fullName: 'k.sweep', schedule: { everyMs: 20 }, payload: undefined });
+
+    await waitFor(() => warned.length >= 1);
+    await backend.stop();
+
+    expect(ran).toEqual([]);
+    // 宁可这一轮不跑（下一轮补），也不能因数据库抖一下变成重复执行。
+    expect(warned[0]).toContain('connection reset');
+  });
+
+  it('不给租约时保持原行为（单副本开发）', async () => {
+    const ran: string[] = [];
+    const backend = new MemoryJobBackend({ logger: { warn: () => undefined } });
+    backend.handle('k.sweep', () => {
+      ran.push('tick');
+    });
+    await backend.schedule({ fullName: 'k.sweep', schedule: { everyMs: 20 }, payload: undefined });
+
+    await waitFor(() => ran.length >= 2);
+    await backend.stop();
+    expect(ran.length).toBeGreaterThanOrEqual(2);
+  });
+
+  it('租约时长取一个间隔：不把有效频率压到几倍间隔以下', () => {
+    expect(leaseTtlMs(2_000)).toBe(2_000);
+    expect(leaseTtlMs(60_000)).toBe(60_000);
+    // 下限 1s：2s 一轮的任务不该拿到 20ms 的租约（会与下一轮重叠）。
+    expect(leaseTtlMs(20)).toBe(1_000);
+    // 上限 5 分钟：崩溃后最多 5 分钟就有别的副本接手。
+    expect(leaseTtlMs(24 * 60 * 60_000)).toBe(5 * 60_000);
+  });
+});

packages/db/prisma/migrations/20261027120000_kernel_job_lease/migration.sql

@@ -0,0 +1,14 @@
+-- 对标 #12.5：无 Redis 时的跨副本单实例保证。
+--
+-- 进程内任务后端的每个副本各有一份定时器，同一 tick 会执行 N 次。执行前抢
+-- 一次租约（这里的表就是租约本身），抢到才干活；租约按 expiresAtMs 自然过期，
+-- 持有者崩溃后由下一个副本接手，不需要死节点检测。
+--
+-- 有 Redis 时走 BullMQ 抢占，本表不参与。
+CREATE TABLE "kernel_job_leases" (
+    "name" TEXT NOT NULL,
+    "holder" TEXT NOT NULL,
+    "expiresAtMs" BIGINT NOT NULL,
+
+    CONSTRAINT "kernel_job_leases_pkey" PRIMARY KEY ("name")
+);