1/**
2 * The spilled-pack reader (src/Git/Store/PackSource.ts): windowed random
3 * access over blob storage. Reads within a window must be VIEWS of the
4 * cached slab (a copy per probe read was the 40 s spilled-push ingest), and
5 * a parse through tiny windows — many window boundaries, straddling
6 * entries — must produce exactly what the in-memory parse produces.
7 */
8import {
9 hashObject,
10 makeSha1,
11 encodeTypeSize,
12 type Oid,
13} from "@/Git/Protocol/ObjectCodec.ts";
14import {
15 bufferRandomAccess,
16 ingestPack,
17 SINK_BATCH,
18} from "@/Git/Protocol/PackParser.ts";
19import { packHeader } from "@/Git/Protocol/PackWriter.ts";
20import * as Zlib from "@/Git/Protocol/Zlib.ts";
21import { makeObjectStore } from "@/Git/Store/ObjectStore.ts";
22import { blobRandomAccess, sliceRandomAccess } from "@/Git/Store/PackSource.ts";
23import { makeStreamingSource } from "@/Git/Store/StreamingSource.ts";
24import * as Fiber from "effect/Fiber";
25import { describe, expect, test } from "alchemy-test";
26import * as Effect from "effect/Effect";
27import { RuntimeContext } from "@/RuntimeContext.ts";
28import { concat } from "./harness/pack.ts";
29import { makeMemoryBlobStore, makeTestSqlClient } from "./harness/store.ts";
30
31/** A synthetic non-delta pack of `n` blobs with sizes cycling 100..5000. */
32const buildPack = (
33 n: number,
34 options?: { readonly incompressible?: boolean },
35) =>
36 Effect.gen(function* () {
37 const pieces: Array<Uint8Array> = [packHeader(n)];
38 const oids: Array<Oid> = [];
39 for (let i = 0; i < n; i++) {
40 const size = 100 + ((i * 977) % 4900);
41 const content = new Uint8Array(size);
42 if (options?.incompressible) crypto.getRandomValues(content);
43 else for (let j = 0; j < size; j++) content[j] = (j * 7 + i) & 0xff;
44 content[0] = i & 0xff;
45 oids.push(yield* hashObject(3, content));
46 pieces.push(encodeTypeSize(3, size), yield* Zlib.deflate(content));
47 }
48 const body = concat(pieces);
49 const sha = makeSha1();
50 sha.update(body);
51 return { pack: concat([body, sha.digest()]), oids };
52 });
53
54const parseWith = (source: ReturnType<typeof bufferRandomAccess>) =>
55 Effect.gen(function* () {
56 const store = makeObjectStore({
57 sql: makeTestSqlClient(),
58 blobs: makeMemoryBlobStore(),
59 repoId: "R",
60 });
61 const seen: Array<string> = [];
62 const offsets: Array<{
63 dataOffset: number;
64 zdata: Uint8Array;
65 fromDelta: boolean;
66 }> = [];
67 const summary = yield* ingestPack({
68 source,
69 store,
70 sink: (entry) =>
71 Effect.sync(() => {
72 seen.push(entry.oid);
73 offsets.push({
74 dataOffset: entry.dataOffset,
75 zdata: entry.zdata,
76 fromDelta: entry.fromDelta,
77 });
78 }),
79 });
80 return { count: summary.count, seen, offsets };
81 });
82
83describe("blobRandomAccess", { tags: ["unit", "local"] }, () => {
84 test("reads inside one window are views of the cached slab, not copies", async () => {
85 await Effect.runPromise(
86 Effect.gen(function* () {
87 const blobs = makeMemoryBlobStore();
88 const bytes = new Uint8Array(10_000);
89 for (let i = 0; i < bytes.length; i++) bytes[i] = i & 0xff;
90 yield* blobs.put("k", bytes);
91 const source = blobRandomAccess({
92 blobs,
93 key: "k",
94 size: bytes.length,
95 windowBytes: 4096,
96 maxWindows: 2,
97 });
98 const a = yield* source.read(10, 100);
99 const b = yield* source.read(200, 1000);
100 expect(a.buffer).toBe(b.buffer); // same slab
101 expect(a[0]).toBe(10);
102 expect(b[0]).toBe(200);
103 expect(blobs.gets.length).toBe(1); // one window fetch served both
104 // Crossing a window boundary assembles a fresh buffer.
105 const c = yield* source.read(4000, 200);
106 expect(c.length).toBe(200);
107 expect(c[96]).toBe(4096 & 0xff);
108 expect(blobs.gets.length).toBe(2);
109 // A read past the end is clipped.
110 const tail = yield* source.read(9_990, 100);
111 expect(tail.length).toBe(10);
112 }).pipe(Effect.provide(RuntimeContext.phantom)),
113 );
114 });
115
116 test("a parse through 1 KiB windows equals the in-memory parse (boundaries, straddles, eviction)", async () => {
117 await Effect.runPromise(
118 Effect.gen(function* () {
119 const { pack, oids } = yield* buildPack(300);
120 expect(pack.length).toBeGreaterThan(50_000);
121 const memory = yield* parseWith(bufferRandomAccess(pack));
122 const blobs = makeMemoryBlobStore();
123 // Emulate the wire body: a request prefix before the pack, then the pack.
124 const prefix = new TextEncoder().encode("0000".repeat(37));
125 yield* blobs.put("incoming", concat([prefix, pack]));
126 const spilled = sliceRandomAccess(
127 blobRandomAccess({
128 blobs,
129 key: "incoming",
130 size: prefix.length + pack.length,
131 windowBytes: 1024,
132 maxWindows: 3,
133 }),
134 prefix.length,
135 );
136 expect(spilled.readSync).toBeDefined();
137 const windowed = yield* parseWith(spilled);
138 expect(windowed.count).toBe(300);
139 // dataOffset addresses the compressed span in the SOURCE (promotion
140 // relies on this): reading it back yields exactly zdata.
141 for (const entry of windowed.offsets.slice(0, 50)) {
142 expect(entry.fromDelta).toBe(false);
143 const span = yield* spilled.read(
144 entry.dataOffset,
145 entry.zdata.length,
146 );
147 expect(Array.from(span)).toEqual(Array.from(entry.zdata));
148 }
149 expect(windowed.seen).toEqual(memory.seen);
150 expect(new Set(windowed.seen)).toEqual(new Set(oids));
151 }).pipe(Effect.provide(RuntimeContext.phantom)),
152 );
153 });
154
155 test("with production geometry (large windows) a sequential parse fetches about one window each", async () => {
156 await Effect.runPromise(
157 Effect.gen(function* () {
158 const { pack } = yield* buildPack(600, { incompressible: true });
159 const blobs = makeMemoryBlobStore();
160 yield* blobs.put("incoming", pack);
161 const windowBytes = 64 * 1024;
162 const windows = Math.ceil(pack.length / windowBytes);
163 expect(windows).toBeGreaterThan(3);
164 const parsed = yield* parseWith(
165 blobRandomAccess({
166 blobs,
167 key: "incoming",
168 size: pack.length,
169 windowBytes,
170 maxWindows: 4,
171 }),
172 );
173 expect(parsed.count).toBe(600);
174 // Every window once, plus at most one re-read per window edge an
175 // entry's probe straddles (the LRU holds 4, the parse is sequential).
176 expect(blobs.gets.length).toBeLessThanOrEqual(windows * 2);
177 }).pipe(Effect.provide(RuntimeContext.phantom)),
178 );
179 });
180});
181
182describe("synchronous fast path", { tags: ["unit", "local"] }, () => {
183 test("with a batched sink, non-delta entries arrive in runs of ≤ SINK_BATCH and the per-entry sink sees none", async () => {
184 await Effect.runPromise(
185 Effect.gen(function* () {
186 const { pack, oids } = yield* buildPack(700);
187 const store = makeObjectStore({
188 sql: makeTestSqlClient(),
189 blobs: makeMemoryBlobStore(),
190 repoId: "R",
191 });
192 const batches: Array<number> = [];
193 const single: Array<string> = [];
194 const seen = new Set<string>();
195 const summary = yield* ingestPack({
196 source: bufferRandomAccess(pack),
197 store,
198 sink: (entry) =>
199 Effect.sync(() => {
200 single.push(entry.oid);
201 }),
202 sinkBatch: (entries) =>
203 Effect.sync(() => {
204 batches.push(entries.length);
205 for (const e of entries) seen.add(e.oid);
206 }),
207 });
208 expect(summary.count).toBe(700);
209 expect(single).toEqual([]);
210 expect(batches.every((n) => n <= SINK_BATCH && n > 0)).toBe(true);
211 expect(batches.reduce((a, b) => a + b, 0)).toBe(700);
212 expect(seen).toEqual(new Set(oids));
213 }),
214 );
215 });
216});
217
218describe(
219 "parsing a pack while it streams in (DESIGN §22.6)",
220 { tags: ["unit", "local"] },
221 () => {
222 test(
223 "parse runs concurrently with the feed, in random chunk sizes, and verifies the trailer",
224 async () => {
225 await Effect.runPromise(
226 Effect.gen(function* () {
227 const { pack, oids } = yield* buildPack(500, {
228 incompressible: true,
229 });
230 const feeder = makeStreamingSource({
231 slabBytes: 64 * 1024,
232 retainBytes: 256 * 1024,
233 backpressureBytes: 128 * 1024,
234 });
235 const prefix = new TextEncoder().encode(
236 "0021push refs/heads/main x\n0000".padEnd(37, "\0"),
237 );
238 const parse = yield* Effect.forkChild(
239 parseWith(sliceRandomAccess(feeder.source, prefix.length)),
240 );
241 const body = concat([prefix, pack]);
242 let at = 0;
243 let seed = 7;
244 while (at < body.length) {
245 seed = (seed * 1103515245 + 12345) & 0x7fffffff;
246 const n = 1 + (seed % 20_000);
247 yield* feeder.push(body.subarray(at, at + n));
248 at += n;
249 if (seed % 5 === 0) yield* Effect.yieldNow;
250 }
251 feeder.end();
252 const parsed = yield* Fiber.join(parse);
253 expect(parsed.count).toBe(500);
254 expect(new Set(parsed.seen)).toEqual(new Set(oids));
255 }),
256 );
257 },
258 { timeout: 30_000 },
259 );
260
261 test("a body that ends early is a truncated-pack error, not a hang", async () => {
262 await Effect.runPromise(
263 Effect.gen(function* () {
264 const { pack } = yield* buildPack(50);
265 const feeder = makeStreamingSource({ slabBytes: 4096 });
266 const parse = yield* Effect.forkChild(
267 Effect.result(parseWith(feeder.source)),
268 );
269 yield* feeder.push(pack.subarray(0, Math.floor(pack.length / 2)));
270 feeder.end();
271 const r = yield* Fiber.join(parse);
272 expect(r._tag).toBe("Failure");
273 }),
274 );
275 });
276
277 test("a corrupted trailer is rejected even though the hash is accumulated incrementally", async () => {
278 await Effect.runPromise(
279 Effect.gen(function* () {
280 const { pack } = yield* buildPack(20);
281 const bad = Uint8Array.from(pack);
282 bad[bad.length - 1] ^= 0xff;
283 const r = yield* Effect.result(parseWith(bufferRandomAccess(bad)));
284 expect(r._tag).toBe("Failure");
285 if (r._tag === "Failure")
286 expect(String(r.failure._tag)).toBe("PackChecksumMismatch");
287 }),
288 );
289 });
290 },
291);