diff --git a/.changeset/stream-channel-downstream-cancel.md b/.changeset/stream-channel-downstream-cancel.md new file mode 100644 index 0000000000..d4fcdaa936 --- /dev/null +++ b/.changeset/stream-channel-downstream-cancel.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Stream channels: tolerate the consumer cancelling the readable side. Fire-and-forget write() and close() on a downstream-cancelled channel no longer surface as unhandled promise rejections (`Error: undefined`), the `closed` getter now reflects downstream cancellation, and the audio forwarder cancels its reader with a real reason. diff --git a/agents/src/log.ts b/agents/src/log.ts index 4bece4b564..f60f5301c3 100644 --- a/agents/src/log.ts +++ b/agents/src/log.ts @@ -5,7 +5,7 @@ import { Writable } from 'node:stream'; import type { DestinationStream, Logger } from 'pino'; import { multistream, pino } from 'pino'; import { build as pinoPretty } from 'pino-pretty'; -import { type LoggerOptions, log, loggerOptions, setLoggerState } from './log_core.js'; +import { type LoggerOptions, log, loggerOptions, setLoggerState, tryLog } from './log_core.js'; import { type PinoLogObject, emitToOtel } from './telemetry/pino_otel_transport.js'; const OTEL_ENABLED_KEY = Symbol.for('@livekit/agents:otelEnabled'); @@ -19,7 +19,7 @@ const globals = globalThis as typeof globalThis & GlobalState; // LiveKit Cloud injects this into deployed agents. Child processes inherit it. const deployedRegion = process.env.LIVEKIT_REGION_NAME || undefined; -export { log, loggerOptions, type LoggerOptions }; +export { log, loggerOptions, tryLog, type LoggerOptions }; const createLogger = ({ pretty, level }: LoggerOptions): Logger => { const logLevel = level || 'info'; diff --git a/agents/src/log_core.ts b/agents/src/log_core.ts index 16f71a3e0f..7fa6d43695 100644 --- a/agents/src/log_core.ts +++ b/agents/src/log_core.ts @@ -34,6 +34,12 @@ export const log = (): Logger => { return logger; }; +/** + * Like {@link log}, but returns undefined before initializeLogger() has run. + * For best-effort logging from code that may run outside a worker. + */ +export const tryLog = (): Logger | undefined => globals[LOGGER_KEY]; + /** @internal */ export const setLoggerState = (logger: Logger, options: LoggerOptions): void => { globals[LOGGER_OPTIONS_KEY] = options; diff --git a/agents/src/stream/stream_channel.test.ts b/agents/src/stream/stream_channel.test.ts index 7acad4e93d..33d0a6b9ad 100644 --- a/agents/src/stream/stream_channel.test.ts +++ b/agents/src/stream/stream_channel.test.ts @@ -163,4 +163,40 @@ describe('StreamChannel', () => { const result3 = await read3; expect(result3.done).toBe(true); }); + + it('should not produce unhandled rejections when the reader cancels mid-stream', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + const channel = createStreamChannel(); + const reader = channel.stream().getReader(); + + await channel.write('before'); + await reader.read(); + await reader.cancel(); + + // Fire-and-forget write and close after the consumer is gone, as + // realtime producers do. + channel.write('after-cancel'); + await channel.close(); + + await new Promise((resolve) => setTimeout(resolve, 20)); + expect(unhandled).toEqual([]); + expect(channel.closed).toBe(true); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + + it('should still reject an awaited write after the reader cancelled', async () => { + const channel = createStreamChannel(); + const reader = channel.stream().getReader(); + + await reader.cancel(new Error('consumer gone')); + + await expect(channel.write('late')).rejects.toThrow('consumer gone'); + }); }); diff --git a/agents/src/stream/stream_channel.ts b/agents/src/stream/stream_channel.ts index edaeaa8569..7765a0ef6f 100644 --- a/agents/src/stream/stream_channel.ts +++ b/agents/src/stream/stream_channel.ts @@ -2,6 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import type { ReadableStream } from 'node:stream/web'; +import { tryLog } from '../log.js'; import { IdentityTransform } from './identity_transform.js'; export interface StreamChannel { @@ -18,8 +19,29 @@ export function createStreamChannel(): StreamChannel const writer = transform.writable.getWriter(); let isClosed = false; + // The consumer side can cancel the readable (or the stream can error), which + // errors the writable and would otherwise reject every later write() and + // close() with the cancel reason - surfacing as unhandled rejections from + // fire-and-forget producers. + writer.closed.catch((error: unknown) => { + if (!isClosed) { + isClosed = true; + tryLog()?.debug({ error }, 'stream channel writable errored or was cancelled downstream'); + } + }); + return { - write: (chunk: T) => writer.write(chunk), + write: (chunk: T) => { + const result = writer.write(chunk); + // Mark the rejection as handled for fire-and-forget producers; callers + // that await the returned promise still observe it. + result.catch((error: unknown) => { + if (!isClosed) { + tryLog()?.debug({ error }, 'stream channel write failed'); + } + }); + return result; + }, stream: () => transform.readable, abort: async (error: E) => { if (isClosed) return; @@ -53,16 +75,15 @@ export function createStreamChannel(): StreamChannel }, close: async () => { try { - const result = await writer.close(); - isClosed = true; - return result; + return await writer.close(); } catch (e) { - if (e instanceof Error && e.name === 'TypeError') { - // Ignore error if the stream is already closed - isClosed = true; - return; + // Ignore error if the stream is already closed or errored - either way + // no more data flows, which is what close() asks for. + if (!isClosed && !(e instanceof Error && e.name === 'TypeError')) { + tryLog()?.debug({ error: e }, 'stream channel close failed'); } - throw e; + } finally { + isClosed = true; } }, get closed() { diff --git a/agents/src/voice/generation.ts b/agents/src/voice/generation.ts index f5432932b9..af859eb692 100644 --- a/agents/src/voice/generation.ts +++ b/agents/src/voice/generation.ts @@ -1049,7 +1049,9 @@ async function forwardAudio( const reader = ttsStream.getReader(); let resampler: AudioResampler | null = null; const cancelReader = () => { - void reader.cancel().catch((error) => { + // Cancel with a reason so producers still writing into the stream see a + // real error instead of `undefined`. + void reader.cancel(new Error('audio forwarding aborted')).catch((error) => { logger.debug({ error }, 'failed to cancel TTS stream reader after abort'); }); };