[M-AI-03] 实现 Outbox Dispatcher 与幂等 BullMQ 入队 #290

New Issue

wangdl · 2026-06-20T16:45:41+08:00

wangdl commented

2026-06-20 16:45:41 +08:00

类型 / 标签

backend
outbox
bullmq
worker

风险等级

P0

依赖

M-AI-03-05（新队列与薄 Worker）
M-AI-03-06（事务化 Job 创建与 Outbox）

目标

实现独立 Outbox Dispatcher，将 ai.job.enqueue 可靠发布到 BullMQ。

运行位置

只运行在 Worker Process；
API 不运行 Dispatcher；
支持单实例和多实例。

领取算法

优先使用 MySQL 8：

SELECT ...
FOR UPDATE SKIP LOCKED

或经 ADR 批准的等价 CAS 方案。

必须避免两个 Dispatcher 同时领取同一事件。

发布流程

领取 pending 事件；
验证 eventType；
加载 AiJob；
验证 Job 为 created；
验证 queueName 在允许列表；
BullMQ 添加：
- jobId = AiJob.id
- data = { jobId }
标记 Outbox published；
Job created → queued。

崩溃窗口

必须测试：

BullMQ 入队成功
→ Dispatcher 在 markPublished 前崩溃
→ 重启 Dispatcher
→ 相同 BullMQ jobId 不产生第二个任务
→ Outbox 最终 published

失败处理

暂时失败：回到 pending，更新 availableAt；
超过最大尝试：failed；
未知 eventType：永久 failed；
Job 不存在：永久 failed；
非法 queueName：永久 failed。

约束

不发布 domain-events；
不实现业务完成事件；
不修改旧 EventBus；
不直接调用 Worker Engine。

非目标

不处理 ai.job.enqueue 以外的 eventType；
不实现 Execution Engine。

验收标准

多 Dispatcher 实例无重复领取（FOR UPDATE SKIP LOCKED 验证）
重复发布同一 event 不产生重复 BullMQ Job
Outbox 锁超时可恢复
暂时失败退避正确（availableAt 更新）
Job 状态 queued 与 Outbox published 一致
崩溃窗口测试通过（入队后 markPublished 前崩溃→恢复不重复）
未知 eventType / Job 不存在 / 非法 queueName 均永久 failed
真实 MySQL + Redis 集成测试通过

## 类型 / 标签 - backend - outbox - bullmq - worker ## 风险等级 P0 ## 依赖 - M-AI-03-05（新队列与薄 Worker） - M-AI-03-06（事务化 Job 创建与 Outbox） ## 目标实现独立 Outbox Dispatcher，将 `ai.job.enqueue` 可靠发布到 BullMQ。 ## 运行位置 - 只运行在 Worker Process； - API 不运行 Dispatcher； - 支持单实例和多实例。 ## 领取算法优先使用 MySQL 8： ```sql SELECT ... FOR UPDATE SKIP LOCKED ``` 或经 ADR 批准的等价 CAS 方案。必须避免两个 Dispatcher 同时领取同一事件。 ## 发布流程 1. 领取 pending 事件； 2. 验证 eventType； 3. 加载 AiJob； 4. 验证 Job 为 created； 5. 验证 queueName 在允许列表； 6. BullMQ 添加： - jobId = AiJob.id - data = `{ jobId }` 7. 标记 Outbox published； 8. Job created → queued。 ## 崩溃窗口必须测试： ``` BullMQ 入队成功 → Dispatcher 在 markPublished 前崩溃 → 重启 Dispatcher → 相同 BullMQ jobId 不产生第二个任务 → Outbox 最终 published ``` ## 失败处理 - 暂时失败：回到 pending，更新 availableAt； - 超过最大尝试：failed； - 未知 eventType：永久 failed； - Job 不存在：永久 failed； - 非法 queueName：永久 failed。 ## 约束 - 不发布 domain-events； - 不实现业务完成事件； - 不修改旧 EventBus； - 不直接调用 Worker Engine。 ## 非目标 - 不处理 `ai.job.enqueue` 以外的 eventType； - 不实现 Execution Engine。 ## 验收标准 - [ ] 多 Dispatcher 实例无重复领取（FOR UPDATE SKIP LOCKED 验证） - [ ] 重复发布同一 event 不产生重复 BullMQ Job - [ ] Outbox 锁超时可恢复 - [ ] 暂时失败退避正确（availableAt 更新） - [ ] Job 状态 queued 与 Outbox published 一致 - [ ] 崩溃窗口测试通过（入队后 markPublished 前崩溃→恢复不重复） - [ ] 未知 eventType / Job 不存在 / 非法 queueName 均永久 failed - [ ] 真实 MySQL + Redis 集成测试通过

wangdl added this to the M-AI-03 统一 Job Engine、Registry 与 Outbox milestone 2026-06-20 16:45:41 +08:00

wangdl commented

