From 3b6b805fe6e46ed89b5c7077963af10001585386 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 03:04:46 +0000 Subject: [PATCH 1/2] fix(peer): copy octet-stream chunks that view part of a larger buffer `OctetStreamTransmitter` forwarded each chunk as-is. A chunk that is a view onto a larger ArrayBuffer (Node pooled `Buffer` slices from `Buffer.from`, `Buffer.concat`, `allocUnsafe`, `subarray`) sent over a transport that structured-clones raw messages (Worker, MessagePort, Electron IPC) cloned the whole backing buffer to the receiver, including unrelated pooled data. Copy a chunk when its view doesn't cover its whole buffer. Chunks that already own their buffer are still sent without a copy. The other binary send paths are unaffected: atomic bodies are always Blobs, and `encodePeerMessage` copies into a fresh frame. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011cCB5ionSFz6U6a1SUr1T8 --- packages/peer/src/octet-stream.test.ts | 43 ++++++++++++++++++++++++++ packages/peer/src/octet-stream.ts | 17 +++++++++- 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/packages/peer/src/octet-stream.test.ts b/packages/peer/src/octet-stream.test.ts index bf910b4..11370b8 100644 --- a/packages/peer/src/octet-stream.test.ts +++ b/packages/peer/src/octet-stream.test.ts @@ -1,3 +1,5 @@ +import { Buffer } from 'node:buffer' + import { Queue } from '@standard-server/shared' import { OctetStreamTransmitter, toOctetStream } from './octet-stream' @@ -141,6 +143,47 @@ describe('octetStreamTransmitter', () => { }) }) + it('copies chunks that view part of a larger buffer', async () => { + const pooled = Buffer.from('hello') + const view = new Uint8Array(new ArrayBuffer(16), 4, 3).fill(7) + expect(pooled.buffer.byteLength).toBeGreaterThan(pooled.byteLength) // Node's shared Buffer pool + + const send = vi.fn(async (_message: PeerOctetStreamMessage) => {}) + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(pooled) + controller.enqueue(view) + controller.close() + }, + }) + + await new OctetStreamTransmitter(stream, 'msg-1', send).transmit() + + const sent = send.mock.calls.map(([message]) => message.binary as Uint8Array) + + expect(sent[0]!.buffer.byteLength).toBe(sent[0]!.byteLength) + expect(structuredClone(sent[0]!).buffer.byteLength).toBe(5) + expect(new TextDecoder().decode(sent[0])).toBe('hello') + + expect(sent[1]!.buffer.byteLength).toBe(sent[1]!.byteLength) + expect(sent[1]).toEqual(new Uint8Array([7, 7, 7])) + }) + + it('sends chunks that own their whole buffer as-is', async () => { + const chunk = new Uint8Array([1, 2, 3]) + const send = vi.fn(async (_message: PeerOctetStreamMessage) => {}) + const stream = new ReadableStream({ + start(controller) { + controller.enqueue(chunk) + controller.close() + }, + }) + + await new OctetStreamTransmitter(stream, 'msg-1', send).transmit() + + expect(send.mock.calls[0]![0].binary).toBe(chunk) + }) + it('cancel stops transmission and cancels reader', async () => { const send = vi.fn(async () => {}) const cancel = vi.fn() diff --git a/packages/peer/src/octet-stream.ts b/packages/peer/src/octet-stream.ts index 95439f9..c045350 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -32,6 +32,21 @@ export function toOctetStream( }) } +/** + * Copy a chunk that views only part of its `ArrayBuffer` (e.g. a slice of Node's + * pooled `Buffer` memory), so transports that structured-clone messages don't send + * the whole backing buffer, including unrelated data, to the remote peer. + * Chunks that already own their whole buffer are returned as-is. + */ +function toStandaloneBytes(bytes: Uint8Array): Uint8Array { + if (bytes.byteOffset === 0 && bytes.byteLength === bytes.buffer.byteLength) { + return bytes + } + + // Not `bytes.slice()`: `Buffer#slice` returns a view, not a copy + return new Uint8Array(bytes) +} + export class OctetStreamTransmitter { private isDone = false private readonly reader: ReadableStreamDefaultReader> @@ -67,7 +82,7 @@ export class OctetStreamTransmitter { try { await this.send({ json: { close: item.done ? true : undefined }, - binary: item.value, + binary: item.value && toStandaloneBytes(item.value), kind: 'octet-stream', id: this.messageId, }) From b72bd8de326b90c5c24388616688bee4faedc23c Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 7 Oct 2026 03:10:50 +0000 Subject: [PATCH 2/2] refactor(peer): simplify standalone-bytes check and its tests Drop the `byteOffset === 0` clause (implied by a view covering its whole buffer), shorten the helper's doc comment, and fold the two transmitter tests into one that covers both the copy and the pass-through branch. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_011cCB5ionSFz6U6a1SUr1T8 --- packages/peer/src/octet-stream.test.ts | 26 ++++---------------------- packages/peer/src/octet-stream.ts | 8 +++----- 2 files changed, 7 insertions(+), 27 deletions(-) diff --git a/packages/peer/src/octet-stream.test.ts b/packages/peer/src/octet-stream.test.ts index 11370b8..c6a72ee 100644 --- a/packages/peer/src/octet-stream.test.ts +++ b/packages/peer/src/octet-stream.test.ts @@ -143,16 +143,16 @@ describe('octetStreamTransmitter', () => { }) }) - it('copies chunks that view part of a larger buffer', async () => { + it('copies chunks that view part of a larger buffer, sends whole-buffer chunks as-is', async () => { const pooled = Buffer.from('hello') - const view = new Uint8Array(new ArrayBuffer(16), 4, 3).fill(7) + const whole = new Uint8Array([1, 2, 3]) expect(pooled.buffer.byteLength).toBeGreaterThan(pooled.byteLength) // Node's shared Buffer pool const send = vi.fn(async (_message: PeerOctetStreamMessage) => {}) const stream = new ReadableStream({ start(controller) { controller.enqueue(pooled) - controller.enqueue(view) + controller.enqueue(whole) controller.close() }, }) @@ -162,26 +162,8 @@ describe('octetStreamTransmitter', () => { const sent = send.mock.calls.map(([message]) => message.binary as Uint8Array) expect(sent[0]!.buffer.byteLength).toBe(sent[0]!.byteLength) - expect(structuredClone(sent[0]!).buffer.byteLength).toBe(5) expect(new TextDecoder().decode(sent[0])).toBe('hello') - - expect(sent[1]!.buffer.byteLength).toBe(sent[1]!.byteLength) - expect(sent[1]).toEqual(new Uint8Array([7, 7, 7])) - }) - - it('sends chunks that own their whole buffer as-is', async () => { - const chunk = new Uint8Array([1, 2, 3]) - const send = vi.fn(async (_message: PeerOctetStreamMessage) => {}) - const stream = new ReadableStream({ - start(controller) { - controller.enqueue(chunk) - controller.close() - }, - }) - - await new OctetStreamTransmitter(stream, 'msg-1', send).transmit() - - expect(send.mock.calls[0]![0].binary).toBe(chunk) + expect(sent[1]).toBe(whole) }) it('cancel stops transmission and cancels reader', async () => { diff --git a/packages/peer/src/octet-stream.ts b/packages/peer/src/octet-stream.ts index c045350..88fe818 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -33,13 +33,11 @@ export function toOctetStream( } /** - * Copy a chunk that views only part of its `ArrayBuffer` (e.g. a slice of Node's - * pooled `Buffer` memory), so transports that structured-clone messages don't send - * the whole backing buffer, including unrelated data, to the remote peer. - * Chunks that already own their whole buffer are returned as-is. + * Copy a chunk that views only part of its `ArrayBuffer` (e.g. Node's pooled `Buffer`), + * so structured-clone transports don't send the whole backing buffer to the remote peer. */ function toStandaloneBytes(bytes: Uint8Array): Uint8Array { - if (bytes.byteOffset === 0 && bytes.byteLength === bytes.buffer.byteLength) { + if (bytes.byteLength === bytes.buffer.byteLength) { return bytes }