From 7d438683939f96c32b7d5e8b124e2bbd25835d34 Mon Sep 17 00:00:00 2001 From: Philip Trauner Date: Mon, 12 Oct 2026 14:55:06 +0200 Subject: [PATCH] fix: prevent ingestion stall on invalid `content-length` 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 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 --- .../server/src/service/snapshot/defer/base.ts | 4 + .../src/service/snapshot/defer/ingest.test.ts | 282 +++++++++++++++++- .../src/service/snapshot/defer/ingest.ts | 238 +++++++++------ .../service/snapshot/defer/object-store.ts | 36 ++- .../src/utility/line-split-transform.test.ts | 175 +++++++++++ .../src/utility/line-split-transform.ts | 46 +++ 6 files changed, 681 insertions(+), 100 deletions(-) create mode 100644 project/server/src/utility/line-split-transform.test.ts create mode 100644 project/server/src/utility/line-split-transform.ts diff --git a/project/server/src/service/snapshot/defer/base.ts b/project/server/src/service/snapshot/defer/base.ts index 6cdb9d60..b4277fae 100644 --- a/project/server/src/service/snapshot/defer/base.ts +++ b/project/server/src/service/snapshot/defer/base.ts @@ -8,12 +8,15 @@ import type { SnapshotRequestTransformOut, } from "../stream"; +/** holds handle, dispose to prevent handle leakage when not consuming */ export type SnapshotDeferTargetDeferred = { voucher: SnapshotVoucher; hassVersion: string; hash: SnapshotHash; snapshot: AsyncIterable; createdAt: Date; + /** releases the handle backing `snapshot` */ + [Symbol.asyncDispose]: () => Promise; }; export type ISnapshotDeferTarget = { @@ -22,6 +25,7 @@ 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 22bb63a5..e4728437 100644 --- a/project/server/src/service/snapshot/defer/ingest.test.ts +++ b/project/server/src/service/snapshot/defer/ingest.test.ts @@ -1,20 +1,142 @@ import { type TestContext, test } from "node:test"; +import { logger } from "../../../logger"; import { StubIntrospection } from "../../introspect/stub"; import { SuspendableHandle } from "../../suspendable"; +import { Voucher } from "../../voucher"; import { SnapshotDeferIngest } from "./ingest"; -import type { ISnapshot } from ".."; -import type { ISnapshotDeferTarget } from "./base"; +import type { Uuid } from "../../../type/codec/uuid"; +import type { ISnapshot, SnapshotHash } from ".."; +import type { SnapshotRequestTransformOut } from "../stream"; +import type { ISnapshotDeferTarget, SnapshotDeferTargetDeferred } from "./base"; -const buildIngest = (target: ISnapshotDeferTarget) => { +logger.silent = true; + +const buildIngest = (target: ISnapshotDeferTarget, snapshot?: ISnapshot) => { return new SnapshotDeferIngest( - {} as ISnapshot, + snapshot ?? ({} as ISnapshot), target, new StubIntrospection(), ); }; +const id = "00000000-0000-4000-8000-000000000000" as Uuid; +const sub = "00000000-0000-4000-8000-000000000001" as Uuid; + +/** a tripwire iterable, consumption means the handle is held for longer than the test assumes */ +const nonconsumable = (): AsyncIterable => ({ + [Symbol.asyncIterator]: () => ({ + next: () => + Promise.reject(new Error("snapshot must not be consumed in this test")), + }), +}); + +/** an iterable that never delivers, mirroring an s3 object the client keeps open forever */ +const undeliverable = (): AsyncIterable => ({ + [Symbol.asyncIterator]: () => ({ + next: () => new Promise(() => {}), + }), +}); + +/** forges handle */ +const buildHandle = () => + new Proxy({}, { get: () => ({ kind: "attachable-hashed" }) }); + +/** a deferred snapshot that records whether it was disposed */ +const buildDeferred = ( + snapshot: AsyncIterable = nonconsumable(), +) => { + const state = { disposed: 0 }; + + const deferred: SnapshotDeferTargetDeferred = { + voucher: new Voucher("signing-key").create( + "snapshot-submission", + new Date(0), + { id, sub }, + ), + hassVersion: "2024.1.0", + hash: {} as SnapshotHash, + createdAt: new Date(0), + snapshot, + [Symbol.asyncDispose]: async () => { + state.disposed += 1; + }, + }; + + return { deferred, state }; +}; + +/** a target handing out `deferred` exactly once, recording terminal calls */ +const buildTarget = (deferred: SnapshotDeferTargetDeferred) => { + const state = { completed: [] as Uuid[], archived: [] as Uuid[] }; + + let handed = false; + const target = { + async deferred() { + if (handed) { + return null; + } + handed = true; + + return deferred; + }, + async complete(completed: Uuid) { + state.completed.push(completed); + }, + async archive(archived: Uuid) { + state.archived.push(archived); + }, + } as ISnapshotDeferTarget; + + return { target, state }; +}; + +/** snapshot service handing out an attachable handle */ +const buildSnapshot = () => { + const state = { + attached: [] as string[], + finalized: [] as string[], + deleted: [] as Uuid[], + }; + + const snapshot = { + voucher: { + expiresAt: () => new Date(0), + }, + create: async () => ({ kind: "success", handle: buildHandle() }), + finalize: async (_handle: unknown, hassVersion: string) => { + state.finalized.push(hassVersion); + }, + delete: async (deleted: Uuid) => { + state.deleted.push(deleted); + }, + attach: { + device: async (_handle: unknown, integration: string) => { + state.attached.push(`device:${integration}`); + }, + entity: async (_handle: unknown, integration: string) => { + state.attached.push(`entity:${integration}`); + }, + }, + } as unknown as ISnapshot; + + return { snapshot, state }; +}; + +const device = { + integration: "hue", + device: {}, + entities: [], +} as unknown as SnapshotRequestTransformOut; +const entity = { + integration: "mqtt", + entity: {}, +} as unknown as SnapshotRequestTransformOut; + +/** drains the microtask queue, letting the ingest loop reach its next real suspension point */ +const settled = () => new Promise((resolve) => setImmediate(resolve)); + test("suspend", async (t: TestContext) => { // use a deferred target that blocks on `deferred()` to control timing let entered: Promise | undefined; @@ -84,3 +206,155 @@ test("suspend", async (t: TestContext) => { value: "idle", }); }); + +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 iterated + 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", + }); + + // handle must be released before the step is handed to the consumer + t.assert.strictEqual(state.disposed, 1); + t.assert.deepStrictEqual(calls.completed, [id]); + }); + + await t.test("failed ingestion", async (t: TestContext) => { + const { deferred, state } = buildDeferred(); + const { target } = buildTarget(deferred); + + const snapshot = { + voucher: { + expiresAt: () => new Date(0), + }, + create: async () => { + throw new Error("error"); + }, + } as unknown as ISnapshot; + + const iterator = buildIngest(target, snapshot) + .ingest() + [Symbol.asyncIterator](); + + await t.assert.rejects(iterator.next(), /error/); + + t.assert.strictEqual(state.disposed, 1); + }); +}); + +test("consume", async (t: TestContext) => { + await t.test("attaches every item", async (t: TestContext) => { + const { deferred, state } = buildDeferred( + (async function* () { + yield device; + yield entity; + })(), + ); + const { target, state: calls } = buildTarget(deferred); + 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.attached, ["device:hue", "entity:mqtt"]); + t.assert.deepStrictEqual(recorded.finalized, ["2024.1.0"]); + t.assert.deepStrictEqual(recorded.deleted, []); + t.assert.deepStrictEqual(calls.completed, [id]); + t.assert.strictEqual(state.disposed, 1); + }); + + await t.test( + "gives up on stream that never delivers", + async (t: TestContext) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + + const { deferred, state } = buildDeferred(undeliverable()); + 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.mock.timers.tick(30_000); + + t.assert.deepStrictEqual(await step, { done: false, value: "acted" }); + + 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( + "drops deadline once a chunk arrives", + async (t: TestContext) => { + t.mock.timers.enable({ apis: ["setTimeout"] }); + + let release: (() => void) | undefined; + const { deferred } = buildDeferred( + (async function* () { + yield entity; + await new Promise((resolve) => { + release = resolve; + }); + yield device; + })(), + ); + 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"]); + + // beyond the deadline + t.mock.timers.tick(30_000 * 10); + await settled(); + + t.assert.deepStrictEqual(calls.completed, []); + t.assert.deepStrictEqual(recorded.deleted, []); + + release?.(); + + t.assert.deepStrictEqual(await step, { done: false, value: "acted" }); + t.assert.deepStrictEqual(recorded.attached, [ + "entity:mqtt", + "device:hue", + ]); + 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 bfd807b5..6ea4b131 100644 --- a/project/server/src/service/snapshot/defer/ingest.ts +++ b/project/server/src/service/snapshot/defer/ingest.ts @@ -3,6 +3,7 @@ import { createType, inject } from "@lppedd/di-wise-neo"; import { logger as parentLogger } from "../../../logger"; import { isNone, isSome } from "../../../type/maybe"; import { injectOrStub } from "../../../utility/dependency-injection"; +import race from "../../../utility/race-as-promised"; import { IIntrospection, type IntrospectionMetricCounter, @@ -10,11 +11,17 @@ import { import { StubIntrospection } from "../../introspect/stub"; import { type ISuspendable, Suspendable } from "../../suspendable"; import { Voucher } from "../../voucher"; -import { ISnapshot, Snapshot } from ".."; +import { ISnapshot, Snapshot, type SnapshotHandleAttachable } from ".."; import { ISnapshotDeferTarget } from "./base"; +import type { SnapshotRequestTransformOut } from "../stream"; + type SnapshotDeferIngestIngestStep = "idle" | "acted"; +type ConsumeResult = { + stalled: boolean; +}; + export interface ISnapshotDeferIngest extends ISuspendable { ingest(): AsyncIterable; } @@ -104,7 +111,8 @@ export class SnapshotDeferIngest } async *ingest(): AsyncIterable { - if (typeof this.snapshotDeferTarget === "undefined") { + const target = this.snapshotDeferTarget; + if (typeof target === "undefined") { return; } @@ -126,101 +134,155 @@ export class SnapshotDeferIngest this.acting = true; try { - const deferred = await this.snapshotDeferTarget?.deferred(); - if (isNone(deferred)) { - yield "idle"; + yield await this.act(target); + } finally { + this.acting = false; + this.settle(); + } + } + } finally { + this.ingesting = false; + } + } - continue; - } + private async act( + target: ISnapshotDeferTarget, + ): Promise { + await using deferred = await target.deferred(); + if (isNone(deferred)) { + return "idle"; + } - const { id, sub, at } = Voucher.peek(deferred.voucher); + const { id, sub, at } = Voucher.peek(deferred.voucher); - let completed = false; - const created = await this.snapshot.create( - deferred.voucher, - deferred.hash, - deferred.createdAt, - ); - if (created.kind !== "success") { - logger.warn("handle acquisition failed for deferred ingest", { + let completed = false; + const created = await this.snapshot.create( + deferred.voucher, + deferred.hash, + deferred.createdAt, + ); + if (created.kind !== "success") { + logger.warn("handle acquisition failed for deferred ingest", { + id, + sub, + reason: created.reason, + hassVersion: deferred.hassVersion, + wantedAt: at, + persistedAt: deferred.createdAt, + expiresAt: this.snapshot.voucher.expiresAt(deferred.voucher), + }); + + this.metrics.handleAcquisitionFailure.increment({ + version: deferred.hassVersion, + reason: created.reason, + }); + + await target.complete(id); + + return "acted"; + } + + try { + if (!Snapshot.isDuplicate(created.handle)) { + const consumed = await this.consume(created.handle, deferred.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 + // 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 + if (consumed.stalled) { + logger.warn( + `gave up on stalled deferred ingest <${id}> by <${sub}>`, + { id, sub, - reason: created.reason, hassVersion: deferred.hassVersion, - wantedAt: at, persistedAt: deferred.createdAt, - expiresAt: this.snapshot.voucher.expiresAt(deferred.voucher), - }); - - this.metrics.handleAcquisitionFailure.increment({ - version: deferred.hassVersion, - reason: created.reason, - }); - - // TODO: figure out alternative to fully consuming that doesn't slowly leak handles - for await (const _ of deferred.snapshot) { - } - - await this.snapshotDeferTarget.complete(id); - - yield "acted"; - continue; - } - - try { - if (!Snapshot.isDuplicate(created.handle)) { - for await (const item of deferred.snapshot) { - if ("device" in item) { - await this.snapshot.attach.device( - created.handle, - item.integration, - item.device, - item.entities, - ); - } else { - await this.snapshot.attach.entity( - created.handle, - item.integration, - item.entity, - ); - } - } - } 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); - - await this.snapshotDeferTarget.complete(id); - completed = true; - - logger.info(`ingested <${id}> by <${sub}>`, { id, sub }); - } catch (err) { - logger.error("ingestion failure", { - message: - typeof err === "object" && isSome(err) && "message" in err - ? err.message - : "unknown error", - }); - console.error(err); - - if (!completed) { - await this.snapshotDeferTarget.archive(id); - } - - await this.snapshot.delete(id); - } - - yield "acted"; - } finally { - this.acting = false; - this.settle(); + }, + ); + + await target.complete(id); + await this.snapshot.delete(id); + + return "acted"; } } - } finally { - this.ingesting = false; + + await this.snapshot.finalize(created.handle, deferred.hassVersion); + + await target.complete(id); + completed = true; + + logger.info(`ingested <${id}> by <${sub}>`, { id, sub }); + } catch (err) { + logger.error("ingestion failure", { + message: + typeof err === "object" && isSome(err) && "message" in err + ? err.message + : "unknown error", + }); + console.error(err); + + if (!completed) { + await target.archive(id); + } + + await this.snapshot.delete(id); } + + return "acted"; + } + + private async consume( + handle: SnapshotHandleAttachable, + snapshot: AsyncIterable, + ): Promise { + const iterator = snapshot[Symbol.asyncIterator](); + + 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; + { + const { resolve, promise: stalled } = + Promise.withResolvers(); + const deadline = setTimeout(() => resolve(StalledSymbol), 30_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]); + } 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) { + const item = next.value; + if ("device" in item) { + await this.snapshot.attach.device( + handle, + item.integration, + item.device, + item.entities, + ); + } else { + await this.snapshot.attach.entity( + handle, + item.integration, + item.entity, + ); + } + + next = await iterator.next(); + } + + return { stalled: false }; } } diff --git a/project/server/src/service/snapshot/defer/object-store.ts b/project/server/src/service/snapshot/defer/object-store.ts index 618ea836..d6d1a36f 100644 --- a/project/server/src/service/snapshot/defer/object-store.ts +++ b/project/server/src/service/snapshot/defer/object-store.ts @@ -1,4 +1,3 @@ -import { createInterface } from "node:readline/promises"; import type { TransformCallback } from "node:stream"; import { pipeline, Readable, Transform } from "node:stream"; import { setTimeout } from "node:timers/promises"; @@ -16,6 +15,7 @@ import { Schema } from "effect/index"; import { ConfigProvider } from "../../../config"; import { logger as parentLogger } from "../../../logger"; import { isNone, type Maybe } from "../../../type/maybe"; +import { LineSplitTransform } from "../../../utility/line-split-transform"; import { Voucher } from "../../voucher"; import { ISnapshot, Snapshot, type SnapshotVoucher } from ".."; import { @@ -356,6 +356,7 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { } picked = { + key: descriptor.Key, voucher: voucherDeserialized.voucher, hassVersion, hash: hashDeserialized, @@ -392,16 +393,19 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { return null; } - const rl = createInterface({ - input: Readable.fromWeb(picked.body.transformToWebStream()), - crlfDelay: Infinity, - }); - const chained = pipeline( - Readable.from(rl), + Readable.fromWeb(picked.body.transformToWebStream()), + new LineSplitTransform(), new NdJsonDecodeTransform(), new SnapshotPersistedTransform(), - () => {}, + (err) => { + if (err !== null && typeof err !== "undefined") { + logger.warn("deferred snapshot stream terminated early", { + key: picked.key, + message: err.message, + }); + } + }, ); return { @@ -410,6 +414,22 @@ export class SnapshotDeferTargetObjectStore implements ISnapshotDeferTarget { hassVersion: picked.hassVersion, createdAt: picked.createdAt, snapshot: chained, + [Symbol.asyncDispose]: async () => { + if (chained.destroyed) { + return; + } + + const closed = new Promise((resolve) => { + chained.once("close", resolve); + }); + + // an in flight failure is already reported by the pipeline callback, it must not resurface here + chained.once("error", () => {}); + chained.destroy(); + + // resolves once teardown has propagated back + await closed; + }, }; } diff --git a/project/server/src/utility/line-split-transform.test.ts b/project/server/src/utility/line-split-transform.test.ts new file mode 100644 index 00000000..b4f417d7 --- /dev/null +++ b/project/server/src/utility/line-split-transform.test.ts @@ -0,0 +1,175 @@ +import { pipeline, Readable, Transform } from "node:stream"; +import { type TestContext, test } from "node:test"; +import { setTimeout as delay } from "node:timers/promises"; + +import { LineSplitTransform } from "./line-split-transform"; + +/** feeds `chunks` through the transform and collects every pushed line */ +const split = async ( + chunks: readonly (string | Buffer)[], +): Promise => { + const transform = new LineSplitTransform(); + + for (const chunk of chunks) { + transform.write( + typeof chunk === "string" ? Buffer.from(chunk, "utf8") : chunk, + ); + } + transform.end(); + + const lines: string[] = []; + for await (const line of transform) { + lines.push(line as string); + } + + return lines; +}; + +/** slices `source` into `size` *byte* chunks, so boundaries can land mid-line and mid-code-point */ +const chunked = (source: string, size: number): Buffer[] => { + const buffer = Buffer.from(source, "utf8"); + + const chunks: Buffer[] = []; + for (let index = 0; index < buffer.byteLength; index += size) { + chunks.push(buffer.subarray(index, index + size)); + } + + return chunks; +}; + +test("line split transform", async (t: TestContext) => { + await t.test("empty input yields no lines", async (t: TestContext) => { + t.assert.deepStrictEqual(await split([]), []); + t.assert.deepStrictEqual(await split([""]), []); + }); + + await t.test("splits on newline", async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\nb\nc\n"]), ["a", "b", "c"]); + }); + + await t.test( + "a trailing fragment without a terminator is still a line", + async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\nb\nc"]), ["a", "b", "c"]); + }, + ); + + await t.test( + "a trailing terminator yields no empty line", + async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\n"]), ["a"]); + }, + ); + + await t.test( + "strips the carriage return of a crlf", + async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\r\nb\r\n"]), ["a", "b"]); + }, + ); + + await t.test( + "a lone carriage return is not a terminator", + async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\rb\n"]), ["a\rb"]); + }, + ); + + await t.test("preserves interior empty lines", async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\n\nb\n"]), ["a", "", "b"]); + }); + + await t.test("handles a crlf split across chunks", async (t: TestContext) => { + t.assert.deepStrictEqual(await split(["a\r", "\nb\r", "\n"]), ["a", "b"]); + }); + + await t.test("is insensitive to chunk boundaries", async (t: TestContext) => { + const source = "alpha\nbravo\r\ncharlie\n\ndelta"; + const expected = ["alpha", "bravo", "charlie", "", "delta"]; + + for (const size of [1, 2, 3, 5, 7, 64]) { + t.assert.deepStrictEqual( + await split(chunked(source, size)), + expected, + `chunk size ${size}`, + ); + } + }); + + await t.test( + "handles a multi byte code point split across chunks", + async (t: TestContext) => { + // é and 你 and 😀 are 2, 3 and 4 bytes respectively + const source = '{"a":"éè你好😀"}\n{"b":2}\n'; + const expected = ['{"a":"éè你好😀"}', '{"b":2}']; + + for (const size of [1, 2, 3, 5, 7]) { + t.assert.deepStrictEqual( + await split(chunked(source, size)), + expected, + `chunk size ${size}`, + ); + } + }, + ); + + await t.test( + "does not terminate on the separators readline splits on", + async (t: TestContext) => { + // `JSON.stringify` emits u+2028 and u+2029 unescaped, and `readline` treats both as line terminators + const record = { name: "a\u2028b\u2029c" }; + const lines = await split([`${JSON.stringify(record)}\n`]); + + t.assert.strictEqual(lines.length, 1); + t.assert.deepStrictEqual(JSON.parse(lines[0]), record); + }, + ); + + await t.test( + "destroying the pipeline tail destroys the source", + async (t: TestContext) => { + let destroyed = false; + + const source = new Readable({ + read() { + // never ends, so teardown can only come from a destroy + this.push(Buffer.from("line\n", "utf8")); + }, + destroy(error, callback) { + destroyed = true; + callback(error); + }, + }); + + const tail = pipeline( + source, + new LineSplitTransform(), + new Transform({ + objectMode: true, + transform(chunk, _, callback) { + this.push(chunk); + callback(); + }, + }), + () => {}, + ); + + let seen = 0; + for await (const line of tail) { + t.assert.strictEqual(line, "line"); + + seen += 1; + if (seen === 4) { + // `break` invokes the iterator's `return()`, which destroys `tail` + break; + } + } + + await delay(10); + + t.assert.strictEqual(seen, 4); + t.assert.strictEqual(destroyed, true, "source was not destroyed"); + t.assert.strictEqual(source.destroyed, true); + }, + ); +}); diff --git a/project/server/src/utility/line-split-transform.ts b/project/server/src/utility/line-split-transform.ts new file mode 100644 index 00000000..1f91f9a8 --- /dev/null +++ b/project/server/src/utility/line-split-transform.ts @@ -0,0 +1,46 @@ +import type { TransformCallback } from "node:stream"; +import { Transform } from "node:stream"; +import { StringDecoder } from "node:string_decoder"; + +/** splits a byte stream into lines, pushing each one as a string + * + * can safely be dropped mid-stream, as opposed to {@link import("node:readline").createInterface} */ +export class LineSplitTransform extends Transform { + private readonly decoder = new StringDecoder("utf8"); + private buffered = ""; + + constructor() { + super({ readableObjectMode: true, writableObjectMode: false }); + } + + override _transform( + chunk: Buffer, + _: BufferEncoding, + callback: TransformCallback, + ): void { + this.buffered += this.decoder.write(chunk); + + let index = this.buffered.indexOf("\n"); + while (index !== -1) { + const line = this.buffered.slice(0, index); + this.buffered = this.buffered.slice(index + 1); + + this.push(line.endsWith("\r") ? line.slice(0, -1) : line); + + index = this.buffered.indexOf("\n"); + } + + callback(); + } + + /** a trailing fragment without a terminator is still a line */ + override _flush(callback: TransformCallback): void { + this.buffered += this.decoder.end(); + if (this.buffered.length > 0) { + this.push(this.buffered); + this.buffered = ""; + } + + callback(); + } +}