rootandClaude Fable 5.1
1262340856
- TaskEngine port + OpenWorkflowEngine ([email protected].1, schema openworkflow): enqueue/signal/cancel/get/list/startWorker/stop - defineTask registry with zod input schema; 9 stub task definitions (§3.1-8) - TaskContext: step.run/sleep/waitForFact, redacting logger, traceId, attemptToken, abortSignal driven by PG cancel flag - signal persists fact to tasks_fact first; waitForFact checks DB before waiting - OutboxRelay: FOR UPDATE SKIP LOCKED, idempotencyKey outbox:<id> - withAttemptToken fencing on tasks_op_ledger (own migration, merged in W2) - tests: kill -9 crash recovery (child worker), cancel, idempotency, concurrent relays, early fact, lease-loss fencing Co-Authored-By: Claude Fable 5.1 <[email protected]>
58 lines
5.6 KiB
Markdown
58 lines
5.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` 里未投递的行,按 type 路由后 enqueue(幂等键 `outbox:<id>`),再标记 `published_at`;多进程并发时不会重复 |
|
|
| `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 规则)
|
|
|
|
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,会无限退避等待,不会失败。
|
|
- 同一版本里,只允许在末尾追加步骤,或插入幂等步骤。
|
|
|
|
## 已知缺口(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。
|
|
|
|
## 测试
|
|
|
|
`pnpm test` 会连接开发库(`DATABASE_URL`,默认 `postgres://umt:umt@localhost:55433/umt`)。每个测试文件使用独立的 `tasks_test_*` schema,测试前后都会清理;库不可达时跳过集成测试。覆盖范围:kill -9 后续跑(子进程 worker)、取消、幂等 enqueue、两个 relay 并发、事实先到达、fencing。
|