perf(android): queue fragmented scrcpy video packets (#20230)

* perf(android): queue fragmented scrcpy video packets

* fix(android): release consumed scrcpy chunk storage

* fix(android): bound queued scrcpy fragment count

- Coalesce pending video fragments once more than MAX_PENDING_CHUNKS
  (1024) are queued, so a large frame delivered in tiny socket chunks
  cannot retain millions of Buffer objects below the 16 MiB byte guard.
- Add a regression test feeding a 256 KiB frame one byte at a time and
  asserting the retained fragment count stays bounded.

---------

Co-authored-by: m4air <m4air@Mac.localdomain>
Co-authored-by: Neil <neil@stably.ai>

config/scripts/scrcpy-frame-buffering-benchmark.mjs

@@ -0,0 +1,134 @@
+import assert from 'node:assert/strict'
+import { readFileSync } from 'node:fs'
+import { performance } from 'node:perf_hooks'
+import { transform } from 'esbuild'
+import { buildCounterbalancedSchedule } from './counterbalanced-benchmark-schedule.mjs'
+import { summarizeBenchmarkSamples } from './benchmark-sample-summary.mjs'
+
+// git show <ref>:src/main/emulator/android/scrcpy-video-frame-parser.ts | node config/scripts/scrcpy-frame-buffering-benchmark.mjs
+async function load(source) {
+  const { code } = await transform(source, { loader: 'ts', format: 'esm' })
+  return import(`data:text/javascript;base64,${Buffer.from(code).toString('base64')}`)
+}
+const before = (await load(readFileSync(0, 'utf8'))).parseScrcpyVideoFrames
+const after = (
+  await load(readFileSync('src/main/emulator/android/scrcpy-video-frame-parser.ts', 'utf8'))
+).parseScrcpyVideoFrames
+const { RelayFrameBuffer } = await load(readFileSync('src/shared/relay-frame-buffer.ts', 'utf8'))
+
+function reader(arm) {
+  if (arm === 'before') {
+    let pending = Buffer.alloc(0)
+    return {
+      read(chunk) {
+        // Match the baseline session's copy before invoking its production parser.
+        const result = before(Buffer.alloc(0), Buffer.concat([pending, chunk]))
+        pending = result.pending
+        return result.frames
+      },
+      pending: () => pending
+    }
+  }
+  const pending = new RelayFrameBuffer()
+  return {
+    read(chunk) {
+      // Include the session's mandatory ownership copy in the queued parser arm.
+      if (chunk.length > 0) {
+        pending.append(Buffer.from(chunk))
+      }
+      return after(pending)
+    },
+    pending: () => (pending.length > 0 ? pending.peek(pending.length) : Buffer.alloc(0))
+  }
+}
+
+function packet(size, meta = 123n) {
+  const frame = Buffer.alloc(size + 12, 7)
+  frame.writeBigUInt64BE(meta, 0)
+  frame.writeUInt32BE(size, 8)
+  return frame
+}
+
+let seed = 42
+const random = (max) => {
+  seed = (Math.imul(seed, 1664525) + 1013904223) >>> 0
+  return seed % max
+}
+let differentialChunks = 0
+for (let trial = 0; trial < 1000; trial += 1) {
+  const stream = Buffer.concat(
+    Array.from({ length: 1 + random(8) }, (_, index) =>
+      packet(random(256), (BigInt(random(4)) << 62n) | BigInt(index))
+    )
+  )
+  const oldReader = reader('before')
+  const newReader = reader('after')
+  for (let offset = 0; offset < stream.length;) {
+    const size = 1 + random(128)
+    const chunk = stream.subarray(offset, offset + size)
+    assert.deepEqual(newReader.read(chunk), oldReader.read(chunk))
+    assert.deepEqual(newReader.pending(), oldReader.pending())
+    assert.deepEqual(newReader.read(Buffer.alloc(0)), oldReader.read(Buffer.alloc(0)))
+    differentialChunks += 1
+    offset += size
+  }
+}
+
+const results = []
+for (const frameBytes of [32, 4096, 65_536, 1_048_576]) {
+  const frame = packet(frameBytes)
+  const expected = reader('before').read(frame)
+  for (const chunkBytes of new Set([frame.length, 65_536, 4096, 1024])) {
+    if (chunkBytes > frame.length) {
+      continue
+    }
+    const chunks = []
+    for (let offset = 0; offset < frame.length; offset += chunkBytes) {
+      chunks.push(frame.subarray(offset, offset + chunkBytes))
+    }
+    const iterations = Math.max(10, Math.floor(4_194_304 / frameBytes))
+    function run(arm, repeats) {
+      const parser = reader(arm)
+      let frames
+      const started = performance.now()
+      for (let iteration = 0; iteration < repeats; iteration += 1) {
+        for (const chunk of chunks) {
+          frames = parser.read(chunk)
+        }
+      }
+      const ms = performance.now() - started
+      assert.deepEqual(frames, expected)
+      assert.equal(parser.pending().length, 0)
+      return ms
+    }
+    run('before', iterations)
+    run('after', iterations)
+    /** @type {{ before: number[], after: number[] }} */
+    const samples = { before: [], after: [] }
+    for (const pair of buildCounterbalancedSchedule(8, 'before', 'after')) {
+      for (const arm of pair) {
+        samples[arm].push(run(arm, iterations))
+      }
+    }
+    results.push({
+      frameBytes,
+      chunkBytes,
+      iterations,
+      meanMicrosecondsPerFrame: Object.fromEntries(
+        Object.entries(samples).map(([arm, values]) => [
+          arm,
+          (values.reduce((sum, ms) => sum + ms, 0) * 1000) / values.length / iterations
+        ])
+      ),
+      before: summarizeBenchmarkSamples(samples.before),
+      after: summarizeBenchmarkSamples(samples.after)
+    })
+  }
+}
+console.log(
+  JSON.stringify(
+    { node: process.version, platform: process.platform, differentialChunks, results },
+    null,
+    2
+  )
+)

src/main/emulator/android/scrcpy-stream-session.test.ts

@@ -0,0 +1,172 @@
+import { EventEmitter } from 'node:events'
+import { beforeEach, describe, expect, it, vi } from 'vitest'
+import { ScrcpyStreamSession } from './scrcpy-stream-session'
+
+const io = vi.hoisted(() => ({ spawn: vi.fn(), connect: vi.fn() }))
+vi.mock('node:child_process', () => ({ spawn: io.spawn }))
+vi.mock('node:net', () => ({ connect: io.connect }))
+vi.mock('../emulator-probe', () => ({ emulatorProbe: vi.fn(), emulatorProbeError: vi.fn() }))
+
+class TestSocket extends EventEmitter {
+  destroy = vi.fn()
+  setTimeout = vi.fn()
+}
+
+function packet(size: number, meta = 123n): Buffer {
+  const result = Buffer.alloc(12 + size, 7)
+  result.writeBigUInt64BE(meta, 0)
+  result.writeUInt32BE(size, 8)
+  return result
+}
+
+function handshake(): Buffer {
+  const result = Buffer.alloc(77)
+  result.write('test-device', 1)
+  result.write('h264', 65)
+  result.writeUInt32BE(1080, 69)
+  result.writeUInt32BE(2400, 73)
+  return result
+}
+
+async function startSession() {
+  const video = new TestSocket()
+  const control = new TestSocket()
+  const server = Object.assign(new EventEmitter(), {
+    stdout: new EventEmitter(),
+    stderr: new EventEmitter(),
+    kill: vi.fn()
+  })
+  io.spawn.mockReturnValue(server)
+  io.connect.mockReturnValueOnce(video).mockReturnValueOnce(control)
+  const callbacks = { onMeta: vi.fn(), onFrame: vi.fn(), onError: vi.fn(), onClose: vi.fn() }
+  const runner = vi.fn().mockResolvedValue({ code: 0, stdout: '', stderr: '' })
+  const started = ScrcpyStreamSession.start(
+    {
+      runner,
+      sdk: { sdkRoot: 'sdk', adb: 'adb', emulator: 'emulator', avdmanager: 'avdmanager' },
+      serial: 'test-device',
+      localJarPath: 'server.jar',
+      localPort: 12345
+    },
+    callbacks
+  )
+  await vi.waitFor(() => expect(io.connect).toHaveBeenCalledTimes(1))
+  return { video, control, server, callbacks, started }
+}
+
+beforeEach(() => {
+  io.spawn.mockReset()
+  io.connect.mockReset()
+})
+
+describe('ScrcpyStreamSession video buffering', () => {
+  it('accepts bytewise handshake and frames, including empty chunks and empty frames', async () => {
+    const { video, callbacks, started } = await startSession()
+    const header = handshake()
+    for (let index = 0; index < header.length - 1; index += 1) {
+      video.emit('data', header.subarray(index, index + 1))
+    }
+    expect(callbacks.onMeta).not.toHaveBeenCalled()
+    video.emit('data', header.subarray(-1))
+    const session = await started
+    expect(callbacks.onMeta).toHaveBeenCalledWith({ codecId: 'h264', width: 1080, height: 2400 })
+    const stream = Buffer.concat([packet(0, 1n << 63n), packet(3, (1n << 62n) | 5n)])
+    for (const byte of stream) {
+      video.emit('data', Buffer.alloc(0))
+      video.emit('data', Buffer.from([byte]))
+    }
+    expect(callbacks.onFrame.mock.calls.map(([frame]) => frame)).toEqual([
+      { config: true, keyFrame: false, pts: 0n, data: Buffer.alloc(0) },
+      { config: false, keyFrame: true, pts: 5n, data: Buffer.alloc(3, 7) }
+    ])
+    session.close()
+  })
+
+  it('owns pending bytes and emitted frames independently of input chunks', async () => {
+    const { video, callbacks, started } = await startSession()
+    video.emit('data', handshake())
+    const session = await started
+    const first = packet(4)
+    const second = packet(6, 456n)
+    const chunk = Buffer.concat([first, second.subarray(0, 14)])
+    video.emit('data', chunk)
+    chunk.fill(0)
+    const tail = Buffer.from(second.subarray(14))
+    video.emit('data', tail)
+    tail.fill(0)
+    expect(callbacks.onFrame.mock.calls.map(([frame]) => frame)).toEqual([
+      { config: false, keyFrame: false, pts: 123n, data: Buffer.alloc(4, 7) },
+      { config: false, keyFrame: false, pts: 456n, data: Buffer.alloc(6, 7) }
+    ])
+    session.close()
+  })
+
+  it('emits initial metadata and frames before startup resolves', async () => {
+    const { video, callbacks, started } = await startSession()
+    const events: string[] = []
+    callbacks.onMeta.mockImplementation(() => events.push('meta'))
+    callbacks.onFrame.mockImplementation(() => events.push('frame'))
+    const ready = started.then((session) => {
+      events.push('ready')
+      return session
+    })
+    video.emit('data', Buffer.concat([handshake(), packet(3), packet(5)]))
+    expect(events).toEqual(['meta', 'frame', 'frame'])
+    const session = await ready
+    expect(events).toEqual(['meta', 'frame', 'frame', 'ready'])
+    session.close()
+  })
+
+  it('fails an already started session on a corrupt batch without delivering partial results', async () => {
+    const { video, callbacks, started } = await startSession()
+    video.emit('data', handshake())
+    const session = await started
+    const corrupt = packet(0)
+    corrupt.writeUInt32BE(16 * 1024 * 1024 + 1, 8)
+    video.emit('data', Buffer.concat([packet(1), corrupt]))
+    expect(callbacks.onFrame).not.toHaveBeenCalled()
+    expect(callbacks.onError).toHaveBeenCalledExactlyOnceWith(expect.stringMatching(/desynced/))
+    expect(callbacks.onClose).toHaveBeenCalledTimes(1)
+    session.close()
+    expect(callbacks.onClose).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not resolve startup or deliver earlier frames if the first batch is desynced', async () => {
+    const { video, callbacks, started, server } = await startSession()
+    const corrupt = packet(0)
+    corrupt.writeUInt32BE(16 * 1024 * 1024 + 1, 8)
+    const rejected = expect(started).rejects.toThrow(/desynced/)
+    video.emit('data', Buffer.concat([handshake(), packet(1), corrupt]))
+    await rejected
+    expect(callbacks.onMeta).toHaveBeenCalledTimes(1)
+    expect(callbacks.onFrame).not.toHaveBeenCalled()
+    expect(callbacks.onError).toHaveBeenCalledTimes(1)
+    expect(callbacks.onClose).toHaveBeenCalledTimes(1)
+    expect(server.kill).toHaveBeenCalledTimes(1)
+    expect(video.destroy).toHaveBeenCalledTimes(1)
+  })
+
+  it('does not repeatedly concatenate a growing fragmented frame', async () => {
+    const { video, callbacks, started } = await startSession()
+    video.emit('data', handshake())
+    const session = await started
+    const frame = packet(1024 * 1024)
+    const concat = Buffer.concat
+    let concatenatedBytes = 0
+    const spy = vi.spyOn(Buffer, 'concat').mockImplementation((buffers, length) => {
+      concatenatedBytes += length ?? buffers.reduce((sum, part) => sum + part.length, 0)
+      return concat(buffers, length)
+    })
+    try {
+      for (let offset = 0; offset < frame.length; offset += 4096) {
+        video.emit('data', frame.subarray(offset, offset + 4096))
+      }
+    } finally {
+      spy.mockRestore()
+      session.close()
+    }
+    expect(callbacks.onFrame).toHaveBeenCalledTimes(1)
+    expect(callbacks.onFrame.mock.calls[0][0].data).toEqual(frame.subarray(12))
+    expect(concatenatedBytes).toBeLessThanOrEqual(frame.length * 2)
+  })
+})

src/main/emulator/android/scrcpy-stream-session.ts

@@ -1,6 +1,7 @@
 import { spawn, type ChildProcess } from 'node:child_process'
 import { connect, type Socket } from 'node:net'
 import { randomBytes } from 'node:crypto'
+import { RelayFrameBuffer } from '../../../shared/relay-frame-buffer'
 import type { AndroidCommandRunner } from './android-command-runner'
 import type { AndroidSdkPaths } from './android-sdk-discovery'
 import { ensureAdbOk } from './android-adb-result'
@@ -14,7 +15,6 @@ import {
 import {
   parseScrcpyVideoFrames,
   parseScrcpyVideoMeta,
-  type ScrcpyFrameParseResult,
   type ScrcpyVideoFrame,
   type ScrcpyVideoMeta
 } from './scrcpy-video-frame-parser'
@@ -57,7 +57,7 @@ export class ScrcpyStreamSession {
   private server: ChildProcess | null = null
   private videoSocket: Socket | null = null
   private controlSocket: Socket | null = null
-  private pendingVideo: Buffer = Buffer.alloc(0)
+  private readonly pendingVideo = new RelayFrameBuffer()
   private metaSeen = false
   private headerStripped = false
   private closed = false
@@ -206,47 +206,48 @@ export class ScrcpyStreamSession {
   }
 
   private handleVideoChunk(chunk: Buffer): void {
-    let buffer = Buffer.concat([this.pendingVideo, chunk])
+    const buffer = this.pendingVideo
+    if (chunk.length > 0) {
+      // Socket chunks and emitted frames must not share mutable pending storage.
+      buffer.append(Buffer.from(chunk))
+    }
     // The first socket carries a 1-byte readiness marker + the 64-byte device name.
     if (!this.headerStripped) {
       const headerLen = DUMMY_BYTE + DEVICE_NAME_BYTES
       if (buffer.length < headerLen) {
-        this.pendingVideo = buffer
         return
       }
-      buffer = Buffer.from(buffer.subarray(headerLen))
+      buffer.discard(headerLen)
       this.headerStripped = true
     }
     let shouldResolveReady = false
     if (!this.metaSeen) {
-      const meta = parseScrcpyVideoMeta(buffer)
-      if (!meta) {
-        this.pendingVideo = buffer
+      if (buffer.length < 12) {
         return
       }
+      const meta = parseScrcpyVideoMeta(buffer.peek(12))!
       this.metaSeen = true
       emulatorProbe('scrcpy.meta', meta)
       this.callbacks.onMeta(meta)
       shouldResolveReady = true
-      buffer = Buffer.from(buffer.subarray(12))
+      buffer.discard(12)
     }
     // The parser throws on a desynced stream (e.g. an absurd frame size); catch
     // it here so it fails the session via the normal teardown path rather than
     // surfacing as an unhandled exception in this socket 'data' listener.
-    let result: ScrcpyFrameParseResult
+    let frames: ScrcpyVideoFrame[]
     try {
-      result = parseScrcpyVideoFrames(Buffer.alloc(0), buffer)
+      frames = parseScrcpyVideoFrames(buffer)
     } catch (error) {
       this.fail(error instanceof Error ? error.message : String(error))
       return
     }
-    this.pendingVideo = result.pending
     if (shouldResolveReady) {
       this.resolveReady?.()
       this.resolveReady = null
       this.rejectReady = null
     }
-    for (const frame of result.frames) {
+    for (const frame of frames) {
       this.callbacks.onFrame(frame)
     }
   }

src/main/emulator/android/scrcpy-video-frame-parser.test.ts

@@ -1,5 +1,10 @@
 import { describe, expect, it } from 'vitest'
-import { parseScrcpyVideoFrames, parseScrcpyVideoMeta } from './scrcpy-video-frame-parser'
+import { RelayFrameBuffer } from '../../../shared/relay-frame-buffer'
+import {
+  MAX_PENDING_CHUNKS,
+  parseScrcpyVideoFrames,
+  parseScrcpyVideoMeta
+} from './scrcpy-video-frame-parser'
 
 const CONFIG = 1n << 63n
 const KEY = 1n << 62n
@@ -28,7 +33,9 @@ describe('parseScrcpyVideoMeta', () => {
 describe('parseScrcpyVideoFrames', () => {
   it('extracts config and key frames with their flags and data', () => {
     const stream = Buffer.concat([frame(CONFIG, [0, 0, 0, 1]), frame(KEY | 123n, [1, 2, 3])])
-    const { frames, pending } = parseScrcpyVideoFrames(Buffer.alloc(0), stream)
+    const pending = new RelayFrameBuffer()
+    pending.append(stream)
+    const frames = parseScrcpyVideoFrames(pending)
     expect(pending.length).toBe(0)
     expect(frames).toHaveLength(2)
     expect(frames[0]).toMatchObject({ config: true, keyFrame: false })
@@ -39,19 +46,89 @@ describe('parseScrcpyVideoFrames', () => {
 
   it('buffers a partial frame across chunks', () => {
     const full = frame(5n, [9, 9, 9, 9])
-    const r1 = parseScrcpyVideoFrames(Buffer.alloc(0), full.subarray(0, 14))
-    expect(r1.frames).toHaveLength(0)
-    expect(r1.pending.length).toBe(14)
-    const r2 = parseScrcpyVideoFrames(r1.pending, full.subarray(14))
-    expect(r2.frames).toHaveLength(1)
-    expect([...r2.frames[0].data]).toEqual([9, 9, 9, 9])
-    expect(r2.pending.length).toBe(0)
+    const pending = new RelayFrameBuffer()
+    pending.append(full.subarray(0, 14))
+    expect(parseScrcpyVideoFrames(pending)).toHaveLength(0)
+    expect(pending.length).toBe(14)
+    pending.append(full.subarray(14))
+    const frames = parseScrcpyVideoFrames(pending)
+    expect(frames).toHaveLength(1)
+    expect([...frames[0].data]).toEqual([9, 9, 9, 9])
+    expect(pending.length).toBe(0)
+  })
+
+  it('does not retain a consumed large packet behind a one-byte pending suffix', () => {
+    const first = Buffer.alloc(4 * 1024 * 1024 + 12, 7)
+    first.writeBigUInt64BE(123n, 0)
+    first.writeUInt32BE(first.length - 12, 8)
+    const second = frame(KEY | 456n, [1, 2, 3])
+    const chunk = Buffer.concat([first, second.subarray(0, 1)])
+    const pending = new RelayFrameBuffer()
+    pending.append(chunk)
+
+    const frames = parseScrcpyVideoFrames(pending)
+    expect(frames).toHaveLength(1)
+    expect(frames[0]).toMatchObject({ config: false, keyFrame: false, pts: 123n })
+    expect(frames[0].data.equals(first.subarray(12))).toBe(true)
+    expect(pending.length).toBe(1)
+    expect(pending.peek(1)).toEqual(second.subarray(0, 1))
+    expect(pending.peek(1).buffer === chunk.buffer).toBe(false)
+    expect(pending.peek(1).buffer.byteLength).toBeLessThan(chunk.length)
+
+    chunk.fill(0xff)
+    pending.append(second.subarray(1))
+    expect(parseScrcpyVideoFrames(pending)).toEqual([
+      { config: false, keyFrame: true, pts: 456n, data: Buffer.from([1, 2, 3]) }
+    ])
+    expect(pending.length).toBe(0)
+  })
+
+  it('keeps mostly live chunk storage instead of recopying a large pending frame', () => {
+    const first = frame(123n, [1, 2, 3])
+    const second = Buffer.alloc(4 * 1024 * 1024 + 12, 7)
+    second.writeBigUInt64BE(KEY | 456n, 0)
+    second.writeUInt32BE(second.length - 12, 8)
+    const split = 3 * 1024 * 1024
+    const chunk = Buffer.concat([first, second.subarray(0, split)])
+    const pending = new RelayFrameBuffer()
+    pending.append(chunk)
+
+    expect(parseScrcpyVideoFrames(pending)).toHaveLength(1)
+    expect(pending.length).toBe(split)
+    expect(pending.peek(1).buffer === chunk.buffer).toBe(true)
+
+    pending.append(second.subarray(split))
+    const frames = parseScrcpyVideoFrames(pending)
+    expect(frames).toHaveLength(1)
+    expect(frames[0]).toMatchObject({ config: false, keyFrame: true, pts: 456n })
+    expect(frames[0].data.equals(second.subarray(12))).toBe(true)
+    expect(pending.length).toBe(0)
+  })
+
+  it('bounds queued fragment count for a large frame delivered one byte at a time', () => {
+    const full = Buffer.alloc(256 * 1024 + 12, 7)
+    full.writeBigUInt64BE(KEY | 789n, 0)
+    full.writeUInt32BE(full.length - 12, 8)
+    const pending = new RelayFrameBuffer()
+    let maxChunks = 0
+    let frames: ReturnType<typeof parseScrcpyVideoFrames> = []
+    for (const byte of full) {
+      pending.append(Buffer.from([byte]))
+      frames = parseScrcpyVideoFrames(pending)
+      maxChunks = Math.max(maxChunks, pending.chunkCount)
+    }
+    expect(maxChunks).toBeLessThanOrEqual(MAX_PENDING_CHUNKS)
+    expect(frames).toHaveLength(1)
+    expect(frames[0]).toMatchObject({ config: false, keyFrame: true, pts: 789n })
+    expect(frames[0].data.equals(full.subarray(12))).toBe(true)
+    expect(pending.length).toBe(0)
   })
 
   it('holds an incomplete header until more bytes arrive', () => {
-    const result = parseScrcpyVideoFrames(Buffer.alloc(0), Buffer.from([0, 1, 2]))
-    expect(result.frames).toHaveLength(0)
-    expect(result.pending.length).toBe(3)
+    const pending = new RelayFrameBuffer()
+    pending.append(Buffer.from([0, 1, 2]))
+    expect(parseScrcpyVideoFrames(pending)).toHaveLength(0)
+    expect(pending.length).toBe(3)
   })
 
   it('throws on a desynced frame size instead of buffering toward OOM', () => {
@@ -59,6 +136,8 @@ describe('parseScrcpyVideoFrames', () => {
     // never be satisfied, leaving the whole buffer pending forever.
     const header = Buffer.alloc(12)
     header.writeUInt32BE(64 * 1024 * 1024, 8)
-    expect(() => parseScrcpyVideoFrames(Buffer.alloc(0), header)).toThrow(/desynced/)
+    const pending = new RelayFrameBuffer()
+    pending.append(header)
+    expect(() => parseScrcpyVideoFrames(pending)).toThrow(/desynced/)
   })
 })

src/main/emulator/android/scrcpy-video-frame-parser.ts

@@ -3,11 +3,15 @@
 // socket. The socket reader (scrcpy-stream-session) feeds chunks here; this file
 // has no I/O so the framing is unit-testable.
 
+import type { RelayFrameBuffer } from '../../../shared/relay-frame-buffer'
+
 const FRAME_HEADER_SIZE = 12
 const CODEC_META_SIZE = 12
 // scrcpy frames are well under this at the configured max_size; a larger
 // size means a desynced stream — fail fast instead of buffering toward OOM.
 const MAX_FRAME_BYTES = 16 * 1024 * 1024
+// Caps per-object overhead when a socket delivers one frame as many tiny chunks.
+export const MAX_PENDING_CHUNKS = 1024
 // Top two bits of the 64-bit PTS field carry packet flags.
 const CONFIG_FLAG = 1n << 63n
 const KEY_FRAME_FLAG = 1n << 62n
@@ -47,33 +51,38 @@ export type ScrcpyVideoFrame = {
   data: Buffer
 }
 
-export type ScrcpyFrameParseResult = { frames: ScrcpyVideoFrame[]; pending: Buffer }
-
-// Extracts complete frames from `pending + chunk`, returning the leftover bytes
-// of any partially-received frame so the caller can prepend them to the next chunk.
-export function parseScrcpyVideoFrames(pending: Buffer, chunk: Buffer): ScrcpyFrameParseResult {
-  const buffer = pending.length > 0 ? Buffer.concat([pending, chunk]) : chunk
+// Leave partial frames queued so fragmented payloads are not recopied on every chunk.
+export function parseScrcpyVideoFrames(buffer: RelayFrameBuffer): ScrcpyVideoFrame[] {
   const frames: ScrcpyVideoFrame[] = []
-  let offset = 0
 
-  while (buffer.length - offset >= FRAME_HEADER_SIZE) {
-    const meta = buffer.readBigUInt64BE(offset)
-    const size = buffer.readUInt32BE(offset + 8)
+  while (buffer.length >= FRAME_HEADER_SIZE) {
+    const header = buffer.peek(FRAME_HEADER_SIZE)
+    const meta = header.readBigUInt64BE(0)
+    const size = header.readUInt32BE(8)
     if (size > MAX_FRAME_BYTES) {
       throw new Error(`scrcpy frame size ${size} exceeds ${MAX_FRAME_BYTES}; stream desynced`)
     }
-    const dataStart = offset + FRAME_HEADER_SIZE
-    if (buffer.length - dataStart < size) {
+    if (buffer.length < FRAME_HEADER_SIZE + size) {
       break
     }
+    const packet = buffer.take(FRAME_HEADER_SIZE + size)
     frames.push({
       config: (meta & CONFIG_FLAG) !== 0n,
       keyFrame: (meta & KEY_FRAME_FLAG) !== 0n,
       pts: meta & PTS_MASK,
-      data: Buffer.from(buffer.subarray(dataStart, dataStart + size))
+      data: Buffer.from(packet.subarray(FRAME_HEADER_SIZE))
     })
-    offset = dataStart + size
   }
 
-  return { frames, pending: offset > 0 ? Buffer.from(buffer.subarray(offset)) : buffer }
+  if (frames.length > 0 && buffer.length > 0) {
+    const pendingHead = buffer.peek(1)
+    // Compact only mostly consumed allocations larger than the reusable Buffer slab.
+    if (pendingHead.buffer.byteLength > Math.max(Buffer.poolSize, pendingHead.length * 2)) {
+      buffer.append(Buffer.from(buffer.drain()))
+    }
+  }
+  if (buffer.chunkCount > MAX_PENDING_CHUNKS) {
+    buffer.append(buffer.drain())
+  }
+  return frames
 }

src/shared/relay-frame-buffer.ts

@@ -7,6 +7,10 @@ export class RelayFrameBuffer {
     return this.bytes
   }
 
+  get chunkCount(): number {
+    return this.chunks.length - this.head
+  }
+
   append(chunk: Buffer): void {
     this.chunks.push(chunk)
     this.bytes += chunk.length