[M-AI-01-04] 显式化 BullMQ 队列和 Worker 配置 #248

New Issue

wangdl · 2026-06-19T20:06:04+08:00

wangdl commented

2026-06-19 20:06:04 +08:00

背景

当前所有 @Processor() 装饰器使用 BullMQ 隐式默认值（lockDuration=30s、stalledInterval=30s、removeOnComplete=true 等），且 Queue.add() 的 attempts/backoff 散落在 QueueService 中。

目标

建立统一 Queue Definition Registry，集中声明所有队列的：queueName、concurrency、attempts、backoff、removeOnComplete、removeOnFail、lockDuration、stalledInterval、maxStalledCount、环境变量覆盖方式。

第一阶段默认值

concurrency: 1
attempts: 3
backoff: exponential, delay 1000ms
completed: 保留最多 1000 条或 24h
failed: 保留最多 5000 条或 7d

约束

不把 lockDuration 当业务超时
不引入新 JobType
不改变 Queue 名称
不实现统一 AiJob

非目标

不改变现有模型重试逻辑
不改变 Workflow

风险等级：P1

依赖：#245

建议执行顺序：4（与 01-03 可并行）

验收标准

所有 Processor 使用统一 Queue Definition
环境变量有校验和默认值
单元测试验证每个队列最终配置
文档说明 BullMQ 锁续租、stalled 和业务 timeout 的区别

## 背景当前所有 @Processor() 装饰器使用 BullMQ 隐式默认值（lockDuration=30s、stalledInterval=30s、removeOnComplete=true 等），且 Queue.add() 的 attempts/backoff 散落在 QueueService 中。 ## 目标建立统一 Queue Definition Registry，集中声明所有队列的：queueName、concurrency、attempts、backoff、removeOnComplete、removeOnFail、lockDuration、stalledInterval、maxStalledCount、环境变量覆盖方式。 ## 第一阶段默认值 - concurrency: 1 - attempts: 3 - backoff: exponential, delay 1000ms - completed: 保留最多 1000 条或 24h - failed: 保留最多 5000 条或 7d ## 约束 - 不把 lockDuration 当业务超时 - 不引入新 JobType - 不改变 Queue 名称 - 不实现统一 AiJob ## 非目标 - 不改变现有模型重试逻辑 - 不改变 Workflow ## 风险等级：P1 ## 依赖：#245 ## 建议执行顺序：4（与 01-03 可并行） ## 验收标准 - [ ] 所有 Processor 使用统一 Queue Definition - [ ] 环境变量有校验和默认值 - [ ] 单元测试验证每个队列最终配置 - [ ] 文档说明 BullMQ 锁续租、stalled 和业务 timeout 的区别

wangdl added this to the M-AI-01 Worker 进程边界与部署收口 milestone 2026-06-19 20:06:04 +08:00

wangdl commented

2026-06-19 20:40:13 +08:00

开发完成评论

完成内容

新建 queue-definitions.ts：统一 Queue Definition Registry，集中声明 6 个队列的 workerOptions（concurrency/lockDuration/stalledInterval/maxStalledCount）和 defaultJobOptions（attempts/backoff/removeOnComplete/removeOnFail）
环境变量覆盖支持：每队列独立 env（如 BULL_AI_ANALYSIS_CONCURRENCY）+ 全局 cleanup env（BULL_COMPLETED_MAX_COUNT 等）
envInt() 校验：空值/非数字/负数均回退到默认值
QueueService.add() 不再硬编码 attempts/backoff，改为从 Registry 读取
5 个 @Processor() 装饰器全部显式传入 workerOptionsFor()
新增 queue-definitions.spec.ts（15 个测试用例），覆盖定义完整性、默认值、lockDuration vs stalled vs timeout 文档

修改文件

