only open body stream when it will be read in full

project/server/src/service/snapshot/defer/base.ts

@@ -8,15 +8,20 @@ import type {
 	SnapshotRequestTransformOut,
 } from "../stream";
 
-/** holds handle, dispose to prevent handle leakage when not consuming */
+/** body of a deferred snapshot
+ *
+ * must be disposed, whether iterated or not, to release the handle backing it */
+export type SnapshotDeferTargetDeferredSnapshot =
+	AsyncIterable<SnapshotRequestTransformOut> & AsyncDisposable;
+
+/** metadata of a deferred snapshot, obtainable without touching its body */
 export type SnapshotDeferTargetDeferred = {
 	voucher: SnapshotVoucher;
 	hassVersion: string;
 	hash: SnapshotHash;
-	snapshot: AsyncIterable<SnapshotRequestTransformOut>;
 	createdAt: Date;
-	/** releases the handle backing `snapshot` */
-	[Symbol.asyncDispose]: () => Promise<void>;
+	/** the result must be disposed, see {@link SnapshotDeferTargetDeferredSnapshot} */
+	open(): Promise<SnapshotDeferTargetDeferredSnapshot>;
 };
 
 export type ISnapshotDeferTarget = {
@@ -25,7 +30,6 @@ export type ISnapshotDeferTarget = {
 		hassVersion: string,
 		snapshot: SnapshotRequestTransform,
 	): Promise<void>;
-	/** the result must be disposed, see {@link SnapshotDeferTargetDeferred} */
 	deferred(): Promise<Maybe<SnapshotDeferTargetDeferred>>;
 	/** should be called after successful ingestion */
 	complete(id: Uuid): Promise<void>;

project/server/src/service/snapshot/defer/ingest.test.ts

@@ -40,14 +40,14 @@ const undeliverable = (): AsyncIterable<SnapshotRequestTransformOut> => ({
 });
 
 /** forges handle */
-const buildHandle = () =>
-	new Proxy({}, { get: () => ({ kind: "attachable-hashed" }) });
+const buildHandle = (kind = "attachable-hashed") =>
+	new Proxy({}, { get: () => ({ kind }) });
 
-/** a deferred snapshot that records whether it was disposed */
+/** a deferred snapshot that records whether it was opened and disposed */
 const buildDeferred = (
 	snapshot: AsyncIterable<SnapshotRequestTransformOut> = nonconsumable(),
 ) => {
-	const state = { disposed: 0 };
+	const state = { opened: 0, disposed: 0 };
 
 	const deferred: SnapshotDeferTargetDeferred = {
 		voucher: new Voucher("signing-key").create(
@@ -58,9 +58,14 @@ const buildDeferred = (
 		hassVersion: "2024.1.0",
 		hash: {} as SnapshotHash,
 		createdAt: new Date(0),
-		snapshot,
-		[Symbol.asyncDispose]: async () => {
-			state.disposed += 1;
+		open: async () => {
+			state.opened += 1;
+
+			return Object.assign(snapshot, {
+				[Symbol.asyncDispose]: async () => {
+					state.disposed += 1;
+				},
+			});
 		},
 	};
 
@@ -93,7 +98,7 @@ const buildTarget = (deferred: SnapshotDeferTargetDeferred) => {
 };
 
 /** snapshot service handing out an attachable handle */
-const buildSnapshot = () => {
+const buildSnapshot = (kind?: string) => {
 	const state = {
 		attached: [] as string[],
 		finalized: [] as string[],
@@ -104,7 +109,7 @@ const buildSnapshot = () => {
 		voucher: {
 			expiresAt: () => new Date(0),
 		},
-		create: async () => ({ kind: "success", handle: buildHandle() }),
+		create: async () => ({ kind: "success", handle: buildHandle(kind) }),
 		finalize: async (_handle: unknown, hassVersion: string) => {
 			state.finalized.push(hassVersion);
 		},
@@ -208,6 +213,33 @@ test("suspend", async (t: TestContext) => {
 });
 
 test("release", async (t: TestContext) => {
+	await t.test("unconsumed", async (t: TestContext) => {
+		const { deferred, state } = buildDeferred();
+		const { target, state: calls } = buildTarget(deferred);
+
+		const snapshot = {
+			voucher: {
+				expiresAt: () => new Date(0),
+			},
+			// handle acquisition fails, so `snapshot` is never opened
+			create: async () => ({ kind: "failure", reason: "voucher-used" }),
+		} as unknown as ISnapshot;
+
+		const iterator = buildIngest(target, snapshot)
+			.ingest()
+			[Symbol.asyncIterator]();
+
+		t.assert.deepStrictEqual(await iterator.next(), {
+			done: false,
+			value: "acted",
+		});
+
+		// nothing was ever opened, so there is no handle to release
+		t.assert.strictEqual(state.opened, 0);
+		t.assert.strictEqual(state.disposed, 0);
+		t.assert.deepStrictEqual(calls.completed, [id]);
+	});
+
 	await t.test("failed ingestion", async (t: TestContext) => {
 		const { deferred, state } = buildDeferred();
 		const { target } = buildTarget(deferred);
@@ -227,10 +259,69 @@ test("release", async (t: TestContext) => {
 
 		await t.assert.rejects(iterator.next(), /error/);
 
-		t.assert.strictEqual(state.disposed, 1);
+		t.assert.strictEqual(state.opened, 0);
+		t.assert.strictEqual(state.disposed, 0);
 	});
 });
 
+test("open", async (t: TestContext) => {
+	await t.test(
+		"leaves the body of a duplicate untouched",
+		async (t: TestContext) => {
+			const { deferred, state } = buildDeferred();
+			const { target, state: calls } = buildTarget(deferred);
+			const { snapshot, state: recorded } = buildSnapshot("duplicate");
+
+			const iterator = buildIngest(target, snapshot)
+				.ingest()
+				[Symbol.asyncIterator]();
+
+			t.assert.deepStrictEqual(await iterator.next(), {
+				done: false,
+				value: "acted",
+			});
+
+			// a duplicate is attributed without its body ever being fetched
+			t.assert.strictEqual(state.opened, 0);
+			t.assert.deepStrictEqual(recorded.attached, []);
+			t.assert.deepStrictEqual(recorded.finalized, ["2024.1.0"]);
+			t.assert.deepStrictEqual(recorded.deleted, []);
+			t.assert.deepStrictEqual(calls.completed, [id]);
+			t.assert.strictEqual(state.disposed, 0);
+		},
+	);
+
+	await t.test(
+		"archives when the body cannot be fetched",
+		async (t: TestContext) => {
+			const { deferred, state } = buildDeferred();
+			const { target, state: calls } = buildTarget({
+				...deferred,
+				open: async () => {
+					throw new Error("no such key");
+				},
+			});
+			const { snapshot, state: recorded } = buildSnapshot();
+
+			const iterator = buildIngest(target, snapshot)
+				.ingest()
+				[Symbol.asyncIterator]();
+
+			t.assert.deepStrictEqual(await iterator.next(), {
+				done: false,
+				value: "acted",
+			});
+
+			t.assert.deepStrictEqual(recorded.finalized, []);
+			t.assert.deepStrictEqual(recorded.deleted, [id]);
+			t.assert.deepStrictEqual(calls.completed, []);
+			t.assert.deepStrictEqual(calls.archived, [id]);
+			// opening threw, so there is nothing to release
+			t.assert.strictEqual(state.disposed, 0);
+		},
+	);
+});
+
 test("consume", async (t: TestContext) => {
 	await t.test("attaches every item", async (t: TestContext) => {
 		const { deferred, state } = buildDeferred(
@@ -255,6 +346,7 @@ test("consume", async (t: TestContext) => {
 		t.assert.deepStrictEqual(recorded.finalized, ["2024.1.0"]);
 		t.assert.deepStrictEqual(recorded.deleted, []);
 		t.assert.deepStrictEqual(calls.completed, [id]);
+		t.assert.strictEqual(state.opened, 1);
 		t.assert.strictEqual(state.disposed, 1);
 	});
 

project/server/src/service/snapshot/defer/ingest.ts

@@ -148,7 +148,7 @@ export class SnapshotDeferIngest
 	private async act(
 		target: ISnapshotDeferTarget,
 	): Promise<SnapshotDeferIngestIngestStep> {
-		await using deferred = await target.deferred();
+		const deferred = await target.deferred();
 		if (isNone(deferred)) {
 			return "idle";
 		}
@@ -177,18 +177,18 @@ export class SnapshotDeferIngest
 				reason: created.reason,
 			});
 
-			// TODO: figure out alternative to fully consuming that doesn't slowly leak handles
-			for await (const _ of deferred.snapshot) {
-			}
-
 			await target.complete(id);
 
 			return "acted";
 		}
 
 		try {
 			if (!Snapshot.isDuplicate(created.handle)) {
-				const consumed = await this.consume(created.handle, deferred.snapshot);
+				// only now is it certain the body will be read in full
+				await using snapshot = await deferred.open();
+
+				const consumed = await this.consume(created.handle, snapshot);
+
 				// production deployment uses scaleway's s3 offering
 				// said offering has been (very rarely) observed to corrupt objects
 				// corrupted objects can still be requested, though their purported `content-length` will be smaller than their real length
@@ -210,10 +210,6 @@ export class SnapshotDeferIngest
 
 					return "acted";
 				}
-			} else {
-				// TODO: figure out alternative to fully consuming that doesn't slowly leak handles
-				for await (const _ of deferred.snapshot) {
-				}
 			}
 
 			await this.snapshot.finalize(created.handle, deferred.hassVersion);
@@ -229,7 +225,6 @@ export class SnapshotDeferIngest
 						? err.message
 						: "unknown error",
 			});
-			console.error(err);
 
 			if (!completed) {
 				await target.archive(id);