From 5714995cce51beed94e71bf1a012fab12f07c7a2 Mon Sep 17 00:00:00 2001 From: azank Date: Tue, 13 Oct 2026 20:43:22 +0500 Subject: [PATCH] builder: copy pack bytes into blocks as they arrive, not keep the network chunks The receive loop kept every network chunk until a pack was whole: tens of thousands of 16-64 KB allocations, freed together at the end among allocations that live on, a heap glibc cannot give back. It is part of why a production instance sat at 1,171 MB resident with 22 MB live after google/site-kit-wp and google/filament were refused as too big. Pack bytes are now copied out of each chunk into blocks that double from 64 KB to 64 MB, joined once at the end, and the chunk goes as soon as it is read. The pkt-line parser keeps its place across chunks instead of joining them, so a length, side band or text line split anywhere still parses. Replayed locally from the same recorded GitHub responses, one repo after another in one process as a warm instance runs them, each turn opened with the last-resort collection (peak / after, MB): google/site-kit-wp (too big) 781 / 436 on main 705 / 364 here google/dawn 1,129 / 467 1,021 / 371 anomalyco/opencode 1,082 / 480 994 / 372 The records are byte for byte the same on main and here, and the times are the same. --- web/lib/engine/git.ts | 113 ++++++++++++++++++++++++++++-------------- web/test/git.test.ts | 61 +++++++++++++++++++---- 2 files changed, 127 insertions(+), 47 deletions(-) diff --git a/web/lib/engine/git.ts b/web/lib/engine/git.ts index ab60893..bf14bc9 100644 --- a/web/lib/engine/git.ts +++ b/web/lib/engine/git.ts @@ -97,66 +97,105 @@ export async function fetchPack(t: Transport, o: FetchOptions): Promise, o: FetchOptions, ac: AbortController): Promise { const shallow: string[] = [], unshallow: string[] = []; - const packChunks: Buffer[] = []; + const blocks: Buffer[] = []; + let block = Buffer.alloc(0), used = 0; let packBytes = 0; let section = ''; - let pending: Buffer = Buffer.alloc(0); let done = false; const maxBytes = o.maxBytes ?? Infinity; + // the pkt-line being read: its 4-digit length (which can be split between chunks), the payload bytes still to + // come, a packfile line's side band once its first byte is in, and the pieces of a text line + const head = Buffer.alloc(4); + let headLen = 0, left = 0, band = -1; + let text: Buffer[] = []; - const handle = (payload: Buffer | null, special: number) => { - if (payload === null) { // 0000 flush, 0001 delim, 0002 response end - if (special === 1) section = ''; - else done = true; - return; + const keep = (data: Buffer) => { + packBytes += data.length; + if (packBytes > maxBytes) { + ac.abort(); + throw new GitError(`pack exceeds ${Math.round(maxBytes / 1e6)} MB`, 'too_big'); } - if (section === 'packfile') { - const band = payload[0]; - if (band === 1) { - packBytes += payload.length - 1; - if (packBytes > maxBytes) { - ac.abort(); - throw new GitError(`pack exceeds ${Math.round(maxBytes / 1e6)} MB`, 'too_big'); - } - packChunks.push(payload.subarray(1)); - o.onProgress?.(packBytes); - } else if (band === 3) { - throw new GitError(`remote error: ${payload.subarray(1).toString('utf8').trim()}`, 'protocol'); + for (let i = 0; i < data.length;) { + if (used === block.length) { + if (used) blocks.push(block); + block = Buffer.allocUnsafeSlow(Math.min(MAX_BLOCK, Math.max(FIRST_BLOCK, block.length * 2))); + used = 0; } - return; + const n = data.copy(block, used, i); + used += n; + i += n; } - const line = payload.toString('utf8').replace(/\n$/, ''); - if (line.startsWith('ERR ')) { - const msg = line.slice(4); + o.onProgress?.(packBytes); + }; + + const line = (payload: Buffer) => { + const l = payload.toString('utf8').replace(/\n$/, ''); + if (l.startsWith('ERR ')) { + const msg = l.slice(4); throw new GitError(`remote error: ${msg}`, /no commits selected|not our ref|not found/i.test(msg) ? 'empty' : 'protocol'); } - if (!section) { section = line; return; } + if (!section) { section = l; return; } if (section === 'shallow-info') { - if (line.startsWith('shallow ')) shallow.push(line.slice(8)); - else if (line.startsWith('unshallow ')) unshallow.push(line.slice(10)); + if (l.startsWith('shallow ')) shallow.push(l.slice(8)); + else if (l.startsWith('unshallow ')) unshallow.push(l.slice(10)); } // acknowledgments / wanted-refs / packfile-uris: nothing to do with `done` sent }; for await (const chunk of body) { - pending = pending.length ? Buffer.concat([pending, Buffer.from(chunk)]) : Buffer.from(chunk); + const buf = Buffer.from(chunk.buffer, chunk.byteOffset, chunk.byteLength); let p = 0; - while (!done && pending.length - p >= 4) { - const len = parseInt(pending.toString('latin1', p, p + 4), 16); - if (Number.isNaN(len)) throw new GitError('malformed pkt-line', 'protocol'); - if (len < 4) { handle(null, len); p += 4; continue; } - if (pending.length - p < len) break; - handle(pending.subarray(p + 4, p + len), -1); - p += len; + while (p < buf.length && !done) { + if (!left) { + const n = buf.copy(head, headLen, p, Math.min(buf.length, p + 4 - headLen)); + p += n; + headLen += n; + if (headLen < 4) break; + headLen = 0; + const len = parseInt(head.toString('latin1'), 16); + if (Number.isNaN(len)) throw new GitError('malformed pkt-line', 'protocol'); + if (len < 4) { // 0000 flush, 0001 delim, 0002 response end + if (len === 1) section = ''; + else done = true; + } else { + left = len - 4; + band = -1; + if (!left && section !== 'packfile') line(Buffer.alloc(0)); + } + continue; + } + const n = Math.min(left, buf.length - p); + let part = buf.subarray(p, p + n); + p += n; + left -= n; + if (section !== 'packfile') { + text.push(Buffer.from(part)); + if (!left) { line(Buffer.concat(text)); text = []; } + continue; + } + if (band === -1) { band = part[0]; part = part.subarray(1); } + if (band === 1) { if (part.length) keep(part); } + else if (band === 3) { + text.push(Buffer.from(part)); + if (!left) throw new GitError(`remote error: ${Buffer.concat(text).toString('utf8').trim()}`, 'protocol'); + } } - pending = p < pending.length ? pending.subarray(p) : Buffer.alloc(0); if (done) break; } - if (!packChunks.length && section !== 'packfile') { + if (!packBytes && section !== 'packfile') { // git answers a deepen-since that selects no commits by ending the stream after shallow-info throw new GitError(`no packfile in response (last section: ${section || 'none'})`, o.deepenSince != null ? 'empty' : 'protocol'); } - return { pack: Buffer.concat(packChunks, packBytes), shallow, unshallow }; + blocks.push(block.subarray(0, used)); + return { pack: Buffer.concat(blocks, packBytes), shallow, unshallow }; } diff --git a/web/test/git.test.ts b/web/test/git.test.ts index 8a48d1a..e3a8b7a 100644 --- a/web/test/git.test.ts +++ b/web/test/git.test.ts @@ -1,19 +1,26 @@ -// The pack download: pkt-lines split anywhere across network chunks, and the progress the replay guard watches. +// The pack download: pkt-lines split anywhere across network chunks, the progress the replay guard watches, and +// the ways a download stops. import assert from 'node:assert/strict'; import { test } from 'node:test'; -import { DELIM, FLUSH, fetchPack, pkt, type Transport } from '../lib/engine/git'; +import { DELIM, FLUSH, fetchPack, GitError, pkt, type Transport } from '../lib/engine/git'; const SHA = 'a'.repeat(40); -/** A protocol-v2 fetch response: shallow-info, then the pack in side-band pkt-lines of the biggest size git sends. */ +/** One side-band pkt-line: band 1 carries pack bytes, band 3 an error. */ +function band(n: number, data: Buffer): Buffer { + return Buffer.concat([Buffer.from((data.length + 5).toString(16).padStart(4, '0')), Buffer.from([n]), data]); +} + +/** The packfile section: the pack in side-band pkt-lines of the biggest size git sends. */ +function packfile(pack: Buffer): Buffer[] { + const parts = [pkt('packfile\n')]; + for (let i = 0; i < pack.length; i += 65515) parts.push(band(1, pack.subarray(i, i + 65515))); + return parts; +} + +/** A protocol-v2 fetch response: shallow-info, then the pack. */ function response(pack: Buffer): Buffer { - const parts = [pkt('shallow-info\n'), pkt(`shallow ${SHA}\n`), DELIM, pkt('packfile\n')]; - for (let i = 0; i < pack.length; i += 65515) { - const data = pack.subarray(i, i + 65515); - parts.push(Buffer.from((data.length + 5).toString(16).padStart(4, '0')), Buffer.from([1]), data); - } - parts.push(FLUSH); - return Buffer.concat(parts); + return Buffer.concat([pkt('shallow-info\n'), pkt(`shallow ${SHA}\n`), DELIM, ...packfile(pack), FLUSH]); } /** Serves `body` in chunks of the given sizes, round and round: 3 bytes splits a pkt-line's length in two. */ @@ -47,3 +54,37 @@ test('a pack split anywhere across chunks arrives whole, and its progress counts assert.equal(seen.at(-1), pack.length); assert.ok(seen.every((n, i) => i === 0 || n > seen[i - 1])); }); + +const code = (c: string) => (e: unknown) => e instanceof GitError && e.code === c; + +test('a response a byte at a time: lengths, side bands and text lines split everywhere', async () => { + const pack = pattern(66_000); // two pkt-lines + const body = Buffer.concat([pkt('shallow-info\n'), pkt(`shallow ${SHA}\n`), pkt(`unshallow ${'b'.repeat(40)}\n`), DELIM, ...packfile(pack), FLUSH]); + const r = await fetchPack(chunked(body, [1]), { want: SHA }); + assert.ok(r.pack.equals(pack)); + assert.deepEqual(r.shallow, [SHA]); + assert.deepEqual(r.unshallow, ['b'.repeat(40)]); +}); + +test('an error on the side band, split across chunks, stops the fetch with its message', async () => { + const body = Buffer.concat([pkt('packfile\n'), band(1, pattern(1000)), band(3, Buffer.from('fatal: the remote end hung up\n')), FLUSH]); + await assert.rejects(fetchPack(chunked(body, [7]), { want: SHA }), + (e: unknown) => code('protocol')(e) && (e as Error).message === 'remote error: fatal: the remote end hung up'); +}); + +test('a pack over its budget stops with too_big and aborts the download', async () => { + let signal: AbortSignal | undefined; + const t = chunked(response(pattern(300_000)), [65536]); + const spy: Transport = { rpc: (b, s) => { signal = s; return t.rpc(b, s); } }; + await assert.rejects(fetchPack(spy, { want: SHA, maxBytes: 200_000 }), code('too_big')); + assert.equal(signal?.aborted, true); +}); + +test('an ERR line, and a response with no pack, say why', async () => { + const err = (m: string) => chunked(Buffer.concat([pkt(`ERR ${m}\n`), FLUSH]), [5]); + await assert.rejects(fetchPack(err(`upload-pack: not our ref ${SHA}`), { want: SHA }), code('empty')); + await assert.rejects(fetchPack(err('upload-pack: something else'), { want: SHA }), code('protocol')); + // git answers a deepen-since that selects no commits with shallow-info and no pack + const none = chunked(Buffer.concat([pkt('shallow-info\n'), pkt(`shallow ${SHA}\n`)]), [9]); + await assert.rejects(fetchPack(none, { want: SHA, deepenSince: 1 }), code('empty')); +});