rootandClaude Fable 5.1
31729f2bac
The relay only looked at published_at and ignored the ADR-002 §2.1 columns added in db 0002_foundation: rows stayed status='pending' forever after dispatch, failures were retried every tick without backoff, and the row-level idempotency_key was never used as the engine enqueue key. - claim status='pending' AND next_attempt_at<=now() - enqueue with idempotency_key (fallback outbox:<id>) - success: status=dispatched, published_at, attempts+1 - failure: attempts+1, exponential backoff (cap 5 min), last_error; dead at maxAttempts - test helper table now mirrors db 0001+0002; tests cover backoff and dead Co-Authored-By: Claude Fable 5.1 <[email protected]>
@umt/tasks
持久任务引擎封装。依据 ADR-002:OpenWorkflow 0.10.1(精确锁定)+ postgres.js,引擎表固定在 openworkflow schema。业务代码只依赖本包导出的 TaskEngine 端口,不直接 import openworkflow。
引擎边界
- 引擎只管过程:推进步骤、定时唤醒、重试、租约。业务状态不放引擎,download_task / approval / asset_operation 等都写业务表。
- UI 读业务表。
TaskEngine.get/list只是只读投影,给 task.inspect 和管理页看状态、attempts、错误用。不要读openworkflow.step_attempts来拼时间线:被中断的 attempt 会一直停在running。 - 引擎用自己的 postgres.js 连接池,不能和业务写入放在同一个事务里入队。入队要走 outbox。
组成
| 导出 | 作用 |
|---|---|
OpenWorkflowEngine.connect({ databaseUrl, db, registry }) |
TaskEngine 实现:enqueue / signal / cancel / get / list / startWorker / stop |
defineTask / TaskRegistry / createDefaultRegistry() |
任务定义与注册表;默认注册表里有 §3.1-8 的 9 个任务(目前是空壳) |
OutboxRelay |
用 FOR UPDATE SKIP LOCKED 认领 outbox_event 中 status=pending 且 next_attempt_at 已到的行,按 type 路由后 enqueue(幂等键取行的 idempotency_key,缺省 outbox:<id>),成功标记 dispatched + published_at;失败 attempts+1 指数退避(上限 5 分钟),达 maxAttempts 置 dead;多进程并发时不会重复 |
withAttemptToken(db, opKey, fn, { commit }) |
操作账本 fencing,表是 tasks_op_ledger |
migrateTasksTables(db) |
本包迁移(migrations/):tasks_run_meta、tasks_fact、tasks_op_ledger。W2 并入 packages/db |
TaskContext 包含 input(已经过 zod 校验)、runId、traceId、priority、attemptToken(每次执行 handler 都会重新生成)、abortSignal、logger(自动脱敏)、step.run / step.sleep / step.waitForFact、throwIfCanceled()。
- 取消:
cancel(runId, reason)先写tasks_run_meta.cancel_requested_at,再取消引擎里的 run。handler 执行期间会轮询这个标志(默认 1s)并触发abortSignal;每个step.run/sleep开始前也会检查一次。进行中的步骤只能靠自己检查abortSignal来中止。已完成的外部动作不会撤销,补偿要写成显式步骤。 - 等待外部条件:
signal(runId, name, payload)先把事实写进tasks_fact,再发送引擎信号。step.waitForFact(name, { check?, timeout })每一轮都先查库,查不到才带超时等待信号。所以事实比等待方先到也不会丢。审批、下载完成这类业务事实,先写业务表,再用check去读业务表。
新增/填充任务(W2 规则)
- 在
src/definitions.ts里用defineTask({ name, version, inputSchema, retryPolicy, handler })定义任务。输入只放业务 id,handler 自己去读业务表。重试策略要显式写:外部网络类 5 次(1s→60s),文件类 3 次。 - 步骤体必须幂等。worker 被 kill 后,正在执行的步骤会在新 worker 上从头重跑(最长要等约 30s 租约到期)。已完成的步骤不会重跑。
- 有外部副作用的步骤都走
withAttemptToken,opKey 按语义生成,例如submit:<planId>:- 已经有结果时直接返回;
info.resumed === true表示上一次 attempt 没留下结果。这时先对账外部状态(qB 按 hash 或管理标签查,文件按目标路径 + 大小/hash 查),再决定是收尾还是重做;- 业务状态放在
commit(trx, result)里提交,和 fencing 校验在同一个事务。租约丢失的旧 attempt 提交时会抛FencingError。
- 等授权、等下载完成,用
step.waitForFact,不要直接等信号。同一个 run 里,fact 名不要重复使用。 - 长步骤(比如复制大文件)要分块检查
ctx.abortSignal,被中止时throw ctx.abortSignal.reason。 - 步骤名在同一版本内保持稳定。重放是按步骤名匹配的。
- 日志只用
ctx.logger。passkey、token 这类字段和 URL 参数会被脱敏,但也不要主动把凭证传给 logger。
版本升级规则
- 不兼容的改动(改步骤名或顺序、改输入结构、改已有步骤的语义)必须把
version+1,同时保留旧版本定义并注册。worker 用startWorker({ versions: ["submitDownload@1", "submitDownload@2"] })或者省略参数(表示全部版本)。等list({ taskName, status })查不到旧版本的 run 了,再删除旧定义。 - 新 run 一律用最高版本。旧版本 run 如果找不到对应的 worker,会无限退避等待,不会失败。
- 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。
已知缺口(OpenWorkflow 0.10.1)
- 引擎信号不持久,同名信号同一时间只能有一个等待者。已经用
tasks_fact+waitForFact补上。 - 进行中的步骤拿不到引擎的 AbortSignal 和租约状态。已经用 PG 取消标志和操作账本 fencing 补上。
- 租约固定 30s,不可配置。崩溃后最长约 30s 才会续跑。
- 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。
priority只记录在tasks_run_meta和 run 输入里,引擎仍按available_at先后认领,不支持真正的优先级调度。- 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。
runMigrations默认 true(幂等)。等并入统一迁移后,启动时改成 false。
测试
deno task test(或本包 deno task test)会连接开发库(DATABASE_URL,默认 postgres://umt:umt@localhost:55433/umt)。每个测试文件使用独立的 tasks_test_<名>_<pid+随机> schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。