1/**
2 * The streaming pack source (src/Git/Store/StreamingSource.ts): reads
3 * block until bytes arrive, in-slab reads are views, retention drops
4 * consumed slabs and falls back to the spilled reader, and the feeder is
5 * throttled by backpressure.
6 */
7import { bufferRandomAccess } from "@/Git/Protocol/PackParser.ts";
8import { StoreError } from "@/Git/Protocol/Store.ts";
9import { makeStreamingSource } from "@/Git/Store/StreamingSource.ts";
10import { describe, expect, test } from "alchemy-test";
11import * as Effect from "effect/Effect";
12import * as Fiber from "effect/Fiber";
13
14const bytes = (n: number, seed = 1) => {
15 const out = new Uint8Array(n);
16 for (let i = 0; i < n; i++) out[i] = (i * 31 + seed) & 0xff;
17 return out;
18};
19
20describe("StreamingSource", { tags: ["unit", "local"] }, () => {
21 test("a read past what has arrived waits; in-slab reads are views", async () => {
22 await Effect.runPromise(
23 Effect.gen(function* () {
24 const feeder = makeStreamingSource({ slabBytes: 1024 });
25 const data = bytes(3000);
26 // Reader starts first and blocks.
27 const pending = yield* Effect.forkChild(feeder.source.read(1000, 500));
28 expect(feeder.source.readSync!(0, 10)).toBeUndefined();
29 yield* feeder.push(data.subarray(0, 700));
30 yield* feeder.push(data.subarray(700, 1600));
31 const got = yield* Fiber.join(pending);
32 expect(Array.from(got)).toEqual(Array.from(data.subarray(1000, 1500)));
33 // Inside one slab: a view of the slab's buffer.
34 const view = feeder.source.readSync!(1024, 100)!;
35 expect(view.byteOffset).toBe(0);
36 expect(Array.from(view)).toEqual(Array.from(data.subarray(1024, 1124)));
37 // Across a slab edge, readSync declines; read assembles.
38 expect(feeder.source.readSync!(1000, 100)).toBeUndefined();
39 const across = yield* feeder.source.read(1000, 100);
40 expect(Array.from(across)).toEqual(
41 Array.from(data.subarray(1000, 1100)),
42 );
43 yield* feeder.push(data.subarray(1600));
44 feeder.end();
45 expect(yield* feeder.source.awaitEnd).toBe(3000);
46 // Short read at the end.
47 expect((yield* feeder.source.read(2990, 100)).length).toBe(10);
48 expect(feeder.source.readSync!(2990, 100)!.length).toBe(10);
49 }),
50 );
51 });
52
53 test("retention drops consumed slabs; evicted reads go to the fallback after end", async () => {
54 await Effect.runPromise(
55 Effect.gen(function* () {
56 const feeder = makeStreamingSource({
57 slabBytes: 1000,
58 retainBytes: 2500,
59 backpressureBytes: 1 << 20,
60 });
61 const data = bytes(6000, 5);
62 for (let at = 0; at < data.length; at += 1500)
63 yield* feeder.push(data.subarray(at, at + 1500));
64 // Nothing consumed: everything retained.
65 expect(feeder.source.readSync!(0, 10)).toBeDefined();
66 feeder.source.release(4000);
67 // Now the oldest slabs are gone.
68 expect(feeder.source.readSync!(0, 10)).toBeUndefined();
69 expect(feeder.source.readSync!(4500, 10)).toBeDefined();
70 // A read below the window waits for the fallback + end.
71 const pending = yield* Effect.forkChild(feeder.source.read(100, 50));
72 feeder.setFallback(bufferRandomAccess(data));
73 feeder.end();
74 const got = yield* Fiber.join(pending);
75 expect(Array.from(got)).toEqual(Array.from(data.subarray(100, 150)));
76 }),
77 );
78 });
79
80 test("backpressure: push waits until the parser releases", async () => {
81 await Effect.runPromise(
82 Effect.gen(function* () {
83 const feeder = makeStreamingSource({
84 slabBytes: 1000,
85 backpressureBytes: 1500,
86 });
87 yield* feeder.push(bytes(1000));
88 let pushed = false;
89 const blocked = yield* Effect.forkChild(
90 feeder.push(bytes(1000)).pipe(
91 Effect.tap(() =>
92 Effect.sync(() => {
93 pushed = true;
94 }),
95 ),
96 ),
97 );
98 yield* Effect.sleep("20 millis");
99 expect(pushed).toBe(false); // still waiting on backpressure
100 feeder.source.release(1500);
101 yield* Fiber.join(blocked);
102 expect(pushed).toBe(true);
103 }),
104 );
105 });
106
107 test("fail wakes every waiter with the error", async () => {
108 await Effect.runPromise(
109 Effect.gen(function* () {
110 const feeder = makeStreamingSource();
111 const pending = yield* Effect.forkChild(
112 Effect.result(feeder.source.read(0, 10)),
113 );
114 feeder.fail(new StoreError({ reason: "boom" }));
115 const r = yield* Fiber.join(pending);
116 expect(r._tag).toBe("Failure");
117 }),
118 );
119 });
120});