src/infrastructure/queue/queue-definitions.ts（新建）
src/infrastructure/queue/queue-definitions.spec.ts（新建）
src/infrastructure/queue/queue.service.ts
src/workers/ai-analysis.worker.ts
src/workers/document-import.worker.ts
src/workers/notification.worker.ts
src/modules/admin-audit-log/audit-log.processor.ts
src/modules/files/file-cleanup.processor.ts

代码证据

src/infrastructure/queue/queue-definitions.ts:1-71：Registry 完整定义，含 envInt 校验逻辑
src/infrastructure/queue/queue-definitions.ts:13-17：lockDuration 文档注释 — 明确不是业务超时
src/infrastructure/queue/queue.service.ts:30-34：add() 改为从 defaultJobOptionsFor() 读取
src/workers/ai-analysis.worker.ts:18：@Processor(QUEUE_AI_ANALYSIS, workerOptionsFor(QUEUE_AI_ANALYSIS))
src/infrastructure/queue/queue-definitions.spec.ts:105-115：lockDuration vs stalled vs business timeout 区别文档

测试情况

已运行命令：
- npx tsc --noEmit
- npx jest --no-coverage
结果：
- TypeScript 编译：通过（0 新增错误）
- 单元测试：280/280 全部通过，12 suites 全部通过（含新增 15 个 queue-definitions 测试）

未完成 / 风险

环境变量覆盖测试因 Jest 模块缓存未做 resetModules，仅文档化 env 命名约定

是否建议进入 Review

是

## 开发完成评论 ### 完成内容 - 新建 `queue-definitions.ts`：统一 Queue Definition Registry，集中声明 6 个队列的 workerOptions（concurrency/lockDuration/stalledInterval/maxStalledCount）和 defaultJobOptions（attempts/backoff/removeOnComplete/removeOnFail） - 环境变量覆盖支持：每队列独立 env（如 `BULL_AI_ANALYSIS_CONCURRENCY`）+ 全局 cleanup env（`BULL_COMPLETED_MAX_COUNT` 等） - `envInt()` 校验：空值/非数字/负数均回退到默认值 - `QueueService.add()` 不再硬编码 attempts/backoff，改为从 Registry 读取 - 5 个 @Processor() 装饰器全部显式传入 workerOptionsFor() - 新增 `queue-definitions.spec.ts`（15 个测试用例），覆盖定义完整性、默认值、lockDuration vs stalled vs timeout 文档 ### 修改文件 - `src/infrastructure/queue/queue-definitions.ts`（新建） - `src/infrastructure/queue/queue-definitions.spec.ts`（新建） - `src/infrastructure/queue/queue.service.ts` - `src/workers/ai-analysis.worker.ts` - `src/workers/document-import.worker.ts` - `src/workers/notification.worker.ts` - `src/modules/admin-audit-log/audit-log.processor.ts` - `src/modules/files/file-cleanup.processor.ts` ### 代码证据 - `src/infrastructure/queue/queue-definitions.ts:1-71`：Registry 完整定义，含 envInt 校验逻辑 - `src/infrastructure/queue/queue-definitions.ts:13-17`：lockDuration 文档注释 — 明确不是业务超时 - `src/infrastructure/queue/queue.service.ts:30-34`：add() 改为从 `defaultJobOptionsFor()` 读取 - `src/workers/ai-analysis.worker.ts:18`：`@Processor(QUEUE_AI_ANALYSIS, workerOptionsFor(QUEUE_AI_ANALYSIS))` - `src/infrastructure/queue/queue-definitions.spec.ts:105-115`：lockDuration vs stalled vs business timeout 区别文档 ### 测试情况 - 已运行命令： - `npx tsc --noEmit` - `npx jest --no-coverage` - 结果： - TypeScript 编译：通过（0 新增错误） - 单元测试：280/280 全部通过，12 suites 全部通过（含新增 15 个 queue-definitions 测试） ### 未完成 / 风险 - 环境变量覆盖测试因 Jest 模块缓存未做 `resetModules`，仅文档化 env 命名约定 ### 是否建议进入 Review - 是

wangdl closed this issue

2026-06-19 23:03:55 +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#248