fix(core): keep abnormally ended text streams from rethrowing globally

[anzemur]

What

When a participant disconnects while one of its text streams is still open, livekit-client errors the open stream's controller with a DataStreamError (AbnormalEnd) — by design, in validateParticipantHasNoActiveDataStreams.

setupTextStream reads each incoming stream with from(reader).pipe(scan(...)).subscribe(next) — no error handler. RxJS then reports the unhandled observable error by rethrowing it globally from a timer (reportUnhandledError → setTimeout(() => { throw err })), so every abnormally ended transcription stream surfaces as an uncaught DataStreamError: Participant agent-… unexpectedly disconnected in the middle of sending data that no application code can catch.

For voice-agent apps this fires whenever the agent leaves the room mid-segment (interrupted farewell, agent process exit, crash). In our production app it is the #1 error by users impacted (~940 users / week) while being functionally benign — the transcript had already been delivered chunk by chunk and the session is over.

The chat handler's streamObservable.subscribe({ next }) has the same gap.

Fix

Give both subscribes an error callback: keep the already-accumulated text (every chunk was emitted through next; on abnormal end there is nothing left to deliver) and log the error at debug level. For chat, finalize already cleans up attachment state on error.

Tests

New test drives setupTextStream with a reader that yields chunks and then throws the abnormal-end error: the accumulated text survives, the error is logged, and nothing rethrows globally (without the fix, the test run fails with the uncaught error). Full @livekit/components-core suite passes (99 tests).

Live verification

Reproduced end to end against a real livekit-server (Docker, --dev) with real Chromium on both sides (Playwright): a sender participant streams four chunks on lk.transcription without closing the writer, then goes away; the receiver page runs setupTextStream and records all uncaught errors plus emissions. 8/8 deterministic runs across 2 iterations x {unfixed 0.12.15, this branch} x {tab hard-killed, clean room.disconnect() with the stream open}:

• Unfixed: exactly one uncaught page-level DataStreamError: Participant … unexpectedly disconnected in the middle of sending data (surfaced via window.onerror / Playwright pageerror — no application code can catch it).
• Fixed: zero uncaught errors, and the accumulated text still ends complete ("Hello from the agent"), so nothing is lost by handling the error.

The only delta between the two receiver bundles is @livekit/components-core; every other dependency is alias-pinned identically. The live test drives textStream.ts; the identical handler added in chat.ts shares the same mechanism but was not separately driven.

Update: attachment (byte) streams

Review feedback pointed out the fix did not cover attachment streams: a disconnect mid-attachment left the for await rejection unhandled, the attachment's Future pending forever (it was typed Future<…, never> and could not reject), the message pipeline hanging, and the streamIdToAttachments entry leaked.

Addressed by settling the future at the source: the byte stream handler now catches the failed read and rejects the attachment's Future (error type widened to Error). The rejection propagates through the existing message pipeline, so the subscribe error callback records it and finalize cleans up the attachment state — no hang, no leak, no unhandled rejection, and the pipeline stays healthy for subsequent messages. The error callback is also reason-aware now: DataStreamErrorReason.AbnormalEnd logs at debug (expected disconnect churn) while any other failure logs a visible warning.

New tests cover both: an attachment ending abnormally (no unhandled rejection, message dropped, a later message with a completing attachment still delivers) and a non-disconnect attachment failure hitting the warn path.

[changeset-bot]

🦋 Changeset detected

Latest commit: ebbd8d0

The changes in this PR will be included in the next version bump.

This PR includes changesets to release 7 packages

Name	Type
@livekit/components-core	Patch
@livekit/components-react	Patch
@livekit/agents-ui	Patch
@livekit/component-example-next	Patch
@livekit/components-js-docs	Patch
@livekit/component-docs-storybook	Patch
@livekit/components-docs-gen	Patch

Not sure what this means? Click here to learn what changesets are.

Click here if you're a maintainer who wants to add another changeset to this PR

[vercel]

@anzemur is attempting to deploy a commit to the LiveKit Team on Vercel.

A member of the Team first needs to authorize it.

[1egoman]

@anzemur Thanks for the contribution! There's a few issues though which I think would need to be addressed prior to being merged:

1. The change does not correct the problem for data stream attachments.

For context, the way data streams works, the sender transmits each attachment as a separate data stream forming a "tree". The attachedStreamIds field links each attachment stream to its parent text stream, and setupChat makes one Future for each attachment. The chat message pipeline waits for all of these Future objects before resolving.

If the sender disconnects during an attachment stream, the for await loop in the byte stream handler throws an error. The byte stream handler is an async function, and livekit-client does not catch the rejection. Because of this, the error becomes an unhandled rejection, and the new error callback does not receive this error, because the error is not in the text stream.

This means the attachment Future does not resolve or reject, it stays pending indefinitely. This means the pipeline does not complete, finalize does not run, and the key stays in streamIdToAttachments permanently. The user also does not receive the chat message.

So, I don't think a top-level catch is sufficient. Since no error bubbles up to the top level, you first need to settle the attachment Future, then, the top-level handler can record the error in one place.

