1/**
2 * A pack source the parser can read WHILE the body is still arriving
3 * (DESIGN §22.6). The receive-pack handler feeds request chunks in; the
4 * parser reads forward through it; nothing waits for the upload to end.
5 *
6 * - Chunks are coalesced into fixed slabs (`SLAB_BYTES`), so a read inside
7 * one slab is a view (the parser's synchronous fast path needs views).
8 * - A read past what has arrived blocks until enough bytes exist or the
9 * body ends; a read that cannot be satisfied at end returns short.
10 * - Retention is bounded: once the parser has consumed past a slab and the
11 * retained total exceeds `retainBytes`, the oldest slab is dropped. A
12 * later read below the retained window (a delta base far behind) is
13 * served by the `fallback` reader — the spilled object, readable once the
14 * body has ended — or fails when there is none.
15 * - Backpressure: `push` waits while more than `backpressureBytes` of
16 * unconsumed bytes are buffered, so a slow parser throttles the upload
17 * instead of growing memory.
18 */
19import * as Effect from "effect/Effect";
20import { StoreError } from "../Protocol/Store.ts";
21import type { RandomAccess } from "../Protocol/PackParser.ts";
22
23// Slabs match the pump's part size so a part read is a view, not a copy.
24export const SLAB_BYTES = 8 * 1024 * 1024;
25export const RETAIN_BYTES = 16 * 1024 * 1024;
26export const BACKPRESSURE_BYTES = 24 * 1024 * 1024;
27// 24 MiB (was 8): the hasher chain runs behind the upload; with less room
28// ahead of the pump the upload stalled on every part (DESIGN §22.7).
29
30interface Slab {
31 readonly start: number;
32 readonly bytes: Uint8Array; // capacity SLAB_BYTES; `filled` bytes valid
33 filled: number;
34}
35
36interface Waiter {
37 readonly need: number;
38 readonly resume: (result: Effect.Effect<void, StoreError>) => void;
39}
40
41export interface StreamingSource extends RandomAccess {
42 /** Bytes received so far. */
43 readonly received: () => number;
44 /** Resolves with the total once the body has ended. */
45 readonly awaitEnd: Effect.Effect<number, StoreError>;
46 /** True once the feeder has ended the body. */
47 readonly ended: () => boolean;
48 /** The parser is done with everything below `offset`; retention may drop it. */
49 readonly release: (offset: number) => void;
50}
51
52export interface StreamingFeeder {
53 readonly source: StreamingSource;
54 /** Appends a chunk; waits while backpressure applies. */
55 readonly push: (chunk: Uint8Array) => Effect.Effect<void, StoreError>;
56 /**
57 * Appends without ever parking: for receivers whose consumer retains
58 * the whole body anyway (the push pump, DESIGN §22.10), backpressure
59 * buys nothing and the per-chunk Effect hop is the receive path's cost.
60 * Returns false once the source has ended or failed.
61 */
62 readonly pushSync: (chunk: Uint8Array) => boolean;
63 /** Ends the body; readers waiting past the end are woken. */
64 readonly end: () => void;
65 /**
66 * Announces that a fallback reader WILL be set (the body is being
67 * spilled): evicted reads after `end` wait for it instead of failing.
68 */
69 readonly expectFallback: () => void;
70 /** Fails every current and future read. */
71 readonly fail: (error: StoreError) => void;
72 /**
73 * Installs the reader for bytes that retention has dropped (the spilled
74 * object). Reads below the retained window wait for it and for `end`.
75 */
76 readonly setFallback: (reader: RandomAccess) => void;
77}
78
79export const makeStreamingSource = (options?: {
80 readonly retainBytes?: number | undefined;
81 readonly backpressureBytes?: number | undefined;
82 readonly slabBytes?: number | undefined;
83}): StreamingFeeder => {
84 const slabBytes = options?.slabBytes ?? SLAB_BYTES;
85 const retainBytes = options?.retainBytes ?? RETAIN_BYTES;
86 const backpressureBytes = options?.backpressureBytes ?? BACKPRESSURE_BYTES;
87 const slabs: Array<Slab> = [];
88 let received = 0;
89 let retainFrom = 0; // first byte still in memory
90 let consumed = 0; // parser's release watermark
91 let ended = false;
92 let total: number | undefined;
93 let failure: StoreError | undefined;
94 let fallback: RandomAccess | undefined;
95 let fallbackExpected = false;
96 const readWaiters: Array<Waiter> = [];
97 const endWaiters: Array<(r: Effect.Effect<number, StoreError>) => void> = [];
98 const pushWaiters: Array<() => void> = [];
99 const fallbackWaiters: Array<() => void> = [];
100
101 const append = (chunk: Uint8Array) => {
102 let at = 0;
103 while (at < chunk.length) {
104 let slab = slabs[slabs.length - 1];
105 if (slab === undefined || slab.filled === slab.bytes.length) {
106 slab = { start: received, bytes: new Uint8Array(slabBytes), filled: 0 };
107 slabs.push(slab);
108 }
109 const take = Math.min(chunk.length - at, slab.bytes.length - slab.filled);
110 slab.bytes.set(chunk.subarray(at, at + take), slab.filled);
111 slab.filled += take;
112 received += take;
113 at += take;
114 }
115 };
116 const wake = () => {
117 for (let i = readWaiters.length - 1; i >= 0; i--) {
118 const w = readWaiters[i]!;
119 if (failure !== undefined) {
120 readWaiters.splice(i, 1);
121 w.resume(Effect.fail(failure));
122 } else if (received >= w.need || ended) {
123 readWaiters.splice(i, 1);
124 w.resume(Effect.void);
125 }
126 }
127 if (ended || failure !== undefined) {
128 const ws = endWaiters.splice(0);
129 for (const w of ws) {
130 w(
131 failure !== undefined ? Effect.fail(failure) : Effect.succeed(total!),
132 );
133 }
134 }
135 };
136 /**
137 * Backpressure must never park the feeder while a reader is waiting for
138 * bytes past the buffer — that reader is the only thing that will ever
139 * consume, so parking would deadlock (a 64 KiB probe against a 1 KiB
140 * budget, or an entry larger than the budget). A waiting reader always
141 * releases the feeder.
142 */
143 const pushMayProceed = () =>
144 received - consumed <= backpressureBytes ||
145 readWaiters.length > 0 ||
146 ended ||
147 failure !== undefined;
148 const wakePush = () => {
149 if (pushMayProceed()) {
150 const ws = pushWaiters.splice(0);
151 for (const w of ws) w();
152 }
153 };
154 const trim = () => {
155 // Drop whole slabs entirely below the consumed watermark once the
156 // retained total exceeds the budget.
157 while (
158 slabs.length > 1 &&
159 received - retainFrom > retainBytes &&
160 slabs[0]!.start + slabs[0]!.filled <= consumed
161 ) {
162 const dropped = slabs.shift()!;
163 retainFrom = dropped.start + dropped.filled;
164 }
165 };
166 const waitFor = (need: number): Effect.Effect<void, StoreError> =>
167 Effect.suspend(() => {
168 if (failure !== undefined) return Effect.fail(failure);
169 if (received >= need || ended) return Effect.void;
170 return Effect.callback<void, StoreError>((resume) => {
171 readWaiters.push({ need, resume });
172 wakePush(); // a waiting reader releases a parked feeder
173 });
174 });
175 const waitForFallback: Effect.Effect<RandomAccess, StoreError> =
176 Effect.suspend(() => {
177 if (failure !== undefined) return Effect.fail(failure);
178 if (fallback !== undefined && ended) return Effect.succeed(fallback);
179 return Effect.callback<RandomAccess, StoreError>((resume) => {
180 fallbackWaiters.push(() => {
181 resume(
182 failure !== undefined
183 ? Effect.fail(failure)
184 : fallback !== undefined
185 ? Effect.succeed(fallback)
186 : Effect.fail(
187 new StoreError({
188 reason:
189 "streaming source: bytes evicted and no fallback reader",
190 }),
191 ),
192 );
193 });
194 });
195 });
196 const wakeFallback = () => {
197 // Once the body has ended, evicted reads either get the fallback reader
198 // or fail — they must never wait forever. When a fallback is expected
199 // (the spill completes after the body ends), they wait for it.
200 if (
201 ended &&
202 (fallback !== undefined || !fallbackExpected || failure !== undefined)
203 ) {
204 const ws = fallbackWaiters.splice(0);
205 for (const w of ws) w();
206 }
207 };
208
209 /** Assembles [offset, end) from retained slabs; a view when in one slab. */
210 const assemble = (offset: number, end: number): Uint8Array => {
211 let first = -1;
212 for (let i = 0; i < slabs.length; i++) {
213 const s = slabs[i]!;
214 if (offset >= s.start && offset < s.start + s.filled) {
215 first = i;
216 break;
217 }
218 }
219 if (first < 0) return new Uint8Array(0);
220 const s0 = slabs[first]!;
221 if (end <= s0.start + s0.filled) {
222 return s0.bytes.subarray(offset - s0.start, end - s0.start);
223 }
224 const out = new Uint8Array(end - offset);
225 let written = 0;
226 for (let i = first; i < slabs.length && offset + written < end; i++) {
227 const s = slabs[i]!;
228 const from = Math.max(offset, s.start) - s.start;
229 const to = Math.min(end, s.start + s.filled) - s.start;
230 if (to <= from) break;
231 out.set(s.bytes.subarray(from, to), written);
232 written += to - from;
233 }
234 return written === out.length ? out : out.subarray(0, written);
235 };
236
237 const source: StreamingSource = {
238 get size() {
239 return total ?? Number.POSITIVE_INFINITY;
240 },
241 received: () => received,
242 ended: () => ended,
243 awaitEnd: Effect.suspend(() => {
244 if (failure !== undefined) return Effect.fail(failure);
245 if (ended) return Effect.succeed(total!);
246 return Effect.callback<number, StoreError>((resume) => {
247 endWaiters.push(resume);
248 });
249 }),
250 release: (offset) => {
251 if (offset > consumed) {
252 consumed = offset;
253 trim();
254 wakePush();
255 }
256 },
257 readSync: (offset, length) => {
258 if (offset < retainFrom) return undefined;
259 const end = Math.min(offset + length, received);
260 if (end <= offset) return ended ? new Uint8Array(0) : undefined;
261 if (end < offset + length && !ended) return undefined;
262 for (const s of slabs) {
263 if (offset >= s.start && end <= s.start + s.filled) {
264 return s.bytes.subarray(offset - s.start, end - s.start);
265 }
266 }
267 return undefined;
268 },
269 evictedBeforeEnd: (offset) => offset < retainFrom && !ended,
270 read: (offset, length) =>
271 Effect.gen(function* () {
272 if (offset < retainFrom) {
273 const reader = yield* waitForFallback;
274 return yield* reader.read(offset, length);
275 }
276 yield* waitFor(offset + length);
277 if (offset < retainFrom) {
278 const reader = yield* waitForFallback;
279 return yield* reader.read(offset, length);
280 }
281 const end = Math.min(offset + length, received);
282 return end <= offset ? new Uint8Array(0) : assemble(offset, end);
283 }),
284 };
285
286 return {
287 source,
288 pushSync: (chunk) => {
289 if (failure !== undefined || ended) return false;
290 append(chunk);
291 wake();
292 return true;
293 },
294 push: (chunk) =>
295 Effect.gen(function* () {
296 if (failure !== undefined) return yield* Effect.fail(failure);
297 if (ended) {
298 return yield* Effect.fail(
299 new StoreError({ reason: "streaming source: push after end" }),
300 );
301 }
302 append(chunk);
303 wake();
304 if (!pushMayProceed()) {
305 yield* Effect.callback<void>((resume) => {
306 pushWaiters.push(() => resume(Effect.void));
307 });
308 }
309 }),
310 expectFallback: () => {
311 fallbackExpected = true;
312 },
313 end: () => {
314 if (ended) return;
315 ended = true;
316 total = received;
317 wake();
318 wakePush();
319 wakeFallback();
320 },
321 fail: (error) => {
322 if (failure !== undefined) return;
323 failure = error;
324 ended = true;
325 total ??= received;
326 wake();
327 wakePush();
328 wakeFallback();
329 },
330 setFallback: (reader) => {
331 fallback = reader;
332 wakeFallback();
333 },
334 };
335};