# ADR-002:持久任务引擎采用 OpenWorkflow 0.10.1(锁版本),封装在 `TaskEngine` 端口之后

- 状态:已采纳(2026-09-27,W0-B spike 后)
- 关联:REQ §11、`00-first-version-plan`（已删除，见 git `ceb66f8`） §1 与 §3.1-8、验证记录 spike `w0b-openworkflow`（已删除，见 git `ceb66f8`）、代码 `spikes/openworkflow/`（已删除，见 git `ceb66f8`）

## 决定

v0.1 采用 **`openworkflow@0.10.1`(精确锁版本)+ `postgres@^3.4.9`**,表放在同库独立 schema `openworkflow`。所有调用都经过 `packages/tasks` 的 `TaskEngine` 端口,业务代码不 import `openworkflow`。**业务状态不放在引擎里**:引擎只负责"过程推进、定时唤醒、重试、租约";下载任务状态、授权、操作记录都在业务表里。

暂不实现自研租约任务表。下文附录给出备选设计,只在触发复议条件时启用(REQ §11.1:"只保留清晰的执行接口,不提前实现第二套引擎")。

## 验证结论摘要

| 项 | 结果 | 对接入的约束 |
| --- | --- | --- |
| (a) 表初始化 | `BackendPostgres.connect()` 自动幂等迁移 | 迁移放进统一迁移流程;启动时 `runMigrations:false` |
| (b) kill -9 恢复 | 通过。已完成步骤不重跑,正在执行的步骤会在新 worker 上**重跑**(最长等 30s 租约到期) | 步骤体必须幂等 |
| 租约丢失 | 引擎拒收过期 worker 的步骤结果,但**步骤体仍会执行完并产生副作用**,拿不到租约状态 | 业务写入要用操作账本做 fencing |
| (c) 信号提前到达 | **丢失**。没有等待者时 `sendSignal` 直接返回 `[]`,不落库 | 授权、下载完成都**先写业务库**,信号只用来加速唤醒 |
| (d) 取消 | run 变成 `canceled`,后续步骤不再执行;**进行中的步骤不中断**(没有 AbortSignal);已取消的 run 仍会"接收"信号 | 长步骤自己轮询业务侧取消标志;补偿策略显式化 |
| (e) 重试 | 步骤级 `maximumAttempts` 生效,超过后 run 变成 `failed` | 默认 10 次偏多,按任务类型显式配置 |
| (f) 幂等键 | 同 key 返回已有 run;**24h 窗口,不比较输入** | 幂等键由服务端按语义生成,长期去重靠业务表唯一约束 |
| (g) 版本升级 | 旧版本 run 找不到对应 version 的 worker 时无限退避等待,不会失败;同一 version 改代码按步骤名重放,不做确定性检查 | 不兼容的改动必须升 version,worker 同时注册新旧版本 |
| (h) 与 Kysely 共库 | 不同 schema,无冲突;引擎用自己的 postgres.js 连接池,**不能和业务写入同事务入队** | outbox |
| (i) 双 worker 并发 | 60 个 run 执行 60 次,无重复(`SKIP LOCKED`) | — |
| PG 重启 | 自动恢复 | — |

## 给 `packages/tasks` 的接入要求

### 1. 端口(业务只依赖它)

```ts
export interface TaskEngine {
  /** 幂等:同 kind+idempotencyKey 返回同一 taskRunId。调用方负责先在业务事务里写 outbox。 */
  enqueue<K extends TaskKind>(kind: K, input: TaskInput<K>, opts: { idempotencyKey: string; availableAt?: Date }): Promise<{ taskRunId: string }>;
  /** 只做唤醒提示。等待方必须以业务库为准;信号丢了也不能影响正确性。 */
  signal(topic: SignalTopic, payload?: JsonValue): Promise<{ delivered: number }>;
  /** 停止后续步骤。不撤销已完成的外部动作,也不打断进行中的步骤。 */
  cancel(taskRunId: string, reason: string): Promise<void>;
  get(taskRunId: string): Promise<TaskRunView | null>;   // 状态/attempts/error/当前步骤,只读投影
  start(opts: { roles: ("worker")[]; concurrency: number }): Promise<void>;
  stop(): Promise<void>;
}
```

`TaskKind` 对应 plan §3.1-8 的九类任务。每类任务在 `packages/tasks` 里定义 `{ name, version, retryPolicy, handler }`,handler 只能通过注入的业务服务读写 Kysely。

### 2. 必须遵守的模式

1. **Outbox 入队**:业务事务里写 `outbox_event(id, kind, idempotency_key, payload, status, attempts, next_attempt_at)`;分发器轮询 `pending` 行,调用 `enqueue`(引擎幂等键 = outbox 的 idempotency_key),成功后标记 `dispatched`。重复分发由 24h 引擎幂等和业务唯一约束兜底。
2. **等待外部条件 = 查库 + 带超时的信号**。不要只等一个信号:
   ```ts
   while (true) {
     const ok = await step.run({ name: `check-approval-${i}` }, () => approvals.findValid(planId)); // 业务库是唯一真相
     if (ok) break;
     await step.waitForSignal({ name: `wait-approval-${i}`, signal: `approval:${planId}`, timeout: "5m" }); // 信号只加速
     i++;
   }
   ```
   注意:同一信号名同一时间只能有一个等待步骤;步骤名要带序号(引擎也会自动加 `:1` 之类的后缀)。每轮循环都消耗 step attempt 配额(单个 run 上限 1000),长期等待用 `timeout` 拉长间隔,例如下载完成用 10m。
