stop page views waiting on the bucket: list repos with one delimiter listing, read the sync's objects 16 at a time within a byte gate, show the sync's record on /admin and an hourly measurement on /admin/storage; bucket budget 1,010, ceiling 12,710 (v3.0.194)

@@ -1,6 +1,6 @@11 {22 	"name": "kepr",3-	"version": "3.0.193",3+	"version": "3.0.194",44 	"private": true,55 	"type": "module",66 	"scripts": {

@@ -1,10 +1,10 @@11 // §26: fails when any area passes its budget or the TypeScript total passes the2-// 12,600-line ceiling.3-const CEILING = 12_600;2+// 12,710-line ceiling.3+const CEILING = 12_710;44 55 // Prefixes are matched in order, so the specific files come before the directories.66 const TS_AREAS: [string, string[], number][] = [7-	["bucket", ["src/bucket/"], 900],7+	["bucket", ["src/bucket/"], 1010],88 	["mirror", ["src/mirror/"], 1100],99 	["describe", ["src/describe/"], 500],1010 	["pages", ["src/pages/"], 2600],

@@ -1,5 +1,4 @@11 import type { Context, Hono } from "hono";2-import { checkFormat, listRepos } from "../bucket/repo.ts";32 import type { SyncTarget } from "../bucket/sync.ts";43 import { operatorIssues } from "../issues/store.ts";54 import { type Html, html } from "../pages/html.ts";@@ -132,28 +131,36 @@132131 <p class="muted">To revoke access, rotate the write key in Scaleway and update it under <a href="/admin/bucket">bucket</a>.</p>`;133132 }134133 134+/**135+ * The bucket as the sync last read it, and each repo's format and sync result: the sync checks the136+ * format of every repo on every pass, so the console never waits on the bucket itself.137+ */138+function bucketStatus(ctx: AdminContext): Html {139+	const error = setting(ctx.kepr, "bucket_error");140+	if (error) {141+		return html`<p class="issues-error" role="alert">Bucket unreachable at the last sync: ${error}</p>`;142+	}143+	const readAt = Number(setting(ctx.kepr, "bucket_read_at") ?? 0);144+	if (readAt === 0) return html`<p class="muted">The sync has not read the bucket yet.</p>`;145+	// The repos that listing found; Kepr keeps the rows of repos gone from the bucket.146+	const seen = JSON.parse(setting(ctx.kepr, "bucket_repos") ?? "[]") as string[];147+	const errors = new Map(allRepos(ctx.kepr).map((repo) => [repo.name, repo.last_sync_error]));148+	return html`<p>Bucket read by the sync ${relativeTime(readAt)}: ${plural(seen.length, "repo")}.</p>149+<ul>${seen.map((name) => {150+		const failure = errors.get(name);151+		return html`<li><span class="mono">${name}</span> ${152+			failure153+				? html`<span class="issues-error">${failure}</span>`154+				: html`<span class="muted">format ok</span>`155+		}</li>`;156+	})}</ul>`;157+}158+135159 /** §24's Status section at `/admin`: the bucket, each repo's format, backups, the running version. */136160 export function registerStatus(app: Hono, ctx: AdminContext): void {137161 	app.get("/admin", async (c) => {138162 		if (c.req.query("section") === "backups") return c.html(await backupsPage(c, ctx));139-		let bucket: Html;140-		try {141-			const repos = await listRepos(ctx.sync.bucket);142-			const formats = await Promise.all(143-				repos.map(async (repo) => {144-					try {145-						await checkFormat(ctx.sync.bucket, repo);146-						return html`<li><span class="mono">${repo}</span> <span class="muted">format ok</span></li>`;147-					} catch (error) {148-						return html`<li><span class="mono">${repo}</span>149-<span class="issues-error">${String(error)}</span></li>`;150-					}151-				}),152-			);153-			bucket = html`<p>Bucket reachable: ${plural(repos.length, "repo")}.</p><ul>${formats}</ul>`;154-		} catch (error) {155-			bucket = html`<p class="issues-error" role="alert">Bucket unreachable: ${String(error)}</p>`;156-		}163+		const bucket = bucketStatus(ctx);157164 		const { version } = (await Bun.file(`${import.meta.dir}/../../package.json`).json()) as {158165 			version: string;159166 		};

@@ -1,6 +1,6 @@11 import { existsSync, renameSync } from "node:fs";22 import { join } from "node:path";3-import { type Bucket, KEPR_PREFIX } from "../bucket/reader.ts";3+import { type Bucket, inTurn, KEPR_PREFIX } from "../bucket/reader.ts";44 import { holdSync, type SyncTarget } from "../bucket/sync.ts";55 import { closeMirror } from "../mirror/store.ts";66 import { RESERVED_NAMES } from "../server/reserved.ts";@@ -63,7 +63,7 @@6363 		const keys = (await bucket.list(`${from}/`)).filter(6464 			(key) => key !== `${from}/head` && !(IMMUTABLE.test(key) && copied.has(key)),6565 		);66-		await inTurn(keys, (key) => bucket.copy(key, `${to}/${key.slice(from.length + 1)}`));66+		await inTurn(keys, AT_ONCE, (key) => bucket.copy(key, `${to}/${key.slice(from.length + 1)}`));6767 		for (const key of keys) copied.add(key);6868 		await bucket.copy(`${from}/head`, `${to}/head`);6969 		head = now;@@ -85,16 +85,10 @@8585 		const path = join(mirrorDir, `${from}.db${suffix}`);8686 		if (existsSync(path)) renameSync(path, join(mirrorDir, `${to}.db${suffix}`));8787 	}88-	await inTurn(await bucket.list(`${from}/`), (key) => bucket.removeRenamed(key));88+	await inTurn(await bucket.list(`${from}/`), AT_ONCE, (key) => bucket.removeRenamed(key));8989 }9090 9191 async function text(bucket: Bucket, key: string): Promise<string | null> {9292 	const bytes = await bucket.get(key);9393 	return bytes && new TextDecoder().decode(bytes).trim();94-}95-96-async function inTurn(keys: string[], each: (key: string) => Promise<void>): Promise<void> {97-	for (let i = 0; i < keys.length; i += AT_ONCE) {98-		await Promise.all(keys.slice(i, i + AT_ONCE).map(each));99-	}10094 }

@@ -1,48 +1,85 @@11 import type { Context, Hono } from "hono";22 import { BucketWriteError, KEPR_PREFIX } from "../bucket/reader.ts";33 import { listRepos } from "../bucket/repo.ts";4+import { holdSync, type SyncTarget } from "../bucket/sync.ts";45 import { tarballGeneration } from "../mirror/queries.ts";56 import { openMirror } from "../mirror/store.ts";67 import { html } from "../pages/html.ts";7-import { formatSize, plural } from "../pages/layout.ts";8+import { formatSize, plural, relativeTime } from "../pages/layout.ts";9+import { keepState, setting } from "../server/db.ts";810 import { type AdminContext, adminPage, formField, notice, textFields } from "./console.ts";11+import { allRepos } from "./store.ts";912 1013 const TARBALL = /^\.kepr\/tarballs\/([^/]+)\/([^/]+)\/([0-9a-f]{6}|[0-9a-f]{12})\.tar\.gz$/;1114 const HIDDEN = /^\.kepr\/hidden\/([^/]+)$/;1215 1316 type RepoUse = { history: number; builds: number; tarballs: number };1417 type Tarball = { key: string; repo: string; save: string; size: number };18+type Survey = {19+	/** When it was measured, in Unix seconds. */20+	at: number;21+	use: [repo: string, use: RepoUse][];22+	kept: Tarball[];23+	orphans: [key: string, size: number][];24+	kepr: number;25+	total: number;26+};1527 28+// A full listing of the bucket takes seconds (3.2 s for 22k keys on Tylee, 2026-10-02), so it is29+// measured in the background and the page shows the last measurement.30+const MEASURE_MS = 60 * 60 * 1000;31+32+// Measuring and deleting take turns, so neither stores a survey the other has made stale.33+let turn: Promise<unknown> = Promise.resolve();34+function inStorageTurn<T>(work: () => Promise<T>): Promise<T> {35+	const mine = turn.then(work);36+	turn = mine.catch(() => {});37+	return mine;38+}39+40+export function startStorageLoop(target: SyncTarget): void {41+	const tick = () =>42+		measureStorage(target).catch((error) =>43+			console.error(`kepr: measuring storage failed: ${error}`),44+		);45+	tick();46+	setInterval(tick, MEASURE_MS);47+}48+1649 /**17- * The bucket sorted into what §24's Storage section shows. Orphans are only Kepr's own files that18- * nothing reads again; Koh's never are, since a build's files land before its record (§11.2).50+ * Sorts the bucket into what §24's Storage section shows, and keeps it for the page. Orphans are51+ * only Kepr's own files that nothing reads again; Koh's never are, since a build's files land52+ * before its record (§11.2).1953  */20-async function survey(ctx: AdminContext) {21-	const sizes = await ctx.sync.bucket.sizes("");22-	const repos = await listRepos(ctx.sync.bucket);23-	const generations = new Map(24-		repos.map((repo) => [repo, tarballGeneration(openMirror(ctx.mirrorDir, repo))]),25-	);54+export function measureStorage(target: SyncTarget): Promise<void> {55+	return inStorageTurn(() => measure(target));56+}57+58+async function measure(target: SyncTarget): Promise<void> {59+	const sizes = await target.bucket.sizes("");60+	const repos = await listRepos(target.bucket);61+	const isOrphan = orphanTest(target, repos);2662 	const use = new Map<string, RepoUse>(2763 		repos.map((repo) => [repo, { history: 0, builds: 0, tarballs: 0 }]),2864 	);29-	const kept: Tarball[] = [];30-	const orphans: string[] = [];31-	let kepr = 0;32-	let total = 0;65+	const survey: Survey = {66+		at: Math.floor(Date.now() / 1000),67+		use: [],68+		kept: [],69+		orphans: [],70+		kepr: 0,71+		total: 0,72+	};3373 	for (const [key, size] of sizes) {34-		total += size;74+		survey.total += size;3575 		const tarball = TARBALL.exec(key);36-		const hidden = HIDDEN.exec(key);37-		if (tarball) {38-			const [, repo = "", generation, save = ""] = tarball;39-			const owner = use.get(repo);40-			if (owner && generations.get(repo) === generation) {41-				owner.tarballs += size;42-				kept.push({ key, repo, save, size });43-			} else orphans.push(key);44-		} else if (hidden && !use.has(hidden[1] ?? "")) orphans.push(key);45-		else if (key.startsWith(KEPR_PREFIX)) kepr += size;76+		if (isOrphan(key)) survey.orphans.push([key, size]);77+		else if (tarball) {78+			const [, repo = "", , save = ""] = tarball;79+			// Not an orphan, so its repo is in the bucket and in `use`.80+			(use.get(repo) as RepoUse).tarballs += size;81+			survey.kept.push({ key, repo, save, size });82+		} else if (key.startsWith(KEPR_PREFIX)) survey.kepr += size;4683 		else {4784 			const [repo = "", kind] = key.split("/");4885 			const owner = use.get(repo);@@ -50,20 +87,101 @@5087 			else if (owner) owner.history += size;5188 		}5289 	}53-	const orphanBytes = orphans.reduce((sum, key) => sum + (sizes.get(key) ?? 0), 0);54-	return { use, kept, orphans, orphanBytes, kepr, total };90+	survey.use = [...use];91+	keepState(target.kepr, "storage", JSON.stringify(survey));5592 }5693 57-async function storagePage(c: Context, ctx: AdminContext, error: string) {58-	const { use, kept, orphans, orphanBytes, kepr, total } = await survey(ctx);94+/**95+ * Whether a key is an orphan given the repos now in the bucket: a tarball of an older generation,96+ * or Kepr's file of a repo no longer there. Only mirrors of repos Kepr has registered are opened, as97+ * pages do: a rename's new name is registered only once its mirror has moved into place.98+ */99+function orphanTest(target: SyncTarget, repos: string[]): (key: string) => boolean {100+	const registered = new Set(allRepos(target.kepr).map((repo) => repo.name));101+	const generations = new Map(102+		repos103+			.filter((repo) => registered.has(repo))104+			.map((repo) => [repo, tarballGeneration(openMirror(target.mirrorDir, repo))]),105+	);106+	return (key) => {107+		const tarball = TARBALL.exec(key);108+		const [, repo = "", generation] = tarball ?? HIDDEN.exec(key) ?? [];109+		if (!repos.includes(repo)) return tarball !== null || HIDDEN.test(key);110+		// A repo not registered yet has no tarballs of its own making; its keys are left alone.111+		return tarball !== null && generations.has(repo) && generations.get(repo) !== generation;112+	};113+}114+115+function storedSurvey(ctx: AdminContext): Survey | null {116+	const stored = setting(ctx.kepr, "storage");117+	return stored ? (JSON.parse(stored) as Survey) : null;118+}119+120+/** The survey without keys just deleted, so the page never lists what is gone. */121+function withoutKeys(survey: Survey, gone: Set<string>): Survey {122+	const sizeOf = new Map([...survey.orphans, ...survey.kept.map((t) => [t.key, t.size] as const)]);123+	const freed = [...gone].reduce((sum, key) => sum + (sizeOf.get(key) ?? 0), 0);124+	const use = survey.use.map(([repo, u]): [string, RepoUse] => {125+		const dropped = survey.kept.filter((t) => t.repo === repo && gone.has(t.key));126+		return [repo, { ...u, tarballs: u.tarballs - dropped.reduce((sum, t) => sum + t.size, 0) }];127+	});128+	return {129+		...survey,130+		use,131+		kept: survey.kept.filter((t) => !gone.has(t.key)),132+		orphans: survey.orphans.filter(([key]) => !gone.has(key)),133+		total: survey.total - freed,134+	};135+}136+137+/**138+ * Deletes the orphans, or one kept tarball, from the stored survey; returns "unknown" for a key it139+ * doesn't list, the bucket's refusal, or "" when done.140+ */141+async function deleteKeys(ctx: AdminContext, fields: Record<string, string>): Promise<string> {142+	const survey = storedSurvey(ctx);143+	if (!survey) return "unknown";144+	let doomed: string[];145+	if (fields.action === "delete_orphans") {146+		// Checked again now: an older generation can become current again, as unhiding does.147+		const isOrphan = orphanTest(ctx.sync, await listRepos(ctx.sync.bucket));148+		doomed = survey.orphans.map(([key]) => key).filter(isOrphan);149+	} else {150+		doomed = survey.kept.filter((t) => t.key === fields.key).map((t) => t.key);151+		if (doomed.length === 0) return "unknown";152+	}153+	const gone = new Set<string>();154+	try {155+		for (const key of doomed) {156+			await ctx.sync.bucket.remove(key);157+			gone.add(key);158+		}159+	} catch (error) {160+		if (!(error instanceof BucketWriteError)) throw error;161+		return error.message;162+	} finally {163+		keepState(ctx.kepr, "storage", JSON.stringify(withoutKeys(survey, gone)));164+	}165+	return "";166+}167+168+function storagePage(c: Context, ctx: AdminContext, error: string) {169+	const survey = storedSurvey(ctx);170+	if (!survey) {171+		const body = html`${notice(error)}172+<p class="admin-intro">Not measured yet: Kepr measures the bucket when it starts and every hour.</p>`;173+		return adminPage(c, ctx, "storage", body);174+	}175+	const { use, kept, orphans, kepr, total } = survey;176+	const orphanBytes = orphans.reduce((sum, [, size]) => sum + size, 0);59177 	const body = html`${notice(error)}60178 <p class="admin-intro">The bucket holds ${formatSize(total)}; Kepr's own folder ${formatSize(kepr)}61-besides its kept tarballs.</p>179+besides its kept tarballs. Measured ${relativeTime(survey.at)}; Kepr measures it every hour.</p>62180 <section class="repo-section"><h2 class="section-label mono">by repository</h2>63181 <table class="admin-table mono">64182 <thead><tr><th>repo</th><th>history</th><th>builds</th><th>kept tarballs</th></tr></thead>65183 <tbody>66-${[...use].map(184+${use.map(67185 	([repo, u]) => html`<tr>68186 <td>${repo}</td>69187 <td>${formatSize(u.history)}</td>@@ -99,22 +217,19 @@99217 100218 /** §24's Storage section: sizes, kept tarballs and orphan cleanup. */101219 export function registerStorage(app: Hono, ctx: AdminContext): void {102-	app.get("/admin/storage", async (c) => c.html(await storagePage(c, ctx, "")));220+	app.get("/admin/storage", (c) => c.html(storagePage(c, ctx, "")));103221 104222 	app.post("/admin/storage", async (c) => {105223 		const fields = await textFields(c);106-		const { kept, orphans } = await survey(ctx);107-		const doomed =108-			fields.action === "delete_orphans"109-				? orphans110-				: kept.filter((t) => t.key === fields.key).map((t) => t.key);111-		if (doomed.length === 0 && fields.action !== "delete_orphans") return c.notFound();112-		try {113-			for (const key of doomed) await ctx.sync.bucket.remove(key);114-		} catch (error) {115-			if (!(error instanceof BucketWriteError)) throw error;116-			return c.html(await storagePage(c, ctx, error.message), 503);117-		}224+		// While no sync or rename runs: a rename can make a listed orphan live again mid-delete.225+		let outcome = "";226+		await holdSync(() =>227+			inStorageTurn(async () => {228+				outcome = await deleteKeys(ctx, fields);229+			}),230+		);231+		if (outcome === "unknown") return c.notFound();232+		if (outcome) return c.html(storagePage(c, ctx, outcome), 503);118233 		return c.redirect("/admin/storage?saved=1", 303);119234 	});120235 }

@@ -14,6 +14,10 @@1414 export type Bucket = {1515 	get(key: string): Promise<Uint8Array | null>;1616 	list(prefix: string): Promise<string[]>;17+	/** The top-level folders' names: one listing, however many keys they hold. */18+	folders(): Promise<string[]>;19+	/** A key's size in bytes, asked without reading it. */20+	size(key: string): Promise<number>;1721 	/** Every key under `prefix` with its size in bytes, for the console's storage view. */1822 	sizes(prefix: string): Promise<Map<string, number>>;1923 	/** A URL that serves `key` for five minutes, as an attachment with this filename and content type. */@@ -35,6 +39,19 @@3539 3640 export const KEPR_PREFIX = ".kepr/";3741 42+/** `each` over `items`, `atOnce` of them in flight at a time; the results in order. */43+export async function inTurn<T, R>(44+	items: T[],45+	atOnce: number,46+	each: (item: T) => Promise<R>,47+): Promise<R[]> {48+	const results: R[] = [];49+	for (let i = 0; i < items.length; i += atOnce) {50+		results.push(...(await Promise.all(items.slice(i, i + atOnce).map(each))));51+	}52+	return results;53+}54+3855 /** A write Kepr can't make: outside its folder, or without the `[bucket.write]` key. */3956 export class BucketWriteError extends Error {}4057 @@ -131,6 +148,17 @@131148 			} while (continuationToken);132149 			return keys;133150 		},151+		size: (key) => client.size(key),152+		async folders() {153+			const names: string[] = [];154+			let continuationToken: string | undefined;155+			do {156+				const page = await client.list({ delimiter: "/", continuationToken });157+				for (const { prefix } of page.commonPrefixes ?? []) names.push(prefix.slice(0, -1));158+				continuationToken = page.isTruncated ? page.nextContinuationToken : undefined;159+			} while (continuationToken);160+			return names;161+		},134162 		async sizes(prefix) {135163 			const sizes = new Map<string, number>();136164 			let continuationToken: string | undefined;@@ -191,6 +219,17 @@191219 			return (await file.exists()) ? await file.bytes() : null;192220 		},193221 		list,222+		async size(key) {223+			if (key.split("/").includes("..")) throw new Error(`bucket key escapes the folder: ${key}`);224+			return Bun.file(join(root, key)).size;225+		},226+		async folders() {227+			const entries = await readdir(root, { withFileTypes: true });228+			return entries229+				.filter((entry) => entry.isDirectory())230+				.map((entry) => entry.name)231+				.sort();232+		},194233 		async sizes(prefix) {195234 			return new Map((await list(prefix)).map((key) => [key, Bun.file(join(root, key)).size]));196235 		},

@@ -67,8 +67,8 @@6767 /** A save record as it arrived, less the files §5.5 quarantines, named here escaped. */6868 export type ArrivingSave = SaveRecord & { quarantined: string[] };6969 70-/** An object's stored zlib bytes and the inflated bytes they were verified from. */71-export type VerifiedObject = { compressed: Uint8Array; bytes: Uint8Array };70+/** An object's stored zlib bytes, verified against its hash, and its inflated size. */71+export type VerifiedObject = { compressed: Uint8Array; size: number };7272 7373 export class BucketError extends Error {}7474 @@ -95,13 +95,14 @@9595 const JSON_ESCAPE =9696 	/\\u[dD][89abAB][0-9a-fA-F]{2}\\u[dD][c-fC-F][0-9a-fA-F]{2}|\\u[dD][89a-fA-F][0-9a-fA-F]{2}|\\[\s\S]/g;9797 98-// ponytail: one full listing per call; switch to a delimiter listing if the bucket grows large.98+/**99+ * Every repo in the bucket: a top-level folder with a `head`. A folder without one is not a repo100+ * yet, as when a first offer or a rename (which copies `head` last) is still under way.101+ */99102 export async function listRepos(bucket: Bucket): Promise<string[]> {100-	const keys = await bucket.list("");101-	return keys102-		.filter((key) => key.endsWith("/head") && key.split("/").length === 2)103-		.map((key) => key.slice(0, -"/head".length))104-		.filter(isRepoName);103+	const names = (await bucket.folders()).filter(isRepoName);104+	const heads = await Promise.all(names.map((name) => bucket.get(`${name}/head`)));105+	return names.filter((_, i) => heads[i] !== null);105106 }106107 107108 function isRepoName(name: string): boolean {@@ -177,15 +178,27 @@177178 	);178179 }179180 180-/** Fetches an object, inflates it and checks its Blake3 against its key, as steal does (§9.4). */181+function objectKey(repo: string, hash: string): string {182+	requireRepo(repo);183+	if (!HEX64.test(hash)) throw new BucketError(`not an object hash: "${hash}"`);184+	return `${repo}/objects/${hash}`;185+}186+187+/** An object's stored, compressed size, asked without reading it. */188+export function objectSize(bucket: Bucket, repo: string, hash: string): Promise<number> {189+	return bucket.size(objectKey(repo, hash));190+}191+192+/**193+ * Fetches an object, inflates it and checks its Blake3 against its key, as steal does (§9.4). The194+ * inflated bytes never leave this call, so reads in parallel hold one inflated object at a time.195+ */181196 export async function readObject(182197 	bucket: Bucket,183198 	repo: string,184199 	hash: string,185200 ): Promise<VerifiedObject> {186-	requireRepo(repo);187-	if (!HEX64.test(hash)) throw new BucketError(`not an object hash: "${hash}"`);188-	const compressed = await bucket.get(`${repo}/objects/${hash}`);201+	const compressed = await bucket.get(objectKey(repo, hash));189202 	if (compressed === null) throw new BucketError(`${repo}/objects/${hash} is missing`);190203 	let bytes: Uint8Array;191204 	try {@@ -199,7 +212,7 @@199212 	if (bytesToHex(blake3(bytes)) !== hash) {200213 		throw new BucketError(`${repo}/objects/${hash} does not match its hash`);201214 	}202-	return { compressed, bytes };215+	return { compressed, size: bytes.length };203216 }204217 205218 export async function listEventIds(bucket: Bucket, repo: string): Promise<string[]> {

@@ -14,13 +14,14 @@1414 import {1515 	hiddenPaths,1616 	hiddenToHand,17+	keepState,1718 	recordHiddenHanded,1819 	recordSync,1920 	registerRepo,2021 } from "../server/db.ts";2122 import { RESERVED_NAMES } from "../server/reserved.ts";2223 import { dropTarballs } from "../tarball/cache.ts";23-import { type Bucket, BucketWriteError, KEPR_PREFIX } from "./reader.ts";24+import { type Bucket, BucketWriteError, inTurn, KEPR_PREFIX } from "./reader.ts";2425 import {2526 	BucketError,2627 	checkFormat,@@ -31,6 +32,7 @@3132 	listReleaseVersions,3233 	listRepos,3334 	OversizedObject,35+	objectSize,3436 	type ReleaseRecord,3537 	readEvent,3638 	readHead,@@ -45,6 +47,11 @@4547 // Compressed bytes a sync holds before landing them; sized for a 2 GB host.4648 const LANDING_BYTES = 32 * 1024 * 1024;4749 const 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.52+const READS_AT_ONCE = 16;53+// Compressed object bytes being read at once; a larger object waits and is read alone.54+const READ_BYTES = 32 * 1024 * 1024;4855 4956 export type SyncTarget = {5057 	bucket: Bucket;@@ -82,7 +89,18 @@8289 }8390 8491 async function syncEach(target: SyncTarget): Promise<void> {85-	for (const repo of await listRepos(target.bucket)) {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) {86104 		registerRepo(target.kepr, repo);87105 		let error: string | null = null;88106 		if (RESERVED_NAMES.has(repo)) {@@ -147,6 +165,7 @@147165 	repo: string,148166 	hidden: string[] = [],149167 	landingBytes = LANDING_BYTES,168+	readBytes = READ_BYTES,150169 ): Promise<EventRecord[]> {151170 	applyHidden(mirror, hidden);152171 	await checkFormat(bucket, repo);@@ -182,54 +201,80 @@182201 		quarantined = [];183202 		fetched = 0;184203 	};185-	for (const id of [...unknown].reverse()) {186-		const save = await readSave(bucket, repo, id);187-		quarantined.push(...save.quarantined);188-		// One save can hold more than a batch, as a first import does: its objects are stored ahead189-		// of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced.190-		const ahead = new Map<string, StoredObject>();191-		for (const file of save.files) {192-			const { hash } = file;193-			if (194-				objects.has(hash) ||195-				ahead.has(hash) ||196-				oversized.has(hash) ||197-				knownObject(mirror, hash)198-			) {199-				continue;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;200215 			}201-			try {202-				const { compressed, bytes } = await readObject(bucket, repo, hash);203-				ahead.set(hash, { compressed, size: bytes.length });204-				fetched += compressed.length;205-			} catch (cause) {206-				if (!(cause instanceof OversizedObject)) throw cause;207-				oversized.add(hash);208-			}209-			if (fetched >= landingBytes) {210-				const last = saves.at(-1);211-				if (last) landBatch(last.id);212-				storeObjects(mirror, ahead);213-				ahead.clear();214-				fetched = 0;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);216224 		}217-		for (const [hash, object] of ahead) objects.set(hash, object);218-		quarantined.push(219-			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),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),220230 		);221-		save.files = save.files.filter((file) => !oversized.has(file.hash));222-		saves.push(save);223-		if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);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+		}224269 	}225270 	const known = knownEventIds(mirror);226-	const events: EventRecord[] = [];227-	for (const id of await listEventIds(bucket, repo)) {228-		if (known.has(id)) continue;229-		const event = await readEvent(bucket, repo, id);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(230275 		// Koh ignores it too: no save can hold an illegal path (§5.5).231-		if (event.kind !== "strip" || isSafePath(event.path)) events.push(event);232-	}276+		(event) => event.kind !== "strip" || isSafePath(event.path),277+	);233278 	const releasesLatest = (await readReleasesLatest(bucket, repo)) ?? "";234279 	const moved =235280 		mirrorState(mirror, "head") !== head ||@@ -248,15 +293,32 @@248293 	return events;249294 }250295 296+/** Admits reads while their bytes in flight stay within `limit`; a larger one waits to go alone. */297+function 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+}315+251316 async function newReleases(252317 	bucket: Bucket,253318 	mirror: Database,254319 	repo: string,255320 ): Promise<ReleaseRecord[]> {256321 	const known = knownReleases(mirror);257-	const releases: ReleaseRecord[] = [];258-	for (const version of await listReleaseVersions(bucket, repo)) {259-		if (!known.has(version)) releases.push(await readRelease(bucket, repo, version));260-	}261-	return releases;322+	const versions = (await listReleaseVersions(bucket, repo)).filter((v) => !known.has(v));323+	return inTurn(versions, READS_AT_ONCE, (version) => readRelease(bucket, repo, version));262324 }

@@ -1,6 +1,7 @@11 import { join } from "node:path";22 import type { Hono } from "hono";33 import { createSetupApp } from "./admin/setup.ts";4+import { startStorageLoop } from "./admin/storage.ts";45 import { openBucket } from "./bucket/reader.ts";56 import { startSyncLoop } from "./bucket/sync.ts";67 import { warmAtlas } from "./pages/atlas.ts";@@ -23,12 +24,9 @@2324 	startMailLoop(kepr, config);2425 	const mirrorDir = join(config.dataDir, "mirror");2526 	// The first visitor after a new save would otherwise wait a second or two on a big repo.26-	startSyncLoop({27-		bucket,28-		kepr,29-		mirrorDir,30-		synced: (repo) => warmAtlas({ kepr, mirrorDir }, repo),31-	});27+	const sync = { bucket, kepr, mirrorDir };28+	startSyncLoop({ ...sync, synced: (repo) => warmAtlas({ kepr, mirrorDir }, repo) });29+	startStorageLoop(sync);3230 	return createApp(config, kepr, bucket).fetch;3331 }3432 

@@ -317,6 +317,12 @@317317 	return [...paths, ...ignoredNames(db).map((name) => `${ANY_DEPTH}${name}`)];318318 }319319 320+/** Keeps a piece of Kepr's own state in settings, never shown as a setting; null removes it. */321+export function keepState(db: Database, key: string, value: string | null): void {322+	if (value === null) db.query("DELETE FROM settings WHERE key = ?").run(key);323+	else db.query("INSERT OR REPLACE INTO settings (key, value) VALUES (?, ?)").run(key, value);324+}325+320326 /** A repo's hidden paths unless they are known to be the list last handed to Koh, else null. */321327 export function hiddenToHand(db: Database, repo: string): string | null {322328 	const row = db

@@ -68,14 +68,24 @@6868 		expect(admin.status).toBe(200);6969 		expect(admin.headers.get("cache-control")).toBe("no-store");7070 		const text = await admin.text();71-		expect(text).toContain("Bucket reachable: 1 repo.");71+		// As the sync recorded it: the console never waits on the bucket.72+		expect(text).toMatch(/Bucket read by the sync (just now|\d+ \w+ ago): 1 repo\./);7273 		expect(text).toContain("format ok");74+		// A repo gone from the bucket keeps its row, but the last listing didn't find it.75+		inst.kepr.query("INSERT INTO repos (name) VALUES ('gone')").run();76+		expect(await (await browser.get("/admin")).text()).toMatch(/: 1 repo\./);77+		inst.kepr.query("DELETE FROM repos WHERE name = 'gone'").run();78+		inst.kepr.query("INSERT INTO settings (key, value) VALUES ('bucket_error', 'timed out')").run();79+		expect(await (await browser.get("/admin")).text()).toContain(80+			"Bucket unreachable at the last sync: timed out",81+		);82+		inst.kepr.query("DELETE FROM settings WHERE key = 'bucket_error'").run();7383 		expect(text).toMatch(/Running kepr 3\.0\.\d+: [\d,]+ lines of TypeScript/);7484 		expect(text).toContain('href="/admin?section=backups"');7585 		const backupPage = await (await browser.get("/admin?section=backups")).text();7686 		expect(backupPage).toContain("<h1>backups</h1>");7787 		expect(backupPage).toContain("Sessions and bucket credentials are not included");78-		expect(backupPage).not.toContain("Bucket reachable:");88+		expect(backupPage).not.toContain("Bucket read by the sync");7989 	});8090 8191 	test("a wrong password, a foreign origin or a missing sign-in cookie is refused", async () => {

@@ -1,6 +1,7 @@11 import { beforeAll, describe, expect, test } from "bun:test";2-import { writeFile } from "node:fs/promises";2+import { cp, writeFile } from "node:fs/promises";33 import { join } from "node:path";4+import { inflateSync } from "node:zlib";45 import { openBucket } from "../src/bucket/reader.ts";56 import {67 	checkFormat,@@ -26,6 +27,15 @@2627 		expect(await listRepos(bucket())).toEqual([fx.repo]);2728 	});2829 30+	test("a folder without a head, or Kepr's own, is not a repo", async () => {31+		// As a rename leaves the new name until it copies `head`, last.32+		await cp(join(fx.bucketDir, fx.repo, "saves"), join(fx.bucketDir, "half", "saves"), {33+			recursive: true,34+		});35+		await Bun.write(join(fx.bucketDir, ".kepr/hidden", fx.repo), "secret.txt\n");36+		expect(await listRepos(bucket())).toEqual([fx.repo]);37+	});38+2939 	test("reads format, head, the save record and one verified object end to end", async () => {3040 		const b = bucket();3141 		await checkFormat(b, fx.repo);@@ -41,8 +51,10 @@4151 		expect([first.agent, first.session]).toEqual(["test-agent", "fixture"]);4252 4353 		const main = save.files.find((file) => file.path === "src/main.ts");44-		const { bytes } = await readObject(b, fx.repo, main?.hash as string);45-		expect(new TextDecoder().decode(bytes)).toBe("// Entry point.\nconsole.log('hello');\n");54+		const { compressed, size } = await readObject(b, fx.repo, main?.hash as string);55+		const text = "// Entry point.\nconsole.log('hello');\n";56+		expect(new TextDecoder().decode(inflateSync(compressed))).toBe(text);57+		expect(size).toBe(text.length);4658 	});4759 4860 	test("a missing key reads as null and a listing is prefix-scoped", async () => {@@ -152,6 +164,8 @@152164 			{153165 				get: async () => bytes,154166 				list: async () => [],167+				folders: async () => [],168+				size: async () => 0,155169 				sizes: async () => new Map(),156170 				downloadUrl: () => "",157171 				viewUrl: () => null,@@ -207,6 +221,8 @@207221 			{208222 				get: async () => bytes,209223 				list: async () => [],224+				folders: async () => [],225+				size: async () => 0,210226 				sizes: async () => new Map(),211227 				downloadUrl: () => "",212228 				viewUrl: () => null,@@ -288,6 +304,8 @@288304 		const reader = {289305 			get: async () => accepted,290306 			list: async () => [],307+			folders: async () => [],308+			size: async () => 0,291309 			sizes: async () => new Map(),292310 			downloadUrl: () => "",293311 			viewUrl: () => null,@@ -311,6 +329,8 @@311329 			const tampered = {312330 				get: async () => bytes,313331 				list: async () => [],332+				folders: async () => [],333+				size: async () => 0,314334 				sizes: async () => new Map(),315335 				downloadUrl: () => "",316336 				viewUrl: () => null,

@@ -18,15 +18,26 @@1818 	if (bucket !== BUCKET) return new Response("", { status: 404 });1919 	if (url.searchParams.get("list-type") === "2") {2020 		const prefix = url.searchParams.get("prefix") ?? "";21+		const delimiter = url.searchParams.get("delimiter");2122 		const keys = [...objects.keys()].filter((key) => key.startsWith(prefix)).sort();22-		const contents = keys23+		// With a delimiter, keys sharing a folder after the prefix roll up into one CommonPrefixes.24+		const folders = new Set<string>();25+		const listed = keys.filter((key) => {26+			const at = delimiter ? key.indexOf(delimiter, prefix.length) : -1;27+			if (at >= 0) folders.add(key.slice(0, at + 1));28+			return at < 0;29+		});30+		const contents = listed2331 			.map(2432 				(key) =>2533 					`<Contents><Key>${key}</Key><Size>${objects.get(key)?.length ?? 0}</Size></Contents>`,2634 			)2735 			.join("");36+		const common = [...folders]37+			.map((folder) => `<CommonPrefixes><Prefix>${folder}</Prefix></CommonPrefixes>`)38+			.join("");2839 		return new Response(29-			`<?xml version="1.0" encoding="UTF-8"?><ListBucketResult><IsTruncated>false</IsTruncated>${contents}</ListBucketResult>`,40+			`<?xml version="1.0" encoding="UTF-8"?><ListBucketResult><IsTruncated>false</IsTruncated>${contents}${common}</ListBucketResult>`,3041 			{ headers: { "content-type": "application/xml" } },3142 		);3243 	}