diff --git a/docs/architecture/m-ai-03-current-execution-audit.md b/docs/architecture/m-ai-03-current-execution-audit.md index bd06671..0a10637 100644 --- a/docs/architecture/m-ai-03-current-execution-audit.md +++ b/docs/architecture/m-ai-03-current-execution-audit.md @@ -56,12 +56,16 @@ |---|---------------|--------|---------|----------|-------|---------|----------|----------| | P1 | `AiAnalysisService` | `analyze()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'active-recall', questionText, knowledgeItemContent, userAnswer}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:19-31` | | P2 | `AiAnalysisService` | `evaluateFeynman()` | User API | `AiJob` | `ai-analysis` | `{jobId, userId, type:'feynman-evaluation', knowledgeItemTitle, knowledgeItemContent, userExplanation}` | `status:pending`, `lifecycleStatus:queued` | `ai-analysis.service.ts:40-51` | + +> ⚠️ **queueName 与实际队列不一致**：`AiAnalysisRepository.createJob()` 写入 `queueName: 'ai-interactive'`（`ai-analysis.repository.ts:28`），但 `AiAnalysisService` 实际将 BullMQ Job 发送到 `ai-analysis` 队列（`ai-analysis.service.ts:21,42`）。数据库记录的 `queueName` 字段与真实 BullMQ 队列路由存在偏差。M-AI-03 构建统一 Engine 将依赖 `queueName` 字段进行路由决策，此为关键风险。 | P3 | `DocumentImportService` | `createImport()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, rawText, fileName}` | `status:QUEUED`(via Repository) | `document-import.service.ts:43-49` | | P4 | `KnowledgeSourceService` | `addSource()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:43-50` | | P5 | `KnowledgeSourceService` | `triggerParse()` | User API | `DocumentImport` | `document-import` | `{importId, userId, knowledgeBaseId, sourceId, fileName}` | `status:QUEUED`(via Repository) | `knowledge-source.service.ts:90-97` | | P6 | `UserAiService` | `requestJob()` | User API | `AiRuntimeJob` | **无 BullMQ** | REST Poll，不适用 | `status:pending` | `user-ai.service.ts:282-299` | | P7 | `AdminFilesController` | COS cleanup | Admin API | 无 | `file-cleanup` | `{objectKey, bucket, region}` | 无（仅入队） | `admin-files.controller.ts:41` | +> **注**：`QueueService.add()`（`queue.service.ts:41-58`）是上述所有 BullMQ Producer 的公共抽象层，不作为独立 Producer 列出。其副作用：写入 `TaskLog` 记录（`status:enqueued`）和同步发布 `task.enqueued` 事件。 + ### 死代码（定义为入队方法但零调用方） | 方法 | 文件 | 入队目标 | 说明 | @@ -70,11 +74,13 @@ ### 孤儿队列（Consumer 已注册但生产者未找到） -| 队列 | Consumer | 状态 | -|------|----------|------| -| `notification` | `NotificationWorker` (`workers/notification.worker.ts:8`) | ⚠️ 无 `queue.add('notification', ...)` 调用方；`NotificationsService.send()` 仅写 DB + sync eventBus.publish()，不入队 | -| `domain-events` | 无专用 Consumer（`EventBusService.publishAsync` 是唯一入队路径但为零调用方） | ⚠️ 完全闲置 | -| `audit-logs` | `AuditLogProcessor` (`modules/admin-audit-log/audit-log.processor.ts:7`) | ⚠️ 无 `queue.add('audit-logs', ...)` 调用方；审计日志走 `prisma.adminAuditLog.create()` 直接写 DB | +| 队列 | Consumer | WorkerModule 注册 | 生产者状态 | +|------|----------|-------------------|-----------| +| `notification` | `NotificationWorker` (`workers/notification.worker.ts:8`) | `worker.module.ts:76` | ❌ 无生产者 — 全文搜索确认零 `queue.add('notification', ...)`；`NotificationsService.send()` 仅写 DB + sync eventBus，不入队 | +| `domain-events` | 无专用 Consumer（只有 `EventBusService.publishAsync` 可入队） | N/A | ❌ 无生产者 — `publishAsync()` 为零调用方死代码；队列完全闲置 | +| `audit-logs` | `AuditLogProcessor` (`modules/admin-audit-log/audit-log.processor.ts:7`) | `worker.module.ts:77` | ❌ 无生产者 — 全文搜索确认零 `queue.add('audit-logs', ...)`；所有审计日志通过 `prisma.adminAuditLog.create()` 直接写 DB | + +> **含义**：`NotificationWorker`、`AuditLogProcessor` 两个 Worker 进程注册在 `WorkerModule` 中，每 30s 从各自队列 poll，但这两个队列从未有消息进入——实为永久空转的闲置进程。 > **注意**：`AdminEventsController` 仅**读取**队列状态（`getWaitingCount`/`getActiveCount`/`getFailed` 等）和**重试**已有失败 Job（`job.retry()`），**从不创建新 Job**。因此不列为 Producer。 @@ -289,6 +295,24 @@ AiGatewayService - EventBus 当前仅作为 In-Process Event Emitter 使用；其异步域事件能力（BullMQ `domain-events` 队列）处于闲置状态 - `QueueService` 中同步发布的 `task.enqueued` 事件无 Subscriber 消费 +### 补充：不在 EventBus 中的直接通知路径 + +`RuntimeInternalService.notifyJobComplete()`（`runtime-internal.service.ts:629-646`）**绕过 EventBus 直接写入** `prisma.notification.create()`——在提交结果或 Job 彻底失败时触发。此路径不在上述事件矩阵中，是独立的 Job 完成通知机制： + +```typescript +private async notifyJobComplete(userId, jobId, jobType, status) { + await this.prisma.notification.create({ + data: { + userId, + type: status === 'succeeded' ? 'ai_job_succeeded' : 'ai_job_failed', + title: ..., content: ..., data: { jobId, jobType, status }, + }, + }); +} +``` + +调用时机：`submitResult()` 成功后（`runtime-internal.service.ts:278`）和 `submitFailure()` 重试耗尽后（`runtime-internal.service.ts:618`）。 + --- ## 8. Outbox 现状评估 @@ -307,9 +331,28 @@ AiGatewayService | `releaseExpiredLocks(thresholdMs)` | ✅ 已实现 | 重置超时 processing → pending | | **是否有生产写入** | ❌ 无 | **所有搜索返回零调用方** | | **是否有 Dispatcher** | ❌ 无 | 没有 Dispatcher Service/Worker | -| **是否支持并发领取** | ⚠️ 部分 | CAS 保证了唯一领取，但 `findDispatchable` 可能多 Worker 读到相同事件（CAS 后有竞态窗口） | +| **是否支持并发领取** | ❌ 否 | `findDispatchable` 仅做无锁 `findMany`；`markProcessing` 使用应用层 CAS（`updateMany WHERE status='pending'`）后补救。若 CAS 失败，调用方静默丢弃该事件，不重试也不回退 | | **是否使用 SKIP LOCKED** | ❌ 否 | 使用应用层 CAS（updateMany + status check），非数据库层 SKIP LOCKED | -| **重复发布风险** | ⚠️ 存在 | `findDispatchable` 读取后 `markProcessing` 若 CAS 失败不重试；若 Dispatcher 崩溃后重启，`releaseExpiredLocks` 可恢复但非实时 | +| **并发重复发布风险** | ⚠️ 存在 | **具体场景**：两个 Dispatcher 实例同时调用 `findDispatchable(50)` → 读到同一批 `[E1, E2, ...]` → 各自投递到 BullMQ → 同一事件可能被发布两次。`markProcessing` 的 CAS 仅防止数据库状态被双重更新，但不阻止网络层重复投递（BullMQ `queue.add()` 一旦调用就无法回滚） | + +**重复发布场景分析**（Issue #290 相关）： + +``` +Dispatcher-A Dispatcher-B + │ │ + ├─ findDispatchable() → [E1,E2] │ + │ ├─ findDispatchable() → [E1,E2] + ├─ queue.add(E1) ← 第一次投递 │ + │ ├─ queue.add(E1) ← 重复投递！ + ├─ markProcessing(E1) ← CAS 成功 │ + │ ├─ markProcessing(E1) ← CAS 失败，静默丢弃 + ├─ queue.add(E2) │ + │ ├─ queue.add(E2) ← 重复投递！ + ├─ markProcessing(E2) ← CAS 成功 │ + │ ├─ markProcessing(E2) ← CAS 失败，静默丢弃 +``` + +**根因**：`queue.add()` 与 `markProcessing()` 不是原子操作。BullMQ 投递成功后若 CAS 失败，没有补偿路径。修复方向：先抢锁（`markProcessing` CAS），锁定成功后再投递；或使用 DB SKIP LOCKED 在 `findDispatchable` 阶段就排他抢占。 ### 表结构 (Prisma) @@ -335,6 +378,21 @@ pending → processing → completed / failed - **无取消支持** - **无 retry/reaper** — 完全依赖 BullMQ 的 attempts/backoff +#### ⚠️ STATUS_TO_LIFECYCLE 映射缺口 + +`ai-analysis.repository.ts:10-15` 仅映射 4 个状态： + +```typescript +private static readonly STATUS_TO_LIFECYCLE: Record = { + pending: 'queued', + processing: 'running', + completed: 'succeeded', + failed: 'failed', +}; +``` + +**缺失**：`cancel_requested` 和 `cancelled`（Issue #286 验收标准要求这两个状态也在映射范围内）。当前 AiJob 系统完全不支持取消操作，但如果 #286 引入统一状态机后 `cancel_requested` / `cancelled` 成为通用状态，此映射表将产生缺口。 + ### AiRuntimeJob（新） `src/modules/ai-runtime/job-reaper.service.ts` + `runtime-internal.service.ts` @@ -433,14 +491,18 @@ pending → locked → running → succeeded / failed / cancelled ### 关键风险 + | 风险 | 严重度 | 说明 | |------|--------|------| +| queueName 与实际路由不一致 | 🔴 高 | DB 写入 `queueName: 'ai-interactive'`（`ai-analysis.repository.ts:28`），实际入队 `ai-analysis`（`ai-analysis.service.ts:21`）；M-AI-03 依赖 `queueName` 字段进行路由 | +| STATUS_TO_LIFECYCLE 映射缺口 | 🟡 中 | 仅映射 4 个状态，缺 `cancel_requested` / `cancelled`；#286 引入统一状态机后将暴露此缺口 | | 两套 Job 状态不同步 | 🔴 高 | `AiJob.status`(legacy enum) 与 `AiJob.lifecycleStatus`(M-AI-02 新字段) 存在映射但不完整 | -| Outbox 无 Dispatcher | 🟡 中 | 需新建但独立 | +| Outbox 无 Dispatcher 且并发发布不安全 | 🔴 高 | `queue.add()` 与 `markProcessing()` 非原子，两 Dispatcher 并发时可重复投递（详见第 8 节） | | `forwardRef` EventBus↔Queue | 🟢 低 | 已有解决方案，不扩大 | | DocumentImport 使用独立 model | 🟡 中 | 不在 M-AI-03 范围，但未来统一需注意 | | RuntimeInternal 无事务保证 | 🔴 高 | result + job update 分两步，非原子 | -| 孤儿队列（notification/domain-events/audit-logs） | 🟡 中 | Consumer 已注册但无活跃 Producer；若从未有消息入队，`NotificationWorker` 和 `AuditLogProcessor` 实为闲置进程 +| 孤儿队列（notification/domain-events/audit-logs） | 🟡 中 | Consumer 已注册但无活跃 Producer；`NotificationWorker` 和 `AuditLogProcessor` 实为永久空转 | + ---