1/**
2 * Pack ingestion (DESIGN §3.6): a single pass over a pack held behind a
3 * {@link RandomAccess} source, under the push semaphore.
4 *
5 * 1. Validate `PACK`, version 2, count.
6 * 2. Per entry: parse the type/size varint header. Non-delta entries are
7 * streaming-inflated (`Zlib.inflateEntry` — `bytesWritten` gives the exact
8 * compressed span), hashed as `sha1("<type> <size>\0" + content)`, and the
9 * **compressed span is kept verbatim as `zdata`** — no recompression.
10 * Deltas (OFS with the +1-bias offset decoding, REF with a 20-byte base
11 * id) resolve their base from (a) already-ingested entries via a bounded
12 * LRU of resolved contents, (b) re-inflation from the pack on cache miss,
13 * or (c) the object store for thin bases (REF_DELTA to a prior push — the
14 * normal case). The copy/insert instruction stream is applied
15 * (`size==0 ⇒ 0x10000`), the result size verified, the result hashed and
16 * `deflate`d (level 6) for storage.
17 * 3. The trailing pack SHA-1 is verified.
18 *
19 * Resolved entries are emitted to a caller-provided sink as they are
20 * produced; the caller (the Repo DO) stages them with `staged_push = pushId`.
21 *
22 * The parser is written against {@link RandomAccess} from day one so the
23 * v1.x streaming upgrade (R2-multipart tee + ranged reads) swaps one
24 * implementation, not the protocol code (DESIGN §3.6 upgrade seam).
25 */
26import * as Effect from "effect/Effect";
27import * as Result from "effect/Result";
28import * as Schema from "effect/Schema";
29import { applyDelta, DeltaFormatError } from "./Delta.ts";
30import {
31 bytesToHex,
32 decodeOfsDeltaOffset,
33 decodeTypeSize,
34 hashObject,
35 hashObjectSync,
36 isDeltaType,
37 makeSha1,
38 ObjectParseError,
39 type ObjectType,
40 type Oid,
41 type PackEntryType,
42} from "./ObjectCodec.ts";
43import { type ObjectSource, StoreError } from "./Store.ts";
44import {
45 deflate,
46 inflate,
47 inflateEntry,
48 inflateEntrySync,
49 ZlibError,
50} from "./Zlib.ts";
51
52// ─────────────────────────────────────────────────────────────────────────────
53// RandomAccess
54// ─────────────────────────────────────────────────────────────────────────────
55
56/**
57 * Random access over a pack's raw bytes. v1 backs this with the buffered
58 * request body ({@link bufferRandomAccess}); the v1.x streaming upgrade backs
59 * it with R2 ranged reads without touching the parser.
60 */
61export interface RandomAccess {
62 /**
63 * Synchronous read when the range needs no I/O — a view into an
64 * in-memory buffer or an already-cached window — else `undefined`. The
65 * parser's inner loop uses it to stay synchronous between I/O points
66 * (DESIGN §22.5).
67 */
68 readonly readSync?:
69 | ((offset: number, length: number) => Uint8Array | undefined)
70 | undefined;
71 /**
72 * For sources whose total size is not known up front (`size` is
73 * `Infinity` while a body is still arriving — DESIGN §22.6): resolves
74 * with the final size once the body has ended.
75 */
76 readonly awaitEnd?: Effect.Effect<number, StoreError> | undefined;
77 /** The parser is done with every byte below `offset` (retention hint). */
78 readonly release?: ((offset: number) => void) | undefined;
79 /**
80 * True when `offset` has been dropped from memory and the body has NOT
81 * ended yet, i.e. reading it would block on the spill completing. The
82 * parser defers such delta entries until the end (DESIGN §22.6).
83 */
84 readonly evictedBeforeEnd?: ((offset: number) => boolean) | undefined;
85 /** Total size of the pack in bytes (`Infinity` while streaming). */
86 readonly size: number;
87 /**
88 * Reads up to `length` bytes at `offset`. May return fewer bytes only when
89 * the range extends past the end of the pack. Implementations should return
90 * views cheaply (no copy) where possible; callers copy what they retain.
91 */
92 readonly read: (
93 offset: number,
94 length: number,
95 ) => Effect.Effect<Uint8Array, StoreError>;
96}
97
98/**
99 * A {@link RandomAccess} over an in-memory buffer. Reads return subarray
100 * views (zero copy).
101 */
102export const bufferRandomAccess = (buf: Uint8Array): RandomAccess => ({
103 size: buf.length,
104 read: (offset, length) =>
105 Effect.sync(() =>
106 buf.subarray(offset, Math.min(offset + length, buf.length)),
107 ),
108 readSync: (offset, length) =>
109 buf.subarray(offset, Math.min(offset + length, buf.length)),
110 awaitEnd: Effect.succeed(buf.length),
111});
112
113// ─────────────────────────────────────────────────────────────────────────────
114// Errors
115// ─────────────────────────────────────────────────────────────────────────────
116
117/**
118 * Error raised on malformed pack structure: bad magic/version, truncated
119 * entries, entry-count mismatch, size mismatches, or an OFS_DELTA offset that
120 * does not land on an entry boundary.
121 */
122export class PackFormatError extends Schema.TaggedError<PackFormatError>()(
123 "PackFormatError",
124 { reason: Schema.String },
125) {}
126
127/**
128 * Error raised when the trailing pack SHA-1 does not match the hash of the
129 * preceding bytes (`unpack pack checksum mismatch` on the wire).
130 */
131export class PackChecksumMismatch extends Schema.TaggedError<PackChecksumMismatch>()(
132 "PackChecksumMismatch",
133 { expected: Schema.String, actual: Schema.String },
134) {}
135
136/**
137 * Error raised when an object (or a delta result) exceeds the per-object
138 * uncompressed size cap (64 MiB in v1 — DESIGN §3.3).
139 */
140export class ObjectTooLargeError extends Schema.TaggedError<ObjectTooLargeError>()(
141 "ObjectTooLargeError",
142 {
143 size: Schema.Number,
144 limit: Schema.Number,
145 oid: Schema.optional(Schema.String),
146 },
147) {}
148
149/**
150 * Error raised when a REF_DELTA's base cannot be found in the pack or in the
151 * object store (a thin pack whose base we do not have).
152 */
153/**
154 * A delta's base was evicted from a streaming source before the body ended
155 * (DESIGN §22.6). Retried after the end, when the spilled object is
156 * readable; never reaches callers.
157 */
158export class BaseEvictedError extends Schema.TaggedError<BaseEvictedError>()(
159 "BaseEvictedError",
160 { offset: Schema.Number },
161) {}
162
163export class MissingDeltaBaseError extends Schema.TaggedError<MissingDeltaBaseError>()(
164 "MissingDeltaBaseError",
165 { baseOid: Schema.String },
166) {}
167
168/**
169 * The full error union pack ingestion can produce.
170 */
171export type PackIngestError =
172 | PackFormatError
173 | PackChecksumMismatch
174 | ObjectTooLargeError
175 | MissingDeltaBaseError
176 | ZlibError
177 | DeltaFormatError
178 | ObjectParseError
179 | StoreError
180 | BaseEvictedError;
181
182// ─────────────────────────────────────────────────────────────────────────────
183// Resolved entries / options
184// ─────────────────────────────────────────────────────────────────────────────
185
186/**
187 * One fully resolved object produced by pack ingestion.
188 *
189 * `zdata` is `zlib(content)` **without** the loose header — for non-delta
190 * entries it is the pack's compressed span verbatim; for delta-resolved
191 * entries it is a fresh `deflate(content)` at level 6.
192 */
193export interface ResolvedEntry {
194 readonly oid: Oid;
195 readonly type: ObjectType;
196 /** Uncompressed size in bytes. */
197 readonly size: number;
198 /** `zlib(content)`, no loose header. Safe to retain (always a copy). */
199 readonly zdata: Uint8Array;
200 /** `true` when the entry arrived as a delta and was re-deflated. */
201 readonly fromDelta: boolean;
202 /**
203 * Byte offset of `zdata` within the pack SOURCE for non-delta entries
204 * (the compressed span is the pack's bytes verbatim, so a store may
205 * reference it in place — DESIGN §22.5); `-1` for delta-resolved entries,
206 * whose `zdata` is a fresh deflate that exists nowhere in the pack.
207 */
208 readonly dataOffset: number;
209 /**
210 * The inflated bytes. The parser already holds them at this point, so
211 * passing them through spares every consumer a second inflate just to
212 * read a commit or tree.
213 */
214 readonly content: Uint8Array;
215}
216
217/**
218 * Summary returned by {@link ingestPack} after the trailer verifies.
219 */
220export interface IngestSummary {
221 /** Number of entries in the pack (the header count). */
222 readonly count: number;
223 /** The oids of every resolved entry, in emission order. */
224 readonly oids: ReadonlyArray<Oid>;
225}
226
227/** Default per-object uncompressed size cap: 64 MiB (DESIGN §3.3). */
228export const DEFAULT_MAX_OBJECT_SIZE = 64 * 1024 * 1024;
229
230/** Default resolved-content LRU budget: 20 MiB (DESIGN §3.6). */
231export const DEFAULT_CACHE_BYTES = 20 * 1024 * 1024;
232
233/** Per-entry cache admission cap: entries larger than this are never cached. */
234const MAX_CACHE_ENTRY_BYTES = 10 * 1024 * 1024;
235
236/**
237 * Bytes read per entry before growing (see the read in the main pass).
238 * Sized so virtually every git object — a few KiB — is covered by one read,
239 * which is what keeps ingest from a remote source proportional to the pack
240 * size rather than to entries × pack size.
241 */
242/**
243 * Bytes probed per entry before inflating. In-window reads are views on
244 * both sources, so the probe's size costs nothing — except when it crosses
245 * a window edge on the spilled (blob-backed) source, where the reader must
246 * assemble a copy. 64 KiB keeps such crossings rare (only entries within
247 * 64 KiB of a 4 MiB edge) and small; the rare larger entry grows the
248 * window on demand (DESIGN §22.4 — 512 KiB here was ~1 GB of copies per
249 * 15k-object spilled push).
250 */
251const ENTRY_WINDOW_BYTES = 64 * 1024;
252
253/** Non-delta entries handed to `sinkBatch` per fiber hop (DESIGN §22.5). */
254export const SINK_BATCH = 256;
255
256/**
257 * Options for {@link ingestPack}.
258 */
259/** What the parser needs from the live store: thin-delta bases. */
260export interface ThinBaseSource {
261 /** A live object's type and content, or `undefined` when absent. */
262 readonly readBase: (
263 oid: Oid,
264 ) => Effect.Effect<
265 { readonly type: ObjectType; readonly content: Uint8Array } | undefined,
266 StoreError
267 >;
268}
269
270export interface IngestPackOptions<E, R> {
271 /** The pack bytes. */
272 readonly source: RandomAccess;
273 /** Live objects, used to resolve thin REF_DELTA bases. */
274 readonly store: ThinBaseSource;
275 /** Receives each resolved entry as it is produced (staging inserts). */
276 readonly sink: (entry: ResolvedEntry) => Effect.Effect<void, E, R>;
277 /**
278 * Optional batched sink: receives runs of entries the synchronous inner
279 * loop produced without any I/O between them (DESIGN §22.5). When set,
280 * non-delta entries go here in batches of up to `SINK_BATCH` and `sink`
281 * receives only delta-resolved entries.
282 */
283 readonly sinkBatch?:
284 | ((entries: ReadonlyArray<ResolvedEntry>) => Effect.Effect<void, E, R>)
285 | undefined;
286 /** Per-object uncompressed cap; default {@link DEFAULT_MAX_OBJECT_SIZE}. */
287 readonly maxObjectSize?: number | undefined;
288 /**
289 * Optional per-phase CPU accounting (ms), accumulated in place:
290 * `inflate`, `hash`, `delta`, `deflate`, `copy`, `sink`. Cheap enough
291 * to leave on: two `performance.now()` calls per phase per object.
292 */
293 readonly phases?: Record<string, number> | undefined;
294 /** Resolved-content LRU budget; default {@link DEFAULT_CACHE_BYTES}. */
295 readonly cacheBytes?: number | undefined;
296}
297
298// ─────────────────────────────────────────────────────────────────────────────
299// Internals
300// ─────────────────────────────────────────────────────────────────────────────
301
302interface CachedContent {
303 readonly type: ObjectType;
304 readonly content: Uint8Array;
305}
306
307/** Insertion-order LRU bounded by total content bytes. */
308class ByteLru {
309 private readonly map = new Map<string, CachedContent>();
310 private bytes = 0;
311 constructor(private readonly capacity: number) {}
312
313 get(key: string): CachedContent | undefined {
314 const hit = this.map.get(key);
315 if (hit !== undefined) {
316 // refresh recency
317 this.map.delete(key);
318 this.map.set(key, hit);
319 }
320 return hit;
321 }
322
323 set(key: string, entry: CachedContent): void {
324 if (entry.content.length > MAX_CACHE_ENTRY_BYTES) return;
325 if (this.map.has(key)) return;
326 this.map.set(key, entry);
327 this.bytes += entry.content.length;
328 while (this.bytes > this.capacity && this.map.size > 1) {
329 const oldest = this.map.keys().next().value as string;
330 const evicted = this.map.get(oldest)!;
331 this.map.delete(oldest);
332 this.bytes -= evicted.content.length;
333 }
334 }
335}
336
337interface IndexedEntry {
338 /** Absolute offset of the entry header in the pack. */
339 readonly offset: number;
340 readonly entryType: PackEntryType;
341 /** Declared uncompressed size (for deltas: of the delta payload). */
342 readonly declaredSize: number;
343 /** Absolute offset of the zlib stream. */
344 readonly dataOffset: number;
345 /** Exact compressed span in bytes. */
346 readonly span: number;
347 /** OFS_DELTA: absolute offset of the base entry. */
348 readonly baseOffset: number | undefined;
349 /** REF_DELTA: base object id. */
350 readonly baseOid: Oid | undefined;
351 /** Set once the entry has been resolved and emitted. */
352 resolved?: { readonly oid: Oid; readonly type: ObjectType };
353}
354
355const readU32BE = (buf: Uint8Array, offset: number): number =>
356 ((buf[offset]! << 24) |
357 (buf[offset + 1]! << 16) |
358 (buf[offset + 2]! << 8) |
359 buf[offset + 3]!) >>>
360 0;
361
362/**
363 * Reads and validates the 12-byte pack header (`PACK`, version 2, entry
364 * count).
365 */
366export const readPackHeader = Effect.fn(function* (source: RandomAccess) {
367 if (Number.isFinite(source.size) && source.size < 12 + 20) {
368 return yield* new PackFormatError({
369 reason: `pack too small: ${source.size} bytes`,
370 });
371 }
372 const header = yield* source.read(0, 12);
373 if (header.length < 12) {
374 return yield* new PackFormatError({
375 reason: `pack too small: ${header.length} bytes`,
376 });
377 }
378 if (
379 header.length < 12 ||
380 header[0] !== 0x50 || // P
381 header[1] !== 0x41 || // A
382 header[2] !== 0x43 || // C
383 header[3] !== 0x4b // K
384 ) {
385 return yield* new PackFormatError({ reason: "bad pack magic" });
386 }
387 const version = readU32BE(header, 4);
388 if (version !== 2) {
389 return yield* new PackFormatError({
390 reason: `unsupported pack version ${version}`,
391 });
392 }
393 const count = readU32BE(header, 8);
394 return { version, count };
395});
396
397// ─────────────────────────────────────────────────────────────────────────────
398// Ingest
399// ─────────────────────────────────────────────────────────────────────────────
400
401/**
402 * Ingests a complete pack: parses every entry, resolves OFS/REF deltas
403 * (including thin bases from the object store), verifies the trailing SHA-1,
404 * and emits each resolved object to `options.sink`.
405 *
406 * Entries are emitted in pack order, except REF_DELTA entries whose base
407 * appears *later* in the pack (legal but rare) — those are deferred and
408 * emitted after the main pass.
409 *
410 * Memory: the resolved-content LRU is bounded by `cacheBytes`; cache misses
411 * re-inflate from the source. Only one resolved object (plus its base) is
412 * otherwise live at a time.
413 */
414export const ingestPack = <E, R>(
415 options: IngestPackOptions<E, R>,
416): Effect.Effect<IngestSummary, PackIngestError | E, R> =>
417 Effect.gen(function* () {
418 const { sink, source, store } = options;
419 const phases = options.phases;
420 const timed = <A, E2, R2>(
421 phase: string,
422 effect: Effect.Effect<A, E2, R2>,
423 ): Effect.Effect<A, E2, R2> =>
424 phases === undefined
425 ? effect
426 : Effect.suspend(() => {
427 const at = performance.now();
428 return Effect.ensuring(
429 effect,
430 Effect.sync(() => {
431 phases[phase] = (phases[phase] ?? 0) + (performance.now() - at);
432 }),
433 );
434 });
435 const maxObjectSize = options.maxObjectSize ?? DEFAULT_MAX_OBJECT_SIZE;
436 const cache = new ByteLru(options.cacheBytes ?? DEFAULT_CACHE_BYTES);
437 const { count } = yield* readPackHeader(source);
438 // Unknown while a streaming body is still arriving (DESIGN §22.6); the
439 // loop is driven by `count`, and the true end is learned from
440 // `awaitEnd` before the trailer is checked.
441 let dataEnd = Number.isFinite(source.size)
442 ? source.size - 20
443 : Number.POSITIVE_INFINITY;
444 // The trailer SHA-1 is accumulated as entries are consumed instead of
445 // re-reading the whole pack afterwards (on a spilled source that was
446 // another full pass of window reads).
447 const trailerSha = makeSha1();
448 trailerSha.update(yield* source.read(0, 12));
449
450 const byOffset = new Map<number, IndexedEntry>();
451 /** oid → entry offset, for in-pack REF_DELTA base lookup. */
452 const offsetByOid = new Map<Oid, number>();
453 const oids: Array<Oid> = [];
454 const deferred: Array<IndexedEntry> = [];
455
456 /** Inflates an already-indexed entry's own zlib stream (exact span). */
457 const inflateOwn = (
458 entry: IndexedEntry,
459 ): Effect.Effect<Uint8Array, PackIngestError> =>
460 Effect.suspend((): Effect.Effect<Uint8Array, PackIngestError> =>
461 source.evictedBeforeEnd?.(entry.dataOffset) === true
462 ? Effect.fail(new BaseEvictedError({ offset: entry.dataOffset }))
463 : source
464 .read(entry.dataOffset, entry.span)
465 .pipe(Effect.flatMap((z) => inflate(z))),
466 );
467
468 /**
469 * Resolves an entry to its full content, walking delta chains. Bases come
470 * from the LRU, from re-inflation of earlier pack entries, or (thin
471 * REF_DELTA) from the object store.
472 */
473 const resolveContent = (
474 entry: IndexedEntry,
475 ): Effect.Effect<CachedContent, PackIngestError> =>
476 Effect.gen(function* () {
477 const key = `ofs:${entry.offset}`;
478 const hit = cache.get(key);
479 if (hit !== undefined) return hit;
480 let result: CachedContent;
481 if (!isDeltaType(entry.entryType)) {
482 const content = yield* inflateOwn(entry);
483 result = { type: entry.entryType as ObjectType, content };
484 } else {
485 const base = yield* resolveBase(entry);
486 const payload = yield* timed("inflate", inflateOwn(entry));
487 const content = yield* timed(
488 "delta",
489 applyDelta(base.content, payload),
490 );
491 if (content.length > maxObjectSize) {
492 return yield* new ObjectTooLargeError({
493 size: content.length,
494 limit: maxObjectSize,
495 });
496 }
497 result = { type: base.type, content };
498 }
499 cache.set(key, result);
500 return result;
501 });
502
503 /** Resolves a delta entry's base content. */
504 const resolveBase = (
505 entry: IndexedEntry,
506 ): Effect.Effect<CachedContent, PackIngestError> =>
507 Effect.gen(function* () {
508 if (entry.baseOffset !== undefined) {
509 const base = byOffset.get(entry.baseOffset);
510 if (base === undefined) {
511 return yield* new PackFormatError({
512 reason: `ofs-delta at ${entry.offset} points at ${entry.baseOffset}, not an entry boundary`,
513 });
514 }
515 return yield* resolveContent(base);
516 }
517 const baseOid = entry.baseOid!;
518 const inPack = offsetByOid.get(baseOid);
519 if (inPack !== undefined) {
520 return yield* resolveContent(byOffset.get(inPack)!);
521 }
522 const cachedThin = cache.get(`oid:${baseOid}`);
523 if (cachedThin !== undefined) return cachedThin;
524 const live = yield* store.readBase(baseOid);
525 if (live === undefined) {
526 return yield* new MissingDeltaBaseError({ baseOid });
527 }
528 if (live.content.length > maxObjectSize) {
529 return yield* new ObjectTooLargeError({
530 size: live.content.length,
531 limit: maxObjectSize,
532 oid: baseOid,
533 });
534 }
535 const thin: CachedContent = { type: live.type, content: live.content };
536 cache.set(`oid:${baseOid}`, thin);
537 return thin;
538 });
539
540 /** Resolves, hashes, deflates, and emits one delta entry. */
541 const emitDelta = (entry: IndexedEntry) =>
542 Effect.gen(function* () {
543 const { content, type } = yield* resolveContent(entry);
544 const oid = yield* timed("hash", hashObject(type, content));
545 const zdata = yield* timed("deflate", deflate(content));
546 entry.resolved = { oid, type };
547 offsetByOid.set(oid, entry.offset);
548 oids.push(oid);
549 yield* timed(
550 "sink",
551 sink({
552 oid,
553 type,
554 size: content.length,
555 zdata,
556 fromDelta: true,
557 dataOffset: -1,
558 content,
559 }),
560 );
561 });
562
563 // ── main pass: index + resolve in pack order ─────────────────────────────
564 const sinkBatch = options.sinkBatch;
565 let pendingSink: Array<ResolvedEntry> = [];
566 // On a streaming source the body arrives through the event loop, and
567 // a synchronous batch monopolizes the isolate: without a real
568 // (macrotask) yield between batches, receive and parse alternate
569 // instead of overlapping (DESIGN §22.6). Fiber yields are not enough —
570 // they stay within the same task.
571 const streaming = !Number.isFinite(source.size);
572 const breathe = streaming
573 ? Effect.promise(
574 () => new Promise<void>((resolve) => setTimeout(resolve, 0)),
575 )
576 : Effect.void;
577 const flushPending = Effect.suspend(() => {
578 if (pendingSink.length === 0 || sinkBatch === undefined) return breathe;
579 const batch = pendingSink;
580 pendingSink = [];
581 return timed("sink", sinkBatch(batch)).pipe(Effect.andThen(breathe));
582 });
583 /**
584 * The synchronous fast path (DESIGN §22.5): a non-delta entry whose
585 * bytes are already in memory is decoded, inflated, hashed and copied
586 * without a single fiber hop. Returns the next offset, `undefined` to
587 * fall through to the general path (a delta, an entry outrunning the
588 * window, or no sync inflate), or an error to fail with.
589 */
590 const syncEntry = (
591 window: Uint8Array,
592 offset: number,
593 i: number,
594 ): number | undefined | PackFormatError | ObjectTooLargeError => {
595 let header;
596 try {
597 header = decodeTypeSize(window, 0);
598 } catch (error) {
599 return new PackFormatError({
600 reason:
601 error instanceof ObjectParseError
602 ? `entry ${i}: ${error.reason}`
603 : `entry ${i}: ${String(error)}`,
604 });
605 }
606 if (isDeltaType(header.type)) return undefined;
607 if (header.size > maxObjectSize) {
608 return new ObjectTooLargeError({
609 size: header.size,
610 limit: maxObjectSize,
611 });
612 }
613 const pos = header.next;
614 const inflated = inflateEntrySync(window, pos, {
615 maxOutput: maxObjectSize,
616 expectedSize: header.size,
617 });
618 if (inflated === undefined) return undefined;
619 const { bytesConsumed, content } = inflated;
620 if (content.length !== header.size) {
621 return new PackFormatError({
622 reason: `entry ${i}: inflated ${content.length} bytes, header declared ${header.size}`,
623 });
624 }
625 const type = header.type as ObjectType;
626 const oid = hashObjectSync(type, content);
627 const dataOffset = offset + pos;
628 const zdata = Uint8Array.from(window.subarray(pos, pos + bytesConsumed));
629 const entry: IndexedEntry = {
630 offset,
631 entryType: header.type,
632 declaredSize: header.size,
633 dataOffset,
634 span: bytesConsumed,
635 baseOffset: undefined,
636 baseOid: undefined,
637 resolved: { oid, type },
638 };
639 byOffset.set(offset, entry);
640 offsetByOid.set(oid, offset);
641 cache.set(`ofs:${offset}`, { type, content });
642 oids.push(oid);
643 pendingSink.push({
644 oid,
645 type,
646 size: content.length,
647 zdata,
648 fromDelta: false,
649 dataOffset,
650 content,
651 });
652 trailerSha.update(window.subarray(0, pos + bytesConsumed));
653 source.release?.(offset);
654 return dataOffset + bytesConsumed;
655 };
656
657 let offset = 12;
658 for (let i = 0; i < count; i++) {
659 if (offset >= dataEnd) {
660 return yield* new PackFormatError({
661 reason: `truncated pack: entry ${i} of ${count} starts past the trailer`,
662 });
663 }
664 // Read a BOUNDED window, not the rest of the pack.
665 //
666 // `source.read(offset, dataEnd - offset)` is free on a buffer (a
667 // subarray view) but catastrophic on a remote source: it copies the
668 // whole remainder once per entry, so a 13.7k-object pack in R2 moved
669 // hundreds of GB and took minutes instead of seconds. An entry is
670 // almost always a few KiB, so one small window covers it; the rare
671 // large object grows the window on demand (below).
672 const probeLength = Math.min(ENTRY_WINDOW_BYTES, dataEnd - offset);
673 const direct = source.readSync?.(offset, probeLength);
674 if (direct !== undefined) {
675 const at = phases === undefined ? 0 : performance.now();
676 const next = syncEntry(direct, offset, i);
677 if (phases !== undefined) {
678 phases["sync"] = (phases["sync"] ?? 0) + (performance.now() - at);
679 }
680 if (typeof next === "number") {
681 offset = next;
682 if (sinkBatch === undefined) {
683 const entry = pendingSink.pop()!;
684 yield* timed("sink", sink(entry));
685 } else if (pendingSink.length >= SINK_BATCH) {
686 yield* flushPending;
687 }
688 continue;
689 }
690 if (next !== undefined) return yield* Effect.fail(next);
691 }
692 let window = yield* source.read(offset, probeLength);
693 if (window.length === 0) {
694 return yield* new PackFormatError({
695 reason: `truncated pack: entry ${i} of ${count} starts past the end`,
696 });
697 }
698 /**
699 * Re-reads a bigger window when an entry's compressed stream runs past
700 * the current one. Doubling from the entry window reaches the whole
701 * remainder in a handful of steps, and only for objects that need it.
702 */
703 const growWindow = Effect.fn(function* (needed: number) {
704 const size = Math.min(
705 Math.max(needed, window.length * 2),
706 dataEnd - offset,
707 );
708 if (size <= window.length) return false;
709 window = yield* source.read(offset, size);
710 return true;
711 });
712 let header;
713 try {
714 header = decodeTypeSize(window, 0);
715 } catch (error) {
716 return yield* new PackFormatError({
717 reason:
718 error instanceof ObjectParseError
719 ? `entry ${i}: ${error.reason}`
720 : `entry ${i}: ${String(error)}`,
721 });
722 }
723 let pos = header.next;
724 let baseOffset: number | undefined;
725 let baseOid: Oid | undefined;
726 if (header.type === 6) {
727 let ofs;
728 try {
729 ofs = decodeOfsDeltaOffset(window, pos);
730 } catch (error) {
731 return yield* new PackFormatError({
732 reason:
733 error instanceof ObjectParseError
734 ? `entry ${i}: ${error.reason}`
735 : `entry ${i}: ${String(error)}`,
736 });
737 }
738 pos = ofs.next;
739 baseOffset = offset - ofs.value;
740 if (baseOffset < 12) {
741 return yield* new PackFormatError({
742 reason: `entry ${i}: ofs-delta offset ${ofs.value} points before the first entry`,
743 });
744 }
745 } else if (header.type === 7) {
746 if (pos + 20 > window.length) {
747 return yield* new PackFormatError({
748 reason: `entry ${i}: truncated ref-delta base id`,
749 });
750 }
751 baseOid = bytesToHex(window.subarray(pos, pos + 20));
752 pos += 20;
753 } else if (header.size > maxObjectSize) {
754 return yield* new ObjectTooLargeError({
755 size: header.size,
756 limit: maxObjectSize,
757 });
758 }
759
760 // An entry whose compressed stream runs past the window grows it and
761 // retries; virtually every object fits the first read.
762 const inflateOptions = {
763 maxOutput: maxObjectSize,
764 // Lets the synchronous fast path verify itself (see Zlib.ts).
765 expectedSize: header.size,
766 };
767 let attempt = yield* Effect.result(
768 timed("inflate", inflateEntry(window, pos, inflateOptions)),
769 );
770 // The header declares the uncompressed size; the compressed span is
771 // at most that plus deflate's own overhead, so one grow reaches an
772 // entry that outran the probe instead of doubling toward it (each
773 // failed attempt below costs up to three inflate passes).
774 while (
775 Result.isFailure(attempt) &&
776 (yield* growWindow(
777 Math.max(window.length * 2, pos + header.size + 64 * 1024),
778 ))
779 ) {
780 attempt = yield* Effect.result(
781 inflateEntry(window, pos, inflateOptions),
782 );
783 }
784 if (Result.isFailure(attempt)) {
785 return yield* Effect.fail(attempt.failure);
786 }
787 const { bytesConsumed, content } = attempt.success;
788 if (content.length !== header.size) {
789 return yield* new PackFormatError({
790 reason: `entry ${i}: inflated ${content.length} bytes, header declared ${header.size}`,
791 });
792 }
793 const dataOffset = offset + pos;
794 const entry: IndexedEntry = {
795 offset,
796 entryType: header.type,
797 declaredSize: header.size,
798 dataOffset,
799 span: bytesConsumed,
800 baseOffset,
801 baseOid,
802 };
803 byOffset.set(offset, entry);
804
805 if (!isDeltaType(header.type)) {
806 const type = header.type as ObjectType;
807 const oid = yield* timed("hash", hashObject(type, content));
808 // keep the compressed span verbatim — copy it out of the shared buffer
809 const zview = yield* source.read(dataOffset, bytesConsumed);
810 const zdata = yield* timed(
811 "copy",
812 Effect.sync(() => Uint8Array.from(zview)),
813 );
814 entry.resolved = { oid, type };
815 offsetByOid.set(oid, entry.offset);
816 cache.set(`ofs:${offset}`, { type, content });
817 oids.push(oid);
818 yield* timed(
819 "sink",
820 sink({
821 oid,
822 type,
823 size: content.length,
824 zdata,
825 fromDelta: false,
826 dataOffset,
827 content,
828 }),
829 );
830 } else {
831 const result = yield* Effect.result(emitDelta(entry));
832 if (Result.isFailure(result)) {
833 if (
834 result.failure instanceof MissingDeltaBaseError ||
835 result.failure instanceof BaseEvictedError
836 ) {
837 // The base may appear later in the pack (thin ref-delta), or
838 // it was evicted from a streaming source and is readable once
839 // the body has ended: retry in the deferred pass.
840 deferred.push(entry);
841 } else {
842 return yield* Effect.fail(result.failure);
843 }
844 }
845 }
846 trailerSha.update(window.subarray(0, pos + bytesConsumed));
847 source.release?.(offset);
848 offset = dataOffset + bytesConsumed;
849 }
850
851 yield* flushPending;
852 if (!Number.isFinite(dataEnd)) {
853 if (source.awaitEnd === undefined) {
854 return yield* new PackFormatError({
855 reason: "pack source has unknown size and no awaitEnd",
856 });
857 }
858 dataEnd = (yield* source.awaitEnd) - 20;
859 }
860 // ── deferred deltas: fixpoint passes AFTER the end ───────────────────────
861 // REF_DELTAs whose base came later in the pack, and (streaming) deltas
862 // whose base had been evicted before the body ended — readable now
863 // through the spilled object.
864 let pending = deferred;
865 while (pending.length > 0) {
866 const next: Array<IndexedEntry> = [];
867 let firstMissing: MissingDeltaBaseError | undefined;
868 for (const entry of pending) {
869 const result = yield* Effect.result(emitDelta(entry));
870 if (Result.isFailure(result)) {
871 if (result.failure instanceof MissingDeltaBaseError) {
872 firstMissing ??= result.failure;
873 next.push(entry);
874 } else if (result.failure instanceof BaseEvictedError) {
875 return yield* new PackFormatError({
876 reason: `delta base at ${result.failure.offset} unreadable after the body ended`,
877 });
878 } else {
879 return yield* Effect.fail(result.failure);
880 }
881 }
882 }
883 if (next.length === pending.length) {
884 // no progress — the base genuinely does not exist anywhere
885 return yield* Effect.fail(firstMissing!);
886 }
887 pending = next;
888 }
889 if (offset !== dataEnd) {
890 return yield* new PackFormatError({
891 reason: `pack has ${dataEnd - offset} unconsumed bytes after ${count} entries`,
892 });
893 }
894
895 // ── trailer verification ─────────────────────────────────────────────────
896 const trailer = yield* source.read(dataEnd, 20);
897 if (trailer.length < 20) {
898 return yield* new PackFormatError({ reason: "truncated pack trailer" });
899 }
900 const expected = bytesToHex(trailer);
901 const actual = yield* Effect.sync(() => trailerSha.digestHex());
902 if (expected !== actual) {
903 return yield* new PackChecksumMismatch({ expected, actual });
904 }
905
906 return { count, oids } satisfies IngestSummary;
907 });