UHD_Media_Tool/packages/tasks

@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;多进程并发时不会重复
migrateEngineSchema(databaseUrl, schema?)	只跑引擎自己的迁移后断开(部署时 --role migrate)
withAttemptToken(db, opKey, fn, { commit })	操作账本 fencing,表是 tasks_op_ledger
表	tasks_run_meta、tasks_fact、tasks_op_ledger 由 packages/db 迁移 0013_tasks_tables 建立(类型见 @umt/db 的 DB);本包不再自带迁移,只通过注入的 Kysely 访问

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 规则)

1. 在 src/definitions.ts 里用 defineTask({ name, version, inputSchema, retryPolicy, handler }) 定义任务。输入只放业务 id,handler 自己去读业务表。重试策略要显式写:外部网络类 5 次(1s→60s),文件类 3 次。
2. 步骤体必须幂等。worker 被 kill 后,正在执行的步骤会在新 worker 上从头重跑(最长要等约 30s 租约到期)。已完成的步骤不会重跑。
3. 有外部副作用的步骤都走 withAttemptToken,opKey 按语义生成,例如 submit:<planId>:
   • 已经有结果时直接返回;
   • info.resumed === true 表示上一次 attempt 没留下结果。这时先对账外部状态(qB 按 hash 或管理标签查,文件按目标路径 + 大小/hash 查),再决定是收尾还是重做;
   • 业务状态放在 commit(trx, result) 里提交,和 fencing 校验在同一个事务。租约丢失的旧 attempt 提交时会抛 FencingError。
4. 等授权、等下载完成,用 step.waitForFact,不要直接等信号。同一个 run 里,fact 名不要重复使用。
5. 长步骤(比如复制大文件)要分块检查 ctx.abortSignal,被中止时 throw ctx.abortSignal.reason。
6. 步骤名在同一版本内保持稳定。重放是按步骤名匹配的。
7. 日志只用 ctx.logger。passkey、token 这类字段和 URL 参数会被脱敏,但也不要主动把凭证传给 logger。

版本升级规则

• 不兼容的改动(改步骤名或顺序、改输入结构、改已有步骤的语义)必须把 version +1,同时保留旧版本定义并注册。worker 用 startWorker({ versions: ["submitDownload@1", "submitDownload@2"] }) 或者省略参数(表示全部版本)。等 list({ taskName, status }) 查不到旧版本的 run 了,再删除旧定义。
• 新 run 一律用最高版本。旧版本 run 如果找不到对应的 worker,会无限退避等待,不会失败。
• 带 versions 的 worker 只认领这些 (任务名, 版本) 的 run(本包在认领 SQL 上加了过滤;OpenWorkflow 0.10.1 原生认领不看任务名, 认领到未注册的 run 会以"未注册"失败并按 5s→5m 退避,分组 worker 会互相拖慢)。server 用它把 worker 分成 io 组与 agent 组。
• 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。

已知缺口(OpenWorkflow 0.10.1)

• 引擎信号不持久,同名信号同一时间只能有一个等待者。已经用 tasks_fact + waitForFact 补上。
• 进行中的步骤拿不到引擎的 AbortSignal 和租约状态。已经用 PG 取消标志和操作账本 fencing 补上。
• 租约固定 30s,不可配置。崩溃后最长约 30s 才会续跑。
• 引擎幂等键只在 24h 内有效,而且不比较输入。长期去重靠业务表的唯一约束。
• priority 只记录在 tasks_run_meta 和 run 输入里,引擎仍按 available_at 先后认领,不支持真正的优先级调度。
• 取消只阻止后续步骤,不撤销已完成的外部动作。在已取消的 run 上,handler 抛出的错误会被引擎拒收,只在 worker 日志里出现。
• 引擎表(openworkflow schema)由 migrateEngineSchema(databaseUrl) 迁移:server 的 --role migrate 调用它;--role all 单进程连接时顺带迁移,api/worker 分进程以 runMigrations:false 连接。

测试

deno task test(或本包 deno task test)会连接开发库,每个测试 schema 先用 @umt/db 的 migrateToLatest 建全部业务表(DATABASE_URL,默认 postgres://umt:umt@localhost:55433/umt)。每个测试文件使用独立的 tasks_test_<名>_<pid+随机> schema(并发跑多份也不冲突),测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing、分组 worker 只认领自己的任务。