fix(agents): per-iteration abort promise in readStream prevents chunk retention#2047
fix(agents): per-iteration abort promise in readStream prevents chunk retention#2047Saravut-Lin wants to merge 1 commit into
Conversation
… retention readStream() hoisted a single waitForAbort(signal) promise outside its read loop; every ThrowsPromise.race([reader.read(), abortPromise]) registered a new reaction on that long-lived promise, and each reaction kept its iteration's settled read result reachable until the signal aborted. AudioRecognition pumps sttInputStream through readStream(), so pipeline AgentSessions retained every inbound AudioFrame (~100/s, ~5.8 MB/min measured in production). Same mechanism as livekit#1950 (Deepgram plugin pump, fixed in livekit#1957); this is the core-utils instance. Fix mirrors livekit#1957: per-iteration abort promise with removeEventListener cleanup after each race. Adds a WeakRef+gc regression test that fails on the old code (20/20 chunks pinned) plus delivery and abort-interrupt coverage. Fixes livekit#2046 Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
🦋 Changeset detectedLatest commit: cd3db7b The changes in this PR will be included in the next version bump. This PR includes changesets to release 37 packages
Not sure what this means? Click here to learn what changesets are. Click here if you're a maintainer who wants to add another changeset to this PR |
|
|
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
toubatbrian
left a comment
toubatbrian
left a comment
There was a problem hiding this comment.
readStream() still leaves the per-iteration abort listener/reaction reachable when reader.read() rejects, because abortCleanup?.() is only reached after a successful Promise.race().
Please wrap the race in try/finally so cleanup runs when the read resolves, rejects, or abort wins, and add a deterministic rejection-path test that asserts the abort listener is removed. I reproduced the failure and verified that this fix passes agents/src/utils.test.ts (50 passed, 1 skipped), formatting, scoped lint, and the full build.
3 participants
Fixes #2046.
What
readStream()hoisted a singlewaitForAbort(signal)promise outside its read loop and raced it against everyreader.read(). EachThrowsPromise.raceregisters a new reaction on that long-lived promise; since it only settles when the signal aborts (normally stream teardown), every reaction — and the settled read result it transitively holds — stays reachable for the stream's entire lifetime.AudioRecognitionpumpssttInputStreamthroughreadStream(), so pipelineAgentSessions retained every inbound AudioFrame (~100/s), ~5.8 MB/min measured in production (details, heap-snapshot retainer chains, and prod before/after numbers in #2046).Same mechanism as #1950 (the Deepgram plugin's private pump, fixed by #1957); this is the core-utils instance. The fix mirrors #1957's idiom exactly: a per-iteration abort promise with
removeEventListenercleanup after each race, so the signal holds at most one listener at a time and no reaction outlives its iteration. Abort semantics are unchanged (pendingread()still interrupted;if (!result) breakpath identical).Tests
Three cases in
utils.test.ts:listenerCount('abort') ≤ 1.it.skipIfwithout--expose-gc) — 20 consumed chunks tracked viaWeakRef; after GC, at most 1 remains reachable while the stream and signal stay live. On the old code all 20 stay pinned, so this is the test that actually catches the bug — worth noting that a listener-count assertion alone cannot (the old code also held exactly one listener; the retention lives in the promise reaction chain, not the listener list — this also applies to the regression test in fix(deepgram): per-iteration abort promise prevents AudioFrame retention on long-running streams #1957).Verified locally: the retention test fails on
main(20/20 chunks alive) and passes with the fix.Production verification
Shipped as a patch on
@livekit/agents@1.4.5in a production voice agent: per-turnheapUsedwent from 72→158 MB climbing over 26 turns (no plateau, forced GC reclaimed nothing) to 72→85 MB flat over 23 turns; job-process RSS high-water 275 MB vs ~360 MB.Note for maintainers
The hoisted-
waitForAbort-raced-in-a-loop shape also appears in several plugin pumps (e.g.cartesia/src/stt.ts,assemblyai/src/stt.ts,inworld/src/stt.ts,sarvam/src/stt.ts,soniox/src/stt.ts,xai/src/stt.ts,baseten/src/tts.ts). Not all are per-frame loops, but they're the same class as #1950 and may deserve an audit; this PR deliberately stays scoped to the corereadStream().