sync.ts

1import type { Database } from "bun:sqlite";2import {3	knownEventIds,4	knownObject,5	knownSave,6	land,7	mirrorState,8	openMirror,9	type StoredObject,10} from "../mirror/store.ts";11import { recordSync, registerRepo } from "../server/db.ts";12import { dropTarballs } from "../tarball/cache.ts";13import type { BucketReader } from "./reader.ts";14import {15	BucketError,16	checkFormat,17	type EventRecord,18	listEventIds,19	listRepos,20	readEvent,21	readHead,22	readObject,23	readSave,24	type SaveRecord,25} from "./repo.ts";2627const INTERVAL_MS = 2 * 60 * 1000;2829export type SyncTarget = {30	bucket: BucketReader;31	kepr: Database;32	mirrorDir: string;33	tarballDir: string;34};3536let running: Promise<void> | null = null;3738/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */39export function syncAll(target: SyncTarget): Promise<void> {40	running ??= syncEach(target).finally(() => {41		running = null;42	});43	return running;44}4546export function startSyncLoop(target: SyncTarget): void {47	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));48	tick();49	setInterval(tick, INTERVAL_MS);50}5152async function syncEach(target: SyncTarget): Promise<void> {53	for (const repo of await listRepos(target.bucket)) {54		registerRepo(target.kepr, repo);55		let error: string | null = null;56		try {57			const events = await syncRepo(target.bucket, openMirror(target.mirrorDir, repo), repo);58			if (events.some((event) => event.kind === "strip")) dropTarballs(target.tarballDir, repo);59		} catch (cause) {60			error = cause instanceof Error ? cause.message : String(cause);61		}62		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);63	}64}6566/**67 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,68 * fetch what is missing, and land it in one transaction. Events are listed every time, since a69 * flag or a strip arrives without moving `head`. Returns the events it landed.70 */71export async function syncRepo(72	bucket: BucketReader,73	mirror: Database,74	repo: string,75): Promise<EventRecord[]> {76	await checkFormat(bucket, repo);77	const head = await readHead(bucket, repo);78	if (head === null) return [];79	const saves: SaveRecord[] = [];80	const objects = new Map<string, StoredObject>();81	const seen = new Set<string>();82	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {83		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);84		seen.add(cursor);85		const save = await readSave(bucket, repo, cursor);86		for (const file of save.files) {87			if (objects.has(file.hash) || knownObject(mirror, file.hash)) continue;88			const { compressed, bytes } = await readObject(bucket, repo, file.hash);89			objects.set(file.hash, { compressed, size: bytes.length });90		}91		saves.push(save);92		cursor = save.parent;93	}94	const known = knownEventIds(mirror);95	const events: EventRecord[] = [];96	for (const id of await listEventIds(bucket, repo)) {97		if (!known.has(id)) events.push(await readEvent(bucket, repo, id));98	}99	if (saves.length === 0 && events.length === 0 && mirrorState(mirror, "head") === head) return [];100	land(mirror, { head, saves: saves.reverse(), objects, events });101	return events;102}