# W0-B spike:OpenWorkflow 0.10.1 兼容性验证记录

- 日期:2026-09-27;环境:Node v24.15.0、PostgreSQL 16(`postgres:16-alpine`,容器 `umt-spike-pg`,端口 55434)、`openworkflow@0.10.1`、`postgres@3.4.x`、`kysely@0.28` + `pg@8`
- 代码:`spikes/openworkflow/`(`common.ts` 定义工作流,`worker.ts` 起 worker,`cli.ts` 投递/信号/取消/查询)
- 业务副作用账本:每个步骤体执行时往 `biz.side_effect_ledger`(Kysely + pg)插一行,用来判断步骤有没有重跑、有没有在失去租约后继续写。
- 结论见 `docs/adr/ADR-002-task-engine.md`。

## 0. API 形状(以 node_modules 内 .d.ts 为准)

```ts
import { OpenWorkflow } from "openworkflow";
import { BackendPostgres } from "openworkflow/postgres";

const backend = await BackendPostgres.connect(url, { schema: "openworkflow", namespaceId?, runMigrations?: true });
// 或 BackendPostgres.fromPool(postgresJsSql, { schema })  —— 共享 postgres.js 连接池,不跑迁移
const ow = new OpenWorkflow({ backend });
const wf = ow.defineWorkflow({ name, version?, schema?: StandardSchema, retryPolicy?: { maximumAttempts, initialInterval, backoffCoefficient, maximumInterval } },
  async ({ input, step, run, version }) => {
    await step.run({ name, retryPolicy? }, async () => ...);          // 结果持久化,重放时直接返回
    await step.sleep(name, "3s" | "7 days");                            // 持久定时
    const s = await step.waitForSignal<T>({ name?, signal, timeout?, schema? }); // 超时返回 null,默认超时 1 年
    await step.sendSignal({ signal, data }); await step.runWorkflow(spec, input, { timeout });
  });
const h = await wf.run(input, { idempotencyKey?, availableAt?, deadlineAt? });   // h.workflowRun.id / h.result() / h.cancel()
await ow.sendSignal({ signal, data, idempotencyKey? });  // → { workflowRunIds }
await ow.cancelWorkflowRun(id);
const worker = ow.newWorker({ concurrency }); await worker.start(); await worker.stop();
backend.getWorkflowRun({ workflowRunId }); backend.listStepAttempts({ workflowRunId }); backend.listWorkflowRuns({ status, workflowName });
```

- 运行状态:`pending | running | sleeping(legacy) | completed | succeeded | failed | canceled`。等待信号 / sleep 期间,run 的 status 仍是 `running`,只是 `worker_id` 为空且 `available_at` 指向唤醒时间。
- 默认值(源码):工作流 `maximumAttempts: 1`;步骤 `maximumAttempts: 10`、`1s` 起指数退避、上限 `100s`;租约 `30s`,由心跳每 15s 续期(不可配置);单个 run 最多 1000 个 step attempt。
- 版本:注册表的键是 `(name, version)`;run 创建时记下 version,只有注册了同一 version 的 worker 才会执行它。

## 验证结果

### (a) 表初始化 / 迁移 —— 通过

`BackendPostgres.connect()` 默认执行幂等迁移(`CREATE SCHEMA IF NOT EXISTS` + `openworkflow_migrations` 版本表)。另有 `dist/postgres/scripts/db-migrate.js`(只认默认 URL,生产不宜直接用)。`fromPool()` 不跑迁移。

```
$ npx tsx openworkflow/cli.ts tables
biz.side_effect_ledger
openworkflow.openworkflow_migrations
openworkflow.step_attempts
openworkflow.workflow_runs
openworkflow.workflow_signals
```

建议:在 `packages/db` 的迁移流程里调用 `migrate(sql, "openworkflow")`(`openworkflow/internal` 导出),应用启动时用 `runMigrations:false`。**未验证** `openworkflow/internal` 的 `migrate` 直接调用。

