From d0cd52430947c25fe10a2ae3fdf5359f7408e142 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 22 Jul 2026 12:53:36 +1000 Subject: [PATCH] fix(server): persist rejection code for tx-dedup replay MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A rejected tx's machine-readable error code was sent on the live socket but never recorded, so a client retrying the same txId replayed the reason message with no code — breaking code-based handling on exactly the retry path dedup exists for. _sync_seen_tx gains an error_code column; initSchema adds it in place (pragma-gated ALTER) so already-deployed DOs upgrade on wake. The replayed rejected frame is now shaped identically to the original: { code, message } when a code was recorded, { message } otherwise. Fixes #21 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 10 ++++++++++ src/server/changes.ts | 20 ++++++++++++++------ src/server/dedup.ts | 14 ++++++++++---- src/server/mixin.ts | 12 ++++++++---- tests/error-paths.test.ts | 30 ++++++++++++++++++++++++++++++ tests/schema-evolution.test.ts | 31 +++++++++++++++++++++++++++++++ 6 files changed, 103 insertions(+), 14 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index a101205..765b013 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,16 @@ While pre-1.0, the public API may change between 0.x releases. ## [Unreleased] +### Fixed + +- **Rejection `code` now survives tx-dedup replay (#21).** The dedup record + persisted only the rejection message, so a client retrying the same `txId` + got the reason with no machine-readable `code` — breaking code-based error + handling on exactly the retry path it exists for. `_sync_seen_tx` gains an + `error_code` column (added in place on wake for already-deployed DOs), and + the replayed `rejected` frame is now shaped identically to the original: + `{ code, message }` when a code was recorded, `{ message }` otherwise. + ### Changed - **Cohosting docs moved to `recipes/cohosting.md` and re-grounded.** The diff --git a/src/server/changes.ts b/src/server/changes.ts index c1fe98a..6df9152 100644 --- a/src/server/changes.ts +++ b/src/server/changes.ts @@ -40,13 +40,21 @@ export function initSchema(sql: SqlStorage): void { sql.exec(`CREATE TABLE IF NOT EXISTS _sync_meta (k TEXT PRIMARY KEY, v TEXT)`) // Mutation/command dedup (exactly-once under retry). See dedup.ts. sql.exec(`CREATE TABLE IF NOT EXISTS _sync_seen_tx ( - tx_id TEXT PRIMARY KEY, - ok INTEGER NOT NULL, - cursor TEXT, - error TEXT, - result TEXT, - ts INTEGER NOT NULL + tx_id TEXT PRIMARY KEY, + ok INTEGER NOT NULL, + cursor TEXT, + error TEXT, + error_code TEXT, + result TEXT, + ts INTEGER NOT NULL )`) + // `error_code` was added after tables shipped without it (issue #21): a DO + // created earlier wakes with the old shape, and CREATE IF NOT EXISTS won't + // touch it. ADD COLUMN isn't idempotent on its own, so gate on table_info. + const seenTxCols = Array.from(sql.exec<{ name: string }>("SELECT name FROM pragma_table_info('_sync_seen_tx')")) + if (!seenTxCols.some((c) => c.name === "error_code")) { + sql.exec("ALTER TABLE _sync_seen_tx ADD COLUMN error_code TEXT") + } } /** diff --git a/src/server/dedup.ts b/src/server/dedup.ts index a6ca546..2b5bb23 100644 --- a/src/server/dedup.ts +++ b/src/server/dedup.ts @@ -13,20 +13,24 @@ export interface SeenTx { ok: boolean cursor: string | null error: string | null + /** Machine-readable rejection code (e.g. VALIDATION); null when the original + * rejection carried none. Persisted so a replay is shaped identically to the + * first response (issue #21). */ + errorCode: string | null /** Value-codec-encoded result for commands; null for none. */ result: string | null } export function lookupTx(sql: SqlStorage, txId: string): SeenTx | null { const rows = Array.from( - sql.exec<{ ok: number; cursor: string | null; error: string | null; result: string | null }>( - "SELECT ok, cursor, error, result FROM _sync_seen_tx WHERE tx_id = ?", + sql.exec<{ ok: number; cursor: string | null; error: string | null; error_code: string | null; result: string | null }>( + "SELECT ok, cursor, error, error_code, result FROM _sync_seen_tx WHERE tx_id = ?", txId, ), ) if (rows.length === 0) return null const r = rows[0]! - return { ok: r.ok === 1, cursor: r.cursor, error: r.error, result: r.result } + return { ok: r.ok === 1, cursor: r.cursor, error: r.error, errorCode: r.error_code, result: r.result } } export function recordTx( @@ -35,16 +39,18 @@ export function recordTx( ok: boolean, cursor: string | null, error: string | null, + errorCode: string | null, result: string | null, ): void { // INSERT OR IGNORE: the first recorded outcome wins; a racing retry never // clobbers it. sql.exec( - "INSERT OR IGNORE INTO _sync_seen_tx(tx_id, ok, cursor, error, result, ts) VALUES (?, ?, ?, ?, ?, unixepoch()*1000)", + "INSERT OR IGNORE INTO _sync_seen_tx(tx_id, ok, cursor, error, error_code, result, ts) VALUES (?, ?, ?, ?, ?, ?, unixepoch()*1000)", txId, ok ? 1 : 0, cursor, error, + errorCode, result, ) } diff --git a/src/server/mixin.ts b/src/server/mixin.ts index fa73478..cb4eb35 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -570,7 +570,7 @@ export function Syncable() { return this.#rejectTx(ws, f.txId, "mutation failed", "EXECUTE_FAILED") } - recordTx(this.#sql, f.txId, true, commitSeq, null, null) + recordTx(this.#sql, f.txId, true, commitSeq, null, null, null) // Enqueue deltas for all subscribers, then flush THIS socket before its // receipt (C1) so its deltas land first. Other subscribers flush on the // coalescer tick. @@ -638,14 +638,14 @@ export function Syncable() { } const commitSeq = String(currentSeq(this.#sql)) - recordTx(this.#sql, f.txId, true, commitSeq, null, stored) + recordTx(this.#sql, f.txId, true, commitSeq, null, null, stored) this.#drainAndBroadcast() this.#broadcaster.flushOne(ws) this.#send(ws, { t: "committed", txId: f.txId, seq: commitSeq, result }) } #rejectTx(ws: WebSocket, txId: string, message: string, code?: string): void { - recordTx(this.#sql, txId, false, null, message, null) + recordTx(this.#sql, txId, false, null, message, code ?? null, null) this.#send(ws, { t: "rejected", txId, error: code ? { code, message } : { message } }) } @@ -653,7 +653,11 @@ export function Syncable() { if (seen.ok) { this.#send(ws, { t: "committed", txId, seq: seen.cursor ?? "0", result: decodeResult(seen.result) }) } else { - this.#send(ws, { t: "rejected", txId, error: { message: seen.error ?? "unknown" } }) + // Shape the replay exactly like #rejectTx's original frame — with the + // persisted code when there was one (issue #21) — so a retrying client's + // code-based handling sees the same outcome either way. + const message = seen.error ?? "unknown" + this.#send(ws, { t: "rejected", txId, error: seen.errorCode ? { code: seen.errorCode, message } : { message } }) } } diff --git a/tests/error-paths.test.ts b/tests/error-paths.test.ts index dbc405f..c567613 100644 --- a/tests/error-paths.test.ts +++ b/tests/error-paths.test.ts @@ -170,6 +170,36 @@ describe("mutation error paths (atomicity, fail-loud, exactly-once)", () => { const r2 = second[second.length - 1]! as Extract expect(r2.t).toBe("rejected") expect(r2.error.message).toBe(r1.error.message) + // A code-less rejection must replay code-less — the replayed frame is shaped + // identically to the original, so a client can't tell replay from first send. + expect(r1.error.code).toBeUndefined() + expect(r2.error.code).toBeUndefined() + ws.close() + }) + + it("a retried CODED rejection replays with the identical code (issue #21)", async () => { + // WHY: the machine-readable `code` exists so clients can branch on the cause + // (VALIDATION vs EXECUTE_FAILED, ...) — and the dedup replay path is exactly + // where a retrying client re-reads the outcome. If the code is dropped from + // the persisted record, code-based handling breaks precisely on the retry it + // was built for, while the first response looks fine in every test. + const ws = await openWs("/sync/err-retry-coded") + // Empty `body` fails the `validated` collection's schema → VALIDATION code. + const frame: ClientFrame = { t: "mut", txId: "v1", collection: "validated", ops: [{ type: "insert", key: "a", cols: { id: "a", body: "" } }] } + + send(ws, frame) + const first = await collectUntil(ws, (f) => f.t === "rejected" || f.t === "committed") + const r1 = first[first.length - 1]! as Extract + expect(r1.t).toBe("rejected") + expect(r1.error.code).toBe("VALIDATION") + + // Retry the exact same frame — the replayed rejection must carry the same code. + send(ws, frame) + const second = await collectUntil(ws, (f) => f.t === "rejected" || f.t === "committed") + const r2 = second[second.length - 1]! as Extract + expect(r2.t).toBe("rejected") + expect(r2.error.code).toBe("VALIDATION") + expect(r2.error.message).toBe(r1.error.message) ws.close() }) }) diff --git a/tests/schema-evolution.test.ts b/tests/schema-evolution.test.ts index f8e50e1..9e54e04 100644 --- a/tests/schema-evolution.test.ts +++ b/tests/schema-evolution.test.ts @@ -1,6 +1,7 @@ import { env, runInDurableObject } from "cloudflare:test" import { describe, expect, it } from "vitest" import { assertSyncCompatible, hydrateRows, initSchema, installTriggers, readChangesSince } from "../src/server/changes.ts" +import { lookupTx, recordTx } from "../src/server/dedup.ts" // WHY: ADR-0007 makes the author own schema + migrations and only validates the // real table at registerSync. Two promises ride on that and are easy to assert @@ -68,6 +69,36 @@ describe("author-owned schema evolution (ADR-0007)", () => { }) }) + it("upgrades a pre-error_code _sync_seen_tx on wake, idempotently", async () => { + // WHY: `_sync_seen_tx` is CREATE IF NOT EXISTS, so a DO deployed before the + // `error_code` column (issue #21) wakes with the old shape and CREATE alone + // will never add it — recordTx/lookupTx would then throw on every dedup hit. + // initSchema must ALTER the old table in place, and stay idempotent since it + // re-runs on every wake. + await inDO("seen-tx-migrate", (sql) => { + // Simulate the already-deployed DO: the table exists in its OLD shape. + sql.exec(`CREATE TABLE _sync_seen_tx ( + tx_id TEXT PRIMARY KEY, + ok INTEGER NOT NULL, + cursor TEXT, + error TEXT, + result TEXT, + ts INTEGER NOT NULL + )`) + sql.exec("INSERT INTO _sync_seen_tx(tx_id,ok,cursor,error,result,ts) VALUES('old',0,null,'boom',null,0)") + + initSchema(sql) // next wake + initSchema(sql) // and the one after — ALTER must not run twice + + const cols = Array.from(sql.exec<{ name: string }>("SELECT name FROM pragma_table_info('_sync_seen_tx')")).map((c) => c.name) + expect(cols).toContain("error_code") + // Pre-migration rows replay code-less; post-migration writes carry the code. + expect(lookupTx(sql, "old")).toMatchObject({ ok: false, error: "boom", errorCode: null }) + recordTx(sql, "new", false, null, "invalid", "VALIDATION", null) + expect(lookupTx(sql, "new")).toMatchObject({ error: "invalid", errorCode: "VALIDATION" }) + }) + }) + it("accepts any TEXT-affinity pk (VARCHAR/CHAR) and rejects INTEGER keys", async () => { await inDO("affinity", (sql) => { sql.exec(`CREATE TABLE v (id VARCHAR PRIMARY KEY)`) // TEXT affinity