corrupted object hang

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

@@ -366,7 +366,7 @@ test("consume", async (t: TestContext) => {
 			const step = iterator.next();
 
 			await settled();
-			t.mock.timers.tick(30_000);
+			t.mock.timers.tick(10_000);
 
 			t.assert.deepStrictEqual(await step, { done: false, value: "acted" });
 
@@ -379,18 +379,60 @@ test("consume", async (t: TestContext) => {
 	);
 
 	await t.test(
-		"drops deadline once a chunk arrives",
+		"gives up on stream that stops delivering midway",
+		async (t: TestContext) => {
+			t.mock.timers.enable({ apis: ["setTimeout"] });
+
+			// mirrors an object whose advertised `content-length` exceeds what is served, which strands the read that would have ended the stream
+			const { deferred, state } = buildDeferred(
+				(async function* () {
+					yield entity;
+					await new Promise<never>(() => {});
+				})(),
+			);
+			const { target, state: calls } = buildTarget(deferred);
+			const { snapshot, state: recorded } = buildSnapshot();
+
+			const iterator = buildIngest(target, snapshot)
+				.ingest()
+				[Symbol.asyncIterator]();
+
+			const step = iterator.next();
+
+			await settled();
+			t.assert.deepStrictEqual(recorded.attached, ["entity:mqtt"]);
+
+			t.mock.timers.tick(10_000);
+
+			t.assert.deepStrictEqual(await step, { done: false, value: "acted" });
+
+			// what was attached before the stall is discarded along with the handle
+			t.assert.deepStrictEqual(calls.completed, [id]);
+			t.assert.deepStrictEqual(calls.archived, []);
+			t.assert.deepStrictEqual(recorded.finalized, []);
+			t.assert.deepStrictEqual(recorded.deleted, [id]);
+			t.assert.strictEqual(state.disposed, 1);
+		},
+	);
+
+	await t.test(
+		"restarts the deadline on every chunk",
 		async (t: TestContext) => {
 			t.mock.timers.enable({ apis: ["setTimeout"] });
 
 			let release: (() => void) | undefined;
+			const pause = () =>
+				new Promise<void>((resolve) => {
+					release = resolve;
+				});
+
 			const { deferred } = buildDeferred(
 				(async function* () {
 					yield entity;
-					await new Promise<void>((resolve) => {
-						release = resolve;
-					});
+					await pause();
 					yield device;
+					await pause();
+					yield entity;
 				})(),
 			);
 			const { target, state: calls } = buildTarget(deferred);
@@ -405,20 +447,21 @@ test("consume", async (t: TestContext) => {
 			await settled();
 			t.assert.deepStrictEqual(recorded.attached, ["entity:mqtt"]);
 
-			// beyond the deadline
-			t.mock.timers.tick(30_000 * 10);
+			// just short of the deadline, twice over, so the elapsed total exceeds it
+			t.mock.timers.tick(9_999);
+			release?.();
 			await settled();
 
-			t.assert.deepStrictEqual(calls.completed, []);
-			t.assert.deepStrictEqual(recorded.deleted, []);
-
+			t.mock.timers.tick(9_999);
 			release?.();
 
 			t.assert.deepStrictEqual(await step, { done: false, value: "acted" });
 			t.assert.deepStrictEqual(recorded.attached, [
 				"entity:mqtt",
 				"device:hue",
+				"entity:mqtt",
 			]);
+			t.assert.deepStrictEqual(recorded.deleted, []);
 			t.assert.deepStrictEqual(recorded.finalized, ["2024.1.0"]);
 			t.assert.deepStrictEqual(calls.completed, [id]);
 		},

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

@@ -191,9 +191,9 @@ export class SnapshotDeferIngest
 
 				// 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
+				// corrupted objects can still be requested, though their purported `content-length` exceeds the amount of bytes actually served
 				// this is problematic when consuming objects in a streaming-fashion, as the s3 sdk does _not like this_
-				// ingestion ends up stalling out, as advertised `content-length` is never reached
+				// ingestion ends up stalling out, as the advertised `content-length` is never reached
 				if (consumed.stalled) {
 					logger.warn(
 						`gave up on stalled deferred ingest <${id}> by <${sub}>`,
@@ -245,28 +245,26 @@ export class SnapshotDeferIngest
 		const StalledSymbol = Symbol("Stalled");
 		type StalledSymbol = typeof StalledSymbol;
 
-		// a stall surfaces on the first read or not at all, so only that read is guarded
-		let first: IteratorResult<SnapshotRequestTransformOut> | StalledSymbol;
-		{
+		// a corrupted object stalls once its real length is exhausted, which can be any read, including the one that would have reported the end of the stream
+		// the deadline therefore guards every read and is restarted by each delivery, so it measures idleness rather than total duration
+		const read = async (): Promise<
+			IteratorResult<SnapshotRequestTransformOut> | StalledSymbol
+		> => {
 			const { resolve, promise: stalled } =
 				Promise.withResolvers<StalledSymbol>();
-			const deadline = setTimeout(() => resolve(StalledSymbol), 30_000);
+			const deadline = setTimeout(() => resolve(StalledSymbol), 10_000);
 			// an ingest loop that is otherwise done must not be held open by the deadline
 			deadline.unref();
 
 			try {
-				first = await race([iterator.next(), stalled]);
+				return await race([iterator.next(), stalled]);
 			} finally {
-				// the stream has shown it delivers, the remainder of it runs unguarded
 				clearTimeout(deadline);
 			}
-		}
-		if (first === StalledSymbol) {
-			return { stalled: true };
-		}
+		};
 
-		let next = first;
-		while (next.done !== true) {
+		let next = await read();
+		while (next !== StalledSymbol && next.done !== true) {
 			const item = next.value;
 			if ("device" in item) {
 				await this.snapshot.attach.device(
@@ -283,9 +281,9 @@ export class SnapshotDeferIngest
 				);
 			}
 
-			next = await iterator.next();
+			next = await read();
 		}
 
-		return { stalled: false };
+		return { stalled: next === StalledSymbol };
 	}
 }

project/server/src/service/snapshot/defer/object-store.test.ts

@@ -0,0 +1,116 @@
+import { Duplex, Readable } from "node:stream";
+import { type TestContext, test } from "node:test";
+
+import { logger } from "../../../logger";
+import { bridgeFailure } from "./object-store";
+
+logger.silent = true;
+
+/** models how the sdk layers `ChecksumStream` on top of a response: a duplex fed by `source.pipe(this)` */
+class PipeLinkedWrapper extends Duplex {
+	constructor(readonly source: Readable) {
+		super();
+
+		this.source.pipe(this);
+	}
+
+	override _read(): void {}
+
+	override _write(
+		chunk: unknown,
+		_: BufferEncoding,
+		callback: (err?: Error) => void,
+	): void {
+		this.push(chunk);
+
+		callback();
+	}
+}
+
+/** a response that is still open, mirroring one whose advertised `content-length` has not been reached */
+const buildResponse = () => {
+	const source = new Readable({ read() {} });
+	source.push(Buffer.from("partial"));
+
+	return source;
+};
+
+/** records how a wrapper settles, if it settles at all */
+const watch = (wrapper: Readable) => {
+	const state = { error: undefined as string | undefined, ended: false };
+
+	wrapper.once("error", (err: Error) => {
+		state.error = err.message;
+	});
+	wrapper.once("end", () => {
+		state.ended = true;
+	});
+	wrapper.resume();
+
+	return state;
+};
+
+/** lets stream events reach their listeners */
+const settled = async () => {
+	for (let i = 0; i < 4; i += 1) {
+		await new Promise((resolve) => setImmediate(resolve));
+	}
+};
+
+test("bridgeFailure", async (t: TestContext) => {
+	await t.test(
+		"a pipe linked wrapper is deaf to the failure of its source",
+		async (t: TestContext) => {
+			const source = buildResponse();
+			const wrapper = new PipeLinkedWrapper(source);
+			const state = watch(wrapper);
+
+			// the sdk's own handler, without which the failure would go uncaught instead of unnoticed
+			source.once("error", () => {});
+
+			source.destroy(new Error("ECONNRESET"));
+			await settled();
+
+			// the premise of the bridge: the wrapper neither fails nor ends, so a reader of it waits forever
+			t.assert.strictEqual(state.error, undefined);
+			t.assert.strictEqual(state.ended, false);
+			t.assert.strictEqual(wrapper.destroyed, false);
+		},
+	);
+
+	await t.test(
+		"carries the failure across the link",
+		async (t: TestContext) => {
+			const source = buildResponse();
+			const wrapper = new PipeLinkedWrapper(source);
+			const state = watch(wrapper);
+
+			bridgeFailure(wrapper);
+
+			source.destroy(new Error("ECONNRESET"));
+			await settled();
+
+			t.assert.strictEqual(state.error, "ECONNRESET");
+		},
+	);
+
+	await t.test("leaves an unwrapped response alone", async (t: TestContext) => {
+		const source = buildResponse();
+		const state = watch(source);
+
+		bridgeFailure(source);
+
+		source.destroy(new Error("ECONNRESET"));
+		await settled();
+
+		// errors already reach a reader of the response itself, and the bridge must not swallow them
+		t.assert.strictEqual(state.error, "ECONNRESET");
+	});
+
+	await t.test("ignores a body that is not a stream", (t: TestContext) => {
+		t.assert.doesNotThrow(() => {
+			bridgeFailure(undefined);
+			bridgeFailure({ source: "not a stream" });
+		});
+	});
+});

project/server/src/service/snapshot/defer/object-store.ts

@@ -90,6 +90,26 @@ class NdJsonDecodeTransform extends Transform {
 	}
 }
 
+/** forwards a failure of the response to the wrapper the sdk layered on top of it
+ *
+ * the sdk replaces the response with a `ChecksumStream` whenever the object carries `x-amz-checksum-*` and links the two with `pipe`, which forwards neither teardown nor errors
+ *
+ * a response that dies mid-flight therefore leaves the wrapper waiting on bytes that will never arrive
+ * this has so far been what a corrupted object looks like: a body shorter than its advertised `content-length`, followed by a reset
+ */
+export const bridgeFailure = (body: unknown): void => {
+	if (!(body instanceof Readable)) {
+		return;
+	}
+
+	const source: unknown = (body as { source?: unknown }).source;
+	if (source instanceof Readable && source !== body) {
+		source.once("error", (err: Error) => {
+			body.destroy(err);
+		});
+	}
+};
+
 class SnapshotPersistedTransform extends Transform {
 	private static validator = Schema.is(SnapshotRequestTransformOut);
 
@@ -432,6 +452,8 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget {
 			throw new SnapshotDeferTargetObjectStoreMissingBodyError(key);
 		}
 
+		bridgeFailure(object.Body);
+
 		// a consumer can deliberately walk away mid-stream, which should not cause a warning
 		let disposing = false;