/** * A range's chunks on the wire (REBUILD.md, prefetch): one streaming * response, a frame per chunk: from (8 bytes), to (8 bytes), length * (4 bytes), then the chunk as the news server sent it (compressed XOVER). */ export interface Frame { from: number to: number data: Buffer } const HEADER = 20 /** An empty frame, sent while the next chunk is still being fetched, so the response doesn't go quiet. */ export const KEEPALIVE: Frame = { from: 0, to: 0, data: Buffer.alloc(0) } export function isKeepalive(frame: Frame) { return frame.data.length === 0 && frame.from === 0 } export function encodeFrame(frame: Frame): Buffer { const head = Buffer.alloc(HEADER) head.writeBigUInt64BE(BigInt(frame.from), 0) head.writeBigUInt64BE(BigInt(frame.to), 8) head.writeUInt32BE(frame.data.length, 16) return Buffer.concat([head, frame.data]) } /** * 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 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) { 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 (length > 0) throw new Error(`Stream ended inside a frame (${length} bytes left)`) }