### 主路径 —— 通过

```
{"runId":"bdc8f92d-…","status":"pending","version":"v1"}
{"status":"running",…}                       ← 停在 wait-download
{"workflowRunIds":["bdc8f92d-…"]}            ← signal download-done:d1
{"status":"completed","attempts":3,"output":{"sub":{"torrentHash":"hash-d1"},"copied":{"dest":"/lib/d1.mkv"},"signal":{"ok":true}}}
submit-download function completed / wait-download signal-wait completed / copy-file function completed
register function completed / retention sleep completed / cleanup function completed
```

### (b) kill -9 后重启恢复 —— 通过(语义:已完成步骤不重跑,正在执行的步骤至少执行一次)

copy-file 步骤内等待 10s;在复制途中对 worker `kill -9`,再启动新 worker A2:

```
kill 之后:  {"status":"running","attempts":2}  copy-file function running;available_at 比 now 晚约 16s(租约尚未到期)
A2 启动约 30s 后:{"status":"completed","attempts":4}
submit-download completed / wait-download completed / copy-file running(孤儿行) / copy-file completed / register / retention / cleanup
账本:
9e50537b submit-download  A      version=v1
9e50537b copy-file        A      start          ← 被 kill 的那次
9e50537b copy-file        A2     start          ← 同一步骤重跑
9e50537b copy-file        A2     end
9e50537b register         A2 / cleanup A2
```

- submit-download 与信号等待没有重跑;copy-file 在新 worker 上从头执行。所以步骤体必须幂等(例如复制前先查目标文件和操作日志)。
- 要等租约到期(最长约 30s)才会被重新认领。
- 被中断的 attempt 行一直停在 `running`,不会被清掉。展示时不要按这张表统计"执行中"。

### 租约丢失后的写入隔离(补充:SIGSTOP 模拟 worker 卡死) —— 引擎层通过,业务层没有隔离

对 A 执行 `kill -STOP`,由 B 接管并完成,再 `kill -CONT` A:

```
d23c7db1 copy-file  A  start
d23c7db1 copy-file  B  start / B end / register B / cleanup B    ← run completed
d23c7db1 copy-file  A  end                                        ← A 恢复后步骤体照常跑完并写了业务账本
Heartbeat failed: Error: Failed to extend lease for workflow run   ← A 的步骤结果被引擎拒绝(worker_id 不再匹配)
```

引擎保证过期 worker 的**步骤结果**不会写入;但步骤体内的**业务副作用**照常发生,而且步骤体拿不到 AbortSignal 或租约状态。REQ §11.1 要求"失去租约的 worker 不得再提交业务状态",这一条要靠业务侧的操作账本做 fencing(见 ADR-002)。

### (c) 信号先于等待到达 —— **会丢**

```
$ cli signal d2 early-key          → {"workflowRunIds":[]}         ← 还没有等待者,直接丢弃
$ cli start-acquire d2             → run 停在 wait-download(running)
$ cli signal d2 early-key          → {"workflowRunIds":["f837c0cb-…"]}   ← 同一个 idempotencyKey 仍能投递(前一次没落库)
$ cli signal d2 key2               → {"workflowRunIds":[]}         ← 等待已结束,再发就丢
```

源码 `sendSignal` 只会 `SELECT … step_attempts WHERE kind='signal-wait' AND status='running' FOR UPDATE`,没有等待者就返回空数组,不落库(类型注释也写了:"If no workflow is waiting, the signal is silently dropped")。同一信号名同一时间只允许一个等待步骤。**不能把 signal 当持久授权或完成通知。**

### (d) 取消传播 —— 部分通过

```
cancel slow(处于 long 步骤中) → {"status":"canceled"}
  账本:long start … long end (still ran after cancel?)   ← 进行中的步骤体跑完了(没有 AbortSignal)
  after-long 未执行;long 的 attempt 行停在 running
cancel acquire(处于 wait-download) → {"status":"canceled"},之后的步骤不再执行
  但随后 signal d4 → {"workflowRunIds":["282dde2b-…"]}   ← 对已取消的 run 仍然"投递成功"
cancel 已完成 run → Error: Cannot cancel workflow run … with status completed
```

