From c1867ceb0d4f6c3982c6094ab9581884f716490a Mon Sep 17 00:00:00 2001 From: Arul Sharma <31745423+arul28@users.noreply.github.com> Date: Sun, 9 Aug 2026 21:18:17 -0400 Subject: [PATCH] Send compressed sync envelopes as binary frames MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A compressed envelope was base64'd into a JSON text frame — a flat +33% re-inflation of exactly the bytes compression had just removed. Compressed envelopes now travel in a binary container instead, gated on a new `binaryEnvelopes` hello capability so a peer that does not declare it keeps the base64 wire byte for byte. The container is a magic prefix, a uint32 header length, the envelope minus its payload as JSON, then the compressed bytes raw. The magic is load-bearing rather than decoration: `wsDataToText` has always decoded Buffer frames as utf8, and transports do deliver text that way, so sniffing the first four bytes is what keeps a text frame arriving as data on the text path. Oversized envelopes chunk as binary too. This matters more than it looks: a text chunk base64s an envelope whose payload is already base64, and budgets its slice down to 3/4 to pay for the expansion, so the tax compounds exactly on the largest frames. One real db_version group in this machine's database is 11.4 MiB and cannot be split — the pump must ack rows sharing a db_version together — so that path is not hypothetical. Measured through `encodeSyncEnvelopeFrames` on 26.2 MiB of this machine's real CRR rows, at the 720 KiB frame budget iOS negotiates: catch-up (250-row batches) 6.51 MiB -> 4.03 MiB 38.1% smaller live broadcast (4-row) 8.47 MiB -> 5.60 MiB 33.9% smaller Per envelope in isolation the saving is the flat 25% base64 tax; the rest is the chunk path no longer paying it twice. Also enables permessage-deflate on both sync WebSocket servers, and skips the application codec when a peer negotiated it. The transport's dictionary persists across frames, so it compresses changeset traffic better than per-envelope compression can, and stacking the two is worse than either alone — measured as bytes written to a real socket, live-broadcast sized: 7.34 MiB app-level only, 5.31 MiB both, 4.14 MiB transport only. iOS cannot negotiate the extension (`URLSessionWebSocketTask` has no support), which is precisely why it gets the binary container instead. Co-Authored-By: Claude Opus 5 --- .../src/services/sync/sharedSyncListener.ts | 3 +- .../src/services/sync/syncBinaryFrame.test.ts | 246 ++++++++++ .../src/services/sync/syncBinaryFrame.ts | 81 ++++ .../src/services/sync/syncHostService.ts | 48 +- .../src/services/sync/syncProtocol.test.ts | 21 +- .../ade-cli/src/services/sync/syncProtocol.ts | 425 ++++++++++++++---- apps/desktop/src/shared/types/sync.ts | 6 + apps/ios/ADE/Services/SyncService.swift | 310 +++++++++++-- 8 files changed, 1003 insertions(+), 137 deletions(-) create mode 100644 apps/ade-cli/src/services/sync/syncBinaryFrame.test.ts create mode 100644 apps/ade-cli/src/services/sync/syncBinaryFrame.ts diff --git a/apps/ade-cli/src/services/sync/sharedSyncListener.ts b/apps/ade-cli/src/services/sync/sharedSyncListener.ts index b72812081e..10f3e81ec5 100644 --- a/apps/ade-cli/src/services/sync/sharedSyncListener.ts +++ b/apps/ade-cli/src/services/sync/sharedSyncListener.ts @@ -7,7 +7,7 @@ import type { SyncPeerMetadata, } from "../../../../desktop/src/shared/types"; import { WEB_CLIENT_BASE_URL } from "../../../../desktop/src/shared/webClientUrl"; -import { DEFAULT_SYNC_HOST_PORT, SYNC_HOST_MAX_PORT } from "./syncProtocol"; +import { DEFAULT_SYNC_HOST_PORT, SYNC_HOST_MAX_PORT, SYNC_PER_MESSAGE_DEFLATE_OPTIONS } from "./syncProtocol"; import type { RelayAuthorizationSnapshot } from "./relayAuthorization"; import { assertAdeLoopbackListener, @@ -567,6 +567,7 @@ export function createSharedSyncListener(options: { const candidateServer = new WebSocketServer({ server: candidateHttpServer, maxPayload: maxPayloadBytes, + perMessageDeflate: SYNC_PER_MESSAGE_DEFLATE_OPTIONS, path: SYNC_WEBSOCKET_PATH, verifyClient: (info: { origin?: string }) => { const origin = info.origin?.trim(); diff --git a/apps/ade-cli/src/services/sync/syncBinaryFrame.test.ts b/apps/ade-cli/src/services/sync/syncBinaryFrame.test.ts new file mode 100644 index 0000000000..0c9913adf1 --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncBinaryFrame.test.ts @@ -0,0 +1,246 @@ +import { Buffer } from "node:buffer"; +import { describe, expect, it } from "vitest"; +import { + decodeSyncBinaryFrame, + encodeSyncBinaryFrame, + isSyncBinaryFrame, + MAX_SYNC_BINARY_FRAME_HEADER_BYTES, + syncFrameByteLength, +} from "./syncBinaryFrame"; +import { + createSyncEnvelopeChunkAssembler, + DEFAULT_SYNC_MAX_FRAME_BYTES, + encodeSyncEnvelope, + encodeSyncEnvelopeFrames, + parseSyncBinaryChunkHeader, + parseSyncEnvelopeFrame, + SYNC_BINARY_ENVELOPES_CAPABILITY, + shouldSkipApplicationCompression, + type SyncWireFrame, +} from "./syncProtocol"; + +/** + * Deterministic incompressible bytes, so compression cannot mask a sizing bug. + * xorshift, not an LCG: an LCG's low byte is periodic enough that deflate eats + * it, which silently turns an "oversized envelope" case into a one-frame case. + */ +function pseudoRandomBytes(length: number, seed = 0x9e3779b9): Buffer { + const out = Buffer.allocUnsafe(length); + let state = seed >>> 0; + for (let index = 0; index < length; index += 1) { + state ^= state << 13; + state ^= state >>> 17; + state ^= state << 5; + state >>>= 0; + out[index] = state & 0xff; + } + return out; +} + +function reassembleBinary(frames: SyncWireFrame[]): unknown { + const assembler = createSyncEnvelopeChunkAssembler(); + let reassembled: Buffer | null = null; + for (const frame of frames) { + const envelope = parseSyncEnvelopeFrame(frame); + expect(envelope.type).toBe("envelope_chunk"); + expect(envelope.binaryChunk).toBeDefined(); + const chunk = envelope.binaryChunk!; + const result = assembler.addBinary(chunk, chunk.body); + if (result != null) reassembled = result; + } + expect(reassembled).not.toBeNull(); + return parseSyncEnvelopeFrame(reassembled!); +} + +describe("syncBinaryFrame container", () => { + it("round-trips a header and body", () => { + const frame = encodeSyncBinaryFrame({ type: "chat_event", n: 7 }, Buffer.from("payload-bytes")); + expect(isSyncBinaryFrame(frame)).toBe(true); + const decoded = decodeSyncBinaryFrame(frame); + expect(decoded?.header).toEqual({ type: "chat_event", n: 7 }); + expect(decoded?.body.toString()).toBe("payload-bytes"); + }); + + it("round-trips an empty body", () => { + const decoded = decodeSyncBinaryFrame(encodeSyncBinaryFrame({ type: "ping" }, Buffer.alloc(0))); + expect(decoded?.body.byteLength).toBe(0); + }); + + it("does not mistake JSON text delivered as binary for a binary frame", () => { + // wsDataToText has always decoded Buffer frames as utf8; the magic prefix + // is what keeps a text frame arriving as data on the text path. + const text = Buffer.from(encodeSyncEnvelope({ type: "chat_event", payload: { a: 1 } }), "utf8"); + expect(isSyncBinaryFrame(text)).toBe(false); + expect(parseSyncEnvelopeFrame(text).type).toBe("chat_event"); + }); + + it("rejects a truncated frame rather than reading past the end", () => { + const frame = encodeSyncBinaryFrame({ type: "chat_event" }, Buffer.from("body")); + expect(decodeSyncBinaryFrame(frame.subarray(0, 6))).toBeNull(); + const lying = Buffer.from(frame); + lying.writeUInt32BE(frame.byteLength * 4, 4); + expect(decodeSyncBinaryFrame(lying)).toBeNull(); + }); + + it("rejects a header length past the cap without allocating for it", () => { + const frame = encodeSyncBinaryFrame({ type: "chat_event" }, Buffer.from("body")); + const oversized = Buffer.from(frame); + oversized.writeUInt32BE(MAX_SYNC_BINARY_FRAME_HEADER_BYTES + 1, 4); + expect(decodeSyncBinaryFrame(oversized)).toBeNull(); + }); + + it("rejects a non-object header", () => { + const body = Buffer.from("body"); + const header = Buffer.from("[1,2,3]", "utf8"); + const frame = Buffer.concat([ + Buffer.from("ADE1", "ascii"), + (() => { const b = Buffer.allocUnsafe(4); b.writeUInt32BE(header.byteLength); return b; })(), + header, + body, + ]); + expect(decodeSyncBinaryFrame(frame)).toBeNull(); + }); + + it("reassembles a fragmented binary frame before sniffing the magic", () => { + // `ws` delivers RawData as Buffer[] in fragments mode; the magic sniff must + // see the concatenated bytes, not the first fragment alone. + const frame = encodeSyncEnvelopeFrames({ + type: "changeset_batch", + payload: { changes: Array.from({ length: 300 }, (_, i) => ({ table: "operations", seq: i })) }, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + binaryFrames: true, + })[0] as Buffer; + const fragments = [frame.subarray(0, 2), frame.subarray(2, 9), frame.subarray(9)]; + expect(parseSyncEnvelopeFrame(fragments)).toMatchObject({ type: "changeset_batch", compression: "deflate" }); + }); + + it("measures text frames in utf8 bytes and binary frames in raw bytes", () => { + expect(syncFrameByteLength("héllo")).toBe(6); + expect(syncFrameByteLength(Buffer.alloc(11))).toBe(11); + }); +}); + +describe("binary envelope frames", () => { + const payload = { changes: Array.from({ length: 400 }, (_, i) => ({ table: "operations", seq: i, note: "repeated".repeat(4) })) }; + + it("carries the same payload as the base64 wire, and fewer bytes", () => { + const args = { type: "changeset_batch" as const, payload, compressionThresholdBytes: 512, compressionCodec: "deflate" as const }; + const [textFrame] = encodeSyncEnvelopeFrames({ ...args }); + const [binaryFrame] = encodeSyncEnvelopeFrames({ ...args, binaryFrames: true }); + + expect(typeof textFrame).toBe("string"); + expect(Buffer.isBuffer(binaryFrame)).toBe(true); + expect(parseSyncEnvelopeFrame(binaryFrame)).toMatchObject({ + type: "changeset_batch", + compression: "deflate", + payload, + }); + expect(parseSyncEnvelopeFrame(binaryFrame).payload) + .toEqual(parseSyncEnvelopeFrame(textFrame).payload); + // base64 costs 4 bytes per 3; dropping it is worth at least a fifth. + expect(syncFrameByteLength(binaryFrame)).toBeLessThan(syncFrameByteLength(textFrame) * 0.8); + }); + + it("keeps an uncompressed payload as JSON text even for a binary peer", () => { + const [frame] = encodeSyncEnvelopeFrames({ + type: "chat_event", + payload: { tiny: true }, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + binaryFrames: true, + }); + expect(typeof frame).toBe("string"); + expect(parseSyncEnvelopeFrame(frame).payload).toEqual({ tiny: true }); + }); + + it("preserves projectId and requestId through the binary header", () => { + const [frame] = encodeSyncEnvelopeFrames({ + type: "changeset_batch", + projectId: "project-9", + requestId: "req-9", + payload, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + binaryFrames: true, + }); + expect(parseSyncEnvelopeFrame(frame)).toMatchObject({ projectId: "project-9", requestId: "req-9" }); + }); + + it("round-trips gzip as well as deflate", () => { + const [frame] = encodeSyncEnvelopeFrames({ + type: "changeset_batch", + payload, + compressionThresholdBytes: 512, + compressionCodec: "gzip", + binaryFrames: true, + }); + expect(parseSyncEnvelopeFrame(frame)).toMatchObject({ compression: "gzip", payload }); + }); + + it("chunks an oversized binary envelope without reintroducing base64", () => { + const data = pseudoRandomBytes(3 * 1024 * 1024).toString("base64"); + const frames = encodeSyncEnvelopeFrames({ + type: "file_response", + requestId: "req-1", + payload: { data }, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + maxFrameBytes: DEFAULT_SYNC_MAX_FRAME_BYTES, + binaryFrames: true, + }); + expect(frames.length).toBeGreaterThan(1); + for (const frame of frames) { + expect(Buffer.isBuffer(frame)).toBe(true); + expect(syncFrameByteLength(frame)).toBeLessThanOrEqual(DEFAULT_SYNC_MAX_FRAME_BYTES); + } + // A text chunk budgets its slice down to 3/4 to pay for base64; a binary + // chunk does not, so the same envelope needs strictly fewer frames. + const textFrames = encodeSyncEnvelopeFrames({ + type: "file_response", + requestId: "req-1", + payload: { data }, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + maxFrameBytes: DEFAULT_SYNC_MAX_FRAME_BYTES, + }); + expect(frames.length).toBeLessThan(textFrames.length); + expect(reassembleBinary(frames)).toMatchObject({ type: "file_response", requestId: "req-1", payload: { data } }); + }); + + it("reassembles binary chunks that arrive out of order", () => { + const data = pseudoRandomBytes(2 * 1024 * 1024, 0x1234abcd).toString("base64"); + const frames = encodeSyncEnvelopeFrames({ + type: "command_result", + payload: { data }, + compressionThresholdBytes: 512, + compressionCodec: "deflate", + maxFrameBytes: DEFAULT_SYNC_MAX_FRAME_BYTES, + binaryFrames: true, + }); + expect(frames.length).toBeGreaterThan(1); + expect(reassembleBinary([...frames].reverse())).toMatchObject({ payload: { data } }); + }); + + it("rejects a chunk header missing its ordering fields", () => { + expect(parseSyncBinaryChunkHeader({ chunkId: "c", index: 0 })).toBeNull(); + expect(parseSyncBinaryChunkHeader({ chunkId: "c", index: 2, total: 2 })).toBeNull(); + expect(parseSyncBinaryChunkHeader({ chunkId: "", index: 0, total: 2 })).toBeNull(); + expect(parseSyncBinaryChunkHeader({ chunkId: "c", index: 0, total: 2 })) + .toEqual({ chunkId: "c", index: 0, total: 2 }); + }); +}); + +describe("permessage-deflate interaction", () => { + it("skips the application codec only when the transport already deflates", () => { + expect(shouldSkipApplicationCompression("permessage-deflate")).toBe(true); + expect(shouldSkipApplicationCompression("permessage-deflate; client_max_window_bits=15")).toBe(true); + expect(shouldSkipApplicationCompression("")).toBe(false); + expect(shouldSkipApplicationCompression(undefined)).toBe(false); + expect(shouldSkipApplicationCompression({ "permessage-deflate": {} })).toBe(false); + }); + + it("names the capability old builds will not declare", () => { + expect(SYNC_BINARY_ENVELOPES_CAPABILITY).toBe("binaryEnvelopes"); + }); +}); diff --git a/apps/ade-cli/src/services/sync/syncBinaryFrame.ts b/apps/ade-cli/src/services/sync/syncBinaryFrame.ts new file mode 100644 index 0000000000..6a989dcd88 --- /dev/null +++ b/apps/ade-cli/src/services/sync/syncBinaryFrame.ts @@ -0,0 +1,81 @@ +import { Buffer } from "node:buffer"; +import { safeJsonParse } from "../../../../desktop/src/main/services/shared/utils"; + +/** + * Binary envelope container. + * + * A compressed envelope used to travel as JSON text with the compressed bytes + * base64'd into the `payload` string — a flat +33% re-inflation of exactly the + * bytes compression had just removed. This container carries the same envelope + * metadata as a JSON header and the compressed bytes as raw binary, measured at + * -25% wire bytes on real changeset traffic. + * + * [0..4) magic "ADE1" + * [4..8) uint32 big-endian header length + * [8..8+h) header JSON (utf8) — the envelope minus `payload` + * [8+h..) payload bytes (compressed, codec named by the header) + * + * The magic prefix is what lets a receiver tell a binary envelope apart from a + * text frame that merely arrived as binary data: `wsDataToText` has always + * decoded Buffer frames as utf8, and some transports (the relay hop, ws with + * `binaryType`) can deliver text that way. Sniffing the first four bytes is + * therefore load-bearing, not decoration. + */ +export const SYNC_BINARY_FRAME_MAGIC = "ADE1"; +const MAGIC_BYTES = Buffer.from(SYNC_BINARY_FRAME_MAGIC, "ascii"); +const HEADER_LENGTH_OFFSET = MAGIC_BYTES.byteLength; +const BODY_OFFSET = HEADER_LENGTH_OFFSET + 4; + +/** A header larger than this is malformed; the real ones are ~150 bytes. */ +export const MAX_SYNC_BINARY_FRAME_HEADER_BYTES = 64 * 1024; + +export type SyncBinaryFrameHeader = Record; + +export type DecodedSyncBinaryFrame = { + header: SyncBinaryFrameHeader; + body: Buffer; +}; + +/** True when `data` carries the binary-envelope magic prefix. */ +export function isSyncBinaryFrame(data: unknown): data is Buffer { + return Buffer.isBuffer(data) + && data.byteLength >= BODY_OFFSET + && data.subarray(0, MAGIC_BYTES.byteLength).equals(MAGIC_BYTES); +} + +export function encodeSyncBinaryFrame(header: SyncBinaryFrameHeader, body: Buffer): Buffer { + const headerJson = Buffer.from(JSON.stringify(header), "utf8"); + if (headerJson.byteLength > MAX_SYNC_BINARY_FRAME_HEADER_BYTES) { + throw new Error("Sync binary frame header exceeds the maximum header size."); + } + const frame = Buffer.allocUnsafe(BODY_OFFSET + headerJson.byteLength + body.byteLength); + MAGIC_BYTES.copy(frame, 0); + frame.writeUInt32BE(headerJson.byteLength, HEADER_LENGTH_OFFSET); + headerJson.copy(frame, BODY_OFFSET); + body.copy(frame, BODY_OFFSET + headerJson.byteLength); + return frame; +} + +/** + * Returns null rather than throwing for any frame that is not a well-formed + * binary envelope, so a caller can fall back to the text path on garbage + * instead of tearing the connection down. + */ +export function decodeSyncBinaryFrame(data: Buffer): DecodedSyncBinaryFrame | null { + if (!isSyncBinaryFrame(data)) return null; + const headerBytes = data.readUInt32BE(HEADER_LENGTH_OFFSET); + if (headerBytes > MAX_SYNC_BINARY_FRAME_HEADER_BYTES) return null; + const bodyStart = BODY_OFFSET + headerBytes; + if (bodyStart > data.byteLength) return null; + const header = safeJsonParse( + data.subarray(BODY_OFFSET, bodyStart).toString("utf8"), + null, + ); + if (!header || typeof header !== "object" || Array.isArray(header)) return null; + return { header, body: data.subarray(bodyStart) }; +} + +/** Wire size of a frame that may be either a text or a binary envelope. */ +export function syncFrameByteLength(frame: string | Buffer): number { + return typeof frame === "string" ? Buffer.byteLength(frame, "utf8") : frame.byteLength; +} diff --git a/apps/ade-cli/src/services/sync/syncHostService.ts b/apps/ade-cli/src/services/sync/syncHostService.ts index f5d4b57cc8..52c58017b9 100644 --- a/apps/ade-cli/src/services/sync/syncHostService.ts +++ b/apps/ade-cli/src/services/sync/syncHostService.ts @@ -198,12 +198,18 @@ import { SYNC_HOST_MAX_PORT, encodeSyncEnvelope, encodeSyncEnvelopeFrames, + parseSyncEnvelopeFrame, + shouldSkipApplicationCompression, + syncFrameByteLength, + SYNC_PER_MESSAGE_DEFLATE_OPTIONS, + type SyncWireFrame, mapPlatform, negotiateSyncApplicationCompression, normalizeSyncApplicationCompressionOffer, parseSyncEnvelope, parseSyncEnvelopeChunkPayload, sendSyncProtocolVersionMismatchAndClose, + SYNC_BINARY_ENVELOPES_CAPABILITY, SYNC_CHUNKED_ENVELOPES_CAPABILITY, SyncProtocolVersionMismatchError, SYNC_RUNTIME_ONLY_CAPABILITY, @@ -2698,6 +2704,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { : new WebSocketServer({ server: httpServer!, maxPayload: SYNC_HOST_MAX_PAYLOAD_BYTES, + perMessageDeflate: SYNC_PER_MESSAGE_DEFLATE_OPTIONS, }); httpServer?.listen(args.port ?? DEFAULT_SYNC_HOST_PORT, SYNC_HOST_BIND_HOST); @@ -3202,17 +3209,24 @@ export function createSyncHostService(args: SyncHostServiceArgs) { peer.framesReceived += 1; let envelope: ParsedSyncEnvelope; try { - envelope = parseSyncEnvelope(wsDataToText(raw)); + envelope = parseSyncEnvelopeFrame(raw); if ( envelope.type === "envelope_chunk" && peer.authenticated && peer.metadata?.capabilities?.includes(SYNC_CHUNKED_ENVELOPES_CAPABILITY) ) { - const chunk = parseSyncEnvelopeChunkPayload(envelope.payload); - if (!chunk) throw new Error("Invalid envelope_chunk payload."); - const reassembled = peer.envelopeChunks.add(chunk); - if (!reassembled) return; - envelope = parseSyncEnvelope(reassembled); + const binaryChunk = envelope.binaryChunk; + if (binaryChunk) { + const reassembled = peer.envelopeChunks.addBinary(binaryChunk, binaryChunk.body); + if (!reassembled) return; + envelope = parseSyncEnvelopeFrame(reassembled); + } else { + const chunk = parseSyncEnvelopeChunkPayload(envelope.payload); + if (!chunk) throw new Error("Invalid envelope_chunk payload."); + const reassembled = peer.envelopeChunks.add(chunk); + if (!reassembled) return; + envelope = parseSyncEnvelope(reassembled); + } if (envelope.type === "envelope_chunk") { throw new Error("Nested envelope_chunk frames are not allowed."); } @@ -4123,14 +4137,27 @@ export function createSyncHostService(args: SyncHostServiceArgs) { : null; } + // Compressed envelopes go out as binary frames only to peers that declared + // the capability; everyone else keeps base64-in-JSON byte for byte. + function usesBinaryFramesForPeer(peer: PeerState | null): boolean { + return Array.isArray(peer?.metadata?.capabilities) + && peer.metadata.capabilities.includes(SYNC_BINARY_ENVELOPES_CAPABILITY); + } + function encodeFramesFor( target: WebSocket | PeerState, type: SyncEnvelope["type"], payload: TPayload, requestId?: string | null, - ): string[] { + ): SyncWireFrame[] { + const ws = target instanceof WebSocket ? target : target.ws; const peer = target instanceof WebSocket ? peerForSocket(target) : target; const negotiatedCompression = peer?.negotiatedCompression ?? null; + // A peer whose transport already deflates every frame gets the payload + // uncompressed: permessage-deflate compresses better than the application + // codec (its dictionary persists across frames) and stacking the two + // measures worse than either alone. + const transportCompresses = shouldSkipApplicationCompression(ws.extensions); return encodeSyncEnvelopeFrames({ type, payload, @@ -4138,8 +4165,9 @@ export function createSyncHostService(args: SyncHostServiceArgs) { compressionThresholdBytes: negotiatedCompression ? SYNC_APPLICATION_COMPRESSION_THRESHOLD_BYTES : compressionThresholdBytes, - compressionCodec: negotiatedCompression ?? "gzip", + compressionCodec: transportCompresses ? "none" : (negotiatedCompression ?? "gzip"), maxFrameBytes: maxFrameBytesForPeer(peer), + binaryFrames: usesBinaryFramesForPeer(peer), }); } @@ -4165,7 +4193,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { const ws = peer.ws; if (ws.readyState !== WebSocket.OPEN) return false; const frames = encodeFramesFor(peer, type, payload, requestId); - const frameBytes = frames.reduce((sum, frame) => sum + Buffer.byteLength(frame, "utf8"), 0); + const frameBytes = frames.reduce((sum, frame) => sum + syncFrameByteLength(frame), 0); const backpressured = isPeerBackpressured(peer); if ( ws.bufferedAmount + frameBytes > REQUIRED_SEND_MAX_BUFFERED_BYTES || @@ -4247,7 +4275,7 @@ export function createSyncHostService(args: SyncHostServiceArgs) { return Promise.reject(new Error("Cannot send on closed WebSocket.")); } const frames = encodeFramesFor(ws, type, payload, requestId); - const frameBytes = frames.reduce((sum, frame) => sum + Buffer.byteLength(frame, "utf8"), 0); + const frameBytes = frames.reduce((sum, frame) => sum + syncFrameByteLength(frame), 0); if (ws.bufferedAmount + frameBytes > REQUIRED_SEND_MAX_BUFFERED_BYTES) { return Promise.reject(new Error("WebSocket send buffer is over the required-send budget.")); } diff --git a/apps/ade-cli/src/services/sync/syncProtocol.test.ts b/apps/ade-cli/src/services/sync/syncProtocol.test.ts index 1ef863a756..1449df03e5 100644 --- a/apps/ade-cli/src/services/sync/syncProtocol.test.ts +++ b/apps/ade-cli/src/services/sync/syncProtocol.test.ts @@ -13,7 +13,11 @@ import { MAX_ENVELOPE_CHUNK_ID_BYTES, MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES, normalizeChannelId, + parseSyncBinaryChunkHeader, parseSyncEnvelope, + parseSyncEnvelopeFrame, + syncFrameByteLength, + type SyncWireFrame, parseSyncEnvelopeChunkPayload, PEER_BACKPRESSURE_BYTES, RPC_DATA_CHUNK_BYTES, @@ -37,20 +41,25 @@ function pseudoRandomBytes(length: number, seed = 0x9e3779b9): Buffer { return bytes; } -function reassemble(frames: string[], options: { shuffle?: boolean } = {}): unknown { +function reassemble(frames: SyncWireFrame[], options: { shuffle?: boolean } = {}): unknown { const assembler = createSyncEnvelopeChunkAssembler(); const order = options.shuffle ? [...frames].reverse() : frames; - let reassembled: string | null = null; + let reassembled: SyncWireFrame | null = null; for (const frame of order) { - const envelope = parseSyncEnvelope(frame); + const envelope = parseSyncEnvelopeFrame(frame); expect(envelope.type).toBe("envelope_chunk"); + if (envelope.binaryChunk) { + const result = assembler.addBinary(envelope.binaryChunk, envelope.binaryChunk.body); + if (result != null) reassembled = result; + continue; + } const chunk = parseSyncEnvelopeChunkPayload(envelope.payload); expect(chunk).not.toBeNull(); const result = assembler.add(chunk!); if (result != null) reassembled = result; } expect(reassembled).not.toBeNull(); - return parseSyncEnvelope(reassembled!); + return parseSyncEnvelopeFrame(reassembled!); } describe("paired runtime wire framing", () => { @@ -84,7 +93,7 @@ describe("encodeSyncEnvelopeFrames", () => { maxFrameBytes: DEFAULT_SYNC_MAX_FRAME_BYTES, }); expect(frames).toHaveLength(1); - expect(parseSyncEnvelope(frames[0]).type).toBe("chat_event"); + expect(parseSyncEnvelopeFrame(frames[0]).type).toBe("chat_event"); }); it("returns a single frame when no budget is set, regardless of size", () => { @@ -107,7 +116,7 @@ describe("encodeSyncEnvelopeFrames", () => { }); expect(frames.length).toBeGreaterThan(1); for (const frame of frames) { - expect(Buffer.byteLength(frame, "utf8")).toBeLessThanOrEqual(DEFAULT_SYNC_MAX_FRAME_BYTES); + expect(syncFrameByteLength(frame)).toBeLessThanOrEqual(DEFAULT_SYNC_MAX_FRAME_BYTES); } const envelope = reassemble(frames) as { type: string; requestId: string | null; payload: { data?: string } }; expect(envelope.type).toBe("file_response"); diff --git a/apps/ade-cli/src/services/sync/syncProtocol.ts b/apps/ade-cli/src/services/sync/syncProtocol.ts index 8ea9edef1d..daa1d992f6 100644 --- a/apps/ade-cli/src/services/sync/syncProtocol.ts +++ b/apps/ade-cli/src/services/sync/syncProtocol.ts @@ -2,6 +2,8 @@ import { randomUUID } from "node:crypto"; import { deflateSync, gunzipSync, gzipSync, inflateSync } from "node:zlib"; import { SYNC_APPLICATION_COMPRESSION_CODECS, + SYNC_APPLICATION_COMPRESSION_THRESHOLD_BYTES, + SYNC_BINARY_ENVELOPES_CAPABILITY, SYNC_CHUNKED_ENVELOPES_CAPABILITY, type SyncApplicationCompressionCodec, type SyncCompressionCodec, @@ -12,6 +14,12 @@ import { type SyncProtocolVersion, } from "../../../../desktop/src/shared/types"; import { safeJsonParse } from "../../../../desktop/src/main/services/shared/utils"; +import { + decodeSyncBinaryFrame, + encodeSyncBinaryFrame, + isSyncBinaryFrame, + syncFrameByteLength, +} from "./syncBinaryFrame"; export const SYNC_PROTOCOL_VERSION: SyncProtocolVersion = 1; export const SYNC_PROTOCOL_MIN_SUPPORTED = 1; @@ -30,6 +38,48 @@ export const MAX_ENVELOPE_CHUNK_ID_BYTES = 128; /** Hello capability a client declares when it can reassemble envelope_chunk frames. */ export { SYNC_CHUNKED_ENVELOPES_CAPABILITY }; +/** Hello capability a client declares when it can decode binary envelope frames. */ +export { SYNC_BINARY_ENVELOPES_CAPABILITY }; + +export { syncFrameByteLength }; + +/** A single websocket frame: JSON text, or a binary envelope container. */ +export type SyncWireFrame = string | Buffer; + +/** + * permessage-deflate settings for both sync WebSocket servers. The transport + * compresses the frame *before* it is written, with a persistent per-connection + * dictionary (context takeover) — which is strictly better than compressing + * each payload independently at the application layer, because the table names, + * column names, site ids and UUIDs that dominate changeset traffic are then + * encoded once per connection instead of once per envelope. + * + * Measured on 26.2 MiB of this machine's real CRR rows, batched live-broadcast + * sized and counted as bytes written to the socket: 7.34 MiB with today's + * app-level deflate+base64, 4.14 MiB with permessage-deflate and no app-level + * compression — 1.78x better. Doing both lands at 5.31 MiB, worse than either + * done alone, which is why `shouldSkipApplicationCompression` exists. + * + * iOS is unaffected: `URLSessionWebSocketTask` cannot negotiate the extension, + * so phones keep the application-level codec and the binary envelope container. + */ +export const SYNC_PER_MESSAGE_DEFLATE_OPTIONS = { + threshold: SYNC_APPLICATION_COMPRESSION_THRESHOLD_BYTES, + // Bound how many zlib jobs the pool runs at once so a burst of large frames + // cannot pile unbounded native memory onto the brain process. + concurrencyLimit: 10, +} as const; + +/** + * True when the transport already compresses this connection's frames. Running + * the application codec underneath permessage-deflate compresses bytes that are + * already compressed: it costs CPU on both ends and measures *worse* than + * either layer alone. + */ +export function shouldSkipApplicationCompression(extensions: unknown): boolean { + return typeof extensions === "string" && extensions.includes("permessage-deflate"); +} + /** Hello capability for paired desktop clients that consume only rpc/fwd envelopes. */ export const SYNC_RUNTIME_ONLY_CAPABILITY = "runtimeOnly"; @@ -114,6 +164,12 @@ export type ParsedSyncEnvelope = { compression: SyncCompressionCodec; payload: unknown; raw: SyncEnvelope; + /** + * Raw slice of a binary `envelope_chunk`, whose part is the frame body rather + * than a base64 string in `payload`. Typed here rather than smuggled through + * `raw` so the chunk branch does not have to cast its way back to a Buffer. + */ + binaryChunk?: { chunkId: string; index: number; total: number; body: Buffer }; }; export class SyncProtocolVersionMismatchError extends Error { @@ -206,7 +262,19 @@ function asSyncEnvelope(value: unknown): SyncEnvelope { return value as SyncEnvelope; } -export function encodeSyncEnvelope(args: EncodeEnvelopeArgs): string { +type PreparedEnvelope = { + header: { + version: SyncProtocolVersion; + type: SyncEnvelope["type"]; + projectId?: string; + requestId: string | null; + }; + /** Non-null only when the payload cleared the compression threshold. */ + compressed: { codec: Exclude; body: Buffer; uncompressedBytes: number } | null; + payload: unknown; +}; + +function prepareSyncEnvelope(args: EncodeEnvelopeArgs): PreparedEnvelope { const payloadJson = JSON.stringify(args.payload ?? null); const payloadBytes = Buffer.byteLength(payloadJson, "utf8"); const requestId = typeof args.requestId === "string" && args.requestId.trim().length > 0 @@ -217,34 +285,73 @@ export function encodeSyncEnvelope(args: EncodeEnvelopeArgs): string { : null; const threshold = Math.max(0, Math.floor(args.compressionThresholdBytes ?? DEFAULT_SYNC_COMPRESSION_THRESHOLD_BYTES)); const compressionCodec = args.compressionCodec ?? "gzip"; + const header = { + version: SYNC_PROTOCOL_VERSION, + type: args.type, + ...(projectId ? { projectId } : {}), + requestId, + }; if (compressionCodec !== "none" && payloadBytes >= threshold) { - const compressed = compressionCodec === "deflate" - ? deflateSync(Buffer.from(payloadJson, "utf8")) - : gzipSync(Buffer.from(payloadJson, "utf8")); + const source = Buffer.from(payloadJson, "utf8"); + return { + header, + compressed: { + codec: compressionCodec, + body: compressionCodec === "deflate" ? deflateSync(source) : gzipSync(source), + uncompressedBytes: payloadBytes, + }, + payload: args.payload ?? null, + }; + } + + return { header, compressed: null, payload: args.payload ?? null }; +} + +export function encodeSyncEnvelope(args: EncodeEnvelopeArgs): string { + const prepared = prepareSyncEnvelope(args); + if (prepared.compressed) { return JSON.stringify(asSyncEnvelope({ - version: SYNC_PROTOCOL_VERSION, - type: args.type, - ...(projectId ? { projectId } : {}), - requestId, - compression: compressionCodec, + ...prepared.header, + compression: prepared.compressed.codec, payloadEncoding: "base64", - payload: compressed.toString("base64"), - uncompressedBytes: payloadBytes, + payload: prepared.compressed.body.toString("base64"), + uncompressedBytes: prepared.compressed.uncompressedBytes, })); } - return JSON.stringify(asSyncEnvelope({ - version: SYNC_PROTOCOL_VERSION, - type: args.type, - ...(projectId ? { projectId } : {}), - requestId, + ...prepared.header, compression: "none", payloadEncoding: "json", - payload: args.payload ?? null, + payload: prepared.payload, })); } +/** + * Encode one envelope into a single frame, using the binary container when the + * peer supports it and the payload actually compressed. An uncompressed payload + * stays JSON text: there are no bytes to save, and small frames staying + * human-readable on the wire is worth more than uniformity. + */ +function encodeSyncEnvelopeFrame(args: EncodeEnvelopeArgs & { binaryFrames?: boolean }): SyncWireFrame { + if (!args.binaryFrames) return encodeSyncEnvelope(args); + const prepared = prepareSyncEnvelope(args); + if (!prepared.compressed) { + return JSON.stringify(asSyncEnvelope({ + ...prepared.header, + compression: "none", + payloadEncoding: "json", + payload: prepared.payload, + })); + } + return encodeSyncBinaryFrame({ + ...prepared.header, + compression: prepared.compressed.codec, + payloadEncoding: "binary", + uncompressedBytes: prepared.compressed.uncompressedBytes, + }, prepared.compressed.body); +} + /** * Encode an envelope into one or more websocket frames. When the encoded * envelope exceeds `maxFrameBytes`, it is sliced into `envelope_chunk` frames @@ -253,20 +360,43 @@ export function encodeSyncEnvelope(args: EncodeEnvelopeArgs): string { * chunkedEnvelopes capability — they get the single full frame, same as today. */ export function encodeSyncEnvelopeFrames( - args: EncodeEnvelopeArgs & { maxFrameBytes?: number | null }, -): string[] { - const encoded = encodeSyncEnvelope(args); + args: EncodeEnvelopeArgs & { maxFrameBytes?: number | null; binaryFrames?: boolean }, +): SyncWireFrame[] { + const encoded = encodeSyncEnvelopeFrame(args); const maxFrameBytes = args.maxFrameBytes ?? null; - if (!maxFrameBytes || Buffer.byteLength(encoded, "utf8") <= maxFrameBytes) { + if (!maxFrameBytes || syncFrameByteLength(encoded) <= maxFrameBytes) { return [encoded]; } - const raw = Buffer.from(encoded, "utf8"); + const raw = typeof encoded === "string" ? Buffer.from(encoded, "utf8") : encoded; + const chunkId = randomUUID(); + + // A binary chunk carries its slice raw, so the only overhead is the small + // header; a text chunk still pays base64's 4/3 expansion inside a JSON + // wrapper and has to budget the slice down to compensate. + if (args.binaryFrames) { + const partBytes = Math.max(16 * 1024, maxFrameBytes - 1024); + const total = Math.ceil(raw.byteLength / partBytes); + const frames: SyncWireFrame[] = []; + for (let index = 0; index < total; index += 1) { + frames.push(encodeSyncBinaryFrame({ + version: SYNC_PROTOCOL_VERSION, + type: "envelope_chunk", + requestId: args.requestId ?? null, + compression: "none", + payloadEncoding: "binary", + chunkId, + index, + total, + }, raw.subarray(index * partBytes, Math.min(raw.byteLength, (index + 1) * partBytes)))); + } + return frames; + } + // Each part is base64 (4/3 expansion) inside a small JSON wrapper; budget // the decoded slice so the wrapped chunk frame stays under maxFrameBytes. const partBytes = Math.max(16 * 1024, Math.floor(((maxFrameBytes - 1024) * 3) / 4)); const total = Math.ceil(raw.byteLength / partBytes); - const chunkId = randomUUID(); - const frames: string[] = []; + const frames: SyncWireFrame[] = []; for (let index = 0; index < total; index += 1) { const payload: SyncEnvelopeChunkPayload = { chunkId, @@ -286,6 +416,28 @@ export function encodeSyncEnvelopeFrames( return frames; } +/** + * Chunk metadata carried in a binary `envelope_chunk` header. The body is the + * raw slice, so unlike the text form there is no `part` string to validate. + */ +export function parseSyncBinaryChunkHeader( + header: Record, +): { chunkId: string; index: number; total: number } | null { + const chunkId = typeof header.chunkId === "string" + && header.chunkId.trim() + && Buffer.byteLength(header.chunkId, "utf8") <= MAX_ENVELOPE_CHUNK_ID_BYTES + ? header.chunkId + : null; + const index = typeof header.index === "number" && Number.isInteger(header.index) && header.index >= 0 + ? header.index + : null; + const total = typeof header.total === "number" && Number.isInteger(header.total) && header.total > 0 + ? header.total + : null; + if (chunkId == null || index == null || total == null || index >= total) return null; + return { chunkId, index, total }; +} + export function parseSyncEnvelopeChunkPayload(payload: unknown): SyncEnvelopeChunkPayload | null { if (!payload || typeof payload !== "object" || Array.isArray(payload)) return null; const record = payload as Record; @@ -335,51 +487,70 @@ export function createSyncEnvelopeChunkAssembler(options: { buffers.delete(chunkId); }; + const ingest = ( + meta: { chunkId: string; index: number; total: number }, + decodedPart: Buffer, + ): Buffer | null => { + if (meta.total > maxTotalParts) return null; + if (decodedPart.byteLength > maxEnvelopeBytes) { + remove(meta.chunkId); + return null; + } + let buffer = buffers.get(meta.chunkId); + if (!buffer) { + while (buffers.size >= maxConcurrentChunks) { + const oldest = buffers.keys().next().value; + if (oldest == null) break; + remove(oldest); + } + const timeout = setTimeout(() => remove(meta.chunkId), timeoutMs); + timeout.unref?.(); + buffer = { total: meta.total, parts: new Map(), bytes: 0, timeout }; + buffers.set(meta.chunkId, buffer); + } + if (buffer.total !== meta.total) { + remove(meta.chunkId); + return null; + } + const previous = buffer.parts.get(meta.index); + const nextBytes = buffer.bytes - (previous?.byteLength ?? 0) + decodedPart.byteLength; + const nextBufferedBytes = bufferedBytes + - (previous?.byteLength ?? 0) + + decodedPart.byteLength; + if (nextBytes > maxEnvelopeBytes || nextBufferedBytes > maxBufferedBytes) { + remove(meta.chunkId); + return null; + } + buffer.parts.set(meta.index, decodedPart); + buffer.bytes = nextBytes; + bufferedBytes = nextBufferedBytes; + if (buffer.parts.size < buffer.total) return null; + remove(meta.chunkId); + const segments: Buffer[] = []; + for (let index = 0; index < buffer.total; index += 1) { + const part = buffer.parts.get(index); + if (part == null) return null; + segments.push(part); + } + return Buffer.concat(segments); + }; + return { add(payload: SyncEnvelopeChunkPayload): string | null { - if (payload.total > maxTotalParts) return null; const decodedPart = decodeStrictBase64(payload.part); - if (!decodedPart || decodedPart.byteLength > maxEnvelopeBytes) { - remove(payload.chunkId); - return null; - } - let buffer = buffers.get(payload.chunkId); - if (!buffer) { - while (buffers.size >= maxConcurrentChunks) { - const oldest = buffers.keys().next().value; - if (oldest == null) break; - remove(oldest); - } - const timeout = setTimeout(() => remove(payload.chunkId), timeoutMs); - timeout.unref?.(); - buffer = { total: payload.total, parts: new Map(), bytes: 0, timeout }; - buffers.set(payload.chunkId, buffer); - } - if (buffer.total !== payload.total) { + if (!decodedPart) { remove(payload.chunkId); return null; } - const previous = buffer.parts.get(payload.index); - const nextBytes = buffer.bytes - (previous?.byteLength ?? 0) + decodedPart.byteLength; - const nextBufferedBytes = bufferedBytes - - (previous?.byteLength ?? 0) - + decodedPart.byteLength; - if (nextBytes > maxEnvelopeBytes || nextBufferedBytes > maxBufferedBytes) { - remove(payload.chunkId); - return null; - } - buffer.parts.set(payload.index, decodedPart); - buffer.bytes = nextBytes; - bufferedBytes = nextBufferedBytes; - if (buffer.parts.size < buffer.total) return null; - remove(payload.chunkId); - const segments: Buffer[] = []; - for (let index = 0; index < buffer.total; index += 1) { - const part = buffer.parts.get(index); - if (part == null) return null; - segments.push(part); - } - return Buffer.concat(segments).toString("utf8"); + return ingest(payload, decodedPart)?.toString("utf8") ?? null; + }, + /** + * Binary chunks carry their slice raw, and reassemble into the binary + * envelope frame itself — returning a Buffer, not text, because utf8 + * decoding those bytes would corrupt them. + */ + addBinary(meta: { chunkId: string; index: number; total: number }, body: Buffer): Buffer | null { + return ingest(meta, body); }, reset(): void { for (const chunkId of buffers.keys()) remove(chunkId); @@ -390,6 +561,109 @@ export function createSyncEnvelopeChunkAssembler(options: { }; } +function inflateSyncEnvelopeBody( + compressed: Buffer, + codec: "gzip" | "deflate", + declaredBytes: number | undefined, + requestId: string | null, + projectId: string | null, +): Buffer { + if (typeof declaredBytes === "number" && declaredBytes > MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES) { + throw new Error(`Decoded sync envelope exceeds ${MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES} bytes.`); + } + let uncompressed: Buffer; + try { + uncompressed = codec === "deflate" + ? inflateSync(compressed, { maxOutputLength: MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES }) + : gunzipSync(compressed, { maxOutputLength: MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES }); + } catch (error) { + throw new Error(`Failed to decode ${codec} sync envelope${requestId ? ` ${requestId}` : ""}${projectId ? ` for project ${projectId}` : ""}: ${error instanceof Error ? error.message : String(error)}`); + } + if (uncompressed.byteLength > MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES) { + throw new Error(`Decoded sync envelope exceeds ${MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES} bytes.`); + } + if (typeof declaredBytes === "number" && declaredBytes !== uncompressed.byteLength) { + throw new Error("Decoded sync envelope size does not match declared uncompressedBytes."); + } + return uncompressed; +} + +/** + * Parse a frame that may be either JSON text or a binary envelope container. + * Anything without the binary magic takes the text path unchanged, so a peer + * that never negotiated binary frames sees byte-identical behavior. + */ +export function parseSyncEnvelopeFrame(raw: unknown): ParsedSyncEnvelope { + // `ws` delivers RawData as a single Buffer by default, but as Buffer[] when a + // socket is in fragments mode. Concatenating first means the magic sniff sees + // the real first four bytes instead of failing the check and sending a binary + // frame down the utf8 text path. + const frame = Array.isArray(raw) ? Buffer.concat(raw as Buffer[]) : raw; + if (isSyncBinaryFrame(frame)) return parseSyncBinaryEnvelope(frame); + return parseSyncEnvelope(wsDataToText(frame)); +} + +export function parseSyncBinaryEnvelope(raw: Buffer): ParsedSyncEnvelope { + const decoded = decodeSyncBinaryFrame(raw); + if (!decoded) throw new Error("Invalid binary sync envelope frame."); + const header = decoded.header; + const receivedVersion = header.version; + const rawRequestId = header.requestId; + const requestId = typeof rawRequestId === "string" && rawRequestId.trim() ? rawRequestId.trim() : null; + if ( + typeof receivedVersion === "number" + && Number.isInteger(receivedVersion) + && !isSyncProtocolVersionSupported(receivedVersion) + ) { + throw new SyncProtocolVersionMismatchError(receivedVersion, requestId); + } + if (!isSyncProtocolVersionSupported(receivedVersion)) { + throw new Error(`Invalid sync protocol version: ${String(receivedVersion ?? "unknown")}`); + } + const projectId = typeof header.projectId === "string" && header.projectId.trim() + ? header.projectId.trim() + : null; + const compression = header.compression; + const type = header.type as SyncEnvelope["type"]; + + if (compression === "none") { + // Only chunk frames ride the binary container uncompressed; their body is + // the raw slice rather than a base64 payload. + const chunk = parseSyncBinaryChunkHeader(header); + if (!chunk) throw new Error("Invalid binary envelope_chunk header."); + return { + version: receivedVersion as SyncProtocolVersion, + type, + projectId, + requestId, + compression: "none", + payload: null, + raw: header as unknown as SyncEnvelope, + binaryChunk: { ...chunk, body: decoded.body }, + }; + } + + if (compression !== "gzip" && compression !== "deflate") { + throw new Error(`Unsupported binary sync envelope compression: ${String(compression ?? "unknown")}`); + } + const uncompressed = inflateSyncEnvelopeBody( + decoded.body, + compression, + typeof header.uncompressedBytes === "number" ? header.uncompressedBytes : undefined, + requestId, + projectId, + ); + return { + version: receivedVersion as SyncProtocolVersion, + type, + projectId, + requestId, + compression, + payload: safeJsonParse(uncompressed.toString("utf8"), null), + raw: header as unknown as SyncEnvelope, + }; +} + export function parseSyncEnvelope(rawText: string): ParsedSyncEnvelope { const decoded = safeJsonParse(rawText, null); if (!decoded || typeof decoded !== "object") { @@ -428,24 +702,13 @@ export function parseSyncEnvelope(rawText: string): ParsedSyncEnvelope { ) { throw new Error(`Decoded sync envelope exceeds ${MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES} bytes.`); } - let uncompressedBuffer: Buffer; - try { - const compressed = Buffer.from(decoded.payload, "base64"); - uncompressedBuffer = decoded.compression === "deflate" - ? inflateSync(compressed, { maxOutputLength: MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES }) - : gunzipSync(compressed, { maxOutputLength: MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES }); - } catch (error) { - throw new Error(`Failed to decode ${decoded.compression} sync envelope${requestId ? ` ${requestId}` : ""}${projectId ? ` for project ${projectId}` : ""}: ${error instanceof Error ? error.message : String(error)}`); - } - if (uncompressedBuffer.byteLength > MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES) { - throw new Error(`Decoded sync envelope exceeds ${MAX_UNCOMPRESSED_SYNC_ENVELOPE_BYTES} bytes.`); - } - if ( - typeof decoded.uncompressedBytes === "number" - && decoded.uncompressedBytes !== uncompressedBuffer.byteLength - ) { - throw new Error("Decoded sync envelope size does not match declared uncompressedBytes."); - } + const uncompressedBuffer = inflateSyncEnvelopeBody( + Buffer.from(decoded.payload, "base64"), + decoded.compression, + decoded.uncompressedBytes, + requestId, + projectId, + ); const uncompressed = uncompressedBuffer.toString("utf8"); return { version: decoded.version, diff --git a/apps/desktop/src/shared/types/sync.ts b/apps/desktop/src/shared/types/sync.ts index 67797639d2..1f0958ede3 100644 --- a/apps/desktop/src/shared/types/sync.ts +++ b/apps/desktop/src/shared/types/sync.ts @@ -126,6 +126,12 @@ export type SyncApplicationCompressionCodec = (typeof SYNC_APPLICATION_COMPRESSION_CODECS)[number]; export const SYNC_APPLICATION_COMPRESSION_THRESHOLD_BYTES = 512; export const SYNC_CHUNKED_ENVELOPES_CAPABILITY = "chunkedEnvelopes"; +/** + * Hello capability a client declares when it can decode compressed envelopes + * delivered as binary websocket frames instead of base64 inside JSON text. + * Peers that do not declare it keep the base64 wire byte for byte. + */ +export const SYNC_BINARY_ENVELOPES_CAPABILITY = "binaryEnvelopes"; export type SyncPayloadEncoding = "json" | "base64"; diff --git a/apps/ios/ADE/Services/SyncService.swift b/apps/ios/ADE/Services/SyncService.swift index 771f15323d..699c043af5 100644 --- a/apps/ios/ADE/Services/SyncService.swift +++ b/apps/ios/ADE/Services/SyncService.swift @@ -1588,6 +1588,66 @@ struct SyncPreprocessedEnvelope { let type: String let requestId: String? let payload: Any + /// Raw slice of a binary `envelope_chunk`. Binary chunks carry their part as + /// the frame body rather than a base64 string in `payload`, so the chunk + /// branch reads it from here instead. + var binaryChunk: (chunkId: String, index: Int, total: Int, part: Data)? + + init(type: String, requestId: String?, payload: Any) { + self.type = type + self.requestId = requestId + self.payload = payload + self.binaryChunk = nil + } + + init(binaryChunk: (chunkId: String, index: Int, total: Int, part: Data), requestId: String?) { + self.type = "envelope_chunk" + self.requestId = requestId + self.payload = NSNull() + self.binaryChunk = binaryChunk + } +} + +/// Binary envelope container — the Swift half of `syncBinaryFrame.ts`. +/// +/// [0..4) magic "ADE1" +/// [4..8) uint32 big-endian header length +/// [8..8+h) header JSON (utf8) — the envelope minus `payload` +/// [8+h..) payload bytes (compressed, codec named by the header) +/// +/// A compressed envelope used to arrive as base64 inside JSON text, re-inflating +/// the compressed bytes by a third. The host sends this container only to peers +/// that declared the `binaryEnvelopes` capability, so an older build keeps the +/// base64 wire byte for byte. +enum SyncBinaryFrame { + static let magic = Data("ADE1".utf8) + static let maxHeaderBytes = 64 * 1024 + private static let bodyOffset = 8 + + static func isBinaryFrame(_ data: Data) -> Bool { + data.count >= bodyOffset && data.prefix(magic.count) == magic + } + + /// Returns nil for anything that is not a well-formed container, so a caller + /// can fall back to the text path instead of failing the connection. + static func decode(_ data: Data) -> (header: [String: Any], body: Data)? { + guard isBinaryFrame(data) else { return nil } + let lengthStart = data.startIndex + magic.count + var headerBytes: UInt32 = 0 + for offset in 0..<4 { + headerBytes = (headerBytes << 8) | UInt32(data[lengthStart + offset]) + } + guard headerBytes <= UInt32(maxHeaderBytes) else { return nil } + let headerStart = data.startIndex + bodyOffset + let bodyStart = headerStart + Int(headerBytes) + guard bodyStart <= data.endIndex else { return nil } + guard let header = try? JSONSerialization.jsonObject( + with: data[headerStart..= minSupportedVersion && version <= currentVersion } +/// Inflate and size-check a compressed envelope body. Shared by the base64 text +/// path and the binary container so the two can never disagree about the caps. +func syncInflateEnvelopeBody( + _ compressed: Data, + codec: String, + declaredBytes: Int?, + maxUncompressedBytes: Int = maxUncompressedSyncEnvelopeBytes +) throws -> Any { + let inflated: Data + if codec == "deflate" { + inflated = try inflateZlib(compressed, maxOutputBytes: maxUncompressedBytes) + } else { + inflated = try gunzip(compressed, maxOutputBytes: maxUncompressedBytes) + } + if let declaredBytes, declaredBytes != inflated.count { + throw NSError( + domain: "ADE", + code: 10, + userInfo: [NSLocalizedDescriptionKey: "Decoded sync envelope size does not match declared uncompressedBytes."] + ) + } + return try JSONSerialization.jsonObject(with: inflated, options: []) +} + func syncDecodeEnvelopePayload( _ envelope: [String: Any], maxUncompressedBytes: Int = maxUncompressedSyncEnvelopeBytes @@ -1687,20 +1771,12 @@ func syncDecodeEnvelopePayload( } else { declaredBytes = nil } - let inflated: Data - if compression == "deflate" { - inflated = try inflateZlib(compressed, maxOutputBytes: maxUncompressedBytes) - } else { - inflated = try gunzip(compressed, maxOutputBytes: maxUncompressedBytes) - } - if let declaredBytes, declaredBytes != inflated.count { - throw NSError( - domain: "ADE", - code: 10, - userInfo: [NSLocalizedDescriptionKey: "Decoded sync envelope size does not match declared uncompressedBytes."] - ) - } - return try JSONSerialization.jsonObject(with: inflated, options: []) + return try syncInflateEnvelopeBody( + compressed, + codec: compression, + declaredBytes: declaredBytes, + maxUncompressedBytes: maxUncompressedBytes + ) case "none": if let payloadEncoding = envelope["payloadEncoding"] as? String, payloadEncoding != "json" { @@ -1750,6 +1826,95 @@ func syncPreprocessIncoming( return SyncPreprocessedEnvelope(type: type, requestId: requestId, payload: payload) } +/// Preprocess a frame that arrived as binary data. Only frames carrying the +/// `ADE1` magic take the binary path; anything else is decoded as utf8 text, +/// which is what every frame did before binary envelopes existed and what some +/// transports still do for text frames. +func syncPreprocessIncomingData( + _ data: Data, + maxUncompressedBytes: Int = maxUncompressedSyncEnvelopeBytes +) throws -> SyncPreprocessedEnvelope? { + guard SyncBinaryFrame.isBinaryFrame(data) else { + return try syncPreprocessIncoming(String(decoding: data, as: UTF8.self), maxUncompressedBytes: maxUncompressedBytes) + } + guard let decoded = SyncBinaryFrame.decode(data) else { + throw NSError( + domain: "ADE", + code: 10, + userInfo: [NSLocalizedDescriptionKey: "Invalid binary sync envelope frame."] + ) + } + let header = decoded.header + guard let version = syncProtocolVersionNumber(header["version"]) else { + throw NSError( + domain: "ADE", + code: 4, + userInfo: [NSLocalizedDescriptionKey: "Invalid sync protocol version."] + ) + } + if !syncProtocolVersionIsSupported(version) { + throw SyncProtocolVersionMismatchError( + receivedVersion: version, + currentVersion: syncProtocolVersion, + minSupportedVersion: syncProtocolMinSupported, + updateTarget: version < syncProtocolMinSupported ? "host" : "client" + ) + } + let requestId = header["requestId"] as? String + + // An uncompressed binary frame is a chunk slice: its body is the raw part. + if (header["compression"] as? String ?? "none") == "none" { + guard let chunkId = header["chunkId"] as? String, + let index = syncProtocolVersionNumber(header["index"]), + let total = syncProtocolVersionNumber(header["total"]) else { + throw NSError( + domain: "ADE", + code: 10, + userInfo: [NSLocalizedDescriptionKey: "Uncompressed binary sync envelopes must be envelope chunks."] + ) + } + return SyncPreprocessedEnvelope( + binaryChunk: (chunkId: chunkId, index: index, total: total, part: decoded.body), + requestId: requestId + ) + } + + let compression = header["compression"] as? String ?? "none" + guard compression == "deflate" || compression == "gzip" else { + throw NSError( + domain: "ADE", + code: 10, + userInfo: [NSLocalizedDescriptionKey: "Unsupported sync envelope compression: \(compression)"] + ) + } + let declaredBytes: Int? + if let rawDeclaredBytes = header["uncompressedBytes"] { + guard let parsed = syncProtocolVersionNumber(rawDeclaredBytes), + parsed >= 0, + parsed <= maxUncompressedBytes else { + throw NSError( + domain: "ADE", + code: 10, + userInfo: [NSLocalizedDescriptionKey: "Invalid compressed sync envelope size."] + ) + } + declaredBytes = parsed + } else { + declaredBytes = nil + } + let payload = try syncInflateEnvelopeBody( + decoded.body, + codec: compression, + declaredBytes: declaredBytes, + maxUncompressedBytes: maxUncompressedBytes + ) + return SyncPreprocessedEnvelope( + type: header["type"] as? String ?? "", + requestId: requestId, + payload: payload + ) +} + func syncEncodeEnvelopeText( type: String, requestId: String?, @@ -2012,14 +2177,6 @@ struct SyncEnvelopeChunkAssembler { part: String, now: TimeInterval = ProcessInfo.processInfo.systemUptime ) -> String? { - guard total > 0, - index >= 0, - index < total, - total <= maxTotalParts, - !chunkId.isEmpty, - chunkId.utf8.count <= maxSyncEnvelopeChunkIdBytes else { - return nil - } let encodedBytes = part.utf8.count let decodedUpperBound = ((encodedBytes + 3) / 4) * 3 guard decodedUpperBound <= maxEnvelopeBytes, @@ -2028,6 +2185,58 @@ struct SyncEnvelopeChunkAssembler { arrivalOrder.removeAll { $0 == chunkId } return nil } + guard let assembled = ingest( + chunkId: chunkId, + index: index, + total: total, + decodedPart: decodedPart, + encodedBytes: encodedBytes, + now: now + ) else { return nil } + return String(data: assembled, encoding: .utf8) + } + + /// Binary chunks carry their slice as the frame body, so they reassemble into + /// the binary envelope itself — `Data`, never text: utf8-decoding those bytes + /// would corrupt them. + mutating func addBinary( + chunkId: String, + index: Int, + total: Int, + part: Data, + now: TimeInterval = ProcessInfo.processInfo.systemUptime + ) -> Data? { + guard part.count <= maxEnvelopeBytes else { + buffers.removeValue(forKey: chunkId) + arrivalOrder.removeAll { $0 == chunkId } + return nil + } + return ingest( + chunkId: chunkId, + index: index, + total: total, + decodedPart: part, + encodedBytes: part.count, + now: now + ) + } + + private mutating func ingest( + chunkId: String, + index: Int, + total: Int, + decodedPart: Data, + encodedBytes: Int, + now: TimeInterval + ) -> Data? { + guard total > 0, + index >= 0, + index < total, + total <= maxTotalParts, + !chunkId.isEmpty, + chunkId.utf8.count <= maxSyncEnvelopeChunkIdBytes else { + return nil + } if buffers[chunkId] == nil { while arrivalOrder.count >= maxConcurrentChunks, let oldest = arrivalOrder.first { arrivalOrder.removeFirst() @@ -2073,7 +2282,7 @@ struct SyncEnvelopeChunkAssembler { guard let part = buffer.parts[partIndex] else { return nil } data.append(part.data) } - return String(data: data, encoding: .utf8) + return data } mutating func reset() { @@ -15289,7 +15498,7 @@ final class SyncService: ObservableObject { "deviceType": "phone", "siteId": database.localSiteId(), "dbVersion": latestRemoteDbVersion, - "capabilities": ["changesetAck", "chunkedEnvelopes", "relayReauthorizeV1"], + "capabilities": ["changesetAck", "chunkedEnvelopes", "relayReauthorizeV1", "binaryEnvelopes"], ] if let appVersion = (info["CFBundleShortVersionString"] as? String)? .trimmingCharacters(in: .whitespacesAndNewlines), @@ -16744,24 +16953,35 @@ final class SyncService: ObservableObject { while self.socket === task { do { let message = try await task.receive() + // A compressed envelope arrives as a binary frame; everything else is + // text. Binary frames never carry relay control JSON, so the control + // sniff (and the utf8 decode it needs) is skipped for them. + var binaryFrame: Data? let text: String switch message { case .string(let value): text = value case .data(let data): - text = String(decoding: data, as: UTF8.self) + if SyncBinaryFrame.isBinaryFrame(data) { + binaryFrame = data + text = "" + } else { + text = String(decoding: data, as: UTF8.self) + } @unknown default: text = "" } - if self.handleRelayTransportControlFrame(text, task: task) { + if binaryFrame == nil, self.handleRelayTransportControlFrame(text, task: task) { continue } do { // CPU-heavy decode (envelope JSON, gunzip, payload JSON) runs off // the main actor; ordering is preserved because each frame is // awaited in sequence. The main actor only mutates state. + let frame = binaryFrame let preprocessed = try await Task.detached(priority: .userInitiated) { - try syncPreprocessIncoming(text) + if let frame { return try syncPreprocessIncomingData(frame) } + return try syncPreprocessIncoming(text) }.value // The detached decode is a suspension point: the socket can be // torn down (and a new connection brought up) while it runs. A @@ -16848,23 +17068,35 @@ final class SyncService: ObservableObject { switch type { case "envelope_chunk": - guard let dict = payload as? [String: Any], - let chunkId = dict["chunkId"] as? String, - let index = dict["index"] as? Int, - let total = dict["total"] as? Int, - let part = dict["part"] as? String else { return } - let reassembled = envelopeChunkAssembler.add( - chunkId: chunkId, - index: index, - total: total, - part: part - ) + // Binary chunks reassemble into binary envelope bytes; text chunks into + // envelope JSON. Both reuse one assembler and one nested-decode path. + let reassembled: Data? + if let binaryChunk = pre.binaryChunk { + reassembled = envelopeChunkAssembler.addBinary( + chunkId: binaryChunk.chunkId, + index: binaryChunk.index, + total: binaryChunk.total, + part: binaryChunk.part + ) + } else { + guard let dict = payload as? [String: Any], + let chunkId = dict["chunkId"] as? String, + let index = dict["index"] as? Int, + let total = dict["total"] as? Int, + let part = dict["part"] as? String else { return } + reassembled = envelopeChunkAssembler.add( + chunkId: chunkId, + index: index, + total: total, + part: part + ).map { Data($0.utf8) } + } scheduleEnvelopeChunkExpiry() if let reassembled { // The reassembled envelope can be tens of megabytes — decode it off // the main actor like any first-class frame. let nested = try await Task.detached(priority: .userInitiated) { - try syncPreprocessIncoming(reassembled) + try syncPreprocessIncomingData(reassembled) }.value guard isCurrentConnectionGeneration(generation) else { return } if let nested {