diff --git a/.changeset/readstream-abort-promise-retention.md b/.changeset/readstream-abort-promise-retention.md new file mode 100644 index 000000000..d07fb7ce3 --- /dev/null +++ b/.changeset/readstream-abort-promise-retention.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Fix `readStream()` retaining every chunk until its abort signal fires: racing `reader.read()` against a single long-lived `waitForAbort` promise accumulated one promise reaction (holding that iteration's chunk) per read, pinning every inbound AudioFrame for the lifetime of STT pipelines (#2046). diff --git a/agents/src/utils.test.ts b/agents/src/utils.test.ts index 0e4657baf..7b9136be7 100644 --- a/agents/src/utils.test.ts +++ b/agents/src/utils.test.ts @@ -13,6 +13,7 @@ import { dedent, delay, isPending, + readStream, resampleStream, toStream, } from '../src/utils.js'; @@ -934,3 +935,91 @@ world }); }); }); + +describe('readStream abort-promise retention (issue #2046)', () => { + // Regression: the old implementation hoisted `waitForAbort(signal)` outside the + // read loop, so every `ThrowsPromise.race([reader.read(), abortPromise])` attached + // a new reaction to that single long-lived promise. The reactions - and the read + // results (stream chunks) they transitively held - accumulated until the signal + // aborted, pinning every AudioFrame ever read on audio streams. + it('delivers all chunks and keeps at most one abort listener', async () => { + const controller = new AbortController(); + let enqueueCtl!: ReadableStreamDefaultController; + const stream = new ReadableStream({ + start(c) { + enqueueCtl = c; + }, + }); + const iterator = readStream(stream, controller.signal)[Symbol.asyncIterator](); + + const CHUNKS = 50; + const received: number[] = []; + for (let i = 0; i < CHUNKS; i++) { + enqueueCtl.enqueue(i); + const { value } = await iterator.next(); + received.push(value as number); + } + expect(received).toEqual(Array.from({ length: CHUNKS }, (_, i) => i)); + + // The signal should never accumulate listeners as chunks flow. + const listenerCount = (controller.signal as unknown as NodeJS.EventEmitter).listenerCount?.( + 'abort', + ); + if (listenerCount !== undefined) { + expect(listenerCount).toBeLessThanOrEqual(1); + } + + controller.abort(); + const end = await iterator.next(); + expect(end.done).toBe(true); + }); + + it('abort interrupts a pending read', async () => { + const controller = new AbortController(); + const stream = new ReadableStream({ start() {} }); // never produces + const received: number[] = []; + const consume = (async () => { + for await (const v of readStream(stream, controller.signal)) { + received.push(v); + } + })(); + await delay(10); + controller.abort(); + await consume; // returns instead of hanging + expect(received).toEqual([]); + }); + + // Only meaningful under `node --expose-gc`; the listener-count check above cannot + // catch the reaction leak (the old code also held a single listener - the retention + // lived in the promise reaction chain, not the listener list). + it.skipIf(typeof globalThis.gc !== 'function')( + 'consumed chunks become collectable while the stream stays open', + async () => { + const controller = new AbortController(); + let enqueueCtl!: ReadableStreamDefaultController<{ payload: Uint8Array }>; + const stream = new ReadableStream<{ payload: Uint8Array }>({ + start(c) { + enqueueCtl = c; + }, + }); + const iterator = readStream(stream, controller.signal)[Symbol.asyncIterator](); + + const refs: WeakRef[] = []; + for (let i = 0; i < 20; i++) { + const chunk = { payload: new Uint8Array(1024) }; + refs.push(new WeakRef(chunk)); + enqueueCtl.enqueue(chunk); + await iterator.next(); + } + + globalThis.gc!(); + await delay(0); + globalThis.gc!(); + const alive = refs.filter((r) => r.deref() !== undefined).length; + // With the hoisted abort promise every chunk stayed reachable (alive === 20). + expect(alive).toBeLessThanOrEqual(1); + + controller.abort(); + }, + ); +}); diff --git a/agents/src/utils.ts b/agents/src/utils.ts index 85e351e9c..f79891240 100644 --- a/agents/src/utils.ts +++ b/agents/src/utils.ts @@ -1387,9 +1387,32 @@ export async function* readStream( const reader = stream.getReader(); try { if (signal) { - const abortPromise = waitForAbort(signal); while (true) { + // Create a fresh per-iteration abort promise with explicit cleanup. + // + // A single `waitForAbort(signal)` hoisted outside the loop leaks: every + // `ThrowsPromise.race([reader.read(), abortPromise])` registers a new + // reaction on the long-lived promise, and because that promise only + // settles when the signal aborts (normally at stream teardown), each + // reaction - and the settled read result it transitively holds - stays + // reachable for the lifetime of the stream. On audio streams this pins + // every AudioFrame ever read (#2046; same mechanism as #1950). + // + // A per-iteration promise + `removeEventListener` after each race means + // the signal holds at most one listener at a time and no reaction + // outlives its iteration. + let abortCleanup: (() => void) | undefined; + const abortPromise = new Promise((resolve) => { + if (signal.aborted) { + resolve(undefined); + return; + } + const handler = () => resolve(undefined); + signal.addEventListener('abort', handler, { once: true }); + abortCleanup = () => signal.removeEventListener('abort', handler); + }); const result = await ThrowsPromise.race([reader.read(), abortPromise]); + abortCleanup?.(); if (!result) { break; }