1/**
2 * Scans pack entries out of a BUFFER that starts at an entry boundary and
3 * may end mid-entry (DESIGN §22.7): the unit of work a hashing Worker
4 * receives from the receive-pack pipeline — one spilled part plus the
5 * carry-over of the previous part's incomplete tail.
6 *
7 * Everything that can be settled inside the buffer is settled here:
8 * non-delta entries are inflated and hashed (the expensive part of
9 * ingest), OFS/REF deltas whose base is in the same buffer are applied,
10 * re-deflated and hashed. Deltas whose base lies outside the buffer are
11 * reported as `unresolved` with what the caller needs to settle them
12 * later. The scan stops at the first entry that does not fit; `consumedTo`
13 * is the absolute offset where it starts, so the caller can carry the tail.
14 * An entry only counts as fitting when the buffer extends at least one byte
15 * past it, so the buffer must always end with what FOLLOWS the last entry
16 * to consume (the next part's bytes, or the pack trailer).
17 *
18 * Pure: no store, no I/O. The trailer is NOT verified here (the caller
19 * hashes the whole pack as it streams).
20 */
21import * as Effect from "effect/Effect";
22import { applyDelta } from "./Delta.ts";
23import {
24 bytesToHex,
25 decodeOfsDeltaOffset,
26 decodeTypeSize,
27 hashObjectSync,
28 ObjectParseError,
29 type ObjectType,
30 type Oid,
31} from "./ObjectCodec.ts";
32import {
33 deflate,
34 inflate,
35 inflateEntrySync,
36 inflateExactSpan,
37 type InflatedEntry,
38} from "./Zlib.ts";
39import { ObjectTooLargeError, PackFormatError } from "./PackParser.ts";
40
41export interface ScannedEntry {
42 readonly oid: Oid;
43 readonly type: ObjectType;
44 /** Absolute pack offset of the entry HEADER (what OFS_DELTAs point at). */
45 readonly offset: number;
46 /** Uncompressed size. */
47 readonly size: number;
48 /** Absolute pack offset of the entry's zdata (its header precedes it). */
49 readonly dataOffset: number;
50 /** Compressed span at `dataOffset` — verbatim for non-delta entries. */
51 readonly span: number;
52 /** Set for delta-resolved entries: a fresh deflate to store as the row. */
53 readonly zdata?: Uint8Array | undefined;
54 /** Inflated content for commits, trees and tags (the store parses them). */
55 readonly content?: Uint8Array | undefined;
56 /**
57 * For a delta-resolved entry, its base reference — so a hasher that must
58 * bound its response can demote the entry back to `unresolved`.
59 */
60 readonly baseOffset?: number | undefined;
61 readonly baseOid?: Oid | undefined;
62}
63
64export interface UnresolvedDelta {
65 /** Absolute pack offset of the entry header. */
66 readonly offset: number;
67 /** Absolute offset of the delta payload's zdata and its span. */
68 readonly dataOffset: number;
69 readonly span: number;
70 /** The base: an absolute pack offset (OFS_DELTA) or an oid (REF_DELTA). */
71 readonly baseOffset?: number | undefined;
72 readonly baseOid?: Oid | undefined;
73 /** Declared (uncompressed) size of the delta payload. */
74 readonly size: number;
75}
76
77export interface ScanResult {
78 /** Absolute offset of the first entry scanned (after resync), or -1 if none. */
79 readonly firstOffset: number;
80 readonly entries: ReadonlyArray<ScannedEntry>;
81 readonly unresolved: ReadonlyArray<UnresolvedDelta>;
82 /** Absolute offset of the first entry NOT consumed (carry from here). */
83 readonly consumedTo: number;
84 /** Entries consumed (resolved + unresolved). */
85 readonly count: number;
86}
87
88const isDelta = (type: number) => type === 6 || type === 7;
89
90/**
91 * Scans `buf`, whose byte 0 is at absolute pack offset `base`. `remaining`
92 * is how many entries the pack still owes (so a trailing partial entry
93 * beyond the count is not mistaken for data); `maxObjectSize` bounds
94 * inflated sizes.
95 */
96export const scanPart = (
97 buf: Uint8Array,
98 options: {
99 readonly base: number;
100 readonly remaining: number;
101 readonly maxObjectSize: number;
102 /** Resolved content LRU budget for in-buffer delta bases. */
103 readonly cacheBytes?: number | undefined;
104 /**
105 * `buf` does NOT necessarily start at an entry boundary (a raw chunk of
106 * a pack, DESIGN §22.9): find the first boundary with {@link findBoundary}
107 * and scan from there. No boundary within the chunk ⇒ no entries.
108 */
109 readonly resync?: boolean | undefined;
110 },
111): Effect.Effect<ScanResult, PackFormatError | ObjectTooLargeError> =>
112 Effect.gen(function* () {
113 const entries: Array<ScannedEntry> = [];
114 const unresolved: Array<UnresolvedDelta> = [];
115 let firstOffset = -1;
116 let startAt = 0;
117 if (options.resync === true) {
118 const found = findBoundary(buf, { maxObjectSize: options.maxObjectSize });
119 if (found === undefined) {
120 return {
121 firstOffset: -1,
122 entries,
123 unresolved,
124 consumedTo: options.base,
125 count: 0,
126 };
127 }
128 startAt = found;
129 firstOffset = options.base + found;
130 }
131 // In resync mode `remaining` is only an upper bound and the buffer may
132 // run into the pack trailer: anything that does not parse as an entry
133 // ends the scan instead of failing it (the pipeline validates the count,
134 // the trailer hash and every object id downstream).
135 const lenient = options.resync === true;
136 // Content of resolved entries in this buffer, by absolute offset and by
137 // oid, for in-buffer delta bases. Bounded; a miss becomes unresolved.
138 const byOffset = new Map<
139 number,
140 { type: ObjectType; content: Uint8Array }
141 >();
142 const byOid = new Map<string, { type: ObjectType; content: Uint8Array }>();
143 let cached = 0;
144 const budget = options.cacheBytes ?? 20 * 1024 * 1024;
145 const remember = (
146 offset: number,
147 oid: Oid,
148 type: ObjectType,
149 content: Uint8Array,
150 ) => {
151 if (content.length > budget / 4) return;
152 while (cached + content.length > budget && byOffset.size > 0) {
153 const oldest = byOffset.keys().next().value!;
154 const dropped = byOffset.get(oldest)!;
155 byOffset.delete(oldest);
156 cached -= dropped.content.length;
157 }
158 byOffset.set(offset, { type, content });
159 byOid.set(oid, { type, content });
160 cached += content.length;
161 };
162
163 let pos = startAt;
164 let count = 0;
165 while (count < options.remaining && pos < buf.length) {
166 const offset = options.base + pos;
167 let header;
168 try {
169 header = decodeTypeSize(buf, pos);
170 } catch (error) {
171 // A header cut by the buffer edge: stop here, carry the tail.
172 if (lenient || buf.length - pos < 16) break;
173 return yield* new PackFormatError({
174 reason:
175 error instanceof ObjectParseError
176 ? `entry at ${offset}: ${error.reason}`
177 : `entry at ${offset}: ${String(error)}`,
178 });
179 }
180 if (header.size > options.maxObjectSize) {
181 if (lenient) break;
182 return yield* new ObjectTooLargeError({
183 size: header.size,
184 limit: options.maxObjectSize,
185 });
186 }
187 let at = header.next;
188 let baseOffset: number | undefined;
189 let baseOid: Oid | undefined;
190 if (header.type === 6) {
191 let ofs;
192 try {
193 ofs = decodeOfsDeltaOffset(buf, at);
194 } catch {
195 if (lenient || buf.length - at < 16) break;
196 return yield* new PackFormatError({
197 reason: `entry at ${offset}: bad ofs-delta`,
198 });
199 }
200 at = ofs.next;
201 baseOffset = offset - ofs.value;
202 } else if (header.type === 7) {
203 if (at + 20 > buf.length) break;
204 baseOid = bytesToHex(buf.subarray(at, at + 20)) as Oid;
205 at += 20;
206 }
207 const inflated: InflatedEntry | undefined = inflateEntrySync(buf, at, {
208 maxOutput: options.maxObjectSize,
209 expectedSize: header.size,
210 });
211 if (inflated === undefined) {
212 // The compressed stream runs past the buffer (or is corrupt — the
213 // next attempt, with more bytes, decides): carry from this entry.
214 break;
215 }
216 const { bytesConsumed, content: payload } = inflated;
217 // A stream cut inside its last bytes (end-of-block bits, adler32) can
218 // inflate to exactly the declared size without having ended, and the
219 // consumed count would then be short. An entry is complete only when
220 // the buffer extends PAST it: callers always append what follows
221 // (the next part, or the pack trailer for the last one).
222 if (at + bytesConsumed >= buf.length) break;
223 if (payload.length !== header.size) {
224 if (lenient) break;
225 return yield* new PackFormatError({
226 reason: `entry at ${offset}: inflated ${payload.length} bytes, header declared ${header.size}`,
227 });
228 }
229 const dataOffset = options.base + at;
230 if (!isDelta(header.type)) {
231 const type = header.type as ObjectType;
232 const oid = hashObjectSync(type, payload);
233 entries.push({
234 oid,
235 type,
236 offset,
237 size: payload.length,
238 dataOffset,
239 span: bytesConsumed,
240 content: type === 3 ? undefined : payload,
241 });
242 remember(offset, oid, type, payload);
243 } else {
244 const found =
245 baseOffset !== undefined
246 ? byOffset.get(baseOffset)
247 : byOid.get(baseOid!);
248 if (found === undefined) {
249 unresolved.push({
250 offset,
251 dataOffset,
252 span: bytesConsumed,
253 baseOffset,
254 baseOid,
255 size: header.size,
256 });
257 } else {
258 const content = yield* applyDelta(found.content, payload).pipe(
259 Effect.mapError(
260 (error) =>
261 new PackFormatError({
262 reason: `entry at ${offset}: ${error.reason}`,
263 }),
264 ),
265 );
266 if (content.length > options.maxObjectSize) {
267 return yield* new ObjectTooLargeError({
268 size: content.length,
269 limit: options.maxObjectSize,
270 });
271 }
272 const oid = hashObjectSync(found.type, content);
273 const zdata = yield* deflate(content).pipe(
274 Effect.mapError(
275 (error) => new PackFormatError({ reason: error.reason }),
276 ),
277 );
278 entries.push({
279 oid,
280 type: found.type,
281 offset,
282 size: content.length,
283 dataOffset,
284 span: bytesConsumed,
285 zdata,
286 content: found.type === 3 ? undefined : content,
287 baseOffset,
288 baseOid,
289 });
290 remember(offset, oid, found.type, content);
291 }
292 }
293 pos = at + bytesConsumed;
294 count += 1;
295 }
296 return {
297 firstOffset:
298 firstOffset >= 0 ? firstOffset : entries.length > 0 ? options.base : -1,
299 entries,
300 unresolved,
301 consumedTo: options.base + pos,
302 count,
303 };
304 });
305
306/** One entry's coordinates from a boundary-only scan (DESIGN §22.8). */
307export interface EntryBounds {
308 /** Absolute pack offset of the entry header. */
309 readonly offset: number;
310 /** 1–4 for full objects, 6/7 for OFS/REF deltas. */
311 readonly type: number;
312 /** Declared (uncompressed) size. */
313 readonly size: number;
314 /** Absolute offset and length of the compressed span. */
315 readonly dataOffset: number;
316 readonly span: number;
317 readonly baseOffset?: number | undefined;
318 readonly baseOid?: Oid | undefined;
319}
320
321export interface BoundsResult {
322 readonly entries: ReadonlyArray<EntryBounds>;
323 /** Absolute offset of the first entry NOT consumed (carry from here). */
324 readonly consumedTo: number;
325}
326
327/**
328 * Boundary-only scan (DESIGN §22.8): finds every entry's compressed span in
329 * `buf` without hashing or resolving anything — the one inherently
330 * sequential step of pack ingest (a stream's end is only known by inflating
331 * it). With spans known, hashing every part can run in parallel and can
332 * inflate exact slices with the cheap one-shot path. Same contract as
333 * {@link scanPart}: `buf` starts at an entry boundary and must extend past
334 * the last entry to consume.
335 */
336export const scanBounds = (
337 buf: Uint8Array,
338 options: {
339 readonly base: number;
340 readonly remaining: number;
341 readonly maxObjectSize: number;
342 },
343): Effect.Effect<BoundsResult, PackFormatError | ObjectTooLargeError> =>
344 Effect.gen(function* () {
345 const entries: Array<EntryBounds> = [];
346 let pos = 0;
347 let count = 0;
348 while (count < options.remaining && pos < buf.length) {
349 const offset = options.base + pos;
350 let header;
351 try {
352 header = decodeTypeSize(buf, pos);
353 } catch (error) {
354 if (buf.length - pos < 16) break;
355 return yield* new PackFormatError({
356 reason:
357 error instanceof ObjectParseError
358 ? `entry at ${offset}: ${error.reason}`
359 : `entry at ${offset}: ${String(error)}`,
360 });
361 }
362 if (header.size > options.maxObjectSize) {
363 return yield* new ObjectTooLargeError({
364 size: header.size,
365 limit: options.maxObjectSize,
366 });
367 }
368 let at = header.next;
369 let baseOffset: number | undefined;
370 let baseOid: Oid | undefined;
371 if (header.type === 6) {
372 let ofs;
373 try {
374 ofs = decodeOfsDeltaOffset(buf, at);
375 } catch {
376 if (buf.length - at < 16) break;
377 return yield* new PackFormatError({
378 reason: `entry at ${offset}: bad ofs-delta`,
379 });
380 }
381 at = ofs.next;
382 baseOffset = offset - ofs.value;
383 } else if (header.type === 7) {
384 if (at + 20 > buf.length) break;
385 baseOid = bytesToHex(buf.subarray(at, at + 20)) as Oid;
386 at += 20;
387 }
388 const inflated = inflateEntrySync(buf, at, {
389 maxOutput: options.maxObjectSize,
390 expectedSize: header.size,
391 });
392 if (inflated === undefined) break;
393 if (at + inflated.bytesConsumed >= buf.length) break;
394 if (inflated.content.length !== header.size) {
395 return yield* new PackFormatError({
396 reason: `entry at ${offset}: inflated ${inflated.content.length} bytes, header declared ${header.size}`,
397 });
398 }
399 entries.push({
400 offset,
401 type: header.type,
402 size: header.size,
403 dataOffset: options.base + at,
404 span: inflated.bytesConsumed,
405 baseOffset,
406 baseOid,
407 });
408 pos = at + inflated.bytesConsumed;
409 count += 1;
410 }
411 return { entries, consumedTo: options.base + pos };
412 });
413
414/**
415 * Hashes entries whose spans are already known (DESIGN §22.8): the parallel
416 * half. `bounds` addresses spans inside `buf` (absolute offsets; `base` is
417 * `buf[0]`'s offset). Non-delta entries are inflated with the one-shot
418 * path and hashed; deltas whose base is among these entries (or already
419 * resolved here) are applied, re-deflated and hashed; the rest are reported
420 * unresolved.
421 */
422export const hashBounds = (
423 buf: Uint8Array,
424 bounds: ReadonlyArray<EntryBounds>,
425 options: {
426 readonly base: number;
427 readonly maxObjectSize: number;
428 readonly cacheBytes?: number | undefined;
429 },
430): Effect.Effect<ScanResult, PackFormatError | ObjectTooLargeError> =>
431 Effect.gen(function* () {
432 const entries: Array<ScannedEntry> = [];
433 const unresolved: Array<UnresolvedDelta> = [];
434 const byOffset = new Map<
435 number,
436 { type: ObjectType; content: Uint8Array }
437 >();
438 const byOid = new Map<string, { type: ObjectType; content: Uint8Array }>();
439 const inflateExact = (b: EntryBounds) =>
440 inflateExactSpan(
441 buf.subarray(
442 b.dataOffset - options.base,
443 b.dataOffset - options.base + b.span,
444 ),
445 b.size,
446 ).pipe(
447 Effect.mapError(
448 (error) =>
449 new PackFormatError({
450 reason: `entry at ${b.offset}: ${error.reason}`,
451 }),
452 ),
453 );
454 for (const b of bounds) {
455 const payload = yield* inflateExact(b);
456 if (!isDelta(b.type)) {
457 const type = b.type as ObjectType;
458 const oid = hashObjectSync(type, payload);
459 entries.push({
460 oid,
461 type,
462 offset: b.offset,
463 size: payload.length,
464 dataOffset: b.dataOffset,
465 span: b.span,
466 content: type === 3 ? undefined : payload,
467 });
468 byOffset.set(b.offset, { type, content: payload });
469 byOid.set(oid, { type, content: payload });
470 continue;
471 }
472 const found =
473 b.baseOffset !== undefined
474 ? byOffset.get(b.baseOffset)
475 : byOid.get(b.baseOid!);
476 if (found === undefined) {
477 unresolved.push({
478 offset: b.offset,
479 dataOffset: b.dataOffset,
480 span: b.span,
481 baseOffset: b.baseOffset,
482 baseOid: b.baseOid,
483 size: b.size,
484 });
485 continue;
486 }
487 const content = yield* applyDelta(found.content, payload).pipe(
488 Effect.mapError(
489 (error) =>
490 new PackFormatError({
491 reason: `entry at ${b.offset}: ${error.reason}`,
492 }),
493 ),
494 );
495 if (content.length > options.maxObjectSize) {
496 return yield* new ObjectTooLargeError({
497 size: content.length,
498 limit: options.maxObjectSize,
499 });
500 }
501 const oid = hashObjectSync(found.type, content);
502 const zdata = yield* deflate(content).pipe(
503 Effect.mapError(
504 (error) => new PackFormatError({ reason: error.reason }),
505 ),
506 );
507 entries.push({
508 oid,
509 type: found.type,
510 offset: b.offset,
511 size: content.length,
512 dataOffset: b.dataOffset,
513 span: b.span,
514 zdata,
515 content: found.type === 3 ? undefined : content,
516 });
517 byOffset.set(b.offset, { type: found.type, content });
518 byOid.set(oid, { type: found.type, content });
519 }
520 const last = bounds[bounds.length - 1];
521 return {
522 firstOffset: bounds[0]?.offset ?? -1,
523 entries,
524 unresolved,
525 consumedTo:
526 last === undefined ? options.base : last.dataOffset + last.span,
527 count: bounds.length,
528 };
529 });
530
531/**
532 * Finds the first entry boundary inside `buf` (DESIGN §22.9): the parallel
533 * answer to "where does an entry start?", so chunks of a pack can be scanned
534 * independently. A position is a boundary when it decodes as an entry
535 * header, its stream inflates to exactly the declared size, AND the entry
536 * that follows does too — two consecutive coincidences on arbitrary bytes
537 * are not a realistic false positive, and the pack trailer plus per-object
538 * hashing verify everything downstream regardless.
539 *
540 * Candidates are filtered cheaply: a deflate stream from git starts with
541 * the zlib header `0x78` followed by one of four FLG bytes, right after a
542 * 1–10 byte varint header. Returns the offset into `buf`, or `undefined`
543 * when no boundary lies within `maxSearch` bytes — the whole chunk by
544 * default: a part whose head is inside one large entry still has its own
545 * entries to hash, and a part that reports no boundary is hashed serially
546 * as a region by the caller (a 10 MB blob spanning three parts once cost
547 * a whole part's worth of serial hashing under a 1 MiB limit).
548 */
549export const findBoundary = (
550 buf: Uint8Array,
551 options: {
552 readonly maxObjectSize: number;
553 readonly maxSearch?: number | undefined;
554 },
555): number | undefined => {
556 const limit = Math.min(buf.length, options.maxSearch ?? buf.length);
557 const isZlibHeader = (i: number) =>
558 i + 1 < buf.length &&
559 buf[i] === 0x78 &&
560 (buf[i + 1] === 0x01 ||
561 buf[i + 1] === 0x5e ||
562 buf[i + 1] === 0x9c ||
563 buf[i + 1] === 0xda);
564 const parses = (pos: number): number | undefined => {
565 let header;
566 try {
567 header = decodeTypeSize(buf, pos);
568 } catch {
569 return undefined;
570 }
571 if (header.type < 1 || header.type > 7) return undefined;
572 if (header.size > options.maxObjectSize) return undefined;
573 let at = header.next;
574 if (header.type === 6) {
575 try {
576 at = decodeOfsDeltaOffset(buf, at).next;
577 } catch {
578 return undefined;
579 }
580 } else if (header.type === 7) {
581 at += 20;
582 }
583 if (!isZlibHeader(at)) return undefined;
584 const inflated = inflateEntrySync(buf, at, {
585 maxOutput: options.maxObjectSize,
586 expectedSize: header.size,
587 });
588 if (inflated === undefined || inflated.content.length !== header.size)
589 return undefined;
590 const end = at + inflated.bytesConsumed;
591 return end >= buf.length ? undefined : end;
592 };
593 for (let pos = 0; pos < limit; pos++) {
594 // Cheap pre-filter: a zlib header must appear within the next 32 bytes
595 // (header varint + optional base ref).
596 let plausible = false;
597 for (let k = pos + 1; k <= pos + 32 && k < buf.length; k++) {
598 if (isZlibHeader(k)) {
599 plausible = true;
600 break;
601 }
602 }
603 if (!plausible) continue;
604 const next = parses(pos);
605 if (next === undefined) continue;
606 // Second entry must parse too (or the buffer ends right after — accept
607 // only if the first entry ended cleanly inside the buffer).
608 const after = parses(next);
609 if (after !== undefined) return pos;
610 }
611 return undefined;
612};
613
614// ── delta resolution as a job (DESIGN §22.13) ──────────────────────────────
615
616/** One base a batch of delta jobs refers to: zlib bytes, or already-inflated content. */
617export interface DeltaBase {
618 readonly bytes: Uint8Array;
619 readonly isContent: boolean;
620}
621
622/** One delta to apply: its base (by index into the batch's bases) and its zlib bytes. */
623export interface DeltaJob {
624 readonly id: number;
625 readonly type: ObjectType;
626 readonly base: number;
627 readonly delta: Uint8Array;
628}
629
630/** A resolved delta: the object's oid and a fresh deflate; content for non-blobs. */
631export interface DeltaResolved {
632 readonly id: number;
633 readonly oid: Oid;
634 readonly size: number;
635 readonly zdata: Uint8Array;
636 readonly content?: Uint8Array | undefined;
637}
638
639/**
640 * Applies a batch of deltas whose bases the caller supplies — the work the
641 * receiver would otherwise do itself for every delta whose base lies in
642 * another chunk. Pure over its inputs, so a hasher isolate can run it.
643 */
644export const resolveDeltas = (
645 bases: ReadonlyArray<DeltaBase>,
646 jobs: ReadonlyArray<DeltaJob>,
647 options: { readonly maxObjectSize: number },
648): Effect.Effect<Array<DeltaResolved>, PackFormatError | ObjectTooLargeError> =>
649 Effect.gen(function* () {
650 const inflated = new Map<number, Uint8Array>();
651 const baseContent = (index: number) =>
652 Effect.gen(function* () {
653 const cached = inflated.get(index);
654 if (cached !== undefined) return cached;
655 const base = bases[index];
656 if (base === undefined) {
657 return yield* new PackFormatError({
658 reason: `delta base ${index} missing`,
659 });
660 }
661 const content = base.isContent
662 ? base.bytes
663 : yield* inflate(base.bytes).pipe(
664 Effect.mapError(
665 (error) => new PackFormatError({ reason: error.reason }),
666 ),
667 );
668 inflated.set(index, content);
669 return content;
670 });
671 const out: Array<DeltaResolved> = [];
672 for (const job of jobs) {
673 const base = yield* baseContent(job.base);
674 const delta = yield* inflate(job.delta).pipe(
675 Effect.mapError(
676 (error) => new PackFormatError({ reason: error.reason }),
677 ),
678 );
679 const content = yield* applyDelta(base, delta).pipe(
680 Effect.mapError(
681 (error) => new PackFormatError({ reason: error.reason }),
682 ),
683 );
684 if (content.length > options.maxObjectSize) {
685 return yield* new ObjectTooLargeError({
686 size: content.length,
687 limit: options.maxObjectSize,
688 });
689 }
690 const oid = hashObjectSync(job.type, content);
691 const zdata = yield* deflate(content).pipe(
692 Effect.mapError(
693 (error) => new PackFormatError({ reason: error.reason }),
694 ),
695 );
696 out.push({
697 id: job.id,
698 oid,
699 size: content.length,
700 zdata,
701 content: job.type === 3 ? undefined : content,
702 });
703 }
704 return out;
705 });