diff --git a/packages/peer/src/octet-stream.test.ts b/packages/peer/src/octet-stream.test.ts index bf910b4..c6a72ee 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,29 @@ describe('octetStreamTransmitter', () => { }) }) + it('copies chunks that view part of a larger buffer, sends whole-buffer chunks as-is', async () => { + const pooled = Buffer.from('hello') + 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(whole) + 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(new TextDecoder().decode(sent[0])).toBe('hello') + expect(sent[1]).toBe(whole) + }) + 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..88fe818 100644 --- a/packages/peer/src/octet-stream.ts +++ b/packages/peer/src/octet-stream.ts @@ -32,6 +32,19 @@ export function toOctetStream( }) } +/** + * 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.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 +80,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, })