fix(tts): restart ChunkedStream retries under fresh request IDs#1994
fix(tts): restart ChunkedStream retries under fresh request IDs#1994rosetta-livekit-bot[bot] wants to merge 1 commit into
Conversation
🦋 Changeset detectedLatest commit: 716bd2b The changes in this PR will be included in the next version bump. This PR includes changesets to release 36 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}
There was a problem hiding this comment.
🔴 Closing a chunked TTS stream can crash with an unhandled exception when buffered audio is still being drained
The metrics queue is not closed when the stream is torn down (close() at agents/src/tts/tts.ts:805), so the metrics loop keeps running and tries to write to the already-closed output queue (this.output.put(audio) at agents/src/tts/tts.ts:753), which throws "Queue is closed".
Impact: Calling close() on a chunked TTS stream while audio is still being forwarded causes an unhandled promise rejection that can crash the process.
Mechanism: close() leaves #metricsQueue open while output is already closed
Before this PR, close() closed this.queue, which was the same queue monitorMetrics() iterated. So closing the stream immediately terminated the metrics loop.
After this PR, monitorMetrics() iterates this.#metricsQueue (line 751), but close() at line 805-810 only closes this.queue (the current per-attempt queue) and this.output. It does NOT close this.#metricsQueue.
The sequence that triggers the crash:
close()is called →this.outputis closed (line 807)drainAttemptQueue()(line 632-639) is still running concurrently, forwarding remaining items from the attempt queue into#metricsQueuemonitorMetrics()picks up those items from#metricsQueueand callsthis.output.put(audio)at line 753AsyncIterableQueue.put()(agents/src/utils.ts:327-331) checksthis.#closedand throwsError('Queue is closed')- This exception is thrown inside the
monitorMetrics()promise which is fire-and-forget (called at line 623), resulting in an unhandled promise rejection
Even without the crash, #metricsQueue not being closed means monitorMetrics hangs until mainTask eventually finishes and its finally block closes #metricsQueue (line 629), which is a resource leak compared to the pre-PR behavior.
(Refers to lines 805-810)
Was this helpful? React with 👍 or 👎 to provide feedback.
| const shouldRetry = | ||
| error.retryable && this._connOptions.maxRetry > 0 && i < this._connOptions.maxRetry; | ||
|
|
||
| if (this._connOptions.maxRetry === 0 || !error.retryable) { | ||
| if (!shouldRetry) { | ||
| this.emitError({ error, recoverable: false }); | ||
| throw error; | ||
| } else if (i === this._connOptions.maxRetry) { | ||
| this.emitError({ error, recoverable: false }); | ||
| throw new APIConnectionError({ | ||
| message: `failed to generate TTS completion after ${this._connOptions.maxRetry + 1} attempts`, | ||
| options: { retryable: false }, | ||
| }); | ||
| } else { | ||
| // Don't emit error event for recoverable errors during retry loop | ||
| // to avoid ERR_UNHANDLED_ERROR or premature session termination | ||
| this.logger.warn( | ||
| { tts: this.#tts.label, attempt: i + 1, error }, | ||
| `failed to generate TTS completion, retrying in ${retryInterval}ms`, | ||
| ); | ||
| } |
There was a problem hiding this comment.
🔍 Retry exhaustion now throws the original error instead of a wrapped APIConnectionError
The old code had a dedicated branch for i === maxRetry that threw new APIConnectionError({ message: 'failed to generate TTS completion after N attempts', options: { retryable: false } }). The new shouldRetry logic at agents/src/tts/tts.ts:679-684 collapses this into a single !shouldRetry check that throws the original APIError. This changes the error type that callers receive on retry exhaustion from APIConnectionError to APIError. Meanwhile, SynthesizeStream._mainTaskImpl at agents/src/tts/tts.ts:284-289 still uses the old pattern with the APIConnectionError wrapper. If any caller distinguishes between these error types (e.g., FallbackChunkedStream.run() at agents/src/tts/fallback_adapter.ts:386 checks instanceof APIError || instanceof APIConnectionError), the behavioral change could matter. Worth confirming this inconsistency between ChunkedStream and SynthesizeStream is intentional.
Was this helpful? React with 👍 or 👎 to provide feedback.
Summary
Testing
pnpm build:agentspnpm test -- agents/src/ttspnpm exec prettier --check agents/src/tts/tts.ts .changeset/fresh-tts-chunked-retries.mdcue-clitext-mode runtime validation with a temporary local agent: asserted an assistantconversation_item_addedframework event.Ported from livekit/agents#6346
Original PR description
Problem
test_tts_synthesize[deepgram]failed on main (run): a transient connection error mid-response causedChunkedStreamto retry after partial audio had already been emitted, so the consumer received frames from two attempts with mixedrequest_ids — and frames from the failed attempt could still trickle out after the retry started, interleaving the two attempts.Changes
ChunkedStream._main_task: settle the emitter (aclose) before retrying, so no stale frames from the failed attempt are delivered once the retry begins. The retry restarts synthesis under a freshrequest_id, signaling downstream that any partial audio from the failed attempt is stale. Also skip retries for non-retryable errors (matchingSynthesizeStream).tests/test_tts.py:_do_synthesisnow accepts a request_id restart — it asserts request_ids don't interleave between attempts and validates completeness/audio on the final attempt's frames. New testtest_tts_synthesize_retry_after_partial_audiocovers fail-once-then-succeed: exactly one retry, recoverable error event, two non-interleaved request_ids, full audio from the retry, one metrics event.tests/fake_tts.py:fake_exception_countoption to fail the first N attempts, and yield control between pushes so partial audio reaches the consumer like a real provider streaming over the network.Notes
SynthesizeStreamstill refuses to retry after partial audio (#5242); making it restart under a fresh request_id like this would be a follow-up.