From bedf20d9784001edb40bf493b47c475232fb6f40 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 22 Jul 2026 14:48:44 +1000 Subject: [PATCH 1/3] =?UTF-8?q?feat:=20typed=20oversize-frame=20handling?= =?UTF-8?q?=20=E2=80=94=20client=20pre-send=20guard,=20outbound=20size=20w?= =?UTF-8?q?arning=20(ADR-0018)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Part of #28 (part b; column projection, part a, stays open). An oversize client mutation was dropped silently server-side (ADR-0012's maxFrameBytes guard), surfacing only as a confirmation timeout — and in production Cloudflare's ~1 MiB edge cap on inbound WS messages means the frame may never reach the DO at all, so a server-side rejection cannot be the surface. Outbound had no size awareness whatsoever. - Client: new `maxFrameBytes` transport option (default 1 MiB, aligned with the server and the edge cap). An oversize mut/call rejects locally and immediately with MutationRejectedError code "FRAME_TOO_LARGE" — typed, prompt optimistic rollback, no timeout. Encoded once; bytes reused for the send. String (JSON debug codec) frames are measured in UTF-8 bytes, not UTF-16 code units (codex review). - Server: the ADR-0012 silent drop stays as defense in depth — recovering the txId would mean decoding the very payload the guard exists not to decode; the comment now says so. - Outbound: new `warnOutboundFrameBytes` knob (default 1 MiB, null disables) — console.warn with frame type, size, and collection when an encoded outbound frame exceeds it. Observability only; the frame is still sent whole (column projection, #28a, is the real fix). - A synchronous send() throw now cleans up the pending-receipt waiter and timer before rejecting (codex review). Tests: tests/frame-limits.test.ts pins the typed rejection (raw transport + collection-level rollback), UTF-8 measurement, the outbound warning, and the still-true large-row / read-only / tiny-collection behaviors. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 18 +- docs/adr/0018-oversize-frames.md | 106 ++++++++ docs/adr/README.md | 1 + src/client/transport.ts | 45 +++- src/server/mixin.ts | 36 ++- src/server/sync-do.ts | 1 + tests/frame-limits.test.ts | 447 +++++++++++++++++++++++++++++++ 7 files changed, 648 insertions(+), 6 deletions(-) create mode 100644 docs/adr/0018-oversize-frames.md create mode 100644 tests/frame-limits.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index c7c4e4c..c356322 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,7 +8,23 @@ While pre-1.0, the public API may change between 0.x releases. ## [Unreleased] -### Changed +### Added + +- **Typed oversize-frame handling (ADR-0018; part of + [#28](https://github.com/grrowl/tanstack-durable-object-sync/issues/28)).** + An oversize client mutation used to be dropped silently server-side + (ADR-0012's `maxFrameBytes` guard), surfacing only as a confirmation + timeout — and in production Cloudflare's ~1 MiB edge cap on inbound + WebSocket messages means the frame may never reach the DO at all. The + transport now guards **before sending**: a `mut`/`call` whose encoded size + exceeds the new `maxFrameBytes` option (default 1 MiB, aligned with the + server and the edge cap) rejects immediately with `MutationRejectedError` + (code `"FRAME_TOO_LARGE"`), so the optimistic overlay rolls back promptly. + The server's silent drop stays as defense in depth. Outbound gains a + warn-only threshold, `warnOutboundFrameBytes` (default 1 MiB, `null` + disables): a larger encoded frame logs a `console.warn` with size and + collection but is still sent whole — column projection (#28a) remains the + real fix for oversized full-row re-sends. - **Transport reconnect policy (ADR-0016; fixes #25, #26).** One option, `reconnectDelay?: number | ((attempt, closeCode?, closeReason?) => number | null)` diff --git a/docs/adr/0018-oversize-frames.md b/docs/adr/0018-oversize-frames.md new file mode 100644 index 0000000..35b17fe --- /dev/null +++ b/docs/adr/0018-oversize-frames.md @@ -0,0 +1,106 @@ +# ADR-0018: Oversize frames — client pre-send guard is the rejection surface; outbound is warn-only + +**Status**: Accepted +**Date**: 2026-07-22 +**Issue**: [#28](https://github.com/grrowl/tanstack-durable-object-sync/issues/28) (part b; column projection, part a, remains open) + +## Context + +ADR-0012's `maxFrameBytes` guard drops an oversize inbound frame with a +server-side `console.error` and **no reply**. For a hostile frame that is the +right stance; for an honest client whose mutation happens to carry a large +value (a multi-MB TEXT column, say) it is the worst failure mode we have: the +`mut` vanishes, the client's `sendMut` waits out its full confirmation +timeout, and the optimistic overlay rolls back only via the generic timeout +error — untyped, uncoded, seconds late. + +Outbound had no size awareness at all: `drainAndBroadcast` hydrates full rows +(`SELECT *`), so a small-field UPDATE on a row holding a large column re-sends +the whole row per tick per subscriber, silently. + +The load-bearing asymmetry (verified in workerd; documented in Cloudflare's +WebSocket limits): + +- **Inbound** (client → DO) WebSocket messages are capped at ~1 MiB **at the + edge** in production. An oversize client frame may never reach the DO at + all — workerd may refuse the send or close the socket before + `webSocketMessage` runs. A server-side rejection therefore *cannot* be the + primary surface: there may be no server-side anything. +- **Outbound** (DO → client) is not capped the same way; a >1 MiB frame is + delivered whole (pinned by `tests/frame-limits.test.ts`, snapshot and live + delta). + +## Decisions + +### D1: Client pre-send guard — the reliable half + +`WebSocketTransport` gains `maxFrameBytes` (default `1_048_576`, aligned with +the server's ADR-0012 default and the edge cap). Before sending a `mut` or +`call`, the transport encodes the frame once, checks the encoded size, and on +breach rejects **locally and immediately** with the existing typed surface: +`MutationRejectedError` with code `"FRAME_TOO_LARGE"`. No round trip, no +timeout; the optimistic overlay rolls back promptly through TanStack's normal +rejected-mutation path. The encoded bytes are reused for the actual send, so +the guard costs no extra encode. + +Size is measured on the encoded wire value: `byteLength` for the binary +codec, UTF-8 bytes (not UTF-16 code units) for the JSON debug codec's string +output — `.length` would undercount non-ASCII payloads and let them slip +past the guard only to die at the edge (codex review). + +Only `mut`/`call` are guarded — they are the frames that carry client data +and have a typed rejection surface. `sub`/`fetch`/`unsub` frames are +structurally small in practice; a pathological `where` predicate could in +principle also exceed the cap (codex raised this), but that exposure predates +this ADR, carries no row data, and has no typed rejection surface to reuse — +a bounded-predicate rule would be its own decision. Explicitly out of scope +here. + +### D2: Server drop stays a drop — with the reason on record + +The ADR-0012 inbound guard is unchanged in behavior: oversize → drop + +`console.error`, no reply. A typed `rejected` would require the `txId`, and +recovering it means decoding the very payload the guard exists **not** to +decode — the bound is on memory/CPU, and a partial/prefix decode of +MessagePack would be a hand-rolled parser coupled to field order, for a path +production traffic mostly cannot reach (the edge cap drops it first). The +drop is now explicitly framed as defense in depth behind D1, and the code +comment says so. + +### D3: Outbound is warn-only — observability, not enforcement + +A new tunable, `warnOutboundFrameBytes: number | null` (default `1_048_576`, +`null` disables), following the ADR-0012 `protected readonly` knob pattern. +When an encoded outbound frame exceeds it, the DO logs a `console.warn` with +the frame type, byte size, and the collection (resolved from the socket's +subscription registry only on the warn path — never a hot-path lookup). The +frame is **still sent whole**. + +Why not split or drop: outbound has no edge cap, so delivery works; splitting +would need a reassembly protocol for a problem whose real fix is sending less +— column projection / changed-columns-only patches, which is issue #28 part +(a) and deliberately out of scope here. The warning is the operator's signal +that they are paying the full-row-hydration cost. + +## Consequences + +- An oversize client mutation now fails in milliseconds with + `MutationRejectedError` / `"FRAME_TOO_LARGE"` instead of a 5 s generic + timeout; optimistic state rolls back promptly. Pinned by + `tests/frame-limits.test.ts` (raw transport and collection-level rollback). +- A client with a raised `maxFrameBytes` still hits the production edge cap; + the option exists to *align* with a self-hosted or changed limit, not to + bypass Cloudflare's. +- Large outbound rows keep working exactly as before (whole, single frame) — + now with a warning naming the collection when they cross the threshold. + Operators who consider that noise can set `warnOutboundFrameBytes = null`. +- The server's silent inbound drop is unchanged and remains pinned by + `tests/wire-hardening.test.ts`; its comment now explains why it stays + silent. Its measure for **string** frames is still `message.length` + (UTF-16 code units) per ADR-0012 D2 — an undercount for non-ASCII text + frames, flagged here rather than silently changed; the wire default is + binary, where the measure is exact. +- A socket that refuses a `mut`/`call` send synchronously (workerd does this + for frames over its own cap) now cleans up the pending-receipt waiter and + its timeout before rejecting, instead of leaving a stale entry armed for + `timeoutMs` (codex review). diff --git a/docs/adr/README.md b/docs/adr/README.md index c1dd131..e9cf56f 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -24,3 +24,4 @@ explains the displacement. | [0015](./0015-syncable-mixin.md) | `Syncable` mixin: the sync core as a mixin over any DO base | Accepted (reframes 0001 D13; Actor unsupported) | | [0016](./0016-reconnect-policy.md) | Transport reconnect policy: injectable delay function, terminal 4xxx, jittered backoff | Accepted | | [0017](./0017-blob-wire-normalization.md) | BLOB wire normalization: bare ArrayBuffer becomes Uint8Array at emission | Accepted | +| [0018](./0018-oversize-frames.md) | Oversize frames: client pre-send guard is the rejection surface; outbound is warn-only | Accepted | diff --git a/src/client/transport.ts b/src/client/transport.ts index 6f37468..80655f7 100644 --- a/src/client/transport.ts +++ b/src/client/transport.ts @@ -14,7 +14,7 @@ // The socket opener is injectable: the browser default is `new WebSocket(url)`; // other runtimes (and tests) provide an already-connected socket. -import { createFrameCodec, type FrameCodec } from "../wire/frame-codec.ts" +import { createFrameCodec, type FrameCodec, type WireOut } from "../wire/frame-codec.ts" import type { ClientFrame, RowOp, ServerFrame } from "../wire/frames.ts" /** Minimal structural socket — satisfied by both browser WebSocket and a @@ -74,6 +74,13 @@ export interface TransportOptions { codec?: FrameCodec /** Confirmation/await timeout in ms. */ timeoutMs?: number + /** Maximum encoded size (bytes) of an outgoing mut/call frame (ADR-0018). + * Cloudflare's edge caps INBOUND WebSocket messages at ~1 MiB, so an + * oversize frame may never reach the DO at all — the pre-send guard rejects + * it locally with a typed `MutationRejectedError` (code "FRAME_TOO_LARGE") + * instead of letting the send die into a confirmation timeout. Default + * 1_048_576, aligned with the server's `maxFrameBytes` and the edge cap. */ + maxFrameBytes?: number /** Reconnect pacing. A number is the base delay (ms) for the default * jittered-backoff policy (`defaultReconnectDelay`) — the attempt-1 jitter * ceiling. A function is the full policy; return `null` to stop @@ -128,6 +135,7 @@ export class WebSocketTransport { private connectPromise: Promise | null = null private readonly codec: FrameCodec private readonly timeoutMs: number + private readonly maxFrameBytes: number private readonly open: () => WebSocketLike | Promise private readonly handlers = new Map< @@ -166,6 +174,7 @@ export class WebSocketTransport { constructor(opts: TransportOptions) { this.codec = opts.codec ?? createFrameCodec() this.timeoutMs = opts.timeoutMs ?? 5000 + this.maxFrameBytes = opts.maxFrameBytes ?? 1_048_576 this.reconnectDelay = typeof opts.reconnectDelay === "function" ? opts.reconnectDelay : defaultReconnectDelay(opts.reconnectDelay ?? 250) this.onClosed = opts.onClosed @@ -405,6 +414,23 @@ export class WebSocketTransport { } private async sendAwaitingReceipt(frame: ClientFrame, txId: string): Promise<{ result?: unknown }> { + // Pre-send size guard (ADR-0018): the reliable half of oversize handling. + // Cloudflare's edge caps inbound WS messages at ~1 MiB, so an oversize + // frame may never reach the DO — waiting for a server rejection would just + // be the confirmation timeout. Reject here, typed and immediate, so the + // optimistic overlay rolls back promptly. Encode once; the bytes are + // reused for the actual send. + const encoded = this.codec.encode(frame) + // A string frame (the JSON debug codec) goes over the wire as UTF-8 — + // measure bytes, not UTF-16 code units, or non-ASCII payloads undercount + // and slip past the guard only to die at the edge cap (codex review). + const bytes = typeof encoded === "string" ? new TextEncoder().encode(encoded).byteLength : encoded.byteLength + if (bytes > this.maxFrameBytes) { + throw new MutationRejectedError( + `frame too large (${bytes} bytes > maxFrameBytes ${this.maxFrameBytes})`, + "FRAME_TOO_LARGE", + ) + } await this.connect() return new Promise((resolve, reject) => { const timer = setTimeout(() => { @@ -412,13 +438,26 @@ export class WebSocketTransport { reject(new Error(`confirmation timeout: txId=${txId}`)) }, this.timeoutMs) this.pendingTx.set(txId, { resolve, reject, timer }) - this.sendFrame(frame) + // A socket may refuse a send synchronously (workerd does for frames over + // its own cap). Clean up the waiter/timer before rejecting, or the stale + // entry lingers with an armed timeout for timeoutMs (codex review). + try { + this.sendRaw(encoded) + } catch (e) { + clearTimeout(timer) + this.pendingTx.delete(txId) + reject(e instanceof Error ? e : new Error(String(e))) + } }) } private sendFrame(frame: ClientFrame): void { + this.sendRaw(this.codec.encode(frame)) + } + + private sendRaw(data: WireOut): void { if (!this.ws) throw new Error("transport not connected") - this.ws.send(this.codec.encode(frame)) + this.ws.send(data) } private onMessage(data: unknown): void { diff --git a/src/server/mixin.ts b/src/server/mixin.ts index cb4eb35..b7e342d 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -143,6 +143,13 @@ export function Syncable() { protected readonly maxSubsPerSocket: number = 256 /** Maximum inbound frame size in bytes (ADR-0012). */ protected readonly maxFrameBytes: number = 1_048_576 + /** Outbound frame-size warning threshold in bytes (ADR-0018). + * Observability, not enforcement: outbound WS messages are not + * edge-capped the way inbound ones are, and splitting or dropping a + * hydrated full-row frame would be worse than delivering it — the real + * fix for oversized re-sends is column projection (issue #28a). `null` + * disables the warning. Default 1 MiB (mirrors `maxFrameBytes`). */ + protected readonly warnOutboundFrameBytes: number | null = 1_048_576 // ---- internal machinery (private — off the collision surface) ---------- #compiled: CompiledSync | undefined @@ -325,6 +332,14 @@ export function Syncable() { // Reject oversize frames before decode (ADR-0012): mirrors the // undecodable-frame stance — drop + log, no reply, no crash. + // + // No typed `rejected` here (ADR-0018): recovering the txId would mean + // decoding the very payload this guard exists NOT to decode (the bound + // is on memory/CPU, per ADR-0012). In production Cloudflare's edge caps + // inbound WS messages at ~1 MiB anyway, so an oversize frame usually + // never reaches this handler at all — the client transport's pre-send + // guard (same limit, MutationRejectedError "FRAME_TOO_LARGE") is the + // reliable rejection surface; this drop is defense in depth. const byteLen = typeof message === "string" ? message.length : message.byteLength if (byteLen > this.maxFrameBytes) { console.error(`oversize frame dropped (${byteLen} bytes > maxFrameBytes ${this.maxFrameBytes})`) @@ -868,9 +883,26 @@ export function Syncable() { this.#send(ws, { t: "uptodate", seq }) } - /** Encode and send a server frame on one socket. */ + /** Encode and send a server frame on one socket. Warns — and still sends + * whole — when the encoded frame exceeds `warnOutboundFrameBytes` + * (ADR-0018): observability for hydrated full-row re-sends, whose real + * fix is column projection (issue #28a). */ #send(ws: WebSocket, frame: ServerFrame): void { - ws.send(this.#codec.encode(frame)) + const encoded = this.#codec.encode(frame) + if (this.warnOutboundFrameBytes !== null) { + const bytes = typeof encoded === "string" ? encoded.length : encoded.byteLength + if (bytes > this.warnOutboundFrameBytes) { + // Resolve the sub's collection only on the warn path (linear scan + // over this socket's subs — never on the hot path). + const subId = (frame as { sub?: string }).sub + const collection = subId ? this.#subs.forWs(ws).find((s) => s.subId === subId)?.collection : undefined + const where = collection ? ` for collection '${collection}'` : subId ? ` for sub '${subId}'` : "" + console.warn( + `outbound '${frame.t}' frame is ${bytes} bytes (> warnOutboundFrameBytes ${this.warnOutboundFrameBytes})${where}`, + ) + } + } + ws.send(encoded) } /** The attachment bound at upgrade, surviving hibernation. */ diff --git a/src/server/sync-do.ts b/src/server/sync-do.ts index 353c1de..47593fc 100644 --- a/src/server/sync-do.ts +++ b/src/server/sync-do.ts @@ -40,6 +40,7 @@ export abstract class SyncDurableObject extends declare protected readonly maxOpsPerMutation: number declare protected readonly maxSubsPerSocket: number declare protected readonly maxFrameBytes: number + declare protected readonly warnOutboundFrameBytes: number | null constructor(ctx: ConstructorParameters[0], env: Env) { super(ctx, env) diff --git a/tests/frame-limits.test.ts b/tests/frame-limits.test.ts new file mode 100644 index 0000000..af1626e --- /dev/null +++ b/tests/frame-limits.test.ts @@ -0,0 +1,447 @@ +// WHY: oversize-frame handling (ADR-0018, issue #28b). Cloudflare's edge caps +// INBOUND WebSocket messages at ~1 MiB, so an oversize client frame may never +// reach the DO — a server-side rejection alone can't be the surface. The client +// transport therefore guards BEFORE sending: an oversize mut/call rejects +// immediately with a typed MutationRejectedError (FRAME_TOO_LARGE), rolling the +// optimistic overlay back promptly instead of dying into a confirmation +// timeout. Outbound (server->client) is NOT capped the same way: a large row +// must still arrive whole (no splitting, no dropping — column projection, #28a, +// is the real fix), with a console.warn as observability when a frame exceeds +// `warnOutboundFrameBytes`. These tests pin all three surfaces, plus the +// adjacent read-only-op rejection and tiny-collection snapshots. + +import { createCollection } from "@tanstack/db"; +import { env, runInDurableObject, SELF } from "cloudflare:test"; +import { describe, expect, it, vi } from "vitest"; +import { doCollectionOptions } from "../src/client/do-collection.ts"; +import { + MutationRejectedError, + WebSocketTransport, + type WebSocketLike, +} from "../src/client/transport.ts"; +import { createFrameCodec } from "../src/wire/frame-codec.ts"; +import type { TestApi } from "./test-worker.ts"; + +function makeTransport( + room: string, + timeoutMs?: number, +): WebSocketTransport { + return new WebSocketTransport({ + url: `https://example.com/sync/${room}`, + timeoutMs, + open: async () => { + const res = await SELF.fetch(`https://example.com/sync/${room}`, { + headers: { Upgrade: "websocket" }, + }); + const ws = res.webSocket; + if (!ws) throw new Error("no webSocket"); + ws.accept(); + return ws as unknown as WebSocketLike; + }, + }); +} + +const BIG = 1_500_000; // ~1.5 MB, comfortably over the 1 MiB (1_048_576) boundary + +describe("large TEXT values sync server->client whole", () => { + it("delivers a ~1.5MB row server->client in the initial snapshot, whole", async () => { + const room = "fl-big-snap"; + const t = makeTransport(room); + await t.connect(); + + const bigBody = "S".repeat(BIG); + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => { + // NB: the value must go through a bind parameter — interpolating a + // 1.5MB literal into the statement text hits SQLITE_TOOBIG (DO SQLite + // caps statement TEXT length, not bound-value size). + s.storage.sql.exec( + "INSERT INTO messages(id,body) VALUES(?,?)", + "big", + "S".repeat(BIG), + ); + }, + ); + + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (r) => r.id, + }), + ); + await messages.preload(); + + const row = messages.get("big"); + expect(row).toBeDefined(); + expect((row as { body: string }).body.length).toBe(BIG); + expect((row as { body: string }).body).toBe(bigBody); + t.close(); + }); + + it("delivers a ~1.5MB row server->client as a live delta (runSyncedWrite), whole", async () => { + const room = "fl-big-delta"; + const t = makeTransport(room); + await t.connect(); + + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (r) => r.id, + }), + ); + await messages.preload(); + expect(messages.get("big2")).toBeUndefined(); + + // Server-originated write: runSyncedWrite is protected — reach it at + // runtime, as a DO subclass would call it. + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (i) => { + ( + i as unknown as { + runSyncedWrite: (fn: (sql: SqlStorage) => void) => void; + } + ).runSyncedWrite((sql) => { + sql.exec( + "INSERT INTO messages(id,body) VALUES(?,?)", + "big2", + "D".repeat(BIG), + ); + }); + }, + ); + + const start = Date.now(); + while (!messages.get("big2")) { + if (Date.now() - start > 3000) throw new Error("big delta never arrived"); + await new Promise((r) => setTimeout(r, 10)); + } + expect((messages.get("big2") as { body: string }).body.length).toBe(BIG); + t.close(); + }); + + it("warns (console.warn) when an outbound frame exceeds warnOutboundFrameBytes; small frames stay silent", async () => { + const room = "fl-warn-outbound"; + const warn = vi.spyOn(console, "warn"); + try { + const t = makeTransport(room); + await t.connect(); + + // A small row first: no outbound frame is near the threshold, so the + // guard must stay silent (warn-on-everything would be noise, not signal). + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => { + s.storage.sql.exec( + "INSERT INTO messages(id,body) VALUES('small','tiny')", + ); + }, + ); + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (r) => r.id, + }), + ); + await messages.preload(); + expect(messages.get("small")).toBeDefined(); + expect( + warn.mock.calls.filter((c) => + String(c[0]).includes("warnOutboundFrameBytes"), + ), + ).toHaveLength(0); + + // A >1 MiB live delta must still be DELIVERED whole (warn is + // observability, not enforcement) AND produce exactly the warning. + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (i) => { + ( + i as unknown as { + runSyncedWrite: (fn: (sql: SqlStorage) => void) => void; + } + ).runSyncedWrite((sql) => { + sql.exec( + "INSERT INTO messages(id,body) VALUES(?,?)", + "warned", + "W".repeat(BIG), + ); + }); + }, + ); + const start = Date.now(); + while (!messages.get("warned")) { + if (Date.now() - start > 3000) + throw new Error("big delta never arrived"); + await new Promise((r) => setTimeout(r, 10)); + } + const warned = warn.mock.calls.filter((c) => + String(c[0]).includes("warnOutboundFrameBytes"), + ); + expect(warned.length).toBeGreaterThan(0); + // The warning names the collection (or sub) so the operator can act on it. + expect(String(warned[0]![0])).toMatch(/messages/); + t.close(); + } finally { + warn.mockRestore(); + } + }); +}); + +describe("oversize client mutation frames (ADR-0018)", () => { + it("rejects an oversize mut PROMPTLY with a typed FRAME_TOO_LARGE — never a confirmation timeout", async () => { + const room = "fl-big-mut"; + // Generous confirmation timeout: if the old silent-drop path were still in + // effect, this test would only fail after 5s with a generic timeout Error. + const t = makeTransport(room, 5000); + await t.connect(); + + const start = Date.now(); + let err: unknown; + try { + await t.sendMut({ + t: "mut", + txId: "fl-big-tx", + collection: "messages", + ops: [ + { + type: "insert", + key: "toolarge", + cols: { id: "toolarge", body: "B".repeat(BIG) }, + }, + ], + }); + } catch (e) { + err = e; + } + // The pre-send guard rejects locally — typed, coded, and without waiting + // out the confirmation timeout (the frame may never reach the DO in prod: + // Cloudflare's edge caps inbound WS messages at ~1 MiB). + expect(err).toBeInstanceOf(MutationRejectedError); + expect((err as MutationRejectedError).code).toBe("FRAME_TOO_LARGE"); + expect(Date.now() - start).toBeLessThan(2000); + + // The row must not exist server-side. + const rows = await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => + Array.from( + s.storage.sql.exec( + "SELECT COUNT(*) AS c FROM messages WHERE id='toolarge'", + ), + ), + ); + expect((rows[0] as { c: number }).c).toBe(0); + t.close(); + }); + + it("rolls the optimistic overlay back promptly when a collection insert is FRAME_TOO_LARGE", async () => { + const room = "fl-big-rollback"; + const t = makeTransport(room, 5000); + await t.connect(); + + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (r) => r.id, + }), + ); + await messages.preload(); + + const tx = messages.insert({ id: "huge", body: "H".repeat(BIG) }); + // Optimistic overlay applies immediately... + expect(messages.get("huge")).toBeDefined(); + + let err: unknown; + try { + await tx.isPersisted.promise; + } catch (e) { + err = e; + } + expect(err).toBeDefined(); + // TanStack may wrap the mutationFn error; the root cause is the transport's + // MutationRejectedError with the FRAME_TOO_LARGE reason. + const msg = + err instanceof Error + ? `${err.message} ${String((err as { cause?: unknown }).cause ?? "")}` + : String(err); + expect(msg).toMatch(/frame too large/); + + // ...and rolls back on the local rejection — no 5s timeout window. + expect(messages.get("huge")).toBeUndefined(); + t.close(); + }); + + it("measures string-codec frames in UTF-8 bytes, not UTF-16 code units", async () => { + // The JSON debug codec emits a string; WebSocket sends strings as UTF-8. + // Measuring `.length` (code units) would undercount non-ASCII payloads and + // let them slip past the guard only to die at the edge cap (codex review). + const fake: WebSocketLike = { + send: () => {}, + close: () => {}, + addEventListener: () => {}, + removeEventListener: () => {}, + }; + const t = new WebSocketTransport({ + url: "https://example.com/unused", + open: () => fake, + codec: createFrameCodec({ binary: false }), + maxFrameBytes: 4096, + timeoutMs: 500, + }); + await t.connect(); + // "€" is 1 UTF-16 code unit but 3 UTF-8 bytes: 2000 of them keep the + // frame's code-unit length under 4096 while its byte size is ~6000+. + let err: unknown; + try { + await t.sendMut({ + t: "mut", + txId: "fl-utf8-tx", + collection: "messages", + ops: [ + { type: "insert", key: "u", cols: { id: "u", body: "€".repeat(2000) } }, + ], + }); + } catch (e) { + err = e; + } + // Code-unit measurement would send the frame and die on the 500ms timeout + // with a generic Error instead. + expect(err).toBeInstanceOf(MutationRejectedError); + expect((err as MutationRejectedError).code).toBe("FRAME_TOO_LARGE"); + t.close(); + }); +}); + +describe("read-only collection ops (no mutation handler)", () => { + // `transformed` declares ONLY an insert mutation — update/delete are the + // "no mutations block" case per-op. A collection with NO mutations at all + // takes the identical code path (mutations.get returns undefined). + it("client update on an op with no handler: MutationRejectedError + optimistic rollback", async () => { + const room = "fl-readonly"; + const t = makeTransport(room); + await t.connect(); + + const coll = createCollection( + doCollectionOptions({ + transport: t, + table: "transformed", + getKey: (r) => r.id, + }), + ); + await coll.preload(); + + // Seed via the existing insert mutation (allowed). + await coll.insert({ id: "r1", body: "orig" }).isPersisted.promise; + expect(coll.get("r1")).toMatchObject({ body: "orig" }); + + // update has NO handler on `transformed`. + const tx = coll.update("r1", (d) => { + d.body = "hacked"; + }); + // Optimistic overlay applies immediately... + expect(coll.get("r1")).toMatchObject({ body: "hacked" }); + + let err: unknown; + try { + await tx.isPersisted.promise; + } catch (e) { + err = e; + } + expect(err).toBeDefined(); + // TanStack may wrap the mutationFn error; the root cause is the transport's + // MutationRejectedError carrying the server's message. + const msg = + err instanceof Error + ? `${err.message} ${String((err as { cause?: unknown }).cause ?? "")}` + : String(err); + expect(msg).toMatch(/no mutation handler for 'transformed:update'/); + + // Optimistic overlay rolled back to the confirmed row. + expect(coll.get("r1")).toMatchObject({ body: "orig" }); + + // Server row untouched. + const rows = await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => + Array.from( + s.storage.sql.exec("SELECT body FROM transformed WHERE id='r1'"), + ), + ); + expect((rows[0] as { body: string }).body).toBe("orig"); + t.close(); + }); +}); + +describe("tiny collections", () => { + it("zero-row collection: snapshot completes (snap-end) and preload resolves", async () => { + const room = "fl-empty"; + const t = makeTransport(room); + await t.connect(); + const files = createCollection( + doCollectionOptions({ + transport: t, + table: "files", + getKey: (r) => r.id, + }), + ); + await files.preload(); // would hang if a 0-row snapshot never emitted snap-end + expect(files.size).toBe(0); + t.close(); + }); + + it("single-row collection: snapshot + frequent UPDATEs to the one row stream fine", async () => { + const room = "fl-single"; + const t = makeTransport(room); + await t.connect(); + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => { + s.storage.sql.exec("INSERT INTO files(id,name) VALUES('meta','v0')"); + }, + ); + const files = createCollection( + doCollectionOptions({ + transport: t, + table: "files", + getKey: (r) => r.id, + }), + ); + await files.preload(); + expect(files.size).toBe(1); + expect(files.get("meta")).toMatchObject({ name: "v0" }); + + // Repeated server-side updates to the single row (metadata churn) — the + // coalescer collapses them; the client must land on the last value. + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (i) => { + const inst = i as unknown as { + runSyncedWrite: (fn: (sql: SqlStorage) => void) => void; + }; + for (let n = 1; n <= 5; n++) { + inst.runSyncedWrite((sql) => + sql.exec("UPDATE files SET name=? WHERE id='meta'", `v${n}`), + ); + } + }, + ); + const start = Date.now(); + while ((files.get("meta") as { name: string } | undefined)?.name !== "v5") { + if (Date.now() - start > 3000) + throw new Error( + `never converged: ${JSON.stringify(files.get("meta"))}`, + ); + await new Promise((r) => setTimeout(r, 10)); + } + t.close(); + }); +}); + +// SqlStorage type for the runInDurableObject callbacks above. +type SqlStorage = import("@cloudflare/workers-types").SqlStorage; From 8a38f67dd9e27041d4de4f2b74110ee5a82df143 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 22 Jul 2026 15:19:07 +1000 Subject: [PATCH 2/3] =?UTF-8?q?refactor!:=20frame=20limits=20are=20infra?= =?UTF-8?q?=20facts,=20not=20knobs=20=E2=80=94=20drop=20maxFrameBytes=20op?= =?UTF-8?q?tion=20and=20warnOutboundFrameBytes=20tunable?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Maintainer review of #33: the 1 MiB limit is Cloudflare's inbound WS edge cap — an infrastructure fact, not an application preference. Both wire endpoints ship in this package and exactly one infra is supported, so facts don't get knobs: a client option could only lower the limit pointlessly or raise it into the edge cap's lie. - Client: `maxFrameBytes` TransportOptions option removed; the pre-send guard now checks against a module constant MAX_FRAME_BYTES (1_048_576). The FRAME_TOO_LARGE typed rejection is unchanged; the message cites the edge cap, not an option name. - Server: `warnOutboundFrameBytes: number | null` protected tunable (and its null-disable branch and sync-do.ts redeclare) removed; the outbound warn fires at a fixed WARN_OUTBOUND_FRAME_BYTES (1 MiB) constant and now points at the real fix (column projection, issue #28). Inbound = enforced constant; outbound = deliberately unenforced, warn as the one production breadcrumb. - ADR-0018 gains D0 stating the opinion and recording the rejected configurable-limits alternative; CHANGELOG no longer names options. - Tests: knob usages removed (UTF-8 test now crosses the real 1 MiB cap); added the sync-send-failure cleanup pin (a socket that throws on send surfaces that error immediately with a clean waiter, never a timeout). `!`: removes the `maxFrameBytes` transport option added earlier on this unreleased branch — no released API is affected. Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 16 ++++---- docs/adr/0018-oversize-frames.md | 50 ++++++++++++++++-------- src/client/transport.ts | 21 +++++------ src/server/mixin.ts | 42 ++++++++++----------- src/server/sync-do.ts | 1 - tests/frame-limits.test.ts | 65 +++++++++++++++++++++++++++----- 6 files changed, 129 insertions(+), 66 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c356322..7df302c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -17,14 +17,14 @@ While pre-1.0, the public API may change between 0.x releases. timeout — and in production Cloudflare's ~1 MiB edge cap on inbound WebSocket messages means the frame may never reach the DO at all. The transport now guards **before sending**: a `mut`/`call` whose encoded size - exceeds the new `maxFrameBytes` option (default 1 MiB, aligned with the - server and the edge cap) rejects immediately with `MutationRejectedError` - (code `"FRAME_TOO_LARGE"`), so the optimistic overlay rolls back promptly. - The server's silent drop stays as defense in depth. Outbound gains a - warn-only threshold, `warnOutboundFrameBytes` (default 1 MiB, `null` - disables): a larger encoded frame logs a `console.warn` with size and - collection but is still sent whole — column projection (#28a) remains the - real fix for oversized full-row re-sends. + exceeds the 1 MiB edge cap rejects immediately with + `MutationRejectedError` (code `"FRAME_TOO_LARGE"`), so the optimistic + overlay rolls back promptly. The server's silent drop stays as defense in + depth. Outbound gains a warn-only fixed 1 MiB threshold: a larger encoded + frame logs a `console.warn` with size and collection but is still sent + whole — column projection (#28a) remains the real fix for oversized + full-row re-sends. The limits are infrastructure facts (Cloudflare's edge + cap), not application preferences, so they are constants, not options. - **Transport reconnect policy (ADR-0016; fixes #25, #26).** One option, `reconnectDelay?: number | ((attempt, closeCode?, closeReason?) => number | null)` diff --git a/docs/adr/0018-oversize-frames.md b/docs/adr/0018-oversize-frames.md index 35b17fe..ce65573 100644 --- a/docs/adr/0018-oversize-frames.md +++ b/docs/adr/0018-oversize-frames.md @@ -32,12 +32,34 @@ WebSocket limits): ## Decisions +### D0: The limits are infrastructure facts, not application preferences + +Cloudflare's ~1 MiB inbound edge cap is a **fact about the one infrastructure +this library supports** — and both wire endpoints ship in this package. Facts +don't get knobs. The thresholds below are therefore module-level **constants** +(`MAX_FRAME_BYTES` in the transport, `WARN_OUTBOUND_FRAME_BYTES` in the +mixin), both `1_048_576`. If Cloudflare changes the cap, the constants change +with an ADR note. + +**Rejected alternative — configurable limits** (a `maxFrameBytes` transport +option and a `warnOutboundFrameBytes: number | null` protected tunable, in an +earlier revision of this branch): rejected on review. Since we control both +sides and support exactly one infra, a client-side knob could only be set +*below* the cap (pointless — the server and edge still enforce the fact) or +*above* it (a lie — the guard would wave through frames the edge then kills, +reintroducing exactly the silent-timeout failure this ADR removes). The +warn-threshold knob likewise had no honest setting other than the cap itself; +`null`-disabling it only hid the one production breadcrumb. + +The overall shape: **inbound = enforced constant; outbound = deliberately +unenforced** (breaking a correct broadcast to save bandwidth would invert the +failure hierarchy), with the D3 warning as the only production breadcrumb. + ### D1: Client pre-send guard — the reliable half -`WebSocketTransport` gains `maxFrameBytes` (default `1_048_576`, aligned with -the server's ADR-0012 default and the edge cap). Before sending a `mut` or -`call`, the transport encodes the frame once, checks the encoded size, and on -breach rejects **locally and immediately** with the existing typed surface: +Before sending a `mut` or `call`, the transport encodes the frame once, +checks the encoded size against `MAX_FRAME_BYTES`, and on breach rejects +**locally and immediately** with the existing typed surface: `MutationRejectedError` with code `"FRAME_TOO_LARGE"`. No round trip, no timeout; the optimistic overlay rolls back promptly through TanStack's normal rejected-mutation path. The encoded bytes are reused for the actual send, so @@ -69,12 +91,11 @@ comment says so. ### D3: Outbound is warn-only — observability, not enforcement -A new tunable, `warnOutboundFrameBytes: number | null` (default `1_048_576`, -`null` disables), following the ADR-0012 `protected readonly` knob pattern. -When an encoded outbound frame exceeds it, the DO logs a `console.warn` with -the frame type, byte size, and the collection (resolved from the socket's -subscription registry only on the warn path — never a hot-path lookup). The -frame is **still sent whole**. +When an encoded outbound frame exceeds `WARN_OUTBOUND_FRAME_BYTES` (fixed +1 MiB, per D0), the DO logs a `console.warn` with the frame type, byte size, +the collection (resolved from the socket's subscription registry only on the +warn path — never a hot-path lookup), and a pointer at the real fix (column +projection, issue #28). The frame is **still sent whole**. Why not split or drop: outbound has no edge cap, so delivery works; splitting would need a reassembly protocol for a problem whose real fix is sending less @@ -88,12 +109,11 @@ that they are paying the full-row-hydration cost. `MutationRejectedError` / `"FRAME_TOO_LARGE"` instead of a 5 s generic timeout; optimistic state rolls back promptly. Pinned by `tests/frame-limits.test.ts` (raw transport and collection-level rollback). -- A client with a raised `maxFrameBytes` still hits the production edge cap; - the option exists to *align* with a self-hosted or changed limit, not to - bypass Cloudflare's. +- No new public API surface: both thresholds are constants (D0), so there is + nothing to configure and nothing to misconfigure. - Large outbound rows keep working exactly as before (whole, single frame) — - now with a warning naming the collection when they cross the threshold. - Operators who consider that noise can set `warnOutboundFrameBytes = null`. + now with a warning naming the collection and pointing at column projection + (#28) when they cross the threshold. - The server's silent inbound drop is unchanged and remains pinned by `tests/wire-hardening.test.ts`; its comment now explains why it stays silent. Its measure for **string** frames is still `message.length` diff --git a/src/client/transport.ts b/src/client/transport.ts index 80655f7..6ae64e2 100644 --- a/src/client/transport.ts +++ b/src/client/transport.ts @@ -35,6 +35,14 @@ export interface SubHandler { onReset(): void } +/** Cloudflare's inbound WebSocket edge cap, ~1 MiB (ADR-0018). An + * infrastructure FACT, not an application preference: both wire endpoints + * ship in this package and the only supported infra fixes the number, so it + * is a constant, not a knob — a knob could only lower it pointlessly or + * raise it into the edge cap's lie. If Cloudflare changes the cap, this + * constant changes with an ADR note. Mirrors the server's ADR-0012 default. */ +const MAX_FRAME_BYTES = 1_048_576 + export class MutationRejectedError extends Error { constructor( message: string, @@ -74,13 +82,6 @@ export interface TransportOptions { codec?: FrameCodec /** Confirmation/await timeout in ms. */ timeoutMs?: number - /** Maximum encoded size (bytes) of an outgoing mut/call frame (ADR-0018). - * Cloudflare's edge caps INBOUND WebSocket messages at ~1 MiB, so an - * oversize frame may never reach the DO at all — the pre-send guard rejects - * it locally with a typed `MutationRejectedError` (code "FRAME_TOO_LARGE") - * instead of letting the send die into a confirmation timeout. Default - * 1_048_576, aligned with the server's `maxFrameBytes` and the edge cap. */ - maxFrameBytes?: number /** Reconnect pacing. A number is the base delay (ms) for the default * jittered-backoff policy (`defaultReconnectDelay`) — the attempt-1 jitter * ceiling. A function is the full policy; return `null` to stop @@ -135,7 +136,6 @@ export class WebSocketTransport { private connectPromise: Promise | null = null private readonly codec: FrameCodec private readonly timeoutMs: number - private readonly maxFrameBytes: number private readonly open: () => WebSocketLike | Promise private readonly handlers = new Map< @@ -174,7 +174,6 @@ export class WebSocketTransport { constructor(opts: TransportOptions) { this.codec = opts.codec ?? createFrameCodec() this.timeoutMs = opts.timeoutMs ?? 5000 - this.maxFrameBytes = opts.maxFrameBytes ?? 1_048_576 this.reconnectDelay = typeof opts.reconnectDelay === "function" ? opts.reconnectDelay : defaultReconnectDelay(opts.reconnectDelay ?? 250) this.onClosed = opts.onClosed @@ -425,9 +424,9 @@ export class WebSocketTransport { // measure bytes, not UTF-16 code units, or non-ASCII payloads undercount // and slip past the guard only to die at the edge cap (codex review). const bytes = typeof encoded === "string" ? new TextEncoder().encode(encoded).byteLength : encoded.byteLength - if (bytes > this.maxFrameBytes) { + if (bytes > MAX_FRAME_BYTES) { throw new MutationRejectedError( - `frame too large (${bytes} bytes > maxFrameBytes ${this.maxFrameBytes})`, + `frame too large (${bytes} bytes > the ${MAX_FRAME_BYTES}-byte inbound WebSocket edge cap)`, "FRAME_TOO_LARGE", ) } diff --git a/src/server/mixin.ts b/src/server/mixin.ts index b7e342d..f43c7ee 100644 --- a/src/server/mixin.ts +++ b/src/server/mixin.ts @@ -47,6 +47,14 @@ import { SubscriptionRegistry, type Sub } from "./subscriptions.ts" * this tag, so tddc never touches a host's sockets and vice versa (ADR-0015). */ export const SYNC_TAG = "_tddc" +/** Outbound frame-size warning threshold (ADR-0018): observability only, and + * deliberately NOT a knob. Inbound is enforced at the edge-cap constant; + * outbound is deliberately unenforced — breaking a correct broadcast to save + * bandwidth would invert the failure hierarchy — so this warn is the one + * production breadcrumb. The real fix for oversize full-row rebroadcasts is + * column projection (issue #28). */ +const WARN_OUTBOUND_FRAME_BYTES = 1_048_576 + /** Runtime options for the mixin. `configure()` in your constructor. * Numeric tuning knobs stay protected overridable fields (see ADR-0015). */ export interface SyncableOptions { @@ -143,13 +151,6 @@ export function Syncable() { protected readonly maxSubsPerSocket: number = 256 /** Maximum inbound frame size in bytes (ADR-0012). */ protected readonly maxFrameBytes: number = 1_048_576 - /** Outbound frame-size warning threshold in bytes (ADR-0018). - * Observability, not enforcement: outbound WS messages are not - * edge-capped the way inbound ones are, and splitting or dropping a - * hydrated full-row frame would be worse than delivering it — the real - * fix for oversized re-sends is column projection (issue #28a). `null` - * disables the warning. Default 1 MiB (mirrors `maxFrameBytes`). */ - protected readonly warnOutboundFrameBytes: number | null = 1_048_576 // ---- internal machinery (private — off the collision surface) ---------- #compiled: CompiledSync | undefined @@ -884,23 +885,22 @@ export function Syncable() { } /** Encode and send a server frame on one socket. Warns — and still sends - * whole — when the encoded frame exceeds `warnOutboundFrameBytes` + * whole — when the encoded frame exceeds `WARN_OUTBOUND_FRAME_BYTES` * (ADR-0018): observability for hydrated full-row re-sends, whose real - * fix is column projection (issue #28a). */ + * fix is column projection (issue #28). */ #send(ws: WebSocket, frame: ServerFrame): void { const encoded = this.#codec.encode(frame) - if (this.warnOutboundFrameBytes !== null) { - const bytes = typeof encoded === "string" ? encoded.length : encoded.byteLength - if (bytes > this.warnOutboundFrameBytes) { - // Resolve the sub's collection only on the warn path (linear scan - // over this socket's subs — never on the hot path). - const subId = (frame as { sub?: string }).sub - const collection = subId ? this.#subs.forWs(ws).find((s) => s.subId === subId)?.collection : undefined - const where = collection ? ` for collection '${collection}'` : subId ? ` for sub '${subId}'` : "" - console.warn( - `outbound '${frame.t}' frame is ${bytes} bytes (> warnOutboundFrameBytes ${this.warnOutboundFrameBytes})${where}`, - ) - } + const bytes = typeof encoded === "string" ? encoded.length : encoded.byteLength + if (bytes > WARN_OUTBOUND_FRAME_BYTES) { + // Resolve the sub's collection only on the warn path (linear scan + // over this socket's subs — never on the hot path). + const subId = (frame as { sub?: string }).sub + const collection = subId ? this.#subs.forWs(ws).find((s) => s.subId === subId)?.collection : undefined + const where = collection ? ` for collection '${collection}'` : subId ? ` for sub '${subId}'` : "" + console.warn( + `oversize outbound '${frame.t}' frame (${bytes} bytes > ${WARN_OUTBOUND_FRAME_BYTES})${where} — ` + + `full-row rebroadcast; column projection (issue #28) is the real fix`, + ) } ws.send(encoded) } diff --git a/src/server/sync-do.ts b/src/server/sync-do.ts index 47593fc..353c1de 100644 --- a/src/server/sync-do.ts +++ b/src/server/sync-do.ts @@ -40,7 +40,6 @@ export abstract class SyncDurableObject extends declare protected readonly maxOpsPerMutation: number declare protected readonly maxSubsPerSocket: number declare protected readonly maxFrameBytes: number - declare protected readonly warnOutboundFrameBytes: number | null constructor(ctx: ConstructorParameters[0], env: Env) { super(ctx, env) diff --git a/tests/frame-limits.test.ts b/tests/frame-limits.test.ts index af1626e..ad70922 100644 --- a/tests/frame-limits.test.ts +++ b/tests/frame-limits.test.ts @@ -7,8 +7,10 @@ // timeout. Outbound (server->client) is NOT capped the same way: a large row // must still arrive whole (no splitting, no dropping — column projection, #28a, // is the real fix), with a console.warn as observability when a frame exceeds -// `warnOutboundFrameBytes`. These tests pin all three surfaces, plus the -// adjacent read-only-op rejection and tiny-collection snapshots. +// the fixed 1 MiB threshold. The limits are infrastructure facts (Cloudflare's +// edge cap), not application preferences — constants, not knobs. These tests +// pin all three surfaces, plus the adjacent read-only-op rejection and +// tiny-collection snapshots. import { createCollection } from "@tanstack/db"; import { env, runInDurableObject, SELF } from "cloudflare:test"; @@ -123,7 +125,7 @@ describe("large TEXT values sync server->client whole", () => { t.close(); }); - it("warns (console.warn) when an outbound frame exceeds warnOutboundFrameBytes; small frames stay silent", async () => { + it("warns (console.warn) when an outbound frame exceeds the fixed 1 MiB threshold; small frames stay silent", async () => { const room = "fl-warn-outbound"; const warn = vi.spyOn(console, "warn"); try { @@ -151,7 +153,7 @@ describe("large TEXT values sync server->client whole", () => { expect(messages.get("small")).toBeDefined(); expect( warn.mock.calls.filter((c) => - String(c[0]).includes("warnOutboundFrameBytes"), + String(c[0]).includes("oversize outbound"), ), ).toHaveLength(0); @@ -180,11 +182,13 @@ describe("large TEXT values sync server->client whole", () => { await new Promise((r) => setTimeout(r, 10)); } const warned = warn.mock.calls.filter((c) => - String(c[0]).includes("warnOutboundFrameBytes"), + String(c[0]).includes("oversize outbound"), ); expect(warned.length).toBeGreaterThan(0); - // The warning names the collection (or sub) so the operator can act on it. + // The warning names the collection (or sub) so the operator can act on + // it, and points at the real fix (column projection, issue #28). expect(String(warned[0]![0])).toMatch(/messages/); + expect(String(warned[0]![0])).toMatch(/column projection/); t.close(); } finally { warn.mockRestore(); @@ -291,12 +295,12 @@ describe("oversize client mutation frames (ADR-0018)", () => { url: "https://example.com/unused", open: () => fake, codec: createFrameCodec({ binary: false }), - maxFrameBytes: 4096, timeoutMs: 500, }); await t.connect(); - // "€" is 1 UTF-16 code unit but 3 UTF-8 bytes: 2000 of them keep the - // frame's code-unit length under 4096 while its byte size is ~6000+. + // "€" is 1 UTF-16 code unit but 3 UTF-8 bytes: 400k of them keep the + // frame's code-unit length (~400k) under the 1 MiB cap while its byte + // size (~1.2M) is over it. let err: unknown; try { await t.sendMut({ @@ -304,7 +308,11 @@ describe("oversize client mutation frames (ADR-0018)", () => { txId: "fl-utf8-tx", collection: "messages", ops: [ - { type: "insert", key: "u", cols: { id: "u", body: "€".repeat(2000) } }, + { + type: "insert", + key: "u", + cols: { id: "u", body: "€".repeat(400_000) }, + }, ], }); } catch (e) { @@ -316,6 +324,43 @@ describe("oversize client mutation frames (ADR-0018)", () => { expect((err as MutationRejectedError).code).toBe("FRAME_TOO_LARGE"); t.close(); }); + + it("surfaces a synchronous send() failure immediately with a clean waiter — never a confirmation timeout", async () => { + // workerd refuses some sends synchronously (e.g. frames over its own cap). + // The transport must reject with THAT error right away and clean up the + // pending-receipt waiter/timer (codex review) — not sit on an armed + // timeout and report a generic "confirmation timeout" later. + const boom = new Error("socket refused the send"); + const fake: WebSocketLike = { + send: () => { + throw boom; + }, + close: () => {}, + addEventListener: () => {}, + removeEventListener: () => {}, + }; + const t = new WebSocketTransport({ + url: "https://example.com/unused", + open: () => fake, + timeoutMs: 500, + }); + await t.connect(); + const start = Date.now(); + let err: unknown; + try { + await t.sendMut({ + t: "mut", + txId: "fl-syncfail-tx", + collection: "messages", + ops: [{ type: "insert", key: "k", cols: { id: "k", body: "small" } }], + }); + } catch (e) { + err = e; + } + expect(err).toBe(boom); + expect(Date.now() - start).toBeLessThan(400); + t.close(); + }); }); describe("read-only collection ops (no mutation handler)", () => { From f470b81c6bda5e53985c6f82b60c5f0b3802c151 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 22 Jul 2026 15:50:38 +1000 Subject: [PATCH 3/3] =?UTF-8?q?test:=20single-topic=20frame-limits=20?= =?UTF-8?q?=E2=80=94=20move=20client-altitude=20no-handler=20pin=20to=20er?= =?UTF-8?q?ror-paths,=20drop=20duplicate=20tiny-collection=20pins?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Final tidy for #33. tests/frame-limits.test.ts is now ADR-0018 frame limits only: - The "no mutation handler" client-altitude test moves to tests/error-paths.test.ts, beside its wire-level sibling ("unknown mutation collection → rejected"). Same why, kept intact: the wire pin proves the server answers /no mutation handler/; the moved pin proves that rejection surfaces through a real @tanstack/db collection as MutationRejectedError with prompt optimistic rollback. Room name unchanged (fl-readonly). - The "tiny collections" pair is deleted as confirmed duplicates: zero-row snapshot is already pinned at wire level (sync-read: "empty collection emits only snap-end") and client level (do-collection: "readies an empty collection without any write") — a third pin cannot fail uniquely; single-row-frequent-updates is a weaker restatement of coalesce.test.ts's burst pin, and its original why (migration due-diligence) was already banked. Full suite: 47 files / 225 tests (227 − 2 deleted; the moved one keeps counting); typecheck clean. Co-Authored-By: Claude Fable 5 --- tests/error-paths.test.ts | 75 ++++++++++++++++++++++ tests/frame-limits.test.ts | 128 +------------------------------------ 2 files changed, 76 insertions(+), 127 deletions(-) diff --git a/tests/error-paths.test.ts b/tests/error-paths.test.ts index c567613..17ea8e5 100644 --- a/tests/error-paths.test.ts +++ b/tests/error-paths.test.ts @@ -1,7 +1,11 @@ +import { createCollection } from "@tanstack/db" import { env, runInDurableObject, SELF } from "cloudflare:test" import { describe, expect, it } from "vitest" +import { doCollectionOptions } from "../src/client/do-collection.ts" +import { WebSocketTransport, type WebSocketLike } from "../src/client/transport.ts" import { createFrameCodec } from "../src/wire/frame-codec.ts" import type { ClientFrame, ServerFrame } from "../src/wire/frames.ts" +import type { TestApi } from "./test-worker.ts" // WHY: these tests pin three load-bearing invariants in the server's error paths: // 1. Atomicity — a failed mutation leaves no partial state; nothing is applied @@ -204,6 +208,77 @@ describe("mutation error paths (atomicity, fail-loud, exactly-once)", () => { }) }) +describe("no mutation handler, at client altitude", () => { + // WHY: the wire-level sibling above ("unknown mutation collection → + // rejected") pins that the server answers /no mutation handler/ with a + // `rejected` frame. This pins the CLIENT altitude of the SAME server path: + // the rejection surfaces through a real @tanstack/db collection as a + // MutationRejectedError and the optimistic overlay rolls back promptly — + // a regression in either half (transport rejection plumbing, TanStack + // rollback wiring) ships silently past the wire-level pin alone. + // + // `transformed` declares ONLY an insert mutation — update/delete are the + // "no mutations block" case per-op. A collection with NO mutations at all + // takes the identical code path (mutations.get returns undefined). + it("client update on an op with no handler: MutationRejectedError + optimistic rollback", async () => { + const room = "fl-readonly" + const t = new WebSocketTransport({ + url: `https://example.com/sync/${room}`, + open: async () => { + const res = await SELF.fetch(`https://example.com/sync/${room}`, { headers: { Upgrade: "websocket" } }) + const ws = res.webSocket + if (!ws) throw new Error("no webSocket") + ws.accept() + return ws as unknown as WebSocketLike + }, + }) + await t.connect() + + const coll = createCollection( + doCollectionOptions({ + transport: t, + table: "transformed", + getKey: (r) => r.id, + }), + ) + await coll.preload() + + // Seed via the existing insert mutation (allowed). + await coll.insert({ id: "r1", body: "orig" }).isPersisted.promise + expect(coll.get("r1")).toMatchObject({ body: "orig" }) + + // update has NO handler on `transformed`. + const tx = coll.update("r1", (d) => { + d.body = "hacked" + }) + // Optimistic overlay applies immediately... + expect(coll.get("r1")).toMatchObject({ body: "hacked" }) + + let err: unknown + try { + await tx.isPersisted.promise + } catch (e) { + err = e + } + expect(err).toBeDefined() + // TanStack may wrap the mutationFn error; the root cause is the transport's + // MutationRejectedError carrying the server's message. + const msg = + err instanceof Error ? `${err.message} ${String((err as { cause?: unknown }).cause ?? "")}` : String(err) + expect(msg).toMatch(/no mutation handler for 'transformed:update'/) + + // Optimistic overlay rolled back to the confirmed row. + expect(coll.get("r1")).toMatchObject({ body: "orig" }) + + // Server row untouched. + const rows = await runInDurableObject(env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), (_i, s) => + Array.from(s.storage.sql.exec("SELECT body FROM transformed WHERE id='r1'")), + ) + expect((rows[0] as { body: string }).body).toBe("orig") + t.close() + }) +}) + describe("command error paths", () => { it("command execute throws → rejected with generic message, detail not leaked", async () => { const ws = await openWs("/sync/err-cmd-boom") diff --git a/tests/frame-limits.test.ts b/tests/frame-limits.test.ts index ad70922..f2a3920 100644 --- a/tests/frame-limits.test.ts +++ b/tests/frame-limits.test.ts @@ -9,8 +9,7 @@ // is the real fix), with a console.warn as observability when a frame exceeds // the fixed 1 MiB threshold. The limits are infrastructure facts (Cloudflare's // edge cap), not application preferences — constants, not knobs. These tests -// pin all three surfaces, plus the adjacent read-only-op rejection and -// tiny-collection snapshots. +// pin all three surfaces. import { createCollection } from "@tanstack/db"; import { env, runInDurableObject, SELF } from "cloudflare:test"; @@ -363,130 +362,5 @@ describe("oversize client mutation frames (ADR-0018)", () => { }); }); -describe("read-only collection ops (no mutation handler)", () => { - // `transformed` declares ONLY an insert mutation — update/delete are the - // "no mutations block" case per-op. A collection with NO mutations at all - // takes the identical code path (mutations.get returns undefined). - it("client update on an op with no handler: MutationRejectedError + optimistic rollback", async () => { - const room = "fl-readonly"; - const t = makeTransport(room); - await t.connect(); - - const coll = createCollection( - doCollectionOptions({ - transport: t, - table: "transformed", - getKey: (r) => r.id, - }), - ); - await coll.preload(); - - // Seed via the existing insert mutation (allowed). - await coll.insert({ id: "r1", body: "orig" }).isPersisted.promise; - expect(coll.get("r1")).toMatchObject({ body: "orig" }); - - // update has NO handler on `transformed`. - const tx = coll.update("r1", (d) => { - d.body = "hacked"; - }); - // Optimistic overlay applies immediately... - expect(coll.get("r1")).toMatchObject({ body: "hacked" }); - - let err: unknown; - try { - await tx.isPersisted.promise; - } catch (e) { - err = e; - } - expect(err).toBeDefined(); - // TanStack may wrap the mutationFn error; the root cause is the transport's - // MutationRejectedError carrying the server's message. - const msg = - err instanceof Error - ? `${err.message} ${String((err as { cause?: unknown }).cause ?? "")}` - : String(err); - expect(msg).toMatch(/no mutation handler for 'transformed:update'/); - - // Optimistic overlay rolled back to the confirmed row. - expect(coll.get("r1")).toMatchObject({ body: "orig" }); - - // Server row untouched. - const rows = await runInDurableObject( - env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), - (_i, s) => - Array.from( - s.storage.sql.exec("SELECT body FROM transformed WHERE id='r1'"), - ), - ); - expect((rows[0] as { body: string }).body).toBe("orig"); - t.close(); - }); -}); - -describe("tiny collections", () => { - it("zero-row collection: snapshot completes (snap-end) and preload resolves", async () => { - const room = "fl-empty"; - const t = makeTransport(room); - await t.connect(); - const files = createCollection( - doCollectionOptions({ - transport: t, - table: "files", - getKey: (r) => r.id, - }), - ); - await files.preload(); // would hang if a 0-row snapshot never emitted snap-end - expect(files.size).toBe(0); - t.close(); - }); - - it("single-row collection: snapshot + frequent UPDATEs to the one row stream fine", async () => { - const room = "fl-single"; - const t = makeTransport(room); - await t.connect(); - await runInDurableObject( - env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), - (_i, s) => { - s.storage.sql.exec("INSERT INTO files(id,name) VALUES('meta','v0')"); - }, - ); - const files = createCollection( - doCollectionOptions({ - transport: t, - table: "files", - getKey: (r) => r.id, - }), - ); - await files.preload(); - expect(files.size).toBe(1); - expect(files.get("meta")).toMatchObject({ name: "v0" }); - - // Repeated server-side updates to the single row (metadata churn) — the - // coalescer collapses them; the client must land on the last value. - await runInDurableObject( - env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), - (i) => { - const inst = i as unknown as { - runSyncedWrite: (fn: (sql: SqlStorage) => void) => void; - }; - for (let n = 1; n <= 5; n++) { - inst.runSyncedWrite((sql) => - sql.exec("UPDATE files SET name=? WHERE id='meta'", `v${n}`), - ); - } - }, - ); - const start = Date.now(); - while ((files.get("meta") as { name: string } | undefined)?.name !== "v5") { - if (Date.now() - start > 3000) - throw new Error( - `never converged: ${JSON.stringify(files.get("meta"))}`, - ); - await new Promise((r) => setTimeout(r, 10)); - } - t.close(); - }); -}); - // SqlStorage type for the runInDurableObject callbacks above. type SqlStorage = import("@cloudflare/workers-types").SqlStorage;