From 785bc63d226dcc2782088c0af7708130e81176b4 Mon Sep 17 00:00:00 2001 From: Philip Trauner Date: Mon, 12 Oct 2026 18:42:24 +0200 Subject: [PATCH] fix: corrupted object hang also try to retrieve error from underneath layers that sdk wraps stream in addition to being less permissive with timeouts --- .../src/service/snapshot/defer/ingest.test.ts | 63 ++++++++-- .../src/service/snapshot/defer/ingest.ts | 30 +++-- .../snapshot/defer/object-store.test.ts | 116 ++++++++++++++++++ .../service/snapshot/defer/object-store.ts | 22 ++++ 4 files changed, 205 insertions(+), 26 deletions(-) create mode 100644 project/server/src/service/snapshot/defer/object-store.test.ts diff --git a/project/server/src/service/snapshot/defer/ingest.test.ts b/project/server/src/service/snapshot/defer/ingest.test.ts index 231d1d6..9798c26 100644 --- a/project/server/src/service/snapshot/defer/ingest.test.ts +++ b/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(() => {}); + })(), + ); + 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((resolve) => { + release = resolve; + }); + const { deferred } = buildDeferred( (async function* () { yield entity; - await new Promise((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]); }, diff --git a/project/server/src/service/snapshot/defer/ingest.ts b/project/server/src/service/snapshot/defer/ingest.ts index bd4cfe5..130b198 100644 --- a/project/server/src/service/snapshot/defer/ingest.ts +++ b/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 | 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 | StalledSymbol + > => { const { resolve, promise: stalled } = Promise.withResolvers(); - 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 }; } } diff --git a/project/server/src/service/snapshot/defer/object-store.test.ts b/project/server/src/service/snapshot/defer/object-store.test.ts new file mode 100644 index 0000000..d3fb5e4 --- /dev/null +++ b/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" }); + }); + }); +}); diff --git a/project/server/src/service/snapshot/defer/object-store.ts b/project/server/src/service/snapshot/defer/object-store.ts index 28ae77a..3e18f51 100644 --- a/project/server/src/service/snapshot/defer/object-store.ts +++ b/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;