sync.ts

1import type { Database } from "bun:sqlite";2import {3	clearMirror,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 { hiddenPaths, recordSync, registerRepo } from "../server/db.ts";15import { RESERVED_NAMES } from "../server/reserved.ts";16import { dropTarballs } from "../tarball/cache.ts";17import type { Bucket } from "./reader.ts";18import {19	BucketError,20	checkFormat,21	type EventRecord,22	escapePath,23	isSafePath,24	listEventIds,25	listReleaseVersions,26	listRepos,27	OversizedObject,28	type ReleaseRecord,29	readEvent,30	readHead,31	readObject,32	readRelease,33	readReleasesLatest,34	readSave,35	type SaveRecord,36} from "./repo.ts";3738const INTERVAL_MS = 2 * 60 * 1000;39// Compressed bytes a sync holds before landing them; sized for a 2 GB host.40const LANDING_BYTES = 32 * 1024 * 1024;41const LANDING_SAVES = 100;4243export type SyncTarget = {44	bucket: Bucket;45	kepr: Database;46	mirrorDir: string;47	/** Called after each repo that synced without error. */48	synced?: (repo: string) => void;49};5051let running: Promise<void> | null = null;5253/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */54export function syncAll(target: SyncTarget): Promise<void> {55	return running ?? holdSync(() => syncEach(target));56}5758/**59 * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a60 * rename moves a repo's keys, which a sync must never see half moved.61 */62export function holdSync(work: () => Promise<void>): Promise<void> {63	// Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh.64	const mine = (running ?? Promise.resolve()).then(work).finally(() => {65		if (running === slot) running = null;66	});67	const slot = mine.catch(() => {});68	running = slot;69	return mine;70}7172/** Rebuilds one repo's mirror from the bucket, once any sync under way has finished. */73export async function rebuildRepo(target: SyncTarget, repo: string): Promise<void> {74	await syncAll(target);75	clearMirror(openMirror(target.mirrorDir, repo));76	await syncAll(target);77}7879/**80 * Applies a changed list of names ignored in every repo. Once any sync under way has finished (its81 * later batches would land with the old list), a longer list is applied to each mirror in place; a82 * name taken off needs its files back, so every mirror is rebuilt.83 */84export async function reapplyHidden(target: SyncTarget, rebuild: boolean): Promise<void> {85	const repos = await listRepos(target.bucket);86	await syncAll(target);87	for (const repo of repos) {88		const mirror = openMirror(target.mirrorDir, repo);89		const head = mirrorState(mirror, "head");90		if (rebuild) clearMirror(mirror);91		else if (head) {92			land(mirror, {93				head,94				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",95				saves: [],96				objects: new Map(),97				events: [],98				releases: [],99				quarantined: [],100				hidden: hiddenPaths(target.kepr, repo),101			});102		}103	}104	if (rebuild) await syncAll(target);105}106107export function startSyncLoop(target: SyncTarget): void {108	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));109	tick();110	setInterval(tick, INTERVAL_MS);111}112113async function syncEach(target: SyncTarget): Promise<void> {114	for (const repo of await listRepos(target.bucket)) {115		registerRepo(target.kepr, repo);116		let error: string | null = null;117		if (RESERVED_NAMES.has(repo)) {118			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;119		} else {120			try {121				const mirror = openMirror(target.mirrorDir, repo);122				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));123				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);124			} catch (cause) {125				error = cause instanceof Error ? cause.message : String(cause);126			}127		}128		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);129		if (error === null) target.synced?.(repo);130	}131}132133/**134 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,135 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a136 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`137 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates138 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the139 * events it landed.140 */141export async function syncRepo(142	bucket: Bucket,143	mirror: Database,144	repo: string,145	hidden: string[] = [],146	landingBytes = LANDING_BYTES,147): Promise<EventRecord[]> {148	await checkFormat(bucket, repo);149	const head = await readHead(bucket, repo);150	if (head === null) return [];151	const unknown = new Set<string>();152	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {153		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);154		unknown.add(cursor);155		cursor = (await readSave(bucket, repo, cursor)).parent;156	}157	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES158	// records: a first sync of a big repo never holds its whole history in memory, and one cut short159	// resumes from the last batch. Records are read again here rather than kept from the walk, since160	// each lists every file of its tree.161	let saves: SaveRecord[] = [];162	let objects = new Map<string, StoredObject>();163	let quarantined: string[] = [];164	let fetched = 0;165	const oversized = new Set<string>();166	const landBatch = (batchHead: string) => {167		land(mirror, {168			head: batchHead,169			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",170			saves,171			objects,172			events: [],173			releases: [],174			quarantined,175			hidden,176		});177		saves = [];178		objects = new Map();179		quarantined = [];180		fetched = 0;181	};182	for (const id of [...unknown].reverse()) {183		const save = await readSave(bucket, repo, id);184		quarantined.push(...save.quarantined);185		// One save can hold more than a batch, as a first import does: its objects are stored ahead186		// of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced.187		const ahead = new Map<string, StoredObject>();188		for (const file of save.files) {189			const { hash } = file;190			if (191				objects.has(hash) ||192				ahead.has(hash) ||193				oversized.has(hash) ||194				knownObject(mirror, hash)195			) {196				continue;197			}198			try {199				const { compressed, bytes } = await readObject(bucket, repo, hash);200				ahead.set(hash, { compressed, size: bytes.length });201				fetched += compressed.length;202			} catch (cause) {203				if (!(cause instanceof OversizedObject)) throw cause;204				oversized.add(hash);205			}206			if (fetched >= landingBytes) {207				const last = saves.at(-1);208				if (last) landBatch(last.id);209				storeObjects(mirror, ahead);210				ahead.clear();211				fetched = 0;212			}213		}214		for (const [hash, object] of ahead) objects.set(hash, object);215		quarantined.push(216			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),217		);218		save.files = save.files.filter((file) => !oversized.has(file.hash));219		saves.push(save);220		if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);221	}222	const known = knownEventIds(mirror);223	const events: EventRecord[] = [];224	for (const id of await listEventIds(bucket, repo)) {225		if (known.has(id)) continue;226		const event = await readEvent(bucket, repo, id);227		// Koh ignores it too: no save can hold an illegal path (§5.5).228		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);229	}230	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";231	const moved =232		mirrorState(mirror, "head") !== head ||233		mirrorState(mirror, "releases_latest") !== releasesLatest;234	const releases = moved ? await newReleases(bucket, mirror, repo) : [];235	if (!moved && events.length === 0) return [];236	land(mirror, {237		head,238		releasesLatest,239		saves,240		objects,241		events,242		releases,243		quarantined,244		hidden,245	});246	return events;247}248249async function newReleases(250	bucket: Bucket,251	mirror: Database,252	repo: string,253): Promise<ReleaseRecord[]> {254	const known = knownReleases(mirror);255	const releases: ReleaseRecord[] = [];256	for (const version of await listReleaseVersions(bucket, repo)) {257		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));258	}259	return releases;260}