UHD_Media_Tool/docs/adr/ADR-002-task-engine.md

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 采用 [email protected](精确锁版本)+ 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. 端口(业务只依赖它)

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. 等待外部条件 = 查库 + 带超时的信号。不要只等一个信号:

   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 延迟不可接受;
• 需要可配置租约或步骤级取消信号,但上游不支持。

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

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 认领。