diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0601f9a4..e58321eb 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/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)})" diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index 53873884..bededd77 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/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) """ 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.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