From 31729f2bac5629fdc024e35314e723e3f7de1520 Mon Sep 17 00:00:00 2001 From: root Date: Sun, 27 Sep 2026 18:59:06 +0800 Subject: [PATCH] fix(tasks): align OutboxRelay with outbox_event status/next_attempt_at/idempotency_key MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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:) - 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 --- packages/tasks/README.md | 2 +- packages/tasks/src/outbox-relay.ts | 28 ++++++++++++++++++-------- packages/tasks/test/helpers.ts | 9 +++++++-- packages/tasks/test/outbox.test.ts | 32 ++++++++++++++++++++++++++---- 4 files changed, 56 insertions(+), 15 deletions(-) diff --git a/packages/tasks/README.md b/packages/tasks/README.md index b2af351b..8aa5b80a 100644 --- a/packages/tasks/README.md +++ b/packages/tasks/README.md @@ -14,7 +14,7 @@ | --- | --- | | `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:`),再标记 `published_at`;多进程并发时不会重复 | +| `OutboxRelay` | 用 `FOR UPDATE SKIP LOCKED` 认领 `outbox_event` 中 `status=pending` 且 `next_attempt_at` 已到的行,按 type 路由后 enqueue(幂等键取行的 `idempotency_key`,缺省 `outbox:`),成功标记 `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` | diff --git a/packages/tasks/src/outbox-relay.ts b/packages/tasks/src/outbox-relay.ts index bc74c779..ca2606a7 100644 --- a/packages/tasks/src/outbox-relay.ts +++ b/packages/tasks/src/outbox-relay.ts @@ -31,12 +31,16 @@ export interface OutboxRelayOptions { routes: Record; batchSize?: number; intervalMs?: number; + /** 失败达到该次数后置 status='dead'(默认 10);其间按 2^attempts 秒退避,上限 5 分钟 */ + maxAttempts?: number; logSink?: LogSink; } /** * Outbox 中继:业务事务写 outbox_event,中继在自己的事务里 - * `SELECT … FOR UPDATE SKIP LOCKED` 认领未投递行 → enqueue(幂等键 = outbox:<行 id>)→ 标记 published_at。 + * `SELECT … FOR UPDATE SKIP LOCKED` 认领 status='pending' 且 next_attempt_at 已到的行 + * → enqueue(幂等键 = 行的 idempotency_key,缺省 outbox:<行 id>)→ 标记 dispatched + published_at; + * 失败则 attempts+1、退避 next_attempt_at、记 last_error,超过 maxAttempts 置 dead(ADR-002 §2.1)。 * 多个中继进程并发时互不重复认领;enqueue 成功但标记前崩溃,重投时由引擎幂等键去重(24h), * 更长的去重由业务唯一约束兜底。 */ @@ -61,9 +65,9 @@ export class OutboxRelay { let published = 0; let failed = 0; await this.opts.db.transaction().execute(async (trx) => { - const rows = await sql` - SELECT id, type, payload, at FROM outbox_event - WHERE published_at IS NULL AND type = ANY(${types}::text[]) + const rows = await sql` + SELECT id, type, payload, at, idempotency_key, attempts FROM outbox_event + WHERE status = 'pending' AND next_attempt_at <= now() AND type = ANY(${types}::text[]) ORDER BY at, id LIMIT ${this.opts.batchSize ?? 50} FOR UPDATE SKIP LOCKED`.execute(trx); @@ -73,17 +77,25 @@ export class OutboxRelay { if (route) { const payloadTrace = (ev.payload as { traceId?: unknown } | null)?.traceId; await this.opts.engine.enqueue(route.task, route.input, { - idempotencyKey: OutboxRelay.idempotencyKey(ev.id), + idempotencyKey: ev.idempotency_key ?? OutboxRelay.idempotencyKey(ev.id), traceId: route.traceId ?? (typeof payloadTrace === "string" ? payloadTrace : randomUUID()), ...(route.priority === undefined ? {} : { priority: route.priority }), }); } - await sql`UPDATE outbox_event SET published_at = now() WHERE id = ${ev.id}`.execute(trx); + await sql`UPDATE outbox_event SET status = 'dispatched', published_at = now(), attempts = attempts + 1, + last_error = NULL WHERE id = ${ev.id}`.execute(trx); published++; } catch (err) { failed++; - // 留在未投递状态,下一轮重试 - this.log.error("outbox dispatch failed", { eventId: ev.id, type: ev.type, err }); + // 退避后重试;超过上限置 dead(需人工处理) + const attempts = ev.attempts + 1; + const dead = attempts >= (this.opts.maxAttempts ?? 10); + const delayS = Math.min(2 ** attempts, 300); + const lastError = JSON.stringify({ message: err instanceof Error ? err.message : String(err) }); + await sql`UPDATE outbox_event SET attempts = ${attempts}, status = ${dead ? "dead" : "pending"}, + next_attempt_at = now() + make_interval(secs => ${delayS}), last_error = ${lastError}::jsonb + WHERE id = ${ev.id}`.execute(trx); + this.log.error("outbox dispatch failed", { eventId: ev.id, type: ev.type, attempts, dead, err }); } } }); diff --git a/packages/tasks/test/helpers.ts b/packages/tasks/test/helpers.ts index ad01bb56..cfc307d0 100644 --- a/packages/tasks/test/helpers.ts +++ b/packages/tasks/test/helpers.ts @@ -58,14 +58,19 @@ export async function dropSchemas(schemas: { biz: string; ow: string }) { } } -/** 与 packages/db 0001_init 的 outbox_event 同构 */ +/** 与 packages/db 0001_init + 0002_foundation 的 outbox_event 同构 */ export async function createOutboxTable(db: TestDb) { await sql`CREATE TABLE outbox_event ( id uuid PRIMARY KEY DEFAULT gen_random_uuid(), at timestamptz NOT NULL DEFAULT now(), type text NOT NULL, payload jsonb NOT NULL DEFAULT '{}'::jsonb, - published_at timestamptz)`.execute(db); + published_at timestamptz, + idempotency_key text UNIQUE, + status text NOT NULL DEFAULT 'pending' CHECK (status IN ('pending', 'dispatched', 'failed', 'dead')), + attempts integer NOT NULL DEFAULT 0, + next_attempt_at timestamptz NOT NULL DEFAULT now(), + last_error jsonb)`.execute(db); } export async function createSideEffectTable(db: TestDb) { diff --git a/packages/tasks/test/outbox.test.ts b/packages/tasks/test/outbox.test.ts index a2570a58..4360ab11 100644 --- a/packages/tasks/test/outbox.test.ts +++ b/packages/tasks/test/outbox.test.ts @@ -74,8 +74,13 @@ describe.skipIf(!hasPg)("OutboxRelay", () => { expect(calls).toHaveLength(60); expect(new Set(calls).size).toBe(60); - const pending = await sql<{ type: string }>`SELECT type FROM outbox_event WHERE published_at IS NULL`.execute(db); + const pending = await sql<{ type: string }>`SELECT type FROM outbox_event WHERE status = 'pending'`.execute(db); expect(pending.rows).toEqual([{ type: "other.type" }]); // 未配置路由的 type 不认领 + const dispatched = await sql<{ n: string }>` + SELECT count(*) n FROM outbox_event WHERE status = 'dispatched' AND published_at IS NOT NULL AND attempts = 1`.execute( + db, + ); + expect(dispatched.rows[0]!.n).toBe("60"); const runs = await sql<{ n: string; keys: string }>` SELECT count(*) n, count(DISTINCT idempotency_key) keys FROM tasks_run_meta WHERE task_name = 'echo'`.execute(db); @@ -98,8 +103,27 @@ describe.skipIf(!hasPg)("OutboxRelay", () => { logSink: () => {}, }); expect(await relay.pollOnce()).toEqual({ published: 0, failed: 1 }); - const left = await sql<{ n: string }>` - SELECT count(*) n FROM outbox_event WHERE published_at IS NULL AND type = 'echo.requested'`.execute(db); - expect(left.rows[0]!.n).toBe("1"); + const left = await sql<{ status: string; attempts: number; backoff: boolean; err: string }>` + SELECT status, attempts, next_attempt_at > now() backoff, last_error->>'message' err FROM outbox_event + WHERE published_at IS NULL AND type = 'echo.requested'`.execute(db); + expect(left.rows).toHaveLength(1); + expect(left.rows[0]).toMatchObject({ status: "pending", attempts: 1, backoff: true }); + // 退避期内不再认领;到期后再失败达到上限置 dead + expect(await relay.pollOnce()).toEqual({ published: 0, failed: 0 }); + await sql`UPDATE outbox_event SET next_attempt_at = now() WHERE type = 'echo.requested' AND published_at IS NULL`.execute( + db, + ); + const strict = new OutboxRelay({ + db, + engine, + routes: { "echo.requested": (ev) => ({ task: "echo", input: ev.payload }) }, + logSink: () => {}, + maxAttempts: 2, + }); + expect(await strict.pollOnce()).toEqual({ published: 0, failed: 1 }); + const dead = await sql<{ + status: string; + }>`SELECT status FROM outbox_event WHERE type = 'echo.requested' AND published_at IS NULL`.execute(db); + expect(dead.rows).toEqual([{ status: "dead" }]); }); });