fix(rtc): avoid Room.sid hang on disconnect and deduplicate active speakers

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: 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
+            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,55 @@ 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)
-
-        if cb.connect.error:
-            FfiClient.instance.queue.unsubscribe(self._ffi_queue)
-            raise ConnectError(cb.connect.error)
+            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)
 
-        self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id)
+            if cb.connect.error:
+                FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+                raise ConnectError(cb.connect.error)
 
-        self._e2ee_manager = E2EEManager(
-            self._ffi_handle.handle, options.encryption or options.e2ee
-        )
+            self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id)
 
-        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._e2ee_manager = E2EEManager(
+                self._ffi_handle.handle, options.encryption or options.e2ee
+            )
 
-        for pt in cb.connect.result.participants:
-            rp = self._create_remote_participant(pt.participant)
+            self._info = cb.connect.result.room.info
+            self._resolve_first_sid(self._info.sid)
+            self._connection_state = ConnectionState.CONN_CONNECTED
 
-            # add the initial remote participant tracks
-            for owned_publication_info in pt.publications:
-                publication = RemoteTrackPublication(owned_publication_info)
-                rp._track_publications[publication.sid] = publication
+            self._local_participant = LocalParticipant(
+                self._room_queue, cb.connect.result.local_participant
+            )
 
-        # start listening to room events
-        self._task = self._loop.create_task(self._listen_task())
+            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
+
+            # start listening to room events
+            self._task = self._loop.create_task(self._listen_task())
+        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
 
         # unsubscribe from the FFI queue once the listen task ends.
         # disconnect() unsubscribes too, but it early-returns when the
@@ -713,6 +770,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 +805,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 +832,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 +1050,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", list(speakers))
         elif which == "room_metadata_changed":
             old_metadata = self.metadata
             self._info.metadata = event.room_metadata_changed.metadata
@@ -1126,6 +1196,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 +1317,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 +1338,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)})"

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: