1/**
2 * The pack hasher (DESIGN §22.7): the CPU-heavy half of push ingest —
3 * inflating and hashing entries, applying in-buffer deltas — as a service
4 * the receive-pack pipeline calls per spilled part, so the repo's Durable
5 * Object only receives bytes and stages rows.
6 *
7 * Two layers:
8 *
9 * - {@link HasherInline} runs the scan in the calling isolate — the
10 * reference assembly. On Cloudflare Workers a service-binding subrequest
11 * executes on the caller's own thread (measured, DESIGN §22.10), so
12 * fanning out buys no CPU there; inline avoids the copies and framing.
13 * - {@link HasherSelf} posts each part to the Worker's own
14 * `/_alchemy/git/hash` route through a `Cloudflare.Workers.Self` service
15 * binding: the same work in another invocation of the same script. Kept
16 * for runtimes where such calls do run in parallel, and as the shape a
17 * remote hasher (a Container running native `index-pack`, a multi-core
18 * host behind a cross-zone URL) would take.
19 *
20 * The wire protocol is binary both ways (parts are megabytes; JSON would
21 * base64 them): request body = the raw bytes, coordinates in the query;
22 * response = `u32 jsonLength | json | blob area`, where the JSON's entries
23 * reference their content/zdata as `[offset, length]` into the blob area.
24 */
25import * as Context from "effect/Context";
26import * as Effect from "effect/Effect";
27import * as Layer from "effect/Layer";
28import * as Redacted from "effect/Redacted";
29import * as Schema from "effect/Schema";
30import * as Fiber from "effect/Fiber";
31import { WorkerEnvironment } from "../../Cloudflare/Workers/Worker.ts";
32import { RuntimeContext } from "../../RuntimeContext.ts";
33import { Random } from "../../Random.ts";
34import {
35 BlobStore,
36 type BlobStoreShape,
37 type UploadedPart,
38} from "../BlobStore.ts";
39import type { Oid, ObjectType } from "../Protocol/ObjectCodec.ts";
40import {
41 type EntryBounds,
42 hashBounds,
43 scanPart,
44 type ScanResult,
45 resolveDeltas,
46 type DeltaBase,
47 type DeltaJob,
48 type DeltaResolved,
49} from "../Protocol/PartialScan.ts";
50import {
51 ObjectTooLargeError,
52 PackFormatError,
53} from "../Protocol/PackParser.ts";
54
55export {
56 decodeBoundsRequest,
57 decodeScanResult,
58 encodeBoundsRequest,
59 encodeScanResult,
60 frame,
61 HASH_ROUTE,
62 HashError,
63 makeFrameReader,
64} from "./Protocol.ts";
65import {
66 decodeDeltaResults,
67 decodeScanResult,
68 encodeBoundsRequest,
69 encodeDeltaBatch,
70 frame,
71 HASH_ROUTE,
72 HashError,
73 makeFrameReader,
74} from "./Protocol.ts";
75
76export interface HashPartOptions {
77 /** Pack-relative offset of `payload[skip]`. */
78 readonly base: number;
79 readonly remaining: number;
80 readonly maxObjectSize: number;
81 /** The payload is a raw chunk: find the first boundary first (DESIGN §22.9). */
82 readonly resync?: boolean | undefined;
83 /**
84 * Scanning starts at `payload[skip]`; the bytes before it (the request's
85 * command section and the pack header, in a body-aligned first part) are
86 * spilled but not scanned.
87 */
88 readonly skip?: number | undefined;
89 /**
90 * Write the WHOLE payload as this part of a multipart upload while
91 * scanning it (DESIGN §22.10): the hasher isolate is the spill's writer,
92 * so the receiver sends each byte out once.
93 */
94 readonly spill?:
95 | {
96 readonly key: string;
97 readonly uploadId: string;
98 readonly partNumber: number;
99 }
100 | undefined;
101}
102
103/**
104 * A scan plus, when a spill was requested, the uploaded part's identity —
105 * DEFERRED: the scan is available as soon as the hasher has it, while the
106 * part's upload to blob storage may still be in flight (DESIGN §22.9).
107 */
108export interface HashPartResult extends ScanResult {
109 readonly part?: Effect.Effect<UploadedPart, HashError> | undefined;
110}
111
112export interface HasherShape {
113 /**
114 * Whether `hashPart` honors `spill` by writing the part itself. A hasher
115 * that cannot reach blob storage (a Lambda) answers false; the pump then
116 * uploads the spill parts from its retained bytes.
117 */
118 readonly writesSpill: boolean;
119 /**
120 * The chunk size this hasher wants (a divisor of the spill part size,
121 * 8 MiB); a Lambda is bounded by its 6 MB invoke payload. Default: the
122 * spill part size.
123 */
124 readonly chunkBytes?: number | undefined;
125 /** Chunks this hasher can verify at once (the pump's dispatch gate). */
126 readonly concurrency?: number | undefined;
127 readonly hashPart: (
128 payload: Uint8Array,
129 options: HashPartOptions,
130 ) => Effect.Effect<
131 HashPartResult,
132 HashError | PackFormatError | ObjectTooLargeError
133 >;
134 /**
135 * Applies a batch of deltas whose bases the caller supplies (DESIGN
136 * §22.13): the cross-chunk and thin deltas the scan left unresolved.
137 */
138 readonly resolveDeltas: (
139 bases: ReadonlyArray<DeltaBase>,
140 jobs: ReadonlyArray<DeltaJob>,
141 options: { readonly maxObjectSize: number },
142 ) => Effect.Effect<
143 Array<DeltaResolved>,
144 HashError | PackFormatError | ObjectTooLargeError
145 >;
146 /**
147 * The parallel half (DESIGN §22.8): hash entries whose spans the caller
148 * already found with `scanBounds`. Parts hashed this way have no chain
149 * between them.
150 */
151 readonly hashBoundsPart: (
152 payload: Uint8Array,
153 bounds: ReadonlyArray<EntryBounds>,
154 options: { readonly base: number; readonly maxObjectSize: number },
155 ) => Effect.Effect<
156 ScanResult,
157 HashError | PackFormatError | ObjectTooLargeError
158 >;
159}
160
161export class Hasher extends Context.Service<Hasher, HasherShape>()(
162 "alchemy/Git/Hasher",
163) {}
164
165/** The in-process hasher: scans here; a requested spill part goes to `blobs`. */
166export const makeInlineHasher = (blobs: BlobStoreShape): HasherShape => ({
167 writesSpill: true,
168 hashPart: (payload, options) =>
169 Effect.gen(function* () {
170 const spill = options.spill;
171 // Detached: the caller's fiber ends with the scan; the part is joined
172 // later, before the multipart upload is completed.
173 const upload =
174 spill === undefined
175 ? undefined
176 : yield* Effect.forkDetach(
177 blobs
178 .uploadPart(
179 spill.key,
180 spill.uploadId,
181 spill.partNumber,
182 payload,
183 )
184 .pipe(
185 Effect.mapError(
186 (error) =>
187 new HashError({
188 reason: `spill part ${spill.partNumber}: ${error.reason}`,
189 }),
190 ),
191 Effect.provide(RuntimeContext.phantom),
192 ),
193 );
194 const skip = options.skip ?? 0;
195 const scan = yield* scanPart(
196 skip === 0 ? payload : payload.subarray(skip),
197 options,
198 );
199 return {
200 ...scan,
201 part: upload === undefined ? undefined : Fiber.join(upload),
202 };
203 }),
204 resolveDeltas: (bases, jobs, options) => resolveDeltas(bases, jobs, options),
205 hashBoundsPart: (payload, bounds, options) =>
206 hashBounds(payload, bounds, options),
207});
208
209/** Runs the scan in-process (tests, or a deployment without a self binding). */
210export const HasherInline: Layer.Layer<Hasher, never, BlobStore> = Layer.effect(
211 Hasher,
212 Effect.map(BlobStore, makeInlineHasher),
213);
214
215/**
216 * Request framing for the bounds mode: `u32 jsonLength | json(bounds) |
217 * bytes`.
218 */
219/** The `env` name of the self service binding (`GIT_WORKER_OPTIONS`). */
220/**
221 * The secret the push pipeline's internal hash route requires: a
222 * deploy-time random value bound onto the Worker (an `Alchemy.Random`
223 * resource, minted once and stable across deploys). Yielded at layer build
224 * by both the route and the self-calling hasher, so no user credential is
225 * ever involved.
226 */
227export const InternalSecret: Effect.Effect<
228 Effect.Effect<Redacted.Redacted<string>>
229> = Effect.gen(function* () {
230 // Yielding the resource class gives a constructor whose providers are
231 // the host stack's (the same way the Cloudflare `*Http` bindings mint
232 // tokens), so declaring the secret here needs nothing from the caller.
233 const R = yield* Random;
234 const resource = yield* R("GitInternalSecret");
235 return yield* resource.text;
236});
237
238export const HASHER_BINDING = "GIT_SELF";
239
240// ── self-binding layer ──────────────────────────────────────────────────────
241
242interface SelfFetcher {
243 readonly fetch: (input: string, init?: RequestInit) => Promise<Response>;
244}
245
246/**
247 * Posts each part to the Worker's own hash route through the self service
248 * binding (`GIT_WORKER_OPTIONS` declares it as `GIT_SELF`). The internal
249 * call authenticates with {@link InternalSecret}, a deploy-time random
250 * value no user ever holds. Falls back to {@link HasherInline} when the
251 * binding is absent from the environment.
252 */
253export const HasherSelf: Layer.Layer<
254 Hasher,
255 never,
256 WorkerEnvironment | BlobStore
257> = Layer.effect(
258 Hasher,
259 Effect.gen(function* () {
260 const env = yield* WorkerEnvironment;
261 const blobs = yield* BlobStore;
262 const fetcher = (env as Record<string, unknown>)[HASHER_BINDING] as
263 | SelfFetcher
264 | undefined;
265 if (fetcher === undefined) {
266 return makeInlineHasher(blobs);
267 }
268 const secret = yield* InternalSecret;
269 const send = (url: string, body: Uint8Array) =>
270 Effect.gen(function* () {
271 const token = Redacted.value(yield* secret);
272 const response = yield* Effect.tryPromise({
273 try: () =>
274 fetcher.fetch(url, {
275 method: "POST",
276 headers: {
277 "content-type": "application/octet-stream",
278 authorization: `Bearer ${token}`,
279 },
280 body: body as unknown as BodyInit,
281 }),
282 catch: (error) =>
283 new HashError({ reason: `hash part fetch: ${String(error)}` }),
284 });
285 if (response.status !== 200 || response.body === null) {
286 const text = yield* Effect.tryPromise({
287 try: () => response.text(),
288 catch: () => new HashError({ reason: "hash part: unreadable" }),
289 });
290 return yield* new HashError({
291 reason: `hash part: status ${response.status}: ${text.slice(0, 200)}`,
292 });
293 }
294 return makeFrameReader(response.body);
295 });
296 const post = (url: string, body: Uint8Array) =>
297 Effect.gen(function* () {
298 const next = yield* send(url, body);
299 const first = yield* next();
300 if (first === undefined) {
301 return yield* new HashError({ reason: "hash part: empty response" });
302 }
303 return decodeScanResult(first);
304 });
305 return {
306 writesSpill: true,
307 hashPart: (payload, opts) =>
308 Effect.gen(function* () {
309 const spill =
310 opts.spill === undefined
311 ? ""
312 : `&key=${encodeURIComponent(opts.spill.key)}&uploadId=${encodeURIComponent(opts.spill.uploadId)}&part=${opts.spill.partNumber}`;
313 const next = yield* send(
314 `https://self${HASH_ROUTE}?base=${opts.base}&remaining=${opts.remaining}&max=${opts.maxObjectSize}${opts.resync ? "&resync=1" : ""}${opts.skip ? `&skip=${opts.skip}` : ""}${spill}`,
315 payload,
316 );
317 const first = yield* next();
318 if (first === undefined) {
319 return yield* new HashError({
320 reason: "hash part: empty response",
321 });
322 }
323 const scan = decodeScanResult(first);
324 if (opts.spill === undefined) return scan;
325 // The part frame arrives once the hasher's upload has finished.
326 const part = next().pipe(
327 Effect.flatMap((bytes) =>
328 bytes === undefined
329 ? Effect.fail(
330 new HashError({ reason: "hash part: no part frame" }),
331 )
332 : Effect.succeed(
333 JSON.parse(new TextDecoder().decode(bytes)) as UploadedPart,
334 ),
335 ),
336 );
337 return { ...scan, part };
338 }),
339 hashBoundsPart: (payload, bounds, opts) =>
340 post(
341 `https://self${HASH_ROUTE}?mode=bounds&base=${opts.base}&max=${opts.maxObjectSize}`,
342 encodeBoundsRequest(payload, bounds),
343 ),
344 resolveDeltas: (bases, jobs, opts) =>
345 Effect.gen(function* () {
346 const next = yield* send(
347 `https://self${HASH_ROUTE}?mode=deltas&max=${opts.maxObjectSize}`,
348 encodeDeltaBatch(bases, jobs),
349 );
350 const first = yield* next();
351 if (first === undefined) {
352 return yield* new HashError({
353 reason: "delta batch: empty response",
354 });
355 }
356 return decodeDeltaResults(first);
357 }),
358 };
359 }),
360);