Rebuild: the worker joins stream pieces once per frame, gzips posts off its thread

decodeFrames joined each network piece (~64 KB) onto the frame so far:
a chunk of MBs copied over and over, 5.4% of a worker's time in an
end-to-end profile (the bench reads files, so it never showed). Pieces
are now joined once a frame is complete. A batch post's gzip (1.6%) runs
on a thread-pool thread.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

commands/edge_node.ts

@@ -4,11 +4,12 @@ import { setTimeout as sleep } from 'node:timers/promises'
 import { spawn, type ChildProcess } from 'node:child_process'
 import { availableParallelism } from 'node:os'
 import { promisify } from 'node:util'
-import { gzipSync, unzip } from 'node:zlib'
+import { gzip, unzip } from 'node:zlib'
 import { RangeIndexer, decodeFrames, isKeepalive, toWire, type EdgeBatch, type Lease } from '#lib/edge/index'
 
 const IDLE_MS = 30_000
 const unzipAsync = promisify(unzip)
+const gzipAsync = promisify(gzip)
 const RETRIES = 8
 
 class LeaseLost extends Error {}
@@ -219,8 +220,8 @@ class Api {
   }
 
   async batch(rangeId: number, batch: EdgeBatch): Promise<{ releases: number; candidates: number; bytes: number }> {
-    // Gzipped: chunk payloads are compressed already, the rest (names, fragments, candidates) compresses well.
-    const body = gzipSync(JSON.stringify(toWire(batch)), { level: 1 })
+    // Gzipped (on a thread-pool thread): chunk payloads are compressed already, the rest compresses well.
+    const body = await gzipAsync(JSON.stringify(toWire(batch)), { level: 1 })
     const response = await this.request(`ranges/${rangeId}/batches`, { method: 'POST', body, headers: { 'Content-Encoding': 'gzip' } })
     if (!response.ok) throw new Error(`batch: HTTP ${response.status}`)
     return { ...((await response.json()) as { releases: number; candidates: number }), bytes: body.length }

lib/edge/stream.ts

@@ -26,17 +26,33 @@ export function encodeFrame(frame: Frame): Buffer {
   return Buffer.concat([head, frame.data])
 }
 
-/** The frames of a byte stream, as they complete. */
+/**
+ * The frames of a byte stream, as they complete. Pieces are kept in a list
+ * and joined once a frame is complete: joining each piece onto what came
+ * before copied a chunk of MBs once per 64 KB (~5% of a worker's time;
+ * LOG.md 2026-10-09).
+ */
 export async function* decodeFrames(stream: AsyncIterable<Uint8Array> | Iterable<Uint8Array>): AsyncGenerator<Frame> {
-  let buffer = Buffer.alloc(0)
+  let pieces: Buffer[] = []
+  let length = 0
+  /** Bytes needed before the next frame can be cut: its header first, then all of it. */
+  let needed = HEADER
   for await (const piece of stream) {
-    buffer = buffer.length === 0 ? Buffer.from(piece) : Buffer.concat([buffer, piece])
-    while (buffer.length >= HEADER) {
-      const length = buffer.readUInt32BE(16)
-      if (buffer.length < HEADER + length) break
-      yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, HEADER + length) }
-      buffer = buffer.subarray(HEADER + length)
+    pieces.push(Buffer.isBuffer(piece) ? piece : Buffer.from(piece.buffer, piece.byteOffset, piece.byteLength))
+    length += piece.byteLength
+    while (length >= needed) {
+      let buffer = pieces.length === 1 ? pieces[0] : Buffer.concat(pieces, length)
+      pieces = [buffer]
+      if (needed === HEADER) {
+        needed = HEADER + buffer.readUInt32BE(16)
+        if (length < needed) break
+      }
+      yield { from: Number(buffer.readBigUInt64BE(0)), to: Number(buffer.readBigUInt64BE(8)), data: buffer.subarray(HEADER, needed) }
+      buffer = buffer.subarray(needed)
+      pieces = buffer.length > 0 ? [buffer] : []
+      length = buffer.length
+      needed = HEADER
     }
   }
-  if (buffer.length > 0) throw new Error(`Stream ended inside a frame (${buffer.length} bytes left)`)
+  if (length > 0) throw new Error(`Stream ended inside a frame (${length} bytes left)`)
 }

tests/unit/edge/stream.spec.ts

@@ -25,4 +25,15 @@ test.group('edge / stream frames', () => {
       }
     }, /inside a frame/)
   })
+  test('frames come out whole whatever the piece sizes, also pieces holding several frames', async ({ assert }) => {
+    const sent = Array.from({ length: 12 }, (_, i) => ({ from: i * 10 + 1, to: i * 10 + 10, data: Buffer.alloc((i * 997) % 5000, i) }))
+    const whole = Buffer.concat(sent.map(encodeFrame))
+    for (const size of [1, 3, 20, 21, 64, 1000, 4096, whole.length]) {
+      const pieces = []
+      for (let at = 0; at < whole.length; at += size) pieces.push(new Uint8Array(whole.subarray(at, at + size)))
+      const frames = []
+      for await (const frame of decodeFrames(pieces)) frames.push(frame)
+      assert.deepEqual(frames.map((f) => [f.from, f.to, f.data.toString('hex')]), sent.map((f) => [f.from, f.to, f.data.toString('hex')]), `pieces of ${size}`)
+    }
+  })
 })