1/**
2 * side-band-64k multiplexing.
3 *
4 * When `side-band-64k` is negotiated, binary data is carried as pkt-lines
5 * whose first payload byte is the band:
6 *
7 * - **band 1** — pack data (or the receive-pack report)
8 * - **band 2** — progress messages (client prints to stderr as `remote: `)
9 * - **band 3** — fatal error message; the client aborts
10 *
11 * Frame arithmetic: max pkt-line total 65520 = 4 length bytes + 1 band byte
12 * + **65515** data bytes, so every frame carries at most 65515 payload bytes.
13 * Legacy `side-band` (1000-byte packets) is never advertised.
14 */
15import * as Stream from "effect/Stream";
16import { concatBytes, utf8Encode } from "./ObjectCodec.ts";
17import { flushPkt, pktLine } from "./Pkt.ts";
18
19/**
20 * Maximum data bytes per side-band-64k frame (65520 − 4 length − 1 band).
21 */
22export const SIDEBAND_DATA_MAX = 65515;
23
24/**
25 * The three side-band channels.
26 */
27export type SidebandBand = 1 | 2 | 3;
28
29/**
30 * Encodes a single side-band frame. The data must be at most
31 * {@link SIDEBAND_DATA_MAX} bytes — use {@link sidebandFrames} to split
32 * arbitrary payloads.
33 */
34const HEX = "0123456789abcdef";
35
36/** The 5-byte sideband frame header for `dataLength` payload bytes. */
37const sidebandHeader = (band: SidebandBand, dataLength: number): Uint8Array => {
38 const total = dataLength + 5;
39 const out = new Uint8Array(5);
40 out[0] = HEX.charCodeAt((total >> 12) & 0xf);
41 out[1] = HEX.charCodeAt((total >> 8) & 0xf);
42 out[2] = HEX.charCodeAt((total >> 4) & 0xf);
43 out[3] = HEX.charCodeAt(total & 0xf);
44 out[4] = band;
45 return out;
46};
47
48export const sidebandFrame = (
49 band: SidebandBand,
50 data: Uint8Array,
51): Uint8Array => {
52 // One allocation, one copy: header + band + payload in place.
53 const out = new Uint8Array(5 + data.length);
54 out.set(sidebandHeader(band, data.length), 0);
55 out.set(data, 5);
56 return out;
57};
58
59/**
60 * Splits a payload into as many side-band frames as needed (each carrying at
61 * most {@link SIDEBAND_DATA_MAX} data bytes). An empty payload produces no
62 * frames.
63 */
64export const sidebandFrames = (
65 band: SidebandBand,
66 data: Uint8Array,
67): Array<Uint8Array> => {
68 const frames: Array<Uint8Array> = [];
69 for (let pos = 0; pos < data.length; pos += SIDEBAND_DATA_MAX) {
70 frames.push(
71 sidebandFrame(band, data.subarray(pos, pos + SIDEBAND_DATA_MAX)),
72 );
73 }
74 return frames;
75};
76
77/**
78 * Stream transform: wraps every chunk of a byte stream into side-band frames
79 * on the given band. Used to mux the pack stream (band 1) into the response
80 * body; the caller appends the outer flush-pkt after the wrapped stream.
81 */
82export const wrapSideband =
83 (band: SidebandBand) =>
84 <E, R>(
85 stream: Stream.Stream<Uint8Array, E, R>,
86 ): Stream.Stream<Uint8Array, E, R> =>
87 Stream.flatMap(stream, (chunk) =>
88 chunk.length === 0
89 ? Stream.empty
90 : Stream.fromArray(sidebandFrames(band, chunk)),
91 );
92
93/**
94 * Encodes a band-2 progress message (e.g. `Counting objects: 42, done.`).
95 * A trailing newline is appended when missing so the client's stderr stays
96 * line-buffered.
97 */
98export const progressMessage = (message: string): Uint8Array =>
99 concatBytes(
100 sidebandFrames(
101 2,
102 utf8Encode(
103 message.endsWith("\n") || message.endsWith("\r")
104 ? message
105 : `${message}\n`,
106 ),
107 ),
108 );
109
110/**
111 * Encodes a band-3 fatal error message. The client prints it and aborts the
112 * transfer; the server should end the response after sending it.
113 */
114export const sidebandError = (message: string): Uint8Array =>
115 concatBytes(
116 sidebandFrames(
117 3,
118 utf8Encode(message.endsWith("\n") ? message : `${message}\n`),
119 ),
120 );
121
122/** Target bytes per native write from {@link pumpPackBody}. */
123export const PUMP_READ_BYTES = 1 << 20;
124
125/**
126 * Sideband-frames a whole chunk into ONE buffer: `ceil(n / 65515)` frames,
127 * each `header(5) + data`, laid out back to back. One memcpy per byte.
128 */
129export const sidebandFrameAll = (band: SidebandBand, data: Uint8Array) => {
130 const frames = Math.ceil(data.length / SIDEBAND_DATA_MAX);
131 const out = new Uint8Array(data.length + frames * 5);
132 let at = 0;
133 for (let pos = 0; pos < data.length; pos += SIDEBAND_DATA_MAX) {
134 const end = Math.min(pos + SIDEBAND_DATA_MAX, data.length);
135 out.set(sidebandHeader(band, end - pos), at);
136 out.set(data.subarray(pos, end), at + 5);
137 at += 5 + (end - pos);
138 }
139 return out;
140};
141
142/**
143 * Pumps a pack body to the client through a native `IdentityTransformStream`
144 * (DESIGN §22.3). Returns the response body.
145 *
146 * Why this shape, measured on a 40 MiB bundle (DESIGN §22.4):
147 *
148 * - An Effect `Stream` body tops out around 15–20 MiB/s: every chunk is a
149 * fiber round trip. The raw case is a platform-to-platform `pipeTo` —
150 * zero JS per chunk — and runs at the client's line rate (71 MiB/s here).
151 * - Sideband cannot be a bare pipe: the protocol caps a frame at 65515
152 * data bytes. Writing frames one at a time into the identity stream —
153 * whether from a `TransformStream` or a manual loop — also measured
154 * ~15 MiB/s: the identity stream is unbuffered, so each `write` waits
155 * for the consumer, and 64 KiB per round trip is the ceiling.
156 * - So: read ~1 MiB at a time (BYOB `readAtLeast` on byte sources, which
157 * R2 bodies are), frame the whole read into one buffer, one `write`.
158 * 16× fewer round trips; one extra memcpy per byte (~1 ms/MiB).
159 *
160 * Errors on the source abort the writable, which errors the body the
161 * client is reading — the right signal for a truncated pack.
162 */
163export const pumpPackBody = (options: {
164 /** Bytes before the pack (NAK/ACK lines, progress) — sent as-is. */
165 readonly prefix: Uint8Array;
166 readonly source: ReadableStream<Uint8Array>;
167 readonly sideband: boolean;
168 /**
169 * The source is ALREADY sideband-framed (a `bundleSidebandKey` object):
170 * pipe it through natively and only append the flush packet.
171 */
172 readonly framed?: boolean | undefined;
173}): ReadableStream<Uint8Array> => {
174 const { readable, writable } = new IdentityTransformStream();
175 void (async () => {
176 const writer = writable.getWriter();
177 try {
178 await writer.write(options.prefix);
179 if (!options.sideband) {
180 writer.releaseLock();
181 await options.source.pipeTo(writable);
182 return;
183 }
184 if (options.framed === true) {
185 writer.releaseLock();
186 await options.source.pipeTo(writable, { preventClose: true });
187 const tail = writable.getWriter();
188 await tail.write(flushPkt);
189 await tail.close();
190 return;
191 }
192 for await (const chunk of readBig(options.source)) {
193 await writer.write(sidebandFrameAll(1, chunk));
194 }
195 await writer.write(flushPkt);
196 await writer.close();
197 } catch (error) {
198 await writable.abort(error).catch(() => {});
199 }
200 })();
201 return readable;
202};
203
204/**
205 * Yields chunks of about {@link PUMP_READ_BYTES} from a source: BYOB
206 * `readAtLeast` (a workerd extension) fills a whole buffer per read on byte
207 * streams; other streams are read as they come.
208 */
209async function* readBig(
210 source: ReadableStream<Uint8Array>,
211): AsyncGenerator<Uint8Array> {
212 let byob: ReadableStreamBYOBReader | undefined;
213 try {
214 byob = source.getReader({ mode: "byob" });
215 } catch {
216 byob = undefined;
217 }
218 if (byob === undefined) {
219 const reader = source.getReader();
220 for (;;) {
221 const { done, value } = await reader.read();
222 if (done) return;
223 if (value.length > 0) yield value;
224 }
225 }
226 const reader = byob as ReadableStreamBYOBReader & {
227 readAtLeast?: (
228 min: number,
229 view: Uint8Array,
230 ) => Promise<ReadableStreamReadResult<Uint8Array>>;
231 };
232 let buffer = new ArrayBuffer(PUMP_READ_BYTES);
233 for (;;) {
234 const view = new Uint8Array(buffer);
235 const result =
236 reader.readAtLeast !== undefined
237 ? await reader.readAtLeast(view.byteLength, view)
238 : await reader.read(view);
239 if (result.value !== undefined && result.value.byteLength > 0) {
240 yield result.value;
241 // The buffer was transferred into `value`; take it back for reuse.
242 buffer = result.value.buffer as ArrayBuffer;
243 }
244 if (result.done) return;
245 }
246}
247
248/** Byte length of {@link sidebandRechunk}'s output for `size` input bytes. */
249export const sidebandFramedLength = (size: number): number =>
250 size + 5 * Math.ceil(size / SIDEBAND_DATA_MAX);
251
252/**
253 * Re-frames a byte stream into band frames of EXACTLY
254 * {@link SIDEBAND_DATA_MAX} data bytes (the last one shorter), independent
255 * of how the input is chunked — so the output length is a pure function
256 * of the input length ({@link sidebandFramedLength}), which an R2 put of
257 * a stream needs up front. Used to write a bundle's pre-framed twin.
258 */
259export const sidebandRechunk =
260 (band: SidebandBand) =>
261 <E, R>(
262 stream: Stream.Stream<Uint8Array, E, R>,
263 ): Stream.Stream<Uint8Array, E, R> => {
264 let carry: Uint8Array = new Uint8Array(0);
265 const frames = (chunk: Uint8Array): Array<Uint8Array> => {
266 const data = carry.length === 0 ? chunk : concatBytes([carry, chunk]);
267 const out: Array<Uint8Array> = [];
268 let pos = 0;
269 while (data.length - pos >= SIDEBAND_DATA_MAX) {
270 out.push(
271 sidebandFrame(band, data.subarray(pos, pos + SIDEBAND_DATA_MAX)),
272 );
273 pos += SIDEBAND_DATA_MAX;
274 }
275 carry = data.slice(pos);
276 return out;
277 };
278 return Stream.flatMap(stream, (chunk) =>
279 Stream.fromArray(frames(chunk)),
280 ).pipe(
281 Stream.concat(
282 Stream.suspend(() =>
283 carry.length === 0
284 ? Stream.empty
285 : Stream.succeed(sidebandFrame(band, carry)),
286 ),
287 ),
288 );
289 };