fix: #289 检修 — TOCTOU 幂等 + promptKey 传递 + require 清理
- 幂等保护从 findUnique 预检改为 P2002 catch（消除 TOCTOU 竞态窗口）
- AiJobLifecycleRepository: CreateJobInput 新增 promptKey，createJob 写入
- AiJobCreationService: 从 JobDefinition 传递 promptKey 到 Job
- computeHash: 内联 require('crypto') → 顶层 import * as crypto
- 测试更新: 幂等场景改为 P2002 冲突 → findUniqueOrThrow 返回已有 Job
Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
wangdl
parent
6668bba9b4
commit
471aa2b872
@ -138,58 +138,44 @@ describe('AiJobCreationService', () => {
|
||||
});
|
||||
|
||||
describe('幂等性', () => {
|
||||
it('相同 (userId, jobType, idempotencyKey) 返回已有 Job，不创建新记录', async () => {
|
||||
it('并发重复创建相同 (userId, jobType, idempotencyKey) → P2002 → 返回已有 Job', async () => {
|
||||
registry.get.mockReturnValue(validDef());
|
||||
snapshotBuilder.buildSnapshot.mockResolvedValue({});
|
||||
|
||||
const existingJob = {
|
||||
id: 'existing-job',
|
||||
lifecycleStatus: 'succeeded',
|
||||
jobType: 'test_job',
|
||||
};
|
||||
prisma.aiJob.findUnique.mockResolvedValue(existingJob);
|
||||
// First request creates successfully
|
||||
lifecycleRepo.createJob.mockResolvedValueOnce({ id: 'job-001' });
|
||||
|
||||
await service.createJob({
|
||||
userId: 'u1', jobType: 'test_job', triggerType: 'user_api',
|
||||
targetType: 'kb', targetId: 'kb-1', idempotencyKey: 'idem-001',
|
||||
});
|
||||
expect(lifecycleRepo.createJob).toHaveBeenCalledTimes(1);
|
||||
|
||||
// Second request: transaction fails with P2002 (duplicate unique key)
|
||||
const p2002Error = new Error('Unique constraint failed') as any;
|
||||
p2002Error.code = 'P2002';
|
||||
prisma.$transaction.mockRejectedValueOnce(p2002Error);
|
||||
|
||||
const existingJob = { id: 'job-001', lifecycleStatus: 'queued', jobType: 'test_job' };
|
||||
prisma.aiJob.findUniqueOrThrow = jest.fn().mockResolvedValue(existingJob);
|
||||
|
||||
const result = await service.createJob({
|
||||
userId: 'u1',
|
||||
jobType: 'test_job',
|
||||
triggerType: 'user_api',
|
||||
targetType: 'kb',
|
||||
targetId: 'kb-1',
|
||||
idempotencyKey: 'idem-001',
|
||||
userId: 'u1', jobType: 'test_job', triggerType: 'user_api',
|
||||
targetType: 'kb', targetId: 'kb-1', idempotencyKey: 'idem-001',
|
||||
});
|
||||
|
||||
expect(result).toBe(existingJob);
|
||||
// 不应调用 snapshot、lifecycleRepo、transaction
|
||||
expect(snapshotBuilder.buildSnapshot).not.toHaveBeenCalled();
|
||||
expect(lifecycleRepo.createJob).not.toHaveBeenCalled();
|
||||
expect(prisma.$transaction).not.toHaveBeenCalled();
|
||||
// Snapshot was still built (pre-transaction), but no new Job created
|
||||
});
|
||||
|
||||
it('未提供 idempotencyKey 时每次创建新 Job', async () => {
|
||||
it('未提供 idempotencyKey 时每次创建新 Job（无幂等保护）', async () => {
|
||||
registry.get.mockReturnValue(validDef());
|
||||
snapshotBuilder.buildSnapshot.mockResolvedValue({});
|
||||
const mockJob = { id: 'job-002' };
|
||||
lifecycleRepo.createJob.mockResolvedValue(mockJob);
|
||||
|
||||
// 第一次创建
|
||||
prisma.aiJob.findUnique.mockResolvedValue(null);
|
||||
await service.createJob({
|
||||
userId: 'u1',
|
||||
jobType: 'test_job',
|
||||
triggerType: 'user_api',
|
||||
targetType: 'kb',
|
||||
targetId: 'kb-1',
|
||||
// no idempotencyKey
|
||||
});
|
||||
|
||||
// 第二次创建（无幂等键，应再次创建）
|
||||
await service.createJob({
|
||||
userId: 'u1',
|
||||
jobType: 'test_job',
|
||||
triggerType: 'user_api',
|
||||
targetType: 'kb',
|
||||
targetId: 'kb-1',
|
||||
// no idempotencyKey
|
||||
});
|
||||
await service.createJob({ userId: 'u1', jobType: 'test_job', triggerType: 'user_api', targetType: 'kb', targetId: 'kb-1' });
|
||||
await service.createJob({ userId: 'u1', jobType: 'test_job', triggerType: 'user_api', targetType: 'kb', targetId: 'kb-1' });
|
||||
|
||||
expect(lifecycleRepo.createJob).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
|
||||
@ -1,6 +1,7 @@
|
||||
import { Injectable, Logger, BadRequestException } from '@nestjs/common';
|
||||
import * as crypto from 'crypto';
|
||||
import { PrismaService } from '../../infrastructure/database/prisma.service';
|
||||
import { OutboxRepository, CreateOutboxInput } from '../../infrastructure/outbox/outbox.repository';
|
||||
import { OutboxRepository } from '../../infrastructure/outbox/outbox.repository';
|
||||
import { SnapshotBuilderService } from '../ai-runtime/snapshot-builder.service';
|
||||
import { AiJobLifecycleRepository, CreateJobInput } from './ai-job-lifecycle.repository';
|
||||
import { JobDefinitionRegistry } from './job-definition-registry';
|
||||
@ -63,32 +64,12 @@ export class AiJobCreationService {
|
||||
// 1. Resolve Definition（fail-fast：未知 JobType 在此拦截）
|
||||
const def = this.registry.get(input.jobType);
|
||||
|
||||
// 2. Validate input schema version consistency
|
||||
// 2. Validate triggerType（TypeScript 编译期 + 运行时双重保护）
|
||||
if (input.triggerType && !['user_api', 'admin_manual', 'system_scheduled', 'parent_job'].includes(input.triggerType)) {
|
||||
throw new BadRequestException(`Invalid triggerType: ${input.triggerType}`);
|
||||
}
|
||||
|
||||
// 3. Idempotency check（fast-return before snapshot build）
|
||||
if (input.idempotencyKey) {
|
||||
const existing = await this.prisma.aiJob.findUnique({
|
||||
where: {
|
||||
userId_jobType_idempotencyKey: {
|
||||
userId: input.userId,
|
||||
jobType: input.jobType,
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
},
|
||||
},
|
||||
});
|
||||
if (existing) {
|
||||
this.logger.log(
|
||||
`Idempotent return: jobId=${existing.id} ` +
|
||||
`key=${input.idempotencyKey}`,
|
||||
);
|
||||
return existing;
|
||||
}
|
||||
}
|
||||
|
||||
// 4. Build snapshot（事务外：依赖外部 DB 读取，不应占用事务时间）
|
||||
// 3. Build snapshot（事务外：依赖外部 DB 读取，不应占用事务时间）
|
||||
const snapshot = await this.snapshotBuilder.buildSnapshot(
|
||||
input.userId,
|
||||
input.targetType,
|
||||
@ -97,9 +78,9 @@ export class AiJobCreationService {
|
||||
|
||||
const contentHash = this.computeHash(JSON.stringify(snapshot));
|
||||
|
||||
// 5. Transaction: Job + Snapshot + Outbox
|
||||
const job = await this.prisma.$transaction(async (tx) => {
|
||||
// 5a. Create AiJob
|
||||
// 4. Transaction: Job + Snapshot + Outbox（含幂等保护）
|
||||
const createFn = async (tx: Prisma.TransactionClient) => {
|
||||
// 4a. Create AiJob
|
||||
const jobInput: CreateJobInput = {
|
||||
userId: input.userId,
|
||||
jobType: input.jobType,
|
||||
@ -114,6 +95,7 @@ export class AiJobCreationService {
|
||||
credentialId: input.overrides?.credentialId,
|
||||
modelProvider: def.model.modelProvider,
|
||||
modelName: def.model.modelName,
|
||||
promptKey: def.prompt.promptKey,
|
||||
promptVersion: def.prompt.promptVersion,
|
||||
outputSchemaVersion: def.output.schemaVersion,
|
||||
inputSchemaVersion: def.input.schemaVersion,
|
||||
@ -123,7 +105,7 @@ export class AiJobCreationService {
|
||||
|
||||
const job = await this.lifecycleRepo.createJob(tx, jobInput);
|
||||
|
||||
// 5b. Create AiJobSnapshot
|
||||
// 4b. Create AiJobSnapshot
|
||||
await tx.aiJobSnapshot.create({
|
||||
data: {
|
||||
jobId: job.id,
|
||||
@ -135,7 +117,7 @@ export class AiJobCreationService {
|
||||
},
|
||||
});
|
||||
|
||||
// 5c. Create OutboxEvent
|
||||
// 4c. Create OutboxEvent
|
||||
const dedupeKey = `ai.job.enqueue:${job.id}`;
|
||||
await this.outboxRepo.createInTransaction(tx, {
|
||||
eventType: 'ai.job.enqueue',
|
||||
@ -146,21 +128,41 @@ export class AiJobCreationService {
|
||||
availableAt: new Date(),
|
||||
});
|
||||
|
||||
return job; // job 在事务成功路径中使用，P2002 catch 路径中不用
|
||||
};
|
||||
|
||||
try {
|
||||
const job = await this.prisma.$transaction(createFn);
|
||||
this.logger.log(
|
||||
`Job created: id=${job.id} type=${input.jobType} ` +
|
||||
`queue=${def.queue.queueName} snapshotHash=${contentHash}`,
|
||||
);
|
||||
return job;
|
||||
});
|
||||
|
||||
this.logger.log(
|
||||
`Job created: id=${job.id} type=${input.jobType} ` +
|
||||
`queue=${def.queue.queueName} snapshotHash=${contentHash}`,
|
||||
);
|
||||
|
||||
return job;
|
||||
} catch (err: any) {
|
||||
// 幂等：并发重复创建 → UNIQUE 冲突（P2002）→ 返回已有 Job
|
||||
if (err?.code === 'P2002' && input.idempotencyKey) {
|
||||
const existing = await this.prisma.aiJob.findUniqueOrThrow({
|
||||
where: {
|
||||
userId_jobType_idempotencyKey: {
|
||||
userId: input.userId,
|
||||
jobType: input.jobType,
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
},
|
||||
},
|
||||
});
|
||||
this.logger.log(
|
||||
`Idempotent return (P2002): jobId=${existing.id} ` +
|
||||
`key=${input.idempotencyKey}`,
|
||||
);
|
||||
return existing;
|
||||
}
|
||||
throw err;
|
||||
}
|
||||
}
|
||||
|
||||
// ── helpers ──
|
||||
|
||||
private computeHash(content: string): string {
|
||||
const crypto = require('crypto');
|
||||
return crypto.createHash('sha256').update(content).digest('hex').substring(0, 16);
|
||||
}
|
||||
}
|
||||
|
||||
@ -31,6 +31,7 @@ export interface CreateJobInput {
|
||||
credentialId?: string;
|
||||
modelProvider?: string;
|
||||
modelName?: string;
|
||||
promptKey?: string;
|
||||
promptVersion?: string;
|
||||
outputSchemaVersion?: string;
|
||||
inputSchemaVersion?: string;
|
||||
@ -98,6 +99,7 @@ export class AiJobLifecycleRepository {
|
||||
// 若未传入，依赖 Prisma schema @default 保证列非空。
|
||||
...(input.modelProvider ? { modelProvider: input.modelProvider } : {}),
|
||||
...(input.modelName ? { modelName: input.modelName } : {}),
|
||||
promptKey: input.promptKey ?? null,
|
||||
promptVersion: input.promptVersion ?? null,
|
||||
outputSchemaVersion: input.outputSchemaVersion ?? null,
|
||||
inputSchemaVersion: input.inputSchemaVersion ?? null,
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user