1/**
2 * `Hasher` over dynamically loaded Workers (DESIGN §22.12). A
3 * `WorkerLoader` binding on the Git Worker instantiates the hasher module
4 * (`WorkerLoaderModule.ts`, bundled into a string at build time by the `?worker`
5 * import) under a few fixed names; each name is its own isolate with its
6 * own 128 MB, and — measured — distinct isolates run in parallel with
7 * the caller, up to the runtime's four concurrent dynamic invocations per
8 * request. No cross-cloud hop, no credentials, no network for the hasher.
9 *
10 * Chunks are 4 MiB and the pump writes the spill parts (`writesSpill:
11 * false`); any failure in the loaded worker hashes that chunk inline.
12 *
13 * **Example:** A Git host hashing pushes on dynamic Workers
14 * ```typescript
15 * import * as GitHasher from "alchemy/Git/Hasher";
16 *
17 * const GitLive = Git.ApiLive.pipe(
18 * Layer.provide(Git.ApiHandlersLive),
19 * Layer.provide(Authentication.layer),
20 * Layer.provide(Git.ReposDurableObject),
21 * Layer.provide(Git.RegistryDurableObject),
22 * Layer.provide(GitHasher.HasherWorkerLoader()),
23 * Layer.provide(Git.BlobStoreR2(GitObjects)),
24 * );
25 * ```
26 */
27import * as Effect from "effect/Effect";
28import * as Layer from "effect/Layer";
29import * as Semaphore from "effect/Semaphore";
30import * as HttpClientRequest from "effect/http/HttpClientRequest";
31import { WorkerLoader } from "../../Cloudflare/Workers/WorkerLoader.ts";
32import type {
33 Worker,
34 WorkerEnvironment,
35} from "../../Cloudflare/Workers/Worker.ts";
36import {
37 hashBounds,
38 resolveDeltas,
39 scanPart,
40} from "../Protocol/PartialScan.ts";
41import { Hasher, type HasherShape } from "./Hasher.ts";
42import {
43 decodeDeltaResults,
44 decodeScanResult,
45 encodeDeltaBatch,
46 HASH_ROUTE,
47 HashError,
48} from "./Protocol.ts";
49import hasherSource from "./WorkerLoaderModule.ts?worker";
50
51/** The runtime allows four concurrent dynamic-worker invocations per request. */
52export const LOADER_MAX_CONCURRENCY = 4;
53/** Chunk size: four in flight × 4 MiB keeps the working set small. */
54export const LOADER_CHUNK_BYTES = 4 * 1024 * 1024;
55
56export interface HasherWorkerLoaderOptions {
57 /** The `worker_loader` binding name on the Git Worker. @default "GIT_HASHER_LOADER" */
58 readonly binding?: string | undefined;
59 /** Isolates (and concurrent chunks) per push, at most four. @default 4 */
60 readonly concurrency?: number | undefined;
61 /** Compatibility date of the loaded hasher. @default "2026-03-17" */
62 readonly compatibilityDate?: string | undefined;
63}
64
65export const HasherWorkerLoader = (
66 options: HasherWorkerLoaderOptions = {},
67): Layer.Layer<Hasher, never, Worker | WorkerEnvironment> =>
68 Layer.effect(
69 Hasher,
70 Effect.gen(function* () {
71 const loader = yield* WorkerLoader(
72 options.binding ?? "GIT_HASHER_LOADER",
73 );
74 const concurrency = Math.max(
75 1,
76 Math.min(
77 options.concurrency ?? LOADER_MAX_CONCURRENCY,
78 LOADER_MAX_CONCURRENCY,
79 ),
80 );
81 // Fixed names: the runtime caches an isolate per name, so a slot is
82 // warm across pushes; distinct names are what makes them parallel.
83 const idle: Array<string> = Array.from(
84 { length: concurrency },
85 (_, i) => `git-hasher-${i}`,
86 );
87 const gate = yield* Semaphore.make(concurrency);
88 const code = () => ({
89 compatibilityDate: options.compatibilityDate ?? "2026-03-17",
90 compatibilityFlags: ["nodejs_compat"],
91 mainModule: "hasher.js",
92 modules: { "hasher.js": hasherSource },
93 globalOutbound: null,
94 });
95 /** One call on a free slot: POST `body` to `path` on the loaded hasher, answer bytes (sans frame). */
96 const call = (path: string, body: Uint8Array) =>
97 Semaphore.withPermits(
98 gate,
99 1,
100 )(
101 Effect.acquireUseRelease(
102 Effect.sync(() => idle.pop()!),
103 (slot) =>
104 Effect.gen(function* () {
105 const worker = yield* loader.get(slot, code);
106 const response = yield* worker
107 .fetch(
108 HttpClientRequest.post(`https://hasher${path}`).pipe(
109 HttpClientRequest.bodyUint8Array(body),
110 ),
111 )
112 .pipe(
113 Effect.mapError(
114 (error) =>
115 new HashError({
116 reason: `dynamic hasher: ${String(error)}`,
117 }),
118 ),
119 );
120 const bytes = new Uint8Array(
121 yield* response.arrayBuffer.pipe(
122 Effect.mapError(
123 (error) =>
124 new HashError({
125 reason: `dynamic hasher body: ${String(error)}`,
126 }),
127 ),
128 ),
129 );
130 if (response.status !== 200) {
131 return yield* new HashError({
132 reason: `dynamic hasher: status ${response.status}: ${new TextDecoder().decode(bytes.subarray(0, 200))}`,
133 });
134 }
135 // One frame: `u32 len | payload`.
136 return bytes.subarray(4);
137 }),
138 (slot) => Effect.sync(() => void idle.push(slot)),
139 ),
140 );
141 const remote = (
142 payload: Uint8Array,
143 opts: Parameters<HasherShape["hashPart"]>[1],
144 ) =>
145 call(
146 `${HASH_ROUTE}?base=${opts.base}&remaining=${opts.remaining}&max=${opts.maxObjectSize}${opts.resync ? "&resync=1" : ""}${opts.skip ? `&skip=${opts.skip}` : ""}`,
147 payload,
148 ).pipe(Effect.map(decodeScanResult));
149 return {
150 writesSpill: false,
151 chunkBytes: LOADER_CHUNK_BYTES,
152 concurrency,
153 hashPart: (payload, opts) =>
154 remote(payload, opts).pipe(
155 Effect.catchTag("HashError", (error) => {
156 console.warn(`[hasher] ${error.reason}; hashing inline`);
157 const skip = opts.skip ?? 0;
158 return scanPart(
159 skip > 0 ? payload.subarray(skip) : payload,
160 opts,
161 );
162 }),
163 ),
164 resolveDeltas: (bases, jobs, opts) =>
165 call(
166 `${HASH_ROUTE}?mode=deltas&max=${opts.maxObjectSize}`,
167 encodeDeltaBatch(bases, jobs),
168 ).pipe(
169 Effect.map(decodeDeltaResults),
170 Effect.catchTag("HashError", (error) => {
171 console.warn(`[hasher] ${error.reason}; resolving inline`);
172 return resolveDeltas(bases, jobs, opts);
173 }),
174 ),
175 hashBoundsPart: (payload, bounds, opts) =>
176 hashBounds(payload, bounds, opts),
177 } satisfies HasherShape;
178 }),
179 );