diff --git a/.changeset/git-index-memory.md b/.changeset/git-index-memory.md new file mode 100644 index 0000000..bc156a6 --- /dev/null +++ b/.changeset/git-index-memory.md @@ -0,0 +1,19 @@ +--- +'@pdsjs/git': patch +--- + +Indexing a pack keeps only the objects the caller asked for. `indexPacks` +held every object it materialized, which costs the whole inflated graph: +bluesky-social/atproto holds 1.3 GB of objects across 158,213 entries, and +the walk peaked at 3.0 GB of resident memory. + +A `retain` option names what the returned map keeps. The index builder asks +for 'none', and the write guard's ancestry check asks for 'commits', because +it walks commit parents alone. Both modes hold a bounded cache of delta bases +and build an evicted base again from the pack bytes. `cacheBytes` sets that +bound for a runtime with tighter memory. Only 4% of that repository's objects +are ever wanted as a base at one time, so the same walk peaks at 1.4 GB and +runs 15% faster. + +`indexPacks` also names objects in batches of 512. One batch for each pack +held a second copy of every object in it. diff --git a/packages/git/src/objects.js b/packages/git/src/objects.js index 2b9768f..130d826 100644 --- a/packages/git/src/objects.js +++ b/packages/git/src/objects.js @@ -28,6 +28,26 @@ const TYPE_NAMES = { [OBJ_TAG]: 'tag', }; +/** How many objects indexPacks names in one digest batch. */ +const HASH_BATCH = 512; + +/** + * How many bytes of materialized objects indexPacks keeps as delta bases + * when it is not retaining the whole graph. A base evicted below this line + * is built again from the pack, so the number trades work for memory. + */ +const CACHE_BYTES = 64 * 1024 * 1024; + +/** + * Whether an entry is part of the commit graph, which is what a caller + * walking history keeps. + * @param {PackEntryMeta} meta + * @returns {boolean} + */ +function isHistory(meta) { + return meta.type === 'commit' || meta.type === 'tag'; +} + const decoder = new TextDecoder(); /** @@ -227,16 +247,33 @@ export function readEntryHeader(bytes) { * a ref-delta base absent from the given packs. An incremental bundle may * delta against its prerequisites, which a caller holding the full * repository can materialize. - * @returns {Promise>} objects by SHA-1 id + * @param {{retain?: 'all'|'commits'|'none', cacheBytes?: number}} [options] - + * `retain` is which objects the returned map holds. A caller that only + * walks history asks for 'commits'; one that only wants the collector asks + * for 'none'. Both cost a bounded amount of memory over the pack itself, + * where 'all' costs the whole materialized object graph. `cacheBytes` + * bounds the delta bases those two modes keep, for a runtime whose memory + * is tighter than the default. + * @returns {Promise>} the retained objects by SHA-1 id */ -export async function indexPacks(packs, onObject, resolveBase) { +export async function indexPacks(packs, onObject, resolveBase, options) { + const retain = options?.retain ?? 'all'; + /** Objects the caller asked to keep. */ /** @type {Map} */ const bySha = new Map(); + /** + * Where each named object sits, which is what lets an evicted delta base + * be built again from the pack bytes. + * @type {Map} + */ + const placeBySha = new Map(); const digest = await platformDigest(); /** * Objects walked but not yet named. Hashing them together lets the - * platform digest run over the whole batch at once. + * platform digest run over the whole batch at once. The batch is bounded: + * flush copies every object it holds into a header-prefixed buffer, so a + * batch the size of a pack costs a second copy of that pack. * @type {Array<{object: GitObject, meta: PackEntryMeta}>} */ let pending = []; @@ -246,7 +283,10 @@ export async function indexPacks(packs, onObject, resolveBase) { * @param {{object: GitObject, meta: PackEntryMeta}} entry */ function emit(sha, entry) { - bySha.set(sha, entry.object); + if (retain === 'all' || (retain === 'commits' && isHistory(entry.meta))) { + bySha.set(sha, entry.object); + } + placeBySha.set(sha, { pack: entry.meta.pack, offset: entry.meta.offset }); onObject?.(sha, entry.meta); } @@ -274,19 +314,138 @@ export async function indexPacks(packs, onObject, resolveBase) { for (let i = 0; i < batch.length; i++) emit(ids[i], batch[i]); } + /** + * Materialized entries, holding what the deltas still to come need. + * Retaining every object costs the whole inflated graph, so 'commits' and + * 'none' bound this and build an evicted base again from the pack bytes. + * A Map iterates in insertion order, and a hit reinserts, so the first key + * is the least recently used one. + * @type {Map} + */ + const cache = new Map(); + const cacheLimit = + retain === 'all' + ? Number.POSITIVE_INFINITY + : (options?.cacheBytes ?? CACHE_BYTES); + let cacheBytes = 0; + + /** + * One key across every pack. A pack offset is below 2^32. + * @param {number} p + * @param {number} offset + */ + const cacheKey = (p, offset) => p * 2 ** 32 + offset; + + /** + * @param {number} p + * @param {number} offset + * @param {GitObject} object + */ + function cachePut(p, offset, object) { + const key = cacheKey(p, offset); + cache.set(key, object); + cacheBytes += object.data.length; + while (cacheBytes > cacheLimit && cache.size > 1) { + const oldest = cache.keys().next().value; + if (oldest === undefined || oldest === key) break; + cacheBytes -= /** @type {GitObject} */ (cache.get(oldest)).data.length; + cache.delete(oldest); + } + } + + /** + * @param {number} p + * @param {number} offset + * @returns {GitObject|undefined} + */ + function cacheGet(p, offset) { + const key = cacheKey(p, offset); + const object = cache.get(key); + if (object === undefined) return undefined; + cache.delete(key); + cache.set(key, object); + return object; + } + + /** + * The entry at this offset, inflated and with its deltas applied. A base + * still cached costs nothing; an evicted one is built again from the pack, + * down its whole delta chain. + * @param {number} p + * @param {number} offset + * @returns {Promise} + */ + async function materialize(p, offset) { + const cached = cacheGet(p, offset); + if (cached) return cached; + + /** The chain from the wanted entry down to its deepest missing base. */ + /** @type {Array<{pack: number, offset: number}>} */ + const chain = []; + /** @type {GitObject|undefined} */ + let base; + let at = { pack: p, offset }; + for (;;) { + chain.push(at); + const header = readEntryHeader(packs[at.pack].subarray(at.offset)); + if (header.baseRel !== null) { + const next = { pack: at.pack, offset: at.offset - header.baseRel }; + const hit = cacheGet(next.pack, next.offset); + if (hit) { + base = hit; + break; + } + at = next; + continue; + } + if (header.baseSha !== null) { + base = await baseBySha(header.baseSha); + break; + } + break; + } + + for (let i = chain.length - 1; i >= 0; i--) { + const { pack: cp, offset: co } = chain[i]; + const header = readEntryHeader(packs[cp].subarray(co)); + const { data } = inflateAt(packs[cp], co + header.dataStart); + if (header.baseRel === null && header.baseSha === null) { + base = { type: header.type, data }; + } else { + if (!base) throw new Error(`delta base for entry ${co} not found`); + base = { type: base.type, data: applyDelta(data, base.data) }; + } + cachePut(cp, co, base); + } + return /** @type {GitObject} */ (base); + } + + /** + * @param {string} sha + * @returns {Promise} + */ + async function baseBySha(sha) { + let place = placeBySha.get(sha); + if (!place) { + // The base is usually an object from an earlier pack, already named. + // A pack that ref-deltas against itself needs this one named first. + await flush(); + place = placeBySha.get(sha); + } + if (place) return await materialize(place.pack, place.offset); + if (resolveBase) return (await resolveBase(sha)) ?? undefined; + return undefined; + } + for (let packIndex = 0; packIndex < packs.length; packIndex++) { const pack = packs[packIndex]; const { count } = splitPack(pack); - /** @type {Map} */ - const byOffset = new Map(); let offset = 12; for (let i = 0; i < count; i++) { const entryStart = offset; - let byte = pack[offset++]; - const typeCode = (byte >> 4) & 0x07; - // The size in the header is the inflated size; the boundary walk does - // not need it, but the varint must still be consumed. - while (byte & 0x80) byte = pack[offset++]; + const header = readEntryHeader(pack.subarray(entryStart)); + const { data, end } = inflateAt(pack, entryStart + header.dataStart); + offset = end; /** @type {GitObject} */ let object; @@ -294,46 +453,22 @@ export async function indexPacks(packs, onObject, resolveBase) { let baseOffset = null; /** @type {string|null} */ let baseSha = null; - if (typeCode === OBJ_OFS_DELTA) { - byte = pack[offset++]; - let rel = byte & 0x7f; - while (byte & 0x80) { - byte = pack[offset++]; - rel = ((rel + 1) << 7) | (byte & 0x7f); - } - const { data, end } = inflateAt(pack, offset); - offset = end; - baseOffset = entryStart - rel; - const base = byOffset.get(baseOffset); - if (!base) throw new Error('ofs-delta base not found'); + if (header.baseRel !== null) { + baseOffset = entryStart - header.baseRel; + const base = + cacheGet(packIndex, baseOffset) ?? + (await materialize(packIndex, baseOffset)); object = { type: base.type, data: applyDelta(data, base.data) }; - } else if (typeCode === OBJ_REF_DELTA) { - baseSha = Array.from(pack.subarray(offset, offset + 20)) - .map((b) => b.toString(16).padStart(2, '0')) - .join(''); - offset += 20; - const { data, end } = inflateAt(pack, offset); - offset = end; - let base = bySha.get(baseSha); - if (!base) { - // The base is usually an object from an earlier pack, already - // named. A pack that ref-deltas against itself needs this one - // named first. - await flush(); - base = bySha.get(baseSha); - } - if (!base && resolveBase) - base = (await resolveBase(baseSha)) ?? undefined; + } else if (header.baseSha !== null) { + baseSha = header.baseSha; + const base = await baseBySha(baseSha); if (!base) throw new Error(`ref-delta base ${baseSha} not found`); object = { type: base.type, data: applyDelta(data, base.data) }; } else { - const typeName = TYPE_NAMES[typeCode]; - if (!typeName) throw new Error(`unknown pack entry type ${typeCode}`); - const { data, end } = inflateAt(pack, offset); - offset = end; - object = { type: typeName, data }; + object = { type: header.type, data }; } - byOffset.set(entryStart, object); + cachePut(packIndex, entryStart, object); + if (pending.length >= HASH_BATCH) await flush(); pending.push({ object, meta: { diff --git a/packages/git/src/pack-index.js b/packages/git/src/pack-index.js index 2aafc57..90b3a91 100644 --- a/packages/git/src/pack-index.js +++ b/packages/git/src/pack-index.js @@ -88,7 +88,9 @@ export async function buildBundleIndex(bundleBytes, resolveBase) { type: meta.type, }); }; - await indexPacks([pack], collect, resolveBase); + // The index is the whole product here, so no object outlives its use as a + // delta base. + await indexPacks([pack], collect, resolveBase, { retain: 'none' }); entries.sort((a, b) => a.offset - b.offset); return { packStart, entries }; } diff --git a/packages/git/test/pack-index.test.js b/packages/git/test/pack-index.test.js index cc1d523..c19a4f4 100644 --- a/packages/git/test/pack-index.test.js +++ b/packages/git/test/pack-index.test.js @@ -160,3 +160,65 @@ describe('buildBundleIndex', () => { ); }); }); + +describe('indexPacks retention', () => { + /** + * Every entry the walk names, in order, whatever it keeps. + * @param {object} [options] + */ + async function walk(options) { + /** @type {Array<{sha: string, type: string, size: number}>} */ + const seen = []; + const kept = await indexPacks( + [parseBundle(full).pack], + (sha, meta) => seen.push({ sha, type: meta.type, size: meta.size }), + undefined, + options, + ); + return { seen, kept }; + } + + it('names the same objects whatever it retains', async () => { + const all = await walk({ retain: 'all' }); + const none = await walk({ retain: 'none' }); + expect(none.seen).toEqual(all.seen); + expect(none.kept.size).toBe(0); + expect(all.kept.size).toBe(all.seen.length); + }); + + it('keeps the commit graph alone for a history walk', async () => { + const { seen, kept } = await walk({ retain: 'commits' }); + const history = seen.filter( + (entry) => entry.type === 'commit' || entry.type === 'tag', + ); + expect(history.length).toBeGreaterThan(0); + expect(kept.size).toBe(history.length); + for (const entry of history) { + expect(kept.get(entry.sha)?.type, entry.sha).toBe(entry.type); + } + }); + + it('builds an evicted delta base again from the pack', async () => { + // A one-byte budget keeps no base between entries, so every delta walks + // its chain back down to a whole object. + const all = await walk({ retain: 'all' }); + const starved = await walk({ retain: 'none', cacheBytes: 1 }); + expect(starved.seen).toEqual(all.seen); + }); + + it('resolves a cross-bundle base under a starved cache', async () => { + const prior = await indexPacks([parseBundle(full).pack]); + const index = await buildBundleIndex(incremental, async (sha) => { + return prior.get(sha) ?? null; + }); + /** @type {string[]} */ + const seen = []; + await indexPacks( + [parseBundle(incremental).pack], + (sha) => seen.push(sha), + async (sha) => prior.get(sha) ?? null, + { retain: 'none', cacheBytes: 1 }, + ); + expect(seen).toEqual(index.entries.map((entry) => entry.sha)); + }); +});