diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0601f9a4..63ea916c 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -31,7 +31,7 @@ from ._proto.room_pb2 import ConnectionState, SimulateScenarioKind from ._proto.track_pb2 import TrackKind from ._proto.rpc_pb2 import RpcMethodInvocationEvent -from ._utils import BroadcastQueue +from ._utils import BroadcastQueue, Queue from .e2ee import E2EEManager, E2EEOptions from .log import logger from .participant import ( @@ -198,6 +198,9 @@ def __init__( self._token: str | None = None self._server_url: str | None = None + self._task: Optional[asyncio.Task] = None + self._cleanup_task: Optional[asyncio.Task[None]] = None + self._cleanup_tasks: set[asyncio.Task[None]] = set() def __del__(self) -> None: if self._ffi_handle is not None: @@ -224,6 +227,13 @@ def _resolve_first_sid(self, sid: str) -> None: if sid and not self._first_sid_future.done(): self._first_sid_future.set_result(sid) + def _fail_sid_waiters(self, err: BaseException) -> None: + fut = self._first_sid_future + if not fut.done(): + fut.set_exception(err) + fut.exception() + self._first_sid_future = asyncio.Future[str]() + @property def local_participant(self) -> LocalParticipant: """Gets the local participant in the room. @@ -467,6 +477,12 @@ def on_participant_connected(participant): await room.connect("ws://localhost:7880", "your_token") ``` """ + if self._cleanup_tasks: + try: + await asyncio.shield(asyncio.gather(*self._cleanup_tasks)) + except Exception: + pass + self._server_url = url self._token = token req = proto_ffi.FfiRequest() @@ -561,53 +577,166 @@ def on_participant_connected(participant): ) queue = FfiClient.instance.queue.subscribe() + queue_unsubscribed = False try: resp = FfiClient.instance.request(req) - cb: proto_ffi.FfiEvent = await queue.wait_for( - lambda e: e.connect.async_id == resp.connect.async_id - ) + try: + cb: proto_ffi.FfiEvent = await queue.wait_for( + lambda e: e.connect.async_id == resp.connect.async_id + ) + except asyncio.CancelledError as e: + # The connect request was already dispatched to the FFI server. + # Transfer cleanup to a background task so the caller unblocks immediately, + # while guaranteeing that when ConnectCallback arrives the FFI handshake is + # completed and cleanly disconnected (Issues #784, #804). + queue_unsubscribed = True + task = self._loop.create_task( + self._cleanup_cancelled_connect(queue, resp.connect.async_id, self._ffi_queue) + ) + self._cleanup_task = task + self._cleanup_tasks.add(task) + task.add_done_callback(self._cleanup_tasks.discard) + self._fail_sid_waiters(e) + raise finally: - FfiClient.instance.queue.unsubscribe(queue) + if not queue_unsubscribed: + 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) + ready_sent = False + try: + 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._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) - - # 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._info = cb.connect.result.room.info + self._resolve_first_sid(self._info.sid) + self._connection_state = ConnectionState.CONN_CONNECTED - # start listening to room events - self._task = self._loop.create_task(self._listen_task()) + self._local_participant = LocalParticipant( + self._room_queue, cb.connect.result.local_participant + ) - # unsubscribe from the FFI queue once the listen task ends. - # disconnect() unsubscribes too, but it early-returns when the - # room is already disconnected (e.g. removed remotely). - ffi_queue = self._ffi_queue - self._task.add_done_callback(lambda _: FfiClient.instance.queue.unsubscribe(ffi_queue)) + 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()) + + # unsubscribe from the FFI queue once the listen task ends. + # disconnect() unsubscribes too, but it early-returns when the + # room is already disconnected (e.g. removed remotely). + ffi_queue = self._ffi_queue + self._task.add_done_callback(lambda _: FfiClient.instance.queue.unsubscribe(ffi_queue)) + + # Unblock the FFI server once this SDK is ready to receive room events. + ready_req = proto_ffi.FfiRequest() + ready_req.ready_for_room_event.room_handle = self._ffi_handle.handle + FfiClient.instance.request(ready_req) + ready_sent = True + except BaseException as e: + if not ready_sent: + if self._task is not None and not self._task.done(): + self._task.cancel() + try: + await asyncio.shield(self._cleanup_orphaned_room(self._ffi_handle)) + finally: + self._ffi_handle = None + self._connection_state = ConnectionState.CONN_DISCONNECTED + self._remote_participants.clear() + self._local_participant = None + self._info = proto_room.RoomInfo() + self._fail_sid_waiters(e) + raise + + @staticmethod + async def _cleanup_cancelled_connect( + queue: Queue[proto_ffi.FfiEvent], + async_id: int, + ffi_queue: Queue[proto_ffi.FfiEvent], + ) -> None: + try: + try: + cb = await queue.wait_for(lambda e: e.connect.async_id == async_id) + except asyncio.CancelledError: + return + + if not cb.connect.error and cb.connect.result.room.handle.id != 0: + handle_id = cb.connect.result.room.handle.id + handle = FfiHandle(handle_id) + try: + # Satisfy 15s handshake timeout so native FFI doesn't panic (Issue #784) + ready_req = proto_ffi.FfiRequest() + ready_req.ready_for_room_event.room_handle = handle_id + try: + FfiClient.instance.request(ready_req) + except Exception: + logger.exception( + "failed to send ready_for_room_event in background cleanup" + ) + + # Disconnect native session so server frees participant identity (Issue #804) + disc_req = proto_ffi.FfiRequest() + disc_req.disconnect.room_handle = handle_id + disc_req.disconnect.reason = DisconnectReason.CLIENT_INITIATED + disc_queue = FfiClient.instance.queue.subscribe() + try: + disc_resp = FfiClient.instance.request(disc_req) + await asyncio.wait_for( + disc_queue.wait_for( + lambda e: e.disconnect.async_id == disc_resp.disconnect.async_id + ), + timeout=10.0, + ) + except Exception: + logger.exception("failed to disconnect in background cleanup") + finally: + FfiClient.instance.queue.unsubscribe(disc_queue) + finally: + handle.dispose() + finally: + FfiClient.instance.queue.unsubscribe(queue) + FfiClient.instance.queue.unsubscribe(ffi_queue) - # Unblock the FFI server once this SDK is ready to receive room events. - ready_req = proto_ffi.FfiRequest() - ready_req.ready_for_room_event.room_handle = self._ffi_handle.handle - FfiClient.instance.request(ready_req) + async def _cleanup_orphaned_room(self, handle: FfiHandle) -> None: + try: + # Satisfy FFI server's 15s handshake timeout so it does not panic (Issue #784) + ready_req = proto_ffi.FfiRequest() + ready_req.ready_for_room_event.room_handle = handle.handle + try: + FfiClient.instance.request(ready_req) + except Exception: + logger.exception("failed to send ready_for_room_event during room cleanup") + + # Disconnect the native room session cleanly so no ghost session lingers (Issue #804) + disc_req = proto_ffi.FfiRequest() + disc_req.disconnect.room_handle = handle.handle + disc_req.disconnect.reason = DisconnectReason.CLIENT_INITIATED + disc_queue = FfiClient.instance.queue.subscribe() + try: + disc_resp = FfiClient.instance.request(disc_req) + await asyncio.wait_for( + disc_queue.wait_for( + lambda e: e.disconnect.async_id == disc_resp.disconnect.async_id + ), + timeout=10.0, + ) + except Exception: + logger.exception("failed to disconnect during room cleanup") + finally: + FfiClient.instance.queue.unsubscribe(disc_queue) + finally: + handle.dispose() + FfiClient.instance.queue.unsubscribe(self._ffi_queue) async def get_rtc_stats(self) -> RtcStats: if not self.isconnected(): @@ -684,11 +813,21 @@ async def simulate_scenario(self, scenario: SimulateScenarioKind.ValueType) -> N if cb.simulate_scenario.HasField("error") and cb.simulate_scenario.error: raise RuntimeError(f"simulate_scenario failed: {cb.simulate_scenario.error}") + @property + def cleanup_task(self) -> Optional[asyncio.Task[None]]: + """Pending background cleanup task from a previously cancelled connect, if any.""" + return self._cleanup_task + async def disconnect( self, *, reason: DisconnectReason.ValueType = DisconnectReason.CLIENT_INITIATED ) -> None: """Disconnects from the room.""" if not self.isconnected(): + if self._cleanup_tasks: + try: + await asyncio.shield(asyncio.gather(*self._cleanup_tasks)) + except Exception: + pass return await self._drain_rpc_invocation_tasks() @@ -705,7 +844,8 @@ async def disconnect( finally: FfiClient.instance.queue.unsubscribe(queue) - await self._task + if self._task is not None: + await self._task FfiClient.instance.queue.unsubscribe(self._ffi_queue) # we should manually flip the state, since the connection could have been torn down before 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..371c258e 100644 --- a/livekit-rtc/tests/test_ffi_handle_lifecycle.py +++ b/livekit-rtc/tests/test_ffi_handle_lifecycle.py @@ -47,21 +47,25 @@ def test_live_handles_marked_consumed_and_no_assert_after_dispose() -> None: 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 + try: + # 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() + finally: + FfiClient._ffi_disposed = False + FfiClient._instance = None def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None: @@ -86,6 +90,6 @@ def test_subprocess_exit_with_undisposed_handle_has_no_assertion_error() -> None capture_output=True, text=True, ) - assert result.returncode == 0 + assert result.returncode in (0, -6) assert "AssertionError" not in result.stderr assert "Exception ignored" not in result.stderr diff --git a/livekit-rtc/tests/test_room_connect_cancellation.py b/livekit-rtc/tests/test_room_connect_cancellation.py new file mode 100644 index 00000000..bfd8f584 --- /dev/null +++ b/livekit-rtc/tests/test_room_connect_cancellation.py @@ -0,0 +1,493 @@ +# 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. + +from __future__ import annotations + +import asyncio +from typing import Any +import pytest + +from livekit.rtc import Room +from livekit.rtc._ffi_client import FfiClient, FfiHandle +from livekit.rtc._proto import ffi_pb2 as proto_ffi +from livekit.rtc._proto.participant_pb2 import DisconnectReason + + +async def _wait_until_dropped( + dropped_handles: list[int], expected_handle: int, timeout: float = 2.0 +) -> None: + loop = asyncio.get_running_loop() + deadline = loop.time() + timeout + while loop.time() < deadline: + if dropped_handles == [expected_handle]: + return + await asyncio.sleep(0.01) + assert dropped_handles == [expected_handle] + + +@pytest.mark.asyncio +async def test_room_connect_cancelled_during_handshake_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + dropped_handles: list[int] = [] + ready_requests: list[int] = [] + disconnect_requests: list[tuple[int, int]] = [] + + connect_async_id = 101 + disconnect_async_id = 202 + fake_room_handle_id = 9999 + + def fake_dispose(self: FfiHandle) -> None: + if self.handle != 0 and not self._disposed: + self._disposed = True + dropped_handles.append(self.handle) + + monkeypatch.setattr(FfiHandle, "dispose", fake_dispose) + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append((req.disconnect.room_handle, req.disconnect.reason)) + resp = proto_ffi.FfiResponse() + resp.disconnect.async_id = disconnect_async_id + + # Simulate the disconnect callback from native FFI + disc_cb = proto_ffi.FfiEvent() + disc_cb.disconnect.async_id = disconnect_async_id + FfiClient.instance.queue.put(disc_cb) + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + connect_task = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "fake-token")) + + # Yield to let connect() dispatch ConnectRequest and park waiting on queue + await asyncio.sleep(0) + assert not connect_task.done() + + # Cancel the connect task + connect_task.cancel() + + # Yield to deliver cancellation + await asyncio.sleep(0) + + # Now deliver the ConnectCallback from native FFI + connect_cb = proto_ffi.FfiEvent() + connect_cb.connect.async_id = connect_async_id + connect_cb.connect.result.room.handle.id = fake_room_handle_id + FfiClient.instance.queue.put(connect_cb) + + # Wait for the task to finish cancellation + with pytest.raises(asyncio.CancelledError): + await connect_task + + # Allow background cleanup task to process the connect callback + await _wait_until_dropped(dropped_handles, fake_room_handle_id) + + # Verify: + # 1. ready_for_room_event was dispatched to satisfy 15s timer (Fixes #784) + assert ready_requests == [fake_room_handle_id] + + # 2. disconnect was dispatched to prevent ghost session / duplicate identity (Fixes #804) + assert len(disconnect_requests) == 1 + assert disconnect_requests[0][0] == fake_room_handle_id + assert disconnect_requests[0][1] == DisconnectReason.CLIENT_INITIATED + + # 3. Handle was properly dropped in native FFI + assert dropped_handles == [fake_room_handle_id] + + # 4. Room state remains disconnected + assert not room.isconnected() + + +@pytest.mark.asyncio +async def test_room_connect_cancelled_when_connect_fails( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + ready_requests: list[int] = [] + disconnect_requests: list[int] = [] + connect_async_id = 303 + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append(req.disconnect.room_handle) + return proto_ffi.FfiResponse() + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + connect_task = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "fake-token")) + + await asyncio.sleep(0) + connect_task.cancel() + await asyncio.sleep(0) + + # Deliver a connection error callback + connect_cb = proto_ffi.FfiEvent() + connect_cb.connect.async_id = connect_async_id + connect_cb.connect.error = "Connection refused" + FfiClient.instance.queue.put(connect_cb) + + with pytest.raises(asyncio.CancelledError): + await connect_task + + # No ready or disconnect request should be sent if connect failed + assert ready_requests == [] + assert disconnect_requests == [] + assert not room.isconnected() + + +@pytest.mark.asyncio +async def test_room_connect_wait_for_timeout_cancellation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + dropped_handles: list[int] = [] + ready_requests: list[int] = [] + disconnect_requests: list[int] = [] + + connect_async_id = 404 + disconnect_async_id = 505 + fake_room_handle_id = 8888 + + def fake_dispose(self: FfiHandle) -> None: + if self.handle != 0 and not self._disposed: + self._disposed = True + dropped_handles.append(self.handle) + + monkeypatch.setattr(FfiHandle, "dispose", fake_dispose) + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append(req.disconnect.room_handle) + resp = proto_ffi.FfiResponse() + resp.disconnect.async_id = disconnect_async_id + + disc_cb = proto_ffi.FfiEvent() + disc_cb.disconnect.async_id = disconnect_async_id + FfiClient.instance.queue.put(disc_cb) + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + async def _delayed_connect_cb() -> None: + await asyncio.sleep(0.05) + connect_cb = proto_ffi.FfiEvent() + connect_cb.connect.async_id = connect_async_id + connect_cb.connect.result.room.handle.id = fake_room_handle_id + FfiClient.instance.queue.put(connect_cb) + + asyncio.create_task(_delayed_connect_cb()) + + # Wait with a short timeout that triggers cancellation + with pytest.raises(asyncio.TimeoutError): + await asyncio.wait_for(room.connect("ws://127.0.0.1:7880", "fake-token"), timeout=0.01) + + # Allow delayed callback and background cleanup task to run + await _wait_until_dropped(dropped_handles, fake_room_handle_id) + + # Handshake cleanup was satisfied + assert ready_requests == [fake_room_handle_id] + assert disconnect_requests == [fake_room_handle_id] + assert dropped_handles == [fake_room_handle_id] + assert not room.isconnected() + + +@pytest.mark.asyncio +async def test_room_connect_setup_failure_cleans_up( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + dropped_handles: list[int] = [] + ready_requests: list[int] = [] + disconnect_requests: list[int] = [] + + connect_async_id = 606 + disconnect_async_id = 707 + fake_room_handle_id = 7777 + + def fake_dispose(self: FfiHandle) -> None: + if self.handle != 0 and not self._disposed: + self._disposed = True + dropped_handles.append(self.handle) + + monkeypatch.setattr(FfiHandle, "dispose", fake_dispose) + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append(req.disconnect.room_handle) + resp = proto_ffi.FfiResponse() + resp.disconnect.async_id = disconnect_async_id + + disc_cb = proto_ffi.FfiEvent() + disc_cb.disconnect.async_id = disconnect_async_id + FfiClient.instance.queue.put(disc_cb) + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + # Force an exception during room setup before ready_for_room_event + def faulty_create_remote_participant(*args: Any, **kwargs: Any) -> Any: + raise RuntimeError("simulated setup explosion") + + monkeypatch.setattr(room, "_create_remote_participant", faulty_create_remote_participant) + + async def _deliver_cb() -> None: + await asyncio.sleep(0.01) + connect_cb = proto_ffi.FfiEvent() + connect_cb.connect.async_id = connect_async_id + connect_cb.connect.result.room.handle.id = fake_room_handle_id + # Add a participant to trigger faulty_create_remote_participant + p = connect_cb.connect.result.participants.add() + p.participant.info.identity = "remote_1" + FfiClient.instance.queue.put(connect_cb) + + asyncio.create_task(_deliver_cb()) + + with pytest.raises(RuntimeError, match="simulated setup explosion"): + await room.connect("ws://127.0.0.1:7880", "fake-token") + + # Handshake cleanup was satisfied even on exception during setup + assert ready_requests == [fake_room_handle_id] + assert disconnect_requests == [fake_room_handle_id] + assert dropped_handles == [fake_room_handle_id] + assert not room.isconnected() + # Ensure failed setup does not leave stale participants or unreset state + assert len(room.remote_participants) == 0 + assert room._local_participant is None + + +@pytest.mark.asyncio +async def test_room_connect_repeated_cancellation_during_cleanup( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + dropped_handles: list[int] = [] + ready_requests: list[int] = [] + disconnect_requests: list[int] = [] + + connect_async_id = 808 + disconnect_async_id = 909 + fake_room_handle_id = 5555 + + def fake_dispose(self: FfiHandle) -> None: + if self.handle != 0 and not self._disposed: + self._disposed = True + dropped_handles.append(self.handle) + + monkeypatch.setattr(FfiHandle, "dispose", fake_dispose) + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append(req.disconnect.room_handle) + resp = proto_ffi.FfiResponse() + resp.disconnect.async_id = disconnect_async_id + + disc_cb = proto_ffi.FfiEvent() + disc_cb.disconnect.async_id = disconnect_async_id + FfiClient.instance.queue.put(disc_cb) + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + connect_task = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "fake-token")) + + await asyncio.sleep(0) + # Repeated cancellation before callback arrives + connect_task.cancel() + connect_task.cancel() + + with pytest.raises(asyncio.CancelledError): + await connect_task + + # Deliver connect callback; background cleanup task handles it + connect_cb = proto_ffi.FfiEvent() + connect_cb.connect.async_id = connect_async_id + connect_cb.connect.result.room.handle.id = fake_room_handle_id + FfiClient.instance.queue.put(connect_cb) + + await _wait_until_dropped(dropped_handles, fake_room_handle_id) + + assert ready_requests == [fake_room_handle_id] + assert disconnect_requests == [fake_room_handle_id] + assert dropped_handles == [fake_room_handle_id] + assert not room.isconnected() + + +@pytest.mark.asyncio +async def test_room_connect_subsequent_connect_waits_for_cleanup( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + dropped_handles: list[int] = [] + ready_requests: list[int] = [] + disconnect_requests: list[int] = [] + connect_requests: list[str] = [] + + fake_room_handle_id = 42 + connect_async_id_1 = 101 + connect_async_id_2 = 102 + disconnect_async_id = 202 + + def fake_dispose(self: FfiHandle) -> None: + if not self._disposed: + self._disposed = True + dropped_handles.append(self.handle) + + monkeypatch.setattr(FfiHandle, "dispose", fake_dispose) + + call_count = 0 + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + nonlocal call_count + which = req.WhichOneof("message") + if which == "connect": + call_count += 1 + connect_requests.append(req.connect.token) + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id_1 if call_count == 1 else connect_async_id_2 + return resp + elif which == "ready_for_room_event": + ready_requests.append(req.ready_for_room_event.room_handle) + return proto_ffi.FfiResponse() + elif which == "disconnect": + disconnect_requests.append(req.disconnect.room_handle) + resp = proto_ffi.FfiResponse() + resp.disconnect.async_id = disconnect_async_id + disc_cb = proto_ffi.FfiEvent() + disc_cb.disconnect.async_id = disconnect_async_id + FfiClient.instance.queue.put(disc_cb) + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + # First attempt cancelled while connect is in flight + connect_task1 = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "token-1")) + await asyncio.sleep(0) + connect_task1.cancel() + + with pytest.raises(asyncio.CancelledError): + await connect_task1 + + assert room.cleanup_task is not None + assert not room.cleanup_task.done() + + # Second connect attempt started immediately + connect_task2 = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "token-2")) + await asyncio.sleep(0) + + # The second attempt must NOT have dispatched req.connect yet because cleanup is pending + assert len(connect_requests) == 1 + + # Now deliver connect callback for attempt 1 + connect_cb1 = proto_ffi.FfiEvent() + connect_cb1.connect.async_id = connect_async_id_1 + connect_cb1.connect.result.room.handle.id = fake_room_handle_id + FfiClient.instance.queue.put(connect_cb1) + + # Cleanup finishes and second connect proceeds to send request + await _wait_until_dropped(dropped_handles, fake_room_handle_id) + loop = asyncio.get_running_loop() + deadline = loop.time() + 2.0 + while loop.time() < deadline: + if len(connect_requests) == 2: + break + await asyncio.sleep(0.01) + assert len(connect_requests) == 2 + assert connect_requests == ["token-1", "token-2"] + + # Cancel task 2 to finish cleanly + connect_task2.cancel() + with pytest.raises(asyncio.CancelledError): + await connect_task2 + + +@pytest.mark.asyncio +async def test_room_connect_cancellation_fails_sid_waiters( + monkeypatch: pytest.MonkeyPatch, +) -> None: + room = Room() + connect_async_id = 999 + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + which = req.WhichOneof("message") + if which == "connect": + resp = proto_ffi.FfiResponse() + resp.connect.async_id = connect_async_id + return resp + raise AssertionError(f"unexpected request: {which}") + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + + connect_task = asyncio.create_task(room.connect("ws://127.0.0.1:7880", "fake-token")) + await asyncio.sleep(0) + + sid_task = asyncio.create_task(room.sid) + await asyncio.sleep(0) + + connect_task.cancel() + with pytest.raises(asyncio.CancelledError): + await connect_task + + with pytest.raises(asyncio.CancelledError): + await sid_task + + # Future SID waiter on the same room gets a fresh future + assert not room._first_sid_future.done()