1/**
2 * The zlib boundary — the ONLY module in `src/Git/Protocol/` allowed to touch
3 * `node:zlib` (per DESIGN §7).
4 *
5 * | Need | Tool |
6 * |-----------------------------------------------|-------------------------------|
7 * | Pack entry parsing (unknown compressed span) | `createInflate()` per entry |
8 * | One-shot inflate of an exactly-known span | `inflateSync` |
9 * | Compression at rest (delta-resolved objects) | `deflateSync` (level 6) |
10 *
11 * The streaming-inflate trick (workerd-verified on 1.20260801.1): each pack
12 * entry's zlib stream (RFC 1950) is self-terminating; `createInflate()`'s
13 * `end` event fires at `Z_STREAM_END` mid-`write` without `.end()`, and
14 * `inflate.bytesWritten` then reports the **exact compressed bytes
15 * consumed** — the only way to find the next entry.
16 * `DecompressionStream("deflate")` is disqualified for this: it errors on
17 * trailing bytes and does not report consumption.
18 */
19import * as Effect from "effect/Effect";
20import * as Schema from "effect/Schema";
21import zlib from "node:zlib";
22
23/**
24 * Error raised on zlib failures: corrupt streams, truncated input, or an
25 * inflated output exceeding the caller's cap.
26 */
27export class ZlibError extends Schema.TaggedError<ZlibError>()("ZlibError", {
28 reason: Schema.String,
29}) {}
30
31/**
32 * The result of inflating one pack entry: the inflated content and the exact
33 * number of compressed bytes consumed from the input (which locates the next
34 * entry).
35 */
36export interface InflatedEntry {
37 readonly content: Uint8Array;
38 readonly bytesConsumed: number;
39}
40
41/** Feed granularity for the streaming inflater. */
42const CHUNK = 65536;
43
44/**
45 * Streaming-inflates exactly one zlib stream from `buf` starting at
46 * `offset`, tolerating arbitrary trailing bytes (the next pack entries).
47 *
48 * Input is fed in 64 KiB chunks; when the stream reaches `Z_STREAM_END` the
49 * `end` event fires and `bytesWritten` gives the exact compressed span.
50 * Truncated input (no `Z_STREAM_END` before the buffer ends) and corrupt
51 * streams fail with {@link ZlibError}, as does output growing past
52 * `options.maxOutput`.
53 */
54/**
55 * Synchronous single-entry inflate — the fast path.
56 *
57 * The streaming implementation below builds a `zlib.createInflate()`
58 * Transform (EventEmitter, internal buffers) and takes an async event-loop
59 * round trip **per object**. Measured on a 13.7k-object push: ~1.49 ms of
60 * CPU per object, roughly 50x the cost of actually inflating ~2.7 KB.
61 *
62 * `_processChunk` is what Node's own `zlib.inflateSync` uses internally: it
63 * drives the engine synchronously and leaves the consumed-input count in
64 * `bytesWritten`, which is exactly the pair a pack parser needs. It is an
65 * internal API, so this returns `undefined` when it is unavailable (or
66 * behaves unexpectedly) and the caller falls back to the stream path —
67 * notably relevant on workerd, whose `node:zlib` is a compatibility shim.
68 */
69/**
70 * Second synchronous path, on the PUBLIC API: `inflateSync(.., { info: true })`
71 * returns `{ buffer, engine }`, and `engine.bytesWritten` is the input
72 * consumed — the same pair `_processChunk` gives, without the internal.
73 * zlib stops at the deflate stream's end, so trailing pack bytes are
74 * ignored. Same `expectedSize` verification as above (a shim that returns
75 * only a first chunk is caught by it). `undefined` = use the stream path.
76 */
77const inflateEntryInfoSync = (
78 buf: Uint8Array,
79 offset: number,
80 maxOutput: number | undefined,
81 expectedSize: number | undefined,
82): InflatedEntry | undefined => {
83 try {
84 const result = zlib.inflateSync(buf.subarray(offset), {
85 info: true,
86 } as zlib.ZlibOptions) as unknown as {
87 buffer?: Uint8Array;
88 engine?: { bytesWritten?: number };
89 };
90 const output = result?.buffer;
91 const bytesConsumed = result?.engine?.bytesWritten;
92 if (
93 output === undefined ||
94 typeof bytesConsumed !== "number" ||
95 bytesConsumed <= 0 ||
96 (maxOutput !== undefined && output.length > maxOutput) ||
97 (expectedSize !== undefined && output.length !== expectedSize)
98 ) {
99 return undefined;
100 }
101 return {
102 content: new Uint8Array(output.buffer, output.byteOffset, output.length),
103 bytesConsumed,
104 };
105 } catch {
106 return undefined;
107 }
108};
109
110const inflateEntryUnsafeSync = (
111 buf: Uint8Array,
112 offset: number,
113 maxOutput: number | undefined,
114 expectedSize: number | undefined,
115): InflatedEntry | undefined => {
116 const engine = zlib.createInflate() as unknown as {
117 _processChunk?: (chunk: Uint8Array, flushFlag: number) => Uint8Array;
118 bytesWritten?: number;
119 close?: () => void;
120 };
121 if (typeof engine._processChunk !== "function") {
122 engine.close?.();
123 return inflateEntryInfoSync(buf, offset, maxOutput, expectedSize);
124 }
125 try {
126 const output = engine._processChunk(
127 buf.subarray(offset),
128 zlib.constants.Z_SYNC_FLUSH,
129 );
130 const bytesConsumed = engine.bytesWritten;
131 if (typeof bytesConsumed !== "number" || bytesConsumed <= 0) {
132 return undefined;
133 }
134 if (maxOutput !== undefined && output.length > maxOutput) {
135 // Let the caller's streaming path produce the typed error.
136 return undefined;
137 }
138 // MUST verify: workerd's `node:zlib` shim returns only the first chunk
139 // (observed: 525,107 bytes of a 1,337,267-byte object) instead of
140 // looping like Node's implementation, and reports a plausible
141 // `bytesWritten` with it. Checking against the size the pack header
142 // already declares turns that silent truncation into a fallback.
143 if (expectedSize !== undefined && output.length !== expectedSize) {
144 // workerd's shim returned only the first output chunk: the public
145 // `inflateSync({ info })` loops to completion — try it before paying
146 // for the streaming path (measured: the entries that hit this were
147 // the only async work left in a production ingest).
148 return inflateEntryInfoSync(buf, offset, maxOutput, expectedSize);
149 }
150 return {
151 content: new Uint8Array(output.buffer, output.byteOffset, output.length),
152 bytesConsumed,
153 };
154 } catch {
155 // Corrupt stream, truncated window, or an unsupported shim — the
156 // stream path re-runs it and reports the real error.
157 return undefined;
158 } finally {
159 engine.close?.();
160 }
161};
162
163/**
164 * The synchronous fast path alone (DESIGN §22.5): the exact-span inflate
165 * when the whole compressed stream lies within `buf` and the platform
166 * offers a sync engine; `undefined` means "use {@link inflateEntry}".
167 */
168export const inflateEntrySync = (
169 buf: Uint8Array,
170 offset: number,
171 options: {
172 readonly maxOutput?: number | undefined;
173 readonly expectedSize: number;
174 },
175): InflatedEntry | undefined =>
176 offset >= 0 && offset < buf.length
177 ? inflateEntryUnsafeSync(
178 buf,
179 offset,
180 options.maxOutput,
181 options.expectedSize,
182 )
183 : undefined;
184
185export const inflateEntry = (
186 buf: Uint8Array,
187 offset: number,
188 options?: {
189 readonly maxOutput?: number | undefined;
190 /**
191 * The uncompressed size the pack header declares for this entry. Used
192 * to validate the synchronous fast path (see above); without it the
193 * fast path is skipped, because an unverified result cannot be trusted.
194 */
195 readonly expectedSize?: number | undefined;
196 },
197): Effect.Effect<InflatedEntry, ZlibError> =>
198 Effect.suspend(() => {
199 const fast =
200 offset >= 0 && offset < buf.length && options?.expectedSize !== undefined
201 ? inflateEntryUnsafeSync(
202 buf,
203 offset,
204 options.maxOutput,
205 options.expectedSize,
206 )
207 : undefined;
208 return fast === undefined
209 ? inflateEntryStreaming(buf, offset, options)
210 : Effect.succeed(fast);
211 });
212
213/** The portable fallback: a real zlib stream per entry. */
214const inflateEntryStreaming = (
215 buf: Uint8Array,
216 offset: number,
217 options?: { readonly maxOutput?: number | undefined },
218): Effect.Effect<InflatedEntry, ZlibError> =>
219 Effect.callback<InflatedEntry, ZlibError>((resume) => {
220 if (offset < 0 || offset >= buf.length) {
221 resume(
222 Effect.fail(
223 new ZlibError({
224 reason: `inflate offset ${offset} out of range (buffer ${buf.length})`,
225 }),
226 ),
227 );
228 return;
229 }
230 const maxOutput = options?.maxOutput;
231 const inflater = zlib.createInflate();
232 const chunks: Array<Uint8Array> = [];
233 let outBytes = 0;
234 let settled = false;
235
236 const fail = (reason: string): void => {
237 if (settled) return;
238 settled = true;
239 inflater.destroy();
240 resume(Effect.fail(new ZlibError({ reason })));
241 };
242
243 inflater.on("data", (chunk: Uint8Array) => {
244 if (settled) return;
245 outBytes += chunk.length;
246 if (maxOutput !== undefined && outBytes > maxOutput) {
247 fail(`inflated output exceeds ${maxOutput} bytes`);
248 return;
249 }
250 chunks.push(chunk);
251 });
252 inflater.on("error", (error: Error) => {
253 fail(`inflate failed at offset ${offset}: ${error.message}`);
254 });
255 inflater.once("end", () => {
256 if (settled) return;
257 settled = true;
258 const bytesConsumed = inflater.bytesWritten;
259 const content = new Uint8Array(outBytes);
260 let pos = 0;
261 for (const chunk of chunks) {
262 content.set(chunk, pos);
263 pos += chunk.length;
264 }
265 inflater.destroy();
266 resume(Effect.succeed({ content, bytesConsumed }));
267 });
268
269 const end = buf.length;
270 let pos = offset;
271 const feed = (): void => {
272 if (settled) return;
273 if (pos >= end) {
274 // ran out of input without Z_STREAM_END — flushing EOF surfaces the
275 // "unexpected end of file" error through the error handler above.
276 inflater.end();
277 return;
278 }
279 const next = Math.min(pos + CHUNK, end);
280 const chunk = buf.subarray(pos, next);
281 pos = next;
282 inflater.write(chunk, () => feed());
283 };
284 feed();
285
286 return Effect.sync(() => {
287 // interruption: tear the native stream down
288 if (!settled) {
289 settled = true;
290 inflater.destroy();
291 }
292 });
293 });
294
295/**
296 * One-shot inflate of an exactly-known zlib span (e.g. a stored `zdata`
297 * BLOB). Strict: trailing garbage or truncation fails.
298 */
299export const inflate = (
300 data: Uint8Array,
301): Effect.Effect<Uint8Array, ZlibError> =>
302 Effect.try({
303 try: () => new Uint8Array(zlib.inflateSync(data)),
304 catch: (error) =>
305 new ZlibError({
306 reason: `inflateSync failed: ${error instanceof Error ? error.message : String(error)}`,
307 }),
308 });
309
310/**
311 * One-shot deflate at the given level (default 6 — the at-rest compression
312 * level for delta-resolved objects; non-delta pack entries are stored
313 * verbatim and never re-compressed).
314 */
315export const deflate = (
316 data: Uint8Array,
317 level = 6,
318): Effect.Effect<Uint8Array, ZlibError> =>
319 Effect.try({
320 try: () => new Uint8Array(zlib.deflateSync(data, { level })),
321 catch: (error) =>
322 new ZlibError({
323 reason: `deflateSync failed: ${error instanceof Error ? error.message : String(error)}`,
324 }),
325 });
326
327/**
328 * Inflates a COMPLETE deflate stream whose exact span is known (DESIGN
329 * §22.8): the plain one-shot `inflateSync`, measured ~4x cheaper than the
330 * consumed-count paths on production. Verified against the declared size.
331 */
332export const inflateExactSpan = (
333 span: Uint8Array,
334 expectedSize: number,
335): Effect.Effect<Uint8Array, ZlibError> =>
336 Effect.try({
337 try: () => {
338 const out = new Uint8Array(zlib.inflateSync(span));
339 if (out.length !== expectedSize) {
340 throw new Error(
341 `inflated ${out.length} bytes, expected ${expectedSize}`,
342 );
343 }
344 return out;
345 },
346 catch: (error) =>
347 new ZlibError({
348 reason: `inflateSync failed: ${error instanceof Error ? error.message : String(error)}`,
349 }),
350 });