2. Errors in the text stream are safe, but errors in an attachment stream are not

If the text stream fails during a read, finalize runs for all error types and deletes all state in any persistent maps. This is correct today.

However, if an attachment stream fails during a read, the state stays. This is true for all error types, not only for a disconnection. So I think you would need to make this case disconenction error specific, which you can do by reading the DataStreamErrorReason enum on the error.

General thoughts on these issues? I think as part of this change it might be worthwhile to introduce at least one more (if not more) test(s) with a data stream that has an attachment and validate that errors propegate properly in the above situations.

[1egoman]

Thanks for the continued progress on this! I think you are on the right track, but there's a few small bugs with with what you have now which I think would need to be addressed prior to being merged.

[1egoman]

I suspect here you want something like the below, otherwise if you have multiple attachments which are included on chat message, there's a narrow period of time in the concatMap in chat.ts line 134 where if a supposed attachment index > 0 were to reject, it would get raised as an unhandled rejection. I think to work around this, you'd want something like the below here:

Suggested change

	const attachments = new Map(
	(attachedStreamIds ?? []).map((id) => [
	id,
	new Future<{ fileName: string; mimeType: string; buffer: Array<Uint8Array> }, never>(),
	new Future<{ fileName: string; mimeType: string; buffer: Array<Uint8Array> }, Error>(),
	]),
	);
	const attachments = new Map(
	(attachedStreamIds ?? []).map((id) => {
	const future = new Future<{ fileName: string; mimeType: string; buffer: Array<Uint8Array> }, Error>();
	future.promise.catch(() => {}); // Ignore emitting `unhandledRejection` if the promise rejects before the attachments `concatMap` switches to this promise
	return [id, future];
	}),
	);

Example test which I think should exercise this case:

const byteReader = (id: string, body: () => AsyncGenerator<Uint8Array>): ByteReader => ({
  info: { id, name: `${id}.bin`, mimeType: 'application/octet-stream' },
  [Symbol.asyncIterator]: body,
});

it('does not reject an attachment future before anything subscribes to it (multiple attachments)', async () => {
  const rejections: Array<string> = [];
  const onUnhandled = () => rejections.push('unhandledRejection');
  process.on('unhandledRejection', onUnhandled);
  vi.spyOn(log, 'debug').mockImplementation(() => {});
  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')!;

    await textHandler(textReader('msg-1', 'two attachments', ['att-1', 'att-2']), {
      identity: 'sender',
    });
    await settle();

    // `concatMap` subscribes to the attachment futures one at a time, so nothing is
    // listening to att-2 while att-1 is still in flight. Rejecting it here surfaces
    // globally as an unhandled rejection until att-1 settles.
    await byteHandler(
      byteReader('att-2', async function* () {
        throw new DataStreamError(
          'Participant sender unexpectedly disconnected in the middle of sending data',
          DataStreamErrorReason.AbnormalEnd,
        );
      }),
    );
    await settle();

    expect(rejections).toEqual([]);

    subscription.unsubscribe();
  } finally {
    process.off('unhandledRejection', onUnhandled);
    vi.restoreAllMocks();
  }
});

[1egoman]

Another case that I don't think your current solution handles properly - a byte stream fails before the text stream yields its first chunk which results in an unhandled rejection.

The attachment observable is only constructed inside mergeMap here, which doesn't fire until a text chunk arrives. Similarly to my other comment, if the downstream attachment is rejected before that and nobody upstream is listening with a `.catch()1, then you will get an unhandledRejection.

I think a test which should assert this:

const byteReader = (id: string, body: () => AsyncGenerator<Uint8Array>): ByteReader => ({
  info: { id, name: `${id}.bin`, mimeType: 'application/octet-stream' },
  [Symbol.asyncIterator]: body,
});

it('settles the message when an attachment byte stream never arrives at all', async () => {
  vi.spyOn(log, 'debug').mockImplementation(() => {});
  vi.spyOn(log, 'warn').mockImplementation(() => {});
  try {
    const { room, textHandlers } = makeRoom();
    const chat = setupChat(room);
    const emissions: ReceivedChatMessage[][] = [];
    const subscription = chat.messageObservable.subscribe((me
      emissions.push(messages);
    });

    const textHandler = textHandlers.get('lk.chat')!;

    // The text arrives complete, but the sender drops before
    // byte stream, so its future never settles. Nothing errors, so no error callback
    // can rescue this: the pipeline never completes, `finali
    // `streamIdToAttachments` entry is retained for the lifetime of the page.
    await textHandler(textReader('msg-3', 'text arrived, atta']), {
      identity: 'sender',
    });
    await settle();

    // The received text must not be silently swallowed. (If the preferred behaviour is
    // to drop the message instead, assert that the pipeline
    // that it must reach *some* terminal state rather than hanging forever.)
    expect(emissions.flat().map((message) => message.message)
      'text arrived, attachment never did',
    ]);

    subscription.unsubscribe();
  } finally {
    vi.restoreAllMocks();
  }
});