Diffinery

codician/diffinerypublic

Open-source, agent-first GitHub and GitHub Actions alternative that runs on your own Cloudflare account: git hosting, pull requests, code review and drop-in GitHub Actions YAML on Workers, Durable Objects, R2 and Containers.

a807a7e

Branches

claude/community-replies claude/connector-bad-gateway-doc claude/early-access-cta claude/github-discoverability claude/github-star-buttons claude/graphql-issues claude/mcp-links claude/merged-pr-diff claude/numeric-run-ids claude/readme-reorganize claude/readme-screenshots-up claude/showcase-agents-plugins community-replies lists-archived main marketplace-builds mcp-registry-auth mcp-registry-docs plugins-any-instance privacy-page

Tags

v0.1.0
diffinery/packages/git-core/src/pack-stream.ts
970 lines · 39.9 KBRaw History
1import pako from "pako";
2import { equalBytes, fromHex, toHex } from "./bytes.js";
3import { applyDelta, deltaResultSize } from "./delta.js";
4import { CODE_TYPES, TYPE_CODES, hashObject, objectHeader, type GitObject, type ObjectType } from "./objects.js";
5import { MAX_HELD_BYTES, MAX_UNPACKED_BYTES, tooBigUnpacked, tooMuchMemory, type ExternalLookup } from "./pack.js";
6import { deflateChunks } from "./zlib.js";
7import { Sha1 } from "./sha1.js";
8
9/**
10 * Unpacking a pack that is never held whole: it is read in ranges (from R2, where a large push is
11 * staged), indexed in one pass, and each object is inflated again, and its deltas applied again,
12 * whenever it is needed, with a bounded cache of resolved objects.
13 *
14 * The same limits as `parsePack` apply: each object stops inflating at its declared size, and the
15 * pack may unpack to at most `maxBytes` counted the same way (every entry's declared size, then
16 * every delta's result, and a large base once for all its deltas), and deltas are resolved in one
17 * pass from their bases, whatever order the pack lists them in. Inflating objects again is counted
18 * separately, against `maxWork`.
19 *
20 * Everything it holds is charged against `maxHeldBytes` before it is allocated: the caches, about
21 * ENTRY_OVERHEAD bytes for each object, the zlib data it keeps to inflate again, SCRATCH_BYTES for
22 * hashing and storing an object, and every object while it is in use (a delta's instructions, its
23 * base and its result, a base from the repository). Requests are counted and capped too: each R2
24 * read, each base read from the repository and each object storing writes to R2 on its own is a
25 * subrequest, and an invocation has a limited number.
26 */
27
28export interface PackSource {
29 /** The pack's length, its 20-byte checksum included. */
30 readonly size: number;
31 read(offset: number, length: number): Promise<Uint8Array>;
32}
33
34export interface StreamedObject {
35 oid: string;
36 type: ObjectType;
37 size: number;
38}
39
40export interface UnpackOptions {
41 external?: ExternalLookup;
42 maxBytes?: number;
43 /** Bytes inflated or rebuilt from deltas in all, re-reads included (default 4 × maxBytes). */
44 maxWork?: number;
45 /** Blobs at least this large are hashed and handed over as they inflate, never whole (like `parsePack`'s). */
46 largeObjectBytes?: number;
47 /** Objects a pack may hold. */
48 maxObjects?: number;
49 /** The pack is read in blocks of this size for its one sequential pass (and for storing it). */
50 blockBytes?: number;
51 maxBlocks?: number;
52 /** Resolved objects kept for reuse as delta bases, in bytes. */
53 cacheBytes?: number;
54 /** Zlib data of deltas kept from the first pass, so they are inflated again without reading the source (default 16 MiB). */
55 keepBytes?: number;
56 /** Bytes read at once when an entry not kept is read again, so entries near each other take one read. */
57 windowBytes?: number;
58 /** What it may hold in all (default MAX_HELD_BYTES, as parsePack). */
59 maxHeldBytes?: number;
60 /**
61 * Requests in all (default 500, well within an invocation's 1,000 subrequests): reads of the
62 * source, storing's included, bases read from the repository, and objects stored on their own.
63 */
64 maxReads?: number;
65 /** Bytes read from the source in all, storing included (default 6 × the pack's size + 64 MiB). */
66 maxReadBytes?: number;
67 /**
68 * Storing writes an object whose zlib data is larger than this with a request of its own (to R2):
69 * each one is counted with the reads. Default: none is.
70 */
71 writesOver?: number;
72 /** Bytes of inflating between yields to the event loop (default 4 MiB), so a long unpack doesn't hold up everything else in the isolate. */
73 yieldEvery?: number;
74}
75
76/** Hands a large blob's content to `onChunk` slice by slice. */
77export type ChunkFeed = (onChunk: (chunk: Uint8Array) => void) => Promise<void>;
78
79const OFS_DELTA = 6;
80const REF_DELTA = 7;
81export const MAX_STREAMED_OBJECTS = 1_000_000;
82
83/**
84 * What one object costs to hold beyond its content: its slots in the index's typed arrays and its
85 * oid, and, for a delta, its place in the lists of deltas waiting on a base and in the replay plan.
86 * Measured at about 100 bytes in Node at the end of indexing 200,000 small objects, half of them
87 * OFS_DELTAs.
88 */
89export const ENTRY_OVERHEAD = 128;
90
91/**
92 * What a REF_DELTA costs besides: its base's oid, kept by entry and as the key it waits on. Measured
93 * at about 510 bytes more than an OFS_DELTA.
94 */
95export const REF_DELTA_OVERHEAD = 640;
96
97/** What keeping one object in a cache costs beyond its bytes: a map entry, its key, and the Uint8Array. */
98const CACHED_OVERHEAD = 192;
99
100/** Zlib data longer than this isn't kept from the first pass: it is read again when needed. */
101const KEEP_ENTRY_MAX = 1024 * 1024;
102
103/** Kept zlib data is packed into buffers of this size; an entry over a quarter of it gets its own. */
104const KEEP_CHUNK = 1024 * 1024;
105
106/**
107 * Held throughout for the one object being hashed or stored: hashing copies one of up to
108 * HASH_COPY_MAX bytes after its header (a larger one is hashed in pieces), and storing compresses
109 * one, whole up to 1 MiB (its zlib chunks and their join), larger ones a chunk at a time.
110 */
111export const SCRATCH_BYTES = 3 * 1024 * 1024;
112
113const DEFAULTS = { blockBytes: 2 * 1024 * 1024, maxBlocks: 4, cacheBytes: 12 * 1024 * 1024, keepBytes: 16 * 1024 * 1024, windowBytes: 1024 * 1024, maxReads: 500 };
114
115/** The most zlib data `n` bytes compress to (zlib's deflateBound, with its 6 bytes of header and trailer). */
116function deflateBound(n: number): number {
117 return n + Math.floor(n / 4096) + Math.floor(n / 16384) + Math.floor(n / 33554432) + 13 + 6;
118}
119
120/** Fixed-size blocks of the source, the most recently used kept: for reading it in order. */
121class Blocks {
122 private cache = new Map<number, Uint8Array>();
123 constructor(
124 private src: PackSource,
125 private blockBytes: number,
126 private maxBlocks: number,
127 private onRead: (bytes: number) => void,
128 ) {}
129
130 /** Forget the blocks kept, so what follows reads exactly as it would from the start. */
131 clear(): void {
132 this.cache.clear();
133 }
134
135 async block(i: number): Promise<Uint8Array> {
136 const hit = this.cache.get(i);
137 if (hit) {
138 this.cache.delete(i);
139 this.cache.set(i, hit);
140 return hit;
141 }
142 const start = i * this.blockBytes;
143 const len = Math.min(this.blockBytes, this.src.size - start);
144 if (len <= 0) throw new Error("read past the end of the pack");
145 this.onRead(len);
146 const data = await this.src.read(start, len);
147 if (data.length !== len) throw new Error(`short read of the pack at ${start}: ${data.length} of ${len} bytes`);
148 this.cache.set(i, data);
149 while (this.cache.size > this.maxBlocks) this.cache.delete(this.cache.keys().next().value as number);
150 return data;
151 }
152
153 /** Views of the source from `offset` on, block by block, up to `end`. */
154 async *views(offset: number, end = this.src.size): AsyncGenerator<Uint8Array> {
155 let at = offset;
156 while (at < end) {
157 const i = Math.floor(at / this.blockBytes);
158 const b = await this.block(i);
159 const from = at - i * this.blockBytes;
160 const to = Math.min(b.length, end - i * this.blockBytes);
161 yield b.subarray(from, to);
162 at += to - from;
163 }
164 }
165
166 async bytes(offset: number, n: number): Promise<Uint8Array> {
167 const end = Math.min(offset + n, this.src.size);
168 const out = new Uint8Array(end - offset);
169 let at = 0;
170 for await (const v of this.views(offset, end)) {
171 out.set(v, at);
172 at += v.length;
173 }
174 return out;
175 }
176}
177
178/**
179 * An inflater for one zlib stream (zlib only: pako also takes gzip unless told not to, the zlib data
180 * is stored as sent, and git can't read gzip) that stops at `expectedSize` and writes into one
181 * buffer of that size, or hands its output to `onChunk`.
182 */
183function inflater(offset: number, expectedSize: number, onChunk?: (chunk: Uint8Array) => void) {
184 const inflator = new pako.Inflate(onChunk ? { chunkSize: 256 * 1024, windowBits: 15 } : { windowBits: 15 }) as pako.Inflate & { ended: boolean };
185 const out = onChunk ? null : new Uint8Array(expectedSize);
186 let produced = 0;
187 inflator.onData = (chunk) => {
188 const c = chunk as Uint8Array;
189 // A small zlib stream can't be made to inflate past the size the pack declared.
190 if (produced + c.length > expectedSize) throw new Error(`object at ${offset} is larger than its declared size (${expectedSize} bytes)`);
191 if (onChunk) onChunk(c);
192 else out!.set(c, produced);
193 produced += c.length;
194 };
195 return {
196 push(v: Uint8Array) {
197 inflator.push(v, false);
198 if (inflator.err) throw new Error(`zlib error at ${offset}: ${inflator.msg}`);
199 return inflator.ended;
200 },
201 /** Input pushed but not used: what follows the stream. */
202 unused() {
203 return (inflator as unknown as { strm: { avail_in: number } }).strm.avail_in;
204 },
205 finish(pushed: number) {
206 if (!inflator.ended) throw new Error(`truncated zlib stream at offset ${offset}`);
207 if (produced !== expectedSize) throw new Error(`size mismatch at ${offset}: ${produced} != ${expectedSize}`);
208 return { consumed: pushed - this.unused(), out };
209 },
210 };
211}
212
213/**
214 * One zlib stream starting at `offset` in the source: its input length, and its output (whole, or
215 * to `onChunk`). `onInput` gets exactly the stream's bytes, as they are read.
216 */
217async function inflateFrom(blocks: Blocks, offset: number, expectedSize: number, onChunk?: (chunk: Uint8Array) => void, onInput?: (bytes: Uint8Array) => void): Promise<{ consumed: number; out: Uint8Array | null }> {
218 const z = inflater(offset, expectedSize, onChunk);
219 let pushed = 0;
220 for await (const v of blocks.views(offset)) {
221 pushed += v.length;
222 const ended = z.push(v);
223 onInput?.(ended ? v.subarray(0, v.length - z.unused()) : v);
224 if (ended) break;
225 }
226 return z.finish(pushed);
227}
228
229/** LRU by entry (or by oid, for a base from outside the pack), bounded in bytes, each entry counted with what holding it costs. */
230class Lru {
231 private map = new Map<number | string, Uint8Array>();
232 private bytes = 0;
233 constructor(private max: number) {}
234 get(i: number | string): Uint8Array | undefined {
235 const hit = this.map.get(i);
236 if (hit) {
237 this.map.delete(i);
238 this.map.set(i, hit);
239 }
240 return hit;
241 }
242 set(i: number | string, content: Uint8Array): void {
243 if (content.length > this.max / 4 || this.map.has(i)) return;
244 this.map.set(i, content);
245 this.bytes += content.length + CACHED_OVERHEAD;
246 while (this.bytes > this.max) {
247 const [k, v] = this.map.entries().next().value as [number | string, Uint8Array];
248 this.map.delete(k);
249 this.bytes -= v.length + CACHED_OVERHEAD;
250 }
251 }
252}
253
254export class StreamedPack {
255 private blocks: Blocks;
256 private cache: Lru;
257 // The index, one slot per entry in pack order, in typed arrays: the pack says how many entries it has.
258 private offsets = new Float64Array(0);
259 /** Bytes from the entry's start to its zlib data. */
260 private headLen = new Uint8Array(0);
261 private zlen = new Float64Array(0);
262 private codes = new Uint8Array(0);
263 /** The size each entry's header declares: an object's, or a delta's instructions'. */
264 private declared = new Float64Array(0);
265 /** Each object's size (a delta's once it is resolved). */
266 private sizes = new Float64Array(0);
267 /** Each object's type code, once known. */
268 private typeCodes = new Uint8Array(0);
269 /** Each delta's base: an entry, or EXTERNAL for one from outside the pack. */
270 private baseOf = new Int32Array(0);
271 /** Each object's oid, 20 bytes apiece, once known. */
272 private oidBytes = new Uint8Array(0);
273 /**
274 * Zlib data kept from the first pass: deltas', packed into a few large buffers (no per-entry
275 * object). An entry's is at `keptOff` in buffer `keptChunk` (-1: not kept).
276 */
277 private keptChunks: Uint8Array[] = [];
278 private keptChunk = new Int32Array(0);
279 private keptOff = new Uint32Array(0);
280 /** Where the next small entry goes in the last shared buffer. */
281 private keptFill = { chunk: -1, at: 0 };
282 /** The source last read again, from `start`. */
283 private window: { start: number; data: Uint8Array } | null = null;
284 private keptBytes = 0;
285 /** A REF_DELTA's base oid, until its base is found in the pack (kept for one outside it). */
286 private refBase = new Map<number, string>();
287 /** Deltas waiting on a base in the pack, by its entry, or by oid (a REF_DELTA's). */
288 private waitingOnEntry = new Map<number, number[]>();
289 private waitingOnOid = new Map<string, number[]>();
290 /** The zlib data's size of each delta result large enough that storing may write it on its own. */
291 private zlibSizes = new Map<number, number>();
292 /** Bases from outside the pack (a thin pack's), by oid: their type. */
293 private externals = new Map<string, ObjectType>();
294 /**
295 * What resolving the deltas did, in order, for `replayDeltas` to do again exactly: a base (its
296 * entry, or -1 - k for the k-th base from outside the pack), the number of deltas against it, then
297 * those deltas.
298 */
299 private plan: number[] = [];
300 private externalOids: string[] = [];
301 private entries = 0;
302 private unpacked = 0;
303 private work = 0;
304 /** What is held, as charged against `maxHeldBytes`, and the most it has been. */
305 private held = 0;
306 private peak = 0;
307 private reads = 0;
308 private readBytes = 0;
309 /** What a read counts for: twice while resolving (replaying reads it again), and nothing while replaying. */
310 private readWeight = 1;
311 /** Storing's reads of the objects that aren't deltas, counted in advance. */
312 private inSlices = false;
313 /** Work, what is held, and reads once the pack was indexed, where resolving the deltas starts. */
314 private afterIndex = { work: 0, held: 0 };
315 private yieldedAt = 0;
316 private readonly maxBytes: number;
317 private readonly maxWork: number;
318 private readonly maxHeld: number;
319 private readonly maxReads: number;
320 private readonly maxReadBytes: number;
321 private readonly large: number;
322 private readonly cacheBytes: number;
323
324 private constructor(
325 private src: PackSource,
326 private o: UnpackOptions,
327 ) {
328 this.blocks = new Blocks(src, o.blockBytes ?? DEFAULTS.blockBytes, o.maxBlocks ?? DEFAULTS.maxBlocks, (n) => {
329 if (!this.inSlices) this.read_(n, this.readWeight);
330 });
331 this.cacheBytes = o.cacheBytes ?? DEFAULTS.cacheBytes;
332 this.cache = new Lru(this.cacheBytes);
333 this.maxBytes = o.maxBytes ?? MAX_UNPACKED_BYTES;
334 this.maxWork = o.maxWork ?? 4 * this.maxBytes;
335 this.maxHeld = o.maxHeldBytes ?? MAX_HELD_BYTES;
336 this.maxReads = o.maxReads ?? DEFAULTS.maxReads;
337 this.maxReadBytes = o.maxReadBytes ?? 6 * src.size + 64 * 1024 * 1024;
338 this.large = o.largeObjectBytes ?? 4 * 1024 * 1024;
339 }
340
341 /** The index, once it is known to fit. */
342 private allocate(count: number) {
343 this.offsets = new Float64Array(count);
344 this.headLen = new Uint8Array(count);
345 this.zlen = new Float64Array(count);
346 this.codes = new Uint8Array(count);
347 this.declared = new Float64Array(count);
348 this.sizes = new Float64Array(count);
349 this.typeCodes = new Uint8Array(count);
350 this.baseOf = new Int32Array(count).fill(NONE);
351 this.oidBytes = new Uint8Array(count * 20);
352 this.keptChunk = new Int32Array(count).fill(-1);
353 this.keptOff = new Uint32Array(count);
354 }
355
356 get count(): number {
357 return this.entries;
358 }
359
360 /** Bytes inflated or rebuilt so far, as counted against `maxWork`. */
361 get workDone(): number {
362 return this.work;
363 }
364
365 /** What is held now, as charged against `maxHeldBytes`. */
366 get heldBytes(): number {
367 return this.held;
368 }
369
370 /** The most it has held at once, as charged against `maxHeldBytes`. */
371 get peakHeldBytes(): number {
372 return this.peak;
373 }
374
375 /** Reads of the source so far, those storing will make counted in advance. */
376 get readsDone(): number {
377 return this.reads;
378 }
379
380 /**
381 * At most what storing the push adds: the zlib data of each object that isn't a delta, stored as
382 * sent, and each delta's content, stored compressed on its own (counted uncompressed).
383 */
384 get storedBytes(): number {
385 let n = 0;
386 for (let i = 0; i < this.entries; i++) n += this.codes[i] === OFS_DELTA || this.codes[i] === REF_DELTA ? (this.zlibSizes.get(i) ?? this.sizes[i]!) : this.zlen[i]!;
387 return n;
388 }
389
390 /** A rebuilt delta's size compressed, when it was large enough to be measured (for storing it). */
391 zlibSizeOf(entry: number): number | undefined {
392 return this.zlibSizes.get(entry);
393 }
394
395 /**
396 * Index the pack and resolve every object, handing each one to `visit` once: with its content,
397 * or, for a large blob, with a feed of its content in slices.
398 */
399 static async open(src: PackSource, o: UnpackOptions, visit: Visit): Promise<StreamedPack> {
400 if (src.size < 32) throw new Error("not a packfile");
401 const p = new StreamedPack(src, o);
402 p.read_(12, 1);
403 const head = await src.read(0, 12);
404 if (head.length !== 12 || head[0] !== 0x50 || head[1] !== 0x41 || head[2] !== 0x43 || head[3] !== 0x4b) throw new Error("not a packfile");
405 const view = new DataView(head.buffer, head.byteOffset, 12);
406 const version = view.getUint32(4);
407 if (version !== 2 && version !== 3) throw new Error(`unsupported pack version ${version}`);
408 const count = view.getUint32(8);
409 if (count > (o.maxObjects ?? MAX_STREAMED_OBJECTS)) throw new Error(`the pack has ${count} objects, more than one push may hold; split it into smaller pushes`);
410 // An entry is at least a byte of header and a zlib stream (8 bytes): no more can fit.
411 if (count > (src.size - 32) / 9) throw new Error(`truncated pack: ${src.size} bytes can't hold ${count} objects`);
412 // Its caches, scratch and index are held throughout: refused up front if they can't fit.
413 p.hold((o.blockBytes ?? DEFAULTS.blockBytes) * (o.maxBlocks ?? DEFAULTS.maxBlocks) + p.cacheBytes + (o.windowBytes ?? DEFAULTS.windowBytes) + SCRATCH_BYTES + count * ENTRY_OVERHEAD);
414 p.allocate(count);
415 // Storing reads the pack once more in order, for the objects that aren't deltas: counted now.
416 p.reads += Math.ceil(src.size / (o.blockBytes ?? DEFAULTS.blockBytes));
417 p.readBytes += src.size;
418 p.checkReads();
419 await p.index(count, visit);
420 p.afterIndex = { work: p.work, held: p.held };
421 p.readWeight = 2;
422 await p.resolveDeltas(visit);
423 p.readWeight = 1;
424 return p;
425 }
426
427 private tooBig() {
428 return tooBigUnpacked(this.maxBytes);
429 }
430
431 /** Let other work in the isolate run now and then: an unpack can take most of a minute of CPU. */
432 async breathe(): Promise<void> {
433 if (this.work - this.yieldedAt < (this.o.yieldEvery ?? 4 * 1024 * 1024)) return;
434 this.yieldedAt = this.work;
435 await new Promise((r) => setTimeout(r, 0));
436 }
437
438 private count_(n: number) {
439 this.work += n;
440 if (this.work > this.maxWork) throw new Error(`pack takes more than ${this.maxWork} bytes of inflating to unpack; split it into smaller pushes`);
441 }
442
443 private unpack_(n: number) {
444 if (this.unpacked + n > this.maxBytes) throw this.tooBig();
445 this.unpacked += n;
446 }
447
448 private hold(n: number) {
449 if (this.held + n > this.maxHeld) throw tooMuchMemory(this.maxHeld);
450 this.held += n;
451 if (this.held > this.peak) this.peak = this.held;
452 }
453
454 private release(n: number) {
455 this.held -= n;
456 }
457
458 /** What is left to hold an object with now. */
459 private room(): number {
460 return this.maxHeld - this.held;
461 }
462
463 private read_(bytes: number, weight: number) {
464 this.reads += weight;
465 this.readBytes += bytes * weight;
466 this.checkReads();
467 }
468
469 /** Storing will write an object of `zlibBytes` (at most) with a request of its own: counted now. */
470 private willWrite(zlibBytes: number) {
471 if (zlibBytes > (this.o.writesOver ?? Infinity)) this.read_(0, 1);
472 }
473
474 private checkReads() {
475 if (this.reads > this.maxReads || this.readBytes > this.maxReadBytes) throw new Error(`pack takes more than ${this.maxReads} reads (or ${this.maxReadBytes} bytes) of storage to unpack; split it into smaller pushes`);
476 }
477
478 private setOid(i: number, oid: string) {
479 this.oidBytes.set(fromHex(oid), i * 20);
480 }
481
482 private dataAt(i: number): number {
483 return this.offsets[i]! + this.headLen[i]!;
484 }
485
486 /** Keep entry `i`'s zlib data from the first pass, if it is small and there is room. */
487 private async keep(i: number): Promise<void> {
488 const len = this.zlen[i]!;
489 if (len > KEEP_ENTRY_MAX) return;
490 const f = this.keptFill;
491 if (len > KEEP_CHUNK / 4 || f.chunk < 0 || f.at + len > KEEP_CHUNK) {
492 // A new buffer: of its own for a large entry, else shared by the small ones that follow.
493 const size = len > KEEP_CHUNK / 4 ? len : KEEP_CHUNK;
494 if (this.keptBytes + size > (this.o.keepBytes ?? DEFAULTS.keepBytes) || size + CACHED_OVERHEAD > this.room()) return;
495 this.hold(size + CACHED_OVERHEAD);
496 this.keptBytes += size + CACHED_OVERHEAD;
497 this.keptChunks.push(new Uint8Array(size));
498 if (size === KEEP_CHUNK) this.keptFill = { chunk: this.keptChunks.length - 1, at: 0 };
499 }
500 const own = len > KEEP_CHUNK / 4;
501 const c = own ? this.keptChunks.length - 1 : this.keptFill.chunk;
502 const at = own ? 0 : this.keptFill.at;
503 let w = at;
504 for await (const v of this.blocks.views(this.dataAt(i), this.dataAt(i) + len)) {
505 this.keptChunks[c]!.set(v, w);
506 w += v.length;
507 }
508 if (!own) this.keptFill.at += len;
509 this.keptChunk[i] = c;
510 this.keptOff[i] = at;
511 }
512
513 /** Entry `i`'s zlib data as the first pass kept it, or null. */
514 private keptData(i: number): Uint8Array | null {
515 const c = this.keptChunk[i]!;
516 return c < 0 ? null : this.keptChunks[c]!.subarray(this.keptOff[i]!, this.keptOff[i]! + this.zlen[i]!);
517 }
518
519 private async index(count: number, visit: Visit): Promise<void> {
520 const size = this.src.size;
521 // Hashed as it is read, so nothing is read twice.
522 const checksum = new Sha1().update(await this.blocks.bytes(0, 12));
523 const hash = (bytes: Uint8Array) => void checksum.update(bytes);
524 let pos = 12;
525 for (let n = 0; n < count; n++) {
526 const offset = pos;
527 // An entry's header is at most 10 bytes of size, then 20 of base oid (or 10 of base offset).
528 const h = await this.blocks.bytes(pos, 32);
529 let k = 0;
530 let c = h[k++]!;
531 const code = (c >> 4) & 7;
532 let objSize = c & 15;
533 let shift = 4;
534 while (c & 0x80) {
535 if (k >= 11) throw new Error(`bad object header at ${offset}`);
536 c = h[k++]!;
537 objSize += (c & 0x7f) * 2 ** shift;
538 shift += 7;
539 }
540 let baseEntry = NONE;
541 let baseOid: string | undefined;
542 if (code === OFS_DELTA) {
543 c = h[k++]!;
544 let ofs = c & 0x7f;
545 while (c & 0x80) {
546 if (k >= 22) throw new Error(`bad delta base offset at ${offset}`);
547 c = h[k++]!;
548 ofs = (ofs + 1) * 128 + (c & 0x7f);
549 }
550 const base = offset - ofs;
551 if (base < 12 || base >= offset) throw new Error(`bad delta base offset at ${offset}`);
552 baseEntry = this.entryAt(base, n);
553 } else if (code === REF_DELTA) {
554 if (k + 20 > h.length) throw new Error(`truncated pack at ${offset}`);
555 baseOid = toHex(h.subarray(k, k + 20));
556 k += 20;
557 } else if (!CODE_TYPES[code]) {
558 throw new Error(`bad object type ${code} at ${offset}`);
559 }
560 checksum.update(h.subarray(0, k));
561 pos += k;
562 this.unpack_(objSize);
563 const type = CODE_TYPES[code];
564 let consumed: number;
565 if (type === "blob" && objSize >= this.large) {
566 // Hashed as it inflates, never held whole.
567 const sha = new Sha1().update(objectHeader("blob", objSize));
568 const r = await inflateFrom(this.blocks, pos, objSize, (chunk) => sha.update(chunk), hash);
569 this.count_(objSize);
570 consumed = r.consumed;
571 const oid = toHex(sha.digest());
572 this.setOid(n, oid);
573 this.willWrite(consumed);
574 const at = pos;
575 await visit({ oid, type, size: objSize }, null, (onChunk) => this.feed(at, objSize, onChunk));
576 } else if (type) {
577 // Held from before it is inflated (at its declared size) until it has been visited.
578 this.hold(objSize);
579 const r = await inflateFrom(this.blocks, pos, objSize, undefined, hash);
580 this.count_(objSize);
581 consumed = r.consumed;
582 const oid = await hashObject(type, r.out!);
583 this.setOid(n, oid);
584 this.willWrite(consumed);
585 await visit({ oid, type, size: objSize }, r.out!, null);
586 this.release(objSize);
587 } else {
588 // A delta: read through to find where it ends; it is inflated again when resolved.
589 const r = await inflateFrom(this.blocks, pos, objSize, () => {}, hash);
590 this.count_(objSize);
591 consumed = r.consumed;
592 if (baseOid !== undefined) {
593 this.hold(REF_DELTA_OVERHEAD);
594 this.refBase.set(n, baseOid);
595 waitOn(this.waitingOnOid, baseOid, n);
596 } else waitOn(this.waitingOnEntry, baseEntry, n);
597 }
598 this.offsets[n] = offset;
599 this.headLen[n] = k;
600 this.zlen[n] = consumed;
601 this.codes[n] = code;
602 this.declared[n] = objSize;
603 this.sizes[n] = type ? objSize : 0;
604 this.typeCodes[n] = type ? code : 0;
605 this.baseOf[n] = baseEntry;
606 this.entries = n + 1;
607 // A delta is inflated again: kept while it is at hand, if there is room.
608 if (!type) await this.keep(n);
609 pos += consumed;
610 if (pos > size - 20) throw new Error(`truncated pack at ${offset}`);
611 await this.breathe();
612 }
613 if (pos !== size - 20) throw new Error("trailing garbage in pack");
614 const trailer = await this.blocks.bytes(size - 20, 20);
615 if (!equalBytes(trailer, checksum.digest())) throw new Error("pack checksum mismatch");
616 }
617
618 /** Entry number of the entry starting at `offset`, among the first `n` (offsets ascend). */
619 private entryAt(offset: number, n: number): number {
620 let lo = 0;
621 let hi = n - 1;
622 while (lo <= hi) {
623 const mid = (lo + hi) >> 1;
624 const v = this.offsets[mid]!;
625 if (v === offset) return mid;
626 if (v < offset) lo = mid + 1;
627 else hi = mid - 1;
628 }
629 throw new Error(`no object starts at delta base offset ${offset}`);
630 }
631
632 /**
633 * One pass: each object, once known, has the deltas waiting on it applied, depth first, as git's
634 * index-pack does: a delta's own deltas are rebuilt from it while it is at hand. The objects that
635 * aren't deltas start it in pack order, so those read again are read in order. What is left
636 * waits on bases from outside the pack, each read once.
637 */
638 private async resolveDeltas(visit: Visit): Promise<void> {
639 this.window = null;
640 this.blocks.clear();
641 const onKid = (kid: number, content: Uint8Array) => this.visitResolved(kid, content, visit);
642 const takeKids = (kid: number) => this.takeWaiting(kid);
643 for (let i = 0; i < this.entries; i++) {
644 if (this.codes[i] === OFS_DELTA || this.codes[i] === REF_DELTA) continue;
645 const kids = this.takeWaiting(i);
646 if (!kids) continue;
647 // A large blob is inflated once for all its deltas: counted once, as parsePack does.
648 if (this.codes[i] === TYPE_CODES.blob && this.sizes[i]! >= this.large) this.unpack_(this.sizes[i]!);
649 await this.descend({ base: await this.content_(i) }, i, this.typeCodes[i]!, kids, onKid, takeKids);
650 }
651 for (const oid of [...this.waitingOnOid.keys()]) {
652 if (!this.waitingOnOid.has(oid) || !this.o.external) continue;
653 const start = await this.externalStart(oid);
654 if (!start) continue;
655 const kids = this.waitingOnOid.get(oid)!;
656 this.waitingOnOid.delete(oid);
657 this.planGroup(-1 - this.externalOids.length, kids);
658 this.externalOids.push(oid);
659 this.unpack_(start.first.base!.length);
660 await this.descend(start.first, EXTERNAL, start.typeCode, kids, onKid, takeKids);
661 }
662 const unresolved = [...this.waitingOnOid.keys(), ...[...this.waitingOnEntry.keys()].map((e) => `@${this.offsets[e]}`)];
663 if (unresolved.length) throw new Error(`unresolvable delta base(s): ${unresolved.join(", ")}`);
664 }
665
666 /** The deltas waiting on entry `b`, now known, taken (and recorded for the replay); null if none. */
667 private takeWaiting(b: number): number[] | null {
668 const byEntry = this.waitingOnEntry.get(b);
669 const oid = this.oidOf(b);
670 const byOid = this.waitingOnOid.get(oid);
671 if (!byEntry && !byOid) return null;
672 this.waitingOnEntry.delete(b);
673 this.waitingOnOid.delete(oid);
674 const kids = byEntry && byOid ? [...byEntry, ...byOid] : (byEntry ?? byOid)!;
675 this.planGroup(b, kids);
676 return kids;
677 }
678
679 private planGroup(b: number, kids: number[]) {
680 this.plan.push(b, kids.length);
681 for (const k of kids) this.plan.push(k);
682 }
683
684 private async visitResolved(kid: number, content: Uint8Array, visit: Visit): Promise<void> {
685 const type = CODE_TYPES[this.typeCodes[kid]!]!;
686 const oid = await hashObject(type, content);
687 this.setOid(kid, oid);
688 // One that may be stored on its own is compressed now to learn its size, which storing reuses.
689 let zlib = deflateBound(content.length);
690 if (zlib > (this.o.writesOver ?? Infinity)) {
691 zlib = 0;
692 for (const c of deflateChunks(content)) zlib += c.length;
693 this.zlibSizes.set(kid, zlib);
694 }
695 this.willWrite(zlib);
696 await visit({ oid, type, size: content.length }, content, null);
697 }
698
699 /**
700 * Rebuild each of `kids` from `base` (held, released here), handing each on, and each one's own
701 * deltas from it while it is held. The same steps when resolving and when replaying, so both do
702 * the same work, hold the same and read the same.
703 */
704 private async descend(
705 first: { base: Uint8Array | null },
706 b: number,
707 typeCode: number,
708 kids: number[],
709 onKid: (kid: number, content: Uint8Array) => Promise<void>,
710 takeKids: (kid: number) => number[] | null,
711 ): Promise<void> {
712 // The stack is the only reference to the first base (the caller's is dropped), so letting go of it frees it.
713 const stack = [{ base: first.base!, b, kids, k: 0 }];
714 first.base = null;
715 while (stack.length) {
716 const top = stack[stack.length - 1]!;
717 const kid = top.kids[top.k++]!;
718 const content = await this.applyOne(kid, top.base, this.maxBytes - this.unpacked);
719 if (top.b !== EXTERNAL) this.refBase.delete(kid);
720 this.baseOf[kid] = top.b;
721 this.typeCodes[kid] = typeCode;
722 this.sizes[kid] = content.length;
723 this.unpack_(content.length);
724 this.cache.set(kid, content);
725 await onKid(kid, content);
726 await this.breathe();
727 // A base is let go once its last delta is rebuilt, before that delta's own deltas: a long
728 // chain of versions of one file holds two of them at a time, not all of them.
729 if (top.k === top.kids.length) {
730 this.release(top.base.length);
731 top.base = EMPTY;
732 stack.pop();
733 }
734 const next = takeKids(kid);
735 if (next) stack.push({ base: content, b: kid, kids: next, k: 0 });
736 else this.release(content.length);
737 }
738 }
739
740 /** Delta `i`'s instructions, inflated again and applied to `base`: its result, held. */
741 private async applyOne(i: number, base: Uint8Array, limit: number): Promise<Uint8Array> {
742 const delta = await this.inflateAgain(i);
743 const size = deltaResultSize(delta);
744 // The pack's limit first, with parsePack's message; then memory, before the result is allocated.
745 if (size > limit) throw this.tooBig();
746 this.hold(size);
747 const content = applyDelta(base, delta, size);
748 this.release(delta.length);
749 this.count_(content.length);
750 return content;
751 }
752
753 /**
754 * Entry `i`'s zlib data inflated again, held. An object that isn't a delta is read through the
755 * blocks: resolving starts from those in pack order, so they are read in order. A delta's comes
756 * from what the first pass kept, or is read again.
757 */
758 private async inflateAgain(i: number): Promise<Uint8Array> {
759 const declared = this.declared[i]!;
760 const len = this.zlen[i]!;
761 this.hold(declared);
762 if (this.codes[i] !== OFS_DELTA && this.codes[i] !== REF_DELTA) {
763 const { out } = await inflateFrom(this.blocks, this.dataAt(i), declared);
764 this.count_(declared);
765 return out!;
766 }
767 const z = this.keptData(i) ?? (await this.reread(i));
768 const data = z instanceof Uint8Array ? z : z.data;
769 const inf = inflater(this.offsets[i]!, declared);
770 inf.push(data);
771 const { out } = inf.finish(len);
772 if (!(z instanceof Uint8Array)) this.release(z.held);
773 this.count_(declared);
774 return out!;
775 }
776
777 /**
778 * Entry `i`'s zlib data read again: from the window of the source last read, or a new window
779 * from it on (exactly its range, held while in use, if larger than a window).
780 */
781 private async reread(i: number): Promise<Uint8Array | { data: Uint8Array; held: number }> {
782 const at = this.dataAt(i);
783 const len = this.zlen[i]!;
784 const w = this.window;
785 if (w && at >= w.start && at + len <= w.start + w.data.length) return w.data.subarray(at - w.start, at - w.start + len);
786 const windowBytes = this.o.windowBytes ?? DEFAULTS.windowBytes;
787 const n = len > windowBytes ? len : Math.max(len, Math.min(windowBytes, this.src.size - 20 - at));
788 if (len > windowBytes) this.hold(len);
789 this.read_(n, this.readWeight);
790 const data = await this.src.read(at, n);
791 if (data.length !== n) throw new Error(`short read of the pack at ${at}: ${data.length} of ${n} bytes`);
792 if (len > windowBytes) return { data, held: len };
793 this.window = { start: at, data };
794 return data.subarray(0, len);
795 }
796
797 private async feed(at: number, size: number, onChunk: (chunk: Uint8Array) => void): Promise<void> {
798 this.count_(size);
799 await inflateFrom(this.blocks, at, size, onChunk);
800 }
801
802 /** Entry `i`'s content, held by the caller until it releases it (a cache spares most of the work). */
803 private async content_(i: number, depth = 0): Promise<Uint8Array> {
804 const hit = this.cache.get(i);
805 if (hit) {
806 this.hold(hit.length);
807 return hit;
808 }
809 if (depth > 10_000) throw new Error("delta chain too deep");
810 const code = this.codes[i]!;
811 let content: Uint8Array;
812 if (code === OFS_DELTA || code === REF_DELTA) {
813 const b = this.baseOf[i]!;
814 let base: Uint8Array;
815 if (b >= 0) base = await this.content_(b, depth + 1);
816 else {
817 const oid = this.refBase.get(i);
818 const ext = oid === undefined ? null : await this.externalBase(oid);
819 if (!ext) throw new Error(`delta base of the object at ${this.offsets[i]} is missing`);
820 base = ext.content;
821 }
822 content = await this.applyOne(i, base, this.maxBytes);
823 this.release(base.length);
824 } else content = await this.inflateAgain(i);
825 this.cache.set(i, content);
826 return content;
827 }
828
829 /** Entry `i`'s content: inflated, and rebuilt from its delta chain, again (a cache spares most of it). */
830 async content(i: number): Promise<Uint8Array> {
831 const c = await this.content_(i);
832 this.release(c.length);
833 return c;
834 }
835
836 /**
837 * A base from outside the pack, read (and inflated) by `external`, held by the caller until it
838 * releases it: kept in the cache like a resolved object, and each read counted against
839 * `maxWork`, so many tiny deltas against one large stored object can't make the server read it
840 * again without bound.
841 */
842 private async externalBase(oid: string): Promise<GitObject | null> {
843 const type = this.externals.get(oid);
844 const hit = type && this.cache.get(oid);
845 if (hit) {
846 this.hold(hit.length);
847 return { type, content: hit };
848 }
849 // Its size is checked before it is read: a small delta can name a huge stored object.
850 const base = (await this.o.external?.(oid, this.room())) ?? null;
851 if (base === null) return null;
852 if (typeof base === "number") throw tooMuchMemory(this.maxHeld);
853 this.read_(0, this.readWeight);
854 this.hold(base.content.length);
855 this.count_(base.content.length);
856 this.externals.set(oid, base.type);
857 this.cache.set(oid, base.content);
858 return base;
859 }
860
861 /** A base from outside the pack, to descend from: its content in a holder `descend` empties, and its type. */
862 private async externalStart(oid: string): Promise<{ first: { base: Uint8Array | null }; typeCode: number } | null> {
863 const ext = await this.externalBase(oid);
864 return ext && { first: { base: ext.content }, typeCode: TYPE_CODES[ext.type] };
865 }
866
867 /**
868 * Every delta's content again, for storing: rebuilt in the order resolving them took and from
869 * the same empty caches, so it does exactly the work, holds exactly as much, reads exactly the
870 * same and reaches exactly the chain depth that already passed in `open`. Nothing in it can go
871 * over a limit `open` didn't, so a push can't be refused partway through being stored.
872 */
873 async replayDeltas(each: (entry: number, content: Uint8Array) => Promise<void>): Promise<void> {
874 this.cache = new Lru(this.cacheBytes);
875 this.window = null;
876 this.blocks.clear();
877 const after = { work: this.work, held: this.held, unpacked: this.unpacked };
878 this.work = this.afterIndex.work;
879 this.held = this.afterIndex.held;
880 // Counted from here again, or it wouldn't yield until the work passed where `open` left off.
881 this.yieldedAt = this.work;
882 // Its reads were counted in `open`, and the limit on what the pack unpacks to checked there.
883 this.readWeight = 0;
884 this.unpacked = 0;
885 const plan = this.plan;
886 let at = 0;
887 const next = () => {
888 const b = plan[at]!;
889 const kids = plan.slice(at + 2, at + 2 + plan[at + 1]!);
890 at += 2 + kids.length;
891 return { b, kids };
892 };
893 // A delta's own group comes right after it in the plan, as `open` took it.
894 const takeKids = (kid: number) => (at < plan.length && plan[at] === kid ? next().kids : null);
895 while (at < plan.length) {
896 const { b, kids } = next();
897 if (b >= 0) await this.descend({ base: await this.content_(b) }, b, this.typeCodes[b]!, kids, each, takeKids);
898 else {
899 const oid = this.externalOids[-1 - b]!;
900 const start = await this.externalStart(oid);
901 if (!start) throw new Error(`delta base ${oid} is missing`);
902 await this.descend(start.first, EXTERNAL, start.typeCode, kids, each, takeKids);
903 }
904 }
905 this.readWeight = 1;
906 this.work = after.work;
907 this.held = after.held;
908 this.unpacked = after.unpacked;
909 }
910
911 /**
912 * Every object in pack order, for storing. One that isn't a delta comes with its zlib data's place
913 * in the pack (checked to inflate to exactly its content), to be stored as it is: neither inflated
914 * nor compressed again.
915 */
916 *objects(): Generator<StreamedObject & { entry: number; deflated?: { at: number; length: number } }> {
917 for (let i = 0; i < this.entries; i++) {
918 const o = { oid: this.oidOf(i), type: this.typeOf(i), size: this.sizes[i]!, entry: i };
919 if (this.codes[i] !== OFS_DELTA && this.codes[i] !== REF_DELTA) yield { ...o, deflated: { at: this.dataAt(i), length: this.zlen[i]! } };
920 else yield o;
921 }
922 }
923
924 /** `length` bytes of the pack from `at`, slice by slice (in order: these reads were counted in `open`). */
925 async *slices(at: number, length: number): AsyncGenerator<Uint8Array> {
926 for (const views = this.blocks.views(at, at + length); ; ) {
927 this.inSlices = true;
928 let next: IteratorResult<Uint8Array>;
929 try {
930 next = await views.next();
931 } finally {
932 this.inSlices = false;
933 }
934 if (next.done) return;
935 yield next.value;
936 }
937 }
938
939 oidOf(i: number): string {
940 return toHex(this.oidBytes.subarray(i * 20, i * 20 + 20));
941 }
942
943 typeOf(i: number): ObjectType {
944 return CODE_TYPES[this.typeCodes[i]!]!;
945 }
946
947 /** The pushed objects with these oids, as entry numbers (one pass over the index). */
948 entriesOf(oids: ReadonlySet<string>): Map<string, number> {
949 const found = new Map<string, number>();
950 if (!oids.size) return found;
951 for (let i = 0; i < this.entries; i++) {
952 const oid = this.oidOf(i);
953 if (oids.has(oid) && !found.has(oid)) found.set(oid, i);
954 }
955 return found;
956 }
957}
958
959type Visit = (obj: StreamedObject, content: Uint8Array | null, feed: ChunkFeed | null) => void | Promise<void>;
960
961const NONE = -1;
962const EMPTY = new Uint8Array(0);
963/** A delta's base from outside the pack. */
964const EXTERNAL = -2;
965
966function waitOn<K>(m: Map<K, number[]>, k: K, i: number) {
967 const list = m.get(k);
968 if (list) list.push(i);
969 else m.set(k, [i]);
970}