1/**
2 * The hasher-driven ingest (src/Git/RepoObject.ts `ingestPackFrom` with a
3 * hasher, DESIGN §22.7) over the store harness: tiny parts force every
4 * delta base into an earlier part, so cross-part resolution after the end
5 * and thin-pack bases from the store are exercised in-process.
6 */
7import * as BunServices from "@effect/platform-bun/BunServices";
8import { bufferRandomAccess } from "@/Git/Protocol/PackParser.ts";
9import { HasherInline, Hasher, type HasherShape } from "@/Git/Hasher/Hasher.ts";
10import { ingestPackFrom, ingestStoreOf } from "@/Git/RepoObject.ts";
11import { makeObjectStore } from "@/Git/Store/ObjectStore.ts";
12import { makeStreamingSource } from "@/Git/Store/StreamingSource.ts";
13import { sliceRandomAccess } from "@/Git/Store/PackSource.ts";
14import {
15 hashObject,
16 encodeTypeSize,
17 makeSha1,
18} from "@/Git/Protocol/ObjectCodec.ts";
19import { packHeader } from "@/Git/Protocol/PackWriter.ts";
20import * as Zlib from "@/Git/Protocol/Zlib.ts";
21import * as Fiber from "effect/Fiber";
22import { concat } from "./harness/pack.ts";
23import { describe, expect, test } from "alchemy-test";
24import { BlobStore } from "@/Git/BlobStore.ts";
25import * as Effect from "effect/Effect";
26import { RuntimeContext } from "@/RuntimeContext.ts";
27import * as Layer from "effect/Layer";
28import * as FileSystem from "effect/FileSystem";
29import * as Path from "effect/Path";
30import { makeMemoryBlobStore, makeTestSqlClient } from "./harness/store.ts";
31
32const fixture = (name: string) =>
33 Effect.gen(function* () {
34 const fs = yield* FileSystem.FileSystem;
35 const path = yield* Path.Path;
36 const dir = path.join(import.meta.dirname, "fixtures", "packs");
37 const pack = yield* fs.readFile(path.join(dir, name));
38 const manifest = JSON.parse(
39 yield* fs.readFileString(path.join(dir, "manifest.json")),
40 ) as {
41 packs: Record<string, { oids: ReadonlyArray<string> }>;
42 };
43 return { pack, manifest };
44 });
45
46describe(
47 "ingestPackFrom through the hasher",
48 { tags: ["unit", "local"] },
49 () => {
50 test("ofs-delta fixture in 64-byte parts: every object staged, cross-part deltas resolved after the end", async () => {
51 await Effect.runPromise(
52 Effect.gen(function* () {
53 const { pack, manifest } = yield* fixture("ofs-delta.pack");
54 const sql = makeTestSqlClient();
55 const store = makeObjectStore({
56 sql,
57 blobs: makeMemoryBlobStore(),
58 repoId: "R",
59 });
60 const hasher = yield* Hasher;
61 const outcome = yield* Effect.result(
62 ingestPackFrom(bufferRandomAccess(pack), {
63 store: ingestStoreOf(store),
64 pushId: "push-1",
65 hasher,
66 partBytes: 64,
67 }),
68 );
69 if (outcome._tag === "Failure")
70 throw new Error(
71 `${outcome.failure._tag}: ${outcome.failure.reason}`,
72 );
73 const result = outcome.success;
74 const expected = manifest.packs["ofs-delta"]!.oids;
75 expect(result.objectCount).toBe(expected.length);
76 const staged = yield* sql.all<{ oid: string }>(
77 `SELECT oid FROM objects WHERE staged_push = 'push-1' ORDER BY oid`,
78 );
79 expect(staged.map((r) => r.oid)).toEqual([...expected].sort());
80 }).pipe(
81 Effect.provide(
82 HasherInline.pipe(
83 Layer.provide(Layer.succeed(BlobStore, makeMemoryBlobStore())),
84 ),
85 ),
86 Effect.provide(BunServices.layer),
87 ),
88 );
89 });
90
91 test("a corrupted trailer is rejected by the pipeline's lagged hash", async () => {
92 await Effect.runPromise(
93 Effect.gen(function* () {
94 const { pack } = yield* fixture("simple.pack");
95 const bad = Uint8Array.from(pack);
96 bad[bad.length - 3] ^= 0x01;
97 const store = makeObjectStore({
98 sql: makeTestSqlClient(),
99 blobs: makeMemoryBlobStore(),
100 repoId: "R",
101 });
102 const hasher = yield* Hasher;
103 const r = yield* Effect.result(
104 ingestPackFrom(bufferRandomAccess(bad), {
105 store: ingestStoreOf(store),
106 pushId: "p",
107 hasher,
108 partBytes: 4096,
109 }),
110 );
111 expect(r._tag).toBe("Failure");
112 if (r._tag === "Failure")
113 expect(r.failure.reason).toContain("checksum");
114 }).pipe(
115 Effect.provide(
116 HasherInline.pipe(
117 Layer.provide(Layer.succeed(BlobStore, makeMemoryBlobStore())),
118 ),
119 ),
120 Effect.provide(BunServices.layer),
121 ),
122 );
123 });
124 },
125);
126
127describe(
128 "hasher pipeline over a streaming source with eviction",
129 { tags: ["unit", "local"] },
130 () => {
131 test(
132 "a multi-part pack whose retained window is smaller than a part stages every inline row and completes (no fallback)",
133 async () => {
134 await Effect.runPromise(
135 Effect.gen(function* () {
136 const n = 1500;
137 const pieces: Array<Uint8Array> = [packHeader(n)];
138 for (let i = 0; i < n; i++) {
139 const c = new Uint8Array(1000 + (i % 700));
140 crypto.getRandomValues(c);
141 yield* hashObject(3, c);
142 pieces.push(encodeTypeSize(3, c.length), yield* Zlib.deflate(c));
143 }
144 const body = concat(pieces);
145 const sha = makeSha1();
146 sha.update(body);
147 const pack = concat([body, sha.digest()]);
148 // Parts of 64 KiB, retention of 128 KiB, no spill/fallback: every
149 // inline row must be read back before its bytes are dropped.
150 const feeder = makeStreamingSource({
151 slabBytes: 32 * 1024,
152 retainBytes: 128 * 1024,
153 backpressureBytes: 1 << 20,
154 });
155 const sql = makeTestSqlClient();
156 const store = makeObjectStore({
157 sql,
158 blobs: makeMemoryBlobStore(),
159 repoId: "R",
160 });
161 const hasher = yield* Hasher;
162 const ingest = yield* Effect.forkChild(
163 Effect.result(
164 ingestPackFrom(feeder.source, {
165 store: ingestStoreOf(store),
166 pushId: "p",
167 hasher,
168 partBytes: 64 * 1024,
169 }),
170 ),
171 );
172 for (let at = 0; at < pack.length; at += 50_000)
173 yield* feeder.push(pack.subarray(at, at + 50_000));
174 feeder.end();
175 const r = yield* Fiber.join(ingest).pipe(
176 Effect.timeout("20 seconds"),
177 );
178 expect(r._tag).toBe("Success");
179 const staged = yield* sql.first<{ n: number; withBytes: number }>(
180 `SELECT COUNT(*) AS n, SUM(LENGTH(zdata) > 0) AS withBytes FROM objects WHERE staged_push = 'p'`,
181 );
182 expect(staged?.n).toBe(n);
183 expect(staged?.withBytes).toBe(n);
184 }).pipe(
185 Effect.provide(
186 HasherInline.pipe(
187 Layer.provide(Layer.succeed(BlobStore, makeMemoryBlobStore())),
188 ),
189 ),
190 ),
191 );
192 },
193 { timeout: 30_000 },
194 );
195 },
196);
197
198describe(
199 "raw-chunk dispatch with resync and stitching (DESIGN §22.9)",
200 { tags: ["unit", "local"] },
201 () => {
202 test("the ofs-delta fixture in 64-byte chunks: every straddler stitched, every delta resolved", async () => {
203 await Effect.runPromise(
204 Effect.gen(function* () {
205 const { pack, manifest } = yield* fixture("ofs-delta.pack");
206 const sql = makeTestSqlClient();
207 const store = makeObjectStore({
208 sql,
209 blobs: makeMemoryBlobStore(),
210 repoId: "R",
211 });
212 const hasher = yield* Hasher;
213 for (const partBytes of [64, 200, 1000, 1 << 20]) {
214 yield* sql.run(`DELETE FROM objects`);
215 const outcome = yield* Effect.result(
216 ingestPackFrom(bufferRandomAccess(pack), {
217 store: ingestStoreOf(store),
218 pushId: "p",
219 hasher,
220 partBytes,
221 }),
222 );
223 if (outcome._tag === "Failure")
224 throw new Error(
225 `parts of ${partBytes}: ${outcome.failure._tag}: ${outcome.failure.reason}`,
226 );
227 const result = outcome.success;
228 const expected = manifest.packs["ofs-delta"]!.oids;
229 expect(result.objectCount, `parts of ${partBytes}`).toBe(
230 expected.length,
231 );
232 const staged = yield* sql.all<{ oid: string }>(
233 `SELECT oid FROM objects WHERE staged_push = 'p' ORDER BY oid`,
234 );
235 expect(
236 staged.map((r) => r.oid),
237 `parts of ${partBytes}`,
238 ).toEqual([...expected].sort());
239 }
240 }).pipe(
241 Effect.provide(
242 HasherInline.pipe(
243 Layer.provide(Layer.succeed(BlobStore, makeMemoryBlobStore())),
244 ),
245 ),
246 Effect.provide(BunServices.layer),
247 ),
248 );
249 });
250 },
251);
252
253describe(
254 "spill by the hasher (DESIGN §22.10)",
255 { tags: ["unit", "local"] },
256 () => {
257 /** A pack of `n` random blobs plus a fake command section in front. */
258 const makeBody = (n: number, blobBytes: number, headBytes: number) =>
259 Effect.gen(function* () {
260 const pieces: Array<Uint8Array> = [packHeader(n)];
261 for (let i = 0; i < n; i++) {
262 const c = new Uint8Array(blobBytes + (i % 97));
263 crypto.getRandomValues(c);
264 pieces.push(encodeTypeSize(3, c.length), yield* Zlib.deflate(c));
265 }
266 const packBody = concat(pieces);
267 const sha = makeSha1();
268 sha.update(packBody);
269 const pack = concat([packBody, sha.digest()]);
270 const head = new Uint8Array(headBytes);
271 head.fill(0x20);
272 return { body: concat([head, pack]), packStart: headBytes };
273 });
274
275 const run = (
276 opts: {
277 readonly n: number;
278 readonly blobBytes: number;
279 readonly headBytes: number;
280 readonly partBytes: number;
281 readonly threshold: number;
282 readonly chunk: number;
283 },
284 hasherOverride?: HasherShape,
285 ) =>
286 Effect.gen(function* () {
287 const { body, packStart } = yield* makeBody(
288 opts.n,
289 opts.blobBytes,
290 opts.headBytes,
291 );
292 // ONE blob store: the hasher writes the parts the pump's upload
293 // collects — in production both are the same BlobStore layer.
294 const blobs = yield* BlobStore;
295 const feeder = makeStreamingSource({
296 slabBytes: opts.partBytes,
297 retainBytes: opts.partBytes * 2,
298 backpressureBytes: 1 << 20,
299 });
300 const sql = makeTestSqlClient();
301 const store = makeObjectStore({ sql, blobs, repoId: "R" });
302 const hasher = hasherOverride ?? (yield* Hasher);
303 const source = sliceRandomAccess(feeder.source, packStart);
304 const ingest = yield* Effect.forkChild(
305 Effect.result(
306 ingestPackFrom(source, {
307 store: ingestStoreOf(store),
308 pushId: "p",
309 hasher,
310 partBytes: opts.partBytes,
311 spill: {
312 body: feeder.source,
313 feeder,
314 packStart,
315 blobs,
316 key: "R/incoming/X.pack",
317 packId: "wire-X",
318 repoId: "R",
319 threshold: opts.threshold,
320 },
321 }),
322 ),
323 );
324 for (let at = 0; at < body.length; at += opts.chunk)
325 yield* feeder.push(body.subarray(at, at + opts.chunk));
326 feeder.end();
327 const r = yield* Fiber.join(ingest).pipe(Effect.timeout("20 seconds"));
328 if (r._tag === "Failure") throw new Error(r.failure.reason);
329 const rows = yield* sql.first<{ n: number; promoted: number }>(
330 `SELECT COUNT(*) AS n, SUM(location = 'pack') AS promoted FROM objects WHERE staged_push = 'p'`,
331 );
332 const spilled = yield* blobs.head("R/incoming/X.pack");
333 return { result: r.success, rows, spilled, body, blobs };
334 }).pipe(
335 Effect.provide(
336 Layer.provideMerge(
337 HasherInline,
338 Layer.succeed(BlobStore, makeMemoryBlobStore()),
339 ),
340 ),
341 Effect.provide(RuntimeContext.phantom),
342 );
343
344 test(
345 "a body past the threshold is written whole by the parts' hashers; blobs are promoted into it",
346 async () => {
347 const out = await Effect.runPromise(
348 run({
349 n: 400,
350 blobBytes: 600,
351 headBytes: 137,
352 partBytes: 32 * 1024,
353 threshold: 8 * 1024,
354 chunk: 7_001,
355 }),
356 );
357 expect(out.result.parkedKey).toBe("R/incoming/X.pack");
358 expect(out.spilled?.size).toBe(out.body.length);
359 const stored = await Effect.runPromise(
360 Effect.gen(function* () {
361 const blob = yield* out.blobs.get("R/incoming/X.pack");
362 return blob === null ? undefined : yield* blob.bytes;
363 }).pipe(Effect.provide(RuntimeContext.phantom)),
364 );
365 expect(stored).toEqual(out.body);
366 expect(out.rows?.n).toBe(400);
367 expect(out.rows?.promoted).toBe(400);
368 expect(out.result.promoted).toBe(400);
369 },
370 { timeout: 30_000 },
371 );
372
373 test(
374 "a hasher that cannot write the spill (writesSpill=false, small chunks): the pump uploads the parts itself",
375 async () => {
376 const out = await Effect.runPromise(
377 Effect.gen(function* () {
378 const inline = yield* Hasher;
379 // A Lambda-shaped hasher: chunks a quarter of a part, no spill.
380 const lambdaLike: HasherShape = {
381 writesSpill: false,
382 chunkBytes: 8 * 1024,
383 hashPart: (payload, opts) => {
384 if (opts.spill !== undefined) {
385 throw new Error(
386 "spill must not be requested from a non-spilling hasher",
387 );
388 }
389 return inline.hashPart(payload, opts);
390 },
391 hashBoundsPart: inline.hashBoundsPart,
392 resolveDeltas: inline.resolveDeltas,
393 };
394 return yield* run(
395 {
396 n: 400,
397 blobBytes: 600,
398 headBytes: 137,
399 partBytes: 32 * 1024,
400 threshold: 8 * 1024,
401 chunk: 7_001,
402 },
403 lambdaLike,
404 );
405 }).pipe(
406 Effect.provide(
407 Layer.provideMerge(
408 HasherInline,
409 Layer.succeed(BlobStore, makeMemoryBlobStore()),
410 ),
411 ),
412 Effect.provide(RuntimeContext.phantom),
413 ),
414 );
415 expect(out.result.parkedKey).toBe("R/incoming/X.pack");
416 expect(out.spilled?.size).toBe(out.body.length);
417 const stored = await Effect.runPromise(
418 Effect.gen(function* () {
419 const blob = yield* out.blobs.get("R/incoming/X.pack");
420 return blob === null ? undefined : yield* blob.bytes;
421 }).pipe(Effect.provide(RuntimeContext.phantom)),
422 );
423 expect(stored).toEqual(out.body);
424 expect(out.rows?.n).toBe(400);
425 expect(out.rows?.promoted).toBe(400);
426 },
427 { timeout: 30_000 },
428 );
429
430 test(
431 "a body that ends within the threshold stays in memory: nothing written, rows inline",
432 async () => {
433 const out = await Effect.runPromise(
434 run({
435 n: 20,
436 blobBytes: 100,
437 headBytes: 50,
438 partBytes: 32 * 1024,
439 threshold: 64 * 1024,
440 chunk: 1_000,
441 }),
442 );
443 expect(out.result.parkedKey).toBeUndefined();
444 expect(out.spilled).toBeNull();
445 expect(out.rows?.n).toBe(20);
446 expect(out.rows?.promoted).toBe(0);
447 },
448 { timeout: 30_000 },
449 );
450 },
451);