clover's git

the-knife

1	import { LIVE_OBJECTS } from "../Store/ObjectStore.ts";
2	/**
3	* The compaction alarm job (DESIGN.md §12.1) — the v2 storage plane.
4	*
5	* v1 keeps every object's bytes in a DO SQLite row, which caps a repo at the
6	* 10 GB DO limit and puts all clone bandwidth behind one single-threaded
7	* object. Compaction moves those bytes into **immutable, content-addressed
8	* R2 packs** and leaves only the index behind:
9	*
10	* ```
11	* objects.location: 'row' → 'pack'
12	* objects.zdata: BLOB → NULL (bytes now live in R2)
13	* objects.pack_id / pack_offset: (where in R2 they live)
14	* ```
15	*
16	* The pack we write is a **real git packfile** (`PACK`, version 2, count,
17	* then `varint(type,size) + zdata` per entry, then the SHA-1 trailer), so it
18	* can later be served verbatim as a clone bundle or a `packfile-uris`
19	* target. That works only because objects are stored **pre-deflated**: a
20	* pack entry is the stored bytes with a small header in front, so compaction
21	* is a copy, never a recompression.
22	*
23	* `pack_offset` deliberately addresses each object's **zdata span**, not its
24	* entry header — a read is then one ranged R2 GET returning exactly the
25	* bytes a `location='row'` object would have held (see `ObjectStore`).
26	*
27	* Bounded per run and alarm-re-armable, like the purge job: at most
28	* {@link MAX_OBJECTS_PER_RUN} objects / {@link MAX_BYTES_PER_RUN} bytes are
29	* moved per alarm, and the job reports whether more remain.
30	*/
31	import type { BlobStoreError, BlobStoreShape } from "../BlobStore.ts";
32	import { RuntimeContext } from "../../RuntimeContext.ts";
33	import * as Effect from "effect/Effect";
34	import {
35	concatBytes,
36	bytesToHex,
37	encodeTypeSize,
38	makeSha1,
39	type PackEntryType,
40	} from "../Protocol/ObjectCodec.ts";
41	import { StoreError } from "../Protocol/Store.ts";
42	import { packKey, isWirePackId, packKeyOf } from "../Store/Keys.ts";
43	import type { SqlClient } from "../Store/Sql.ts";
44
45	/** Objects moved per alarm run. */
46	export const MAX_OBJECTS_PER_RUN = 2_000;
47
48	/**
49	* Byte budget per alarm run (assembled in memory before the R2 put). Well
50	* under the 128 MB isolate limit; the streaming-multipart upgrade lifts it.
51	*/
52	export const MAX_BYTES_PER_RUN = 32 * 1024 * 1024;
53
54	/**
55	* Loose-bytes threshold that arms compaction (DESIGN.md §12.1).
56	*/
57	// Measured (DESIGN §22): a 44k-object repo left loose serves its dynamic
58	// path at ~3 MB/s from 44k SQLite point reads, and every metadata scan
59	// walks pages dense with inline zdata. Packing early is what makes reads
60	// window-coalesced and keeps the objects table small.
61	export const COMPACT_BYTES_THRESHOLD = 64 * 1024 * 1024;
62
63	/** Loose-object-count threshold that arms compaction. */
64	export const COMPACT_COUNT_THRESHOLD = 4_000;
65
66	/**
67	* Only blobs (git type 3) are ever moved into R2 packs. Commits, trees and
68	* tags stay `location='row'` in SQLite for the life of the repo.
69	*
70	* Measured (DESIGN §22): a fetch's want→have closure is a tree walk, and a
71	* tree walk over pack storage is the worst access pattern a window cache
72	* can see — trees are tiny and, packed in oid order, scattered across
73	* every 4 MiB window of every pack. On a 15.6k-object repo the walk alone
74	* cost 12 s of TTFB (27 s before batching); the same walk over rows is a
75	* handful of batched SELECTs. Trees + commits are a few percent of a
76	* repo's bytes, so keeping them local buys sub-second negotiation for
77	* every incremental fetch at negligible storage cost.
78	*/
79	export const BLOB_TYPE = 3;
80
81	/** Outcome of one compaction run. */
82	export interface CompactOutcome {
83	/** Objects moved into the pack this run. */
84	readonly moved: number;
85	/** Bytes of zdata moved. */
86	readonly bytes: number;
87	/** sha1 of the pack written (the rows' `pack_id`), else `undefined`. */
88	readonly packId: string | undefined;
89	/** True when loose objects remain — the caller re-arms the alarm. */
90	readonly more: boolean;
91	}
92
93	/** Dependencies of {@link runCompactJob}. */
94	export interface CompactJobOptions {
95	readonly repoId: string;
96	readonly sql: SqlClient;
97	readonly blobs: BlobStoreShape;
98	readonly maxObjects?: number | undefined;
99	readonly maxBytes?: number | undefined;
100	}
101
102	interface LooseRow extends Record<
103	string,
104	string | number | ArrayBuffer | null
105	> {
106	readonly oid: string;
107	readonly type: number;
108	readonly size: number;
109	readonly zsize: number;
110	readonly zdata: ArrayBuffer | null;
111	}
112
113	const runR2 =
114	(what: string) =>
115	<A>(
116	effect: Effect.Effect<A, BlobStoreError, RuntimeContext>,
117	): Effect.Effect<A, StoreError> =>
118	effect.pipe(
119	Effect.mapError(
120	(error) => new StoreError({ reason: `${what}: ${error.reason}` }),
121	),
122	Effect.provide(RuntimeContext.phantom),
123	);
124
125	const PACK_SIGNATURE = new Uint8Array([0x50, 0x41, 0x43, 0x4b]); // "PACK"
126
127	/** `PACK` + version 2 + object count. */
128	const packHeader = (count: number): Uint8Array => {
129	const header = new Uint8Array(12);
130	header.set(PACK_SIGNATURE, 0);
131	new DataView(header.buffer).setUint32(4, 2);
132	new DataView(header.buffer).setUint32(8, count);
133	return header;
134	};
135
136	/**
137	* Compacts up to one run's worth of loose objects into a single immutable
138	* R2 pack and repoints their rows at it.
139	*
140	* Ordering note: the R2 put happens **before** the row flip, so a crash in
141	* between leaves an unreferenced pack (harmless, content-addressed, GC-able)
142	* rather than rows pointing at bytes that do not exist.
143	*/
144	export const runCompactJob = (
145	options: CompactJobOptions,
146	): Effect.Effect<CompactOutcome, StoreError> =>
147	Effect.gen(function* () {
148	const maxObjects = options.maxObjects ?? MAX_OBJECTS_PER_RUN;
149	const maxBytes = options.maxBytes ?? MAX_BYTES_PER_RUN;
150
151	// Oldest-first by oid keeps runs deterministic and re-runnable.
152	const rows = yield* options.sql.all<LooseRow>(
153	`SELECT oid, type, size, zsize, zdata FROM objects
154	WHERE location = 'row' AND type = ${BLOB_TYPE} AND ${LIVE_OBJECTS}
155	ORDER BY oid LIMIT ?`,
156	maxObjects + 1,
157	);
158	const eligible = rows.slice(0, maxObjects);
159	if (eligible.length === 0) {
160	return { moved: 0, bytes: 0, packId: undefined, more: false };
161	}
162
163	// Assemble the pack, recording each object's zdata offset as we go.
164	const chunks: Array<Uint8Array> = [];
165	const placements: Array<{ oid: string; offset: number; zsize: number }> =
166	[];
167	let offset = 0;
168	let budget = 0;
169	const push = (bytes: Uint8Array) => {
170	chunks.push(bytes);
171	offset += bytes.length;
172	};
173
174	const included: Array<LooseRow> = [];
175	for (const row of eligible) {
176	if (row.zdata === null) continue; // location='row' with NULL zdata: skip
177	const zdata = new Uint8Array(row.zdata);
178	if (budget > 0 && budget + zdata.length > maxBytes) break;
179	included.push(row);
180	budget += zdata.length;
181	push(
182	yield* Effect.sync(() =>
183	// Row types are the 1..4 non-delta git types (the ingest path only
184	// ever stores those), which is exactly `PackEntryType`.
185	encodeTypeSize(row.type as PackEntryType, row.size),
186	),
187	);
188	placements.push({ oid: row.oid, offset, zsize: zdata.length });
189	push(zdata);
190	}
191	if (included.length === 0) {
192	return { moved: 0, bytes: 0, packId: undefined, more: false };
193	}
194
195	const header = packHeader(included.length);
196	const body = [header, ...chunks];
197	// One pass: the raw digest is the pack trailer, its hex is the pack id
198	// (content-addressed, so a retried run rewrites the identical key).
199	const trailer = yield* Effect.sync(() => {
200	const hash = makeSha1();
201	for (const chunk of body) hash.update(chunk);
202	return hash.digest();
203	});
204	const trailerHex = yield* Effect.sync(() => bytesToHex(trailer));
205
206	const total =
207	body.reduce((sum, chunk) => sum + chunk.length, 0) + trailer.length;
208	const pack = yield* Effect.sync(() => {
209	const out = new Uint8Array(total);
210	let at = 0;
211	for (const chunk of body) {
212	out.set(chunk, at);
213	at += chunk.length;
214	}
215	out.set(trailer, at);
216	return out;
217	});
218
219	// Offsets recorded above are relative to the start of the entry stream;
220	// shift them past the 12-byte header to get absolute file offsets.
221	const packId = trailerHex;
222	const key = packKey(options.repoId, packId);
223	yield* runR2(`blob put ${key}`)(
224	options.blobs.put(key, pack, { contentLength: pack.length }),
225	);
226
227	yield* options.sql
228	.transactionSync((raw) => {
229	for (const placement of placements) {
230	raw.exec(
231	`UPDATE objects
232	SET location = 'pack', pack_id = ?, pack_offset = ?, zdata = NULL
233	WHERE oid = ? AND location = 'row' AND ${LIVE_OBJECTS}`,
234	packId,
235	placement.offset + header.length,
236	placement.oid,
237	);
238	}
239	})
240	.pipe(Effect.asVoid);
241
242	return {
243	moved: included.length,
244	bytes: budget,
245	packId,
246	more: rows.length > maxObjects || included.length < eligible.length,
247	};
248	});
249
250	/**
251	* True when the repo has enough loose bytes/objects to be worth compacting
252	* (DESIGN.md §12.1 thresholds).
253	*/
254	export const shouldCompact = (
255	sql: SqlClient,
256	options?: {
257	readonly bytesThreshold?: number | undefined;
258	readonly countThreshold?: number | undefined;
259	},
260	): Effect.Effect<boolean, StoreError> =>
261	Effect.gen(function* () {
262	const row = yield* sql.first<{ n: number; bytes: number }>(
263	`SELECT COUNT(*) AS n, COALESCE(SUM(zsize), 0) AS bytes
264	FROM objects WHERE location = 'row' AND type = ${BLOB_TYPE} AND ${LIVE_OBJECTS}`,
265	);
266	if (row === undefined) return false;
267	return (
268	row.bytes >= (options?.bytesThreshold ?? COMPACT_BYTES_THRESHOLD) ||
269	row.n >= (options?.countThreshold ?? COMPACT_COUNT_THRESHOLD)
270	);
271	});
272
273	// ─────────────────────────────────────────────────────────────────────────────
274	// Geometric pack merging (DESIGN.md §21, Continuity learnings)
275	// ─────────────────────────────────────────────────────────────────────────────
276
277	/**
278	* Merge whenever a pack is smaller than FACTOR × the sum of all packs
279	* smaller than it — the classic geometric invariant. Holding it bounds
280	* pack count at O(log(total bytes)), which bounds the fan-out of ranged
281	* reads on the serving path.
282	*/
283	export const GEOMETRIC_FACTOR = 2;
284
285	/** Per-run ceiling on merged input, keeping a run inside the alarm budget. */
286	export const MAX_MERGE_INPUT_BYTES = 96 * 1024 * 1024;
287
288	export interface MergeOutcome {
289	/** Source packs merged this run (0 = the invariant already holds). */
290	readonly packs: number;
291	/** Objects now pointing at the merged pack. */
292	readonly moved: number;
293	/** Bytes of merged pack written. */
294	readonly bytes: number;
295	readonly packId: string | undefined;
296	/**
297	* R2 keys of the source packs, now unreferenced. NOT deleted here: an
298	* in-flight read planned before the row flip may still hold spans into
299	* them, so the caller deletes these on a LATER run (grace period).
300	*/
301	readonly pendingDelete: ReadonlyArray<string>;
302	/** More merging remains (input cap hit); re-arm the alarm. */
303	readonly more: boolean;
304	}
305
306	const noMerge: MergeOutcome = {
307	packs: 0,
308	moved: 0,
309	bytes: 0,
310	packId: undefined,
311	pendingDelete: [],
312	more: false,
313	};
314
315	/**
316	* Restores the geometric invariant over a repo's compacted packs by
317	* merging the smallest run of violating packs into one.
318	*
319	* Merging is pure concatenation — our packs store non-delta entries
320	* (varint header + zdata), so a merged pack is `header(Σcounts) +
321	* body₁ + … + bodyₙ + sha1`, and every row's offset shifts by a single
322	* per-source delta (the bytes of bodies concatenated before it). No
323	* recompression, no delta re-resolution: O(bytes) of blob IO plus one
324	* UPDATE per source pack.
325	*
326	* Ordering matches {@link runCompactJob}: the merged pack is written
327	* before the row flip (a crash between leaves an unreferenced,
328	* content-addressed, GC-able pack), and source packs are only deleted by
329	* the caller on a later run.
330	*/
331	export const runGeometricMergeJob = (options: {
332	readonly repoId: string;
333	readonly sql: SqlClient;
334	readonly blobs: BlobStoreShape;
335	readonly maxInputBytes?: number | undefined;
336	}): Effect.Effect<MergeOutcome, StoreError> =>
337	Effect.gen(function* () {
338	const maxInput = options.maxInputBytes ?? MAX_MERGE_INPUT_BYTES;
339
340	const rows = yield* options.sql.all<{ pack_id: string; n: number }>(
341	`SELECT pack_id, COUNT(*) AS n FROM objects
342	WHERE location = 'pack' AND pack_id IS NOT NULL
343	GROUP BY pack_id`,
344	);
345	if (rows.length < 2) return noMerge;
346
347	// Exact byte sizes from the store (offsets are absolute file offsets,
348	// so the merge math needs real lengths, not row sums).
349	const packs: Array<{ id: string; count: number; size: number }> = [];
350	for (const row of rows) {
351	const meta = yield* runR2(`merge head ${row.pack_id}`)(
352	options.blobs.head(packKeyOf(options.repoId, row.pack_id)),
353	);
354	// A pack R2 lost (or a crash orphaned) cannot be merged; skip it —
355	// reads through it will surface the real error on their own path.
356	if (meta !== null) {
357	packs.push({ id: row.pack_id, count: row.n, size: meta.size });
358	}
359	}
360	if (packs.length < 2) return noMerge;
361	packs.sort((a, b) => a.size - b.size);
362
363	// Geometric scan from the smallest: everything up to the last
364	// violation must merge.
365	let cumulative = 0;
366	let mergeCount = 0;
367	for (let i = 0; i < packs.length; i++) {
368	if (i >= 1 && packs[i]!.size < GEOMETRIC_FACTOR * cumulative) {
369	mergeCount = i + 1;
370	}
371	cumulative += packs[i]!.size;
372	}
373	if (mergeCount < 2) return noMerge;
374
375	// Cap the run's input; ≥2 sources or there is nothing useful to do.
376	let inputBytes = 0;
377	let take = 0;
378	while (take < mergeCount && inputBytes + packs[take]!.size <= maxInput) {
379	inputBytes += packs[take]!.size;
380	take++;
381	}
382	if (take < 2) return noMerge;
383	const sources = packs.slice(0, take);
384	const more = take < mergeCount;
385
386	// Read + validate each source, strip framing, record body deltas.
387	const bodies: Array<Uint8Array> = [];
388	const deltas: Array<{ id: string; delta: number }> = [];
389	// Promoted wire packs (DESIGN §22.5) also carry the push's trees,
390	// commits and delta entries, so they are REWRITTEN rather than
391	// concatenated: only the rows that reference them are copied, each as
392	// a regenerated typeSize header + its compressed span, and every row
393	// is repointed individually.
394	const rewrites: Array<{ oid: string; offset: number }> = [];
395	let totalCount = 0;
396	let bodyOffset = 0;
397	for (const source of sources) {
398	const object = yield* runR2(`merge get ${source.id}`)(
399	options.blobs.get(packKeyOf(options.repoId, source.id)),
400	);
401	if (object === null) return noMerge; // vanished mid-run: retry later
402	const bytes = yield* runR2(`merge read ${source.id}`)(object.bytes);
403	if (isWirePackId(source.id)) {
404	const rows = yield* options.sql.all<{
405	oid: string;
406	type: number;
407	size: number;
408	zsize: number;
409	pack_offset: number;
410	}>(
411	`SELECT oid, type, size, zsize, pack_offset FROM objects
412	WHERE location = 'pack' AND pack_id = ? ORDER BY pack_offset`,
413	source.id,
414	);
415	const pieces: Array<Uint8Array> = [];
416	let at = bodyOffset;
417	for (const row of rows) {
418	if (row.pack_offset + row.zsize > bytes.length) {
419	return yield* new StoreError({
420	reason: `merge: ${row.oid} points past the end of wire pack ${source.id}`,
421	});
422	}
423	const head = encodeTypeSize(row.type as PackEntryType, row.size);
424	pieces.push(
425	head,
426	bytes.subarray(row.pack_offset, row.pack_offset + row.zsize),
427	);
428	rewrites.push({ oid: row.oid, offset: 12 + at + head.length });
429	at += head.length + row.zsize;
430	}
431	const body = concatBytes(pieces);
432	totalCount += rows.length;
433	bodies.push(body);
434	bodyOffset += body.length;
435	continue;
436	}
437	if (
438	bytes.length < 32 ||
439	bytes[0] !== 0x50 ||
440	bytes[1] !== 0x41 ||
441	bytes[2] !== 0x43 ||
442	bytes[3] !== 0x4b
443	) {
444	return yield* new StoreError({
445	reason: `merge: pack ${source.id} is not a packfile`,
446	});
447	}
448	const view = new DataView(bytes.buffer, bytes.byteOffset);
449	totalCount += view.getUint32(8);
450	deltas.push({ id: source.id, delta: bodyOffset });
451	const body = bytes.subarray(12, bytes.length - 20);
452	bodies.push(body);
453	bodyOffset += body.length;
454	}
455	const header = packHeader(totalCount);
456	const trailer = yield* Effect.sync(() => {
457	const hash = makeSha1();
458	hash.update(header);
459	for (const body of bodies) hash.update(body);
460	return hash.digest();
461	});
462	const packId = yield* Effect.sync(() => bytesToHex(trailer));
463	const merged = yield* Effect.sync(() => {
464	const out = new Uint8Array(12 + bodyOffset + 20);
465	out.set(header, 0);
466	let at = 12;
467	for (const body of bodies) {
468	out.set(body, at);
469	at += body.length;
470	}
471	out.set(trailer, at);
472	return out;
473	});
474
475	const key = packKey(options.repoId, packId);
476	yield* runR2(`merge put ${key}`)(
477	options.blobs.put(key, merged, { contentLength: merged.length }),
478	);
479
480	// One additive UPDATE per source: new_offset = old_offset + delta
481	// (old header and new header are both 12 bytes, so they cancel).
482	let moved = 0;
483	yield* options.sql
484	.transactionSync((raw) => {
485	for (const { id, delta } of deltas) {
486	raw.exec(
487	`UPDATE objects
488	SET pack_id = ?, pack_offset = pack_offset + ?
489	WHERE location = 'pack' AND pack_id = ?`,
490	packId,
491	delta,
492	id,
493	);
494	}
495	for (const { oid, offset } of rewrites) {
496	raw.exec(
497	`UPDATE objects SET pack_id = ?, pack_offset = ? WHERE oid = ?`,
498	packId,
499	offset,
500	oid,
501	);
502	}
503	})
504	.pipe(Effect.asVoid);
505	moved = sources.reduce((sum, source) => sum + source.count, 0);
506
507	return {
508	packs: sources.length,
509	moved,
510	bytes: merged.length,
511	packId,
512	pendingDelete: sources.map((source) =>
513	packKeyOf(options.repoId, source.id),
514	),
515	more,
516	};
517	});