sync.ts

1import type { Database } from "bun:sqlite";2import {3	applyHidden,4	knownEventIds,5	knownObject,6	knownReleases,7	knownSave,8	land,9	mirrorState,10	openMirror,11	type StoredObject,12	storeObjects,13} from "../mirror/store.ts";14import {15	hiddenPaths,16	hiddenToHand,17	recordHiddenHanded,18	recordSync,19	registerRepo,20} from "../server/db.ts";21import { RESERVED_NAMES } from "../server/reserved.ts";22import { dropTarballs } from "../tarball/cache.ts";23import { type Bucket, BucketWriteError, KEPR_PREFIX } from "./reader.ts";24import {25	BucketError,26	checkFormat,27	type EventRecord,28	escapePath,29	isSafePath,30	listEventIds,31	listReleaseVersions,32	listRepos,33	OversizedObject,34	type ReleaseRecord,35	readEvent,36	readHead,37	readObject,38	readRelease,39	readReleasesLatest,40	readSave,41	type SaveRecord,42} from "./repo.ts";4344const INTERVAL_MS = 2 * 60 * 1000;45// Compressed bytes a sync holds before landing them; sized for a 2 GB host.46const LANDING_BYTES = 32 * 1024 * 1024;47const LANDING_SAVES = 100;4849export type SyncTarget = {50	bucket: Bucket;51	kepr: Database;52	mirrorDir: string;53	/** Called after each repo that synced without error. */54	synced?: (repo: string) => void;55};5657let running: Promise<void> | null = null;5859/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */60export function syncAll(target: SyncTarget): Promise<void> {61	return running ?? holdSync(() => syncEach(target));62}6364/**65 * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a66 * rename moves a repo's keys, which a sync must never see half moved.67 */68export function holdSync(work: () => Promise<void>): Promise<void> {69	// Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh.70	const mine = (running ?? Promise.resolve()).then(work).finally(() => {71		if (running === slot) running = null;72	});73	const slot = mine.catch(() => {});74	running = slot;75	return mine;76}7778export function startSyncLoop(target: SyncTarget): void {79	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));80	tick();81	setInterval(tick, INTERVAL_MS);82}8384async function syncEach(target: SyncTarget): Promise<void> {85	for (const repo of await listRepos(target.bucket)) {86		registerRepo(target.kepr, repo);87		let error: string | null = null;88		if (RESERVED_NAMES.has(repo)) {89			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;90		} else {91			try {92				const mirror = openMirror(target.mirrorDir, repo);93				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));94				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);95			} catch (cause) {96				error = cause instanceof Error ? cause.message : String(cause);97			}98		}99		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);100		if (error === null) target.synced?.(repo);101		await handHidden(target, repo);102	}103}104105/**106 * Hands a repo's hidden paths to Koh at `.kepr/hidden/<repo>`, one per line, for it to add to107 * `.kohignore`, unless they are known to be there already, then drops the tarballs kept under the108 * old list. The sync does it rather than the console, so a failed write or a restart is retried at109 * the next pass, and it never overlaps a rename. Without a write key the list stays in Kepr: Koh110 * reads it only from the bucket anyway.111 */112async function handHidden(target: SyncTarget, repo: string): Promise<void> {113	const list = hiddenToHand(target.kepr, repo);114	if (list === null) return;115	const paths = list.split("\n").filter(Boolean);116	const bytes = new TextEncoder().encode(paths.map((path) => `${path}\n`).join(""));117	// Unknown until the write is confirmed, so a restart or failure mid-write is written again118	// even if the list changes back to the one recorded before.119	recordHiddenHanded(target.kepr, repo, null);120	try {121		await target.bucket.put(`${KEPR_PREFIX}hidden/${repo}`, bytes, "text/plain");122	} catch (error) {123		if (!(error instanceof BucketWriteError)) {124			console.error(`kepr: handing ${repo}'s hidden paths to Koh failed: ${error}`);125		}126		return;127	}128	recordHiddenHanded(target.kepr, repo, list);129	// Only storage: a kept tarball is never served once its generation's hidden list has changed.130	await dropTarballs(target.bucket, repo).catch((error) =>131		console.error(`kepr: dropping ${repo}'s kept tarballs failed: ${error}`),132	);133}134135/**136 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,137 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a138 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`139 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates140 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. `hidden` is141 * applied first, so a list changed since the last sync takes effect even with nothing new. Returns142 * the events it landed.143 */144export async function syncRepo(145	bucket: Bucket,146	mirror: Database,147	repo: string,148	hidden: string[] = [],149	landingBytes = LANDING_BYTES,150): Promise<EventRecord[]> {151	applyHidden(mirror, hidden);152	await checkFormat(bucket, repo);153	const head = await readHead(bucket, repo);154	if (head === null) return [];155	const unknown = new Set<string>();156	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {157		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);158		unknown.add(cursor);159		cursor = (await readSave(bucket, repo, cursor)).parent;160	}161	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES162	// records: a first sync of a big repo never holds its whole history in memory, and one cut short163	// resumes from the last batch. Records are read again here rather than kept from the walk, since164	// each lists every file of its tree.165	let saves: SaveRecord[] = [];166	let objects = new Map<string, StoredObject>();167	let quarantined: string[] = [];168	let fetched = 0;169	const oversized = new Set<string>();170	const landBatch = (batchHead: string) => {171		land(mirror, {172			head: batchHead,173			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",174			saves,175			objects,176			events: [],177			releases: [],178			quarantined,179		});180		saves = [];181		objects = new Map();182		quarantined = [];183		fetched = 0;184	};185	for (const id of [...unknown].reverse()) {186		const save = await readSave(bucket, repo, id);187		quarantined.push(...save.quarantined);188		// One save can hold more than a batch, as a first import does: its objects are stored ahead189		// of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced.190		const ahead = new Map<string, StoredObject>();191		for (const file of save.files) {192			const { hash } = file;193			if (194				objects.has(hash) ||195				ahead.has(hash) ||196				oversized.has(hash) ||197				knownObject(mirror, hash)198			) {199				continue;200			}201			try {202				const { compressed, bytes } = await readObject(bucket, repo, hash);203				ahead.set(hash, { compressed, size: bytes.length });204				fetched += compressed.length;205			} catch (cause) {206				if (!(cause instanceof OversizedObject)) throw cause;207				oversized.add(hash);208			}209			if (fetched >= landingBytes) {210				const last = saves.at(-1);211				if (last) landBatch(last.id);212				storeObjects(mirror, ahead);213				ahead.clear();214				fetched = 0;215			}216		}217		for (const [hash, object] of ahead) objects.set(hash, object);218		quarantined.push(219			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),220		);221		save.files = save.files.filter((file) => !oversized.has(file.hash));222		saves.push(save);223		if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);224	}225	const known = knownEventIds(mirror);226	const events: EventRecord[] = [];227	for (const id of await listEventIds(bucket, repo)) {228		if (known.has(id)) continue;229		const event = await readEvent(bucket, repo, id);230		// Koh ignores it too: no save can hold an illegal path (§5.5).231		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);232	}233	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";234	const moved =235		mirrorState(mirror, "head") !== head ||236		mirrorState(mirror, "releases_latest") !== releasesLatest;237	const releases = moved ? await newReleases(bucket, mirror, repo) : [];238	if (!moved && events.length === 0) return [];239	land(mirror, {240		head,241		releasesLatest,242		saves,243		objects,244		events,245		releases,246		quarantined,247	});248	return events;249}250251async function newReleases(252	bucket: Bucket,253	mirror: Database,254	repo: string,255): Promise<ReleaseRecord[]> {256	const known = knownReleases(mirror);257	const releases: ReleaseRecord[] = [];258	for (const version of await listReleaseVersions(bucket, repo)) {259		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));260	}261	return releases;262}