rootandClaude Opus 5.5
a9cee69bd2
OpenWorkflow 0.10.1 claims any run regardless of workflow name and fails unregistered ones with 5s-5m backoff; filter the claim SQL by name@version. Co-Authored-By: Claude Opus 5.5 <[email protected]>
61 lines
6.6 KiB
Markdown
61 lines
6.6 KiB
Markdown
# @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 只认领自己的任务。
|