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};4849let running: Promise<void> | null = null;5051/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */52export function syncAll(target: SyncTarget): Promise<void> {53	running ??= syncEach(target).finally(() => {54		running = null;55	});56	return running;57}5859/** Rebuilds one repo's mirror from the bucket, once any sync under way has finished. */60export async function rebuildRepo(target: SyncTarget, repo: string): Promise<void> {61	await syncAll(target);62	clearMirror(openMirror(target.mirrorDir, repo));63	await syncAll(target);64}6566/**67 * Applies a changed list of names ignored in every repo. Once any sync under way has finished (its68 * later batches would land with the old list), a longer list is applied to each mirror in place; a69 * name taken off needs its files back, so every mirror is rebuilt.70 */71export async function reapplyHidden(target: SyncTarget, rebuild: boolean): Promise<void> {72	const repos = await listRepos(target.bucket);73	await syncAll(target);74	for (const repo of repos) {75		const mirror = openMirror(target.mirrorDir, repo);76		const head = mirrorState(mirror, "head");77		if (rebuild) clearMirror(mirror);78		else if (head) {79			land(mirror, {80				head,81				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",82				saves: [],83				objects: new Map(),84				events: [],85				releases: [],86				quarantined: [],87				hidden: hiddenPaths(target.kepr, repo),88			});89		}90	}91	if (rebuild) await syncAll(target);92}9394export function startSyncLoop(target: SyncTarget): void {95	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));96	tick();97	setInterval(tick, INTERVAL_MS);98}99100async function syncEach(target: SyncTarget): Promise<void> {101	for (const repo of await listRepos(target.bucket)) {102		registerRepo(target.kepr, repo);103		let error: string | null = null;104		if (RESERVED_NAMES.has(repo)) {105			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;106		} else {107			try {108				const mirror = openMirror(target.mirrorDir, repo);109				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));110				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);111			} catch (cause) {112				error = cause instanceof Error ? cause.message : String(cause);113			}114		}115		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);116	}117}118119/**120 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,121 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a122 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`123 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates124 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the125 * events it landed.126 */127export async function syncRepo(128	bucket: Bucket,129	mirror: Database,130	repo: string,131	hidden: string[] = [],132	landingBytes = LANDING_BYTES,133): Promise<EventRecord[]> {134	await checkFormat(bucket, repo);135	const head = await readHead(bucket, repo);136	if (head === null) return [];137	const unknown = new Set<string>();138	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {139		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);140		unknown.add(cursor);141		cursor = (await readSave(bucket, repo, cursor)).parent;142	}143	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES144	// records: a first sync of a big repo never holds its whole history in memory, and one cut short145	// resumes from the last batch. Records are read again here rather than kept from the walk, since146	// each lists every file of its tree.147	let saves: SaveRecord[] = [];148	let objects = new Map<string, StoredObject>();149	let quarantined: string[] = [];150	let fetched = 0;151	const oversized = new Set<string>();152	const landBatch = (batchHead: string) => {153		land(mirror, {154			head: batchHead,155			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",156			saves,157			objects,158			events: [],159			releases: [],160			quarantined,161			hidden,162		});163		saves = [];164		objects = new Map();165		quarantined = [];166		fetched = 0;167	};168	for (const id of [...unknown].reverse()) {169		const save = await readSave(bucket, repo, id);170		quarantined.push(...save.quarantined);171		// One save can hold more than a batch, as a first import does: its objects are stored ahead172		// of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced.173		const ahead = new Map<string, StoredObject>();174		for (const file of save.files) {175			const { hash } = file;176			if (177				objects.has(hash) ||178				ahead.has(hash) ||179				oversized.has(hash) ||180				knownObject(mirror, hash)181			) {182				continue;183			}184			try {185				const { compressed, bytes } = await readObject(bucket, repo, hash);186				ahead.set(hash, { compressed, size: bytes.length });187				fetched += compressed.length;188			} catch (cause) {189				if (!(cause instanceof OversizedObject)) throw cause;190				oversized.add(hash);191			}192			if (fetched >= landingBytes) {193				const last = saves.at(-1);194				if (last) landBatch(last.id);195				storeObjects(mirror, ahead);196				ahead.clear();197				fetched = 0;198			}199		}200		for (const [hash, object] of ahead) objects.set(hash, object);201		quarantined.push(202			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),203		);204		save.files = save.files.filter((file) => !oversized.has(file.hash));205		saves.push(save);206		if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);207	}208	const known = knownEventIds(mirror);209	const events: EventRecord[] = [];210	for (const id of await listEventIds(bucket, repo)) {211		if (known.has(id)) continue;212		const event = await readEvent(bucket, repo, id);213		// Koh ignores it too: no save can hold an illegal path (§5.5).214		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);215	}216	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";217	const moved =218		mirrorState(mirror, "head") !== head ||219		mirrorState(mirror, "releases_latest") !== releasesLatest;220	const releases = moved ? await newReleases(bucket, mirror, repo) : [];221	if (!moved && events.length === 0) return [];222	land(mirror, {223		head,224		releasesLatest,225		saves,226		objects,227		events,228		releases,229		quarantined,230		hidden,231	});232	return events;233}234235async function newReleases(236	bucket: Bucket,237	mirror: Database,238	repo: string,239): Promise<ReleaseRecord[]> {240	const known = knownReleases(mirror);241	const releases: ReleaseRecord[] = [];242	for (const version of await listReleaseVersions(bucket, repo)) {243		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));244	}245	return releases;246}