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