fix(server): check maxOpsPerMutation after the dedup lookup (ADR-0025 amendment)

CHANGELOG.md

@@ -131,6 +131,12 @@ While pre-1.0, the public API may change between 0.x releases.
   `rejected` (`code: "VALIDATION"`) before authorize or any row write, so the client
   gets a prompt `MutationRejectedError` and rolls back. **Behaviour change:** a
   write that used `""` as a pk used to be accepted and is now refused.
+- **A replayed `mut` is no longer rejected by a lowered op limit (ADR-0025
+  amendment).** `maxOpsPerMutation` was checked before the dedup lookup. If the
+  limit was lowered (for example, by a deploy) while a hold-and-replay
+  (ADR-0021) was pending, the replayed frame got `LIMIT_EXCEEDED` and the client
+  rolled back a write that had committed. The check now runs after the lookup,
+  so a resent `mut` gets its stored outcome while that receipt is retained.
 
 ### Internal
 

docs/adr/0012-wire-input-hardening.md

@@ -44,6 +44,8 @@ readonly`, override-able by subclasses at construction time):
 - **`maxOpsPerMutation = 128`**: checked at the top of `handleMut`. **Reject,
   don't truncate** — a partial apply would silently drop client writes. Sends
   `rejected` with `code: "LIMIT_EXCEEDED"`.
+  *(2026-09-25: now checked after the dedup lookup, so a resent `txId` keeps
+  its stored outcome; see [ADR-0025](./0025-empty-op-key-rejection.md).)*
 - **`maxSubsPerSocket = 256`**: checked in `handleSub` before `subs.add`. A
   re-sub on an existing `subId` replaces the old entry
   (`SubscriptionRegistry.add` semantics) and does NOT count against the cap —

docs/adr/0025-empty-op-key-rejection.md

@@ -65,3 +65,27 @@ tracked separately.
   replays identical frames, so only the first case can reach it. It is left as
   a follow-up.
 - Tests: `tests/empty-op-key.test.ts`.
+
+## Amendment — 2026-09-25: every per-tx rejection runs after the dedup lookup
+
+Rule 3 now applies to every check in `#handleMut` and `#handleCall` that can
+answer a `txId` with `rejected`, not only the empty-key check. The one check
+that ran before the lookup was `maxOpsPerMutation` (see the Consequences
+above). It now runs after it. This matters most for ADR-0021's
+hold-and-replay: the client resends an identical frame after a drop, and if
+the limit was lowered in between (a deploy over the same storage), a limit
+check that ran first would answer `LIMIT_EXCEEDED` and the client would roll
+back a write that had committed. `#handleCall` already looked up first.
+
+Checks that run before decode or before the shape guard are not per-tx
+rejections and stay where they are: `maxFrameBytes` and the shape guard drop
+the frame with no reply (ADR-0012 D1, ADR-0018), so they cannot contradict a
+stored receipt. `maxFrameBytes` must run before decode to bound the work.
+
+This rule covers a resend that arrives after the first frame has been
+answered. It does not cover two frames with the same `txId` in flight at
+once: both handlers can pass the lookup before either records an outcome,
+because `authorize` (and a command's `execute`) can await. That needs a
+per-`txId` reservation, which is a separate decision and is not made here.
+
+Tests: `tests/dedup-before-limits.test.ts`.

docs/adr/README.md

@@ -19,7 +19,7 @@ explains the displacement.
 | [0009](./0009-changelog-time-retention.md) | Changelog time-based retention; reset stale reconnects | Accepted |
 | [0010](./0010-typed-mutations-collection-manifest.md) | Typed mutations via a collection-row manifest on `SyncRegistry` | Accepted (manifest superseded by 0014) |
 | [0011](./0011-ssr-dehydrate-hydrate.md) | SSR: dehydrate on the worker, hydrate to the cursor | Accepted (experimental; generalizes 0002 C1 → C1′; amended for merged upstream + 0015/0016 lift; D5 peer floor superseded by 0022) |
-| [0012](./0012-wire-input-hardening.md) | Wire-input hardening: frame-shape guards, inbound limits, sanitized execute errors | Accepted (D1 amended by 0025 for op keys) |
+| [0012](./0012-wire-input-hardening.md) | Wire-input hardening: frame-shape guards, inbound limits, sanitized execute errors | Accepted (D1 amended by 0025 for op keys; D2 limit order amended by 0025) |
 | [0013](./0013-predicate-floor-one-evaluator.md) | Filtered-subscription membership: one evaluator is the source of truth; the floor is the verified-agreeing set | Accepted |
 | [0014](./0014-object-sync-schema.md) | `defineSync`: one schema value, mutations on the collection, commands on the connection | Accepted (supersedes 0001 D11 builder; closes 0010 manifest) |
 | [0015](./0015-syncable-mixin.md) | `Syncable` mixin: the sync core as a mixin over any DO base | Accepted (reframes 0001 D13; Actor unsupported; wake-test gap closed by 0019) |
@@ -31,4 +31,4 @@ explains the displacement.
 | [0021](./0021-in-flight-settlement-unexpected-close.md) | In-flight mut/call settlement across an unexpected close: hold-and-replay + typed `ConnectionLostError` | Accepted (fixes #39; builds on 0020/0016/0011) |
 | [0022](./0022-db-version-support-policy.md) | `@tanstack/db` version support: an uncapped floor, tested at both ends | Accepted (supersedes 0011 D5 peer floor in part) |
 | [0024](./0024-examples-single-tanstack-db.md) | Examples bundle exactly one `@tanstack/db`, and CI proves it | Accepted (examples only) |
-| [0025](./0025-empty-op-key-rejection.md) | An empty `mut` op key is rejected with a reply, after the dedup lookup | Accepted (amends 0012 D1 for op keys; makes 0014 D3 true) |
+| [0025](./0025-empty-op-key-rejection.md) | An empty `mut` op key is rejected with a reply, after the dedup lookup | Accepted (amends 0012 D1 for op keys; makes 0014 D3 true; amended: every per-tx rejection runs after the dedup lookup) |

src/server/mixin.ts

@@ -704,15 +704,17 @@ export function Syncable<Env = unknown, TUser = unknown>() {
        * optimistic overlay is never dropped before the authoritative row lands.
        */
       async #handleMut(ws: WebSocket, f: Extract<ClientFrame, { t: "mut" }>): Promise<void> {
+        // Dedup first: a resent txId gets its stored outcome, never a fresh
+        // rejection from a check below (ADR-0025).
+        const seen = lookupTx(this.#sql, f.txId)
+        if (seen) return this.#replayReceipt(ws, f.txId, seen)
+
         // Inbound limit: reject over-length batches without applying anything
         // (ADR-0012). Reject-don't-truncate: a partial apply silently drops writes.
         if (f.ops.length > this.maxOpsPerMutation) {
           return this.#rejectTx(ws, f.txId, `mutation exceeds maxOpsPerMutation (${this.maxOpsPerMutation})`, "LIMIT_EXCEEDED")
         }
 
-        const seen = lookupTx(this.#sql, f.txId)
-        if (seen) return this.#replayReceipt(ws, f.txId, seen)
-
         // An op key is the row's client-supplied TEXT pk; `""` is never a real
         // identity. Reject with a reply, not a shape-guard drop, so the client
         // rolls back at once instead of timing out (ADR-0025). After the dedup

tests/dedup-before-limits.test.ts

@@ -0,0 +1,126 @@
+import { env, runInDurableObject, SELF } from "cloudflare:test"
+import { describe, expect, it } from "vitest"
+import { createFrameCodec } from "../src/wire/frame-codec.ts"
+import type { ClientFrame, MutOp, ServerFrame } from "../src/wire/frames.ts"
+
+// WHY: a resent txId must get its STORED outcome (exactly-once, ADR-0002 C5), so
+// every per-tx rejection runs after the dedup lookup (ADR-0025, amended). The
+// sharp case is ADR-0021's hold-and-replay: the client resends an identical
+// frame after a drop. If `maxOpsPerMutation` was lowered in between (a deploy
+// over the same storage), a limit check that ran first would answer
+// LIMIT_EXCEEDED, and the client would roll back a write that committed.
+
+const codec = createFrameCodec()
+
+async function openWs(path: string): Promise<WebSocket> {
+  const res = await SELF.fetch(`https://example.com${path}`, { headers: { Upgrade: "websocket" } })
+  expect(res.status).toBe(101)
+  const ws = res.webSocket
+  if (!ws) throw new Error("no webSocket on 101 response")
+  ws.accept()
+  return ws
+}
+
+function collectUntil(ws: WebSocket, done: (f: ServerFrame) => boolean, timeoutMs = 2000): Promise<Array<ServerFrame>> {
+  return new Promise((resolve, reject) => {
+    const out: Array<ServerFrame> = []
+    const timer = setTimeout(() => reject(new Error(`timeout; got [${out.map((f) => f.t).join(",")}]`)), timeoutMs)
+    const onMsg = (e: MessageEvent): void => {
+      out.push(codec.decode(e.data as ArrayBuffer) as ServerFrame)
+      if (done(out[out.length - 1]!)) {
+        clearTimeout(timer)
+        ws.removeEventListener("message", onMsg)
+        resolve(out)
+      }
+    }
+    ws.addEventListener("message", onMsg)
+  })
+}
+
+/** Send a mut and return its receipt (`committed` or `rejected`). */
+async function mut(ws: WebSocket, txId: string, ops: Array<MutOp>): Promise<ServerFrame> {
+  const frame: ClientFrame = { t: "mut", txId, collection: "messages", ops }
+  ws.send(codec.encode(frame))
+  return (await collectUntil(ws, (f) => f.t === "committed" || f.t === "rejected")).at(-1)!
+}
+
+const inserts = (...ids: Array<string>): Array<MutOp> =>
+  ids.map((id) => ({ type: "insert", key: id, cols: { id, body: id } }))
+
+async function rowIds(ns: DurableObjectNamespace, room: string): Promise<Array<string>> {
+  return runInDurableObject(ns.get(ns.idFromName(room)), (_i, s) =>
+    Array.from(s.storage.sql.exec("SELECT id FROM messages ORDER BY id")).map((r) => (r as { id: string }).id),
+  )
+}
+
+describe("dedup lookup runs before the per-mutation op limit (ADR-0025 amendment)", () => {
+  it("(a) committed T replayed with the SAME payload after the limit is lowered → stored committed", async () => {
+    const room = "dbl-lowered"
+    const ws = await openWs(`/sync/${room}`)
+    const first = await mut(ws, "dbl-a1", inserts("a1", "a2"))
+    expect(first.t).toBe("committed")
+
+    // Stand-in for a deploy that lowers the limit over the same storage: the
+    // limit is a construction-time field, and the dedup table lives in SQLite,
+    // so setting the field on the live instance is the same state a new
+    // instance of a redeployed class would see.
+    await runInDurableObject(env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), (i) => {
+      ;(i as unknown as { maxOpsPerMutation: number }).maxOpsPerMutation = 1
+    })
+
+    const again = await mut(ws, "dbl-a1", inserts("a1", "a2"))
+    expect(again).toMatchObject({ t: "committed", txId: "dbl-a1", seq: (first as { seq: string }).seq })
+    expect(await rowIds(env.SYNC_DO, room)).toEqual(["a1", "a2"])
+
+    // The lowered limit does apply to a NEW txId: it is a real limit, not a no-op.
+    const fresh = await mut(ws, "dbl-a2", inserts("a3", "a4"))
+    expect(fresh).toMatchObject({ t: "rejected", error: { code: "LIMIT_EXCEEDED" } })
+    ws.close()
+  })
+
+  it("a LIMIT_EXCEEDED rejection is itself recorded: after the limit is raised, the same frame replays it", async () => {
+    // LimitsTestDO: maxOpsPerMutation = 2.
+    const room = "dbl-raised"
+    const ws = await openWs(`/limits/${room}`)
+    const first = await mut(ws, "dbl-x1", inserts("x1", "x2", "x3"))
+    expect(first).toMatchObject({ t: "rejected", error: { code: "LIMIT_EXCEEDED" } })
+
+    await runInDurableObject(env.LIMITS_DO.get(env.LIMITS_DO.idFromName(room)), (i) => {
+      ;(i as unknown as { maxOpsPerMutation: number }).maxOpsPerMutation = 128
+    })
+
+    const again = await mut(ws, "dbl-x1", inserts("x1", "x2", "x3"))
+    expect(again).toEqual(first)
+    expect(await rowIds(env.LIMITS_DO, room)).toEqual([])
+    ws.close()
+  })
+
+  // The next two resend a DIFFERENT payload under a used txId. Our client never
+  // does that (hold-and-replay resends the identical frame); a non-conforming
+  // client can, and the stored outcome must still win.
+  it("(b) committed T resent with a different, over-limit payload → stored committed, nothing applied", async () => {
+    // LimitsTestDO: maxOpsPerMutation = 2.
+    const room = "dbl-bigger"
+    const ws = await openWs(`/limits/${room}`)
+    const first = await mut(ws, "dbl-b1", inserts("b1"))
+    expect(first.t).toBe("committed")
+
+    const again = await mut(ws, "dbl-b1", inserts("b2", "b3", "b4"))
+    expect(again).toMatchObject({ t: "committed", txId: "dbl-b1", seq: (first as { seq: string }).seq })
+    expect(await rowIds(env.LIMITS_DO, room)).toEqual(["b1"])
+    ws.close()
+  })
+
+  it("a rejected T resent over the limit replays its stored rejection, not LIMIT_EXCEEDED", async () => {
+    const room = "dbl-rejected"
+    const ws = await openWs(`/limits/${room}`)
+    // The test schema's insert authorize denies body "FORBIDDEN".
+    const first = await mut(ws, "dbl-r1", [{ type: "insert", key: "r1", cols: { id: "r1", body: "FORBIDDEN" } }])
+    expect(first).toMatchObject({ t: "rejected", error: { message: "forbidden body" } })
+
+    const again = await mut(ws, "dbl-r1", inserts("r2", "r3", "r4"))
+    expect(again).toEqual(first)
+    expect(await rowIds(env.LIMITS_DO, room)).toEqual([])
+    ws.close()
+  })
+})