From cbe1c96ce41e4f82d24b0331603a6a384b633598 Mon Sep 17 00:00:00 2001 From: Steven Chong <25894545+teamchong@users.noreply.github.com> Date: Sun, 8 Nov 2026 23:37:25 -0500 Subject: [PATCH 1/2] Add opt-in 8-byte-aligned binary framing for WebSocket and HTTP batch RPC --- __tests__/index.test.ts | 427 +++++++++++++++++++++++++++++++++++++++- src/batch.ts | 293 ++++++++++++++++++++------- src/rpc.ts | 7 + src/serialize.ts | 337 +++++++++++++++++++++++++++++-- src/websocket.ts | 110 +++++++++-- 5 files changed, 1074 insertions(+), 100 deletions(-) diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index 48008efd..7c57faba 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -6,8 +6,17 @@ import { expect, it, describe, inject } from "vitest" import { deserialize, serialize, RpcSession, type RpcSessionOptions, RpcTransport, type RpcTransportWithCustomEncoding, RpcTarget, RpcStub, RpcPromise, newWebSocketRpcSession, newMessagePortRpcSession, - newHttpBatchRpcSession} from "../src/index.js" -import { swapByteOrder } from "../src/serialize.js" + newHttpBatchRpcSession, + newHttpBatchRpcResponse} from "../src/index.js" +import { BINARY_CONTENT_TYPE, + decodeBinaryFrame, + encodeBinaryFrame, + Evaluator, + Devaluator, + hasBinaryFrameMagic, + NATIVE_LITTLE_ENDIAN, + stringifyWithBase64Bytes, + swapByteOrder } from "../src/serialize.js" import { MAX_CLOSE_REASON_BYTES } from "../src/websocket.js" import { ErrorStubHook, PayloadStubHook, PromiseStubHook, RpcPayload, RpcStub as RawRpcStub, StubHook, streamImpl, unwrapStubTakingOwnership, type PropertyPath } from "../src/core.js" @@ -4520,3 +4529,417 @@ describe("settled import references", () => { expect(harness.server.getStats()).toStrictEqual({imports: 1, exports: 1}); }); }); + +class BinaryTestService extends RpcTarget { + echo(value: unknown): unknown { + return value; + } + + combineAttachments( + bytes: Uint8Array, + floats: Float32Array, + ints: Int16Array + ): { + bytes: Uint8Array; + floats: Float32Array; + ints: Int16Array; + sum: number; + } { + let sum = 0; + for (let b of bytes) sum += b; + for (let f of floats) sum += f; + for (let i of ints) sum += i; + return { + bytes: new Uint8Array(bytes.map(x => x ^ 0xff)), + floats: new Float32Array(floats.map(x => x * 2)), + ints: new Int16Array(ints.map(x => x + 10)), + sum, + }; + } + + makeCounter(i: number): Counter { + return new Counter(i); + } + + incrementWithBytes(c: RpcStub, payload: Uint8Array): Promise<{ count: number; size: number }> { + return c.increment(payload.byteLength).then(count => ({ count, size: payload.byteLength })); + } +} + +describe("binary framing codec (jsonCompatibleWithBytes)", () => { + it("packs multiple attachments at 8-byte-aligned offsets and provides zero-copy views on decode", () => { + let u8Odd = new Uint8Array([1, 2, 3]); // 3 bytes -> padded to 8 + let u8Empty = new Uint8Array(0); // 0 bytes -> padded to 0 + let f64 = new Float64Array([1.5, -2.25, 3.75]); // 24 bytes (requires 8-byte alignment) + let f32 = new Float32Array([0.5, -1.5, 2.5]); // 12 bytes -> padded to 16 + let i32 = new Int32Array([-100, 200, -300, 400]); // 16 bytes + let i16 = new Int16Array([-10, 20, -30]); // 6 bytes -> padded to 8 + let bi64 = new BigInt64Array([1234567890123456789n, -987654321098765432n]); // 16 bytes + + let original = { + u8Odd, + u8Empty, + f64, + f32, + i32, + i16, + bi64, + literalBytesTuple: ["bytes", 10, 20, 30], + }; + + let devalued = Devaluator.devaluate(original, undefined, undefined, undefined, "jsonCompatibleWithBytes"); + let frame = encodeBinaryFrame(devalued); + expect(hasBinaryFrameMagic(frame)).toBe(true); + expect(frame.byteLength % 8).toBe(0); + + let decodedTree = decodeBinaryFrame(frame.buffer) as Record; + + // Every attachment Uint8Array in the decoded tree must be 8-byte aligned in the frame buffer. + for (let key of ["u8Odd", "u8Empty", "f64", "f32", "i32", "i16", "bi64"]) { + let rawBytes = decodedTree[key][1] as Uint8Array; + expect(rawBytes).toBeInstanceOf(Uint8Array); + expect(rawBytes.buffer).toBe(frame.buffer); + expect(rawBytes.byteOffset % 8).toBe(0); + } + + let payload = new Evaluator( + { + importStub() { throw new Error("unexpected"); }, + importPromise() { throw new Error("unexpected"); }, + getExport() { return undefined; }, + getPipeReadable() { throw new Error("unexpected"); }, + getLimits() { return { maxBigIntDigits: 16384, maxDepth: 256, maxMessageSize: 32 * 1024 * 1024 }; }, + }, + "jsonCompatibleWithBytes" + ).evaluate(decodedTree); + + let evaluated = payload.value as typeof original; + expect(evaluated.u8Odd).toStrictEqual(u8Odd); + expect(evaluated.u8Empty).toStrictEqual(u8Empty); + expect(evaluated.f64).toStrictEqual(f64); + expect(evaluated.f32).toStrictEqual(f32); + expect(evaluated.i32).toStrictEqual(i32); + expect(evaluated.i16).toStrictEqual(i16); + expect(evaluated.bi64).toStrictEqual(bi64); + expect(evaluated.literalBytesTuple).toStrictEqual(["bytes", 10, 20, 30]); + + if (NATIVE_LITTLE_ENDIAN) { + // TypedArrays and Uint8Array views are zero-copy views directly into the frame's ArrayBuffer! + expect(evaluated.u8Odd.buffer).toBe(frame.buffer); + expect(evaluated.f64.buffer).toBe(frame.buffer); + expect(evaluated.f32.buffer).toBe(frame.buffer); + expect(evaluated.i32.buffer).toBe(frame.buffer); + expect(evaluated.i16.buffer).toBe(frame.buffer); + expect(evaluated.bi64.buffer).toBe(frame.buffer); + } + + payload.dispose(); + }); + + it("stringifies devalued jsonCompatibleWithBytes trees to base64 for text fallback", () => { + let devalued = Devaluator.devaluate( + { a: new Uint8Array([1, 2, 3]), b: new Int32Array([0x01020304]) }, + undefined, + undefined, + undefined, + "jsonCompatibleWithBytes" + ); + let json = stringifyWithBase64Bytes(devalued); + let parsed = JSON.parse(json) as unknown; + expect(parsed).toStrictEqual({ + a: ["bytes", "AQID"], + b: ["bytes", "BAMCAQ", "Int32Array"], + }); + }); + + it("rejects malformed binary frames", () => { + expect(() => decodeBinaryFrame(new Uint8Array([1, 2, 3]))).toThrow(TypeError); + expect(() => decodeBinaryFrame(new Uint8Array([0, 0, 0, 0, 0, 0, 0, 0]))).toThrow(/Invalid binary RPC frame magic/); + + let validFrame = encodeBinaryFrame(["push", ["bytes", new Uint8Array([1, 2, 3, 4])]]); + let truncatedHeader = validFrame.slice(0, 10); + expect(() => decodeBinaryFrame(truncatedHeader)).toThrow(/header length exceeds frame size/); + + let truncatedAttachment = validFrame.slice(0, validFrame.byteLength - 1); + expect(() => decodeBinaryFrame(truncatedAttachment)).toThrow(/exceeds frame bounds/); + + // Craft a frame with an unaligned attachment offset (offset = 4). + let badJson = new TextEncoder().encode(JSON.stringify(["bytes", 0, 4, 4])); + let badFrame = new Uint8Array(8 + 16 + 8); + badFrame.set([0x89, 0x43, 0x57, 0x42], 0); + new DataView(badFrame.buffer).setUint32(4, badJson.byteLength, true); + badFrame.set(badJson, 8); + expect(() => decodeBinaryFrame(badFrame)).toThrow(/Invalid binary attachment offset/); + + // Reject trailing unreferenced bytes at the end of a frame. + let paddedWithExtra = new Uint8Array(validFrame.byteLength + 8); + paddedWithExtra.set(validFrame, 0); + expect(() => decodeBinaryFrame(paddedWithExtra)).toThrow(/does not match header and attachments/); + + // Enforce maxDepth during binary attachment rehydration. + let deepFrame = encodeBinaryFrame([[[[[["bytes", new Uint8Array([1])]]]]]]); + expect(() => decodeBinaryFrame(deepFrame, 2)).toThrow(/exceeded maximum allowed depth/); + + // Preserve own "__proto__" properties without polluting Object.prototype. + let protoJson = new TextEncoder().encode('{"__proto__":{"polluted":123}}'); + let protoFrame = new Uint8Array(8 + protoJson.byteLength); + protoFrame.set([0x89, 0x43, 0x57, 0x42], 0); + new DataView(protoFrame.buffer).setUint32(4, protoJson.byteLength, true); + protoFrame.set(protoJson, 8); + let protoDecoded = decodeBinaryFrame(protoFrame) as Record; + expect(({} as any).polluted).toBeUndefined(); + expect(Object.prototype.hasOwnProperty.call(protoDecoded, "__proto__")).toBe(true); + }); +}); + +describe("HTTP batch binary framing", () => { + it("round-trips typed arrays, multiple attachments, and pipelined calls in binary mode", async () => { + let serverTarget = new BinaryTestService(); + let seenContentTypes: (string | null)[] = []; + let seenResponseTypes: (string | null)[] = []; + let seenRequestMagic: boolean[] = []; + + let originalFetch = globalThis.fetch; + globalThis.fetch = async (input: RequestInfo | URL, init?: RequestInit) => { + let req = new Request(input, init); + seenContentTypes.push(req.headers.get("Content-Type")); + let cloneBuf = await req.clone().arrayBuffer(); + seenRequestMagic.push(hasBinaryFrameMagic(cloneBuf)); + let resp = await newHttpBatchRpcResponse(req, serverTarget); + seenResponseTypes.push(resp.headers.get("Content-Type")); + return resp; + }; + + try { + let stub = newHttpBatchRpcSession("https://example.com/rpc", { binary: true }); + + let u8 = new Uint8Array([10, 20, 30, 40, 50]); + let f32 = new Float32Array([1.25, 2.5, 3.75]); + let i16 = new Int16Array([-5, 15, -25]); + + let combinedPromise = stub.combineAttachments(u8, f32, i16); + let counter = stub.makeCounter(100); + let pipelinedPromise = stub.incrementWithBytes(counter, new Uint8Array(7)); + + let [combined, pipelined] = await Promise.all([combinedPromise, pipelinedPromise]); + + expect(seenContentTypes).toStrictEqual([BINARY_CONTENT_TYPE]); + expect(seenRequestMagic).toStrictEqual([true]); + expect(seenResponseTypes).toStrictEqual([BINARY_CONTENT_TYPE]); + + expect(combined.bytes).toStrictEqual(new Uint8Array(u8.map(x => x ^ 0xff))); + expect(combined.floats).toStrictEqual(new Float32Array([2.5, 5.0, 7.5])); + expect(combined.ints).toStrictEqual(new Int16Array([5, 25, -15])); + expect(combined.sum).toBe(10 + 20 + 30 + 40 + 50 + 1.25 + 2.5 + 3.75 - 5 + 15 - 25); + expect(pipelined).toStrictEqual({ count: 107, size: 7 }); + } finally { + globalThis.fetch = originalFetch; + } + }); + + it("supports mixed text/binary negotiation and preserves text default for HTTP batch", async () => { + let serverTarget = new BinaryTestService(); + let lastReqContentType: string | null = null; + let lastRespContentType: string | null = null; + let serverOptions: RpcSessionOptions | undefined; + + let originalFetch = globalThis.fetch; + globalThis.fetch = async (input: RequestInfo | URL, init?: RequestInit) => { + let req = new Request(input, init); + lastReqContentType = req.headers.get("Content-Type"); + let resp = await newHttpBatchRpcResponse(req, serverTarget, serverOptions); + lastRespContentType = resp.headers.get("Content-Type"); + return resp; + }; + + try { + // 1. Default client -> default server uses text framing. + { + serverOptions = undefined; + let stub = newHttpBatchRpcSession("https://example.com/rpc"); + let echoed = (await stub.echo(new Uint8Array([1, 2, 3]))) as Uint8Array; + expect(new Uint8Array(echoed)).toStrictEqual(new Uint8Array([1, 2, 3])); + expect(lastReqContentType).not.toBe(BINARY_CONTENT_TYPE); + expect(lastRespContentType).not.toBe(BINARY_CONTENT_TYPE); + } + + // 2. Text client -> binary-configured server: request is text, response is binary, client decodes it. + { + serverOptions = { binary: true }; + let stub = newHttpBatchRpcSession("https://example.com/rpc"); + let echoed = await stub.echo(new Float32Array([42.5, -10.25])); + expect(echoed).toStrictEqual(new Float32Array([42.5, -10.25])); + expect(lastReqContentType).not.toBe(BINARY_CONTENT_TYPE); + expect(lastRespContentType).toBe(BINARY_CONTENT_TYPE); + } + + // 3. Request with Accept: application/x-capnweb-bin -> default server replies in binary. + { + serverOptions = undefined; + let req = new Request("https://example.com/rpc", { + headers: { Accept: BINARY_CONTENT_TYPE }, + }); + let stub = newHttpBatchRpcSession(req); + let echoed = await stub.echo(new Int16Array([11, 22, 33])); + expect(echoed).toStrictEqual(new Int16Array([11, 22, 33])); + expect(lastReqContentType).not.toBe(BINARY_CONTENT_TYPE); + expect(lastRespContentType).toBe(BINARY_CONTENT_TYPE); + } + } finally { + globalThis.fetch = originalFetch; + } + }); + + it("performs a binary HTTP batch request against the Node HTTP test server", async () => { + let cap = newHttpBatchRpcSession(`http://${inject("testServerHost")}`, { + binary: true, + }); + let promise1 = cap.square(6); + let counter = cap.makeCounter(2); + let promise2 = counter.increment(3); + let promise3 = cap.incrementCounter(counter, 4); + + await expect(Promise.all([promise1, promise2, promise3])).resolves.toStrictEqual([36, 5, 9]); + }); + + it("aborts the RPC session and disposes the main target when HTTP batch receives a malformed frame", async () => { + let disposed = false; + class DisposableService extends BinaryTestService { + [Symbol.dispose]() { + disposed = true; + } + } + let badReq = new Request("https://example.com/rpc", { + method: "POST", + headers: { "Content-Type": BINARY_CONTENT_TYPE }, + body: new Uint8Array([0x89, 0x43, 0x57, 0x42, 0xff, 0xff, 0xff, 0x7f]), + }); + await expect( + newHttpBatchRpcResponse(badReq, new DisposableService()), + ).rejects.toThrow(/header length exceeds frame size/); + expect(disposed).toBe(true); + }); +}); + +describe("WebSocket binary framing", () => { + it("round-trips binary frames over the live WebSocket server with auto-detection", async () => { + let url = `ws://${inject("testServerHost")}`; + using cap = newWebSocketRpcSession(url, undefined, { binary: true }); + + expect(await cap.square(5)).toBe(25); + let counter = cap.makeCounter(2); + expect(await counter.increment(3)).toBe(5); + expect(await cap.incrementCounter(counter, 4)).toBe(9); + + let blob = new Blob([new Uint8Array([1, 3, 5, 7, 9])], { type: "application/octet-stream" }); + let echoed = await cap.echoBlob(blob); + expect(new Uint8Array(await echoed.arrayBuffer())).toStrictEqual(new Uint8Array([1, 3, 5, 7, 9])); + }); + + it("supports binary mode, auto-upgrading, and mixed text/binary peers over fake WebSockets", async () => { + type WsEventCallback = (ev: { data?: unknown; code?: number; reason?: string }) => void; + + function createLinkedWebSockets() { + let clientListeners: Record = {}; + let serverListeners: Record = {}; + let clientSentFrames: (string | ArrayBuffer)[] = []; + let serverSentFrames: (string | ArrayBuffer)[] = []; + + let clientWs = { + readyState: 1, + binaryType: "blob", + addEventListener(type: string, cb: WsEventCallback) { + (clientListeners[type] ||= []).push(cb); + }, + send(data: string | ArrayBuffer | Uint8Array) { + let payload = data instanceof Uint8Array + ? data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength) as ArrayBuffer + : data; + clientSentFrames.push(payload); + queueMicrotask(() => { + for (let cb of serverListeners["message"] || []) { + cb({ data: payload }); + } + }); + }, + close(code?: number, reason?: string) { + queueMicrotask(() => { + for (let cb of serverListeners["close"] || []) { + cb({ code: code ?? 1000, reason: reason ?? "" }); + } + }); + }, + } as unknown as WebSocket; + + let serverWs = { + readyState: 1, + binaryType: "blob", + addEventListener(type: string, cb: WsEventCallback) { + (serverListeners[type] ||= []).push(cb); + }, + send(data: string | ArrayBuffer | Uint8Array) { + let payload = data instanceof Uint8Array + ? data.buffer.slice(data.byteOffset, data.byteOffset + data.byteLength) as ArrayBuffer + : data; + serverSentFrames.push(payload); + queueMicrotask(() => { + for (let cb of clientListeners["message"] || []) { + cb({ data: payload }); + } + }); + }, + close(code?: number, reason?: string) { + queueMicrotask(() => { + for (let cb of clientListeners["close"] || []) { + cb({ code: code ?? 1000, reason: reason ?? "" }); + } + }); + }, + } as unknown as WebSocket; + + return { clientWs, serverWs, clientSentFrames, serverSentFrames }; + } + + // Case 1: Binary client -> default (auto-detect) server. + // Both client frames and server replies must be ArrayBuffers! + { + let { clientWs, serverWs, clientSentFrames, serverSentFrames } = createLinkedWebSockets(); + using _serverStub = newWebSocketRpcSession(serverWs, new BinaryTestService()); + using clientStub = newWebSocketRpcSession(clientWs, undefined, { binary: true }); + + let res = await clientStub.combineAttachments( + new Uint8Array([1, 2, 3]), + new Float32Array([10.5, 20.5]), + new Int16Array([100, 200]) + ); + expect(res.bytes).toStrictEqual(new Uint8Array([254, 253, 252])); + expect(res.floats).toStrictEqual(new Float32Array([21, 41])); + expect(res.ints).toStrictEqual(new Int16Array([110, 210])); + expect(res.sum).toBe(1 + 2 + 3 + 10.5 + 20.5 + 100 + 200); + expect(clientSentFrames.length).toBeGreaterThan(0); + expect(serverSentFrames.length).toBeGreaterThan(0); + expect(clientSentFrames.every(f => f instanceof ArrayBuffer)).toBe(true); + expect(serverSentFrames.every(f => f instanceof ArrayBuffer)).toBe(true); + } + + // Case 2: Explicit text client ({ binary: false }) -> explicit binary server ({ binary: true }). + // Client always sends strings, server always sends ArrayBuffers, both sides interoperate. + { + let { clientWs, serverWs, clientSentFrames, serverSentFrames } = createLinkedWebSockets(); + using _serverStub = newWebSocketRpcSession(serverWs, new BinaryTestService(), { binary: true }); + using clientStub = newWebSocketRpcSession(clientWs, undefined, { binary: false }); + + let echoed = await clientStub.echo({ + f32: new Float32Array([1.5, -2.5]), + u8: new Uint8Array([9, 8, 7]), + }); + expect(echoed).toStrictEqual({ + f32: new Float32Array([1.5, -2.5]), + u8: new Uint8Array([9, 8, 7]), + }); + + expect(clientSentFrames.every(f => typeof f === "string")).toBe(true); + expect(serverSentFrames.every(f => f instanceof ArrayBuffer)).toBe(true); + } + }); +}); diff --git a/src/batch.ts b/src/batch.ts index 1a8b3ddb..f517b54a 100644 --- a/src/batch.ts +++ b/src/batch.ts @@ -3,10 +3,21 @@ // https://opensource.org/license/mit import { RpcStub } from "./core.js"; -import { RpcTransport, RpcSession, RpcSessionOptions } from "./rpc.js"; +import { RpcTransportWithCustomEncoding, RpcSession, RpcSessionOptions } from "./rpc.js"; +import { + BINARY_CONTENT_TYPE, + DEFAULT_LIMITS, + decodeBinaryFrame, + encodeBinaryFrame, + hasBinaryFrameMagic, + stringifyWithBase64Bytes, +} from "./serialize.js"; import type { IncomingMessage, ServerResponse, OutgoingHttpHeader, OutgoingHttpHeaders } from "node:http"; -type SendBatchFunc = (batch: string[]) => Promise; +type RawBatch = + | { isBinary: true; buffer: ArrayBuffer } + | { isBinary: false; body: string }; +type SendBatchFunc = (batch: unknown[]) => Promise; // Yield a full macrotask; prefer setImmediate because Node/Bun clamp setTimeout(0) to 1ms. const yieldToMacrotask: () => Promise = @@ -14,18 +25,48 @@ const yieldToMacrotask: () => Promise = ? () => new Promise(resolve => setImmediate(resolve)) : () => new Promise(resolve => setTimeout(resolve, 0)); -class BatchClientTransport implements RpcTransport { - constructor(sendBatch: SendBatchFunc) { +function hasBinaryMimeType(headerValue: string | null | undefined): boolean { + return typeof headerValue === "string" && + headerValue.toLowerCase().includes(BINARY_CONTENT_TYPE); +} + +function decodeBinaryBatch( + buffer: ArrayBuffer, + maxMessageSize: number, + maxDepth: number +): unknown[] { + if (buffer.byteLength === 0) return []; + if (buffer.byteLength > maxMessageSize) { + throw new TypeError( + `Incoming message exceeds maximum size of ${maxMessageSize} bytes.` + ); + } + let decoded = decodeBinaryFrame(buffer, maxDepth * 2 + 2); + if (!Array.isArray(decoded)) { + throw new TypeError("Binary HTTP batch frame must decode to an array of messages."); + } + return decoded; +} + +class BatchClientTransport implements RpcTransportWithCustomEncoding { + readonly encodingLevel = "jsonCompatibleWithBytes" as const; + + constructor(sendBatch: SendBatchFunc, options?: RpcSessionOptions) { + this.#maxMessageSize = options?.limits?.maxMessageSize ?? DEFAULT_LIMITS.maxMessageSize; + this.#maxDepth = options?.limits?.maxDepth ?? DEFAULT_LIMITS.maxDepth; this.#promise = this.#scheduleBatch(sendBatch); } + #maxMessageSize: number; + #maxDepth: number; #promise: Promise; - #aborted: any; - - #batchToSend: string[] | null = []; - #batchToReceive: string[] | null = null; + #aborted: unknown; + #batchToSend: unknown[] | null = []; + #rawBatch: RawBatch | null = null; + #batchToReceive?: unknown[]; + #linesToReceive?: string[]; - send(message: string): void { + send(message: unknown): void { // If the batch was already sent, we just ignore the message, because throwing may cause the // RPC system to abort prematurely. Once the last receive() is done then we'll throw an error // that aborts the RPC system at the right time and will propagate to all other requests. @@ -34,21 +75,44 @@ class BatchClientTransport implements RpcTransport { } } - async receive(): Promise { - if (!this.#batchToReceive) { + async receive(): Promise { + if (!this.#rawBatch) { await this.#promise; } + if (this.#batchToReceive === undefined && this.#linesToReceive === undefined) { + if (this.#rawBatch!.isBinary) { + this.#batchToReceive = decodeBinaryBatch( + this.#rawBatch!.buffer, + this.#maxMessageSize, + this.#maxDepth + ); + } else { + let body = this.#rawBatch!.body; + this.#linesToReceive = body === "" ? [] : body.split("\n"); + } + } - let msg = this.#batchToReceive!.shift(); - if (msg !== undefined) { - return msg; - } else { - // No more messages. An error thrown here will propagate out of any calls that are still - // open. - throw new Error("Batch RPC request ended."); + if (this.#linesToReceive !== undefined) { + let line = this.#linesToReceive.shift(); + if (line !== undefined) { + if (line.length > this.#maxMessageSize) { + throw new TypeError( + `Incoming message exceeds maximum size of ${this.#maxMessageSize} UTF-16 code units.` + ); + } + return JSON.parse(line); + } + } else if (this.#batchToReceive !== undefined) { + let msg = this.#batchToReceive.shift(); + if (msg !== undefined) { + return msg; + } } - } + // No more messages. An error thrown here will propagate out of any calls that are still + // open. + throw new Error("Batch RPC request ended."); + } abort?(reason: any): void { this.#aborted = reason; } @@ -69,66 +133,129 @@ class BatchClientTransport implements RpcTransport { let batch = this.#batchToSend!; this.#batchToSend = null; - this.#batchToReceive = await sendBatch(batch); + this.#rawBatch = await sendBatch(batch); } } export function newHttpBatchRpcSession( - urlOrRequest: string | Request, options?: RpcSessionOptions): RpcStub { - let sendBatch: SendBatchFunc = async (batch: string[]) => { - let response = await fetch(urlOrRequest, { - method: "POST", - body: batch.join("\n"), - }); + urlOrRequest: string | Request, + options?: RpcSessionOptions +): RpcStub { + let reqContentType = urlOrRequest instanceof Request + ? urlOrRequest.headers.get("Content-Type") + : null; + let sendBinary = options?.binary ?? hasBinaryMimeType(reqContentType); + let sendBatch: SendBatchFunc = async (batch: unknown[]) => { + let response: Response; + if (sendBinary) { + let headers = new Headers(urlOrRequest instanceof Request ? urlOrRequest.headers : undefined); + if (!headers.has("Content-Type")) { + headers.set("Content-Type", BINARY_CONTENT_TYPE); + } + if (!headers.has("Accept")) { + headers.set("Accept", `${BINARY_CONTENT_TYPE}, text/plain;q=0.9, */*;q=0.8`); + } + let frame = encodeBinaryFrame(batch); + response = await fetch(urlOrRequest, { + method: "POST", + headers, + body: frame as Uint8Array, + }); + } else { + response = await fetch(urlOrRequest, { + method: "POST", + body: batch.map(msg => stringifyWithBase64Bytes(msg)).join("\n"), + }); + } if (!response.ok) { response.body?.cancel(); throw new Error(`RPC request failed: ${response.status} ${response.statusText}`); } - let body = await response.text(); - return body == "" ? [] : body.split("\n"); + let responseContentType = response.headers.get("Content-Type"); + let buffer = await response.arrayBuffer(); + if (hasBinaryMimeType(responseContentType) || hasBinaryFrameMagic(buffer)) { + return { isBinary: true, buffer }; + } + return { isBinary: false, body: new TextDecoder().decode(buffer) }; }; - let transport = new BatchClientTransport(sendBatch); + let transport = new BatchClientTransport(sendBatch, options); let rpc = new RpcSession(transport, undefined, options); return rpc.getRemoteMain(); } -class BatchServerTransport implements RpcTransport { - constructor(batch: string[]) { - this.#batchToReceive = batch; +class BatchServerTransport implements RpcTransportWithCustomEncoding { + readonly encodingLevel = "jsonCompatibleWithBytes" as const; + + constructor(rawBatch: RawBatch, options?: RpcSessionOptions) { + this.#rawBatch = rawBatch; + this.#maxMessageSize = options?.limits?.maxMessageSize ?? DEFAULT_LIMITS.maxMessageSize; + this.#maxDepth = options?.limits?.maxDepth ?? DEFAULT_LIMITS.maxDepth; } - #batchToSend: string[] = []; - #batchToReceive: string[]; + #rawBatch?: RawBatch; + #maxMessageSize: number; + #maxDepth: number; + #batchToSend: unknown[] = []; + #batchToReceive?: unknown[]; + #linesToReceive?: string[]; #allReceived: PromiseWithResolvers = Promise.withResolvers(); - send(message: string): void { + send(message: unknown): void { this.#batchToSend.push(message); } - async receive(): Promise { - let msg = this.#batchToReceive!.shift(); - if (msg !== undefined) { - return msg; - } else { - // No more messages. - this.#allReceived.resolve(); - return new Promise(r => {}); + async receive(): Promise { + if (this.#rawBatch !== undefined) { + let raw = this.#rawBatch; + this.#rawBatch = undefined; + if (raw.isBinary) { + this.#batchToReceive = decodeBinaryBatch( + raw.buffer, + this.#maxMessageSize, + this.#maxDepth + ); + } else { + this.#linesToReceive = raw.body === "" ? [] : raw.body.split("\n"); + } } - } + if (this.#linesToReceive !== undefined) { + let line = this.#linesToReceive.shift(); + if (line !== undefined) { + if (line.length > this.#maxMessageSize) { + throw new TypeError( + `Incoming message exceeds maximum size of ${this.#maxMessageSize} UTF-16 code units.` + ); + } + return JSON.parse(line); + } + } else if (this.#batchToReceive !== undefined) { + let msg = this.#batchToReceive.shift(); + if (msg !== undefined) { + return msg; + } + } + + // No more messages. + this.#allReceived.resolve(); + return new Promise(r => {}); + } abort?(reason: any): void { this.#allReceived.reject(reason); } - whenAllReceived() { + whenAllReceived(): Promise { return this.#allReceived.promise; } - getResponseBody(): string { - return this.#batchToSend.join("\n"); + return this.#batchToSend.map(msg => stringifyWithBase64Bytes(msg)).join("\n"); + } + + getResponseBinaryBody(): Uint8Array { + return encodeBinaryFrame(this.#batchToSend) as Uint8Array; } } @@ -147,13 +274,16 @@ export async function newHttpBatchRpcResponse( if (request.method !== "POST") { return new Response("This endpoint only accepts POST requests.", { status: 405 }); } - - let body = await request.text(); - let batch = body === "" ? [] : body.split("\n"); - - let transport = new BatchServerTransport(batch); + let contentType = request.headers.get("Content-Type"); + let accept = request.headers.get("Accept"); + let buffer = await request.arrayBuffer(); + let isBinaryRequest = hasBinaryMimeType(contentType) || hasBinaryFrameMagic(buffer); + let rawBatch: RawBatch = isBinaryRequest + ? { isBinary: true, buffer } + : { isBinary: false, body: new TextDecoder().decode(buffer) }; + + let transport = new BatchServerTransport(rawBatch, options); let rpc = new RpcSession(transport, localMain, options); - // TODO: Arguably we should arrange so any attempts to pull promise resolutions from the client // will reject rather than just hang. But it IS valid to make server->client calls in order to // then pipeline the result into something returned to the client. We don't want the errors to @@ -166,6 +296,12 @@ export async function newHttpBatchRpcResponse( // TODO: Ask RpcSession to dispose everything it is still holding on to? + let respondBinary = options?.binary ?? (isBinaryRequest || hasBinaryMimeType(accept)); + if (respondBinary) { + return new Response(transport.getResponseBinaryBody(), { + headers: { "Content-Type": BINARY_CONTENT_TYPE }, + }); + } return new Response(transport.getResponseBody()); } @@ -188,25 +324,46 @@ export async function nodeHttpBatchRpcResponse( response.end(); return; } - - let body = await new Promise((resolve, reject) => { - let chunks: Buffer[] = []; - request.on("data", chunk => { - chunks.push(chunk); - }); - request.on("end", () => { - resolve(Buffer.concat(chunks).toString()); - }); - request.on("error", reject); + let { promise: bodyPromise, resolve, reject } = Promise.withResolvers(); + let chunks: Buffer[] = []; + request.on("data", (chunk: Buffer) => { + chunks.push(chunk); + }); + request.on("end", () => { + resolve(Buffer.concat(chunks)); }); - let batch = body === "" ? [] : body.split("\n"); + request.on("error", reject); + let rawBody = await bodyPromise; + let contentTypeHeader = request.headers["content-type"]; + let contentType = Array.isArray(contentTypeHeader) ? contentTypeHeader.join(", ") : contentTypeHeader; + let acceptHeader = request.headers["accept"]; + let accept = Array.isArray(acceptHeader) ? acceptHeader.join(", ") : acceptHeader; + let isBinaryRequest = hasBinaryMimeType(contentType) || hasBinaryFrameMagic(rawBody); + let rawBatch: RawBatch; + if (isBinaryRequest) { + let buffer = rawBody.byteOffset === 0 && rawBody.byteLength === rawBody.buffer.byteLength && + rawBody.buffer instanceof ArrayBuffer + ? rawBody.buffer + : rawBody.buffer.slice(rawBody.byteOffset, rawBody.byteOffset + rawBody.byteLength) as ArrayBuffer; + rawBatch = { isBinary: true, buffer }; + } else { + rawBatch = { isBinary: false, body: rawBody.toString() }; + } - let transport = new BatchServerTransport(batch); + let transport = new BatchServerTransport(rawBatch, options); let rpc = new RpcSession(transport, localMain, options); await transport.whenAllReceived(); await rpc.drain(); - - response.writeHead(200, options?.headers); - response.end(transport.getResponseBody()); + let respondBinary = options?.binary ?? (isBinaryRequest || hasBinaryMimeType(accept)); + if (respondBinary) { + if (!response.hasHeader("content-type")) { + response.setHeader("Content-Type", BINARY_CONTENT_TYPE); + } + response.writeHead(200, options?.headers); + response.end(transport.getResponseBinaryBody()); + } else { + response.writeHead(200, options?.headers); + response.end(transport.getResponseBody()); + } } diff --git a/src/rpc.ts b/src/rpc.ts index d3fc8267..f72f5e6e 100644 --- a/src/rpc.ts +++ b/src/rpc.ts @@ -464,6 +464,13 @@ export type RpcSessionOptions = { * session. */ limits?: Partial; + + /** + * Enable native binary framing on transports that support it (such as WebSocket and HTTP batch). + * When omitted, transports default to text/JSON framing for backward compatibility while + * automatically accepting (and replying to) binary frames when received. + */ + binary?: boolean; }; class RpcSessionImpl implements Importer, Exporter { diff --git a/src/serialize.ts b/src/serialize.ts index c8f7118d..9a646270 100644 --- a/src/serialize.ts +++ b/src/serialize.ts @@ -398,21 +398,7 @@ export class Devaluator { ? ["bytes", bytes] : ["bytes", bytes, alternateTypeName]; } - let b64: string; - if (bytes.toBase64) { - b64 = bytes.toBase64({omitPadding: true}); - } else if (typeof Buffer !== "undefined") { - let buf = bytes instanceof Buffer ? bytes - : Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength); - b64 = buf.toString("base64"); - } else { - let binary = ""; - for (let i = 0; i < bytes.length; i++) { - binary += String.fromCharCode(bytes[i]); - } - b64 = btoa(binary); - } - b64 = b64.replace(/=+$/, ""); + let b64 = encodeBase64(bytes); return alternateTypeName === undefined ? ["bytes", b64] : ["bytes", b64, alternateTypeName]; } @@ -947,6 +933,26 @@ export class Evaluator { `Invalid byte length ${bytes.byteLength} for ${marker}; ` + `expected a multiple of ${elementSize}`); } + if (value[1] instanceof Uint8Array && NATIVE_LITTLE_ENDIAN && + marker !== "ArrayBuffer" && + (elementSize === undefined || bytes.byteOffset % elementSize === 0)) { + let buf = bytes.buffer; + let off = bytes.byteOffset; + let len = bytes.byteLength; + switch (marker) { + case "DataView": return new DataView(buf, off, len); + case "Int8Array": return new Int8Array(buf, off, len); + case "Uint8ClampedArray": return new Uint8ClampedArray(buf, off, len); + case "Int16Array": return new Int16Array(buf, off, len / 2); + case "Uint16Array": return new Uint16Array(buf, off, len / 2); + case "Int32Array": return new Int32Array(buf, off, len / 4); + case "Uint32Array": return new Uint32Array(buf, off, len / 4); + case "BigInt64Array": return new BigInt64Array(buf, off, len / 8); + case "BigUint64Array": return new BigUint64Array(buf, off, len / 8); + case "Float32Array": return new Float32Array(buf, off, len / 4); + case "Float64Array": return new Float64Array(buf, off, len / 8); + } + } // Copy exactly the decoded range rather than exposing or aliasing a pooled Buffer. let buffer = bytes.buffer.slice(bytes.byteOffset, bytes.byteOffset + bytes.byteLength); @@ -1337,3 +1343,304 @@ export function deserialize(value: string): unknown { payload.dispose(); // should be no-op but just in case return payload.value; } + +// ======================================================================================= +// Binary frame codec for `jsonCompatibleWithBytes` transports (WebSocket & HTTP batch) + +export const BINARY_CONTENT_TYPE = "application/x-capnweb-bin"; + +const BINARY_MAGIC_0 = 0x89; +const BINARY_MAGIC_1 = 0x43; // 'C' +const BINARY_MAGIC_2 = 0x57; // 'W' +const BINARY_MAGIC_3 = 0x42; // 'B' +const BINARY_HEADER_SIZE = 8; + +const textEncoder = new TextEncoder(); +const textDecoder = new TextDecoder(); + +function align8(n: number): number { + let rem = n % 8; + return rem === 0 ? n : n + (8 - rem); +} + +function encodeBase64(bytes: Uint8Array): string { + let b64: string; + if (bytes.toBase64) { + b64 = bytes.toBase64({omitPadding: true}); + } else if (typeof Buffer !== "undefined") { + let buf = bytes instanceof Buffer ? bytes + : Buffer.from(bytes.buffer, bytes.byteOffset, bytes.byteLength); + b64 = buf.toString("base64"); + } else { + let binary = ""; + for (let i = 0; i < bytes.length; i++) { + binary += String.fromCharCode(bytes[i]); + } + b64 = btoa(binary); + } + return b64.replace(/=+$/, ""); +} + +/** + * Stringifies a devaluated `jsonCompatibleWithBytes` tree as standard JSON text, converting any + * raw `["bytes", Uint8Array, typeName?]` leaves into Base64 for compatibility with text/JSON peers. + */ +export function stringifyWithBase64Bytes(value: unknown): string { + return JSON.stringify(value, (_key, val) => { + if (Array.isArray(val) && val[0] === "bytes" && val[1] instanceof Uint8Array) { + let b64 = encodeBase64(val[1]); + return val.length === 3 ? ["bytes", b64, val[2]] : ["bytes", b64]; + } + return val; + }); +} + +/** + * Checks whether a buffer begins with the 4-byte `\x89CWB` binary frame magic sequence. + */ +export function hasBinaryFrameMagic(input: ArrayBuffer | Uint8Array): boolean { + let bytes = input instanceof Uint8Array ? input : new Uint8Array(input); + return bytes.byteLength >= BINARY_HEADER_SIZE && + bytes[0] === BINARY_MAGIC_0 && + bytes[1] === BINARY_MAGIC_1 && + bytes[2] === BINARY_MAGIC_2 && + bytes[3] === BINARY_MAGIC_3; +} + +/** + * Packs a devaluated `jsonCompatibleWithBytes` tree into a compact, 8-byte-aligned binary frame: + * - Bytes 0..3: `\x89CWB` magic + * - Bytes 4..7: little-endian uint32 UTF-8 JSON header byte length + * - Bytes 8..8+headerByteLength: UTF-8 JSON header where `["bytes", Uint8Array, typeName?]` + * leaves are replaced with `["bytes", attachmentIndex, byteOffset, byteLength, typeName?]` + * - 0..7 zero padding bytes (when attachments are present) to align attachments to 8 bytes + * - 8-byte-aligned raw binary attachments + */ +export function encodeBinaryFrame(value: unknown): Uint8Array { + let attachments: Uint8Array[] = []; + let offsets: number[] = []; + let currentOffset = 0; + + let jsonHeader = JSON.stringify(value, (_key, val) => { + if (Array.isArray(val) && val[0] === "bytes") { + if (val[1] instanceof Uint8Array) { + let bytes = val[1]; + let index = attachments.length; + let byteOffset = align8(currentOffset); + let byteLength = bytes.byteLength; + attachments.push(bytes); + offsets.push(byteOffset); + currentOffset = byteOffset + byteLength; + return val.length === 3 + ? ["bytes", index, byteOffset, byteLength, val[2]] + : ["bytes", index, byteOffset, byteLength]; + } + if (typeof val[1] === "number" || val[1] === null) { + return ["bytes", null, ...val.slice(1)]; + } + } + return val; + }); + if (jsonHeader === undefined) { + throw new TypeError("Cannot encode undefined as a binary RPC frame."); + } + let headerBytes = textEncoder.encode(jsonHeader); + let headerByteLength = headerBytes.byteLength; + let attachmentsStart = attachments.length > 0 + ? align8(BINARY_HEADER_SIZE + headerByteLength) + : BINARY_HEADER_SIZE + headerByteLength; + let totalByteLength = attachmentsStart + currentOffset; + + let buffer = new ArrayBuffer(totalByteLength); + let frame = new Uint8Array(buffer); + let view = new DataView(buffer); + + frame[0] = BINARY_MAGIC_0; + frame[1] = BINARY_MAGIC_1; + frame[2] = BINARY_MAGIC_2; + frame[3] = BINARY_MAGIC_3; + view.setUint32(4, headerByteLength, true); + + frame.set(headerBytes, BINARY_HEADER_SIZE); + + for (let i = 0; i < attachments.length; i++) { + frame.set(attachments[i], attachmentsStart + offsets[i]); + } + + return frame; +} + +/** + * Decodes a binary frame created by `encodeBinaryFrame`, replacing attachment markers with + * 8-byte-aligned `Uint8Array` views directly into the incoming `ArrayBuffer`. + */ +export function decodeBinaryFrame( + input: ArrayBuffer | Uint8Array, + maxDepth: number = DEFAULT_MAX_DEPTH +): unknown { + let buffer: ArrayBuffer; + if (input instanceof Uint8Array) { + if (input.byteOffset === 0 && input.byteLength === input.buffer.byteLength && + input.buffer instanceof ArrayBuffer) { + buffer = input.buffer; + } else { + buffer = input.buffer.slice(input.byteOffset, input.byteOffset + input.byteLength) as ArrayBuffer; + } + } else { + buffer = input; + } + + if (buffer.byteLength < BINARY_HEADER_SIZE) { + throw new TypeError("Truncated binary RPC frame."); + } + + let frame = new Uint8Array(buffer); + if (frame[0] !== BINARY_MAGIC_0 || + frame[1] !== BINARY_MAGIC_1 || + frame[2] !== BINARY_MAGIC_2 || + frame[3] !== BINARY_MAGIC_3) { + throw new TypeError("Invalid binary RPC frame magic."); + } + + let view = new DataView(buffer); + let headerByteLength = view.getUint32(4, true); + if (headerByteLength > buffer.byteLength - BINARY_HEADER_SIZE) { + throw new TypeError("Binary RPC frame header length exceeds frame size."); + } + + let attachmentsStart = align8(BINARY_HEADER_SIZE + headerByteLength); + + let jsonHeader = textDecoder.decode(new Uint8Array(buffer, BINARY_HEADER_SIZE, headerByteLength)); + let parsed = JSON.parse(jsonHeader); + let state = { nextIndex: 0, nextOffset: 0 }; + let result = rehydrateBinaryAttachments( + parsed, + buffer, + attachmentsStart, + state, + 0, + maxDepth + ); + let expectedByteLength = state.nextIndex > 0 + ? attachmentsStart + state.nextOffset + : BINARY_HEADER_SIZE + headerByteLength; + if (expectedByteLength !== buffer.byteLength) { + throw new TypeError("Binary RPC frame size does not match header and attachments."); + } + return result; +} + +function rehydrateBinaryAttachments( + node: unknown, + buffer: ArrayBuffer, + attachmentsStart: number, + state: { nextIndex: number; nextOffset: number }, + depth: number, + maxDepth: number +): unknown { + if (depth >= maxDepth) { + throw new TypeError( + "Deserialization exceeded maximum allowed depth. (Does the message contain cycles?)" + ); + } + if (Array.isArray(node)) { + if (node[0] === "bytes") { + if (node[1] === null) { + if ( + node.length < 3 || + (typeof node[2] !== "number" && node[2] !== null) + ) { + throw new TypeError("Invalid binary attachment descriptor."); + } + node.splice(1, 1); + for (let i = 1; i < node.length; i++) { + node[i] = rehydrateBinaryAttachments( + node[i], + buffer, + attachmentsStart, + state, + depth + 1, + maxDepth + ); + } + return node; + } + if (typeof node[1] === "number") { + if ( + (node.length !== 4 && node.length !== 5) || + typeof node[2] !== "number" || + typeof node[3] !== "number" || + (node.length === 5 && typeof node[4] !== "string") + ) { + throw new TypeError("Invalid binary attachment descriptor."); + } + let attachmentIndex = node[1]; + let byteOffset = node[2]; + let byteLength = node[3]; + if (attachmentIndex !== state.nextIndex) { + throw new TypeError( + `Invalid binary attachment index: ${attachmentIndex}` + ); + } + if (byteOffset !== align8(state.nextOffset)) { + throw new TypeError( + `Invalid binary attachment offset: ${byteOffset}` + ); + } + if (!Number.isInteger(byteLength) || byteLength < 0) { + throw new TypeError( + `Invalid binary attachment length: ${byteLength}` + ); + } + let absOffset = attachmentsStart + byteOffset; + if (absOffset + byteLength > buffer.byteLength) { + throw new TypeError("Binary attachment exceeds frame bounds."); + } + state.nextIndex++; + state.nextOffset = byteOffset + byteLength; + let bytes = new Uint8Array(buffer, absOffset, byteLength); + return node.length === 5 ? ["bytes", bytes, node[4]] : ["bytes", bytes]; + } + } + for (let i = 0; i < node.length; i++) { + node[i] = rehydrateBinaryAttachments( + node[i], + buffer, + attachmentsStart, + state, + depth + 1, + maxDepth + ); + } + return node; + } + + if ( + node !== null && + typeof node === "object" && + !(node instanceof Uint8Array) + ) { + let record = node as Record; + for (let key of Object.keys(record)) { + let next = rehydrateBinaryAttachments( + record[key], + buffer, + attachmentsStart, + state, + depth + 1, + maxDepth + ); + if (key === "__proto__") { + Object.defineProperty(record, key, { + value: next, + enumerable: true, + configurable: true, + writable: true, + }); + } else { + record[key] = next; + } + } + } + return node; +} diff --git a/src/websocket.ts b/src/websocket.ts index e7b7cbc6..6d7c4cdf 100644 --- a/src/websocket.ts +++ b/src/websocket.ts @@ -5,7 +5,13 @@ /// import { RpcStub } from "./core.js"; -import { RpcSession, RpcSessionOptions } from "./rpc.js"; +import { RpcSession, RpcSessionOptions, RpcTransportWithCustomEncoding } from "./rpc.js"; +import { + DEFAULT_LIMITS, + decodeBinaryFrame, + encodeBinaryFrame, + stringifyWithBase64Bytes, +} from "./serialize.js"; /** Close-frame reason max UTF-8 bytes: 125 payload - 2-byte status (RFC 6455 section 5.5). */ export const MAX_CLOSE_REASON_BYTES = 125 - 2; @@ -16,7 +22,7 @@ export function newWebSocketRpcSession( webSocket = new WebSocket(webSocket); } - let transport = new WebSocketTransport(webSocket); + let transport = new WebSocketFrameTransport(webSocket, options); let rpc = new RpcSession(transport, localMain, options); return rpc.getRemoteMain(); } @@ -69,13 +75,27 @@ export class WebSocketTransport { webSocket.addEventListener("message", (event: MessageEvent) => { if (this.#error) { // Ignore further messages. - } else if (typeof event.data === "string" || event.data instanceof ArrayBuffer) { + } else if ( + typeof event.data === "string" || + event.data instanceof ArrayBuffer || + ArrayBuffer.isView(event.data) + ) { + let data: string | ArrayBuffer; + if (ArrayBuffer.isView(event.data)) { + let view = event.data; + data = view.byteOffset === 0 && view.byteLength === view.buffer.byteLength && + view.buffer instanceof ArrayBuffer + ? view.buffer + : view.buffer.slice(view.byteOffset, view.byteOffset + view.byteLength) as ArrayBuffer; + } else { + data = event.data; + } if (this.#receiveResolver) { - this.#receiveResolver(event.data as T); + this.#receiveResolver(data as T); this.#receiveResolver = undefined; this.#receiveRejecter = undefined; } else { - this.#receiveQueue.push(event.data as T); + this.#receiveQueue.push(data as T); } } else { this.#receivedError(new TypeError("Received unexpected message type from WebSocket.")); @@ -98,25 +118,30 @@ export class WebSocketTransport { #receiveQueue: T[] = []; #error?: any; - send(message: T): void { + async send(message: T): Promise { if (this.#sendQueue === undefined) { - this.#webSocket.send(message); + try { + this.#webSocket.send(message); + } catch (err) { + this.#receivedError(err); + throw err; + } } else { // Not open yet, queue for later. this.#sendQueue.push(message); } } - receive(): Promise { + async receive(): Promise { if (this.#receiveQueue.length > 0) { - return Promise.resolve(this.#receiveQueue.shift()!); + return this.#receiveQueue.shift()!; } else if (this.#error) { - return Promise.reject(this.#error); + throw this.#error; } else { - return new Promise((resolve, reject) => { - this.#receiveResolver = resolve; - this.#receiveRejecter = reject; - }); + let { promise, resolve, reject } = Promise.withResolvers(); + this.#receiveResolver = resolve; + this.#receiveRejecter = reject; + return promise; } } @@ -151,7 +176,62 @@ export class WebSocketTransport { } } } - // This class is generic, so it can't `implements RpcTransport` (that would require every `T` to // conform, but the ArrayBuffer instantiation intentionally doesn't). The default string // instantiation's conformance is asserted in __type-tests__/rpc-types.test.ts. + +class WebSocketFrameTransport implements RpcTransportWithCustomEncoding { + readonly encodingLevel = "jsonCompatibleWithBytes" as const; + #raw: WebSocketTransport; + #binary: boolean; + #autoDetect: boolean; + #maxMessageSize: number; + #maxDepth: number; + + constructor(webSocket: WebSocket, options?: RpcSessionOptions) { + this.#raw = new WebSocketTransport(webSocket); + this.#binary = options?.binary ?? false; + this.#autoDetect = options?.binary === undefined; + this.#maxMessageSize = options?.limits?.maxMessageSize ?? DEFAULT_LIMITS.maxMessageSize; + this.#maxDepth = options?.limits?.maxDepth ?? DEFAULT_LIMITS.maxDepth; + } + + send(message: unknown): number { + if (this.#binary) { + let frame = encodeBinaryFrame(message); + this.#raw.send(frame.buffer as ArrayBuffer).catch(() => {}); + return frame.byteLength; + } else { + let text = stringifyWithBase64Bytes(message); + this.#raw.send(text).catch(() => {}); + return text.length; + } + } + + async receive(): Promise { + let raw = await this.#raw.receive(); + if (typeof raw === "string") { + if (raw.length > this.#maxMessageSize) { + throw new TypeError( + `Incoming message exceeds maximum size of ${this.#maxMessageSize} UTF-16 code units.` + ); + } + return JSON.parse(raw); + } else { + if (raw.byteLength > this.#maxMessageSize) { + throw new TypeError( + `Incoming message exceeds maximum size of ${this.#maxMessageSize} bytes.` + ); + } + let decoded = decodeBinaryFrame(raw, this.#maxDepth * 2 + 2); + if (this.#autoDetect) { + this.#binary = true; + } + return decoded; + } + } + + abort(reason: unknown): void { + this.#raw.abort(reason); + } +} From 5e80fa92d748b84254e1fc63c526ee6005c8a2f9 Mon Sep 17 00:00:00 2001 From: Steven Chong <25894545+teamchong@users.noreply.github.com> Date: Mon, 9 Nov 2026 00:07:59 -0500 Subject: [PATCH 2/2] fix: remove internal NATIVE_LITTLE_ENDIAN test import and handle CORS preflight --- __tests__/index.test.ts | 23 ++++++++++++----------- src/batch.ts | 12 ++++++++++++ 2 files changed, 24 insertions(+), 11 deletions(-) diff --git a/__tests__/index.test.ts b/__tests__/index.test.ts index 7c57faba..12d4c7e9 100644 --- a/__tests__/index.test.ts +++ b/__tests__/index.test.ts @@ -7,16 +7,17 @@ import { deserialize, serialize, RpcSession, type RpcSessionOptions, RpcTranspor type RpcTransportWithCustomEncoding, RpcTarget, RpcStub, RpcPromise, newWebSocketRpcSession, newMessagePortRpcSession, newHttpBatchRpcSession, - newHttpBatchRpcResponse} from "../src/index.js" -import { BINARY_CONTENT_TYPE, - decodeBinaryFrame, - encodeBinaryFrame, - Evaluator, - Devaluator, - hasBinaryFrameMagic, - NATIVE_LITTLE_ENDIAN, - stringifyWithBase64Bytes, - swapByteOrder } from "../src/serialize.js" + newHttpBatchRpcResponse } from "../src/index.js" +import { + BINARY_CONTENT_TYPE, + decodeBinaryFrame, + encodeBinaryFrame, + Evaluator, + Devaluator, + hasBinaryFrameMagic, + stringifyWithBase64Bytes, + swapByteOrder, +} from "../src/serialize.js" import { MAX_CLOSE_REASON_BYTES } from "../src/websocket.js" import { ErrorStubHook, PayloadStubHook, PromiseStubHook, RpcPayload, RpcStub as RawRpcStub, StubHook, streamImpl, unwrapStubTakingOwnership, type PropertyPath } from "../src/core.js" @@ -4623,7 +4624,7 @@ describe("binary framing codec (jsonCompatibleWithBytes)", () => { expect(evaluated.bi64).toStrictEqual(bi64); expect(evaluated.literalBytesTuple).toStrictEqual(["bytes", 10, 20, 30]); - if (NATIVE_LITTLE_ENDIAN) { + if (new Uint8Array(new Uint16Array([1]).buffer)[0] === 1) { // TypedArrays and Uint8Array views are zero-copy views directly into the frame's ArrayBuffer! expect(evaluated.u8Odd.buffer).toBe(frame.buffer); expect(evaluated.f64.buffer).toBe(frame.buffer); diff --git a/src/batch.ts b/src/batch.ts index f517b54a..10fd2d45 100644 --- a/src/batch.ts +++ b/src/batch.ts @@ -319,6 +319,18 @@ export async function nodeHttpBatchRpcResponse( options?: RpcSessionOptions & { headers?: OutgoingHttpHeaders | OutgoingHttpHeader[], }): Promise { + if (request.method === "OPTIONS" && request.headers["access-control-request-method"]) { + if (!response.hasHeader("access-control-allow-methods")) { + response.setHeader("Access-Control-Allow-Methods", "POST"); + } + let reqHeaders = request.headers["access-control-request-headers"]; + if (reqHeaders && !response.hasHeader("access-control-allow-headers")) { + response.setHeader("Access-Control-Allow-Headers", reqHeaders); + } + response.writeHead(204, options?.headers); + response.end(); + return; + } if (request.method !== "POST") { response.writeHead(405, "This endpoint only accepts POST requests.", options?.headers); response.end();