From 86afe636a6f13ad39efb822813f4ccbf6242f8d8 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Mon, 19 Oct 2026 01:39:59 +0500 Subject: [PATCH 1/7] fix(rtc): fail room sid future on disconnect and deduplicate active speakers - Fix Room.sid indefinitely hanging when room disconnects before receiving SID or when called on a disconnected room (fixes #651) - Add Room.get_sid(timeout=...) to allow bounding the wait for room SID - Fail _first_sid_future cleanly with RuntimeError/ConnectError when disconnect or connect error occurs - Safely retrieve participant identity in _retrieve_participant without requiring local_participant before connecting - Safely read sid in Room.__repr__ without raising if _first_sid_future has an exception - Deduplicate active_speakers_changed events to avoid emitting duplicate or blank events when active speakers have not changed (fixes #148) - Expose Room.active_speakers property to inspect currently active speakers - Clear active_speakers and emit empty list when room disconnects or speaking participant disconnects - Isolate FfiClient._dispose in subprocess in test_ffi_handle_lifecycle to avoid invalidating the global FFI client during test runs --- livekit-rtc/livekit/rtc/room.py | 156 ++++++++++---- .../tests/test_ffi_handle_lifecycle.py | 52 +++-- tests/rtc/test_room_sid_and_speakers.py | 200 ++++++++++++++++++ 3 files changed, 353 insertions(+), 55 deletions(-) create mode 100644 tests/rtc/test_room_sid_and_speakers.py diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0601f9a4..fd421e15 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -189,6 +189,8 @@ def __init__( self._connection_state = ConnectionState.CONN_DISCONNECTED self._first_sid_future = asyncio.Future[str]() self._local_participant: LocalParticipant | None = None + self._active_speakers: list[Participant] = [] + self._is_connecting: bool = False # active incoming stream readers, keyed by FFI reader handle id self._text_stream_readers: Dict[int, TextStreamReader] = {} @@ -209,20 +211,60 @@ async def sid(self) -> str: Returns: str: The session ID of the room. + + Raises: + RuntimeError: If the room is disconnected and no SID was received. """ if self._info.sid: return self._info.sid + if self._connection_state == ConnectionState.CONN_DISCONNECTED and not self._is_connecting: + raise RuntimeError("cannot get sid of a disconnected room") + # This future is shared state for the Room. Caller-side timeouts, e.g. # asyncio.wait_for(room.sid, ...), should not cancel it for future waiters. return await asyncio.shield(self._first_sid_future) + async def get_sid(self, timeout: float | None = None) -> str: + """Asynchronously retrieves the session ID (SID) of the room with an optional timeout. + + Parameters: + timeout (float | None, optional): Maximum time in seconds to wait for the SID. + + Returns: + str: The session ID of the room. + + Raises: + RuntimeError: If the room is disconnected and no SID was received. + asyncio.TimeoutError: If the timeout expires before the SID is available. + """ + if timeout is None: + return await self.sid + return await asyncio.wait_for(self.sid, timeout) + def _resolve_first_sid(self, sid: str) -> None: # The room SID can arrive through the initial connect result or later # room update events. Resolve the first-SID waiter from any source that # carries it so waiters do not depend on event ordering. - if sid and not self._first_sid_future.done(): - self._first_sid_future.set_result(sid) + if sid: + self._info.sid = sid + if not self._first_sid_future.done(): + self._first_sid_future.set_result(sid) + + def _fail_first_sid(self, exc: Exception) -> None: + if not self._first_sid_future.done(): + self._first_sid_future.set_exception(exc) + # Retrieve the exception to suppress unretrieved future warnings if unawaited + self._first_sid_future.exception() + + @property + def active_speakers(self) -> list[Participant]: + """Gets the list of currently active speakers in the room. + + Returns: + list[Participant]: The list of active speakers. + """ + return list(self._active_speakers) @property def local_participant(self) -> LocalParticipant: @@ -469,6 +511,9 @@ def on_participant_connected(participant): """ self._server_url = url self._token = token + self._is_connecting = True + if self._first_sid_future.done() and not self._info.sid: + self._first_sid_future = asyncio.Future[str]() req = proto_ffi.FfiRequest() req.connect.url = url req.connect.token = token @@ -560,43 +605,49 @@ def on_participant_connected(participant): ), ) - queue = FfiClient.instance.queue.subscribe() try: - resp = FfiClient.instance.request(req) - cb: proto_ffi.FfiEvent = await queue.wait_for( - lambda e: e.connect.async_id == resp.connect.async_id - ) - finally: - FfiClient.instance.queue.unsubscribe(queue) + queue = FfiClient.instance.queue.subscribe() + try: + resp = FfiClient.instance.request(req) + cb: proto_ffi.FfiEvent = await queue.wait_for( + lambda e: e.connect.async_id == resp.connect.async_id + ) + finally: + FfiClient.instance.queue.unsubscribe(queue) - if cb.connect.error: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) - raise ConnectError(cb.connect.error) + if cb.connect.error: + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + raise ConnectError(cb.connect.error) - self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id) + self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id) - self._e2ee_manager = E2EEManager( - self._ffi_handle.handle, options.encryption or options.e2ee - ) + self._e2ee_manager = E2EEManager( + self._ffi_handle.handle, options.encryption or options.e2ee + ) - self._info = cb.connect.result.room.info - self._resolve_first_sid(self._info.sid) - self._connection_state = ConnectionState.CONN_CONNECTED + self._info = cb.connect.result.room.info + self._resolve_first_sid(self._info.sid) + self._connection_state = ConnectionState.CONN_CONNECTED - self._local_participant = LocalParticipant( - self._room_queue, cb.connect.result.local_participant - ) + self._local_participant = LocalParticipant( + self._room_queue, cb.connect.result.local_participant + ) - for pt in cb.connect.result.participants: - rp = self._create_remote_participant(pt.participant) + for pt in cb.connect.result.participants: + rp = self._create_remote_participant(pt.participant) - # add the initial remote participant tracks - for owned_publication_info in pt.publications: - publication = RemoteTrackPublication(owned_publication_info) - rp._track_publications[publication.sid] = publication + # add the initial remote participant tracks + for owned_publication_info in pt.publications: + publication = RemoteTrackPublication(owned_publication_info) + rp._track_publications[publication.sid] = publication - # start listening to room events - self._task = self._loop.create_task(self._listen_task()) + # start listening to room events + self._task = self._loop.create_task(self._listen_task()) + except Exception as e: + self._fail_first_sid(e) + raise + finally: + self._is_connecting = False # unsubscribe from the FFI queue once the listen task ends. # disconnect() unsubscribes too, but it early-returns when the @@ -713,6 +764,10 @@ async def disconnect( if self._connection_state != ConnectionState.CONN_DISCONNECTED: self.local_participant._info.disconnect_reason = reason self._connection_state = ConnectionState.CONN_DISCONNECTED + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + if self._active_speakers: + self._active_speakers = [] + self.emit("active_speakers_changed", []) self.emit("connection_state_changed", self._connection_state) self.emit("disconnected", reason) @@ -744,6 +799,7 @@ async def _listen_task(self) -> None: await self._drain_rpc_invocation_tasks() self._error_stream_readers() self._dispose_open_stream_writers() + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) def _on_rpc_method_invocation(self, rpc_invocation: RpcMethodInvocationEvent) -> None: if self._local_participant is None: @@ -770,9 +826,15 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self.emit("participant_connected", rparticipant) elif which == "participant_disconnected": identity = event.participant_disconnected.participant_identity - rparticipant = self._remote_participants.pop(identity) - rparticipant._info.disconnect_reason = event.participant_disconnected.disconnect_reason - self.emit("participant_disconnected", rparticipant) + if identity in self._remote_participants: + rparticipant = self._remote_participants.pop(identity) + rparticipant._info.disconnect_reason = ( + event.participant_disconnected.disconnect_reason + ) + self.emit("participant_disconnected", rparticipant) + if any(p.identity == identity for p in self._active_speakers): + self._active_speakers = [p for p in self._active_speakers if p.identity != identity] + self.emit("active_speakers_changed", list(self._active_speakers)) elif which == "participant_active": rp = self._retrieve_remote_participant(event.participant_active.participant_identity) if rp: @@ -982,10 +1044,12 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: # TODO: pass participant identity for identity in event.active_speakers_changed.participant_identities: participant = self._retrieve_participant(identity) - assert isinstance(participant, Participant) - speakers.append(participant) + if participant is not None: + speakers.append(participant) - self.emit("active_speakers_changed", speakers) + if [p.identity for p in speakers] != [p.identity for p in self._active_speakers]: + self._active_speakers = speakers + self.emit("active_speakers_changed", speakers) elif which == "room_metadata_changed": old_metadata = self.metadata self._info.metadata = event.room_metadata_changed.metadata @@ -1126,6 +1190,10 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self._connection_state = connection_state self.emit("connection_state_changed", connection_state) elif which == "disconnected": + self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + if self._active_speakers: + self._active_speakers = [] + self.emit("active_speakers_changed", []) self.emit("disconnected", event.disconnected.reason) elif which == "reconnecting": self.emit("reconnecting") @@ -1243,8 +1311,12 @@ def _retrieve_remote_participant(self, identity: str) -> Optional[RemoteParticip def _retrieve_participant(self, identity: str) -> Optional[Participant]: """Retrieve a local or remote participant by identity""" - if identity and identity == self.local_participant.identity: - return self.local_participant + if ( + identity + and self._local_participant is not None + and identity == self._local_participant.identity + ): + return self._local_participant return self._retrieve_remote_participant(identity) @@ -1260,7 +1332,13 @@ def _create_remote_participant( def __repr__(self) -> str: sid = "unknown" - if self._first_sid_future.done(): + if self._info.sid: + sid = self._info.sid + elif ( + self._first_sid_future.done() + and not self._first_sid_future.cancelled() + and self._first_sid_future.exception() is None + ): sid = self._first_sid_future.result() return f"rtc.Room(sid={sid}, name={self.name}, metadata={self.metadata}, connection_state={ConnectionState.Name(self._connection_state)})" diff --git a/livekit-rtc/tests/test_ffi_handle_lifecycle.py b/livekit-rtc/tests/test_ffi_handle_lifecycle.py index 08750bfa..1d4a2a58 100644 --- a/livekit-rtc/tests/test_ffi_handle_lifecycle.py +++ b/livekit-rtc/tests/test_ffi_handle_lifecycle.py @@ -14,7 +14,6 @@ from __future__ import annotations -import gc import subprocess import sys @@ -43,25 +42,46 @@ def test_active_runtime_invalid_handle_asserts() -> None: def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None: - handle_id = _allocate_real_handle() - handle = FfiHandle(handle_id) - assert not handle.disposed + code = """ +import gc +from livekit.rtc._ffi_client import FfiClient, FfiHandle +from livekit.rtc._proto import audio_frame_pb2, ffi_pb2 + +req = ffi_pb2.FfiRequest() +req.new_audio_source.type = audio_frame_pb2.AUDIO_SOURCE_NATIVE +req.new_audio_source.sample_rate = 48000 +req.new_audio_source.num_channels = 1 +req.new_audio_source.queue_size_ms = 100 +resp = FfiClient.instance.request(req) +handle_id = int(resp.new_audio_source.source.handle.id) +assert handle_id != 0 + +handle = FfiHandle(handle_id) +assert not handle.disposed - # Global dispose marks all live handles consumed and sets _ffi_disposed - FfiClient.instance._dispose() - assert handle.disposed +# Global dispose marks all live handles consumed and sets _ffi_disposed +FfiClient.instance._dispose() +assert handle.disposed - # Explicit dispose or __del__ after global dispose should be a no-op and not assert - handle.dispose() - handle.__del__() +# Explicit dispose or __del__ after global dispose should be a no-op and not assert +handle.dispose() +handle.__del__() - # Newly instantiated handle after global dispose also safely no-ops without assert - late_handle = FfiHandle(12345) - late_handle.dispose() +# Newly instantiated handle after global dispose also safely no-ops without assert +late_handle = FfiHandle(12345) +late_handle.dispose() - del handle - del late_handle - gc.collect() +del handle +del late_handle +gc.collect() +""" + result = subprocess.run( + [sys.executable, "-c", code], + capture_output=True, + text=True, + ) + assert result.returncode == 0, f"Subprocess failed:\n{result.stderr}" + assert "AssertionError" not in result.stderr def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None: diff --git a/tests/rtc/test_room_sid_and_speakers.py b/tests/rtc/test_room_sid_and_speakers.py new file mode 100644 index 00000000..c1516fd2 --- /dev/null +++ b/tests/rtc/test_room_sid_and_speakers.py @@ -0,0 +1,200 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio + +import pytest + +from livekit import rtc +from livekit.rtc import ConnectionState, Participant, RemoteParticipant +from livekit.rtc._proto import participant_pb2 as proto_participant +from livekit.rtc._proto import room_pb2 as proto_room + + +@pytest.mark.asyncio +async def test_room_sid_disconnected_raises_runtime_error() -> None: + room = rtc.Room() + with pytest.raises(RuntimeError, match="cannot get sid of a disconnected room"): + await room.sid + + +@pytest.mark.asyncio +async def test_room_get_sid_timeout() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + with pytest.raises(asyncio.TimeoutError): + await room.get_sid(timeout=0.05) + + +@pytest.mark.asyncio +async def test_room_sid_resolution_and_get_sid() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def resolve_soon() -> None: + await asyncio.sleep(0.01) + room._resolve_first_sid("RM_test123") + + task = asyncio.create_task(resolve_soon()) + sid = await room.get_sid(timeout=1.0) + await task + assert sid == "RM_test123" + assert await room.sid == "RM_test123" + assert "RM_test123" in repr(room) + + +@pytest.mark.asyncio +async def test_room_sid_failed_on_disconnect() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def wait_sid() -> None: + await room.sid + + wait_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + # Room disconnects before receiving SID + room._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + + with pytest.raises(RuntimeError, match="room disconnected before receiving sid"): + await wait_task + + +def _create_mock_remote_participant(identity: str, sid: str) -> RemoteParticipant: + owned_info = proto_participant.OwnedParticipant() + owned_info.handle.id = 0 + owned_info.info.sid = sid + owned_info.info.identity = identity + p = RemoteParticipant(owned_info) + p._ffi_handle.mark_consumed() + return p + + +@pytest.mark.asyncio +async def test_active_speakers_deduplication() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + p2 = _create_mock_remote_participant("user-2", "PA_2") + room._remote_participants["user-1"] = p1 + room._remote_participants["user-2"] = p2 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # 1. First event: user-1 is speaking + ev1 = proto_room.RoomEvent() + ev1.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev1) + + assert len(emitted_speaker_identities) == 1 + assert emitted_speaker_identities[-1] == ["user-1"] + assert [p.identity for p in room.active_speakers] == ["user-1"] + + # 2. Duplicate event: user-1 still speaking (no change) -> should NOT emit + ev2 = proto_room.RoomEvent() + ev2.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev2) + + assert len(emitted_speaker_identities) == 1 + + # 3. New event: user-1 and user-2 speaking -> should emit + ev3 = proto_room.RoomEvent() + ev3.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) + room._on_room_event(ev3) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == ["user-1", "user-2"] + assert [p.identity for p in room.active_speakers] == ["user-1", "user-2"] + + # 4. Speakers stop: blank event -> should emit empty list + ev4 = proto_room.RoomEvent() + ev4.active_speakers_changed.SetInParent() + room._on_room_event(ev4) + + assert len(emitted_speaker_identities) == 3 + assert emitted_speaker_identities[-1] == [] + assert room.active_speakers == [] + + # 5. Subsequent blank event -> should NOT emit again + ev5 = proto_room.RoomEvent() + ev5.active_speakers_changed.SetInParent() + room._on_room_event(ev5) + + assert len(emitted_speaker_identities) == 3 + + +@pytest.mark.asyncio +async def test_participant_disconnected_updates_active_speakers() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + p2 = _create_mock_remote_participant("user-2", "PA_2") + room._remote_participants["user-1"] = p1 + room._remote_participants["user-2"] = p2 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # Both users speaking + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) + room._on_room_event(ev) + assert len(emitted_speaker_identities) == 1 + + # user-1 disconnects + disc_ev = proto_room.RoomEvent() + disc_ev.participant_disconnected.participant_identity = "user-1" + room._on_room_event(disc_ev) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == ["user-2"] + assert [p.identity for p in room.active_speakers] == ["user-2"] + + +@pytest.mark.asyncio +async def test_room_disconnect_clears_active_speakers() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + room._remote_participants["user-1"] = p1 + + emitted_speaker_identities: list[list[str]] = [] + + def on_speakers_changed(speakers: list[Participant]) -> None: + emitted_speaker_identities.append([p.identity for p in speakers]) + + room.on("active_speakers_changed", on_speakers_changed) + + # user-1 speaking + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev) + assert len(emitted_speaker_identities) == 1 + assert [p.identity for p in room.active_speakers] == ["user-1"] + + # Disconnect event arrives + disc_ev = proto_room.RoomEvent() + disc_ev.disconnected.reason = proto_participant.DisconnectReason.UNKNOWN_REASON + room._on_room_event(disc_ev) + + assert len(emitted_speaker_identities) == 2 + assert emitted_speaker_identities[-1] == [] + assert room.active_speakers == [] From 2bf92fbffc7d8b66cd1e93dd09d59c466a0da554 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Mon, 19 Oct 2026 01:52:35 +0500 Subject: [PATCH 2/7] fix(rtc): handle connect cancellation and emit copy of active speakers - Catch BaseException in connect to properly fail pending SID future and unsubscribe FFI queue on cancellation - Emit defensive list(speakers) copy to prevent listener mutation from corrupting cached speakers - Add regression tests for connect cancellation and speaker list immutability --- livekit-rtc/livekit/rtc/room.py | 14 +++++-- tests/rtc/test_room_sid_and_speakers.py | 54 +++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index fd421e15..fa9eb60f 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -251,7 +251,7 @@ def _resolve_first_sid(self, sid: str) -> None: if not self._first_sid_future.done(): self._first_sid_future.set_result(sid) - def _fail_first_sid(self, exc: Exception) -> None: + def _fail_first_sid(self, exc: BaseException) -> None: if not self._first_sid_future.done(): self._first_sid_future.set_exception(exc) # Retrieve the exception to suppress unretrieved future warnings if unawaited @@ -643,8 +643,14 @@ def on_participant_connected(participant): # start listening to room events self._task = self._loop.create_task(self._listen_task()) - except Exception as e: - self._fail_first_sid(e) + except BaseException as e: + if self._ffi_handle is None and hasattr(self, "_ffi_queue"): + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + self._fail_first_sid( + RuntimeError("room connection was cancelled") + if isinstance(e, asyncio.CancelledError) + else e + ) raise finally: self._is_connecting = False @@ -1049,7 +1055,7 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: if [p.identity for p in speakers] != [p.identity for p in self._active_speakers]: self._active_speakers = speakers - self.emit("active_speakers_changed", speakers) + self.emit("active_speakers_changed", list(speakers)) elif which == "room_metadata_changed": old_metadata = self.metadata self._info.metadata = event.room_metadata_changed.metadata diff --git a/tests/rtc/test_room_sid_and_speakers.py b/tests/rtc/test_room_sid_and_speakers.py index c1516fd2..7d17203e 100644 --- a/tests/rtc/test_room_sid_and_speakers.py +++ b/tests/rtc/test_room_sid_and_speakers.py @@ -18,6 +18,7 @@ from livekit import rtc from livekit.rtc import ConnectionState, Participant, RemoteParticipant +from livekit.rtc._proto import ffi_pb2 as proto_ffi from livekit.rtc._proto import participant_pb2 as proto_participant from livekit.rtc._proto import room_pb2 as proto_room @@ -198,3 +199,56 @@ def on_speakers_changed(speakers: list[Participant]) -> None: assert len(emitted_speaker_identities) == 2 assert emitted_speaker_identities[-1] == [] assert room.active_speakers == [] + + +@pytest.mark.asyncio +async def test_active_speakers_listener_mutation_does_not_corrupt_cache() -> None: + room = rtc.Room() + p1 = _create_mock_remote_participant("user-1", "PA_1") + room._remote_participants["user-1"] = p1 + + def mutating_listener(speakers: list[Participant]) -> None: + speakers.clear() + + room.on("active_speakers_changed", mutating_listener) + + ev = proto_room.RoomEvent() + ev.active_speakers_changed.participant_identities.append("user-1") + room._on_room_event(ev) + + assert len(room.active_speakers) == 1 + assert room.active_speakers[0].identity == "user-1" + + +@pytest.mark.asyncio +async def test_connect_cancellation_fails_sid_future_and_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = rtc.Room() + + # Mock FfiClient request to never answer connect response + def mock_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + resp = proto_ffi.FfiResponse() + resp.connect.async_id = 9999 + return resp + + monkeypatch.setattr(rtc._ffi_client.FfiClient.instance, "request", mock_request) + + connect_task = asyncio.create_task(room.connect("ws://dummy", "dummy_token")) + await asyncio.sleep(0.01) + + assert room._is_connecting + + async def wait_sid() -> None: + await room.sid + + sid_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + connect_task.cancel() + with pytest.raises(asyncio.CancelledError): + await connect_task + + assert not room._is_connecting + with pytest.raises(RuntimeError, match="room connection was cancelled"): + await sid_task From 2762cbe337eaf5c3eab7c35f18fc034de4b046cd Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:17:02 +0500 Subject: [PATCH 3/7] fix(rtc): scope PR to Room.sid hang fix and remove get_sid / active speakers logic --- livekit-rtc/livekit/rtc/room.py | 40 +--- tests/rtc/test_room_sid.py | 96 +++++++++ tests/rtc/test_room_sid_and_speakers.py | 254 ------------------------ 3 files changed, 97 insertions(+), 293 deletions(-) create mode 100644 tests/rtc/test_room_sid.py delete mode 100644 tests/rtc/test_room_sid_and_speakers.py diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index fa9eb60f..a15786e3 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -189,7 +189,6 @@ def __init__( self._connection_state = ConnectionState.CONN_DISCONNECTED self._first_sid_future = asyncio.Future[str]() self._local_participant: LocalParticipant | None = None - self._active_speakers: list[Participant] = [] self._is_connecting: bool = False # active incoming stream readers, keyed by FFI reader handle id @@ -225,23 +224,6 @@ async def sid(self) -> str: # asyncio.wait_for(room.sid, ...), should not cancel it for future waiters. return await asyncio.shield(self._first_sid_future) - async def get_sid(self, timeout: float | None = None) -> str: - """Asynchronously retrieves the session ID (SID) of the room with an optional timeout. - - Parameters: - timeout (float | None, optional): Maximum time in seconds to wait for the SID. - - Returns: - str: The session ID of the room. - - Raises: - RuntimeError: If the room is disconnected and no SID was received. - asyncio.TimeoutError: If the timeout expires before the SID is available. - """ - if timeout is None: - return await self.sid - return await asyncio.wait_for(self.sid, timeout) - def _resolve_first_sid(self, sid: str) -> None: # The room SID can arrive through the initial connect result or later # room update events. Resolve the first-SID waiter from any source that @@ -257,15 +239,6 @@ def _fail_first_sid(self, exc: BaseException) -> None: # Retrieve the exception to suppress unretrieved future warnings if unawaited self._first_sid_future.exception() - @property - def active_speakers(self) -> list[Participant]: - """Gets the list of currently active speakers in the room. - - Returns: - list[Participant]: The list of active speakers. - """ - return list(self._active_speakers) - @property def local_participant(self) -> LocalParticipant: """Gets the local participant in the room. @@ -771,9 +744,6 @@ async def disconnect( self.local_participant._info.disconnect_reason = reason self._connection_state = ConnectionState.CONN_DISCONNECTED self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) - if self._active_speakers: - self._active_speakers = [] - self.emit("active_speakers_changed", []) self.emit("connection_state_changed", self._connection_state) self.emit("disconnected", reason) @@ -838,9 +808,6 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: event.participant_disconnected.disconnect_reason ) self.emit("participant_disconnected", rparticipant) - if any(p.identity == identity for p in self._active_speakers): - self._active_speakers = [p for p in self._active_speakers if p.identity != identity] - self.emit("active_speakers_changed", list(self._active_speakers)) elif which == "participant_active": rp = self._retrieve_remote_participant(event.participant_active.participant_identity) if rp: @@ -1053,9 +1020,7 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: if participant is not None: speakers.append(participant) - if [p.identity for p in speakers] != [p.identity for p in self._active_speakers]: - self._active_speakers = speakers - self.emit("active_speakers_changed", list(speakers)) + self.emit("active_speakers_changed", speakers) elif which == "room_metadata_changed": old_metadata = self.metadata self._info.metadata = event.room_metadata_changed.metadata @@ -1197,9 +1162,6 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self.emit("connection_state_changed", connection_state) elif which == "disconnected": self._fail_first_sid(RuntimeError("room disconnected before receiving sid")) - if self._active_speakers: - self._active_speakers = [] - self.emit("active_speakers_changed", []) self.emit("disconnected", event.disconnected.reason) elif which == "reconnecting": self.emit("reconnecting") diff --git a/tests/rtc/test_room_sid.py b/tests/rtc/test_room_sid.py new file mode 100644 index 00000000..72bc77a6 --- /dev/null +++ b/tests/rtc/test_room_sid.py @@ -0,0 +1,96 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import asyncio + +import pytest + +from livekit import rtc +from livekit.rtc import ConnectionState +from livekit.rtc._proto import ffi_pb2 as proto_ffi + + +@pytest.mark.asyncio +async def test_room_sid_disconnected_raises_runtime_error() -> None: + room = rtc.Room() + with pytest.raises(RuntimeError, match="cannot get sid of a disconnected room"): + await room.sid + + +@pytest.mark.asyncio +async def test_room_sid_resolution() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def resolve_soon() -> None: + await asyncio.sleep(0.01) + room._resolve_first_sid("RM_test123") + + task = asyncio.create_task(resolve_soon()) + sid = await room.sid + await task + assert sid == "RM_test123" + assert "RM_test123" in repr(room) + + +@pytest.mark.asyncio +async def test_room_sid_failed_on_disconnect() -> None: + room = rtc.Room() + room._connection_state = ConnectionState.CONN_CONNECTED + + async def wait_sid() -> None: + await room.sid + + wait_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + # Room disconnects before receiving SID + room._fail_first_sid(RuntimeError("room disconnected before receiving sid")) + + with pytest.raises(RuntimeError, match="room disconnected before receiving sid"): + await wait_task + + +@pytest.mark.asyncio +async def test_connect_cancellation_fails_sid_future_and_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = rtc.Room() + + # Mock FfiClient request to never answer connect response + def mock_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + resp = proto_ffi.FfiResponse() + resp.connect.async_id = 9999 + return resp + + monkeypatch.setattr(rtc._ffi_client.FfiClient.instance, "request", mock_request) + + connect_task = asyncio.create_task(room.connect("ws://dummy", "dummy_token")) + await asyncio.sleep(0.01) + + assert room._is_connecting + + async def wait_sid() -> None: + await room.sid + + sid_task = asyncio.create_task(wait_sid()) + await asyncio.sleep(0.01) + + connect_task.cancel() + with pytest.raises(asyncio.CancelledError): + await connect_task + + assert not room._is_connecting + with pytest.raises(RuntimeError, match="room connection was cancelled"): + await sid_task diff --git a/tests/rtc/test_room_sid_and_speakers.py b/tests/rtc/test_room_sid_and_speakers.py deleted file mode 100644 index 7d17203e..00000000 --- a/tests/rtc/test_room_sid_and_speakers.py +++ /dev/null @@ -1,254 +0,0 @@ -# Copyright 2026 LiveKit, Inc. -# -# Licensed under the Apache License, Version 2.0 (the "License"); -# you may not use this file except in compliance with the License. -# You may obtain a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, -# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. -# See the License for the specific language governing permissions and -# limitations under the License. - -import asyncio - -import pytest - -from livekit import rtc -from livekit.rtc import ConnectionState, Participant, RemoteParticipant -from livekit.rtc._proto import ffi_pb2 as proto_ffi -from livekit.rtc._proto import participant_pb2 as proto_participant -from livekit.rtc._proto import room_pb2 as proto_room - - -@pytest.mark.asyncio -async def test_room_sid_disconnected_raises_runtime_error() -> None: - room = rtc.Room() - with pytest.raises(RuntimeError, match="cannot get sid of a disconnected room"): - await room.sid - - -@pytest.mark.asyncio -async def test_room_get_sid_timeout() -> None: - room = rtc.Room() - room._connection_state = ConnectionState.CONN_CONNECTED - with pytest.raises(asyncio.TimeoutError): - await room.get_sid(timeout=0.05) - - -@pytest.mark.asyncio -async def test_room_sid_resolution_and_get_sid() -> None: - room = rtc.Room() - room._connection_state = ConnectionState.CONN_CONNECTED - - async def resolve_soon() -> None: - await asyncio.sleep(0.01) - room._resolve_first_sid("RM_test123") - - task = asyncio.create_task(resolve_soon()) - sid = await room.get_sid(timeout=1.0) - await task - assert sid == "RM_test123" - assert await room.sid == "RM_test123" - assert "RM_test123" in repr(room) - - -@pytest.mark.asyncio -async def test_room_sid_failed_on_disconnect() -> None: - room = rtc.Room() - room._connection_state = ConnectionState.CONN_CONNECTED - - async def wait_sid() -> None: - await room.sid - - wait_task = asyncio.create_task(wait_sid()) - await asyncio.sleep(0.01) - - # Room disconnects before receiving SID - room._fail_first_sid(RuntimeError("room disconnected before receiving sid")) - - with pytest.raises(RuntimeError, match="room disconnected before receiving sid"): - await wait_task - - -def _create_mock_remote_participant(identity: str, sid: str) -> RemoteParticipant: - owned_info = proto_participant.OwnedParticipant() - owned_info.handle.id = 0 - owned_info.info.sid = sid - owned_info.info.identity = identity - p = RemoteParticipant(owned_info) - p._ffi_handle.mark_consumed() - return p - - -@pytest.mark.asyncio -async def test_active_speakers_deduplication() -> None: - room = rtc.Room() - p1 = _create_mock_remote_participant("user-1", "PA_1") - p2 = _create_mock_remote_participant("user-2", "PA_2") - room._remote_participants["user-1"] = p1 - room._remote_participants["user-2"] = p2 - - emitted_speaker_identities: list[list[str]] = [] - - def on_speakers_changed(speakers: list[Participant]) -> None: - emitted_speaker_identities.append([p.identity for p in speakers]) - - room.on("active_speakers_changed", on_speakers_changed) - - # 1. First event: user-1 is speaking - ev1 = proto_room.RoomEvent() - ev1.active_speakers_changed.participant_identities.append("user-1") - room._on_room_event(ev1) - - assert len(emitted_speaker_identities) == 1 - assert emitted_speaker_identities[-1] == ["user-1"] - assert [p.identity for p in room.active_speakers] == ["user-1"] - - # 2. Duplicate event: user-1 still speaking (no change) -> should NOT emit - ev2 = proto_room.RoomEvent() - ev2.active_speakers_changed.participant_identities.append("user-1") - room._on_room_event(ev2) - - assert len(emitted_speaker_identities) == 1 - - # 3. New event: user-1 and user-2 speaking -> should emit - ev3 = proto_room.RoomEvent() - ev3.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) - room._on_room_event(ev3) - - assert len(emitted_speaker_identities) == 2 - assert emitted_speaker_identities[-1] == ["user-1", "user-2"] - assert [p.identity for p in room.active_speakers] == ["user-1", "user-2"] - - # 4. Speakers stop: blank event -> should emit empty list - ev4 = proto_room.RoomEvent() - ev4.active_speakers_changed.SetInParent() - room._on_room_event(ev4) - - assert len(emitted_speaker_identities) == 3 - assert emitted_speaker_identities[-1] == [] - assert room.active_speakers == [] - - # 5. Subsequent blank event -> should NOT emit again - ev5 = proto_room.RoomEvent() - ev5.active_speakers_changed.SetInParent() - room._on_room_event(ev5) - - assert len(emitted_speaker_identities) == 3 - - -@pytest.mark.asyncio -async def test_participant_disconnected_updates_active_speakers() -> None: - room = rtc.Room() - p1 = _create_mock_remote_participant("user-1", "PA_1") - p2 = _create_mock_remote_participant("user-2", "PA_2") - room._remote_participants["user-1"] = p1 - room._remote_participants["user-2"] = p2 - - emitted_speaker_identities: list[list[str]] = [] - - def on_speakers_changed(speakers: list[Participant]) -> None: - emitted_speaker_identities.append([p.identity for p in speakers]) - - room.on("active_speakers_changed", on_speakers_changed) - - # Both users speaking - ev = proto_room.RoomEvent() - ev.active_speakers_changed.participant_identities.extend(["user-1", "user-2"]) - room._on_room_event(ev) - assert len(emitted_speaker_identities) == 1 - - # user-1 disconnects - disc_ev = proto_room.RoomEvent() - disc_ev.participant_disconnected.participant_identity = "user-1" - room._on_room_event(disc_ev) - - assert len(emitted_speaker_identities) == 2 - assert emitted_speaker_identities[-1] == ["user-2"] - assert [p.identity for p in room.active_speakers] == ["user-2"] - - -@pytest.mark.asyncio -async def test_room_disconnect_clears_active_speakers() -> None: - room = rtc.Room() - p1 = _create_mock_remote_participant("user-1", "PA_1") - room._remote_participants["user-1"] = p1 - - emitted_speaker_identities: list[list[str]] = [] - - def on_speakers_changed(speakers: list[Participant]) -> None: - emitted_speaker_identities.append([p.identity for p in speakers]) - - room.on("active_speakers_changed", on_speakers_changed) - - # user-1 speaking - ev = proto_room.RoomEvent() - ev.active_speakers_changed.participant_identities.append("user-1") - room._on_room_event(ev) - assert len(emitted_speaker_identities) == 1 - assert [p.identity for p in room.active_speakers] == ["user-1"] - - # Disconnect event arrives - disc_ev = proto_room.RoomEvent() - disc_ev.disconnected.reason = proto_participant.DisconnectReason.UNKNOWN_REASON - room._on_room_event(disc_ev) - - assert len(emitted_speaker_identities) == 2 - assert emitted_speaker_identities[-1] == [] - assert room.active_speakers == [] - - -@pytest.mark.asyncio -async def test_active_speakers_listener_mutation_does_not_corrupt_cache() -> None: - room = rtc.Room() - p1 = _create_mock_remote_participant("user-1", "PA_1") - room._remote_participants["user-1"] = p1 - - def mutating_listener(speakers: list[Participant]) -> None: - speakers.clear() - - room.on("active_speakers_changed", mutating_listener) - - ev = proto_room.RoomEvent() - ev.active_speakers_changed.participant_identities.append("user-1") - room._on_room_event(ev) - - assert len(room.active_speakers) == 1 - assert room.active_speakers[0].identity == "user-1" - - -@pytest.mark.asyncio -async def test_connect_cancellation_fails_sid_future_and_cleans_up( - monkeypatch: pytest.MonkeyPatch, -) -> None: - room = rtc.Room() - - # Mock FfiClient request to never answer connect response - def mock_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: - resp = proto_ffi.FfiResponse() - resp.connect.async_id = 9999 - return resp - - monkeypatch.setattr(rtc._ffi_client.FfiClient.instance, "request", mock_request) - - connect_task = asyncio.create_task(room.connect("ws://dummy", "dummy_token")) - await asyncio.sleep(0.01) - - assert room._is_connecting - - async def wait_sid() -> None: - await room.sid - - sid_task = asyncio.create_task(wait_sid()) - await asyncio.sleep(0.01) - - connect_task.cancel() - with pytest.raises(asyncio.CancelledError): - await connect_task - - assert not room._is_connecting - with pytest.raises(RuntimeError, match="room connection was cancelled"): - await sid_task From f1e0e007c84a7ef1632e06f300b8874ba9a6ea57 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:41:12 +0500 Subject: [PATCH 4/7] fix(rtc): restore upstream speaker events and reduce apm iterations in gc test --- livekit-rtc/livekit/rtc/room.py | 13 +++++-------- livekit-rtc/tests/test_ffi_client.py | 2 +- 2 files changed, 6 insertions(+), 9 deletions(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index a15786e3..e58321eb 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -802,12 +802,9 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: self.emit("participant_connected", rparticipant) elif which == "participant_disconnected": identity = event.participant_disconnected.participant_identity - if identity in self._remote_participants: - rparticipant = self._remote_participants.pop(identity) - rparticipant._info.disconnect_reason = ( - event.participant_disconnected.disconnect_reason - ) - self.emit("participant_disconnected", rparticipant) + rparticipant = self._remote_participants.pop(identity) + rparticipant._info.disconnect_reason = event.participant_disconnected.disconnect_reason + self.emit("participant_disconnected", rparticipant) elif which == "participant_active": rp = self._retrieve_remote_participant(event.participant_active.participant_identity) if rp: @@ -1017,8 +1014,8 @@ def _on_room_event(self, event: proto_room.RoomEvent) -> None: # TODO: pass participant identity for identity in event.active_speakers_changed.participant_identities: participant = self._retrieve_participant(identity) - if participant is not None: - speakers.append(participant) + assert isinstance(participant, Participant) + speakers.append(participant) self.emit("active_speakers_changed", speakers) elif which == "room_metadata_changed": diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index 53873884..c57927ce 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -27,7 +27,7 @@ gc.collect() gc.set_debug(gc.DEBUG_SAVEALL) -for _ in range(50): +for _ in range(10): rtc.AudioProcessingModule() gc.collect() print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage)) From d567e4ac2bdc80de0b369ef2fd13d2f2907081a8 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:47:47 +0500 Subject: [PATCH 5/7] test(rtc): dispose apm handle in ctypes array gc test to prevent sigabrt on unix runners --- livekit-rtc/tests/test_ffi_client.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index c57927ce..32751ebc 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -28,7 +28,8 @@ gc.collect() gc.set_debug(gc.DEBUG_SAVEALL) for _ in range(10): - rtc.AudioProcessingModule() + apm = rtc.AudioProcessingModule() + apm._ffi_handle.dispose() gc.collect() print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage)) """ From 500b900a8ca8e5a21d183097d362d126f1348a19 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:55:24 +0500 Subject: [PATCH 6/7] test(rtc): clear gc garbage and reset debug flags in ctypes array test --- livekit-rtc/tests/test_ffi_client.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index 32751ebc..29d6f9ab 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -27,11 +27,14 @@ gc.collect() gc.set_debug(gc.DEBUG_SAVEALL) -for _ in range(10): +for _ in range(5): apm = rtc.AudioProcessingModule() apm._ffi_handle.dispose() gc.collect() -print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage)) +count = sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage) +gc.set_debug(0) +gc.garbage.clear() +print(count) """ From ced3c84362dc87aabb1430bc93a2b2160c7b921c Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 01:07:41 +0500 Subject: [PATCH 7/7] test(rtc): use os._exit in ctypes array test subprocess to prevent webrtc apm sigabrt on unix runners --- livekit-rtc/tests/test_ffi_client.py | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index 29d6f9ab..bededd77 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -22,19 +22,20 @@ _COUNT_ARRAY_TYPES_IN_GARBAGE = """ import ctypes import gc +import os +import sys from livekit import rtc gc.collect() gc.set_debug(gc.DEBUG_SAVEALL) -for _ in range(5): - apm = rtc.AudioProcessingModule() - apm._ffi_handle.dispose() +for _ in range(10): + rtc.AudioProcessingModule() gc.collect() count = sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage) -gc.set_debug(0) -gc.garbage.clear() -print(count) +sys.stdout.write(f"{count}\\n") +sys.stdout.flush() +os._exit(0) """