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 type { BucketSource } from "../server/config.ts";15import {16	hiddenPaths,17	hiddenToHand,18	isRegistered,19	keepState,20	recordHiddenHanded,21	recordSync,22	registerRepo,23	renamedTo,24} from "../server/db.ts";25import { RESERVED_NAMES } from "../server/reserved.ts";26import { dropTarballs } from "../tarball/cache.ts";27import { type Bucket, BucketWriteError, inTurn, KEPR_PREFIX, openBucket } from "./reader.ts";28import {29	BucketError,30	checkFormat,31	type EventRecord,32	escapePath,33	isSafePath,34	listEventIds,35	listReleaseVersions,36	listRepos,37	OversizedObject,38	objectSize,39	type ReleaseRecord,40	readEvent,41	readHead,42	readObject,43	readRelease,44	readReleasesLatest,45	readSave,46	type SaveRecord,47} from "./repo.ts";4849const INTERVAL_MS = 2 * 60 * 1000;50// Compressed bytes a sync holds before landing them; sized for a 2 GB host.51const LANDING_BYTES = 32 * 1024 * 1024;52const LANDING_SAVES = 100;53// Bucket reads in flight at once. Each waits about 170 ms on Tylee's round trip to the bucket54// (measured 2026-10-02), so reading one at a time left the line idle.55const READS_AT_ONCE = 16;56// Compressed object bytes being read at once; a larger object waits and is read alone.57const READ_BYTES = 32 * 1024 * 1024;5859export type SyncTarget = {60	bucket: Bucket;61	kepr: Database;62	mirrorDir: string;63	/** Called after each repo that synced without error. */64	synced?: (repo: string) => void;65};6667let running: Promise<void> | null = null;6869/** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */70export function syncAll(target: SyncTarget): Promise<void> {71	return running ?? holdSync(() => syncEach(target));72}7374/**75 * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a76 * rename moves a repo's keys, which a sync must never see half moved.77 */78export function holdSync(work: () => Promise<void>): Promise<void> {79	// Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh.80	const mine = (running ?? Promise.resolve()).then(work).finally(() => {81		if (running === slot) running = null;82	});83	const slot = mine.catch(() => {});84	running = slot;85	return mine;86}8788/**89 * What the app asks of the sync: one now, syncs held off while some work runs, a changed hidden90 * list applied, new bucket settings followed. Mirrors have one writer, the sync, so hidden lists go91 * through it too.92 */93export type Syncer = {94	syncAll(): Promise<void>;95	hold(work: () => Promise<void>): Promise<void>;96	applyHidden(repos: string[]): Promise<void>;97	setBucket(source: BucketSource): void;98};99100/** The sync in this thread, as the worker and the tests run it. */101export function localSyncer(target: SyncTarget): Syncer {102	return {103		syncAll: () => syncAll(target),104		hold: holdSync,105		applyHidden: async (repos) => {106			for (const asked of repos) {107				// A rename may have moved the repo while this waited on its hold.108				const repo = renamedTo(target.kepr, asked) ?? asked;109				if (!isRegistered(target.kepr, repo)) continue;110				applyHidden(openMirror(target.mirrorDir, repo), hiddenPaths(target.kepr, repo));111			}112		},113		setBucket: (source) => Object.assign(target.bucket, openBucket(source)),114	};115}116117/** A request to the sync's worker (`worker.ts`), and its reply. */118export type SyncRequest =119	| { kind: "start"; dataDir: string; source: BucketSource }120	| { kind: "sync" | "hold" | "release"; id: number }121	| { kind: "hidden"; id: number; repos: string[] }122	| { kind: "bucket"; source: BucketSource };123export type SyncReply =124	| { kind: "done"; id: number; error?: string }125	| { kind: "synced"; repo: string };126127/**128 * The sync on its own thread (2026-10-02), so a landing, a recount or a migration never stalls a129 * request. `synced` hears of each repo that synced cleanly. A worker that dies takes Kepr down with130 * it, for the service manager to restart, rather than leaving requests waiting on it.131 */132export function startSyncWorker(133	dataDir: string,134	source: BucketSource,135	synced: (repo: string) => void,136): Syncer {137	const worker = new Worker(new URL("./worker.ts", import.meta.url));138	// The server keeps Kepr running; the worker alone never should.139	worker.unref();140	const waiting = new Map<number, { resolve: () => void; reject: (error: Error) => void }>();141	let next = 0;142	worker.addEventListener("error", (event) => {143		console.error(`kepr: the sync worker failed: ${event.message}`);144		process.exit(1);145	});146	worker.addEventListener("message", ({ data }: MessageEvent<SyncReply>) => {147		if (data.kind === "synced") return synced(data.repo);148		const reply = waiting.get(data.id);149		waiting.delete(data.id);150		if (data.error === undefined) reply?.resolve();151		else reply?.reject(new Error(data.error));152	});153	const send = (request: SyncRequest) => worker.postMessage(request);154	const ask = (request: (id: number) => SyncRequest) =>155		new Promise<number>((resolve, reject) => {156			const id = next++;157			waiting.set(id, { resolve: () => resolve(id), reject });158			send(request(id));159		});160	send({ kind: "start", dataDir, source });161	return {162		syncAll: async () => {163			await ask((id) => ({ kind: "sync", id }));164		},165		async hold(work) {166			const id = await ask((id) => ({ kind: "hold", id }));167			try {168				await work();169			} finally {170				send({ kind: "release", id });171			}172		},173		applyHidden: async (repos) => {174			await ask((id) => ({ kind: "hidden", id, repos }));175		},176		setBucket: (source) => send({ kind: "bucket", source }),177	};178}179180export function startSyncLoop(target: SyncTarget): void {181	const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`));182	tick();183	setInterval(tick, INTERVAL_MS);184}185186async function syncEach(target: SyncTarget): Promise<void> {187	let repos: string[];188	try {189		repos = await listRepos(target.bucket);190	} catch (error) {191		keepState(target.kepr, "bucket_error", String(error));192		throw error;193	}194	// What the console's status shows, so it never asks the bucket itself.195	keepState(target.kepr, "bucket_error", null);196	keepState(target.kepr, "bucket_read_at", String(Math.floor(Date.now() / 1000)));197	keepState(target.kepr, "bucket_repos", JSON.stringify(repos));198	for (const repo of repos) {199		registerRepo(target.kepr, repo);200		let error: string | null = null;201		if (RESERVED_NAMES.has(repo)) {202			error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`;203		} else {204			try {205				const mirror = openMirror(target.mirrorDir, repo);206				const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo));207				if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo);208			} catch (cause) {209				error = cause instanceof Error ? cause.message : String(cause);210			}211		}212		recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error);213		if (error === null) target.synced?.(repo);214		await handHidden(target, repo);215	}216}217218/**219 * Hands a repo's hidden paths to Koh at `.kepr/hidden/<repo>`, one per line, for it to add to220 * `.kohignore`, unless they are known to be there already, then drops the tarballs kept under the221 * old list. The sync does it rather than the console, so a failed write or a restart is retried at222 * the next pass, and it never overlaps a rename. Without a write key the list stays in Kepr: Koh223 * reads it only from the bucket anyway.224 */225async function handHidden(target: SyncTarget, repo: string): Promise<void> {226	const list = hiddenToHand(target.kepr, repo);227	if (list === null) return;228	const paths = list.split("\n").filter(Boolean);229	const bytes = new TextEncoder().encode(paths.map((path) => `${path}\n`).join(""));230	// Unknown until the write is confirmed, so a restart or failure mid-write is written again231	// even if the list changes back to the one recorded before.232	recordHiddenHanded(target.kepr, repo, null);233	try {234		await target.bucket.put(`${KEPR_PREFIX}hidden/${repo}`, bytes, "text/plain");235	} catch (error) {236		if (!(error instanceof BucketWriteError)) {237			console.error(`kepr: handing ${repo}'s hidden paths to Koh failed: ${error}`);238		}239		return;240	}241	recordHiddenHanded(target.kepr, repo, list);242	// Only storage: a kept tarball is never served once its generation's hidden list has changed.243	await dropTarballs(target.bucket, repo).catch((error) =>244		console.error(`kepr: dropping ${repo}'s kept tarballs failed: ${error}`),245	);246}247248/**249 * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,250 * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a251 * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`252 * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates253 * past Koh's file limit, is quarantined: the save lands without it, as if stripped. `hidden` is254 * applied first, so a list changed since the last sync takes effect even with nothing new. Returns255 * the events it landed.256 */257export async function syncRepo(258	bucket: Bucket,259	mirror: Database,260	repo: string,261	hidden: string[] = [],262	landingBytes = LANDING_BYTES,263	readBytes = READ_BYTES,264): Promise<EventRecord[]> {265	applyHidden(mirror, hidden);266	await checkFormat(bucket, repo);267	const head = await readHead(bucket, repo);268	if (head === null) return [];269	const unknown = new Set<string>();270	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {271		if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);272		unknown.add(cursor);273		cursor = (await readSave(bucket, repo, cursor)).parent;274	}275	// Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES276	// records: a first sync of a big repo never holds its whole history in memory, and one cut short277	// resumes from the last batch. Records are read again here rather than kept from the walk, since278	// each lists every file of its tree.279	let saves: SaveRecord[] = [];280	let objects = new Map<string, StoredObject>();281	let quarantined: string[] = [];282	let fetched = 0;283	const oversized = new Set<string>();284	const landBatch = (batchHead: string) => {285		land(mirror, {286			head: batchHead,287			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",288			saves,289			objects,290			events: [],291			releases: [],292			quarantined,293		});294		saves = [];295		objects = new Map();296		quarantined = [];297		fetched = 0;298	};299	const gate = byteGate(readBytes);300	let failure: unknown = null;301	// Reads one object into `ahead`. A read the gate admits after the batch filled, or after another302	// read failed, puts its hash back for the next batch instead.303	const readInto = async (ahead: Map<string, StoredObject>, wanted: string[], hash: string) => {304		const size = await objectSize(bucket, repo, hash);305		await gate.enter(size);306		try {307			if (failure !== null || fetched >= landingBytes) {308				wanted.unshift(hash);309				return;310			}311			const object = await readObject(bucket, repo, hash);312			ahead.set(hash, object);313			fetched += object.compressed.length;314		} catch (cause) {315			if (!(cause instanceof OversizedObject)) throw cause;316			oversized.add(hash);317		} finally {318			gate.leave(size);319		}320	};321	const ids = [...unknown].reverse();322	for (let start = 0; start < ids.length; start += READS_AT_ONCE) {323		const records = await inTurn(ids.slice(start, start + READS_AT_ONCE), READS_AT_ONCE, (id) =>324			readSave(bucket, repo, id),325		);326		for (const save of records) {327			quarantined.push(...save.quarantined);328			// One save can hold more than a batch, as a first import does: its objects are stored329			// ahead of it, after the saves before it have landed, so no landing's cleanup sees them330			// unreferenced.331			const ahead = new Map<string, StoredObject>();332			const wanted = [...new Set(save.files.map((file) => file.hash))].filter(333				(hash) => !objects.has(hash) && !oversized.has(hash) && !knownObject(mirror, hash),334			);335			while (wanted.length > 0) {336				// No read starts once the batch is full, so a batch passes `landingBytes` by at most337				// the reads under way, which the gate holds to `readBytes` or one larger object.338				const reader = async () => {339					while (failure === null && fetched < landingBytes && wanted.length > 0) {340						await readInto(ahead, wanted, wanted.shift() as string).catch((cause) => {341							failure ??= cause;342						});343					}344				};345				// Every reader settles before a failure is reported, so none outlives this sync.346				await Promise.all(Array.from({ length: READS_AT_ONCE }, reader));347				if (failure !== null) throw failure;348				if (fetched >= landingBytes) {349					const last = saves.at(-1);350					if (last) landBatch(last.id);351					storeObjects(mirror, ahead);352					ahead.clear();353					fetched = 0;354				}355			}356			for (const [hash, object] of ahead) objects.set(hash, object);357			quarantined.push(358				...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),359			);360			save.files = save.files.filter((file) => !oversized.has(file.hash));361			saves.push(save);362			if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);363		}364	}365	const known = knownEventIds(mirror);366	const newEvents = (await listEventIds(bucket, repo)).filter((id) => !known.has(id));367	const events = (368		await inTurn(newEvents, READS_AT_ONCE, (id) => readEvent(bucket, repo, id))369	).filter(370		// Koh ignores it too: no save can hold an illegal path (§5.5).371		(event) => event.kind !== "strip" || isSafePath(event.path),372	);373	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";374	const moved =375		mirrorState(mirror, "head") !== head ||376		mirrorState(mirror, "releases_latest") !== releasesLatest;377	const releases = moved ? await newReleases(bucket, mirror, repo) : [];378	if (!moved && events.length === 0) return [];379	land(mirror, {380		head,381		releasesLatest,382		saves,383		objects,384		events,385		releases,386		quarantined,387	});388	return events;389}390391/** Admits reads while their bytes in flight stay within `limit`; a larger one waits to go alone. */392function byteGate(limit: number) {393	let inFlight = 0;394	let waiting: (() => void)[] = [];395	return {396		async enter(size: number): Promise<void> {397			while (inFlight > 0 && inFlight + size > limit) {398				await new Promise<void>((resolve) => waiting.push(resolve));399			}400			inFlight += size;401		},402		leave(size: number): void {403			inFlight -= size;404			const woken = waiting;405			waiting = [];406			for (const wake of woken) wake();407		},408	};409}410411async function newReleases(412	bucket: Bucket,413	mirror: Database,414	repo: string,415): Promise<ReleaseRecord[]> {416	const known = knownReleases(mirror);417	const versions = (await listReleaseVersions(bucket, repo)).filter((v) => !known.has(v));418	return inTurn(versions, READS_AT_ONCE, (version) => readRelease(bucket, repo, version));419}