1/**
2 * The pack hasher on dynamically loaded Workers, end to end (DESIGN
3 * §22.12): pushes stream through four loaded hasher isolates, and the
4 * result clones back byte-identical under `fsck --strict`.
5 */
6import * as Cloudflare from "@/Cloudflare";
7import { GitApi } from "@/Git/Api.ts";
8import * as Test from "@/Test/Alchemy";
9import { expect } from "alchemy-test";
10import * as Data from "effect/Data";
11import * as Effect from "effect/Effect";
12import * as FileSystem from "effect/FileSystem";
13import { MinimumLogLevel } from "effect/References";
14import * as Schedule from "effect/Schedule";
15import * as Stream from "effect/Stream";
16import * as HttpClient from "effect/http/HttpClient";
17import * as HttpClientRequest from "effect/http/HttpClientRequest";
18import * as HttpApiClient from "effect/http-api/HttpApiClient";
19import * as ChildProcess from "effect/process/ChildProcess";
20import { makeLoaderTestStack, TEST_SECRET } from "./fixtures/loader-stack.ts";
21
22const { test, beforeAll, afterAll, deploy, destroy } = Test.make({
23 providers: Cloudflare.providers(),
24});
25
26const logLevel = Effect.provideService(
27 MinimumLogLevel,
28 process.env.DEBUG ? "Debug" : "Info",
29);
30
31const Stack = makeLoaderTestStack("GitHasherLoaderStack");
32
33const makeClient = (url: string, token: string) =>
34 HttpApiClient.make(GitApi, {
35 baseUrl: url,
36 transformClient: HttpClient.mapRequest((request) =>
37 request.pipe(HttpClientRequest.bearerToken(token)),
38 ),
39 });
40
41const edgeRetry = <A, E, R>(effect: Effect.Effect<A, E, R>) =>
42 effect.pipe(
43 Effect.timeout("10 seconds"),
44 Effect.retry({
45 while: (error: unknown) =>
46 (error as { _tag?: string })._tag === "TimeoutError" ||
47 (error as { _tag?: string })._tag === "HttpClientError",
48 schedule: Schedule.spaced("1500 millis"),
49 times: 20,
50 }),
51 );
52
53class ShellError extends Data.TaggedError("ShellError")<{
54 readonly script: string;
55 readonly exitCode: number;
56 readonly stdout: string;
57 readonly stderr: string;
58}> {
59 override get message(): string {
60 return `shell script exited ${this.exitCode}\n--- script ---\n${this.script}\n--- stdout ---\n${this.stdout}\n--- stderr ---\n${this.stderr}`;
61 }
62}
63
64const sh = Effect.fn(function* (cwd: string, script: string) {
65 const handle = yield* ChildProcess.make("sh", ["-ec", script], {
66 cwd,
67 env: {
68 GIT_TERMINAL_PROMPT: "0",
69 GIT_CONFIG_NOSYSTEM: "1",
70 GIT_CONFIG_GLOBAL: "/dev/null",
71 GIT_AUTHOR_NAME: "Hasher Bot",
72 GIT_AUTHOR_EMAIL: "hasher@example.com",
73 GIT_COMMITTER_NAME: "Hasher Bot",
74 GIT_COMMITTER_EMAIL: "hasher@example.com",
75 GIT_AUTHOR_DATE: "2026-01-01T00:00:00Z",
76 GIT_COMMITTER_DATE: "2026-01-01T00:00:00Z",
77 },
78 extendEnv: true,
79 });
80 const [exitCode, stdout, stderr] = yield* Effect.all(
81 [
82 handle.exitCode,
83 Stream.mkString(Stream.decodeText(handle.stdout)),
84 Stream.mkString(Stream.decodeText(handle.stderr)),
85 ],
86 { concurrency: 3 },
87 );
88 if (exitCode !== 0) {
89 return yield* new ShellError({ script, exitCode, stdout, stderr });
90 }
91 return { stdout: stdout.trim(), stderr };
92}, Effect.timeout("180 seconds"));
93
94const stack = beforeAll(
95 deploy(Stack).pipe(
96 Effect.tap(({ url }) =>
97 Effect.gen(function* () {
98 yield* Effect.logInfo(`git-service (loader hasher) deployed at ${url}`);
99 const admin = yield* makeClient(url, TEST_SECRET);
100 yield* admin.repos.list({ query: {} }).pipe(
101 edgeRetry,
102 Effect.retry({
103 schedule: Schedule.spaced("1500 millis"),
104 times: 40,
105 }),
106 );
107 }),
108 ),
109 logLevel,
110 ),
111);
112afterAll.skipIf(!!process.env.NO_DESTROY)(destroy(Stack).pipe(logLevel));
113
114test(
115 "a multi-part push is hashed on loaded Workers and clones back under fsck --strict",
116 Effect.gen(function* () {
117 const { url } = yield* stack;
118 const admin = yield* makeClient(url, TEST_SECRET);
119 const owner = "loader";
120 const name = "hashed";
121 yield* admin.repos.delete({ params: { owner, repo: name } }).pipe(
122 Effect.catchTag("RepoNotFound", () => Effect.void),
123 edgeRetry,
124 );
125 const created = yield* admin.repos
126 .create({ payload: { owner, name } })
127 .pipe(edgeRetry);
128 const parsed = new URL(url);
129 const remote = `${parsed.protocol}//x:${TEST_SECRET}@${parsed.host}/${owner}/${name}.git`;
130 const fs = yield* FileSystem.FileSystem;
131 const dir = yield* fs.makeTempDirectory({ prefix: "git-lambda-" });
132 // ~14 MiB of incompressible blobs plus text files that delta well: a
133 // body that spills (four Lambda chunks, two R2 parts) and carries
134 // both non-delta and delta entries across chunk edges.
135 yield* sh(
136 dir,
137 `git init -q -b main src && cd src
138 for i in $(seq 1 14); do head -c 1000000 /dev/urandom > bin-$i.dat; done
139 for i in $(seq 1 200); do seq 1 400 | sed "s/^/line $i /" > text-$i.txt; done
140 git add -A && git commit -qm base
141 for i in $(seq 1 60); do echo "edit $i" >> text-$i.txt; done
142 git add -A && git commit -qm edits
143 git push -q ${remote} main`,
144 );
145 const back = yield* sh(
146 dir,
147 `git clone -q ${remote} back && cd back && git fsck --strict && git rev-parse HEAD`,
148 );
149 const head = yield* sh(dir, "cd src && git rev-parse HEAD");
150 expect(back.stdout.trim().split("\n").pop()).toBe(head.stdout.trim());
151 const repo = yield* admin.repos
152 .get({ params: { owner, repo: name } })
153 .pipe(edgeRetry);
154 const push = repo.lastPush;
155 expect(push).not.toBeNull();
156 // Four 4 MiB chunks were dispatched to the loaded hashers.
157 expect(push?.phases?.chunks ?? 0).toBeGreaterThanOrEqual(4);
158 yield* Effect.logInfo(`lastPush: ${JSON.stringify(push)}`);
159 }),
160 {
161 tags: [
162 "provider:cloudflare",
163 "provider:cloudflare:r2",
164 "provider:cloudflare:worker",
165 "live",
166 ],
167 timeout: 600_000,
168 },
169);