import type { Database } from "bun:sqlite"; import { applyHidden, knownEventIds, knownObject, knownReleases, knownSave, land, mirrorState, openMirror, type StoredObject, storeObjects, } from "../mirror/store.ts"; import { hiddenPaths, hiddenToHand, recordHiddenHanded, recordSync, registerRepo, } from "../server/db.ts"; import { RESERVED_NAMES } from "../server/reserved.ts"; import { dropTarballs } from "../tarball/cache.ts"; import { type Bucket, BucketWriteError, KEPR_PREFIX } from "./reader.ts"; import { BucketError, checkFormat, type EventRecord, escapePath, isSafePath, listEventIds, listReleaseVersions, listRepos, OversizedObject, type ReleaseRecord, readEvent, readHead, readObject, readRelease, readReleasesLatest, readSave, type SaveRecord, } from "./repo.ts"; const INTERVAL_MS = 2 * 60 * 1000; // Compressed bytes a sync holds before landing them; sized for a 2 GB host. const LANDING_BYTES = 32 * 1024 * 1024; const LANDING_SAVES = 100; export type SyncTarget = { bucket: Bucket; kepr: Database; mirrorDir: string; /** Called after each repo that synced without error. */ synced?: (repo: string) => void; }; let running: Promise | null = null; /** Syncs every repo in the bucket; the admin console's "sync now" calls this too. Never overlaps itself. */ export function syncAll(target: SyncTarget): Promise { return running ?? holdSync(() => syncEach(target)); } /** * Runs `work` once any sync under way has finished, and keeps syncs waiting until it is done: a * rename moves a repo's keys, which a sync must never see half moved. */ export function holdSync(work: () => Promise): Promise { // Cleared before anyone awaiting `work` resumes, so a sync asked for next starts afresh. const mine = (running ?? Promise.resolve()).then(work).finally(() => { if (running === slot) running = null; }); const slot = mine.catch(() => {}); running = slot; return mine; } export function startSyncLoop(target: SyncTarget): void { const tick = () => syncAll(target).catch((error) => console.error(`kepr: sync failed: ${error}`)); tick(); setInterval(tick, INTERVAL_MS); } async function syncEach(target: SyncTarget): Promise { for (const repo of await listRepos(target.bucket)) { registerRepo(target.kepr, repo); let error: string | null = null; if (RESERVED_NAMES.has(repo)) { error = `not synced: the name "${repo}" clashes with Kepr's own /${repo} route; rename the repo`; } else { try { const mirror = openMirror(target.mirrorDir, repo); const events = await syncRepo(target.bucket, mirror, repo, hiddenPaths(target.kepr, repo)); if (events.some((event) => event.kind === "strip")) await dropTarballs(target.bucket, repo); } catch (cause) { error = cause instanceof Error ? cause.message : String(cause); } } recordSync(target.kepr, repo, Math.floor(Date.now() / 1000), error); if (error === null) target.synced?.(repo); await handHidden(target, repo); } } /** * Hands a repo's hidden paths to Koh at `.kepr/hidden/`, one per line, for it to add to * `.kohignore`, unless they are known to be there already, then drops the tarballs kept under the * old list. The sync does it rather than the console, so a failed write or a restart is retried at * the next pass, and it never overlaps a rename. Without a write key the list stays in Kepr: Koh * reads it only from the bucket anyway. */ async function handHidden(target: SyncTarget, repo: string): Promise { const list = hiddenToHand(target.kepr, repo); if (list === null) return; const paths = list.split("\n").filter(Boolean); const bytes = new TextEncoder().encode(paths.map((path) => `${path}\n`).join("")); // Unknown until the write is confirmed, so a restart or failure mid-write is written again // even if the list changes back to the one recorded before. recordHiddenHanded(target.kepr, repo, null); try { await target.bucket.put(`${KEPR_PREFIX}hidden/${repo}`, bytes, "text/plain"); } catch (error) { if (!(error instanceof BucketWriteError)) { console.error(`kepr: handing ${repo}'s hidden paths to Koh failed: ${error}`); } return; } recordHiddenHanded(target.kepr, repo, list); // Only storage: a kept tarball is never served once its generation's hidden list has changed. await dropTarballs(target.bucket, repo).catch((error) => console.error(`kepr: dropping ${repo}'s kept tarballs failed: ${error}`), ); } /** * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has, * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest` * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates * past Koh's file limit, is quarantined: the save lands without it, as if stripped. `hidden` is * applied first, so a list changed since the last sync takes effect even with nothing new. Returns * the events it landed. */ export async function syncRepo( bucket: Bucket, mirror: Database, repo: string, hidden: string[] = [], landingBytes = LANDING_BYTES, ): Promise { applyHidden(mirror, hidden); await checkFormat(bucket, repo); const head = await readHead(bucket, repo); if (head === null) return []; const unknown = new Set(); for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) { if (unknown.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`); unknown.add(cursor); cursor = (await readSave(bucket, repo, cursor)).parent; } // Oldest first, landing whenever the batch passes `landingBytes` of objects or LANDING_SAVES // records: a first sync of a big repo never holds its whole history in memory, and one cut short // resumes from the last batch. Records are read again here rather than kept from the walk, since // each lists every file of its tree. let saves: SaveRecord[] = []; let objects = new Map(); let quarantined: string[] = []; let fetched = 0; const oversized = new Set(); const landBatch = (batchHead: string) => { land(mirror, { head: batchHead, releasesLatest: mirrorState(mirror, "releases_latest") ?? "", saves, objects, events: [], releases: [], quarantined, }); saves = []; objects = new Map(); quarantined = []; fetched = 0; }; for (const id of [...unknown].reverse()) { const save = await readSave(bucket, repo, id); quarantined.push(...save.quarantined); // One save can hold more than a batch, as a first import does: its objects are stored ahead // of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced. const ahead = new Map(); for (const file of save.files) { const { hash } = file; if ( objects.has(hash) || ahead.has(hash) || oversized.has(hash) || knownObject(mirror, hash) ) { continue; } try { const { compressed, bytes } = await readObject(bucket, repo, hash); ahead.set(hash, { compressed, size: bytes.length }); fetched += compressed.length; } catch (cause) { if (!(cause instanceof OversizedObject)) throw cause; oversized.add(hash); } if (fetched >= landingBytes) { const last = saves.at(-1); if (last) landBatch(last.id); storeObjects(mirror, ahead); ahead.clear(); fetched = 0; } } for (const [hash, object] of ahead) objects.set(hash, object); quarantined.push( ...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)), ); save.files = save.files.filter((file) => !oversized.has(file.hash)); saves.push(save); if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id); } const known = knownEventIds(mirror); const events: EventRecord[] = []; for (const id of await listEventIds(bucket, repo)) { if (known.has(id)) continue; const event = await readEvent(bucket, repo, id); // Koh ignores it too: no save can hold an illegal path (§5.5). if (event.kind !== "strip" || isSafePath(event.path)) events.push(event); } const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? ""; const moved = mirrorState(mirror, "head") !== head || mirrorState(mirror, "releases_latest") !== releasesLatest; const releases = moved ? await newReleases(bucket, mirror, repo) : []; if (!moved && events.length === 0) return []; land(mirror, { head, releasesLatest, saves, objects, events, releases, quarantined, }); return events; } async function newReleases( bucket: Bucket, mirror: Database, repo: string, ): Promise { const known = knownReleases(mirror); const releases: ReleaseRecord[] = []; for (const version of await listReleaseVersions(bucket, repo)) { if (!known.has(version)) releases.push(await readRelease(bucket, repo, version)); } return releases; }