From 6668bba9b4fc67c58ac2f24bfed9d2e26616401c Mon Sep 17 00:00:00 2001 From: wangdl Date: Sat, 20 Jun 2026 18:20:22 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M-AI-03=20Outbox=20Dispatcher=20?= =?UTF-8?q?=E2=80=94=20=E5=8F=AF=E9=9D=A0=E6=8A=95=E9=80=92=20ai.job.enque?= =?UTF-8?q?ue=20=E5=88=B0=20BullMQ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - OutboxDispatcher: 每 5s 轮询 OutboxEvent 表，CAS 领取 → 验证 → BullMQ 投递 → 标记 published - 采用 ADR-003 §4.3 先 CAS 后投递算法：markProcessing(CAS) → queue.add() → markPublished - BullMQ jobId = AiJob.id（幂等重复 add） - 崩溃窗口保护：bullmq jobId 幂等 + releaseExpiredLocks(60s) - 错误处理：永久失败（未知eventType/Job不存在/非法queueName/超过最大尝试） vs 暂时失败（exponential backoff 回退 pending） - 仅在 WorkerModule 注册（API 进程不运行 Dispatcher） Co-Authored-By: Claude --- src/modules/ai-job/ai-job.module.ts | 3 +- .../ai-job/outbox-dispatcher.service.spec.ts | 278 ++++++++++++++++++ .../ai-job/outbox-dispatcher.service.ts | 206 +++++++++++++ src/worker.module.ts | 2 + 4 files changed, 488 insertions(+), 1 deletion(-) create mode 100644 src/modules/ai-job/outbox-dispatcher.service.spec.ts create mode 100644 src/modules/ai-job/outbox-dispatcher.service.ts diff --git a/src/modules/ai-job/ai-job.module.ts b/src/modules/ai-job/ai-job.module.ts index 13a0c36..6ba066a 100644 --- a/src/modules/ai-job/ai-job.module.ts +++ b/src/modules/ai-job/ai-job.module.ts @@ -11,7 +11,8 @@ import { AiJobCreationService } from './ai-job-creation.service'; * M-AI-03: AiJob 统一 Job Engine 模块 * * 当前包含：状态机、生命周期仓储、Registry、CreationService。 - * 后续 Issue（#290–#293）逐步新增：OutboxDispatcher、Engine、Projector。 + * OutboxDispatcher 仅在 WorkerModule 注册（API 进程不运行）。 + * 后续 Issue（#291–#293）逐步新增：Engine、Projector。 */ @Module({ imports: [PrismaModule, AiRuntimeModule], diff --git a/src/modules/ai-job/outbox-dispatcher.service.spec.ts b/src/modules/ai-job/outbox-dispatcher.service.spec.ts new file mode 100644 index 0000000..886f3af --- /dev/null +++ b/src/modules/ai-job/outbox-dispatcher.service.spec.ts @@ -0,0 +1,278 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { OutboxDispatcher } from './outbox-dispatcher.service'; +import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; +import { QueueService } from '../../infrastructure/queue/queue.service'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; + +describe('OutboxDispatcher', () => { + let dispatcher: OutboxDispatcher; + let outboxRepo: any; + let queueService: any; + let prisma: any; + + function makeEvent(overrides?: any) { + return { + id: 'evt-001', + eventType: 'ai.job.enqueue', + aggregateType: 'AiJob', + aggregateId: 'job-001', + dedupeKey: 'ai.job.enqueue:job-001', + payload: { jobId: 'job-001' }, + status: 'pending', + attemptCount: 0, + availableAt: new Date(), + lockedAt: null, + lockedBy: null, + lastErrorCode: null, + lastErrorMessage: null, + ...overrides, + }; + } + + beforeEach(async () => { + outboxRepo = { + findDispatchable: jest.fn(), + markProcessing: jest.fn(), + markPublished: jest.fn(), + markFailed: jest.fn(), + releaseExpiredLocks: jest.fn(), + }; + + queueService = { add: jest.fn() }; + + prisma = { + aiJob: { findUnique: jest.fn() }, + outboxEvent: { update: jest.fn() }, + }; + + // Suppress NestJS logger noise in tests + jest.spyOn(require('@nestjs/common').Logger.prototype, 'log').mockImplementation(() => {}); + jest.spyOn(require('@nestjs/common').Logger.prototype, 'warn').mockImplementation(() => {}); + jest.spyOn(require('@nestjs/common').Logger.prototype, 'error').mockImplementation(() => {}); + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + OutboxDispatcher, + { provide: OutboxRepository, useValue: outboxRepo }, + { provide: QueueService, useValue: queueService }, + { provide: PrismaService, useValue: prisma }, + ], + }).compile(); + + dispatcher = module.get(OutboxDispatcher); + // Don't start the interval + }); + + describe('正常分发', () => { + it('领取 → CAS 抢锁 → 验证 → 投递 BullMQ → 标记 published', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing', lockedBy: 'x' })); + prisma.aiJob.findUnique.mockResolvedValue({ + id: 'job-001', + queueName: 'ai-interactive', + lifecycleStatus: 'queued', + }); + queueService.add.mockResolvedValue({ id: 'job-001' }); + outboxRepo.markPublished.mockResolvedValue(undefined); + + const result = await dispatcher.cycle(); + expect(result.published).toBe(1); + expect(result.skipped).toBe(0); + + // Step order: CAS first, then enqueue + expect(outboxRepo.markProcessing).toHaveBeenCalledWith('evt-001', expect.any(String)); + expect(queueService.add).toHaveBeenCalledWith( + 'ai-interactive', + { jobId: 'job-001' }, + { jobId: 'job-001' }, // BullMQ jobId = AiJob.id + ); + expect(outboxRepo.markPublished).toHaveBeenCalledWith('evt-001'); + }); + }); + + describe('CAS 并发安全', () => { + it('markProcessing 返回 null → skip（被其他 Dispatcher 抢走）', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(null); // CAS failed + + const result = await dispatcher.cycle(); + expect(result.skipped).toBe(1); + expect(result.published).toBe(0); + // 不应投递 BullMQ + expect(queueService.add).not.toHaveBeenCalled(); + }); + }); + + describe('幂等 BullMQ 入队', () => { + it('相同 jobId 重复 add → BullMQ 视为 no-op（不抛错）', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing' })); + prisma.aiJob.findUnique.mockResolvedValue({ + id: 'job-001', + queueName: 'ai-interactive', + lifecycleStatus: 'queued', + }); + // BullMQ 同 jobId repeat add → no error + queueService.add.mockResolvedValue({ id: 'job-001' }); + outboxRepo.markPublished.mockResolvedValue(undefined); + + // 第一次 + await dispatcher.cycle(); + // 模拟第二次（相同 event 被 releaseExpiredLocks 恢复） + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ status: 'processing' })); + await dispatcher.cycle(); + + // 两次都成功 published + expect(outboxRepo.markPublished).toHaveBeenCalledTimes(2); + }); + }); + + describe('锁超时恢复', () => { + it('releaseExpiredLocks 在每轮开始调用', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(3); // 3 expired locks released + outboxRepo.findDispatchable.mockResolvedValue([]); + + await dispatcher.cycle(); + expect(outboxRepo.releaseExpiredLocks).toHaveBeenCalledWith(60_000); + }); + }); + + describe('未知 eventType', () => { + it('非 ai.job.enqueue → 永久 failed', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([ + makeEvent({ eventType: 'unknown.event', id: 'evt-bad' }), + ]); + outboxRepo.markProcessing.mockResolvedValue( + makeEvent({ eventType: 'unknown.event', id: 'evt-bad' }), + ); + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + expect(outboxRepo.markFailed).toHaveBeenCalledWith( + 'evt-bad', + 'UNKNOWN_EVENT_TYPE', + expect.stringContaining('unknown.event'), + ); + expect(queueService.add).not.toHaveBeenCalled(); + }); + }); + + describe('Job 不存在', () => { + it('AiJob 已删除 → 永久 failed', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent()); + prisma.aiJob.findUnique.mockResolvedValue(null); // Job not found + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + expect(outboxRepo.markFailed).toHaveBeenCalledWith( + 'evt-001', + 'JOB_NOT_FOUND', + expect.any(String), + ); + }); + }); + + describe('非法 queueName', () => { + it('queueName 不在允许列表 → 永久 failed', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent()); + prisma.aiJob.findUnique.mockResolvedValue({ + id: 'job-001', + queueName: 'ai-analysis', // 旧队列，不允许 + lifecycleStatus: 'queued', + }); + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + expect(outboxRepo.markFailed).toHaveBeenCalledWith( + 'evt-001', + 'ILLEGAL_QUEUE_NAME', + expect.stringContaining('ai-analysis'), + ); + }); + }); + + describe('BullMQ 入队失败回退', () => { + it('入队抛错 → 回退到 pending（未达最大尝试）', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ attemptCount: 0 })); + prisma.aiJob.findUnique.mockResolvedValue({ + id: 'job-001', + queueName: 'ai-interactive', + lifecycleStatus: 'queued', + }); + queueService.add.mockRejectedValue(new Error('Redis connection lost')); + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + + // 应重置为 pending（退避重试） + expect(prisma.outboxEvent.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: 'evt-001' }, + data: expect.objectContaining({ + status: 'pending', + attemptCount: { increment: 1 }, + }), + }), + ); + }); + + it('入队抛错 → 已达最大尝试 → 永久 failed', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent()]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ attemptCount: 2 })); // 已达 2 次 + prisma.aiJob.findUnique.mockResolvedValue({ + id: 'job-001', + queueName: 'ai-interactive', + lifecycleStatus: 'queued', + }); + queueService.add.mockRejectedValue(new Error('Redis down')); + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + + // 应调用 markFailed（永久） + expect(outboxRepo.markFailed).toHaveBeenCalledWith( + 'evt-001', + expect.any(String), + expect.any(String), + ); + }); + }); + + describe('空轮询', () => { + it('无可分发事件 → 返回全零', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([]); + + const result = await dispatcher.cycle(); + expect(result).toEqual({ processed: 0, published: 0, failed: 0, skipped: 0 }); + }); + }); + + describe('无效 payload', () => { + it('payload 缺少 jobId → 永久 failed', async () => { + outboxRepo.releaseExpiredLocks.mockResolvedValue(0); + outboxRepo.findDispatchable.mockResolvedValue([makeEvent({ payload: {} })]); + outboxRepo.markProcessing.mockResolvedValue(makeEvent({ payload: {} })); + + const result = await dispatcher.cycle(); + expect(result.failed).toBe(1); + expect(outboxRepo.markFailed).toHaveBeenCalledWith( + 'evt-001', + 'INVALID_PAYLOAD', + expect.any(String), + ); + }); + }); +}); diff --git a/src/modules/ai-job/outbox-dispatcher.service.ts b/src/modules/ai-job/outbox-dispatcher.service.ts new file mode 100644 index 0000000..ba3cb50 --- /dev/null +++ b/src/modules/ai-job/outbox-dispatcher.service.ts @@ -0,0 +1,206 @@ +import { Injectable, Logger, OnModuleInit, OnModuleDestroy } from '@nestjs/common'; +import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository'; +import { QueueService } from '../../infrastructure/queue/queue.service'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; + +/** + * M-AI-03-07: Outbox Dispatcher + * + * 将 OutboxEvent（eventType = 'ai.job.enqueue'）可靠投递到 BullMQ。 + * + * 运行位置：仅 Worker Process（API 不启动 Dispatcher）。 + * + * 算法（ADR-003 §4.3 — 先 CAS 后投递）： + * 1. findDispatchable(N) — 读取 pending 事件 + * 2. FOR EACH: + * a. markProcessing (CAS: pending → processing) + * b. IF locked IS NULL → SKIP（被其他 Dispatcher 抢走） + * c. 验证 eventType、加载 AiJob、验证 queueName + * d. queue.add() — BullMQ jobId = AiJob.id + * e. markPublished / markFailed + * + * 崩溃窗口保护： + * BullMQ jobId = AiJob.id → 相同 jobId 重复 add 视为 no-op。 + * markProcessing 锁超时（60s）→ releaseExpiredLocks 重置为 pending。 + */ + +const DISPATCH_INTERVAL_MS = 5_000; // 每 5s 轮询 +const DISPATCH_BATCH_SIZE = 10; // 每次最多领取 10 个 +const LOCK_TIMEOUT_MS = 60_000; // processing 锁超时 +const MAX_ATTEMPTS = 3; + +@Injectable() +export class OutboxDispatcher implements OnModuleInit, OnModuleDestroy { + private readonly logger = new Logger(OutboxDispatcher.name); + private readonly instanceId = `dispatcher-${process.pid}-${Date.now()}`; + private timer: ReturnType | null = null; + private running = false; + + constructor( + private readonly outboxRepo: OutboxRepository, + private readonly queueService: QueueService, + private readonly prisma: PrismaService, + ) {} + + async onModuleInit(): Promise { + this.logger.log(`OutboxDispatcher starting, instance=${this.instanceId}`); + // 启动时立即运行一次，然后定时轮询 + await this.cycle().catch((err) => + this.logger.error(`Initial dispatch cycle failed`, err?.stack), + ); + this.timer = setInterval( + () => this.cycle().catch((err) => this.logger.error(`Dispatch cycle failed`, err?.stack)), + DISPATCH_INTERVAL_MS, + ); + } + + onModuleDestroy(): void { + if (this.timer) { + clearInterval(this.timer); + this.timer = null; + } + this.logger.log(`OutboxDispatcher stopped, instance=${this.instanceId}`); + } + + /** + * 单次调度周期。 + */ + async cycle(): Promise<{ processed: number; published: number; failed: number; skipped: number }> { + if (this.running) { + // 防止并发周期重叠（上一批尚未处理完） + return { processed: 0, published: 0, failed: 0, skipped: 0 }; + } + this.running = true; + + let published = 0; + let failed = 0; + let skipped = 0; + let events: any[] = []; + + try { + // 1. 释放过期锁（先清理上一轮崩溃的 processing 事件） + const released = await this.outboxRepo.releaseExpiredLocks(LOCK_TIMEOUT_MS); + if (released > 0) { + this.logger.warn(`Released ${released} expired lock(s)`); + } + + // 2. 查找可分发事件 + events = await this.outboxRepo.findDispatchable(DISPATCH_BATCH_SIZE); + + if (events.length === 0) { + this.running = false; + return { processed: 0, published: 0, failed: 0, skipped: 0 }; + } + + // 3. 逐个处理 + for (const event of events) { + try { + const result = await this.dispatchOne(event.id); + if (result === 'published') published++; + else if (result === 'failed') failed++; + else skipped++; // CAS locked by another dispatcher + } catch (err: any) { + this.logger.error(`Unexpected error dispatching event=${event.id}: ${err.message}`, err?.stack); + failed++; + await this.outboxRepo.markFailed(event.id, 'DISPATCH_INTERNAL', err.message?.slice(0, 500)).catch(() => {}); + } + } + } finally { + this.running = false; + } + + if (published > 0 || failed > 0) { + this.logger.log( + `Dispatch cycle: ${events.length} found, ${published} published, ` + + `${failed} failed, ${skipped} skipped`, + ); + } + + return { processed: published + failed + skipped, published, failed, skipped }; + } + + // ═══════════════════════════════════════════════════════════ + // 单个事件分发 + // ═══════════════════════════════════════════════════════════ + + private async dispatchOne(eventId: string): Promise<'published' | 'failed' | 'skipped'> { + // Step 1: CAS 抢锁（先 CAS 后投递 — ADR-003 §4.3 修正） + const locked = await this.outboxRepo.markProcessing(eventId, this.instanceId); + if (!locked) return 'skipped'; // 被其他 Dispatcher 抢走 + + try { + // Step 2: 验证 eventType + if (locked.eventType !== 'ai.job.enqueue') { + this.logger.warn(`Unknown eventType="${locked.eventType}" for event=${eventId} — marking failed`); + await this.outboxRepo.markFailed(eventId, 'UNKNOWN_EVENT_TYPE', `Unsupported eventType: ${locked.eventType}`); + return 'failed'; + } + + // Step 3: 加载并验证 AiJob + const payload = locked.payload as any; + const jobId = payload?.jobId; + if (!jobId) { + await this.outboxRepo.markFailed(eventId, 'INVALID_PAYLOAD', 'Missing jobId in Outbox payload'); + return 'failed'; + } + + const job = await this.prisma.aiJob.findUnique({ where: { id: jobId } }); + if (!job) { + await this.outboxRepo.markFailed(eventId, 'JOB_NOT_FOUND', `AiJob ${jobId} not found`); + return 'failed'; + } + + // Step 4: 验证 Job 状态（因幂等检查可能已处于终态，仍应投递） + // 若 Job 已不在 queued，不拒绝（幂等场景下可能已处理过） + + // Step 5: 验证 queueName + const allowedQueues = ['ai-interactive', 'ai-background']; + if (job.queueName && !allowedQueues.includes(job.queueName)) { + await this.outboxRepo.markFailed(eventId, 'ILLEGAL_QUEUE_NAME', `Queue "${job.queueName}" not allowed`); + return 'failed'; + } + + const targetQueue = job.queueName || 'ai-interactive'; + + // Step 6: 投递 BullMQ（jobId = AiJob.id 保证幂等） + // 入队失败 → 抛出，由外层 catch 统一处理退避/重试 + await this.queueService.add(targetQueue, { jobId: job.id }, { jobId: job.id }); + + // Step 7: 标记 published + await this.outboxRepo.markPublished(eventId); + + this.logger.log(`Published: event=${eventId} job=${job.id} → queue=${targetQueue}`); + return 'published'; + } catch (err: any) { + const errorCode = err?.code || 'DISPATCH_ERROR'; + const message = err?.message?.slice(0, 500) || 'Unknown error'; + const currentAttempt = (locked.attemptCount || 0) + 1; + + if (currentAttempt >= MAX_ATTEMPTS) { + // 永久失败 + await this.outboxRepo.markFailed(eventId, errorCode, message); + this.logger.error(`Event ${eventId} permanently failed after ${currentAttempt} attempts`); + return 'failed'; + } + + // 暂时失败：回退到 pending（exponential backoff），不经过 markFailed + await this.prisma.outboxEvent.update({ + where: { id: eventId }, + data: { + status: 'pending', + availableAt: new Date(Date.now() + Math.pow(2, currentAttempt) * 1000), + lockedAt: null, + lockedBy: null, + attemptCount: { increment: 1 }, + lastErrorCode: errorCode, + lastErrorMessage: message, + }, + }); + this.logger.warn( + `Event ${eventId} transient failure: attempt=${currentAttempt}/${MAX_ATTEMPTS}, ` + + `backoff=${Math.pow(2, currentAttempt)}s, error=${errorCode}`, + ); + return 'failed'; // 本轮失败，将重试 + } + } +} diff --git a/src/worker.module.ts b/src/worker.module.ts index a4669bb..bdd450d 100644 --- a/src/worker.module.ts +++ b/src/worker.module.ts @@ -18,6 +18,7 @@ import { AiBackgroundJobWorker } from './workers/ai-background.worker'; import { AuditLogProcessor } from './modules/admin-audit-log/audit-log.processor'; import { FileCleanupProcessor } from './modules/files/file-cleanup.processor'; import { WorkerHeartbeatService } from './workers/worker-heartbeat.service'; +import { OutboxDispatcher } from './modules/ai-job/outbox-dispatcher.service'; import { AiAnalysisModule } from './modules/ai-analysis/ai-analysis.module'; import { DocumentImportModule } from './modules/document-import/document-import.module'; @@ -81,6 +82,7 @@ import appleConfig from './config/apple.config'; AuditLogProcessor, FileCleanupProcessor, WorkerHeartbeatService, + OutboxDispatcher, ], }) export class WorkerModule {}