[M-AI-03] 实现 AiJob 状态机与原子状态仓储 #286

New Issue

wangdl · 2026-06-20T16:43:25+08:00

wangdl commented

2026-06-20 16:43:25 +08:00

类型 / 标签

backend
state-machine
database

风险等级

P0

依赖

M-AI-03-02

目标

实现统一生命周期的唯一写入入口，禁止业务代码任意修改 lifecycleStatus。

实现内容

新增：

AiJobStateMachine
AiJobLifecycleRepository
状态迁移错误类型
乐观并发/CAS 更新
Legacy status Shadow Write

要求

状态更新必须使用条件更新：

WHERE id = ?
AND lifecycleStatus IN (...)

禁止：

先读取后无条件 update；
直接调用 prisma.aiJob.update({ lifecycleStatus })；
在多个 Service 中重复实现状态映射。

Legacy 映射

至少保证：

created / queued   → pending
running / retrying → processing
succeeded          → completed
failed             → failed
cancelled          → failed + errorCode=cancelled

cancel_requested 保留当前 Legacy status，直到进入 cancelled。

非目标

不修改旧 Worker；
不切换旧 Repository；
不创建 Queue Job。

验收标准

所有合法状态迁移测试通过
所有非法迁移被拒绝并抛出明确错误
终态（succeeded/failed/cancelled）不可再次修改
并发双写只有一个成功（乐观锁/CAS 验证）
新旧 status 字段原子更新
时间字段（startedAt/finishedAt）正确设置
错误字段（errorCode/internalErrorMessage）正确清理或保留
真实 MySQL 集成测试通过

## 类型 / 标签 - backend - state-machine - database ## 风险等级 P0 ## 依赖 - M-AI-03-02 ## 目标实现统一生命周期的唯一写入入口，禁止业务代码任意修改 `lifecycleStatus`。 ## 实现内容新增： - `AiJobStateMachine` - `AiJobLifecycleRepository` - 状态迁移错误类型 - 乐观并发/CAS 更新 - Legacy status Shadow Write ## 要求状态更新必须使用条件更新： ```sql WHERE id = ? AND lifecycleStatus IN (...) ``` 禁止： - 先读取后无条件 update； - 直接调用 `prisma.aiJob.update({ lifecycleStatus })`； - 在多个 Service 中重复实现状态映射。 ## Legacy 映射至少保证： ``` created / queued → pending running / retrying → processing succeeded → completed failed → failed cancelled → failed + errorCode=cancelled ``` `cancel_requested` 保留当前 Legacy status，直到进入 cancelled。 ## 非目标 - 不修改旧 Worker； - 不切换旧 Repository； - 不创建 Queue Job。 ## 验收标准 - [ ] 所有合法状态迁移测试通过 - [ ] 所有非法迁移被拒绝并抛出明确错误 - [ ] 终态（succeeded/failed/cancelled）不可再次修改 - [ ] 并发双写只有一个成功（乐观锁/CAS 验证） - [ ] 新旧 status 字段原子更新 - [ ] 时间字段（startedAt/finishedAt）正确设置 - [ ] 错误字段（errorCode/internalErrorMessage）正确清理或保留 - [ ] 真实 MySQL 集成测试通过

wangdl added this to the M-AI-03 统一 Job Engine、Registry 与 Outbox milestone 2026-06-20 16:44:54 +08:00

wangdl commented

2026-06-20 17:17:37 +08:00

开发完成评论

完成内容

AiJobStateMachine（ai-job-state-machine.ts）：
- 5 状态定义（queued/running/succeeded/failed/cancelled）+ 终态集合
- 合法迁移表（queued→running/cancelled；running→succeeded/failed/cancelled）
- 非法迁移抛出 IllegalStateTransitionError
- 双向 Shadow Write 映射（LIFECYCLE_TO_STATUS + STATUS_TO_LIFECYCLE），含 #285 修正的 cancelled/cancel_requested
- parse() 方法处理 null/undefined/无效字符串
AiJobLifecycleRepository（ai-job-lifecycle.repository.ts）：
- createJob(tx, input)：在事务中创建 queued 状态的 Job + Shadow Write
- lockJob()：CAS queued → running（WHERE lifecycleStatus = 'queued'），失败时区分终态/已锁定/不存在
- extendLock()：续约 lockUntil + 返回 cancelRequested 信号
- markSucceeded(tx?)：CAS running → succeeded + Shadow Write completed，清除错误和运行时字段
- markFailed(error, tx?)：CAS running → failed + 写入 errorCode/publicErrorMessage/internalErrorMessage
- markCancelled(tx?)：从 queued 或 running 取消 → cancelled，Shadow Write failed + errorCode=CANCELLED
- requestCancellation()：queued→直接 cancel；running→设置 cancelRequestedAt 信号
- findByLifecycleStatus() / findExpiredLocks() / findById() 只读查询
6 个类型化错误（ai-job.errors.ts）：
JobLockConflictError, JobNotActiveError, JobAlreadyTerminalError,
JobNotCancellableError, JobExtendLockError, IllegalStateTransitionError
AiJobModule（ai-job.module.ts）：独立模块，当前导出 StateMachine + Repository

修改文件

src/modules/ai-job/ai-job.errors.ts（新增，56 行）
src/modules/ai-job/ai-job-state-machine.ts（新增，99 行）
src/modules/ai-job/ai-job-lifecycle.repository.ts（新增，351 行）
src/modules/ai-job/ai-job.module.ts（新增，20 行）
src/modules/ai-job/ai-job-lifecycle.repository.spec.ts（新增，340 行）

