1/**
2 * The clone-bundle alarm job (DESIGN.md §12.2) — the v2 serving plane.
3 *
4 * A clone is, on the wire, exactly one pack containing the closure of the
5 * advertised refs. v1 recomputes that pack per clone (walk the commit graph,
6 * walk every tree, emit every object). A **bundle** computes it once, after
7 * pushes settle, and stores it in R2 under a key derived from the ref
8 * snapshot it covers:
9 *
10 * ```
11 * {repoId}/bundles/bundle-{refsHash}.pack
12 * ```
13 *
14 * Serving a clone then becomes "stream these bytes" — no closure walk, no
15 * per-object reads, no compression. Because the key is content-addressed by
16 * its ref snapshot, the bytes are immutable and therefore cacheable at every
17 * edge PoP, and the Worker can stream them straight to the client without
18 * the Durable Object touching a single pack byte.
19 *
20 * The pack is streamed to R2 with a **precomputed content length** (every
21 * entry is `varint(type,size) + zdata`, and both are known from the object
22 * index), so bundling a large repo never buffers the pack in memory.
23 */
24import { BlobStoreError, type BlobStoreShape } from "../BlobStore.ts";
25import { RuntimeContext } from "../../RuntimeContext.ts";
26import * as Effect from "effect/Effect";
27import * as Stream from "effect/Stream";
28import {
29 encodeTypeSize,
30 makeSha1,
31 type Oid,
32 type PackEntryType,
33} from "../Protocol/ObjectCodec.ts";
34import { StoreError, type ManifestEntry } from "../Protocol/Store.ts";
35import { bundleKey, bundleSidebandKey } from "../Store/Keys.ts";
36import { sidebandFramedLength, sidebandRechunk } from "../Protocol/Sideband.ts";
37
38/** A ref as it appears in a bundle's covered snapshot. */
39export interface BundleRef {
40 readonly name: string;
41 readonly oid: string;
42}
43
44/**
45 * What a repo records about its current bundle (stored as JSON in the
46 * `config` table under `bundle`).
47 */
48export interface BundleInfo {
49 /** sha1 over the sorted `name oid` lines the bundle covers. */
50 readonly refsHash: string;
51 /** Full R2 key of the pack. */
52 readonly key: string;
53 /** Object count inside the pack. */
54 readonly objectCount: number;
55 /** Exact byte length of the pack. */
56 readonly size: number;
57 /** The ref snapshot this bundle covers. */
58 readonly refs: ReadonlyArray<BundleRef>;
59 /** Epoch millis the bundle was cut. */
60 readonly createdAt: number;
61}
62
63/**
64 * Repos larger than this are not bundled in v2.0 — the win is clone-storm
65 * absorption on ordinary repos, and an unbounded bundle job would monopolise
66 * the alarm budget. (The streaming multipart upgrade lifts this.)
67 */
68export const MAX_BUNDLE_BYTES = 512 * 1024 * 1024;
69
70/**
71 * sha1 over the sorted `"<name> <oid>"` lines — the identity of a ref
72 * snapshot. Two repos states with the same refs produce the same hash, so
73 * the bundle key is stable and immutable.
74 */
75export const hashRefs = (refs: ReadonlyArray<BundleRef>): string => {
76 const hash = makeSha1();
77 const lines = [...refs]
78 .map((ref) => `${ref.name} ${ref.oid}`)
79 .sort()
80 .join("\n");
81 hash.update(new TextEncoder().encode(lines));
82 return hash.digestHex();
83};
84
85/**
86 * Exact byte length of the pack `entries` will produce: the 12-byte header,
87 * each entry's varint type/size header plus its stored zdata, and the
88 * 20-byte trailer. Lets the bundle stream straight into R2 without
89 * buffering.
90 */
91export const packByteLength = (
92 entries: ReadonlyArray<ManifestEntry>,
93): number => {
94 let total = 12 + 20;
95 for (const entry of entries) {
96 total +=
97 encodeTypeSize(entry.type as PackEntryType, entry.size).length +
98 entry.zsize;
99 }
100 return total;
101};
102
103/** Dependencies of {@link runBundleJob}. */
104export interface BundleJobOptions {
105 readonly repoId: string;
106 readonly refs: ReadonlyArray<BundleRef>;
107 /** The full-clone manifest (closure of every ref). */
108 readonly entries: ReadonlyArray<ManifestEntry>;
109 /** Emits the pack for `entries` — `RepoObject`'s `packStream`. */
110 readonly packStream: (
111 entries: ReadonlyArray<ManifestEntry>,
112 ) => Stream.Stream<Uint8Array, StoreError>;
113 readonly blobs: BlobStoreShape;
114 readonly maxBytes?: number | undefined;
115}
116
117/**
118 * Cuts a clone bundle for the given ref snapshot and returns what to record
119 * on the repo. Returns `undefined` when the repo is too large to bundle in
120 * v2.0 or has no refs (nothing to clone).
121 */
122export const runBundleJob = (
123 options: BundleJobOptions,
124): Effect.Effect<BundleInfo | undefined, StoreError> =>
125 Effect.gen(function* () {
126 if (options.refs.length === 0 || options.entries.length === 0) {
127 return undefined;
128 }
129 const size = packByteLength(options.entries);
130 if (size > (options.maxBytes ?? MAX_BUNDLE_BYTES)) {
131 return undefined;
132 }
133 const refsHash = yield* Effect.sync(() => hashRefs(options.refs));
134 const key = bundleKey(options.repoId, refsHash);
135
136 // The pack streams straight into R2 (its byte length is known up front),
137 // so a large bundle is never buffered. StoreError may surface from the
138 // stream itself (an object read failing mid-pack); R2Error from the put.
139 yield* options.blobs
140 .put(
141 key,
142 options
143 .packStream(options.entries)
144 .pipe(
145 Stream.mapError(
146 (error) => new BlobStoreError({ reason: error.reason }),
147 ),
148 ),
149 { contentLength: size },
150 )
151 .pipe(
152 Effect.mapError(
153 (error: BlobStoreError) =>
154 new StoreError({ reason: `blob put ${key}: ${error.reason}` }),
155 ),
156 Effect.provide(RuntimeContext.phantom),
157 );
158
159 // The pre-framed twin (Keys.ts `bundleSidebandKey`): one more pass over
160 // the same entries, framed deterministically so the length is known.
161 const sidebandKey = bundleSidebandKey(options.repoId, refsHash);
162 yield* options.blobs
163 .put(
164 sidebandKey,
165 options.packStream(options.entries).pipe(
166 Stream.mapError(
167 (error) => new BlobStoreError({ reason: error.reason }),
168 ),
169 sidebandRechunk(1),
170 ),
171 { contentLength: sidebandFramedLength(size) },
172 )
173 .pipe(
174 Effect.mapError(
175 (error: BlobStoreError) =>
176 new StoreError({
177 reason: `blob put ${sidebandKey}: ${error.reason}`,
178 }),
179 ),
180 Effect.provide(RuntimeContext.phantom),
181 );
182
183 return {
184 refsHash,
185 key,
186 objectCount: options.entries.length,
187 size,
188 refs: options.refs,
189 createdAt: Date.now(),
190 } satisfies BundleInfo;
191 });
192
193/**
194 * Decides whether a recorded bundle can serve this fetch (DESIGN.md §12.2,
195 * tier 1 "exact match").
196 *
197 * The bundle holds the closure of the refs it covers, so it is safe for any
198 * request that:
199 *
200 * - brings no `have`s and no shallow state (it is a *clone*, not an
201 * incremental fetch — otherwise the client would receive objects it
202 * already has, which is legal but wasteful, or miss the negotiation the
203 * protocol promised), and
204 * - wants only oids the bundle actually covers.
205 *
206 * Extra objects in a pack are legal, so a superset bundle serves a
207 * single-branch clone fine.
208 */
209export const bundleCovers = (
210 bundle: BundleInfo,
211 request: {
212 readonly wants: ReadonlyArray<Oid>;
213 readonly haves: ReadonlyArray<Oid>;
214 readonly depth?: number | undefined;
215 readonly clientShallow: ReadonlyArray<Oid>;
216 },
217): boolean => {
218 const uncovered = bundleUncovered(bundle, request);
219 return uncovered !== undefined && uncovered.length === 0;
220};
221
222/**
223 * How much of a request this bundle can serve (DESIGN.md §21.2).
224 *
225 * - `undefined` — unusable: the request negotiates (haves) or is
226 * shallow. Serve dynamically.
227 * - `[]` — the bundle covers every want: stream its bytes verbatim.
228 * - non-empty — the wants the bundle does not name directly. Their
229 * closure, computed with the bundle's refs as haves, is a delta pack
230 * the Worker concatenates onto the bundle.
231 *
232 * A want the bundle does not *name* is usually still mostly IN it — the
233 * ordinary case is `main` having moved a few commits past the snapshot,
234 * so the bundle holds all but the new objects. Union completeness holds
235 * regardless: the bundle carries closure(bundle refs), the delta carries
236 * closure(wants) minus that, so together they cover closure(wants) with
237 * no duplicate entries. Only the *size* of the delta varies, and the
238 * caller drops the splice when it stops being a saving.
239 *
240 * The partial case is what keeps the fast path alive on a repo that is
241 * being pushed to: previously ANY ref moving after the bundle was cut
242 * made a clone's want-set uncovered, dropping every clone back onto the
243 * Durable Object until the next full re-cut.
244 */
245export const bundleUncovered = (
246 bundle: BundleInfo,
247 request: {
248 readonly wants: ReadonlyArray<Oid>;
249 readonly haves: ReadonlyArray<Oid>;
250 readonly depth?: number | undefined;
251 readonly clientShallow: ReadonlyArray<Oid>;
252 },
253): ReadonlyArray<Oid> | undefined => {
254 if (
255 request.haves.length > 0 ||
256 request.depth !== undefined ||
257 request.clientShallow.length > 0 ||
258 request.wants.length === 0
259 ) {
260 return undefined;
261 }
262 const covered = new Set(bundle.refs.map((ref) => ref.oid));
263 return request.wants.filter((want) => !covered.has(want));
264};