1/**
2 * Receiving a git wire request body (DESIGN.md §3.6, §22.6, §22.10).
3 *
4 * A push body must NEVER be materialized: `request.arrayBuffer` on a
5 * 100 MiB pack inside a 128 MB isolate is an OOM (the failure that rejected
6 * the full alchemy-history push with a 500). The body is fed chunk by
7 * chunk into a {@link StreamingFeeder} the parser reads from as bytes
8 * arrive; the parser ends the body's life in memory by dispatching it in
9 * uniform parts to the hasher isolates, which write the parts to blob
10 * storage (the spill) while verifying them.
11 *
12 * The command section of a receive-pack request lives in the first bytes,
13 * so the first {@link HEAD_BYTES} are enough to parse the ref commands
14 * before the pack has arrived.
15 */
16import * as Effect from "effect/Effect";
17import { StoreError } from "../Protocol/Store.ts";
18import type { StreamingFeeder } from "./StreamingSource.ts";
19
20/** Head prefix that is enough for command-section parsing. */
21export const HEAD_BYTES = 1024 * 1024;
22
23/**
24 * Uniform multipart part size — also the hasher part size, since each
25 * hasher isolate writes the part it verifies (DESIGN §22.10). 8 MiB is
26 * above R2's 5 MiB minimum and its 10,000-part limit still allows ~78 GiB.
27 */
28export const SPILL_PART_BYTES = 8 * 1024 * 1024;
29
30/**
31 * Feeds a request body into the streaming source and ends it when the
32 * body ends. The reader respects feeder backpressure while application
33 * authorization or validation is running, and cancellation cancels the
34 * underlying body reader. Resolves with the total; a read failure
35 * fails the feeder so every waiting reader wakes with the error.
36 */
37export const feedBody = (
38 body: ReadableStream<Uint8Array> | null,
39 feeder: StreamingFeeder,
40): Effect.Effect<{ readonly total: number }, StoreError> =>
41 Effect.tryPromise({
42 try: async (signal) => {
43 let total = 0;
44 if (body !== null) {
45 const reader = body.getReader();
46 const cancel = () => {
47 void reader.cancel().catch(() => {});
48 };
49 signal.addEventListener("abort", cancel, { once: true });
50 try {
51 while (!signal.aborted) {
52 const { value, done } = await reader.read();
53 if (done) break;
54 total += value.length;
55 await Effect.runPromise(feeder.push(value), { signal });
56 }
57 } finally {
58 signal.removeEventListener("abort", cancel);
59 reader.releaseLock();
60 }
61 }
62 feeder.end();
63 return { total };
64 },
65 catch: (error) =>
66 new StoreError({ reason: `incoming body read: ${String(error)}` }),
67 }).pipe(Effect.tapError((error) => Effect.sync(() => feeder.fail(error))));