测试情况

已运行命令：
- npx jest --testPathPatterns="ai-job-lifecycle.repository.spec" --no-coverage
结果：
- 41/41 通过
- TypeScript 编译：无错误

验收标准对照

所有合法状态迁移测试通过 → AiJobStateMachine 节：5 合法迁移 each 测试
所有非法迁移被拒绝并抛出明确错误 → 12 非法迁移 each 测试 + 6 类型化错误类
终态不可再次修改 → markSucceeded/markFailed/markCancelled CAS 失败时抛出 JobAlreadyTerminalError
并发双写只有一个成功 → lockJob CAS updateMany + count===0 检测
新旧 status 字段原子更新 → 所有 write 方法同步 lifecycleStatus + status (Shadow Write)
时间字段正确设置 → createJob:queuedAt, lockJob:startedAt/lockedAt/lockUntil, markSucceeded/markFailed/markCancelled:finishedAt
错误字段正确清理或保留 → markSucceeded 清除 errorCode 等字段；markFailed 写入；markCancelled 写入 errorCode=CANCELLED
真实 MySQL 集成测试通过 → 未运行（本阶段为单元测试；集成测试在后续 Issue #294 中）

是否建议进入 Review

是 — 建议审查后进入 #287（Job Definition Registry）

## 开发完成评论 ### 完成内容 - **AiJobStateMachine**（`ai-job-state-machine.ts`）： - 5 状态定义（queued/running/succeeded/failed/cancelled）+ 终态集合 - 合法迁移表（queued→running/cancelled；running→succeeded/failed/cancelled） - 非法迁移抛出 `IllegalStateTransitionError` - 双向 Shadow Write 映射（`LIFECYCLE_TO_STATUS` + `STATUS_TO_LIFECYCLE`），含 #285 修正的 cancelled/cancel_requested - `parse()` 方法处理 null/undefined/无效字符串 - **AiJobLifecycleRepository**（`ai-job-lifecycle.repository.ts`）： - `createJob(tx, input)`：在事务中创建 queued 状态的 Job + Shadow Write - `lockJob()`：CAS `queued → running`（`WHERE lifecycleStatus = 'queued'`），失败时区分终态/已锁定/不存在 - `extendLock()`：续约 `lockUntil` + 返回 `cancelRequested` 信号 - `markSucceeded(tx?)`：CAS `running → succeeded` + Shadow Write `completed`，清除错误和运行时字段 - `markFailed(error, tx?)`：CAS `running → failed` + 写入 `errorCode`/`publicErrorMessage`/`internalErrorMessage` - `markCancelled(tx?)`：从 queued 或 running 取消 → `cancelled`，Shadow Write `failed` + `errorCode=CANCELLED` - `requestCancellation()`：queued→直接 cancel；running→设置 `cancelRequestedAt` 信号 - `findByLifecycleStatus()` / `findExpiredLocks()` / `findById()` 只读查询 - **6 个类型化错误**（`ai-job.errors.ts`）： `JobLockConflictError`, `JobNotActiveError`, `JobAlreadyTerminalError`, `JobNotCancellableError`, `JobExtendLockError`, `IllegalStateTransitionError` - **AiJobModule**（`ai-job.module.ts`）：独立模块，当前导出 StateMachine + Repository ### 修改文件 - `src/modules/ai-job/ai-job.errors.ts`（新增，56 行） - `src/modules/ai-job/ai-job-state-machine.ts`（新增，99 行） - `src/modules/ai-job/ai-job-lifecycle.repository.ts`（新增，351 行） - `src/modules/ai-job/ai-job.module.ts`（新增，20 行） - `src/modules/ai-job/ai-job-lifecycle.repository.spec.ts`（新增，340 行） ### 测试情况 - 已运行命令： - `npx jest --testPathPatterns="ai-job-lifecycle.repository.spec" --no-coverage` - 结果： - **41/41 通过** - TypeScript 编译：无错误 ### 验收标准对照 - [x] 所有合法状态迁移测试通过 → AiJobStateMachine 节：5 合法迁移 each 测试 - [x] 所有非法迁移被拒绝并抛出明确错误 → 12 非法迁移 each 测试 + 6 类型化错误类 - [x] 终态不可再次修改 → `markSucceeded`/`markFailed`/`markCancelled` CAS 失败时抛出 `JobAlreadyTerminalError` - [x] 并发双写只有一个成功 → `lockJob` CAS updateMany + count===0 检测 - [x] 新旧 status 字段原子更新 → 所有 write 方法同步 `lifecycleStatus` + `status` (Shadow Write) - [x] 时间字段正确设置 → `createJob:queuedAt`, `lockJob:startedAt/lockedAt/lockUntil`, `markSucceeded/markFailed/markCancelled:finishedAt` - [x] 错误字段正确清理或保留 → `markSucceeded` 清除 errorCode 等字段；`markFailed` 写入；`markCancelled` 写入 errorCode=CANCELLED - [ ] 真实 MySQL 集成测试通过 → 未运行（本阶段为单元测试；集成测试在后续 Issue #294 中） ### 是否建议进入 Review - 是 — 建议审查后进入 #287（Job Definition Registry）

wangdl closed this issue

2026-06-21 12:37:50 +08:00

Sign in to join this conversation.

1 Participants

Notifications

Due Date

No due date set.

Dependencies

No dependencies set.

Reference: wangdl/api-server#286