From cbbae8eb5af6d1f479db1fc5eb10ac96070f12b1 Mon Sep 17 00:00:00 2001 From: Philip Trauner Date: Mon, 12 Oct 2026 18:08:07 +0200 Subject: [PATCH] fix: only open body stream when it will be read in full --- .../server/src/service/snapshot/defer/base.ts | 14 ++- .../src/service/snapshot/defer/ingest.test.ts | 112 ++++++++++++++++-- .../src/service/snapshot/defer/ingest.ts | 17 +-- .../service/snapshot/defer/object-store.ts | 82 ++++++++++--- 4 files changed, 181 insertions(+), 44 deletions(-) diff --git a/project/server/src/service/snapshot/defer/base.ts b/project/server/src/service/snapshot/defer/base.ts index b4277fa..e75662f 100644 --- a/project/server/src/service/snapshot/defer/base.ts +++ b/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 & AsyncDisposable; + +/** metadata of a deferred snapshot, obtainable without touching its body */ export type SnapshotDeferTargetDeferred = { voucher: SnapshotVoucher; hassVersion: string; hash: SnapshotHash; - snapshot: AsyncIterable; createdAt: Date; - /** releases the handle backing `snapshot` */ - [Symbol.asyncDispose]: () => Promise; + /** the result must be disposed, see {@link SnapshotDeferTargetDeferredSnapshot} */ + open(): Promise; }; export type ISnapshotDeferTarget = { @@ -25,7 +30,6 @@ export type ISnapshotDeferTarget = { hassVersion: string, snapshot: SnapshotRequestTransform, ): Promise; - /** the result must be disposed, see {@link SnapshotDeferTargetDeferred} */ deferred(): Promise>; /** should be called after successful ingestion */ complete(id: Uuid): Promise; diff --git a/project/server/src/service/snapshot/defer/ingest.test.ts b/project/server/src/service/snapshot/defer/ingest.test.ts index 268afa6..231d1d6 100644 --- a/project/server/src/service/snapshot/defer/ingest.test.ts +++ b/project/server/src/service/snapshot/defer/ingest.test.ts @@ -40,14 +40,14 @@ const undeliverable = (): AsyncIterable => ({ }); /** 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 = 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); }); diff --git a/project/server/src/service/snapshot/defer/ingest.ts b/project/server/src/service/snapshot/defer/ingest.ts index 8426189..bd4cfe5 100644 --- a/project/server/src/service/snapshot/defer/ingest.ts +++ b/project/server/src/service/snapshot/defer/ingest.ts @@ -148,7 +148,7 @@ export class SnapshotDeferIngest private async act( target: ISnapshotDeferTarget, ): Promise { - await using deferred = await target.deferred(); + const deferred = await target.deferred(); if (isNone(deferred)) { return "idle"; } @@ -177,10 +177,6 @@ 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"; @@ -188,7 +184,11 @@ export class SnapshotDeferIngest 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); diff --git a/project/server/src/service/snapshot/defer/object-store.ts b/project/server/src/service/snapshot/defer/object-store.ts index d6d1a36..28ae77a 100644 --- a/project/server/src/service/snapshot/defer/object-store.ts +++ b/project/server/src/service/snapshot/defer/object-store.ts @@ -5,6 +5,7 @@ import { setTimeout } from "node:timers/promises"; import { MetadataDirective, NoSuchKey, + NotFound, paginateListObjectsV2, S3, } from "@aws-sdk/client-s3"; @@ -24,7 +25,11 @@ import { } from "../stream"; import type { Uuid } from "../../../type/codec/uuid"; -import type { ISnapshotDeferTarget, SnapshotDeferTargetDeferred } from "./base"; +import type { + ISnapshotDeferTarget, + SnapshotDeferTargetDeferred, + SnapshotDeferTargetDeferredSnapshot, +} from "./base"; const logger = parentLogger.child({ label: "snapshot-defer-target-object-store", @@ -40,6 +45,16 @@ class SnapshotDeferTargetObjectStoreTooFewTargetsError extends Error { } } +class SnapshotDeferTargetObjectStoreMissingBodyError extends Error { + constructor(key: string) { + super(`object <${key}> was fetched without a body`); + Object.setPrototypeOf( + this, + SnapshotDeferTargetObjectStoreMissingBodyError.prototype, + ); + } +} + class NdJsonEncodeTransform extends Transform { constructor() { super({ readableObjectMode: false, writableObjectMode: true }); @@ -282,23 +297,20 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { let object; try { - object = await this.targets[0].s3.getObject({ + // the body is left untouched until a consumer commits to it + object = await this.targets[0].s3.headObject({ Bucket: this.targets[0].bucket, Key: descriptor.Key, }); } catch (e) { // should not happen when there is only one consumer, but happened regardless (during load testing with seaweedfs) - if (e instanceof NoSuchKey) { + if (e instanceof NotFound || e instanceof NoSuchKey) { continue; } throw e; } - if (typeof object.Body === "undefined") { - continue; - } - inner: { const voucherSerialized = object.Metadata?.voucher; if (typeof voucherSerialized === "undefined") { @@ -361,7 +373,6 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { hassVersion, hash: hashDeserialized, createdAt, - body: object.Body, } as const; break outer; @@ -393,32 +404,67 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { return null; } + const key = picked.key; + + return { + voucher: picked.voucher, + hash: picked.hash, + hassVersion: picked.hassVersion, + createdAt: picked.createdAt, + open: () => this.open(key), + }; + } + + private async open( + key: string, + ): Promise { + // the sdk layers wrappers of its own on top of the response (`ChecksumStream` when the object carries `x-amz-checksum-*`) + // those link to their source with `pipe`, which does not propagate `destroy` backwards + // aborting the request is the only teardown that reliably reaches the socket + const abort = new AbortController(); + + const object = await this.targets[0].s3.getObject( + { Bucket: this.targets[0].bucket, Key: key }, + { abortSignal: abort.signal }, + ); + + if (typeof object.Body === "undefined") { + throw new SnapshotDeferTargetObjectStoreMissingBodyError(key); + } + + // a consumer can deliberately walk away mid-stream, which should not cause a warning + let disposing = false; + const chained = pipeline( - Readable.fromWeb(picked.body.transformToWebStream()), + Readable.fromWeb(object.Body.transformToWebStream()), new LineSplitTransform(), new NdJsonDecodeTransform(), new SnapshotPersistedTransform(), (err) => { - if (err !== null && typeof err !== "undefined") { + if (err !== null && typeof err !== "undefined" && !disposing) { logger.warn("deferred snapshot stream terminated early", { - key: picked.key, + key, message: err.message, }); } }, ); - return { - voucher: picked.voucher, - hash: picked.hash, - hassVersion: picked.hassVersion, - createdAt: picked.createdAt, - snapshot: chained, + // `Readable` already carries an async disposer, but it only destroys, which leaves the request hanging + return Object.assign(chained, { [Symbol.asyncDispose]: async () => { + // an in flight failure has already run the pipeline callback by now, so this only ever mutes the teardown that follows + disposing = true; + + // aborting alone releases the socket, but leaves the pipeline running forever + // destroying alone tears down the pipeline, but leaves the socket checked out + abort.abort(); + if (chained.destroyed) { return; } + // teardown of pipeline const closed = new Promise((resolve) => { chained.once("close", resolve); }); @@ -430,7 +476,7 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { // resolves once teardown has propagated back await closed; }, - }; + }); } async complete(id: Uuid): Promise {