feat: M-AI-03 AiJob 状态机与原子状态仓储
- 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 <noreply@anthropic.com>
This commit is contained in:
wangdl
parent
8b6c7cef72
commit
b08d27a449
407
src/modules/ai-job/ai-job-lifecycle.repository.spec.ts
Normal file
407
src/modules/ai-job/ai-job-lifecycle.repository.spec.ts
Normal file
@ -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,
|
||||
}),
|
||||
);
|
||||
});
|
||||
});
|
||||
});
|
||||
430
src/modules/ai-job/ai-job-lifecycle.repository.ts
Normal file
430
src/modules/ai-job/ai-job-lifecycle.repository.ts
Normal file
@ -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<any> {
|
||||
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<any> {
|
||||
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<any> {
|
||||
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 } });
|
||||
}
|
||||
}
|
||||
104
src/modules/ai-job/ai-job-state-machine.ts
Normal file
104
src/modules/ai-job/ai-job-state-machine.ts
Normal file
@ -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<LifecycleStatus> = new Set([
|
||||
'succeeded',
|
||||
'failed',
|
||||
'cancelled',
|
||||
]);
|
||||
|
||||
/** 合法状态迁移（from → to） */
|
||||
const ALLOWED_TRANSITIONS: Record<LifecycleStatus, ReadonlySet<LifecycleStatus>> = {
|
||||
queued: new Set<LifecycleStatus>(['running', 'cancelled']),
|
||||
running: new Set<LifecycleStatus>(['succeeded', 'failed', 'cancelled']),
|
||||
succeeded: new Set<LifecycleStatus>([]), // 终态
|
||||
failed: new Set<LifecycleStatus>([]), // 终态
|
||||
cancelled: new Set<LifecycleStatus>([]), // 终态
|
||||
};
|
||||
|
||||
/**
|
||||
* ── 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<LifecycleStatus, string> = {
|
||||
queued: 'pending',
|
||||
running: 'processing',
|
||||
succeeded: 'completed',
|
||||
failed: 'failed',
|
||||
cancelled: 'failed', // cancelled 在旧 status 中不存在，映射为 failed + errorCode='CANCELLED'
|
||||
};
|
||||
|
||||
export const STATUS_TO_LIFECYCLE: Record<string, LifecycleStatus> = {
|
||||
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');
|
||||
}
|
||||
}
|
||||
58
src/modules/ai-job/ai-job.errors.ts
Normal file
58
src/modules/ai-job/ai-job.errors.ts
Normal file
@ -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';
|
||||
}
|
||||
}
|
||||
22
src/modules/ai-job/ai-job.module.ts
Normal file
22
src/modules/ai-job/ai-job.module.ts
Normal file
@ -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 {}
|
||||
Loading…
x
Reference in New Issue
Block a user