fix(core): keep the cached text stream observable across a room disconnect

[daniel1014]

Summary

setupTextStream caches one observable per room:topic in a global Map, registers the handler in tap({ subscribe }) and unregisters it in finalize (refcounted via share()). But the RoomEvent.Disconnected listener also deletes the cache entry:

room.on(RoomEvent.Disconnected, () => {
  getObservableCache().delete(cacheKey);   // <-- leftover
  textStreams = [];
  textStreamsSubject.next([]);
});

Deleting the entry does not invalidate consumers that already hold the observable. So on a reused Room instance:

1. Consumer A subscribes while connected — handler registered.
2. The room disconnects. A unsubscribes (useTextStream passes undefined when disconnected), the handler is unregistered, and the cache entry is dropped — but A still holds observable wip #1.
3. Consumer B mounts, or an existing consumer changes its topic (useMemo(..., [room, topic]) rebuilds without a remount), while disconnected → cache miss → observable Restructure disconnect #2 for the same topic.
4. The room reconnects. Both observables' tap({ subscribe }) fire → registerTextStreamHandler is called twice → livekit-client throws:

DataStreamError: A text stream handler for topic "lk.transcription" has already been set.

The second subscription dies permanently: that topic never delivers text again for the life of the page, while everything else on the room (tracks, other topics) keeps working — which makes it look like a UI bug rather than a stream-handler collision.

This is a leftover from #1188 ("don't unregister stream handler on disconnect"), whose stated goal is exactly this scenario — "This ensures that a room instance can be reused and the transcription handler stays registered even after a disconnect." That PR correctly introduced the refcount pattern, but kept the getObservableCache().delete(cacheKey) line from the pre-refactor version (where the subject was also completed, so eviction was correct). With the subject now long-lived, the eviction is what breaks reuse.

Fix

Per @lukasIO's review — key the cache on the Room instance itself instead of evicting on disconnect:

const observableCache = new WeakMap<Room, Map<string, Observable<TextStreamData[]>>>();

The entries for a room are collected with the room, so nothing is retained indefinitely, and a reused room keeps its observables across any number of connect/disconnect cycles.

That makes the RoomEvent.Disconnected listener unnecessary and it is removed entirely:

• Cache eviction is now handled by the WeakMap.
• Buffer reset moves into tap({ subscribe }). share() resets on refcount zero, so that callback runs once per subscription window — the first subscriber, and again after every reconnect — which is exactly where a fresh buffer belongs. React consumers already reset their own state on disconnect: useTextStream swaps the observable for undefined, and useObservableState resets to startWith whenever the observable identity changes, so dropping the next([]) emission is not observable to them.

Removing the listener also fixes a leak introduced by #1188, which changed room.once → room.on without a matching off: every cache-key rebuild added another permanent Disconnected listener.

This also deletes the roomInstanceMap / nextRoomId string-key machinery, which only existed to synthesise a key the WeakMap now provides directly.

Repro

Reproduced in an app that mounts useTextStream consumers under a <LiveKitRoom connect={...}> toggle: connect → disconnect → change a consumer's topic (or mount a new one) → reconnect. The regression tests are the minimal version of that, using a fake Room that mirrors livekit-client's one-handler-per-topic contract.

Test plan

New packages/core/src/components/textStream.test.ts:

• reuses one observable per room and topic across disconnect/reconnect — red on main, and it fails with the production error rather than a bare identity mismatch:

  AssertionError: expected "vi.fn()" to be called 1 times, but got 2 times
  Unhandled Errors: A text stream handler for topic "lk.transcription" has already been set.
  

• starts each subscription window with an empty buffer — also red on main (expected [ 2 ] to deeply equal [ 1 ]); guards the buffer-reset behaviour that moved into tap({ subscribe }).
• caches per room instance, so a second room gets its own observable — guards the new WeakMap keying.

Baseline on main (fake room unchanged): Tests 2 failed | 1 passed (3). With the fix: Tests 3 passed (3).

Suites:

• packages/core: pnpm test → Test Files 10 passed (10) / Tests 101 passed (101); tsc --noEmit exit 0; pnpm lint reports only the 14 pre-existing warnings, none in the touched files.
• packages/react: pnpm test → Test Files 4 passed (4) / Tests 14 passed (14).

Changeset included (patch on @livekit/components-core).

[vercel]

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

A member of the Team first needs to authorize it.

[changeset-bot]

🦋 Changeset detected

Latest commit: d704333

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

[CLAassistant]

All committers have signed the CLA.

[daniel1014]

Thank you for your submission! We really appreciate it. Like many open source projects, we ask that you sign our Contributor License Agreement before we can accept your contribution.You have signed the CLA already but the status is still pending? Let us recheck it.

Btw I just tried this link to sign the required CLA but the link appears broke blocking me to sign anything on it. It will be really appreciated to have this PR being reviewed and merged as it's a blocking issue around livekit cache (see the details above) requiring manual workaround patch.

[lukasIO]

Thanks for the PR!
Confirmed it's a bug.

However the fix as you suggest would keep the cache entries around indefinitely.

I'd suggest instead to use a WeakMap keyed on the Room instances like

const cache: WeakMap<Room, Map<string, Observable<TextStreamData[]>>>

so that the entries can get dropped if the Room instance isn't used anymore. This also allows us to drop the RoomEvent.Disconnected handler entirely (buffer reset can move into tap({ subscribe }).

Let me know if you want to tackle that yourself or if I should create a PR with the fix.

[lukasIO]

Btw I just tried this link to sign the required CLA but the link appears broke

This is unexpected, do you get an error message that I can use to investigate what might be wrong?

[daniel1014]

Thanks! I went with the WeakMap keyed on Room and dropped the Disconnected listener entirely as you suggested;
the buffer reset now lives in tap({ subscribe }), which share() runs once per subscription window. That also let me delete the roomInstanceMap/nextRoomId key machinery. Both regression tests are red on main (the first one fails with the real A text stream handler ... has already been set error). Pushed in d704333.

Re the CLA: it shows as signed now — the badge was stale at the time. Sorry for the noise on that.