diff --git a/CHANGELOG.md b/CHANGELOG.md index 4dfa4c9..6788137 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/docs/adr/0012-wire-input-hardening.md b/docs/adr/0012-wire-input-hardening.md index 741b9f1..519a6d1 100644 --- a/docs/adr/0012-wire-input-hardening.md +++ b/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 — diff --git a/docs/adr/0025-empty-op-key-rejection.md b/docs/adr/0025-empty-op-key-rejection.md index 0b1bfb4..7df3d7d 100644 --- a/docs/adr/0025-empty-op-key-rejection.md +++ b/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`. diff --git a/docs/adr/README.md b/docs/adr/README.md index 9c9594c..592200a 100644 --- a/docs/adr/README.md +++ b/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) | diff --git a/src/server/mixin.ts b/src/server/mixin.ts index 19ff404..ca04c1d 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -704,15 +704,17 @@ export function Syncable() { * optimistic overlay is never dropped before the authoritative row lands. */ async #handleMut(ws: WebSocket, f: Extract): Promise { + // 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 diff --git a/tests/dedup-before-limits.test.ts b/tests/dedup-before-limits.test.ts new file mode 100644 index 0000000..e4ca919 --- /dev/null +++ b/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 { + 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> { + return new Promise((resolve, reject) => { + const out: Array = [] + 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): Promise { + 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): Array => + ids.map((id) => ({ type: "insert", key: id, cols: { id, body: id } })) + +async function rowIds(ns: DurableObjectNamespace, room: string): Promise> { + 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() + }) +})