diff --git a/commands/edge_node.ts b/commands/edge_node.ts index f258122..55ca96f 100644 --- a/commands/edge_node.ts +++ b/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 } diff --git a/lib/edge/stream.ts b/lib/edge/stream.ts index 663342d..9a1f7ec 100644 --- a/lib/edge/stream.ts +++ b/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 | Iterable): AsyncGenerator { - 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)`) } diff --git a/tests/unit/edge/stream.spec.ts b/tests/unit/edge/stream.spec.ts index 1f64db7..b1ce1b8 100644 --- a/tests/unit/edge/stream.spec.ts +++ b/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}`) + } + }) })