From a9d0707ff77f48ebe8638e2c4a7ae21aa8fcd069 Mon Sep 17 00:00:00 2001 From: Tom McKenzie Date: Wed, 22 Jul 2026 13:00:32 +1000 Subject: [PATCH] fix(wire): normalize bare ArrayBuffer (workerd BLOB) to Uint8Array in both codecs MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit workerd's SqlStorage returns BLOB columns as bare ArrayBuffer. @msgpack/msgpack only special-cases ArrayBuffer.isView, so a bare ArrayBuffer fell through to encodeMap and arrived on the client as {} — silently, for snapshots and deltas alike. The JSON debug codec had the same gap: classify() tagged Uint8Array ('u8') but not ArrayBuffer. Normalize at emission (ADR-0017) rather than rejecting BLOB columns in assertSyncCompatible: a BLOB is bytes, Uint8Array is the codecs' existing byte type, and the round-trip is lossless — so make it work instead of failing loud. Binary codec registers a msgpack ExtensionCodec entry (custom type 0) that intercepts bare ArrayBuffer before the encodeMap fallthrough and decodes to a copied Uint8Array (a subarray view would alias the wire buffer — caught by codex adversarial review). JSON codec tags ArrayBuffer as 'u8'; its decode side was already symmetric. tests/row-shape.test.ts pins the end-to-end guarantee: BLOB arrives as Uint8Array with exact bytes for cold snapshots AND live deltas, alongside the existing row-shape guarantees (verbatim snake_case column names, JSON TEXT stays string, INTEGER/REAL/NULL fidelity, >2^53 rounding per #10). Fixes #27 Co-Authored-By: Claude Fable 5 --- CHANGELOG.md | 11 ++ docs/adr/0017-blob-wire-normalization.md | 84 +++++++++ docs/adr/README.md | 1 + src/wire/codec.ts | 11 +- src/wire/frame-codec.ts | 20 ++- tests/codec.test.ts | 11 ++ tests/frame-codec.test.ts | 24 +++ tests/row-shape.test.ts | 211 +++++++++++++++++++++++ 8 files changed, 368 insertions(+), 5 deletions(-) create mode 100644 docs/adr/0017-blob-wire-normalization.md create mode 100644 tests/row-shape.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index a82f5e0..71c5e7e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,17 @@ While pre-1.0, the public API may change between 0.x releases. ## [Unreleased] +### Fixed + +- **BLOB columns no longer corrupt to `{}` over the wire (ADR-0017, + [#27](https://github.com/grrowl/tanstack-durable-object-sync/issues/27)).** + workerd's `SqlStorage` returns BLOB values as bare `ArrayBuffer`, which the + msgpack encoder fell through to `encodeMap` on (and the JSON debug codec + stringified as `{}`) — clients silently received an empty object, for + snapshots and deltas alike. Both codecs now normalize `ArrayBuffer` at + emission, so BLOB columns arrive on the client as a `Uint8Array` with the + exact bytes. + ## [0.5.1] — 2026-07-03 ### Fixed diff --git a/docs/adr/0017-blob-wire-normalization.md b/docs/adr/0017-blob-wire-normalization.md new file mode 100644 index 0000000..fed8e23 --- /dev/null +++ b/docs/adr/0017-blob-wire-normalization.md @@ -0,0 +1,84 @@ +# ADR-0017: BLOB wire normalization — bare ArrayBuffer becomes Uint8Array at emission + +**Status**: Accepted +**Date**: 2026-07-22 +**Issue**: [#27](https://github.com/grrowl/tanstack-durable-object-sync/issues/27) + +## Context + +workerd's `SqlStorage` returns BLOB column values as **bare `ArrayBuffer`**. +Both row-reading paths (`SELECT *` in the sql-compiler's snapshot/fetch reader +and `hydrateRows` for deltas) hand that value straight to the frame codec. + +Neither codec handled it: + +- **Binary (msgpack)**: `@msgpack/msgpack`'s `encodeObject` only special-cases + `ArrayBuffer.isView`, so a bare `ArrayBuffer` fell through to `encodeMap` + and encoded as an **empty map**. The client received `{}` — silently, for + snapshots and deltas alike. +- **JSON debug codec**: `classify()` tagged `Uint8Array` (`u8`) but not + `ArrayBuffer`, and `JSON.stringify(ArrayBuffer)` is `{}` — same corruption. + +Silent data loss is the worst failure mode this repo recognises ("fail loud"), +so *something* had to change. Two candidate shapes: + +1. **Normalize at emission**: teach both codecs that a bare `ArrayBuffer` is + bytes, delivered to the decoder as `Uint8Array`. +2. **Reject in `assertSyncCompatible`**: refuse BLOB-affinity columns at + `registerSync` so authors fail loud instead of losing data. + +## Decision + +**Normalize at emission, in both codecs.** Rejection is only the right tool +when the data genuinely cannot be made safe; here it can be, losslessly: + +- A BLOB is bytes. `Uint8Array` is the codecs' existing byte type (msgpack + `bin`; tagged-codec `u8`), and the decode side already produces `Uint8Array` + for it. Normalizing `ArrayBuffer → Uint8Array` at encode time is a pure + view change over the same bytes — nothing to lose. +- The round-trip is stable: a client writing a `Uint8Array` back through a + mutation stores the same bytes in the BLOB column; the next read emits the + same `Uint8Array`. Optimistic and confirmed values agree. + +Implementation: + +- **Binary codec** (`wire/frame-codec.ts`): an `ExtensionCodec` entry + (custom ext type `0`) — `@msgpack/msgpack` consults the extension codec + *before* the `encodeMap` fallthrough, so it intercepts exactly the bare + `ArrayBuffer` case while `Uint8Array` keeps using the native `bin` format. + Decode returns the bytes as `Uint8Array`, never reconstructing an + `ArrayBuffer`. +- **JSON debug codec** (`wire/codec.ts`): `classify()` tags `ArrayBuffer` as + `u8`; `toPlaceholder` reads its bytes through a `Uint8Array` view. Decode + was already symmetric (`u8 → Uint8Array`) and is unchanged. + +The normalization is one-way by design: **the wire has exactly one byte type, +`Uint8Array`**. Clients never see a bare `ArrayBuffer`, regardless of which +codec or which server runtime produced the row. + +### Why `assertSyncCompatible` does not also warn + +Once normalized, a BLOB column round-trips as faithfully as TEXT or INTEGER — +there is nothing for an author to act on, so a warning would be noise +attached to a working feature. Rejection/warning remains the tool for +genuinely unsafe shapes (e.g. the rowid and pk-affinity rules of ADR-0015 / +0001 D9), not for this one. + +### Out of scope + +INTEGER values above 2^53 still lose precision — workerd's `SqlStorage` +returns JS numbers, so the damage happens before the wire is involved. That +is issue [#10](https://github.com/grrowl/tanstack-durable-object-sync/issues/10) +(wontfix), unaffected by this decision; `tests/row-shape.test.ts` pins the +rounding so a behaviour change is noticed. + +## Consequences + +- BLOB columns are now syncable: clients receive `Uint8Array` with the exact + bytes, for cold snapshots, fetches, and live deltas alike. +- Binary wire format: bare `ArrayBuffer` values encode as msgpack ext type + `0` rather than (corrupt) map. Both ends of the wire ship in this package, + so no cross-version concern exists — but a foreign msgpack decoder would + need the same extension registered to read BLOB values. +- `tests/row-shape.test.ts` pins the end-to-end guarantee (snapshot + delta); + `tests/codec.test.ts` / `tests/frame-codec.test.ts` pin it per codec. diff --git a/docs/adr/README.md b/docs/adr/README.md index 58cf345..2660e95 100644 --- a/docs/adr/README.md +++ b/docs/adr/README.md @@ -22,3 +22,4 @@ explains the displacement. | [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) | +| [0017](./0017-blob-wire-normalization.md) | BLOB wire normalization: bare ArrayBuffer becomes Uint8Array at emission | Accepted | diff --git a/src/wire/codec.ts b/src/wire/codec.ts index 127794b..eed65fb 100644 --- a/src/wire/codec.ts +++ b/src/wire/codec.ts @@ -1,8 +1,9 @@ // Tagged value codec (ADR-0001 D17). // // Lossless transport + at-rest encoding for values JSON cannot represent on its -// own: bigint, Date, NaN, ±Infinity, -0, undefined, and Uint8Array. Naive -// `JSON.stringify` silently corrupts every one of these. +// own: bigint, Date, NaN, ±Infinity, -0, undefined, and Uint8Array (plus bare +// ArrayBuffer, normalized to Uint8Array — ADR-0017). Naive `JSON.stringify` +// silently corrupts every one of these. // // Collision-proof by construction: special values are replaced by neutral // placeholders in the payload, and their (path, type) is recorded in a separate @@ -38,6 +39,10 @@ function classify(v: unknown): TypeTag | null { if (v === null) return null if (v instanceof Date) return "date" if (v instanceof Uint8Array) return "u8" + // workerd's SqlStorage returns BLOB columns as bare ArrayBuffer; bare + // JSON stringifies it as {}. Normalize to bytes at emission — decodes as + // Uint8Array, same as a Uint8Array input (ADR-0017). + if (v instanceof ArrayBuffer) return "u8" return null case "function": case "symbol": @@ -54,7 +59,7 @@ function toPlaceholder(tag: TypeTag, v: unknown): unknown { case "date": return (v as Date).getTime() case "u8": - return Array.from(v as Uint8Array) + return Array.from(v instanceof ArrayBuffer ? new Uint8Array(v) : (v as Uint8Array)) case "-0": return 0 case "nan": diff --git a/src/wire/frame-codec.ts b/src/wire/frame-codec.ts index 499fe90..b8df4d7 100644 --- a/src/wire/frame-codec.ts +++ b/src/wire/frame-codec.ts @@ -13,7 +13,7 @@ // saves ~9 bytes/frame, dwarfed by row payloads, so it is deferred to M9 and // measured rather than assumed. Flagged here rather than silently dropped. -import { decode as mpDecode, encode as mpEncode } from "@msgpack/msgpack" +import { decode as mpDecode, encode as mpEncode, ExtensionCodec } from "@msgpack/msgpack" import { decode as valueDecode, encode as valueEncode } from "./codec.ts" import type { ClientFrame, ServerFrame } from "./frames.ts" @@ -36,7 +36,23 @@ function toBytes(d: WireIn): Uint8Array { return new TextEncoder().encode(d) } -const MSGPACK_OPTS = { useBigInt64: true } as const +// workerd's SqlStorage returns BLOB columns as bare ArrayBuffer. @msgpack/msgpack +// only special-cases ArrayBuffer.isView, so a bare ArrayBuffer would fall through +// to encodeMap and arrive as {} — silently (issue #27, ADR-0017). The extension +// codec is consulted before that fallthrough: encode the raw bytes, decode to +// Uint8Array so BLOBs land on the client exactly like a Uint8Array column value. +const ARRAY_BUFFER_EXT_TYPE = 0 +const extensionCodec = new ExtensionCodec() +extensionCodec.register({ + type: ARRAY_BUFFER_EXT_TYPE, + encode: (v) => (v instanceof ArrayBuffer ? new Uint8Array(v) : null), + // Copy out of the decode buffer: `bytes` is a subarray VIEW of the whole + // incoming frame — returning it as-is would alias the wire buffer (mutable + // by the caller) and keep the full frame alive behind a small BLOB. + decode: (bytes) => bytes.slice(), // Uint8Array — normalized, never back to ArrayBuffer +}) + +const MSGPACK_OPTS = { useBigInt64: true, extensionCodec } as const const binaryCodec: FrameCodec = { binary: true, diff --git a/tests/codec.test.ts b/tests/codec.test.ts index ac589a7..25e84b9 100644 --- a/tests/codec.test.ts +++ b/tests/codec.test.ts @@ -57,6 +57,17 @@ describe("tagged value codec", () => { expect(Array.from(r as Uint8Array)).toEqual([0, 1, 2, 254, 255]) }) + it("normalizes a bare ArrayBuffer (workerd BLOB) to Uint8Array", () => { + // workerd's SqlStorage returns BLOB columns as bare ArrayBuffer, which + // bare JSON stringifies to {} — silent corruption (issue #27). Normalized + // at emission: the decoder yields a Uint8Array with the exact bytes. + const r = roundtrip({ payload: new Uint8Array([1, 2, 255]).buffer }) as { + payload: unknown + } + expect(r.payload).toBeInstanceOf(Uint8Array) + expect(Array.from(r.payload as Uint8Array)).toEqual([1, 2, 255]) + }) + it("does not misinterpret user data shaped like an internal tag", () => { // Collision-proofing: this object must survive as plain data. const v = { $type: "bigint", value: "not-a-bigint", m: [[[], "date"]] } diff --git a/tests/frame-codec.test.ts b/tests/frame-codec.test.ts index 15f97bf..ebab509 100644 --- a/tests/frame-codec.test.ts +++ b/tests/frame-codec.test.ts @@ -66,6 +66,30 @@ for (const binary of [true, false]) { expect(row.nested).toEqual({ a: [1, 2, 3], b: "z" }) }) + it("normalizes a bare ArrayBuffer row value (workerd BLOB) to Uint8Array", () => { + // workerd's SqlStorage returns BLOB columns as bare ArrayBuffer. msgpack + // only special-cases ArrayBuffer.isView, so without normalization a bare + // ArrayBuffer fell through to encodeMap and arrived as {} — silently + // (issue #27). Both codecs must deliver the bytes as a Uint8Array. + const f: ServerFrame = { + t: "snap", + sub: "s", + key: "k", + row: { payload: new Uint8Array([1, 2, 254]).buffer }, + seq: "1", + } + const encoded = codec.encode(f) + const r = codec.decode(encoded) as Extract + const payload = (r.row as Record).payload + expect(payload).toBeInstanceOf(Uint8Array) + expect(Array.from(payload as Uint8Array)).toEqual([1, 2, 254]) + // The decoded bytes must be a COPY, not a view aliasing the wire buffer: + // an aliased view is mutable through the transport's buffer and pins the + // whole frame allocation behind a small BLOB. + if (encoded instanceof Uint8Array) encoded.fill(0) + expect(Array.from(payload as Uint8Array)).toEqual([1, 2, 254]) + }) + it(`encodes as ${binary ? "Uint8Array" : "string"}`, () => { const out = codec.encode({ t: "uptodate", seq: "1" }) if (binary) expect(out).toBeInstanceOf(Uint8Array) diff --git a/tests/row-shape.test.ts b/tests/row-shape.test.ts new file mode 100644 index 0000000..da86be7 --- /dev/null +++ b/tests/row-shape.test.ts @@ -0,0 +1,211 @@ +// WHY: the client row IS the raw `SELECT *` row — no mapping layer, no +// auto-parsing, no type coercion between SQLite and the collection. Apps key +// off exact snake_case column names, treat JSON TEXT as strings, and rely on +// scalar fidelity (INTEGER/REAL/NULL) surviving the wire. BLOB columns come +// out of workerd's SqlStorage as bare ArrayBuffer, which the codecs normalize +// to Uint8Array at emission (ADR-0017, issue #27) — before that, msgpack fell +// through to encodeMap and the client silently received {}. If any of these +// guarantees drift, synced apps corrupt data without an error. + +import { createCollection } from "@tanstack/db"; +import { env, runInDurableObject, SELF } from "cloudflare:test"; +import { describe, expect, it } from "vitest"; +import { + doCollectionOptions, + type WebSocketLike, + WebSocketTransport, +} from "../src/client/index.ts"; +import type { SqlStorage } from "@cloudflare/workers-types"; +import type { TestApi } from "./test-worker.ts"; + +type ServerApi = { runSyncedWrite: (fn: (sql: SqlStorage) => T) => T }; +const api = (i: unknown): ServerApi => i as unknown as ServerApi; + +function realTransport(room: string): WebSocketTransport { + return 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; + }, + }); +} + +async function waitFor(pred: () => boolean, timeoutMs = 3000): Promise { + const start = Date.now(); + while (!pred()) { + if (Date.now() - start > timeoutMs) throw new Error("waitFor timeout"); + await new Promise((r) => setTimeout(r, 5)); + } +} + +const TS = 1753056000000; // ms epoch, well under 2^53 +const BIG = 9007199254740993n; // 2^53 + 1 — NOT representable as a JS number + +describe("client row shape is the raw SELECT * row", () => { + it("snapshot rows carry snake_case SQL column names verbatim; JSON TEXT stays a string; INTEGER/REAL/NULL survive; BLOB arrives as Uint8Array", async () => { + const room = "row-shape-1"; + const t = realTransport(room); + await t.connect(); // constructs the DO -> tables exist, registerSync ran + + // Widen the registered `messages` table with realistic columns. + // The declared client Row type (MsgRow = {id, body}) knows NOTHING of these. + await runInDurableObject( + env.SYNC_DO.get(env.SYNC_DO.idFromName(room)), + (_i, s) => { + const sql = s.storage.sql; + sql.exec("ALTER TABLE messages ADD COLUMN parent_ids TEXT"); + sql.exec("ALTER TABLE messages ADD COLUMN tool_output TEXT"); + sql.exec("ALTER TABLE messages ADD COLUMN created_at INTEGER"); + sql.exec("ALTER TABLE messages ADD COLUMN score REAL"); + sql.exec("ALTER TABLE messages ADD COLUMN is_done INTEGER"); + sql.exec("ALTER TABLE messages ADD COLUMN payload BLOB"); + sql.exec( + "INSERT INTO messages(id, body, parent_ids, tool_output, created_at, score, is_done, payload) VALUES (?,?,?,?,?,?,?,?)", + "m1", + "hello", + '["p1","p2"]', + '{"result":{"ok":true},"n":42}', + TS, + 0.5, + 1, + new Uint8Array([1, 2, 3]).buffer, + ); + // Row with NULLs and a beyond-2^53 integer stored via SQL literal. + sql.exec( + `INSERT INTO messages(id, body, created_at) VALUES ('m2', 'nulls', ${BIG})`, + ); + }, + ); + + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (m) => m.id, + }), + ); + await messages.preload(); + + const row = messages.get("m1") as Record; + expect(row).toBeDefined(); + + // 1. Raw SQL column names, snake_case, no mapping layer. (TanStack DB's + // get() decorates rows with enumerable $-prefixed metadata keys — ignore.) + expect( + Object.keys(row) + .filter((k) => !k.startsWith("$")) + .sort(), + ).toEqual([ + "body", + "created_at", + "id", + "is_done", + "parent_ids", + "payload", + "score", + "tool_output", + ]); + expect("parentIds" in row).toBe(false); + + // 2. JSON TEXT arrives as a STRING — no auto-parse anywhere. + expect(row.parent_ids).toBe('["p1","p2"]'); + expect(typeof row.tool_output).toBe("string"); + expect(JSON.parse(row.tool_output as string)).toEqual({ + result: { ok: true }, + n: 42, + }); + + // 3. Scalars: INTEGER -> number, REAL -> number, boolean-as-0/1 -> number. + expect(row.created_at).toBe(TS); + expect(typeof row.created_at).toBe("number"); + expect(row.score).toBe(0.5); + expect(row.is_done).toBe(1); // number 1, NOT boolean true + expect(typeof row.is_done).toBe("number"); + + // 4. BLOB: workerd's SqlStorage yields a bare ArrayBuffer; the codecs + // normalize it at emission so the client always receives a Uint8Array with + // the exact bytes (ADR-0017, issue #27). Anything else here means BLOBs + // are being corrupted silently again. + expect(row.payload).toBeInstanceOf(Uint8Array); + expect(Array.from(row.payload as Uint8Array)).toEqual([1, 2, 3]); + + // 5. NULL columns arrive as JS null. + const m2 = messages.get("m2") as Record; + expect(m2.parent_ids).toBeNull(); + expect(m2.tool_output).toBeNull(); + expect(m2.score).toBeNull(); + + // 6. INTEGER beyond 2^53: workerd's SqlStorage returns JS numbers, so + // 2^53+1 rounds to 2^53 BEFORE the wire is involved. Known-lossy; see + // issue #10 (wontfix) — pinned so a behaviour change is noticed. + expect(m2.created_at).toBe(9007199254740992); + + t.close(); + }); + + it("live delta rows (server-originated UPDATE) carry raw column names, string JSON, and Uint8Array BLOBs", async () => { + const room = "row-shape-2"; + const stub = env.SYNC_DO.get(env.SYNC_DO.idFromName(room)); + const t = realTransport(room); + await t.connect(); + + await runInDurableObject(stub, (_i, s) => { + const sql = s.storage.sql; + sql.exec("ALTER TABLE messages ADD COLUMN tool_output TEXT"); + sql.exec("ALTER TABLE messages ADD COLUMN updated_at INTEGER"); + sql.exec("ALTER TABLE messages ADD COLUMN payload BLOB"); + sql.exec( + "INSERT INTO messages(id, body, tool_output, updated_at) VALUES ('s1','v1',NULL,1)", + ); + }); + + const messages = createCollection( + doCollectionOptions({ + transport: t, + table: "messages", + getKey: (m) => m.id, + }), + ); + await messages.preload(); + expect((messages.get("s1") as Record).body).toBe("v1"); + + // Streaming-style server write: frequent row UPDATE via runSyncedWrite. + await runInDurableObject(stub, (instance) => { + api(instance).runSyncedWrite((sql) => + sql.exec( + "UPDATE messages SET body = ?, tool_output = ?, updated_at = ?, payload = ? WHERE id = ?", + "v2", + '{"chunks":["a","b"]}', + TS, + new Uint8Array([9, 8, 7]).buffer, + "s1", + ), + ); + }); + + await waitFor( + () => (messages.get("s1") as Record).body === "v2", + ); + const row = messages.get("s1") as Record; + // Delta hydration is SELECT * too — raw snake_case keys, string JSON, + // number int, and the same BLOB -> Uint8Array normalization as snapshots. + expect( + Object.keys(row) + .filter((k) => !k.startsWith("$")) + .sort(), + ).toEqual(["body", "id", "payload", "tool_output", "updated_at"]); + expect(row.tool_output).toBe('{"chunks":["a","b"]}'); + expect(typeof row.tool_output).toBe("string"); + expect(row.updated_at).toBe(TS); + expect(row.payload).toBeInstanceOf(Uint8Array); + expect(Array.from(row.payload as Uint8Array)).toEqual([9, 8, 7]); + + t.close(); + }); +});