store a save's objects ahead of it when one save alone passes a batch, so a first import's huge save fits in memory (v3.0.179)

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

@@ -9,6 +9,7 @@99 	mirrorState,1010 	openMirror,1111 	type StoredObject,12+	storeObjects,1213 } from "../mirror/store.ts";1314 import { hiddenPaths, recordSync, registerRepo } from "../server/db.ts";1415 import { RESERVED_NAMES } from "../server/reserved.ts";@@ -148,43 +149,61 @@148149 	let quarantined: string[] = [];149150 	let fetched = 0;150151 	const oversized = new Set<string>();152+	const landBatch = (batchHead: string) => {153+		land(mirror, {154+			head: batchHead,155+			releasesLatest: mirrorState(mirror, "releases_latest") ?? "",156+			saves,157+			objects,158+			events: [],159+			releases: [],160+			quarantined,161+			hidden,162+		});163+		saves = [];164+		objects = new Map();165+		quarantined = [];166+		fetched = 0;167+	};151168 	for (const id of [...unknown].reverse()) {152169 		const save = await readSave(bucket, repo, id);153170 		quarantined.push(...save.quarantined);171+		// One save can hold more than a batch, as a first import does: its objects are stored ahead172+		// of it, after the saves before it have landed, so no landing's cleanup sees them unreferenced.173+		const ahead = new Map<string, StoredObject>();154174 		for (const file of save.files) {155-			if (objects.has(file.hash) || oversized.has(file.hash) || knownObject(mirror, file.hash)) {175+			const { hash } = file;176+			if (177+				objects.has(hash) ||178+				ahead.has(hash) ||179+				oversized.has(hash) ||180+				knownObject(mirror, hash)181+			) {156182 				continue;157183 			}158184 			try {159-				const { compressed, bytes } = await readObject(bucket, repo, file.hash);160-				objects.set(file.hash, { compressed, size: bytes.length });185+				const { compressed, bytes } = await readObject(bucket, repo, hash);186+				ahead.set(hash, { compressed, size: bytes.length });161187 				fetched += compressed.length;162188 			} catch (cause) {163189 				if (!(cause instanceof OversizedObject)) throw cause;164-				oversized.add(file.hash);190+				oversized.add(hash);165191 			}192+			if (fetched >= landingBytes) {193+				const last = saves.at(-1);194+				if (last) landBatch(last.id);195+				storeObjects(mirror, ahead);196+				ahead.clear();197+				fetched = 0;198+			}166199 		}200+		for (const [hash, object] of ahead) objects.set(hash, object);167201 		quarantined.push(168202 			...save.files.filter((f) => oversized.has(f.hash)).map((f) => escapePath(f.path)),169203 		);170204 		save.files = save.files.filter((file) => !oversized.has(file.hash));171205 		saves.push(save);172-		if ((fetched >= landingBytes || saves.length >= LANDING_SAVES) && save.id !== head) {173-			land(mirror, {174-				head: save.id,175-				releasesLatest: mirrorState(mirror, "releases_latest") ?? "",176-				saves,177-				objects,178-				events: [],179-				releases: [],180-				quarantined,181-				hidden,182-			});183-			saves = [];184-			objects = new Map();185-			quarantined = [];186-			fetched = 0;187-		}206+		if (saves.length >= LANDING_SAVES && save.id !== head) landBatch(save.id);188207 	}189208 	const known = knownEventIds(mirror);190209 	const events: EventRecord[] = [];

@@ -187,8 +187,15 @@187187 }188188 189189 /** Lands a sync like Koh's steal: objects, events, saves, then every strip over every save. */190+/** Stores objects on their own, ahead of the save that refers to them. */191+export function storeObjects(db: Database, objects: Map<string, StoredObject>): void {192+	const put = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");193+	db.transaction(() => {194+		for (const [hash, object] of objects) put.run(hash, object.size, object.compressed);195+	})();196+}197+190198 export function land(db: Database, landing: Landing): void {191-	const putObject = db.query("INSERT OR IGNORE INTO objects (hash, size, data) VALUES (?, ?, ?)");192199 	const putEvent = db.query(193200 		"INSERT OR IGNORE INTO events (id, kind, save_id, path, reason, created_at, machine) VALUES " +194201 			"(?, ?, ?, ?, ?, ?, ?)",@@ -203,8 +210,7 @@203210 	const putRelease = db.query("INSERT OR IGNORE INTO releases (version, record) VALUES (?, ?)");204211 	const setState = db.query("INSERT OR REPLACE INTO state (key, value) VALUES (?, ?)");205212 	db.transaction(() => {206-		for (const [hash, object] of landing.objects)207-			putObject.run(hash, object.size, object.compressed);213+		storeObjects(db, landing.objects);208214 		for (const e of landing.events) {209215 			putEvent.run(e.id, e.kind, e.save_id, e.path, e.reason, e.created_at, e.machine);210216 		}

@@ -148,6 +148,33 @@148148 	expect(flagReason(mirror, first)).toBe("holds a secret");149149 });150150 151+test("a save larger than a batch is stored ahead of itself, and a sync cut short within it resumes", async () => {152+	const fx = await makeKohBucket("ahead");153+	const bucket = openBucket({ kind: "folder", path: fx.bucketDir });154+	const mirror = openMirror(await mkdtemp(join(tmpdir(), "kepr-mirror-")), fx.repo);155+	let objectGets = 0;156+	const failing: Bucket = {157+		...bucket,158+		get: (key) =>159+			key.includes("/objects/") && ++objectGets > 2160+				? Promise.reject(new Error("connection reset"))161+				: bucket.get(key),162+	};163+	await expect(syncRepo(failing, mirror, fx.repo, [], 1)).rejects.toThrow("connection reset");164+	const stored = () =>165+		mirror.query<{ n: number }, []>("SELECT COUNT(*) AS n FROM objects").get()?.n;166+	expect(mirrorState(mirror, "head")).toBeNull();167+	expect(stored()).toBe(2);168+169+	await syncRepo(bucket, mirror, fx.repo, [], 1);170+	expect(headId(mirror)).toBe(fx.saveIds[0] as string);171+	expect(history(mirror, 10, 0)).toHaveLength(3);172+	const guide = fileAt(mirror, fx.saveIds[0] as string, "docs/guide.md");173+	expect(text(objectBytes(mirror, guide?.hash as string))).toBe(174+		"# Guide\n\nRead me. It is short.\n",175+	);176+});177+151178 test("a new save syncs incrementally and a bad format is recorded as the last sync error", async () => {152179 	const fx = await makeKohBucket("growing");153180 	const t = await syncTarget(fx);