1/**
2 * A {@link RandomAccess} over a blob-store object, for pack ingestion
3 * (DESIGN.md §3.6 upgrade seam) — this is what removes the push size cap.
4 *
5 * v1 buffered the whole incoming pack in the isolate, which forced a hard
6 * 50 MiB limit: a bigger push could not be held in 128 MB of memory
7 * alongside delta bases. Instead the body is streamed to the blob store
8 * and parsed *from there*, so ingest memory is bounded by the read window
9 * rather than by the pack size. `PackParser` was written against
10 * `RandomAccess` from the start precisely so this swap touches one
11 * implementation and no protocol code.
12 *
13 * Reads are served from window-aligned slabs (the same technique the object
14 * store uses for compacted packs): pack parsing is overwhelmingly
15 * sequential, so one window fetch serves thousands of consecutive reads,
16 * and the occasional backwards seek for an OFS_DELTA base usually lands in
17 * a retained window too.
18 */
19import { RuntimeContext } from "../../RuntimeContext.ts";
20import * as Effect from "effect/Effect";
21import type { BlobStoreShape } from "../BlobStore.ts";
22import type { RandomAccess } from "../Protocol/PackParser.ts";
23import { StoreError } from "../Protocol/Store.ts";
24
25/** Bytes fetched per window. */
26export const PACK_WINDOW_BYTES = 4 * 1024 * 1024;
27
28/** Windows retained (insertion-ordered LRU) — bounds ingest memory. */
29export const PACK_MAX_WINDOWS = 4;
30
31/**
32 * Opens a blob-store object as a {@link RandomAccess}. `size` must be the
33 * object's exact byte length (the caller knows it: it just uploaded it).
34 */
35export const blobRandomAccess = (options: {
36 readonly blobs: BlobStoreShape;
37 readonly key: string;
38 readonly size: number;
39 readonly windowBytes?: number | undefined;
40 readonly maxWindows?: number | undefined;
41}): RandomAccess => {
42 const windowBytes = options.windowBytes ?? PACK_WINDOW_BYTES;
43 const maxWindows = options.maxWindows ?? PACK_MAX_WINDOWS;
44 const windows = new Map<number, { start: number; bytes: Uint8Array }>();
45
46 const fetchWindow = (index: number) =>
47 Effect.gen(function* () {
48 const hit = windows.get(index);
49 if (hit !== undefined) {
50 windows.delete(index);
51 windows.set(index, hit);
52 return hit;
53 }
54 const start = index * windowBytes;
55 const body = yield* options.blobs
56 .get(options.key, { offset: start, length: windowBytes })
57 .pipe(
58 Effect.mapError(
59 (error) =>
60 new StoreError({
61 reason: `incoming pack read ${options.key}: ${error.reason}`,
62 }),
63 ),
64 Effect.provide(RuntimeContext.phantom),
65 );
66 if (body === null) {
67 return yield* Effect.fail(
68 new StoreError({ reason: `incoming pack missing: ${options.key}` }),
69 );
70 }
71 const bytes = yield* body.bytes.pipe(
72 Effect.mapError(
73 (error) =>
74 new StoreError({
75 reason: `incoming pack read ${options.key}: ${error.reason}`,
76 }),
77 ),
78 );
79 const slab = { start, bytes };
80 windows.set(index, slab);
81 while (windows.size > maxWindows) {
82 const oldest = windows.keys().next().value;
83 if (oldest === undefined) break;
84 windows.delete(oldest);
85 }
86 return slab;
87 });
88
89 return {
90 size: options.size,
91 awaitEnd: Effect.succeed(options.size),
92 readSync: (offset, length) => {
93 const end = Math.min(offset + length, options.size);
94 if (end <= offset) return new Uint8Array(0);
95 const first = Math.floor(offset / windowBytes);
96 if (first !== Math.floor((end - 1) / windowBytes)) return undefined;
97 const slab = windows.get(first);
98 if (slab === undefined) return undefined;
99 const from = offset - slab.start;
100 return slab.bytes.subarray(from, from + (end - offset));
101 },
102 read: (offset, length) =>
103 Effect.gen(function* () {
104 const end = Math.min(offset + length, options.size);
105 if (end <= offset) return new Uint8Array(0);
106 const first = Math.floor(offset / windowBytes);
107 const last = Math.floor((end - 1) / windowBytes);
108 if (first === last) {
109 // A VIEW, not a copy (DESIGN §22.4). The parser probes every
110 // entry with a window-sized read and copies only what it keeps;
111 // returning a slice here copied that probe window per entry —
112 // 15k entries × the probe size — which on a spilled 40 MiB push
113 // was the difference between ~6 s and ~40 s of ingest. The slab
114 // stays alive as long as any view does, so eviction is safe.
115 const slab = yield* fetchWindow(first);
116 const from = offset - slab.start;
117 return slab.bytes.subarray(from, from + (end - offset));
118 }
119 // A read spanning window boundaries: stitch the windows it covers.
120 const out = new Uint8Array(end - offset);
121 let written = 0;
122 for (let index = first; index <= last; index++) {
123 const slab = yield* fetchWindow(index);
124 const chunkStart = Math.max(offset, slab.start);
125 const chunkEnd = Math.min(end, slab.start + slab.bytes.length);
126 if (chunkEnd <= chunkStart) continue;
127 out.set(
128 slab.bytes.subarray(chunkStart - slab.start, chunkEnd - slab.start),
129 written,
130 );
131 written += chunkEnd - chunkStart;
132 }
133 return written === out.length ? out : out.subarray(0, written);
134 }),
135 };
136};
137
138/**
139 * A view of `source` starting at `start` — used to address the pack that
140 * follows the command section inside a spilled receive-pack body.
141 */
142export const sliceRandomAccess = (
143 source: RandomAccess,
144 start: number,
145): RandomAccess => ({
146 size: source.size - start,
147 read: (offset, length) => source.read(start + offset, length),
148 readSync:
149 source.readSync === undefined
150 ? undefined
151 : (offset, length) => source.readSync!(start + offset, length),
152 awaitEnd:
153 source.awaitEnd === undefined
154 ? undefined
155 : source.awaitEnd.pipe(Effect.map((total) => total - start)),
156 release:
157 source.release === undefined
158 ? undefined
159 : (offset) => source.release!(start + offset),
160});