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