fix(tasks): align OutboxRelay with outbox_event status/next_attempt_at/idempotency_key
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:<id>) - 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 <[email protected]>
This commit is contained in:
1 parent
608e567edf
commit
31729f2bac
4 files changed
+56
-15
No files matched your search
@@ -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) {
|
||||
|
||||
@@ -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" }]);
|
||||
});
|
||||
});
|
||||
Reference in new issue
Block a user