取消只能保证"后续步骤不再执行"。进行中的步骤不会被中断,子工作流是否级联取消**未验证**。

### (e) 步骤失败重试与最大次数 —— 通过

```
flaky(前 2 次失败, maximumAttempts=3):completed, attempts=3
  flaky-step failed "boom attempt 1" / failed "boom attempt 2" / completed
flaky(前 5 次失败, maximumAttempts=3):failed, attempts=3, error.message="boom attempt 3"
  flaky-step failed ×3,不再重试
```

步骤重试的实现方式是整个 run 按退避重新排期、重放前面的步骤;run 的 `attempts` 会随之增长。重试期间 run 显示为 `pending`,错误写在 run.error 上。

### (f) idempotencyKey 去重 —— 通过(有限制)

```
start-acquire d5 same-key       → runId fd705d54…  pending
start-acquire d5 same-key       → runId fd705d54…  running   ← 返回已有 run
start-acquire d5-other same-key → runId fd705d54…            ← 输入不同也返回旧 run,不报冲突
select count(*) … where idempotency_key='same-key' → 1
```

去重范围是 `(namespace, workflowName, key)`,用 advisory lock 串行化,**窗口只有 24 小时**(`DEFAULT_RUN_IDEMPOTENCY_PERIOD_MS`)。引擎不比较输入。

### (g) 版本升级 —— 通过(要自己管好版本纪律)

1. run 以 `version=v1` 停在等待中;停掉 v1 worker,只起 `WF_VERSION=v2` 的 worker,然后发信号:
   ```
   {"status":"pending","attempts":2,"error":{"message":"Workflow \"acquire\" (version: v1) is not registered"}}
   ```
   run 没有失败,而是按退避无限重排(`MISSING_DEFINITION_RETRY_POLICY maximumAttempts:0`),等能处理 v1 的 worker 出现。
2. 再起一个 v1 worker,但代码改过(同一 version 在 copy 与 register 之间插入 `v2-extra-verify` 步骤):
   ```
   completed;steps: submit-download / wait-download / copy-file / v2-extra-verify / register / retention / cleanup
   ```
   重放按**步骤名**匹配历史,没有确定性检查。新插入的步骤会执行;已完成的步骤只要改了名,就会被当作新步骤重跑。

### (h) 与 Kysely + pg 共库不同 schema —— 通过

OpenWorkflow(postgres.js 连接池,schema `openworkflow`)和 Kysely(pg 连接池,schema `biz`)在同一个库里跑完了以上全部用例,没有冲突。限制:OpenWorkflow 自己管连接(postgres.js),`createWorkflowRun` 不能接收 Kysely 事务,所以**不能和业务写入在同一个事务里入队**,要用 outbox。

### (i) 两个 worker 进程并发 —— 通过

```
worker A、B 同时运行,投递 60 个 burst run
{"total":"60","runs":"60","a":"32","b":"28"}   ← 60 次执行、60 个 run,没有重复
completed|60
```

认领 SQL 用的是 `FOR UPDATE SKIP LOCKED`。

### 补充:PostgreSQL 重启 —— 通过

run 处于 15s 的 retention sleep 时执行 `docker restart umt-spike-pg`,约 30s 后 run 已 `completed`,6 个步骤都 completed;worker 日志里有 2 条 tick 失败,之后自动恢复。

## 未验证

- Deno 下运行(ADR-001 已选 Node)。
- `step.runWorkflow` 子工作流取消级联、`deadlineAt`、OpenTelemetry。
- 大量 sleeping run(>10k)下的轮询开销。
- "qB 接收后断连""复制完成但结果未提交"这类业务故障注入。它们属于业务幂等层,在 `packages/tasks` 集成测试中覆盖(见 ADR-002)。