give each bundle its own statements, so two steals at once no longer split the rows between them (v3.0.214)

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

@@ -181,32 +181,43 @@181181 			if (next.done) controller.close();182182 			else controller.enqueue(next.value);183183 		},184+		// A client that hangs up leaves the generator part way through; this finalizes its queries.185+		cancel() {186+			chunks.return(undefined);187+		},184188 	});185189 }186190 187-/**188- * Each entry is `<key> <length>\n` and then that many bytes, keyed as the single routes are, and189- * the last line is `end 0`.190- */191+/** `<key> <length>\n` and that many bytes per entry, keyed as the single routes are; `end 0` last. */191192 function* bundle(mirror: Database): Generator<Uint8Array> {192193 	const encoder = new TextEncoder();193194 	function* entry(key: string, bytes: Uint8Array): Generator<Uint8Array> {194195 		yield encoder.encode(`${key} ${bytes.length}\n`);195196 		yield bytes;196197 	}197-	for (const row of mirror.query<SaveRow, []>(`SELECT ${SAVE_COLUMNS} FROM saves`).iterate()) {198-		yield* entry(`saves/${row.id}`, encoder.encode(saveRecord(mirror, row)));199-	}200-	for (const row of mirror.query<EventRow, []>(`SELECT ${EVENT_COLUMNS} FROM events`).iterate()) {201-		yield* entry(`events/${row.id}`, encoder.encode(eventRecord(row)));202-	}203-	const objects = mirror.query<{ hash: string; data: Uint8Array }, []>(198+	// `prepare`, never the cached `query`: two bundles streaming at once would share one cursor and199+	// split the rows between them.200+	const saves = mirror.prepare<SaveRow, []>(`SELECT ${SAVE_COLUMNS} FROM saves`);201+	const events = mirror.prepare<EventRow, []>(`SELECT ${EVENT_COLUMNS} FROM events`);202+	const objects = mirror.prepare<{ hash: string; data: Uint8Array }, []>(204203 		`SELECT hash, data FROM objects o205204 		WHERE EXISTS (SELECT 1 FROM save_files f WHERE f.hash = o.hash)`,206205 	);207-	for (const row of objects.iterate()) yield* entry(`objects/${row.hash}`, row.data);208-	// A connection cut between two entries would otherwise pass for the whole repo.209-	yield encoder.encode("end 0\n");206+	try {207+		for (const row of saves.iterate()) {208+			yield* entry(`saves/${row.id}`, encoder.encode(saveRecord(mirror, row)));209+		}210+		for (const row of events.iterate()) {211+			yield* entry(`events/${row.id}`, encoder.encode(eventRecord(row)));212+		}213+		for (const row of objects.iterate()) yield* entry(`objects/${row.hash}`, row.data);214+		// A connection cut between two entries would otherwise pass for the whole repo.215+		yield encoder.encode("end 0\n");216+	} finally {217+		saves.finalize();218+		events.finalize();219+		objects.finalize();220+	}210221 }211222 212223 // Field names and their order are Koh's, and every optional is an empty string, as `json2.encode`

@@ -158,6 +158,15 @@158158 		}159159 	});160160 161+	test("two bundles read at once each hold the whole repo", async () => {162+		const [first, second] = await Promise.all(163+			[get("bundle"), get("bundle")].map(async (response) => (await response).arrayBuffer()),164+		);165+		const whole = await (await get("bundle")).arrayBuffer();166+		expect(first?.byteLength).toBe(whole.byteLength);167+		expect(second?.byteLength).toBe(whole.byteLength);168+	});169+161170 	test("a private repo answers exactly as an unknown one", async () => {162171 		inst.kepr.run("UPDATE repos SET public = 0");163172 		try {