sync.ts

1import type { Database } from "bun:sqlite";2import {3	applyHidden,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 {15	hiddenPaths,16	hiddenToHand,17	keepState,18	recordHiddenHanded,19	recordSync,20	registerRepo,21} from "../server/db.ts";22import { RESERVED_NAMES } from "../server/reserved.ts";23import { dropTarballs } from "../tarball/cache.ts";24import { type Bucket, BucketWriteError, inTurn, KEPR_PREFIX } from "./reader.ts";25import {26	BucketError,27	checkFormat,28	type EventRecord,29	escapePath,30	isSafePath,31	listEventIds,32	listReleaseVersions,33	listRepos,34	OversizedObject,35	objectSize,36	type ReleaseRecord,37	readEvent,38	readHead,39	readObject,40	readRelease,41	readReleasesLatest,42	readSave,43	type SaveRecord,44} from "./repo.ts";4546const INTERVAL_MS = 2 * 60 * 1000;47// Compressed bytes a sync holds before landing them; sized for a 2 GB host.48const LANDING_BYTES = 32 * 1024 * 1024;49const LANDING_SAVES = 100;50// Bucket reads in flight at once. Each waits about 170 ms on Tylee's round trip to the bucket51// (measured 2026-10-02), so reading one at a time left the line idle.52const READS_AT_ONCE = 16;53// Compressed object bytes being read at once; a larger object waits and is read alone.54const READ_BYTES = 32 * 1024 * 1024;5556export type SyncTarget = {57	bucket: Bucket;58	kepr: Database;59	mirrorDir: string;60	/** Called after each repo that synced without error. */61	synced?: (repo: string) => void;62};6364let running: Promise<void> | null = null;6566/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */67export function syncAll(target: SyncTarget): Promise<void> {68	return running ?? holdSync(() => syncEach(target));69}7071/**72 * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a73 * rename moves a repo's keys, which a sync must never see half moved.74 */75export function holdSync(work: () => Promise<void>): Promise<void> {76	// Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh.77	const mine = (running ?? Promise.resolve()).then(work).finally(() => {78		if (running === slot) running = null;79	});80	const slot = mine.catch(() => {});81	running = slot;82	return mine;83}8485export function startSyncLoop(target: SyncTarget): void {86	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));87	tick();88	setInterval(tick, INTERVAL_MS);89}9091async function syncEach(target: SyncTarget): Promise<void> {92	let repos: string[];93	try {94		repos = await listRepos(target.bucket);95	} catch (error) {96		keepState(target.kepr, "bucket_error", String(error));97		throw error;98	}99	// What the console's status shows, so it never asks the bucket itself.100	keepState(target.kepr, "bucket_error", null);101	keepState(target.kepr, "bucket_read_at", String(Math.floor(Date.now() / 1000)));102	keepState(target.kepr, "bucket_repos", JSON.stringify(repos));103	for (const repo of repos) {104		registerRepo(target.kepr, repo);105		let error: string | null = null;106		if (RESERVED_NAMES.has(repo)) {107			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;108		} else {109			try {110				const mirror = openMirror(target.mirrorDir, repo);111				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));112				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);113			} catch (cause) {114				error = cause instanceof Error ? cause.message : String(cause);115			}116		}117		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);118		if (error === null) target.synced?.(repo);119		await handHidden(target, repo);120	}121}122123/**124 * Hands a repo's hidden paths to Koh at `.kepr/hidden/<repo>`, one per line, for it to add to125 * `.kohignore`, unless they are known to be there already, then drops the tarballs kept under the126 * old list. The sync does it rather than the console, so a failed write or a restart is retried at127 * the next pass, and it never overlaps a rename. Without a write key the list stays in Kepr: Koh128 * reads it only from the bucket anyway.129 */130async function handHidden(target: SyncTarget, repo: string): Promise<void> {131	const list = hiddenToHand(target.kepr, repo);132	if (list === null) return;133	const paths = list.split("\n").filter(Boolean);134	const bytes = new TextEncoder().encode(paths.map((path) => `${path}\n`).join(""));135	// Unknown until the write is confirmed, so a restart or failure mid-write is written again136	// even if the list changes back to the one recorded before.137	recordHiddenHanded(target.kepr, repo, null);138	try {139		await target.bucket.put(`${KEPR_PREFIX}hidden/${repo}`, bytes, "text/plain");140	} catch (error) {141		if (!(error instanceof BucketWriteError)) {142			console.error(`kepr: handing ${repo}'s hidden paths to Koh failed: ${error}`);143		}144		return;145	}146	recordHiddenHanded(target.kepr, repo, list);147	// Only storage: a kept tarball is never served once its generation's hidden list has changed.148	await dropTarballs(target.bucket, repo).catch((error) =>149		console.error(`kepr: dropping ${repo}'s kept tarballs failed: ${error}`),150	);151}152153/**154 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,155 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a156 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`157 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates158 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. `hidden` is159 * applied first, so a list changed since the last sync takes effect even with nothing new. Returns160 * the events it landed.161 */162export async function syncRepo(163	bucket: Bucket,164	mirror: Database,165	repo: string,166	hidden: string[] = [],167	landingBytes = LANDING_BYTES,168	readBytes = READ_BYTES,169): Promise<EventRecord[]> {170	applyHidden(mirror, hidden);171	await checkFormat(bucket, repo);172	const head = await readHead(bucket, repo);173	if (head === null) return [];174	const unknown = new Set<string>();175	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {176		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);177		unknown.add(cursor);178		cursor = (await readSave(bucket, repo, cursor)).parent;179	}180	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES181	// records: a first sync of a big repo never holds its whole history in memory, and one cut short182	// resumes from the last batch. Records are read again here rather than kept from the walk, since183	// each lists every file of its tree.184	let saves: SaveRecord[] = [];185	let objects = new Map<string, StoredObject>();186	let quarantined: string[] = [];187	let fetched = 0;188	const oversized = new Set<string>();189	const landBatch = (batchHead: string) => {190		land(mirror, {191			head: batchHead,192			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",193			saves,194			objects,195			events: [],196			releases: [],197			quarantined,198		});199		saves = [];200		objects = new Map();201		quarantined = [];202		fetched = 0;203	};204	const gate = byteGate(readBytes);205	let failure: unknown = null;206	// Reads one object into `ahead`. A read the gate admits after the batch filled, or after another207	// read failed, puts its hash back for the next batch instead.208	const readInto = async (ahead: Map<string, StoredObject>, wanted: string[], hash: string) => {209		const size = await objectSize(bucket, repo, hash);210		await gate.enter(size);211		try {212			if (failure !== null || fetched >= landingBytes) {213				wanted.unshift(hash);214				return;215			}216			const object = await readObject(bucket, repo, hash);217			ahead.set(hash, object);218			fetched += object.compressed.length;219		} catch (cause) {220			if (!(cause instanceof OversizedObject)) throw cause;221			oversized.add(hash);222		} finally {223			gate.leave(size);224		}225	};226	const ids = [...unknown].reverse();227	for (let start = 0; start < ids.length; start += READS_AT_ONCE) {228		const records = await inTurn(ids.slice(start, start + READS_AT_ONCE), READS_AT_ONCE, (id) =>229			readSave(bucket, repo, id),230		);231		for (const save of records) {232			quarantined.push(...save.quarantined);233			// One save can hold more than a batch, as a first import does: its objects are stored234			// ahead of it, after the saves before it have landed, so no landing's cleanup sees them235			// unreferenced.236			const ahead = new Map<string, StoredObject>();237			const wanted = [...new Set(save.files.map((file) => file.hash))].filter(238				(hash) => !objects.has(hash) && !oversized.has(hash) && !knownObject(mirror, hash),239			);240			while (wanted.length > 0) {241				// No read starts once the batch is full, so a batch passes `landingBytes` by at most242				// the reads under way, which the gate holds to `readBytes` or one larger object.243				const reader = async () => {244					while (failure === null && fetched < landingBytes && wanted.length > 0) {245						await readInto(ahead, wanted, wanted.shift() as string).catch((cause) => {246							failure ??= cause;247						});248					}249				};250				// Every reader settles before a failure is reported, so none outlives this sync.251				await Promise.all(Array.from({ length: READS_AT_ONCE }, reader));252				if (failure !== null) throw failure;253				if (fetched >= landingBytes) {254					const last = saves.at(-1);255					if (last) landBatch(last.id);256					storeObjects(mirror, ahead);257					ahead.clear();258					fetched = 0;259				}260			}261			for (const [hash, object] of ahead) objects.set(hash, object);262			quarantined.push(263				...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),264			);265			save.files = save.files.filter((file) => !oversized.has(file.hash));266			saves.push(save);267			if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);268		}269	}270	const known = knownEventIds(mirror);271	const newEvents = (await listEventIds(bucket, repo)).filter((id) => !known.has(id));272	const events = (273		await inTurn(newEvents, READS_AT_ONCE, (id) => readEvent(bucket, repo, id))274	).filter(275		// Koh ignores it too: no save can hold an illegal path (§5.5).276		(event) => event.kind !== "strip" || isSafePath(event.path),277	);278	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";279	const moved =280		mirrorState(mirror, "head") !== head ||281		mirrorState(mirror, "releases_latest") !== releasesLatest;282	const releases = moved ? await newReleases(bucket, mirror, repo) : [];283	if (!moved && events.length === 0) return [];284	land(mirror, {285		head,286		releasesLatest,287		saves,288		objects,289		events,290		releases,291		quarantined,292	});293	return events;294}295296/** Admits reads while their bytes in flight stay within `limit`; a larger one waits to go alone. */297function byteGate(limit: number) {298	let inFlight = 0;299	let waiting: (() => void)[] = [];300	return {301		async enter(size: number): Promise<void> {302			while (inFlight > 0 && inFlight + size > limit) {303				await new Promise<void>((resolve) => waiting.push(resolve));304			}305			inFlight += size;306		},307		leave(size: number): void {308			inFlight -= size;309			const woken = waiting;310			waiting = [];311			for (const wake of woken) wake();312		},313	};314}315316async function newReleases(317	bucket: Bucket,318	mirror: Database,319	repo: string,320): Promise<ReleaseRecord[]> {321	const known = knownReleases(mirror);322	const versions = (await listReleaseVersions(bucket, repo)).filter((v) => !known.has(v));323	return inTurn(versions, READS_AT_ONCE, (version) => readRelease(bucket, repo, version));324}