From efd15105449057222217de6d5edd5669a8e87c68 Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Fri, 28 Aug 2026 14:08:55 +0200 Subject: [PATCH 1/5] 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 | 32 +++++++- packages/core/src/components/textStream.ts | 74 +++++++++++-------- 4 files changed, 84 insertions(+), 33 deletions(-) create mode 100644 .changeset/text-stream-abnormal-end.md 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 125a3648c..0f4f9bab8 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -191,6 +191,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 index 27ea8723b..af07bbf8d 100644 --- a/packages/core/src/components/textStream.test.ts +++ b/packages/core/src/components/textStream.test.ts @@ -1,6 +1,7 @@ import { RoomEvent, type Room } from 'livekit-client'; import { describe, expect, it, vi } from 'vitest'; -import { setupTextStream } from './textStream'; +import { log } from '../logger'; +import { setupTextStream, type TextStreamData } from './textStream'; /** * Minimal stand-in for `Room` that mirrors livekit-client's one-handler-per-topic @@ -106,4 +107,33 @@ describe('setupTextStream', () => { expect(after).toEqual([1]); secondSub.unsubscribe(); }); + + it('keeps the accumulated text when a stream ends abnormally instead of rethrowing', async () => { + const { room, push } = createFakeRoom(); + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + + const emissions: TextStreamData[][] = []; + const subscription = setupTextStream(room, TOPIC).subscribe((streams) => { + emissions.push(streams); + }); + + const abnormalEnd = new Error( + 'Participant agent-x unexpectedly disconnected in the middle of sending data', + ); + await push(TOPIC, { + info: { id: 'stream-1', attributes: {} }, + async *[Symbol.asyncIterator]() { + yield 'Hello '; + yield 'world'; + throw abnormalEnd; + }, + }); + await flush(); + + expect(emissions.at(-1)?.[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 f17227733..899ed8adf 100644 --- a/packages/core/src/components/textStream.ts +++ b/packages/core/src/components/textStream.ts @@ -2,6 +2,7 @@ import { 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; @@ -56,40 +57,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, - }; + 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 + 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]); - } + // 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 38bd24d6f7c080ade978e56127c4d9818494cc73 Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Fri, 28 Aug 2026 23:31:19 +0200 Subject: [PATCH 2/5] 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 0f4f9bab8..3d1426a03 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, @@ -82,7 +82,7 @@ const streamIdToAttachments = new Map< mimeType: string; buffer: Array; }, - never + Error > > >(); @@ -141,7 +141,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); @@ -192,10 +192,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); + } }, }); }); @@ -212,8 +220,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 50fbfb62ae0a9e72bab98131f7f5ceb5c467bdd9 Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Tue, 1 Sep 2026 10:41:10 +0200 Subject: [PATCH 3/5] 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 3d1426a03..91848721a 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, @@ -74,17 +74,20 @@ type ChatTopicState = { 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) { @@ -118,7 +121,28 @@ export function setupChat(room: Room, options?: ChatOptions) { topicSubjectMap.set(room, topicMap); if (isFirstTopicForRoom) { + 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); + room.once(RoomEvent.Disconnected, () => { + room.off(RoomEvent.ParticipantDisconnected, handleParticipantDisconnected); const topics = topicSubjectMap.get(room); topicSubjectMap.delete(room); topics?.forEach(({ messageSubject: subject, onDestroyObservable: onDestroy }, chatTopic) => { @@ -139,12 +163,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) => { @@ -211,8 +241,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; @@ -231,12 +261,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; } From d75d509b70837466b4608ac089b233ef8569fc52 Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Fri, 9 Oct 2026 17:19:09 +0200 Subject: [PATCH 4/5] fix(core): settle never-started attachments with a local error and scope them to their room A sender that leaves before opening an attachment's byte stream gives livekit-client nothing to error, so the future is settled on ParticipantDisconnected. Do that with an AttachmentNotReceivedError instead of a DataStreamError built outside the SDK, and only for attachments of the room the participant left: identities are unique per room, not globally. Co-Authored-By: Claude Opus 5.5 --- .changeset/text-stream-abnormal-end.md | 2 +- packages/core/src/components/chat.test.ts | 53 ++++++++++++++++++++++- packages/core/src/components/chat.ts | 48 +++++++++++++++----- 3 files changed, 90 insertions(+), 13 deletions(-) diff --git a/.changeset/text-stream-abnormal-end.md b/.changeset/text-stream-abnormal-end.md index 704371f56..191f013a0 100644 --- a/.changeset/text-stream-abnormal-end.md +++ b/.changeset/text-stream-abnormal-end.md @@ -2,4 +2,4 @@ '@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. +Handle chat and text streams that end abnormally, for example when the sending participant disconnects mid-stream. Text streams keep the accumulated text and log at debug instead of letting RxJS rethrow the error globally. A chat message whose attachment stream fails, or never starts because its sender left, now settles instead of hanging, without an unhandled rejection. diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts index df9f6fc20..0555b040f 100644 --- a/packages/core/src/components/chat.test.ts +++ b/packages/core/src/components/chat.test.ts @@ -223,7 +223,7 @@ describe('setupChat with attachments', () => { expect(emissions.flat()).toEqual([]); expect(debugSpy).toHaveBeenCalledWith( 'chat message stream ended abnormally', - expect.any(DataStreamError), + expect.objectContaining({ name: 'AttachmentNotReceivedError' }), ); // The pipeline stays healthy for the next message. @@ -247,6 +247,57 @@ describe('setupChat with attachments', () => { } }); + it('leaves pending attachments in other rooms alone when a same-named participant leaves', async () => { + const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + try { + const roomA = makeRoom(); + const roomB = makeRoom(); + const chatA = setupChat(roomA.room); + const chatB = setupChat(roomB.room); + const emissionsB: ReceivedChatMessage[][] = []; + const subscriptionA = chatA.messageObservable.subscribe(() => {}); + const subscriptionB = chatB.messageObservable.subscribe((messages) => { + emissionsB.push(messages); + }); + + const textHandlerB = roomB.textHandlers.get('lk.chat'); + const byteHandlerB = roomB.byteHandlers.get('lk.chat'); + if (!textHandlerB || !byteHandlerB) { + throw new Error('chat stream handlers were not registered'); + } + + // Room B has a message waiting on its attachment from "agent". + await textHandlerB(textReader('msg-b', 'message in room B', ['att-b']), { + identity: 'agent', + }); + await settle(); + + // A participant with the same identity leaves room A. + roomA.emitRoomEvent(RoomEvent.ParticipantDisconnected, { identity: 'agent' }); + await settle(); + expect(debugSpy).not.toHaveBeenCalledWith( + 'chat message stream ended abnormally', + expect.objectContaining({ name: 'AttachmentNotReceivedError' }), + ); + + // Room B's attachment still arrives and completes its message. + await byteHandlerB( + byteReader('att-b', async function* () { + yield new TextEncoder().encode('file body'); + }), + ); + await vi.waitFor(() => { + const messages = emissionsB.at(-1) ?? []; + expect(messages.map((message) => message.message)).toEqual(['message in room B']); + }); + + subscriptionA.unsubscribe(); + subscriptionB.unsubscribe(); + } finally { + 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 91848721a..fd2c701f4 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -72,9 +72,31 @@ type ChatTopicState = { }; const topicSubjectMap: WeakMap> = new WeakMap(); + +/** + * Settles an attachment whose byte stream never opened because its sender + * left first. livekit-client raises nothing for a stream that never started, + * so there is no SDK error to pass along. + */ +class AttachmentNotReceivedError extends Error { + readonly participantIdentity: string; + + readonly attachmentStreamId: string; + + constructor(participantIdentity: string, attachmentStreamId: string) { + super( + `Participant ${participantIdentity} disconnected before sending attachment ${attachmentStreamId}`, + ); + this.name = 'AttachmentNotReceivedError'; + this.participantIdentity = participantIdentity; + this.attachmentStreamId = attachmentStreamId; + } +} + const streamIdToAttachments = new Map< string /* stream id */, { + room: Room; senderIdentity: string; attachments: Map< string /* attachment id */, @@ -122,19 +144,17 @@ export function setupChat(room: Room, options?: ChatOptions) { if (isFirstTopicForRoom) { const handleParticipantDisconnected = (participant: Participant) => { - for (const { senderIdentity, attachments } of streamIdToAttachments.values()) { - if (senderIdentity !== participant.identity) { + for (const entry of streamIdToAttachments.values()) { + // Identities are only unique within a room. + if (entry.room !== room || entry.senderIdentity !== participant.identity) { continue; } - for (const attachment of attachments.values()) { + for (const [attachmentStreamId, attachment] of entry.attachments) { // 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, - ), + new AttachmentNotReceivedError(participant.identity, attachmentStreamId), ); } } @@ -174,7 +194,11 @@ export function setupChat(room: Room, options?: ChatOptions) { return [id, future] as const; }), ); - streamIdToAttachments.set(id, { senderIdentity: participantInfo.identity, attachments }); + streamIdToAttachments.set(id, { + room, + senderIdentity: participantInfo.identity, + attachments, + }); const streamObservable = from(reader).pipe( scan((acc: string, chunk: string) => { @@ -228,7 +252,9 @@ export function setupChat(room: Room, options?: ChatOptions) { // A disconnect mid-stream is expected churn; anything else deserves // a visible warning. const abnormalEnd = - error instanceof DataStreamError && error.reason === DataStreamErrorReason.AbnormalEnd; + error instanceof AttachmentNotReceivedError || + (error instanceof DataStreamError && + error.reason === DataStreamErrorReason.AbnormalEnd); if (abnormalEnd) { log.debug('chat message stream ended abnormally', error); } else { @@ -241,8 +267,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(([, entry]) => - entry.attachments.has(attachmentStreamId), + const foundStreamAttachmentPair = Array.from(streamIdToAttachments).find( + ([, entry]) => entry.room === room && entry.attachments.has(attachmentStreamId), ); if (!foundStreamAttachmentPair) { return; From 921931568428ce4f9bec8cf9b60431ab32a2844c Mon Sep 17 00:00:00 2001 From: Anze Mur Date: Tue, 13 Oct 2026 01:08:35 +0200 Subject: [PATCH 5/5] fix(core): end chat messages when their sender leaves or the room disconnects Replace AttachmentNotReceivedError and its reject loop with takeUntil on the message pipeline, as suggested in review: one notifier is the sender's ParticipantDisconnected (this room, this identity), the other is the room's own disconnect, because a local disconnect clears open stream controllers without erroring them. The pipeline completes instead of erroring, finalize removes the attachment entry, and the entry no longer references the Room. Co-Authored-By: Claude Opus 5.5 --- .changeset/text-stream-abnormal-end.md | 2 +- packages/core/src/components/chat.test.ts | 174 +++++++++++++++++++--- packages/core/src/components/chat.ts | 102 ++++--------- 3 files changed, 184 insertions(+), 94 deletions(-) diff --git a/.changeset/text-stream-abnormal-end.md b/.changeset/text-stream-abnormal-end.md index 191f013a0..1856a00f6 100644 --- a/.changeset/text-stream-abnormal-end.md +++ b/.changeset/text-stream-abnormal-end.md @@ -2,4 +2,4 @@ '@livekit/components-core': patch --- -Handle chat and text streams that end abnormally, for example when the sending participant disconnects mid-stream. Text streams keep the accumulated text and log at debug instead of letting RxJS rethrow the error globally. A chat message whose attachment stream fails, or never starts because its sender left, now settles instead of hanging, without an unhandled rejection. +Handle chat and text streams that end abnormally, for example when the sending participant disconnects mid-stream. Text streams keep the accumulated text and log at debug instead of letting RxJS rethrow the error globally. A chat message whose attachment stream fails now settles instead of hanging, without an unhandled rejection, and a message still waiting on attachments ends when its sender leaves or the local room disconnects. diff --git a/packages/core/src/components/chat.test.ts b/packages/core/src/components/chat.test.ts index 0555b040f..00b08540b 100644 --- a/packages/core/src/components/chat.test.ts +++ b/packages/core/src/components/chat.test.ts @@ -39,6 +39,14 @@ const makeRoom = () => { roomEvents.set(event, [...(roomEvents.get(event) ?? []), handler]); return room; }; + // eslint-disable-next-line @typescript-eslint/no-explicit-any + const remove = (event: string, handler: (...args: any[]) => void) => { + roomEvents.set( + event, + (roomEvents.get(event) ?? []).filter((registered) => registered !== handler), + ); + return room; + }; const room = { serverInfo: { edition: 1 }, registerTextStreamHandler: (topic: string, handler: TextStreamHandler) => { @@ -54,7 +62,7 @@ const makeRoom = () => { byteHandlers.delete(topic); }, on: record, - off: () => room, + off: remove, once: record, getParticipantByIdentity: () => undefined, } as unknown as Room; @@ -63,7 +71,8 @@ const makeRoom = () => { handler(...args); } }; - return { room, textHandlers, byteHandlers, emitRoomEvent }; + const listenerCount = (event: string) => roomEvents.get(event)?.length ?? 0; + return { room, textHandlers, byteHandlers, emitRoomEvent, listenerCount }; }; const textReader = (id: string, text: string, attachedStreamIds: string[]): TextReader => ({ @@ -75,6 +84,19 @@ const textReader = (id: string, text: string, attachedStreamIds: string[]): Text const settle = () => new Promise((resolve) => setTimeout(resolve, 10)); +/** A byte reader that records whether the handler started reading it. */ +const trackedByteReader = (id: string) => { + const reader = { + started: false, + info: { id, name: `${id}.bin`, mimeType: 'application/octet-stream' }, + async *[Symbol.asyncIterator]() { + reader.started = true; + yield new TextEncoder().encode('file body'); + }, + }; + return reader; +}; + describe('setupChat with attachments', () => { it('settles the message pipeline when an attachment stream ends abnormally', async () => { const unhandled: unknown[] = []; @@ -100,23 +122,22 @@ describe('setupChat with attachments', () => { await textHandler(textReader('msg-1', 'hello with attachment', ['att-1']), { identity: 'sender', }); + const abnormalEnd = new DataStreamError( + 'Participant sender unexpectedly disconnected in the middle of sending data', + DataStreamErrorReason.AbnormalEnd, + ); 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, - ); + throw abnormalEnd; }, }); await settle(); expect(unhandled).toEqual([]); expect(emissions.flat()).toEqual([]); - expect(debugSpy).toHaveBeenCalledWith( - 'chat message stream ended abnormally', - expect.any(DataStreamError), - ); + // The SDK's own error object is logged as is, with its reason and message. + expect(debugSpy).toHaveBeenCalledWith('chat message stream ended abnormally', abnormalEnd); // The pipeline stays healthy: a later message with a completing // attachment still delivers. @@ -193,8 +214,9 @@ describe('setupChat with attachments', () => { }; process.on('unhandledRejection', onUnhandled); const debugSpy = vi.spyOn(log, 'debug').mockImplementation(() => {}); + const warnSpy = vi.spyOn(log, 'warn').mockImplementation(() => {}); try { - const { room, textHandlers, byteHandlers, emitRoomEvent } = makeRoom(); + const { room, textHandlers, byteHandlers, emitRoomEvent, listenerCount } = makeRoom(); const emissions: ReceivedChatMessage[][] = []; const chat = setupChat(room); const subscription = chat.messageObservable.subscribe((messages) => { @@ -209,22 +231,32 @@ describe('setupChat with attachments', () => { // 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. + // to error, so only the disconnect event can end the message. await textHandler(textReader('msg-gone', 'text arrived, attachment never did', ['att-x']), { identity: 'sender', }); await settle(); + expect(listenerCount(RoomEvent.ParticipantDisconnected)).toBe(1); + + // Another participant leaving does not end it. + emitRoomEvent(RoomEvent.ParticipantDisconnected, { identity: 'someone-else' }); + await settle(); + expect(listenerCount(RoomEvent.ParticipantDisconnected)).toBe(1); + 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. + // attachment failure) and completes rather than errors, so nothing is + // logged or rejected and its listener and attachment entry are gone. expect(unhandled).toEqual([]); expect(emissions.flat()).toEqual([]); - expect(debugSpy).toHaveBeenCalledWith( - 'chat message stream ended abnormally', - expect.objectContaining({ name: 'AttachmentNotReceivedError' }), - ); + expect(debugSpy).not.toHaveBeenCalled(); + expect(warnSpy).not.toHaveBeenCalled(); + expect(listenerCount(RoomEvent.ParticipantDisconnected)).toBe(0); + const lateAttachment = trackedByteReader('att-x'); + await byteHandler(lateAttachment); + expect(lateAttachment.started).toBe(false); // The pipeline stays healthy for the next message. await textHandler(textReader('msg-after', 'later message', ['att-y']), { @@ -240,6 +272,105 @@ describe('setupChat with attachments', () => { expect(messages.map((message) => message.message)).toEqual(['later message']); }); + subscription.unsubscribe(); + } finally { + process.off('unhandledRejection', onUnhandled); + debugSpy.mockRestore(); + warnSpy.mockRestore(); + } + }); + + it('ends a message still waiting on an attachment when the local room disconnects', async () => { + const unhandled: unknown[] = []; + const onUnhandled = (reason: unknown) => { + unhandled.push(reason); + }; + process.on('unhandledRejection', onUnhandled); + try { + const { room, textHandlers, byteHandlers, emitRoomEvent, listenerCount } = 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-local', 'waiting on attachment', ['att-local']), { + identity: 'sender', + }); + await settle(); + expect(listenerCount(RoomEvent.ParticipantDisconnected)).toBe(1); + + // On a local disconnect livekit-client drops open stream controllers + // without erroring them and emits no ParticipantDisconnected, so only + // the room's own disconnect can end the message. + emitRoomEvent(RoomEvent.Disconnected); + await settle(); + + expect(unhandled).toEqual([]); + expect(listenerCount(RoomEvent.ParticipantDisconnected)).toBe(0); + const lateAttachment = trackedByteReader('att-local'); + await byteHandler(lateAttachment); + expect(lateAttachment.started).toBe(false); + + subscription.unsubscribe(); + } finally { + process.off('unhandledRejection', onUnhandled); + } + }); + + it('keeps the text already delivered when the sender leaves mid-stream', 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, emitRoomEvent } = makeRoom(); + const emissions: ReceivedChatMessage[][] = []; + const chat = setupChat(room); + const subscription = chat.messageObservable.subscribe((messages) => { + emissions.push(messages); + }); + + const textHandler = textHandlers.get('lk.chat'); + if (!textHandler) { + throw new Error('chat stream handlers were not registered'); + } + + let failRead: (error: Error) => void = () => {}; + const pendingRead = new Promise((_, reject) => { + failRead = reject; + }); + await textHandler( + { + info: { id: 'msg-cut', timestamp: Date.now(), attributes: {} }, + async *[Symbol.asyncIterator]() { + yield 'Hello '; + yield await pendingRead; + }, + }, + { identity: 'sender' }, + ); + await settle(); + + // Same order as livekit-client: the open stream is errored, then + // ParticipantDisconnected is emitted in the same tick. + failRead( + new DataStreamError( + 'Participant sender unexpectedly disconnected in the middle of sending data', + DataStreamErrorReason.AbnormalEnd, + ), + ); + emitRoomEvent(RoomEvent.ParticipantDisconnected, { identity: 'sender' }); + await settle(); + + expect(unhandled).toEqual([]); + expect(emissions.at(-1)?.map((message) => message.message)).toEqual(['Hello ']); + subscription.unsubscribe(); } finally { process.off('unhandledRejection', onUnhandled); @@ -267,7 +398,7 @@ describe('setupChat with attachments', () => { } // Room B has a message waiting on its attachment from "agent". - await textHandlerB(textReader('msg-b', 'message in room B', ['att-b']), { + await textHandlerB(textReader('msg-room-b', 'message in room B', ['att-room-b']), { identity: 'agent', }); await settle(); @@ -275,14 +406,11 @@ describe('setupChat with attachments', () => { // A participant with the same identity leaves room A. roomA.emitRoomEvent(RoomEvent.ParticipantDisconnected, { identity: 'agent' }); await settle(); - expect(debugSpy).not.toHaveBeenCalledWith( - 'chat message stream ended abnormally', - expect.objectContaining({ name: 'AttachmentNotReceivedError' }), - ); + expect(roomB.listenerCount(RoomEvent.ParticipantDisconnected)).toBe(1); // Room B's attachment still arrives and completes its message. await byteHandlerB( - byteReader('att-b', async function* () { + byteReader('att-room-b', async function* () { yield new TextEncoder().encode('file body'); }), ); diff --git a/packages/core/src/components/chat.ts b/packages/core/src/components/chat.ts index fd2c701f4..f2bcd5151 100644 --- a/packages/core/src/components/chat.ts +++ b/packages/core/src/components/chat.ts @@ -1,5 +1,5 @@ /* eslint-disable camelcase */ -import type { Participant, Room, SendTextOptions } from 'livekit-client'; +import type { Room, SendTextOptions } from 'livekit-client'; import { compareVersions, DataStreamError, DataStreamErrorReason, RoomEvent } from 'livekit-client'; import { BehaviorSubject, @@ -23,6 +23,7 @@ import { import { log } from '../logger'; import { ChatMessage, ReceivedChatMessage } from '../messages/types'; import { Future } from '../helper/future'; +import { roomEventSelector } from '../observables/room'; /** @public */ export type { ChatMessage, ReceivedChatMessage }; @@ -72,44 +73,19 @@ type ChatTopicState = { }; const topicSubjectMap: WeakMap> = new WeakMap(); - -/** - * Settles an attachment whose byte stream never opened because its sender - * left first. livekit-client raises nothing for a stream that never started, - * so there is no SDK error to pass along. - */ -class AttachmentNotReceivedError extends Error { - readonly participantIdentity: string; - - readonly attachmentStreamId: string; - - constructor(participantIdentity: string, attachmentStreamId: string) { - super( - `Participant ${participantIdentity} disconnected before sending attachment ${attachmentStreamId}`, - ); - this.name = 'AttachmentNotReceivedError'; - this.participantIdentity = participantIdentity; - this.attachmentStreamId = attachmentStreamId; - } -} - const streamIdToAttachments = new Map< string /* stream id */, - { - room: Room; - senderIdentity: string; - attachments: Map< - string /* attachment id */, - Future< - { - fileName: string; - mimeType: string; - buffer: Array; - }, - Error - > - >; - } + Map< + string /* attachment id */, + Future< + { + fileName: string; + mimeType: string; + buffer: Array; + }, + Error + > + > >(); function isIgnorableChatMessage(msg: ReceivedChatMessage | LegacyReceivedChatMessage) { @@ -143,26 +119,7 @@ export function setupChat(room: Room, options?: ChatOptions) { topicSubjectMap.set(room, topicMap); if (isFirstTopicForRoom) { - const handleParticipantDisconnected = (participant: Participant) => { - for (const entry of streamIdToAttachments.values()) { - // Identities are only unique within a room. - if (entry.room !== room || entry.senderIdentity !== participant.identity) { - continue; - } - for (const [attachmentStreamId, attachment] of entry.attachments) { - // 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 AttachmentNotReceivedError(participant.identity, attachmentStreamId), - ); - } - } - }; - room.on(RoomEvent.ParticipantDisconnected, handleParticipantDisconnected); - room.once(RoomEvent.Disconnected, () => { - room.off(RoomEvent.ParticipantDisconnected, handleParticipantDisconnected); const topics = topicSubjectMap.get(room); topicSubjectMap.delete(room); topics?.forEach(({ messageSubject: subject, onDestroyObservable: onDestroy }, chatTopic) => { @@ -194,11 +151,13 @@ export function setupChat(room: Room, options?: ChatOptions) { return [id, future] as const; }), ); - streamIdToAttachments.set(id, { - room, - senderIdentity: participantInfo.identity, - attachments, - }); + streamIdToAttachments.set(id, attachments); + + // An attachment whose byte stream never opened has no controller for + // livekit-client to error, so end the message when its sender leaves. + const senderLeft$ = roomEventSelector(room, RoomEvent.ParticipantDisconnected).pipe( + filter(([participant]) => participant.identity === participantInfo.identity), + ); const streamObservable = from(reader).pipe( scan((acc: string, chunk: string) => { @@ -241,6 +200,10 @@ export function setupChat(room: Room, options?: ChatOptions) { // editTimestamp: type === 'update' ? timestamp : undefined, } satisfies ReceivedChatMessage; }), + takeUntil(senderLeft$), + // A local disconnect drops open stream controllers without erroring + // them, so end the message with the room as well. + takeUntil(onDestroyObservable), finalize(() => streamIdToAttachments.delete(id)), ); streamObservable.subscribe({ @@ -251,11 +214,10 @@ export function setupChat(room: Room, options?: ChatOptions) { // exception; `finalize` has already cleaned up its attachment state. // A disconnect mid-stream is expected churn; anything else deserves // a visible warning. - const abnormalEnd = - error instanceof AttachmentNotReceivedError || - (error instanceof DataStreamError && - error.reason === DataStreamErrorReason.AbnormalEnd); - if (abnormalEnd) { + if ( + error instanceof DataStreamError && + error.reason === DataStreamErrorReason.AbnormalEnd + ) { log.debug('chat message stream ended abnormally', error); } else { log.warn('chat message stream failed', error); @@ -267,8 +229,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( - ([, entry]) => entry.room === room && entry.attachments.has(attachmentStreamId), + const foundStreamAttachmentPair = Array.from(streamIdToAttachments).find(([, attachments]) => + attachments.has(attachmentStreamId), ); if (!foundStreamAttachmentPair) { return; @@ -287,12 +249,12 @@ export function setupChat(room: Room, options?: ChatOptions) { // pending future leaks its `streamIdToAttachments` entry. streamIdToAttachments .get(streamId) - ?.attachments.get(attachmentStreamId) + ?.get(attachmentStreamId) ?.reject?.(error instanceof Error ? error : new Error(String(error))); return; } - const attachment = streamIdToAttachments.get(streamId)?.attachments.get(attachmentStreamId); + const attachment = streamIdToAttachments.get(streamId)?.get(attachmentStreamId); if (!attachment) { return; }