perf(browser): assemble fragmented tunnel frames once

config/scripts/benchmark-browser-tunnel-framing.mjs

@@ -0,0 +1,110 @@
+import assert from 'node:assert/strict'
+import { execFileSync } from 'node:child_process'
+import { readFileSync } from 'node:fs'
+import { stripTypeScriptTypes } from 'node:module'
+import { performance } from 'node:perf_hooks'
+
+// Run from the worktree root: node config/scripts/benchmark-browser-tunnel-framing.mjs [base-ref]
+const path = 'src/shared/browser-network-tunnel-stream-framing.ts'
+const baselineRef = process.argv[2] ?? 'HEAD'
+const beforeSource = execFileSync('git', ['show', `${baselineRef}:${path}`], {
+  encoding: 'utf8'
+})
+const afterSource = readFileSync(path, 'utf8')
+const load = (source) =>
+  import(
+    `data:text/javascript;base64,${Buffer.from(
+      stripTypeScriptTypes(source, { mode: 'transform' })
+    ).toString('base64')}`
+  )
+const before = await load(beforeSource)
+const after = await load(afterSource)
+
+function measure(module, chunks, payload, repetitions) {
+  let frameCount = 0
+  let lastFrame
+  const onFrame = (frame) => {
+    frameCount++
+    lastFrame = frame
+  }
+  const onError = (error) => {
+    throw error
+  }
+  const run = () => {
+    const decoder = new module.BrowserNetworkTunnelStreamFrameDecoder(onFrame, onError)
+    for (const chunk of chunks) {
+      decoder.feed(chunk)
+    }
+  }
+  run()
+  assert.deepEqual(lastFrame, payload)
+  const samples = []
+  for (let sample = 0; sample < 5; sample++) {
+    const start = performance.now()
+    for (let iteration = 0; iteration < repetitions; iteration++) {
+      run()
+    }
+    samples.push((performance.now() - start) / repetitions)
+  }
+  assert.equal(frameCount, 1 + 5 * repetitions)
+  return samples.sort((a, b) => a - b)[2]
+}
+
+function countCopies(module, chunks) {
+  const originalSet = Uint8Array.prototype.set
+  const originalSlice = Uint8Array.prototype.slice
+  let copied = 0
+  Uint8Array.prototype.set = function (source, offset) {
+    copied += source.length
+    return originalSet.call(this, source, offset)
+  }
+  Uint8Array.prototype.slice = function (...args) {
+    const result = originalSlice.apply(this, args)
+    copied += result.length
+    return result
+  }
+  try {
+    const decoder = new module.BrowserNetworkTunnelStreamFrameDecoder(
+      () => {},
+      (error) => {
+        throw error
+      }
+    )
+    for (const chunk of chunks) {
+      decoder.feed(chunk)
+    }
+  } finally {
+    Uint8Array.prototype.set = originalSet
+    Uint8Array.prototype.slice = originalSlice
+  }
+  return copied
+}
+
+const rows = []
+for (const [payloadBytes, chunkBytes, repetitions] of [
+  [1, 5, 10000],
+  [64 * 1024, 65540, 1000],
+  [64 * 1024, 4096, 100],
+  [64 * 1024, 256, 25],
+  [64 * 1024, 16, 5],
+  [64 * 1024, 1, 1]
+]) {
+  const payload = Uint8Array.from({ length: payloadBytes }, (_, index) => index % 251)
+  const encoded = before.encodeBrowserNetworkTunnelStreamFrame(payload)
+  const chunks = []
+  for (let offset = 0; offset < encoded.length; offset += chunkBytes) {
+    chunks.push(encoded.subarray(offset, offset + chunkBytes))
+  }
+  const beforeMs = measure(before, chunks, payload, repetitions)
+  const afterMs = measure(after, chunks, payload, repetitions)
+  rows.push({
+    payloadBytes,
+    chunkBytes,
+    beforeMs: +beforeMs.toFixed(6),
+    afterMs: +afterMs.toFixed(6),
+    speedup: +(beforeMs / afterMs).toFixed(2),
+    beforeCopiedBytes: countCopies(before, chunks),
+    afterCopiedBytes: countCopies(after, chunks)
+  })
+}
+console.log(JSON.stringify({ node: process.version, baselineRef, rows }, null, 2))

