fix(peer): stop leaking pooled buffer memory over structured-clone transports

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<ArrayBuffer>)
+
+    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()

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<ArrayBuffer>): Uint8Array<ArrayBuffer> {
+  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<Uint8Array<ArrayBuffer>>
@@ -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,
           })