From 1acb164195a4dc6348f4728dbaf6a34daa803d48 Mon Sep 17 00:00:00 2001 From: worable Date: Sun, 25 Oct 2026 16:47:35 +0800 Subject: [PATCH] =?UTF-8?q?fix(jobs):=20=E6=97=A0=20Redis=20=E6=97=B6?= =?UTF-8?q?=E7=94=A8=E6=95=B0=E6=8D=AE=E5=BA=93=E7=A7=9F=E7=BA=A6=E4=BF=9D?= =?UTF-8?q?=E8=AF=81=E3=80=8C=E4=B8=80=E4=B8=AA=20tick=20=E5=8F=AA?= =?UTF-8?q?=E8=B7=91=E4=B8=80=E4=B8=AA=E5=89=AF=E6=9C=AC=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对标 #12.5 暴露的真缺口：`MemoryJobBackend` 每个副本各有一份 `setInterval`， 两个副本同时到点就是同一批「到期暂停 / 回收站清理 / outbox 转发」各跑两遍。 之前只有一行启动 warn，把「集群安全」写成了运维自觉 —— 而对标对象 （ZJMF `Cron.php` 的 `cron_lock` + 5 分钟陈旧期）同样场景下反而不会重复。 ## 做法 - `kernel_job_leases(name, holder, expiresAtMs)` + 一条语句完成抢占： `INSERT ... ON CONFLICT DO UPDATE ... WHERE expiresAtMs < now RETURNING name`。 命中即执行，空结果集即跳过 —— 并发时由行锁保证只有一个副本能命中。 - 租约按时间自然过期，**不需要死节点检测**：持有者被 kill，最多一个间隔后 就有别的副本接手。 - 租约时长 = 任务间隔（下限 1s / 上限 5 分钟）。取「间隔的若干倍」会把有效频率 压到几倍间隔以下 —— 抢到的副本每轮都成功、其余副本整个窗口全部跳过。 - 抢租约出错按「没抢到」处理：宁可这一轮不跑（下轮补），也不因数据库抖一下 变成多副本重复执行。 - 只给进程内后端用。BullMQ 本身按队列抢占，再套一层会让重试互相挡住。 ## 验证 - `tests/unit/job-lease-memory.test.ts` 4 例：抢到才执行 / 没抢到静默跳过 / 租约抛错时跳过且告警 / 租约时长取一个间隔（含 1s 下限、5min 上限）。 - `tests/integration/job-lease.test.ts` 4 例真实 DB：两副本只有一个抢到 / 过期后接手 / 不同任务互不阻塞 / 同一副本租约内不能自我放行。 - 既有 `tests/unit/jobs.test.ts`、`tests/unit/jobs-contract.test.ts` 全绿（21 例合计）。 - `tsc` 干净。 顺带一句：写测试时踩到「四个用例共用同一个任务名」——第二例被第一例的未过期租约 挡住，看起来像实现错了。用例已按任务名隔离，并在注释里写明原因。 --- apps/api/src/infra.ts | 5 +- apps/api/src/jobs/index.ts | 19 +++- apps/api/src/jobs/lease.ts | 68 ++++++++++++++ apps/api/src/jobs/memory-backend.ts | 41 ++++++++- apps/api/tests/integration/job-lease.test.ts | 77 ++++++++++++++++ apps/api/tests/unit/job-lease-memory.test.ts | 91 +++++++++++++++++++ .../migration.sql | 14 +++ packages/db/prisma/schema.prisma | 16 ++++ 8 files changed, 324 insertions(+), 7 deletions(-) create mode 100644 apps/api/src/jobs/lease.ts create mode 100644 apps/api/tests/integration/job-lease.test.ts create mode 100644 apps/api/tests/unit/job-lease-memory.test.ts create mode 100644 packages/db/prisma/migrations/20261027120000_kernel_job_lease/migration.sql diff --git a/apps/api/src/infra.ts b/apps/api/src/infra.ts index 3754c998..cb91ad8b 100644 --- a/apps/api/src/infra.ts +++ b/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 { // 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), diff --git a/apps/api/src/jobs/index.ts b/apps/api/src/jobs/index.ts index a01d63e7..7ec9af0c 100644 --- a/apps/api/src/jobs/index.ts +++ b/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 { // `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(); } diff --git a/apps/api/src/jobs/lease.ts b/apps/api/src/jobs/lease.ts new file mode 100644 index 00000000..c04b6572 --- /dev/null +++ b/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; +} + +/** 租约只需要 `$queryRaw`，测试里可以给一个假实现。 */ +export interface LeaseDb { + $queryRaw(query: TemplateStringsArray, ...values: unknown[]): Promise; +} + +/** + * 租约时长：等于任务间隔（下限 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 { + const now = Date.now(); + // 单条语句完成「插入或抢占已过期租约」，不需要事务： + // 并发时 Postgres 行锁保证只有一个 UPDATE 命中 `expiresAtMs < now`。 + const rows = await this.db.$queryRaw>` + 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; + } +} diff --git a/apps/api/src/jobs/memory-backend.ts b/apps/api/src/jobs/memory-backend.ts index 27e43d8d..d595c4ca 100644 --- a/apps/api/src/jobs/memory-backend.ts +++ b/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 { + 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, diff --git a/apps/api/tests/integration/job-lease.test.ts b/apps/api/tests/integration/job-lease.test.ts new file mode 100644 index 00000000..dffd24ef --- /dev/null +++ b/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 => { + 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); + }); +}); diff --git a/apps/api/tests/unit/job-lease-memory.test.ts b/apps/api/tests/unit/job-lease-memory.test.ts new file mode 100644 index 00000000..052d55b7 --- /dev/null +++ b/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 { + 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); + }); +}); diff --git a/packages/db/prisma/migrations/20261027120000_kernel_job_lease/migration.sql b/packages/db/prisma/migrations/20261027120000_kernel_job_lease/migration.sql new file mode 100644 index 00000000..97f03c9a --- /dev/null +++ b/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") +); diff --git a/packages/db/prisma/schema.prisma b/packages/db/prisma/schema.prisma index 2f6e79de..af6de8b3 100644 --- a/packages/db/prisma/schema.prisma +++ b/packages/db/prisma/schema.prisma @@ -914,3 +914,19 @@ model NavEntry { @@index([surface]) @@map("nav_entries") } + +/// 任务调度的跨副本租约（对标 #12.5）：无 Redis 时用数据库做单实例保证。 +/// +/// 进程内后端的每个副本都会按时触发定时器，执行前抢一次租约、抢到才跑； +/// 租约靠 `expiresAtMs` 自然过期，持有者被 kill 后由下一个副本接手。 +/// 有 Redis 时走 BullMQ 抢占，这张表不参与。 +model KernelJobLease { + /// 任务全名（`.`）。 + name String @id + /// 持有者标识，仅用于排查「上一轮是谁在跑」。 + holder String + /// 租约到期时间（epoch ms）。 + expiresAtMs BigInt + // 表名用 snake_case，与其余表一致。 + @@map("kernel_job_leases") +}