src/shared/browser-network-tunnel-stream-framing.test.ts

@@ -55,6 +55,98 @@ describe('browser network tunnel stream framing', () => {
     )
   })
 
+  it.each([1, 2, 3, 16, 256, 4096, 65556])(
+    'preserves maximum-size frames split into %i-byte chunks',
+    (chunkSize) => {
+      const payload = Uint8Array.from({ length: 65552 }, (_, index) => index % 251)
+      const encoded = encodeBrowserNetworkTunnelStreamFrame(payload)
+      const frames: Uint8Array[] = []
+      const onError = vi.fn()
+      const decoder = new BrowserNetworkTunnelStreamFrameDecoder(
+        (frame) => frames.push(frame),
+        onError
+      )
+      for (let offset = 0; offset < encoded.length; offset += chunkSize) {
+        decoder.feed(encoded.subarray(offset, offset + chunkSize))
+      }
+      expect(frames).toEqual([payload])
+      expect(onError).not.toHaveBeenCalled()
+    }
+  )
+
+  it('copies fragmented bytes once instead of recopying the growing carry', () => {
+    const encoded = encodeBrowserNetworkTunnelStreamFrame(new Uint8Array(65536))
+    const decoder = new BrowserNetworkTunnelStreamFrameDecoder(
+      () => {},
+      () => {}
+    )
+    const originalSet = Uint8Array.prototype.set
+    let copiedBytes = 0
+    const set = vi
+      .spyOn(Uint8Array.prototype, 'set')
+      .mockImplementation(function (this: Uint8Array, source, offset) {
+        copiedBytes += source.length
+        originalSet.call(this, source, offset)
+      })
+    try {
+      for (const byte of encoded) {
+        decoder.feed(new Uint8Array([byte]))
+      }
+      expect(copiedBytes).toBe(encoded.length)
+    } finally {
+      set.mockRestore()
+    }
+  })
+
+  it('owns partial input and emitted frames independently of caller buffers', () => {
+    const frames: Uint8Array[] = []
+    const decoder = new BrowserNetworkTunnelStreamFrameDecoder(
+      (frame) => frames.push(frame),
+      () => {}
+    )
+    const first = new Uint8Array([0, 0, 0, 3, 1])
+    decoder.feed(first)
+    first.fill(255)
+    const rest = new Uint8Array([2, 3])
+    decoder.feed(rest)
+    rest.fill(255)
+    decoder.feed(encodeBrowserNetworkTunnelStreamFrame(new Uint8Array([4])))
+    expect(frames).toEqual([new Uint8Array([1, 2, 3]), new Uint8Array([4])])
+  })
+
+  it('enforces the retained cap before decoding complete frames in a feed', () => {
+    const onFrame = vi.fn()
+    const onError = vi.fn()
+    const decoder = new BrowserNetworkTunnelStreamFrameDecoder(onFrame, onError, 16, 8)
+    decoder.feed(new Uint8Array([0, 0]))
+    decoder.feed(new Uint8Array([0, 1, 7, 0, 0, 0, 1]))
+    decoder.feed(new Uint8Array([8]))
+    expect(onFrame).not.toHaveBeenCalled()
+    expect(onError).toHaveBeenCalledExactlyOnceWith(
+      expect.objectContaining({ message: 'browser_tunnel_stream_buffer_overflow' })
+    )
+  })
+
+  it('counts the retained header and payload against the exact cap', () => {
+    const onFrame = vi.fn()
+    const onError = vi.fn()
+    const decoder = new BrowserNetworkTunnelStreamFrameDecoder(onFrame, onError, 16, 7)
+    decoder.feed(new Uint8Array([0, 0, 0, 3, 1]))
+    decoder.feed(new Uint8Array([2, 3]))
+    expect(onFrame).toHaveBeenCalledExactlyOnceWith(new Uint8Array([1, 2, 3]))
+    expect(onError).not.toHaveBeenCalled()
+  })
+
+  it('stops decoding coalesced frames when the callback closes the decoder', () => {
+    const onFrame = vi.fn(() => decoder.close())
+    const onError = vi.fn()
+    const decoder = new BrowserNetworkTunnelStreamFrameDecoder(onFrame, onError)
+    decoder.feed(new Uint8Array([0, 0, 0, 1, 7, 0, 0, 0, 1, 8]))
+    decoder.feed(new Uint8Array([0, 0, 0, 1, 9]))
+    expect(onFrame).toHaveBeenCalledExactlyOnceWith(new Uint8Array([7]))
+    expect(onError).not.toHaveBeenCalled()
+  })
+
   it('serializes writes and rejects bounded queue overflow', () => {
     const callbacks: ((error?: Error | null) => void)[] = []
     const writes: Uint8Array[] = []

src/shared/browser-network-tunnel-stream-framing.ts

@@ -15,7 +15,10 @@ export function encodeBrowserNetworkTunnelStreamFrame(frame: Uint8Array): Uint8A
 }
 
 export class BrowserNetworkTunnelStreamFrameDecoder {
-  private retained = new Uint8Array()
+  private readonly header = new Uint8Array(LENGTH_BYTES)
+  private headerBytes = 0
+  private frame: Uint8Array | null = null
+  private frameBytes = 0
   private closed = false
 
   constructor(
@@ -29,30 +32,50 @@ export class BrowserNetworkTunnelStreamFrameDecoder {
     if (this.closed || chunk.byteLength === 0) {
       return
     }
-    if (this.retained.byteLength + chunk.byteLength > this.maxRetainedBytes) {
+    if (this.headerBytes + this.frameBytes + chunk.byteLength > this.maxRetainedBytes) {
       this.fail(new Error('browser_tunnel_stream_buffer_overflow'))
       return
     }
-    const combined = new Uint8Array(this.retained.byteLength + chunk.byteLength)
-    combined.set(this.retained)
-    combined.set(chunk, this.retained.byteLength)
     let offset = 0
-    while (combined.byteLength - offset >= LENGTH_BYTES) {
-      const length = new DataView(
-        combined.buffer,
-        combined.byteOffset + offset,
-        LENGTH_BYTES
-      ).getUint32(0, false)
-      if (length === 0 || length > this.maxFrameBytes) {
-        this.fail(new Error('browser_tunnel_stream_frame_invalid'))
+    while (offset < chunk.byteLength) {
+      if (this.headerBytes < LENGTH_BYTES) {
+        let length: number
+        if (this.headerBytes === 0 && chunk.byteLength - offset >= LENGTH_BYTES) {
+          length = new DataView(chunk.buffer, chunk.byteOffset + offset, LENGTH_BYTES).getUint32(
+            0,
+            false
+          )
+          this.headerBytes = LENGTH_BYTES
+          offset += LENGTH_BYTES
+        } else {
+          const count = Math.min(LENGTH_BYTES - this.headerBytes, chunk.byteLength - offset)
+          this.header.set(chunk.subarray(offset, offset + count), this.headerBytes)
+          this.headerBytes += count
+          offset += count
+          if (this.headerBytes < LENGTH_BYTES) {
+            return
+          }
+          length = new DataView(this.header.buffer).getUint32(0, false)
+        }
+        if (length === 0 || length > this.maxFrameBytes) {
+          this.fail(new Error('browser_tunnel_stream_frame_invalid'))
+          return
+        }
+        this.frame = new Uint8Array(length)
+      }
+      const frame = this.frame!
+      const count = Math.min(frame.byteLength - this.frameBytes, chunk.byteLength - offset)
+      frame.set(chunk.subarray(offset, offset + count), this.frameBytes)
+      this.frameBytes += count
+      offset += count
+      if (this.frameBytes < frame.byteLength) {
         return
       }
-      const end = offset + LENGTH_BYTES + length
-      if (end > combined.byteLength) {
-        break
-      }
+      this.frame = null
+      this.frameBytes = 0
+      this.headerBytes = 0
       try {
-        this.onFrame(combined.slice(offset + LENGTH_BYTES, end))
+        this.onFrame(frame)
       } catch (error) {
         this.fail(error instanceof Error ? error : new Error(String(error)))
         return
@@ -60,14 +83,14 @@ export class BrowserNetworkTunnelStreamFrameDecoder {
       if (this.closed) {
         return
       }
-      offset = end
     }
-    this.retained = combined.slice(offset)
   }
 
   close(): void {
     this.closed = true
-    this.retained = new Uint8Array()
+    this.frame = null
+    this.frameBytes = 0
+    this.headerBytes = 0
   }
 
   private fail(error: Error): void {