fix(rtc): avoid Room.sid hang when disconnected or cancelled (#651)

livekit-rtc/livekit/rtc/room.py

@@ -189,6 +189,7 @@ def __init__(
         self._connection_state = ConnectionState.CONN_DISCONNECTED
         self._first_sid_future = asyncio.Future[str]()
         self._local_participant: LocalParticipant | None = None
+        self._is_connecting: bool = False
 
         # active incoming stream readers, keyed by FFI reader handle id
         self._text_stream_readers: Dict[int, TextStreamReader] = {}
@@ -209,10 +210,16 @@ 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)
@@ -221,8 +228,16 @@ 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 local_participant(self) -> LocalParticipant:
@@ -469,6 +484,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 +578,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)
-
-        self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id)
+            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._e2ee_manager = E2EEManager(
-            self._ffi_handle.handle, options.encryption or options.e2ee
-        )
+            if cb.connect.error:
+                FfiClient.instance.queue.unsubscribe(self._ffi_queue)
+                raise ConnectError(cb.connect.error)
 
-        self._info = cb.connect.result.room.info
-        self._resolve_first_sid(self._info.sid)
-        self._connection_state = ConnectionState.CONN_CONNECTED
+            self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id)
 
-        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 +743,7 @@ 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"))
             self.emit("connection_state_changed", self._connection_state)
             self.emit("disconnected", reason)
 
@@ -744,6 +775,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:
@@ -1126,6 +1158,7 @@ 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"))
             self.emit("disconnected", event.disconnected.reason)
         elif which == "reconnecting":
             self.emit("reconnecting")
@@ -1243,8 +1276,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 +1297,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_client.py

@@ -22,15 +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(50):
+for _ in range(10):
     rtc.AudioProcessingModule()
 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)
+sys.stdout.write(f"{count}\\n")
+sys.stdout.flush()
+os._exit(0)
 """
 
 

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:

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