From b08d27a449be50807e961ba90acd110196b14057 Mon Sep 17 00:00:00 2001 From: wangdl Date: Sat, 20 Jun 2026 17:17:12 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20M-AI-03=20AiJob=20=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E6=9C=BA=E4=B8=8E=E5=8E=9F=E5=AD=90=E7=8A=B6=E6=80=81=E4=BB=93?= =?UTF-8?q?=E5=82=A8?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - AiJobStateMachine: 5 状态（queued/running/succeeded/failed/cancelled）、 合法/非法迁移验证、终态判断、双向 Shadow Write 映射 - AiJobLifecycleRepository: 唯一 lifecycleStatus 写入入口、 CAS（条件 UPDATE WHERE lifecycleStatus IN (...)）抢锁/续锁/标记成功/失败/取消、 可选 tx 参数支持 Projector 事务内调用 - requestCancellation: queued 直接取消 → cancelled、 running 设置 cancelRequestedAt 信号 - 6 个类型化错误（JobLockConflict/JobNotActive/JobAlreadyTerminal/ JobNotCancellable/JobExtendLock/IllegalStateTransition） - 41 个单元测试全部通过 Co-Authored-By: Claude --- .../ai-job-lifecycle.repository.spec.ts | 407 +++++++++++++++++ .../ai-job/ai-job-lifecycle.repository.ts | 430 ++++++++++++++++++ src/modules/ai-job/ai-job-state-machine.ts | 104 +++++ src/modules/ai-job/ai-job.errors.ts | 58 +++ src/modules/ai-job/ai-job.module.ts | 22 + 5 files changed, 1021 insertions(+) create mode 100644 src/modules/ai-job/ai-job-lifecycle.repository.spec.ts create mode 100644 src/modules/ai-job/ai-job-lifecycle.repository.ts create mode 100644 src/modules/ai-job/ai-job-state-machine.ts create mode 100644 src/modules/ai-job/ai-job.errors.ts create mode 100644 src/modules/ai-job/ai-job.module.ts diff --git a/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts b/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts new file mode 100644 index 0000000..6473c63 --- /dev/null +++ b/src/modules/ai-job/ai-job-lifecycle.repository.spec.ts @@ -0,0 +1,407 @@ +import { Test, TestingModule } from '@nestjs/testing'; +import { AiJobStateMachine, LifecycleStatus } from './ai-job-state-machine'; +import { AiJobLifecycleRepository, CreateJobInput } from './ai-job-lifecycle.repository'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { + JobLockConflictError, + JobNotActiveError, + JobAlreadyTerminalError, + JobNotCancellableError, + JobExtendLockError, + IllegalStateTransitionError, +} from './ai-job.errors'; + +describe('AiJobStateMachine', () => { + let sm: AiJobStateMachine; + + beforeEach(() => { + sm = new AiJobStateMachine(); + }); + + describe('合法迁移', () => { + it.each([ + ['queued', 'running'], + ['queued', 'cancelled'], + ['running', 'succeeded'], + ['running', 'failed'], + ['running', 'cancelled'], + ])('%s → %s', (from, to) => { + expect(() => sm.validate(from as LifecycleStatus, to as LifecycleStatus)).not.toThrow(); + }); + }); + + describe('非法迁移', () => { + it.each([ + ['queued', 'succeeded'], + ['queued', 'failed'], + ['succeeded', 'running'], + ['succeeded', 'failed'], + ['succeeded', 'cancelled'], + ['failed', 'running'], + ['failed', 'succeeded'], + ['failed', 'cancelled'], + ['cancelled', 'running'], + ['cancelled', 'succeeded'], + ['cancelled', 'failed'], + ['running', 'queued'], + ])('%s → %s 抛出 IllegalStateTransitionError', (from, to) => { + expect(() => sm.validateForJob('test-job', from as LifecycleStatus, to as LifecycleStatus)) + .toThrow(IllegalStateTransitionError); + }); + }); + + describe('终态判断', () => { + it.each(['succeeded', 'failed', 'cancelled'] as LifecycleStatus[])('%s 是终态', (s) => { + expect(sm.isTerminal(s)).toBe(true); + }); + + it.each(['queued', 'running'] as LifecycleStatus[])('%s 不是终态', (s) => { + expect(sm.isTerminal(s)).toBe(false); + }); + }); + + describe('parse', () => { + it('有效字符串返回 LifecycleStatus', () => { + expect(sm.parse('queued')).toBe('queued'); + expect(sm.parse('running')).toBe('running'); + expect(sm.parse('succeeded')).toBe('succeeded'); + }); + + it('null/undefined 返回 queued', () => { + expect(sm.parse(null)).toBe('queued'); + expect(sm.parse(undefined)).toBe('queued'); + }); + + it('无效字符串抛出错误', () => { + expect(() => sm.parse('invalid')).toThrow(IllegalStateTransitionError); + }); + }); +}); + +describe('AiJobLifecycleRepository', () => { + let repo: AiJobLifecycleRepository; + let prisma: PrismaService; + let tx: any; // mock transaction client + + // 伪造 Prisma TransactionClient + const makeMockTx = () => ({ + aiJob: { + create: jest.fn(), + updateMany: jest.fn(), + update: jest.fn(), + findUnique: jest.fn(), + findUniqueOrThrow: jest.fn(), + findMany: jest.fn(), + }, + }); + + beforeEach(async () => { + const txMock = makeMockTx(); + + const mockPrisma = { + aiJob: { + create: jest.fn(), + updateMany: jest.fn(), + update: jest.fn(), + findUnique: jest.fn(), + findUniqueOrThrow: jest.fn(), + findMany: jest.fn(), + }, + $transaction: jest.fn(), + }; + + const module: TestingModule = await Test.createTestingModule({ + providers: [ + AiJobStateMachine, + AiJobLifecycleRepository, + { provide: PrismaService, useValue: mockPrisma }, + ], + }).compile(); + + repo = module.get(AiJobLifecycleRepository); + prisma = module.get(PrismaService); + tx = txMock; + }); + + describe('createJob', () => { + it('在事务中创建 queued 状态的 Job，含 Shadow Write', async () => { + const input: CreateJobInput = { + userId: 'user-1', + jobType: 'test_job', + triggerType: 'user_api', + }; + const createdJob = { + id: 'job-001', + lifecycleStatus: 'queued', + status: 'pending', + userId: 'user-1', + jobType: 'test_job', + queuedAt: new Date(), + attemptCount: 0, + progress: 0, + }; + tx.aiJob.create.mockResolvedValue(createdJob); + + const result = await repo.createJob(tx, input); + expect(tx.aiJob.create).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + lifecycleStatus: 'queued', + status: 'pending', // Shadow Write + queuedAt: expect.any(Date), + }), + }), + ); + expect(result.lifecycleStatus).toBe('queued'); + expect(result.status).toBe('pending'); + }); + }); + + describe('lockJob', () => { + it('CAS 抢锁成功：queued → running，含 Shadow Write', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'running', + status: 'processing', + lockedBy: 'worker-1', + lockedAt: expect.any(Date), + lockUntil: expect.any(Date), + }); + + const result = await repo.lockJob('job-001', 'worker-1'); + + const callArgs = (prisma.aiJob.updateMany as jest.Mock).mock.calls[0][0]; + expect(callArgs.where).toEqual({ id: 'job-001', lifecycleStatus: 'queued' }); + expect(callArgs.data.lifecycleStatus).toBe('running'); + expect(callArgs.data.status).toBe('processing'); // Shadow Write + expect(callArgs.data.lockedBy).toBe('worker-1'); + expect(callArgs.data.startedAt).toBeTruthy(); + expect(callArgs.data.lockUntil).toBeTruthy(); + expect(result.lifecycleStatus).toBe('running'); + }); + + it('CAS 抢锁失败（已进入终态）抛出 JobAlreadyTerminalError', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ + lifecycleStatus: 'succeeded', + }); + + await expect(repo.lockJob('job-001', 'worker-1')).rejects.toThrow(JobAlreadyTerminalError); + }); + + it('CAS 抢锁失败（已被其他 Worker 锁定）抛出 JobLockConflictError', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ + lifecycleStatus: 'running', + }); + + await expect(repo.lockJob('job-001', 'worker-1')).rejects.toThrow(JobLockConflictError); + }); + }); + + describe('extendLock', () => { + it('续约成功返回 cancelRequested=false', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'running', + cancelRequestedAt: null, + }); + + const before = Date.now(); + const result = await repo.extendLock('job-001', 'worker-1'); + const after = Date.now(); + + expect(result.cancelRequested).toBe(false); + // lockUntil = now + 60000ms，允许 ±1000ms 的时钟漂移 + expect(result.lockUntil).toBeGreaterThanOrEqual(before + 60000 - 1000); + expect(result.lockUntil).toBeLessThanOrEqual(after + 60000 + 1000); + }); + + it('续约成功返回 cancelRequested=true', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'running', + cancelRequestedAt: new Date(), + }); + + const result = await repo.extendLock('job-001', 'worker-1'); + expect(result.cancelRequested).toBe(true); + }); + + it('续约失败（lockedBy 不匹配）抛出 JobExtendLockError', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ + lifecycleStatus: 'running', + lockedBy: 'worker-2', + }); + + await expect(repo.extendLock('job-001', 'worker-1')).rejects.toThrow(JobExtendLockError); + }); + }); + + describe('markSucceeded', () => { + it('CAS 标记成功：running → succeeded，清除错误和时间字段', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'succeeded', + status: 'completed', + finishedAt: new Date(), + }); + + const result = await repo.markSucceeded('job-001'); + + // 验证 CAS where 条件 + const callArgs = (prisma.aiJob.updateMany as jest.Mock).mock.calls[0][0]; + expect(callArgs.where).toEqual({ id: 'job-001', lifecycleStatus: 'running' }); + expect(callArgs.data.lifecycleStatus).toBe('succeeded'); + expect(callArgs.data.status).toBe('completed'); // Shadow Write + expect(callArgs.data.finishedAt).toBeTruthy(); + expect(callArgs.data.errorCode).toBeNull(); // 成功时清除错误 + expect(callArgs.data.lockedBy).toBeNull(); // 清除运行时字段 + expect(result.lifecycleStatus).toBe('succeeded'); + }); + + it('CAS 失败（已终态）抛出 JobAlreadyTerminalError', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ + lifecycleStatus: 'failed', + }); + + await expect(repo.markSucceeded('job-001')).rejects.toThrow(JobAlreadyTerminalError); + }); + + it('接受可选 tx 参数，复用调用方事务', async () => { + tx.aiJob.updateMany.mockResolvedValue({ count: 1 }); + tx.aiJob.findUniqueOrThrow.mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'succeeded', + }); + + await repo.markSucceeded('job-001', tx); + expect(tx.aiJob.updateMany).toHaveBeenCalled(); + // 不应该调用 this.prisma + }); + }); + + describe('markFailed', () => { + it('CAS 标记失败：running → failed，含错误信息', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'failed', + status: 'failed', + errorCode: 'PROVIDER_TIMEOUT', + }); + + const result = await repo.markFailed('job-001', { + errorCode: 'PROVIDER_TIMEOUT', + publicErrorMessage: 'AI 服务暂时不可用', + internalErrorMessage: 'DeepSeek API timeout after 30s', + }); + + expect(prisma.aiJob.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + data: expect.objectContaining({ + lifecycleStatus: 'failed', + status: 'failed', + errorCode: 'PROVIDER_TIMEOUT', + publicErrorMessage: 'AI 服务暂时不可用', + internalErrorMessage: 'DeepSeek API timeout after 30s', + }), + }), + ); + expect(result.lifecycleStatus).toBe('failed'); + }); + }); + + describe('markCancelled', () => { + it('从 queued 取消：queued → cancelled', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'cancelled', + status: 'failed', + errorCode: 'CANCELLED', + finishedAt: new Date(), + }); + + const result = await repo.markCancelled('job-001'); + expect(prisma.aiJob.updateMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: expect.objectContaining({ + lifecycleStatus: { in: ['queued', 'running'] }, + }), + data: expect.objectContaining({ + lifecycleStatus: 'cancelled', + status: 'failed', // Shadow Write + errorCode: 'CANCELLED', + }), + }), + ); + expect(result.errorCode).toBe('CANCELLED'); + }); + + it('取消已终态的 Job 抛出 JobNotCancellableError', async () => { + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 0 }); + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ + lifecycleStatus: 'succeeded', + }); + + await expect(repo.markCancelled('job-001')).rejects.toThrow(JobNotCancellableError); + }); + }); + + describe('requestCancellation', () => { + it('queued Job 直接取消', async () => { + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ lifecycleStatus: 'queued' }); + (prisma.aiJob.updateMany as jest.Mock).mockResolvedValue({ count: 1 }); + (prisma.aiJob.findUniqueOrThrow as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'cancelled', + errorCode: 'CANCELLED', + }); + + const result = await repo.requestCancellation('job-001'); + expect(result.status).toBe('cancelled'); + }); + + it('running Job 设置 cancelRequestedAt 信号', async () => { + (prisma.aiJob.findUnique as jest.Mock).mockResolvedValue({ lifecycleStatus: 'running' }); + (prisma.aiJob.update as jest.Mock).mockResolvedValue({ + id: 'job-001', + lifecycleStatus: 'running', + cancelRequestedAt: new Date(), + }); + + const result = await repo.requestCancellation('job-001'); + expect(result.status).toBe('cancel_requested'); + expect(prisma.aiJob.update).toHaveBeenCalledWith( + expect.objectContaining({ + where: { id: 'job-001', lifecycleStatus: 'running' }, + }), + ); + }); + }); + + describe('findByLifecycleStatus', () => { + it('按状态过滤并按优先级/创建时间排序', async () => { + (prisma.aiJob.findMany as jest.Mock).mockResolvedValue([]); + + await repo.findByLifecycleStatus(['queued'], { queueName: 'ai-interactive', limit: 10 }); + expect(prisma.aiJob.findMany).toHaveBeenCalledWith( + expect.objectContaining({ + where: { + lifecycleStatus: { in: ['queued'] }, + queueName: 'ai-interactive', + }, + orderBy: [{ priority: 'asc' }, { createdAt: 'asc' }], + take: 10, + }), + ); + }); + }); +}); diff --git a/src/modules/ai-job/ai-job-lifecycle.repository.ts b/src/modules/ai-job/ai-job-lifecycle.repository.ts new file mode 100644 index 0000000..15fcaf8 --- /dev/null +++ b/src/modules/ai-job/ai-job-lifecycle.repository.ts @@ -0,0 +1,430 @@ +import { Injectable } from '@nestjs/common'; +import { PrismaService } from '../../infrastructure/database/prisma.service'; +import { Prisma } from '@prisma/client'; +import { + JobLockConflictError, + JobNotActiveError, + JobAlreadyTerminalError, + JobNotCancellableError, + JobExtendLockError, +} from './ai-job.errors'; +import { + AiJobStateMachine, + LifecycleStatus, + LIFECYCLE_TO_STATUS, +} from './ai-job-state-machine'; + +/** + * ── Job 创建输入 ── + */ +export interface CreateJobInput { + userId: string; + jobType: string; + triggerType?: string; + queueName?: string; + targetType?: string; + targetId?: string; + parentJobId?: string; + idempotencyKey?: string; + priority?: number; + credentialMode?: string; + credentialId?: string; + modelProvider?: string; + modelName?: string; + promptVersion?: string; + outputSchemaVersion?: string; + inputSchemaVersion?: string; + inputRef?: string; + maxAttempts?: number; + timeoutMs?: number; +} + +/** + * ── Job 错误信息 ── + */ +export interface JobErrorInfo { + errorCode: string; + publicErrorMessage?: string; + internalErrorMessage?: string; +} + +/** + * M-AI-03-03: AiJobLifecycleRepository + * + * 统一 lifecycleStatus 的唯一写入入口。 + * + * 约束： + * - 所有状态变更使用 CAS（条件 UPDATE WHERE lifecycleStatus IN (...)） + * - 禁止先读后无条件 update + * - 可选 tx 参数支持在外部 Prisma Transaction 中复用事务 + * - lockJob 不使用 tx（CAS 必须立即对其他 Worker 可见） + */ +@Injectable() +export class AiJobLifecycleRepository { + private readonly LOCK_TTL_MS = 60_000; // ADR-003：60s + + constructor( + private readonly prisma: PrismaService, + private readonly stateMachine: AiJobStateMachine, + ) {} + + // ═══════════════════════════════════════════════════════════ + // 创建 + // ═══════════════════════════════════════════════════════════ + + /** + * 在给定事务中创建 Job（lifecycleStatus = 'queued'）。 + * 调用方负责在同一 tx 中写入 OutboxEvent。 + */ + async createJob( + tx: Prisma.TransactionClient, + input: CreateJobInput, + ) { + const job = await tx.aiJob.create({ + data: { + userId: input.userId, + jobType: input.jobType, + triggerType: input.triggerType ?? 'user_api', + queueName: input.queueName ?? 'ai-interactive', + targetType: input.targetType ?? null, + targetId: input.targetId ?? null, + parentJobId: input.parentJobId ?? null, + idempotencyKey: input.idempotencyKey ?? null, + priority: input.priority ?? 0, + credentialMode: input.credentialMode ?? 'platform_key', + credentialId: input.credentialId ?? null, + modelProvider: input.modelProvider ?? 'deepseek', + modelName: input.modelName ?? 'deepseek-chat', + promptVersion: input.promptVersion ?? null, + outputSchemaVersion: input.outputSchemaVersion ?? null, + inputSchemaVersion: input.inputSchemaVersion ?? null, + inputRef: input.inputRef ?? null, + maxAttempts: input.maxAttempts ?? 3, + timeoutMs: input.timeoutMs ?? 120000, + // ── 生命周期 ── + lifecycleStatus: 'queued', + status: LIFECYCLE_TO_STATUS['queued'], // Shadow Write: pending + queuedAt: new Date(), + attemptCount: 0, + progress: 0, + }, + }); + + return job; + } + + // ═══════════════════════════════════════════════════════════ + // 抢锁（CAS: queued → running） + // ═══════════════════════════════════════════════════════════ + + /** + * CAS 抢锁。只有 lifecycleStatus = 'queued' 的 Job 才能被锁定。 + * + * 必须是独立操作（不传 tx）—— CAS 需要立即对其他 Worker 可见。 + */ + async lockJob(jobId: string, instanceId: string) { + const now = new Date(); + const lockUntil = new Date(now.getTime() + this.LOCK_TTL_MS); + + // CAS: queued → running，仅当 status 仍为 queued 时成功 + const result = await this.prisma.aiJob.updateMany({ + where: { + id: jobId, + lifecycleStatus: 'queued', + }, + data: { + lifecycleStatus: 'running', + status: LIFECYCLE_TO_STATUS['running'], // Shadow Write: processing + lockedBy: instanceId, + lockedAt: now, + lockUntil, + startedAt: now, + }, + }); + + if (result.count === 0) { + // 抢锁失败 —— 检查原因以提供更精确的错误 + const current = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true }, + }); + if (!current) { + throw new JobLockConflictError(jobId); // Job 不存在 + } + if (this.stateMachine.isTerminal(current.lifecycleStatus as LifecycleStatus)) { + throw new JobAlreadyTerminalError(jobId, current.lifecycleStatus!); + } + if (current.lifecycleStatus === 'running') { + throw new JobLockConflictError(jobId); // 已被其他 Worker 抢走 + } + throw new JobLockConflictError(jobId); + } + + const job = await this.prisma.aiJob.findUniqueOrThrow({ + where: { id: jobId }, + }); + + return job; + } + + // ═══════════════════════════════════════════════════════════ + // 续锁（Heartbeat） + // ═══════════════════════════════════════════════════════════ + + /** + * 续约锁定期。仅当 Job 处于 'running' 状态且 lockedBy 匹配时成功。 + */ + async extendLock(jobId: string, instanceId: string) { + const now = new Date(); + const lockUntil = new Date(now.getTime() + this.LOCK_TTL_MS); + + const result = await this.prisma.aiJob.updateMany({ + where: { + id: jobId, + lifecycleStatus: 'running', + lockedBy: instanceId, + }, + data: { lockUntil }, + }); + + if (result.count === 0) { + const current = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true, lockedBy: true }, + }); + if (!current || current.lifecycleStatus !== 'running') { + throw new JobNotActiveError(jobId, current?.lifecycleStatus ?? 'unknown'); + } + throw new JobExtendLockError(jobId, instanceId); + } + + // 读取当前状态以检查 cancelRequestedAt + const job = await this.prisma.aiJob.findUniqueOrThrow({ + where: { id: jobId }, + }); + + return { + jobId, + lockUntil: lockUntil.getTime(), + cancelRequested: job.cancelRequestedAt != null, + }; + } + + // ═══════════════════════════════════════════════════════════ + // 标记成功 + // ═══════════════════════════════════════════════════════════ + + /** + * 标记 Job 成功。 + * + * @param tx 可选事务客户端。Projector 路径传入以共享事务；非 Projector 路径不传。 + */ + async markSucceeded(jobId: string, tx?: Prisma.TransactionClient): Promise { + const client = tx ?? this.prisma; + const now = new Date(); + + const result = await client.aiJob.updateMany({ + where: { + id: jobId, + lifecycleStatus: 'running', + }, + data: { + lifecycleStatus: 'succeeded', + status: LIFECYCLE_TO_STATUS['succeeded'], // Shadow Write: completed + finishedAt: now, + // 清除运行时字段 + lockedBy: null, + lockedAt: null, + lockUntil: null, + errorCode: null, + publicErrorMessage: null, + internalErrorMessage: null, + }, + }); + + if (result.count === 0) { + const current = await client.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true }, + }); + if (current && this.stateMachine.isTerminal(current.lifecycleStatus as LifecycleStatus)) { + throw new JobAlreadyTerminalError(jobId, current.lifecycleStatus!); + } + throw new JobNotActiveError(jobId, current?.lifecycleStatus ?? 'unknown'); + } + + return client.aiJob.findUniqueOrThrow({ where: { id: jobId } }); + } + + // ═══════════════════════════════════════════════════════════ + // 标记失败 + // ═══════════════════════════════════════════════════════════ + + /** + * 标记 Job 失败（终态）。 + * + * @param tx 可选事务客户端。Engine 非 Projector 错误路径不传。 + */ + async markFailed(jobId: string, error: JobErrorInfo, tx?: Prisma.TransactionClient): Promise { + const client = tx ?? this.prisma; + const now = new Date(); + + const result = await client.aiJob.updateMany({ + where: { + id: jobId, + lifecycleStatus: 'running', + }, + data: { + lifecycleStatus: 'failed', + status: LIFECYCLE_TO_STATUS['failed'], // Shadow Write: failed + finishedAt: now, + lockedBy: null, + lockedAt: null, + lockUntil: null, + errorCode: error.errorCode, + publicErrorMessage: error.publicErrorMessage ?? null, + internalErrorMessage: error.internalErrorMessage ?? null, + }, + }); + + if (result.count === 0) { + const current = await client.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true }, + }); + if (current && this.stateMachine.isTerminal(current.lifecycleStatus as LifecycleStatus)) { + throw new JobAlreadyTerminalError(jobId, current.lifecycleStatus!); + } + throw new JobNotActiveError(jobId, current?.lifecycleStatus ?? 'unknown'); + } + + return client.aiJob.findUniqueOrThrow({ where: { id: jobId } }); + } + + // ═══════════════════════════════════════════════════════════ + // 标记取消 + // ═══════════════════════════════════════════════════════════ + + /** + * 标记 Job 取消。 + * + * queued 状态可直接取消；running 状态需先设置 cancelRequestedAt， + * Engine/Heartbeat 检测后调用此方法。 + * + * @param tx 可选事务客户端。取消路径通常不传（独立操作）。 + */ + async markCancelled(jobId: string, tx?: Prisma.TransactionClient): Promise { + const client = tx ?? this.prisma; + const now = new Date(); + + // 尝试从 queued 或 running 取消 + const result = await client.aiJob.updateMany({ + where: { + id: jobId, + lifecycleStatus: { in: ['queued', 'running'] }, + }, + data: { + lifecycleStatus: 'cancelled', + // ── Shadow Write：cancelled 在旧 status 中不存在，映射为 failed + errorCode=CANCELLED ── + status: LIFECYCLE_TO_STATUS['cancelled'], // 'failed' + finishedAt: now, + lockedBy: null, + lockedAt: null, + lockUntil: null, + errorCode: 'CANCELLED', + publicErrorMessage: 'Job was cancelled', + internalErrorMessage: null, + }, + }); + + if (result.count === 0) { + const current = await client.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true }, + }); + if (!current) { + throw new JobLockConflictError(jobId); // Job 不存在，复用此错误 + } + if (this.stateMachine.isTerminal(current.lifecycleStatus as LifecycleStatus)) { + throw new JobNotCancellableError(jobId, current.lifecycleStatus!); + } + throw new JobNotCancellableError(jobId, current.lifecycleStatus ?? 'unknown'); + } + + return client.aiJob.findUniqueOrThrow({ where: { id: jobId } }); + } + + // ═══════════════════════════════════════════════════════════ + // 设置取消请求（不改变 lifecycleStatus） + // ═══════════════════════════════════════════════════════════ + + /** + * 对 running 状态的 Job 设置 cancelRequestedAt 信号。 + * 不改变 lifecycleStatus — Engine/Heartbeat 检测后调用 markCancelled。 + * + * 对 queued 状态的 Job 直接取消。 + */ + async requestCancellation(jobId: string): Promise<{ status: 'cancelled' | 'cancel_requested' }> { + const job = await this.prisma.aiJob.findUnique({ + where: { id: jobId }, + select: { lifecycleStatus: true }, + }); + + if (!job) throw new JobLockConflictError(jobId); + + const status = this.stateMachine.parse(job.lifecycleStatus); + + if (status === 'queued') { + // 直接取消 + await this.markCancelled(jobId); + return { status: 'cancelled' }; + } + + if (status === 'running') { + // 设置取消信号 + await this.prisma.aiJob.update({ + where: { id: jobId, lifecycleStatus: 'running' }, + data: { cancelRequestedAt: new Date() }, + }); + return { status: 'cancel_requested' }; + } + + // 终态不可取消 + throw new JobNotCancellableError(jobId, status); + } + + // ═══════════════════════════════════════════════════════════ + // 只读查询 + // ═══════════════════════════════════════════════════════════ + + async findByLifecycleStatus( + statuses: LifecycleStatus[], + opts?: { queueName?: string; limit?: number }, + ) { + const where: Prisma.AiJobWhereInput = { + lifecycleStatus: { in: statuses }, + }; + if (opts?.queueName) where.queueName = opts.queueName; + + return this.prisma.aiJob.findMany({ + where, + orderBy: [{ priority: 'asc' }, { createdAt: 'asc' }], + take: opts?.limit ?? 50, + }); + } + + async findExpiredLocks(thresholdMs: number = 90_000) { + const cutoff = new Date(Date.now() - thresholdMs); + return this.prisma.aiJob.findMany({ + where: { + lifecycleStatus: 'running', + lockUntil: { lt: cutoff }, + }, + orderBy: { createdAt: 'asc' }, + }); + } + + async findById(jobId: string) { + return this.prisma.aiJob.findUnique({ where: { id: jobId } }); + } +} diff --git a/src/modules/ai-job/ai-job-state-machine.ts b/src/modules/ai-job/ai-job-state-machine.ts new file mode 100644 index 0000000..6107f33 --- /dev/null +++ b/src/modules/ai-job/ai-job-state-machine.ts @@ -0,0 +1,104 @@ +import { Injectable } from '@nestjs/common'; +import { IllegalStateTransitionError } from './ai-job.errors'; + +/** + * M-AI-03-03: AiJob 统一状态机 + * + * 定义合法状态迁移，所有 lifecycleStatus 变更必须经过此验证。 + * + * 状态机（ADR-003 §1.1）： + * queued → running → succeeded / failed / cancelled + * queued → cancelled（直接取消，无需先 running） + * + * cancel_requested 不是独立状态 — 它是 running 状态上的信号字段（cancelRequestedAt）。 + */ + +export const LIFECYCLE_STATUSES = [ + 'queued', + 'running', + 'succeeded', + 'failed', + 'cancelled', +] as const; + +export type LifecycleStatus = (typeof LIFECYCLE_STATUSES)[number]; + +export const TERMINAL_STATUSES: ReadonlySet = new Set([ + 'succeeded', + 'failed', + 'cancelled', +]); + +/** 合法状态迁移（from → to） */ +const ALLOWED_TRANSITIONS: Record> = { + queued: new Set(['running', 'cancelled']), + running: new Set(['succeeded', 'failed', 'cancelled']), + succeeded: new Set([]), // 终态 + failed: new Set([]), // 终态 + cancelled: new Set([]), // 终态 +}; + +/** + * ── M-AI-02-10 Legacy status Shadow Write ── + * + * 新 lifecycleStatus → 旧 status 列映射。 + * 双向均可查询： + * - STATUS_TO_LIFECYCLE: 旧 → 新（ai-analysis.repository.ts 已在用） + * - LIFECYCLE_TO_STATUS: 新 → 旧（本 Repository Shadow Write 使用） + */ +export const LIFECYCLE_TO_STATUS: Record = { + queued: 'pending', + running: 'processing', + succeeded: 'completed', + failed: 'failed', + cancelled: 'failed', // cancelled 在旧 status 中不存在，映射为 failed + errorCode='CANCELLED' +}; + +export const STATUS_TO_LIFECYCLE: Record = { + pending: 'queued', + processing: 'running', + completed: 'succeeded', + failed: 'failed', + cancelled: 'cancelled', + cancel_requested: 'cancelled', +}; + +@Injectable() +export class AiJobStateMachine { + /** + * 校验 from → to 是否为合法迁移。 + * 合法 → 无返回值；非法 → 抛出 IllegalStateTransitionError。 + */ + validate(from: LifecycleStatus, to: LifecycleStatus): void { + const allowed = ALLOWED_TRANSITIONS[from]; + if (!allowed || !allowed.has(to)) { + throw new IllegalStateTransitionError('unknown', from, to); + } + } + + /** + * 校验带 jobId 的迁移。提供更精确的错误信息。 + */ + validateForJob(jobId: string, from: LifecycleStatus, to: LifecycleStatus): void { + const allowed = ALLOWED_TRANSITIONS[from]; + if (!allowed || !allowed.has(to)) { + throw new IllegalStateTransitionError(jobId, from, to); + } + } + + /** 是否为终态 */ + isTerminal(status: LifecycleStatus): boolean { + return TERMINAL_STATUSES.has(status); + } + + /** + * 对于从数据库读取的状态字符串，校验并转换为类型化状态。 + */ + parse(status: string | null | undefined): LifecycleStatus { + if (!status) return 'queued'; + if ((LIFECYCLE_STATUSES as readonly string[]).includes(status)) { + return status as LifecycleStatus; + } + throw new IllegalStateTransitionError('unknown', status, 'any'); + } +} diff --git a/src/modules/ai-job/ai-job.errors.ts b/src/modules/ai-job/ai-job.errors.ts new file mode 100644 index 0000000..75b05a3 --- /dev/null +++ b/src/modules/ai-job/ai-job.errors.ts @@ -0,0 +1,58 @@ +/** + * M-AI-03-03: AiJob 状态机错误类型 + * + * 所有违反状态机的操作抛出这些类型化错误， + * 调用方可区分"Job 已被抢走"、"Job 已结束"等场景。 + */ + +export class JobLockConflictError extends Error { + public readonly code = 'JOB_LOCK_CONFLICT'; + constructor(public readonly jobId: string) { + super(`Job ${jobId} is not in 'queued' status or already locked by another worker`); + this.name = 'JobLockConflictError'; + } +} + +export class JobNotActiveError extends Error { + public readonly code = 'JOB_NOT_ACTIVE'; + constructor(public readonly jobId: string, public readonly currentStatus: string) { + super(`Job ${jobId} is in '${currentStatus}' status, expected 'running' or 'locked'`); + this.name = 'JobNotActiveError'; + } +} + +export class JobAlreadyTerminalError extends Error { + public readonly code = 'JOB_ALREADY_TERMINAL'; + constructor(public readonly jobId: string, public readonly currentStatus: string) { + super(`Job ${jobId} is already in terminal status '${currentStatus}'`); + this.name = 'JobAlreadyTerminalError'; + } +} + +export class JobNotCancellableError extends Error { + public readonly code = 'JOB_NOT_CANCELLABLE'; + constructor(public readonly jobId: string, public readonly currentStatus: string) { + super(`Job ${jobId} in '${currentStatus}' status cannot be cancelled`); + this.name = 'JobNotCancellableError'; + } +} + +export class JobExtendLockError extends Error { + public readonly code = 'JOB_EXTEND_LOCK_ERROR'; + constructor(public readonly jobId: string, public readonly instanceId: string) { + super(`Cannot extend lock for job ${jobId}: not locked by instance '${instanceId}'`); + this.name = 'JobExtendLockError'; + } +} + +export class IllegalStateTransitionError extends Error { + public readonly code = 'ILLEGAL_STATE_TRANSITION'; + constructor( + public readonly jobId: string, + public readonly from: string, + public readonly to: string, + ) { + super(`Illegal state transition for job ${jobId}: '${from}' → '${to}'`); + this.name = 'IllegalStateTransitionError'; + } +} diff --git a/src/modules/ai-job/ai-job.module.ts b/src/modules/ai-job/ai-job.module.ts new file mode 100644 index 0000000..9b93c02 --- /dev/null +++ b/src/modules/ai-job/ai-job.module.ts @@ -0,0 +1,22 @@ +import { Module } from '@nestjs/common'; +import { PrismaModule } from '../../infrastructure/database/prisma.module'; +import { AiJobStateMachine } from './ai-job-state-machine'; +import { AiJobLifecycleRepository } from './ai-job-lifecycle.repository'; + +/** + * M-AI-03: AiJob 统一 Job Engine 模块 + * + * 当前仅包含状态机与生命周期仓储。 + * 后续 Issue（#287–#293）在此模块中逐步新增： + * - JobDefinitionRegistry + * - AiJobService + * - AiJobExecutionEngine + * - OutboxDispatcher + * - ResultProjector + */ +@Module({ + imports: [PrismaModule], + providers: [AiJobStateMachine, AiJobLifecycleRepository], + exports: [AiJobStateMachine, AiJobLifecycleRepository], +}) +export class AiJobModule {}