From 02ac9b6a8135ab89ae8212a335e09bb9eb3b83c6 Mon Sep 17 00:00:00 2001 From: worable Date: Sun, 25 Oct 2026 17:31:10 +0800 Subject: [PATCH] =?UTF-8?q?test(jobs):=20=E7=94=A8=E7=9C=9F=E5=AE=9E=20Red?= =?UTF-8?q?is=20=E5=8F=8C=E5=AE=9E=E4=BE=8B=E5=AE=9E=E6=B5=8B=E3=80=8C?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=20tick=20=E5=8F=AA=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E4=B8=80=E6=AC=A1=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 对标 #12.5 验收门槛里的最后一条：「集群双实例只执行一次」。 在此之前这条只有**推论** —— `bullmq-backend.ts` 的注释写着「BullMQ 保证一个 job 只被一个 worker 取走，`upsertJobScheduler` 每个 id 只有一份调度」。注释不是证据。 新增 `apps/api/tests/integration/jobs-redis-cluster.test.ts`（3 例，真实 Redis）： 1. **两个副本声明同一个调度 → Redis 里只有一份**。否则每个副本按自己的节奏各发 一遍任务，用户会收到双份通知。 2. **两个副本同时运行 → 一次 tick 只执行一次**。除了总数上限，还断言任意两次执行 的间隔 > 400ms（间隔 1s 的调度不会出现「几乎同时两次」，那正是双取的特征）。 同时断言执行次数 ≥ 2 —— 防的是「测试因为根本没跑而通过」。 3. **同一个 `jobId` 从两个副本入队 → 只执行一次**。 ## 为什么需要独立 db 用例清理用 `obliterate`，会删掉整个队列。所以： - 只在 `TEST_REDIS_URL` 指向 **db ≥ 1** 时运行（本机验证过 `…/6379` 与 `…/6379/0` 都会被跳过，不会误清开发库）； - CI 里加 `TEST_REDIS_URL: redis://127.0.0.1:6379/9`（redis service 本来就在）。 不设环境变量时整组跳过，不给本地开发引入常驻 Redis 依赖。 验证：本机对着 `sp-redis` 跑 3 例全绿（4.9s）；不设 / 设 db 0 两种写法均 3 skipped；eslint 干净。 --- .github/workflows/ci.yml | 2 + .../integration/jobs-redis-cluster.test.ts | 126 ++++++++++++++++++ 2 files changed, 128 insertions(+) create mode 100644 apps/api/tests/integration/jobs-redis-cluster.test.ts diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 0bb432be..57b83ae7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,6 +48,8 @@ jobs: DATABASE_URL: postgresql://stackpanel:stackpanel@127.0.0.1:5432/stackpanel TEST_DATABASE_URL: postgresql://stackpanel:stackpanel@127.0.0.1:5432/stackpanel_test REDIS_URL: redis://127.0.0.1:6379 + # 集群语义实测（`jobs-redis-cluster.test.ts`）用独立 db，测试会 `obliterate` 整个队列。 + TEST_REDIS_URL: redis://127.0.0.1:6379/9 JWT_SECRET: ci_jwt_secret_at_least_32_characters_0123456789 SETTINGS_ENCRYPTION_KEY: ci_settings_key_at_least_32_characters_0123456789 diff --git a/apps/api/tests/integration/jobs-redis-cluster.test.ts b/apps/api/tests/integration/jobs-redis-cluster.test.ts new file mode 100644 index 00000000..21e31146 --- /dev/null +++ b/apps/api/tests/integration/jobs-redis-cluster.test.ts @@ -0,0 +1,126 @@ +import { Queue } from 'bullmq'; +import { afterAll, afterEach, describe, expect, it } from 'vitest'; +import { REDIS_KEY_PREFIX } from '@stackpanel/sdk'; +import { BullMqJobBackend, KERNEL_QUEUE_NAME } from '../../src/jobs/bullmq-backend.ts'; +import { redisConnectionOptions } from '../../src/jobs/redis-connection.ts'; + +/** + * 集群语义的实测（对标 #12.5 的验收门槛「双实例只执行一次」）。 + * + * 之前这条只有**推论**：`bullmq-backend.ts` 的注释写着「BullMQ 保证一个 job 只被 + * 一个 worker 取走，`upsertJobScheduler` 每个 id 只有一份调度」。注释不是证据 —— + * 这里用同一个 Redis 起两个独立的 backend（各自 worker，等价于两个副本）， + * 把「只执行一次」变成可复现的断言。 + * + * 需要真实 Redis：设 `TEST_REDIS_URL`（如 `redis://127.0.0.1:6379/9`）才跑， + * 否则整组跳过 —— 不给 CI 引入必须常驻的 Redis 依赖。务必用**独立 db**， + * 测试结束会 `obliterate` 整个队列。 + */ +const REDIS_URL = process.env.TEST_REDIS_URL ?? ''; + +/** + * 只允许在**独立 db** 上跑：清理用 `obliterate` 会删掉整个队列， + * 指到 db 0（开发/生产的默认库）就等于把线上排队任务清空。 + */ +function dedicatedDb(url: string): boolean { + if (!url) return false; + const db = new URL(url).pathname.replace(/^\//, ''); + return Number(db) >= 1; +} + +const suite = dedicatedDb(REDIS_URL) ? describe : describe.skip; + +const silent = { info: () => {}, warn: () => {} }; + +function backend(): BullMqJobBackend { + return new BullMqJobBackend({ redisUrl: REDIS_URL, logger: silent }); +} + +function inspectQueue(): Queue { + return new Queue(KERNEL_QUEUE_NAME, { + connection: redisConnectionOptions(REDIS_URL), + prefix: REDIS_KEY_PREFIX.bull, + }); +} + +const started: BullMqJobBackend[] = []; + +function track(b: BullMqJobBackend): BullMqJobBackend { + started.push(b); + return b; +} + +suite('BullMQ 集群语义（双实例共用一个 Redis）', () => { + afterEach(async () => { + while (started.length) await started.pop()?.stop(); + const queue = inspectQueue(); + await queue.obliterate({ force: true }).catch(() => undefined); + await queue.close(); + }); + + afterAll(() => { + // 组内已清理；此处仅保留给 future 用例挂全局钩子。 + }); + + it('两个副本都声明同一个调度时，Redis 里只有一份', async () => { + const name = `test.redis.single-scheduler.${Date.now()}`; + const a = track(backend()); + const b = track(backend()); + for (const instance of [a, b]) { + await instance.schedule({ fullName: name, schedule: { everyMs: 60_000 }, payload: null }); + } + await a.start(); + await b.start(); + + const queue = inspectQueue(); + const schedulers = await queue.getJobSchedulers(0, -1, true); + await queue.close(); + + // 关键：两个副本各自 `upsertJobScheduler` 同一个 id，结果仍是一份调度， + // 否则每个副本都会按自己的节奏各发一遍任务。 + expect(schedulers.filter((entry) => entry.key === name)).toHaveLength(1); + }, 30_000); + + it('两个副本同时运行时，一次 tick 只执行一次', async () => { + const name = `test.redis.single-run.${Date.now()}`; + const runs: number[] = []; + const everyMs = 1_000; + for (const instance of [track(backend()), track(backend())]) { + instance.handle(name, async () => { + runs.push(Date.now()); + }); + await instance.schedule({ fullName: name, schedule: { everyMs }, payload: null }); + } + for (const instance of started) await instance.start(); + + // 每个副本的并发都是默认值，所以「同一毫秒附近被执行两次」只可能来自 + // 「两个副本各取走了同一个 job」——这正是要证伪的东西。 + await new Promise((resolve) => setTimeout(resolve, 3_200)); + + expect(runs.length).toBeGreaterThanOrEqual(2); + // 间隔 1s 的调度在 3.2s 内不可能出现 4 次以上；重复执行会是 2 倍。 + expect(runs.length).toBeLessThanOrEqual(4); + const gap = runs.slice(1).map((at, index) => at - (runs[index] ?? 0)); + expect(Math.min(...gap)).toBeGreaterThan(400); + }, 30_000); + + it('两个副本用同一个 jobId 入队，只执行一次', async () => { + const name = `test.redis.dedup.${Date.now()}`; + const jobId = `dedupe-${Date.now()}`; + const runs: unknown[] = []; + const [a, b] = [track(backend()), track(backend())]; + for (const instance of [a, b]) { + instance.handle(name, async (payload) => { + runs.push(payload); + }); + } + await a.start(); + await b.start(); + + await Promise.all([a.enqueue(name, { n: 1 }, { jobId }), b.enqueue(name, { n: 2 }, { jobId })]); + await new Promise((resolve) => setTimeout(resolve, 1_500)); + + // BullMQ 的 jobId 去重是「等待中/延迟中的同 id 只留一个」。 + expect(runs).toHaveLength(1); + }, 30_000); +});