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} from "../mirror/store.ts";13import { hiddenPaths, recordSync, registerRepo } from "../server/db.ts";14import { RESERVED_NAMES } from "../server/reserved.ts";15import { dropTarballs } from "../tarball/cache.ts";16import type { Bucket } from "./reader.ts";17import {18	BucketError,19	checkFormat,20	type EventRecord,21	escapePath,22	isSafePath,23	listEventIds,24	listReleaseVersions,25	listRepos,26	OversizedObject,27	type ReleaseRecord,28	readEvent,29	readHead,30	readObject,31	readRelease,32	readReleasesLatest,33	readSave,34	type SaveRecord,35} from "./repo.ts";3637const INTERVAL_MS = 2 * 60 * 1000;38// Compressed bytes a sync holds before landing them; sized for a 2 GB host.39const LANDING_BYTES = 32 * 1024 * 1024;40const LANDING_SAVES = 100;4142export type SyncTarget = {43	bucket: Bucket;44	kepr: Database;45	mirrorDir: string;46};4748let running: Promise<void> | null = null;4950/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */51export function syncAll(target: SyncTarget): Promise<void> {52	running ??= syncEach(target).finally(() => {53		running = null;54	});55	return running;56}5758/** Rebuilds one repo's mirror from the bucket, once any sync under way has finished. */59export async function rebuildRepo(target: SyncTarget, repo: string): Promise<void> {60	await syncAll(target);61	clearMirror(openMirror(target.mirrorDir, repo));62	await syncAll(target);63}6465/**66 * Applies a changed list of names ignored in every repo. Once any sync under way has finished (its67 * later batches would land with the old list), a longer list is applied to each mirror in place; a68 * name taken off needs its files back, so every mirror is rebuilt.69 */70export async function reapplyHidden(target: SyncTarget, rebuild: boolean): Promise<void> {71	const repos = await listRepos(target.bucket);72	await syncAll(target);73	for (const repo of repos) {74		const mirror = openMirror(target.mirrorDir, repo);75		const head = mirrorState(mirror, "head");76		if (rebuild) clearMirror(mirror);77		else if (head) {78			land(mirror, {79				head,80				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",81				saves: [],82				objects: new Map(),83				events: [],84				releases: [],85				quarantined: [],86				hidden: hiddenPaths(target.kepr, repo),87			});88		}89	}90	if (rebuild) await syncAll(target);91}9293export function startSyncLoop(target: SyncTarget): void {94	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));95	tick();96	setInterval(tick, INTERVAL_MS);97}9899async function syncEach(target: SyncTarget): Promise<void> {100	for (const repo of await listRepos(target.bucket)) {101		registerRepo(target.kepr, repo);102		let error: string | null = null;103		if (RESERVED_NAMES.has(repo)) {104			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;105		} else {106			try {107				const mirror = openMirror(target.mirrorDir, repo);108				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));109				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);110			} catch (cause) {111				error = cause instanceof Error ? cause.message : String(cause);112			}113		}114		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);115	}116}117118/**119 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,120 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a121 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`122 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates123 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the124 * events it landed.125 */126export async function syncRepo(127	bucket: Bucket,128	mirror: Database,129	repo: string,130	hidden: string[] = [],131	landingBytes = LANDING_BYTES,132): Promise<EventRecord[]> {133	await checkFormat(bucket, repo);134	const head = await readHead(bucket, repo);135	if (head === null) return [];136	const unknown = new Set<string>();137	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {138		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);139		unknown.add(cursor);140		cursor = (await readSave(bucket, repo, cursor)).parent;141	}142	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES143	// records: a first sync of a big repo never holds its whole history in memory, and one cut short144	// resumes from the last batch. Records are read again here rather than kept from the walk, since145	// each lists every file of its tree.146	let saves: SaveRecord[] = [];147	let objects = new Map<string, StoredObject>();148	let quarantined: string[] = [];149	let fetched = 0;150	const oversized = new Set<string>();151	for (const id of [...unknown].reverse()) {152		const save = await readSave(bucket, repo, id);153		quarantined.push(...save.quarantined);154		for (const file of save.files) {155			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {156				continue;157			}158			try {159				const { compressed, bytes } = await readObject(bucket, repo, file.hash);160				objects.set(file.hash, { compressed, size: bytes.length });161				fetched += compressed.length;162			} catch (cause) {163				if (!(cause instanceof OversizedObject)) throw cause;164				oversized.add(file.hash);165			}166		}167		quarantined.push(168			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),169		);170		save.files = save.files.filter((file) => !oversized.has(file.hash));171		saves.push(save);172		if ((fetched >= landingBytes || saves.length >= LANDING_SAVES) && save.id !== head) {173			land(mirror, {174				head: save.id,175				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",176				saves,177				objects,178				events: [],179				releases: [],180				quarantined,181				hidden,182			});183			saves = [];184			objects = new Map();185			quarantined = [];186			fetched = 0;187		}188	}189	const known = knownEventIds(mirror);190	const events: EventRecord[] = [];191	for (const id of await listEventIds(bucket, repo)) {192		if (known.has(id)) continue;193		const event = await readEvent(bucket, repo, id);194		// Koh ignores it too: no save can hold an illegal path (§5.5).195		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);196	}197	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";198	const moved =199		mirrorState(mirror, "head") !== head ||200		mirrorState(mirror, "releases_latest") !== releasesLatest;201	const releases = moved ? await newReleases(bucket, mirror, repo) : [];202	if (!moved && events.length === 0) return [];203	land(mirror, {204		head,205		releasesLatest,206		saves,207		objects,208		events,209		releases,210		quarantined,211		hidden,212	});213	return events;214}215216async function newReleases(217	bucket: Bucket,218	mirror: Database,219	repo: string,220): Promise<ReleaseRecord[]> {221	const known = knownReleases(mirror);222	const releases: ReleaseRecord[] = [];223	for (const version of await listReleaseVersions(bucket, repo)) {224		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));225	}226	return releases;227}