diff --git a/src/modules/ai-job/ai-job-execution-engine.spec.ts b/src/modules/ai-job/ai-job-execution-engine.spec.ts index 0886273..cf0aab4 100644 --- a/src/modules/ai-job/ai-job-execution-engine.spec.ts +++ b/src/modules/ai-job/ai-job-execution-engine.spec.ts @@ -166,14 +166,15 @@ describe('AiJobExecutionEngineImpl', () => { expect(lifecycleRepo.markFailed).not.toHaveBeenCalled(); }); - it('schema validation → permanent → markFailed', async () => { + it('schema validation → permanent → markFailed(不抛错,BullMQ 不再重试)', async () => { prisma.aiJob.findUnique.mockResolvedValue(makeJob()); - lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running' })); + lifecycleRepo.lockJob.mockResolvedValue(makeJob({ lifecycleStatus: 'running', attemptCount: 1 })); prisma.aiJobSnapshot.findUnique.mockResolvedValue({ schemaVersion: '1.0', content: {} }); aiGateway.generate.mockRejectedValue(new Error('schema validation failed')); - await expect(engine.execute('job-001', ctx)).rejects.toThrow(); + // 永久错误:不应抛出(避免 BullMQ retry) + await engine.execute('job-001', ctx); expect(lifecycleRepo.markFailed).toHaveBeenCalledWith( 'job-001', expect.objectContaining({ errorCode: 'schema_validation_failed' }), diff --git a/src/modules/ai-job/ai-job-execution-engine.ts b/src/modules/ai-job/ai-job-execution-engine.ts index 8490175..6b5d1c4 100644 --- a/src/modules/ai-job/ai-job-execution-engine.ts +++ b/src/modules/ai-job/ai-job-execution-engine.ts @@ -230,22 +230,19 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { ); if (classified.retryable) { - // 重试:抛给 BullMQ(不调用 markFailed — BullMQ 会重新投递) + // 重试:先解锁回 queued(BullMQ retry → lockJob 可再次抢锁),然后抛给 BullMQ + await this.unlockForRetry(aiJobId); throw execErr; } - // 永久错误 - const currentAttempt = (lockedJob.attemptCount || 0) + 1; - if (currentAttempt >= (def.execution.maxRetries + 1)) { - await this.lifecycleRepo.markFailed(aiJobId, { - errorCode: classified.errorCode, - publicErrorMessage: classified.publicMessage, - internalErrorMessage: execErr.message?.slice(0, 500), - }); - } else { - // 还有重试配额 → 抛给 BullMQ - throw execErr; - } + // 永久错误:无论 attemptCount 多少,直接 markFailed(不重试) + await this.lifecycleRepo.markFailed(aiJobId, { + errorCode: classified.errorCode, + publicErrorMessage: classified.publicMessage, + internalErrorMessage: execErr.message?.slice(0, 500), + }); + // markFailed 后不抛错 → BullMQ 认为 job 完成(不再重试) + return; } } catch (outerErr: any) { // 捕获 RESOLVE 阶段的错误 @@ -266,6 +263,24 @@ export class AiJobExecutionEngineImpl implements AiJobExecutionEngine { // ── helpers ── + /** + * 解锁 Job 回 queued 状态,供 BullMQ 重试。 + * 保持 attemptCount(已在 lockJob 中原子递增)。 + */ + private async unlockForRetry(jobId: string): Promise { + // 仅持有锁的 Worker 调用 — 无并发竞争,update 即可。 + await this.prisma.aiJob.update({ + where: { id: jobId }, + data: { + lifecycleStatus: 'queued', + status: 'pending', + lockedBy: null, + lockedAt: null, + lockUntil: null, + }, + }); + } + private computeHash(content: string): string { const crypto = require('crypto'); return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16); diff --git a/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts b/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts index f8851d6..2bcb939 100644 --- a/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts +++ b/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts @@ -178,6 +178,7 @@ describe('AiJobLifecycleRepository', () => { expect(callArgs.data.lockedBy).toBe('worker-1'); expect(callArgs.data.startedAt).toBeTruthy(); expect(callArgs.data.lockUntil).toBeTruthy(); + expect(callArgs.data.attemptCount).toEqual({ increment: 1 }); // 每次 lockJob 原子递增 expect(result.lifecycleStatus).toBe('running'); }); diff --git a/src/modules/ai-job/ai-job-lifecycle.repository.ts b/src/modules/ai-job/ai-job-lifecycle.repository.ts index d35b44c..8b3a11d 100644 --- a/src/modules/ai-job/ai-job-lifecycle.repository.ts +++ b/src/modules/ai-job/ai-job-lifecycle.repository.ts @@ -144,6 +144,7 @@ export class AiJobLifecycleRepository { lockedAt: now, lockUntil, startedAt: now, + attemptCount: { increment: 1 }, // 原子递增,每次 lockJob 真实执行次数 +1 }, });