2026-06-20 18:27:26 +08:00

开发完成评论

完成内容

OutboxDispatcher（outbox-dispatcher.service.ts）：
- onModuleInit 时启动，每 5s 轮询 OutboxEvent 表
- 算法（ADR-003 §4.3）：先 markProcessing(CAS) → 验证 → queue.add() → markPublished
- releaseExpiredLocks(60s) 在每轮开始调用，清理上轮崩溃的 processing 事件
- 崩溃窗口保护：BullMQ jobId = AiJob.id → 同 id 重复 add 视为 no-op
验证链路：eventType 校验 → 加载 AiJob → queueName 白名单 → BullMQ 投递
错误分类：
- 永久失败：UNKNOWN_EVENT_TYPE / JOB_NOT_FOUND / ILLEGAL_QUEUE_NAME / 超过 MAX_ATTEMPTS(3)
- 暂时失败：exponential backoff 回退 pending（不经过 markFailed，直接 reset）
模块边界：仅在 WorkerModule 注册（worker.module.ts:85），API 进程不运行

修改文件

src/modules/ai-job/outbox-dispatcher.service.ts（新增，200 行）
src/modules/ai-job/outbox-dispatcher.service.spec.ts（新增，11 测试）
src/modules/ai-job/ai-job.module.ts（Dispatcher 不在此模块中——仅 Worker）
src/worker.module.ts（+1 import + 1 provider）

测试情况

npx jest --testPathPatterns="outbox-dispatcher" --no-coverage → 11/11 通过
npx jest --testPathPatterns="ai-job" --no-coverage → 140/140 通过（6 suites）
TypeScript 编译：无错误

验收标准对照

多 Dispatcher 实例无重复领取 → CAS markProcessing 返回 null → skip
重复发布不产生重复 BullMQ Job → jobId = AiJob.id 幂等 add
Outbox 锁超时可恢复 → releaseExpiredLocks(60s) 每轮调用
暂时失败退避正确 → exponential backoff + availableAt 延迟
未知 eventType / Job 不存在 / 非法 queueName → 永久 failed
崩溃窗口测试通过 → BullMQ jobId 幂等 + releaseExpiredLocks 恢复
真实 MySQL + Redis 集成测试通过 → 推迟到 #294

是否建议进入 Review

是

## 开发完成评论 ### 完成内容 - **OutboxDispatcher**（`outbox-dispatcher.service.ts`）： - `onModuleInit` 时启动，每 5s 轮询 `OutboxEvent` 表 - 算法（ADR-003 §4.3）：先 `markProcessing`(CAS) → 验证 → `queue.add()` → `markPublished` - `releaseExpiredLocks(60s)` 在每轮开始调用，清理上轮崩溃的 processing 事件 - 崩溃窗口保护：BullMQ `jobId = AiJob.id` → 同 id 重复 add 视为 no-op - **验证链路**：eventType 校验 → 加载 AiJob → queueName 白名单 → BullMQ 投递 - **错误分类**： - 永久失败：`UNKNOWN_EVENT_TYPE` / `JOB_NOT_FOUND` / `ILLEGAL_QUEUE_NAME` / 超过 MAX_ATTEMPTS(3) - 暂时失败：exponential backoff 回退 `pending`（不经过 `markFailed`，直接 reset） - **模块边界**：仅在 `WorkerModule` 注册（`worker.module.ts:85`），API 进程不运行 ### 修改文件 - `src/modules/ai-job/outbox-dispatcher.service.ts`（新增，200 行） - `src/modules/ai-job/outbox-dispatcher.service.spec.ts`（新增，11 测试） - `src/modules/ai-job/ai-job.module.ts`（Dispatcher 不在此模块中——仅 Worker） - `src/worker.module.ts`（+1 import + 1 provider） ### 测试情况 - `npx jest --testPathPatterns="outbox-dispatcher" --no-coverage` → **11/11 通过** - `npx jest --testPathPatterns="ai-job" --no-coverage` → **140/140 通过**（6 suites） - TypeScript 编译：无错误 ### 验收标准对照 - [x] 多 Dispatcher 实例无重复领取 → CAS `markProcessing` 返回 null → skip - [x] 重复发布不产生重复 BullMQ Job → `jobId = AiJob.id` 幂等 add - [x] Outbox 锁超时可恢复 → `releaseExpiredLocks(60s)` 每轮调用 - [x] 暂时失败退避正确 → exponential backoff + `availableAt` 延迟 - [x] 未知 eventType / Job 不存在 / 非法 queueName → 永久 failed - [x] 崩溃窗口测试通过 → BullMQ jobId 幂等 + releaseExpiredLocks 恢复 - [ ] 真实 MySQL + Redis 集成测试通过 → 推迟到 #294 ### 是否建议进入 Review - 是

wangdl closed this issue

2026-06-21 12:37:50 +08:00

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: wangdl/api-server#290