From 8313f2aee33ecefcd4b98ea2acc48203b4d18aba Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Fri, 28 Aug 2026 14:08:55 +0200 Subject: [PATCH 1/3] fix(core): keep abnormally ended text streams from rethrowing globally --- .changeset/text-stream-abnormal-end.md | 5 ++ packages/core/src/components/chat.ts | 6 ++ .../core/src/components/textStream.test.ts | 64 +++++++++++++++ packages/core/src/components/textStream.ts | 78 +++++++++++-------- 4 files changed, 119 insertions(+), 34 deletions(-) create mode 100644 .changeset/text-stream-abnormal-end.md create mode 100644 packages/core/src/components/textStream.test.ts diff --git a/.changeset/text-stream-abnormal-end.md b/.changeset/text-stream-abnormal-end.md new file mode 100644 index 000000000..704371f56 --- /dev/null +++ b/.changeset/text-stream-abnormal-end.md @@ -0,0 +1,5 @@ +--- +'@livekit/components-core': patch +--- + +Handle text streams that end abnormally (e.g. the sending participant disconnects mid-stream): keep the accumulated text and log at debug instead of letting RxJS rethrow the DataStreamError globally as an uncaught exception. diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index 038957c75..d9a64b2a2 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -162,6 +162,12 @@ export function setupChat(room: Room, options?: ChatOptions) { ); streamObservable.subscribe({ next: (value) => messageSubject.next(value), + error: (error) => { + // Keep an abnormally ended stream (e.g. the sending participant + // disconnected mid-stream) from rethrowing globally as an uncaught + // exception; `finalize` has already cleaned up its attachment state. + log.debug('chat text stream ended abnormally', error); + }, }); }); // NOTE: Attachment byte streams are guaranteed to arrive after their parent text stream diff --git a/packages/core/src/components/textStream.test.ts b/packages/core/src/components/textStream.test.ts new file mode 100644 index 000000000..8d9e45cd8 --- /dev/null +++ b/packages/core/src/components/textStream.test.ts @@ -0,0 +1,64 @@ +import type { Room } from 'livekit-client'; +import { describe, expect, it, vi } from 'vitest'; +import { log } from '../logger'; +import { setupTextStream, type TextStreamData } from './textStream'; + +type TextStreamHandler = ( + reader: AsyncIterable & { info: { id: string; attributes?: Record } }, + participantInfo: { identity: string }, +) => Promise; + +const makeRoom = () => { + const handlers = new Map(); + const room = { + registerTextStreamHandler: (topic: string, handler: TextStreamHandler) => { + handlers.set(topic, handler); + }, + unregisterTextStreamHandler: (topic: string) => { + handlers.delete(topic); + }, + on: () => room, + } as unknown as Room; + return { room, handlers }; +}; + +describe('setupTextStream', () => { + it('keeps the accumulated text when a stream ends abnormally instead of rethrowing', async () => { + const { room, handlers } = makeRoom(); + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + + const emissions: TextStreamData[][] = []; + const subscription = setupTextStream(room, 'lk.transcription').subscribe((streams) => { + emissions.push(streams); + }); + + const handler = handlers.get('lk.transcription'); + if (!handler) { + throw new Error('text stream handler was not registered'); + } + + const abnormalEnd = new Error( + 'Participant agent-x unexpectedly disconnected in the middle of sending data', + ); + await handler( + { + info: { id: 'stream-1', attributes: {} }, + // eslint-disable-next-line @typescript-eslint/no-explicit-any + async *[Symbol.asyncIterator]() { + yield 'Hello '; + yield 'world'; + throw abnormalEnd; + }, + }, + { identity: 'agent-x' }, + ); + await new Promise((resolve) => setTimeout(resolve, 0)); + + const lastEmission = emissions.at(-1); + expect(lastEmission?.[0]?.text).toBe('Hello world'); + expect(debugSpy).toHaveBeenCalledWith('text stream ended abnormally', abnormalEnd); + + subscription.unsubscribe(); + debugSpy.mockRestore(); + }); +}); diff --git a/packages/core/src/components/textStream.ts b/packages/core/src/components/textStream.ts index 736e6bdcf..dbe26f835 100644 --- a/packages/core/src/components/textStream.ts +++ b/packages/core/src/components/textStream.ts @@ -2,6 +2,7 @@ import { RoomEvent, type Room, type TextStreamInfo } from 'livekit-client'; import { from, scan, Subject, type Observable } from 'rxjs'; import { share, tap } from 'rxjs/operators'; import { ParticipantAgentAttributes } from '../helper'; +import { log } from '../logger'; export interface TextStreamData { text: string; @@ -74,40 +75,49 @@ export function setupTextStream(room: Room, topic: string): Observable { - // Find and update the stream in our array - const index = textStreams.findIndex( - (stream) => - stream.streamInfo.id === reader.info.id || - (isTranscription && - stream.streamInfo.attributes?.[segmentAttribute] === - reader.info.attributes?.[segmentAttribute]), - ); - if (index !== -1) { - textStreams[index] = { - ...textStreams[index], - text: accumulatedText, - // Carry the latest streamInfo forward. Transcription updates for a - // segment arrive as separate streams sharing the same lk.segment_id; - // keeping the original streamInfo would freeze attributes that change - // over the segment's lifetime — notably lk.transcription_final flipping - // "false" -> "true" on the final user STT result. - streamInfo: reader.info, - }; - - // Emit the updated array - textStreamsSubject.next([...textStreams]); - } else { - // Handle case where stream ID wasn't found (new stream) - textStreams.push({ - text: accumulatedText, - participantInfo, - streamInfo: reader.info, - }); - - // Emit the updated array with the new stream - textStreamsSubject.next([...textStreams]); - } + streamObservable.subscribe({ + next: (accumulatedText) => { + // Find and update the stream in our array + const index = textStreams.findIndex( + (stream) => + stream.streamInfo.id === reader.info.id || + (isTranscription && + stream.streamInfo.attributes?.[segmentAttribute] === + reader.info.attributes?.[segmentAttribute]), + ); + if (index !== -1) { + textStreams[index] = { + ...textStreams[index], + text: accumulatedText, + // Carry the latest streamInfo forward. Transcription updates for a + // segment arrive as separate streams sharing the same lk.segment_id; + // keeping the original streamInfo would freeze attributes that change + // over the segment's lifetime — notably lk.transcription_final flipping + // "false" -> "true" on the final user STT result. + streamInfo: reader.info, + }; + + // Emit the updated array + textStreamsSubject.next([...textStreams]); + } else { + // Handle case where stream ID wasn't found (new stream) + textStreams.push({ + text: accumulatedText, + participantInfo, + streamInfo: reader.info, + }); + + // Emit the updated array with the new stream + textStreamsSubject.next([...textStreams]); + } + }, + error: (error) => { + // A stream that ends abnormally (e.g. the sending participant + // disconnected mid-stream) has already delivered its chunks through + // `next`; keep the accumulated text instead of letting RxJS rethrow + // the error globally as an uncaught exception. + log.debug('text stream ended abnormally', error); + }, }); }); }, From ebbd8d0bac0644b1a90252f315d3a7af902872e6 Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Fri, 28 Aug 2026 23:31:19 +0200 Subject: [PATCH 2/3] fix(core): settle attachment futures when a byte stream fails --- packages/core/src/components/chat.test.ts | 156 ++++++++++++++++++++++ packages/core/src/components/chat.ts | 36 +++-- 2 files changed, 184 insertions(+), 8 deletions(-) create mode 100644 packages/core/src/components/chat.test.ts diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts new file mode 100644 index 000000000..908aadc73 --- /dev/null +++ b/packages/core/src/components/chat.test.ts @@ -0,0 +1,156 @@ +import type { Room } from 'livekit-client'; +import { DataStreamError, DataStreamErrorReason } from 'livekit-client'; +import { describe, expect, it, vi } from 'vitest'; +import { log } from '../logger'; +import { setupChat } from './chat'; +import type { ReceivedChatMessage } from '../messages/types'; + +type TextReader = AsyncIterable & { + info: { + id: string; + timestamp: number; + attributes?: Record; + attachedStreamIds?: string[]; + }; +}; + +type ByteReader = AsyncIterable & { + info: { id: string; name: string; mimeType: string }; +}; + +type TextStreamHandler = ( + reader: TextReader, + participantInfo: { identity: string }, +) => Promise; +type ByteStreamHandler = (reader: ByteReader) => Promise; + +const makeRoom = () => { + const textHandlers = new Map(); + const byteHandlers = new Map(); + const room = { + serverInfo: { edition: 1 }, + registerTextStreamHandler: (topic: string, handler: TextStreamHandler) => { + textHandlers.set(topic, handler); + }, + registerByteStreamHandler: (topic: string, handler: ByteStreamHandler) => { + byteHandlers.set(topic, handler); + }, + unregisterTextStreamHandler: (topic: string) => { + textHandlers.delete(topic); + }, + unregisterByteStreamHandler: (topic: string) => { + byteHandlers.delete(topic); + }, + on: () => room, + off: () => room, + once: () => room, + getParticipantByIdentity: () => undefined, + } as unknown as Room; + return { room, textHandlers, byteHandlers }; +}; + +const textReader = (id: string, text: string, attachedStreamIds: string[]): TextReader => ({ + info: { id, timestamp: Date.now(), attributes: {}, attachedStreamIds }, + async *[Symbol.asyncIterator]() { + yield text; + }, +}); + +const settle = () => new Promise((resolve) => setTimeout(resolve, 10)); + +describe('setupChat with attachments', () => { + it('settles the message pipeline when an attachment stream ends abnormally', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + try { + const { room, textHandlers, byteHandlers } = makeRoom(); + const emissions: ReceivedChatMessage[][] = []; + const chat = setupChat(room); + const subscription = chat.messageObservable.subscribe((messages) => { + emissions.push(messages); + }); + + const textHandler = textHandlers.get('lk.chat'); + const byteHandler = byteHandlers.get('lk.chat'); + if (!textHandler || !byteHandler) { + throw new Error('chat stream handlers were not registered'); + } + + await textHandler(textReader('msg-1', 'hello with attachment', ['att-1']), { + identity: 'sender', + }); + await byteHandler({ + info: { id: 'att-1', name: 'photo.png', mimeType: 'image/png' }, + async *[Symbol.asyncIterator]() { + throw new DataStreamError( + 'Participant sender unexpectedly disconnected in the middle of sending data', + DataStreamErrorReason.AbnormalEnd, + ); + }, + }); + await settle(); + + expect(unhandled).toEqual([]); + expect(emissions.flat()).toEqual([]); + expect(debugSpy).toHaveBeenCalledWith( + 'chat message stream ended abnormally', + expect.any(DataStreamError), + ); + + // The pipeline stays healthy: a later message with a completing + // attachment still delivers. + await textHandler(textReader('msg-2', 'second message', ['att-2']), { + identity: 'sender', + }); + await byteHandler({ + info: { id: 'att-2', name: 'notes.txt', mimeType: 'text/plain' }, + async *[Symbol.asyncIterator]() { + yield new TextEncoder().encode('file body'); + }, + }); + await vi.waitFor(() => { + const messages = emissions.at(-1) ?? []; + expect(messages.map((message) => message.message)).toEqual(['second message']); + expect(messages[0]?.attachedFiles?.[0]?.name).toBe('notes.txt'); + }); + + subscription.unsubscribe(); + } finally { + process.off('unhandledRejection', onUnhandled); + debugSpy.mockRestore(); + } + }); + + it('warns instead of debug-logging when an attachment fails for another reason', async () => { + const warnSpy = vi.spyOn(log, 'warn').mockImplementation(() => {}); + try { + const { room, textHandlers, byteHandlers } = makeRoom(); + const chat = setupChat(room); + const subscription = chat.messageObservable.subscribe(() => {}); + + const textHandler = textHandlers.get('lk.chat'); + const byteHandler = byteHandlers.get('lk.chat'); + if (!textHandler || !byteHandler) { + throw new Error('chat stream handlers were not registered'); + } + + await textHandler(textReader('msg-3', 'message', ['att-3']), { identity: 'sender' }); + await byteHandler({ + info: { id: 'att-3', name: 'photo.png', mimeType: 'image/png' }, + async *[Symbol.asyncIterator]() { + throw new Error('decompression failed'); + }, + }); + await settle(); + + expect(warnSpy).toHaveBeenCalledWith('chat message stream failed', expect.any(Error)); + subscription.unsubscribe(); + } finally { + warnSpy.mockRestore(); + } + }); +}); diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index d9a64b2a2..4d8fe9387 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -1,6 +1,6 @@ /* eslint-disable camelcase */ import type { Room, SendTextOptions } from 'livekit-client'; -import { compareVersions, RoomEvent } from 'livekit-client'; +import { compareVersions, DataStreamError, DataStreamErrorReason, RoomEvent } from 'livekit-client'; import { BehaviorSubject, Subject, @@ -67,7 +67,7 @@ const streamIdToAttachments = new Map< mimeType: string; buffer: Array; }, - never + Error > > >(); @@ -112,7 +112,7 @@ export function setupChat(room: Room, options?: ChatOptions) { const attachments = new Map( (attachedStreamIds ?? []).map((id) => [ id, - new Future<{ fileName: string; mimeType: string; buffer: Array }, never>(), + new Future<{ fileName: string; mimeType: string; buffer: Array }, Error>(), ]), ); streamIdToAttachments.set(id, attachments); @@ -163,10 +163,18 @@ export function setupChat(room: Room, options?: ChatOptions) { streamObservable.subscribe({ next: (value) => messageSubject.next(value), error: (error) => { - // Keep an abnormally ended stream (e.g. the sending participant - // disconnected mid-stream) from rethrowing globally as an uncaught + // Keep a failed message (text stream errored, or an attachment + // future rejected) from rethrowing globally as an uncaught // exception; `finalize` has already cleaned up its attachment state. - log.debug('chat text stream ended abnormally', error); + // A disconnect mid-stream is expected churn; anything else deserves + // a visible warning. + const abnormalEnd = + error instanceof DataStreamError && error.reason === DataStreamErrorReason.AbnormalEnd; + if (abnormalEnd) { + log.debug('chat message stream ended abnormally', error); + } else { + log.warn('chat message stream failed', error); + } }, }); }); @@ -183,8 +191,20 @@ export function setupChat(room: Room, options?: ChatOptions) { const streamId = foundStreamAttachmentPair[0]; const bufferList = []; - for await (const buffer of reader) { - bufferList.push(buffer); + try { + for await (const buffer of reader) { + bufferList.push(buffer); + } + } catch (error) { + // Settle the attachment future so the message pipeline errors instead + // of hanging forever - its error callback logs and `finalize` cleans up + // the attachment state. Without this the rejection is unhandled and the + // pending future leaks its `streamIdToAttachments` entry. + streamIdToAttachments + .get(streamId) + ?.get(attachmentStreamId) + ?.reject?.(error instanceof Error ? error : new Error(String(error))); + return; } const attachment = streamIdToAttachments.get(streamId)?.get(attachmentStreamId); From 3469a612f5a62f40082d2fa95c7cd52042997f8c Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Tue, 1 Sep 2026 10:41:10 +0200 Subject: [PATCH 3/3] fix(core): settle attachments that reject early or never start --- packages/core/src/components/chat.test.ts | 130 +++++++++++++++++++++- packages/core/src/components/chat.ts | 72 ++++++++---- 2 files changed, 177 insertions(+), 25 deletions(-) diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts index 908aadc73..df9f6fc20 100644 --- a/packages/core/src/components/chat.test.ts +++ b/packages/core/src/components/chat.test.ts @@ -1,5 +1,5 @@ import type { Room } from 'livekit-client'; -import { DataStreamError, DataStreamErrorReason } from 'livekit-client'; +import { DataStreamError, DataStreamErrorReason, RoomEvent } from 'livekit-client'; import { describe, expect, it, vi } from 'vitest'; import { log } from '../logger'; import { setupChat } from './chat'; @@ -18,6 +18,11 @@ type ByteReader = AsyncIterable & { info: { id: string; name: string; mimeType: string }; }; +const byteReader = (id: string, body: () => AsyncGenerator): ByteReader => ({ + info: { id, name: `${id}.bin`, mimeType: 'application/octet-stream' }, + [Symbol.asyncIterator]: body, +}); + type TextStreamHandler = ( reader: TextReader, participantInfo: { identity: string }, @@ -27,6 +32,13 @@ type ByteStreamHandler = (reader: ByteReader) => Promise; const makeRoom = () => { const textHandlers = new Map(); const byteHandlers = new Map(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const roomEvents = new Map void>>(); + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const record = (event: string, handler: (...args: any[]) => void) => { + roomEvents.set(event, [...(roomEvents.get(event) ?? []), handler]); + return room; + }; const room = { serverInfo: { edition: 1 }, registerTextStreamHandler: (topic: string, handler: TextStreamHandler) => { @@ -41,12 +53,17 @@ const makeRoom = () => { unregisterByteStreamHandler: (topic: string) => { byteHandlers.delete(topic); }, - on: () => room, + on: record, off: () => room, - once: () => room, + once: record, getParticipantByIdentity: () => undefined, } as unknown as Room; - return { room, textHandlers, byteHandlers }; + const emitRoomEvent = (event: string, ...args: unknown[]) => { + for (const handler of roomEvents.get(event) ?? []) { + handler(...args); + } + }; + return { room, textHandlers, byteHandlers, emitRoomEvent }; }; const textReader = (id: string, text: string, attachedStreamIds: string[]): TextReader => ({ @@ -125,6 +142,111 @@ describe('setupChat with attachments', () => { } }); + it('does not surface an unhandled rejection when a later attachment fails first', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + try { + const { room, textHandlers, byteHandlers } = makeRoom(); + const chat = setupChat(room); + const subscription = chat.messageObservable.subscribe(() => {}); + + const textHandler = textHandlers.get('lk.chat'); + const byteHandler = byteHandlers.get('lk.chat'); + if (!textHandler || !byteHandler) { + throw new Error('chat stream handlers were not registered'); + } + + await textHandler(textReader('msg-race', 'two attachments', ['att-a', 'att-b']), { + identity: 'sender', + }); + await settle(); + + // `concatMap` subscribes to the attachment futures one at a time, so + // nothing is listening to att-b while att-a is still in flight - + // rejecting it must not surface globally. + await byteHandler( + byteReader('att-b', async function* () { + throw new DataStreamError( + 'Participant sender unexpectedly disconnected in the middle of sending data', + DataStreamErrorReason.AbnormalEnd, + ); + }), + ); + await settle(); + + expect(unhandled).toEqual([]); + subscription.unsubscribe(); + } finally { + process.off('unhandledRejection', onUnhandled); + debugSpy.mockRestore(); + } + }); + + it('settles the message when an attachment byte stream never arrives at all', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + try { + const { room, textHandlers, byteHandlers, emitRoomEvent } = makeRoom(); + const emissions: ReceivedChatMessage[][] = []; + const chat = setupChat(room); + const subscription = chat.messageObservable.subscribe((messages) => { + emissions.push(messages); + }); + + const textHandler = textHandlers.get('lk.chat'); + const byteHandler = byteHandlers.get('lk.chat'); + if (!textHandler || !byteHandler) { + throw new Error('chat stream handlers were not registered'); + } + + // The text arrives complete, but the sender drops before ever opening + // the attachment byte stream - no controller exists for livekit-client + // to error, so only the disconnect event can settle the future. + await textHandler(textReader('msg-gone', 'text arrived, attachment never did', ['att-x']), { + identity: 'sender', + }); + await settle(); + emitRoomEvent(RoomEvent.ParticipantDisconnected, { identity: 'sender' }); + await settle(); + + // Terminal state: the message is dropped (consistent with a mid-transfer + // attachment failure), nothing hangs, nothing rejects unhandled. + expect(unhandled).toEqual([]); + expect(emissions.flat()).toEqual([]); + expect(debugSpy).toHaveBeenCalledWith( + 'chat message stream ended abnormally', + expect.any(DataStreamError), + ); + + // The pipeline stays healthy for the next message. + await textHandler(textReader('msg-after', 'later message', ['att-y']), { + identity: 'sender-2', + }); + await byteHandler( + byteReader('att-y', async function* () { + yield new TextEncoder().encode('file body'); + }), + ); + await vi.waitFor(() => { + const messages = emissions.at(-1) ?? []; + expect(messages.map((message) => message.message)).toEqual(['later message']); + }); + + subscription.unsubscribe(); + } finally { + process.off('unhandledRejection', onUnhandled); + debugSpy.mockRestore(); + } + }); + it('warns instead of debug-logging when an attachment fails for another reason', async () => { const warnSpy = vi.spyOn(log, 'warn').mockImplementation(() => {}); try { diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index 4d8fe9387..2f6dce0b5 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -1,5 +1,5 @@ /* eslint-disable camelcase */ -import type { Room, SendTextOptions } from 'livekit-client'; +import type { Participant, Room, SendTextOptions } from 'livekit-client'; import { compareVersions, DataStreamError, DataStreamErrorReason, RoomEvent } from 'livekit-client'; import { BehaviorSubject, @@ -59,17 +59,20 @@ export type ChatOptions = { const topicSubjectMap: WeakMap>> = new WeakMap(); const streamIdToAttachments = new Map< string /* stream id */, - Map< - string /* attachment id */, - Future< - { - fileName: string; - mimeType: string; - buffer: Array; - }, - Error - > - > + { + senderIdentity: string; + attachments: Map< + string /* attachment id */, + Future< + { + fileName: string; + mimeType: string; + buffer: Array; + }, + Error + > + >; + } >(); function isIgnorableChatMessage(msg: ReceivedChatMessage | LegacyReceivedChatMessage) { @@ -110,12 +113,18 @@ export function setupChat(room: Room, options?: ChatOptions) { // Store a future for each attachment to be later resolved once the corresponding file data // stream completes. const attachments = new Map( - (attachedStreamIds ?? []).map((id) => [ - id, - new Future<{ fileName: string; mimeType: string; buffer: Array }, Error>(), - ]), + (attachedStreamIds ?? []).map((id) => { + const future = new Future< + { fileName: string; mimeType: string; buffer: Array }, + Error + >(); + // Ignore emitting `unhandledRejection` if the promise rejects before + // the attachments `concatMap` switches to this promise. + future.promise.catch(() => {}); + return [id, future] as const; + }), ); - streamIdToAttachments.set(id, attachments); + streamIdToAttachments.set(id, { senderIdentity: participantInfo.identity, attachments }); const streamObservable = from(reader).pipe( scan((acc: string, chunk: string) => { @@ -182,8 +191,8 @@ export function setupChat(room: Room, options?: ChatOptions) { // has initialized the attachment map (per client SDK sending implementation) room.registerByteStreamHandler(topic, async (reader) => { const { id: attachmentStreamId } = reader.info; - const foundStreamAttachmentPair = Array.from(streamIdToAttachments).find(([, attachments]) => - attachments.has(attachmentStreamId), + const foundStreamAttachmentPair = Array.from(streamIdToAttachments).find(([, entry]) => + entry.attachments.has(attachmentStreamId), ); if (!foundStreamAttachmentPair) { return; @@ -202,12 +211,12 @@ export function setupChat(room: Room, options?: ChatOptions) { // pending future leaks its `streamIdToAttachments` entry. streamIdToAttachments .get(streamId) - ?.get(attachmentStreamId) + ?.attachments.get(attachmentStreamId) ?.reject?.(error instanceof Error ? error : new Error(String(error))); return; } - const attachment = streamIdToAttachments.get(streamId)?.get(attachmentStreamId); + const attachment = streamIdToAttachments.get(streamId)?.attachments.get(attachmentStreamId); if (!attachment) { return; } @@ -317,6 +326,26 @@ export function setupChat(room: Room, options?: ChatOptions) { } }; + const handleParticipantDisconnected = (participant: Participant) => { + for (const { senderIdentity, attachments } of streamIdToAttachments.values()) { + if (senderIdentity !== participant.identity) { + continue; + } + for (const attachment of attachments.values()) { + // A byte stream that never opened has no controller livekit-client + // could error - settle it here so the message pipeline reaches a + // terminal state instead of hanging. Settled futures ignore this. + attachment.reject?.( + new DataStreamError( + `Participant ${participant.identity} disconnected before sending all attachments`, + DataStreamErrorReason.AbnormalEnd, + ), + ); + } + } + }; + room.on(RoomEvent.ParticipantDisconnected, handleParticipantDisconnected); + function destroy() { onDestroyObservable.next(); onDestroyObservable.complete(); @@ -324,6 +353,7 @@ export function setupChat(room: Room, options?: ChatOptions) { topicSubjectMap.delete(room); room.unregisterTextStreamHandler(topic); room.unregisterByteStreamHandler(topic); + room.off(RoomEvent.ParticipantDisconnected, handleParticipantDisconnected); } room.once(RoomEvent.Disconnected, destroy);