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			// §22.3: after the strip has landed in the mirror, so no rebuild can bring the old bytes back.59			for (const event of events) {60				if (event.kind === "strip") dropTarballs(target.tarballDir, repo, event.save_id);61			}62		} catch (cause) {63			error = cause instanceof Error ? cause.message : String(cause);64		}65		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);66	}67}6869/**70 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,71 * fetch what is missing, and land it in one transaction. Events are listed every time, since a72 * flag or a strip arrives without moving `head`. Returns the events it landed.73 */74export async function syncRepo(75	bucket: BucketReader,76	mirror: Database,77	repo: string,78): Promise<EventRecord[]> {79	await checkFormat(bucket, repo);80	const head = await readHead(bucket, repo);81	if (head === null) return [];82	const saves: SaveRecord[] = [];83	const objects = new Map<string, StoredObject>();84	const seen = new Set<string>();85	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {86		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);87		seen.add(cursor);88		const save = await readSave(bucket, repo, cursor);89		for (const file of save.files) {90			if (objects.has(file.hash) || knownObject(mirror, file.hash)) continue;91			const { compressed, bytes } = await readObject(bucket, repo, file.hash);92			objects.set(file.hash, { compressed, size: bytes.length });93		}94		saves.push(save);95		cursor = save.parent;96	}97	const known = knownEventIds(mirror);98	const events: EventRecord[] = [];99	for (const id of await listEventIds(bucket, repo)) {100		if (!known.has(id)) events.push(await readEvent(bucket, repo, id));101	}102	if (saves.length === 0 && events.length === 0 && mirrorState(mirror, "head") === head) return [];103	land(mirror, { head, saves: saves.reverse(), objects, events });104	return events;105}