land a sync's new history in 32 MB batches oldest first, so a big first sync fits in memory and resumes if cut short (v3.0.175)

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

@@ -15,6 +15,7 @@1515 import { dropTarballs } from "../tarball/cache.ts";1616 import type { Bucket } from "./reader.ts";1717 import {18+	type ArrivingSave,1819 	BucketError,1920 	checkFormat,2021 	type EventRecord,@@ -35,6 +36,8 @@3536 } from "./repo.ts";3637 3738 const INTERVAL_MS = 2 * 60 * 1000;39+// Compressed bytes a sync holds before landing them; sized for a 2 GB host.40+const LANDING_BYTES = 32 * 1024 * 1024;3841 3942 export type SyncTarget = {4043 	bucket: Bucket;@@ -86,7 +89,7 @@8689 8790 /**8891  * Koh's steal flow (§9.4) against a mirror: walk back from the remote head to a save the mirror has,89- * fetch what is missing, and land it in one transaction. Events are listed every time, since a92+ * fetch what is missing, and land it, in batches oldest first. Events are listed every time, since a9093  * flag or a strip arrives without moving `head`; releases only when `head` or `releases/latest`9194  * moved (§18), since publishing one moves both. A path §5.5 refuses, or one whose object inflates9295  * past Koh's file limit, is quarantined: the save lands without it, as if stripped. Returns the@@ -97,19 +100,28 @@97100 	mirror: Database,98101 	repo: string,99102 	hidden: string[] = [],103+	landingBytes = LANDING_BYTES,100104 ): Promise<EventRecord[]> {101105 	await checkFormat(bucket, repo);102106 	const head = await readHead(bucket, repo);103107 	if (head === null) return [];104-	const saves: SaveRecord[] = [];105-	const objects = new Map<string, StoredObject>();106-	const oversized = new Set<string>();107-	const quarantined: string[] = [];108+	const unknown: ArrivingSave[] = [];108109 	const seen = new Set<string>();109110 	for (let cursor = head; cursor !== "" && !knownSave(mirror, cursor); ) {110111 		if (seen.has(cursor)) throw new BucketError(`${repo}: save ${cursor} is its own ancestor`);111112 		seen.add(cursor);112113 		const save = await readSave(bucket, repo, cursor);114+		unknown.push(save);115+		cursor = save.parent;116+	}117+	// Oldest first, landing whenever the fetched objects pass `landingBytes`: a first sync of a big118+	// repo never holds its whole history in memory, and one cut short resumes from the last batch.119+	let saves: SaveRecord[] = [];120+	let objects = new Map<string, StoredObject>();121+	let quarantined: string[] = [];122+	let fetched = 0;123+	const oversized = new Set<string>();124+	for (const save of unknown.reverse()) {113125 		quarantined.push(...save.quarantined);114126 		for (const file of save.files) {115127 			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {@@ -118,6 +130,7 @@118130 			try {119131 				const { compressed, bytes } = await readObject(bucket, repo, file.hash);120132 				objects.set(file.hash, { compressed, size: bytes.length });133+				fetched += compressed.length;121134 			} catch (cause) {122135 				if (!(cause instanceof OversizedObject)) throw cause;123136 				oversized.add(file.hash);@@ -128,7 +141,22 @@128141 		);129142 		save.files = save.files.filter((file) => !oversized.has(file.hash));130143 		saves.push(save);131-		cursor = save.parent;144+		if (fetched >= landingBytes && save.id !== head) {145+			land(mirror, {146+				head: save.id,147+				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",148+				saves,149+				objects,150+				events: [],151+				releases: [],152+				quarantined,153+				hidden,154+			});155+			saves = [];156+			objects = new Map();157+			quarantined = [];158+			fetched = 0;159+		}132160 	}133161 	const known = knownEventIds(mirror);134162 	const events: EventRecord[] = [];@@ -147,7 +175,7 @@147175 	land(mirror, {148176 		head,149177 		releasesLatest,150-		saves: saves.reverse(),178+		saves,151179 		objects,152180 		events,153181 		releases,

@@ -59,7 +59,7 @@5959 /** An object as the bucket stores it: zlib bytes, already checked against its hash. */6060 export type StoredObject = { compressed: Uint8Array; size: number };6161 62-/** What one sync brings in, landed in one transaction. Saves are oldest first. */62+/** One batch of a sync, landed in one transaction. Saves are oldest first. */6363 export type Landing = {6464 	head: string;6565 	releasesLatest: string;

@@ -4,8 +4,8 @@44 import { tmpdir } from "node:os";55 import { join } from "node:path";66 import { deflateSync } from "node:zlib";7-import { openBucket } from "../src/bucket/reader.ts";8-import { syncAll } from "../src/bucket/sync.ts";7+import { type Bucket, openBucket } from "../src/bucket/reader.ts";8+import { syncAll, syncRepo } from "../src/bucket/sync.ts";99 import {1010 	fileAt,1111 	filesAt,@@ -123,6 +123,29 @@123123 		expect(objectBytes(t.mirror(), secret)).toBeNull();124124 		expect(fileAt(t.mirror(), head(), "src/main.ts")).not.toBeNull();125125 	});126+});127+128+test("a sync lands history in batches, so one cut short resumes where it stopped", async () => {129+	const fx = await makeKohBucket("batched");130+	const bucket = openBucket({ kind: "folder", path: fx.bucketDir });131+	const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo);132+	// The bucket goes away as soon as the first batch has landed.133+	const failing: Bucket = {134+		...bucket,135+		get: (key) =>136+			key.includes("/objects/") && mirrorState(mirror, "head") !== null137+				? Promise.reject(new Error("connection reset"))138+				: bucket.get(key),139+	};140+	await expect(syncRepo(failing, mirror, fx.repo, [], 1)).rejects.toThrow("connection reset");141+	const first = fx.saveIds[3] as string;142+	expect(mirrorState(mirror, "head")).toBe(first);143+	expect(history(mirror, 10, 0).map((s) => s.id)).toEqual([first]);144+145+	await syncRepo(bucket, mirror, fx.repo, [], 1);146+	expect(headId(mirror)).toBe(fx.saveIds[0] as string);147+	expect(history(mirror, 10, 0)).toHaveLength(3);148+	expect(flagReason(mirror, first)).toBe("holds a secret");126149 });127150 128151 test("a new save syncs incrementally and a bad format is recorded as the last sync error", async () => {