fix(server): persist rejection code for tx-dedup replay

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

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")
+  }
 }
 
 /**

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,
   )
 }

src/server/mixin.ts

@@ -570,7 +570,7 @@ export function Syncable<Env = unknown, TUser = unknown>() {
           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,22 +638,26 @@ export function Syncable<Env = unknown, TUser = unknown>() {
         }
 
         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 } })
       }
 
       #replayReceipt(ws: WebSocket, txId: string, seen: SeenTx): void {
         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 } })
         }
       }
 

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<ServerFrame, { t: "rejected" }>
     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<ServerFrame, { t: "rejected" }>
+    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<ServerFrame, { t: "rejected" }>
+    expect(r2.t).toBe("rejected")
+    expect(r2.error.code).toBe("VALIDATION")
+    expect(r2.error.message).toBe(r1.error.message)
     ws.close()
   })
 })

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