Skip to content

fix(tts): restart ChunkedStream retries under fresh request IDs #1994

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rosetta-livekit-bot wants to merge 1 commit into
base: main
Choose a base branch
from
+44 −18
Open

fix(tts): restart ChunkedStream retries under fresh request IDs #1994

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/fresh-tts-chunked-retries.md
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
'@livekit/agents': patch
---

Restart chunked TTS retries with a fresh attempt queue so retried synthesis uses a fresh request ID and does not write to a closed failed-attempt queue.
57 changes: 39 additions & 18 deletions agents/src/tts/tts.ts
View file
Open in desktop

@devin-ai-integration devin-ai-integration Bot Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔴 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:

  1. close() is called → this.output is closed (line 807)
  2. drainAttemptQueue() (line 632-639) is still running concurrently, forwarding remaining items from the attempt queue into #metricsQueue
  3. monitorMetrics() picks up those items from #metricsQueue and calls this.output.put(audio) at line 753
  4. AsyncIterableQueue.put() (agents/src/utils.ts:327-331) checks this.#closed and throws Error('Queue is closed')
  5. 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)

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Original file line number Diff line number Diff line change
Expand Up @@ -601,6 +601,7 @@ export abstract class ChunkedStream implements AsyncIterableIterator<Synthesized
#inputTokens = 0;
#outputTokens = 0;
#startedTime: number;
#metricsQueue = new AsyncIterableQueue<SynthesizedAudio>();

protected abortController = new AbortController();

Expand All @@ -625,7 +626,17 @@ export abstract class ChunkedStream implements AsyncIterableIterator<Synthesized
// is run **after** the constructor has finished. Otherwise we get
// runtime error when trying to access class variables in the
// `run` method.
ThrowsPromise.resolve().then(() => this.mainTask().finally(() => this.queue.close()));
ThrowsPromise.resolve().then(() => this.mainTask().finally(() => this.#metricsQueue.close()));
}

private drainAttemptQueue(attemptQueue: AsyncIterableQueue<SynthesizedAudio>): Promise<void> {
return ThrowsPromise.resolve().then(async () => {
for await (const audio of attemptQueue) {
if (!this.#metricsQueue.closed) {
this.#metricsQueue.put(audio);
}
}
});
}

private _mainTaskImpl = async (span: Span) => {
Expand All @@ -636,8 +647,12 @@ export abstract class ChunkedStream implements AsyncIterableIterator<Synthesized
});

for (let i = 0; i < this._connOptions.maxRetry + 1; i++) {
const attemptQueue = new AsyncIterableQueue<SynthesizedAudio>();
this.queue = attemptQueue;
const drainAttemptQueue = this.drainAttemptQueue(attemptQueue);

try {
return await tracer.startActiveSpan(
const result = await tracer.startActiveSpan(
async (attemptSpan) => {
attemptSpan.setAttribute(traceTypes.ATTR_RETRY_COUNT, i);
try {
Expand All @@ -649,28 +664,34 @@ export abstract class ChunkedStream implements AsyncIterableIterator<Synthesized
},
{ name: 'tts_request_run' },
);
if (!attemptQueue.closed) {
attemptQueue.close();
}
await drainAttemptQueue;
return result;
} catch (error) {
if (!attemptQueue.closed) {
attemptQueue.close();
}
await drainAttemptQueue;

if (error instanceof APIError) {
const retryInterval = intervalForRetry(this._connOptions, i);
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`,
);
}
Comment on lines +679 to 685

@devin-ai-integration devin-ai-integration Bot Jul 8, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔍 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.

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.


const retryInterval = intervalForRetry(this._connOptions, i);
// 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`,
);

if (retryInterval > 0) {
await delay(retryInterval);
}
Expand Down Expand Up @@ -727,7 +748,7 @@ export abstract class ChunkedStream implements AsyncIterableIterator<Synthesized
let ttfb: bigint = BigInt(-1);
let requestId = '';

for await (const audio of this.queue) {
for await (const audio of this.#metricsQueue) {
audio.frame.userdata[USERDATA_TTS_STARTED_TIME] = this.#startedTime;
this.output.put(audio);
requestId = audio.requestId;
Expand Down
Loading