3. **副作用步骤走操作账本**(REQ §11.2):`asset_operation`/`download_task` 记录 `plan → attempt(started) → result`,步骤体开头先查账本:
   - 已有 result:直接返回,不重做;
   - 有 attempt 没有 result:先对账外部状态(qB 按 hash 或管理标签查询,文件按目标路径 + 大小/hash 核验),再决定收尾还是重做;
   - 提交业务状态时带 fencing 条件,例如 `UPDATE … WHERE id=? AND attempt_token=?`。attempt_token 在步骤开始时写入,重跑的新 attempt 会覆盖它,老 worker 的提交因此失败。
4. **取消**:`TaskEngine.cancel` = 先写业务侧 `cancel_requested`,再调用引擎 cancel。长步骤(复制大文件)分块时检查这个标志,自己中止;补偿(暂停 qB、删临时文件)作为显式步骤或单独任务。
5. **版本纪律**:handler 有不兼容改动时 `version` +1;worker 同时注册当前和上一个版本,直到旧版本的 run 清空(用 `listWorkflowRuns({ workflowName, status })` 查,并在 `/admin` 显示)。同版本内只允许在末尾追加步骤,或插入幂等步骤,不允许改已有步骤名。
6. **重试策略显式化**:每类任务显式给出 `retryPolicy`。外部网络类 5 次,1s→60s;文件类 3 次。用尽后 run 变成 `failed`,由 `resolveDecision`/待处理中心接管。死信视图 = `status='failed'` 的 run 加业务上下文。
7. **展示**:UI 的业务时间线读业务表,不读 `step_attempts`,因为里面会有停在 `running` 的孤儿行。
8. **连接**:引擎用自己的 postgres.js 连接池(同一个 DATABASE_URL);Kysely 用 pg。两边的连接数都计入 PG `max_connections`。

### 3. 测试要求

- vitest 集成测试(testcontainers 或 docker PG):kill worker 后续跑、提前信号 + 查库路径、取消、重试上限、幂等 enqueue、双 worker。
- 故障注入(REQ §11.1):qB 接收后断连、复制完成但结果未提交、DB 断连、重复投递、租约丢失(SIGSTOP)、运行中升级。断言对象是**业务账本和文件系统**,不是引擎 `completed`。

## 已知风险

- 0.x 版本,API 仍在变(0.1→0.10 一年内发布 25 个版本)。锁精确版本,升级前重跑 spike 用例。
- 租约固定 30s,不可配置;worker 挂掉后最长约 30s 才会续跑。
- 步骤体拿不到 AbortSignal 和租约状态,fencing 全靠业务层。
- 信号不持久,同名只能有一个等待者。
- 引擎自带的 idempotency 只有 24h。

## 复议条件(触发后启用附录中的自研表)

- 升级到新版本后恢复/重试语义回归,且短期修不好;
- 需要与业务写入同事务入队,且 outbox 延迟不可接受;
- 需要可配置租约或步骤级取消信号,但上游不支持。

## 附录:备选的自研租约任务表(本版不实现)

```sql
create table task (
  id uuid primary key, kind text not null, handler_version int not null,
  idempotency_key text not null, input jsonb not null, state jsonb not null default '{}', -- 步骤游标 + 已完成步骤结果
  status text not null check (status in ('pending','running','waiting','succeeded','failed','canceled','dead')),
  attempts int not null default 0, max_attempts int not null,
  next_run_at timestamptz not null default now(), lease_owner text, lease_until timestamptz, lease_token uuid,
  wait_topic text, wait_until timestamptz, cancel_requested_at timestamptz,
  last_error jsonb, dead_letter_at timestamptz, created_at timestamptz default now(), updated_at timestamptz default now(),
  unique (kind, idempotency_key)
);
create index on task (status, next_run_at);
create table task_signal (id bigserial primary key, topic text not null, payload jsonb, created_at timestamptz default now(), consumed_by uuid);
-- 信号持久化:等待方进入 waiting 前先查未消费的 task_signal,解决"提前到达"
-- outbox_event 同 §2.1
```

- 认领:`UPDATE task SET status='running', lease_owner=$w, lease_token=gen_random_uuid(), lease_until=now()+$lease, attempts=attempts+1 WHERE id=(SELECT id FROM task WHERE status IN('pending','waiting') AND next_run_at<=now() AND cancel_requested_at IS NULL ORDER BY priority, next_run_at FOR UPDATE SKIP LOCKED LIMIT 1) RETURNING *`。
- 心跳:`UPDATE … SET lease_until=… WHERE id=$id AND lease_token=$tok`,影响 0 行说明租约已丢,立即中止并禁止提交。所有业务提交都带 `lease_token` 条件,在同一事务里完成。
- 恢复算法:扫描 `status='running' AND lease_until<now()` 的行,`attempts<max_attempts` 时回到 `pending`(next_run_at 按退避计算),否则置为 `dead`(dead_letter_at=now());handler 按 `state` 里的步骤游标续跑,已完成步骤直接取结果。
- 版本:`handler_version` 不匹配的任务只由声明兼容该版本的 worker 认领。