From 95764c119f8015661d15f8e59c976b8bea57d1ac Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 22:56:39 +0500 Subject: [PATCH 1/8] fix(rtc): make Room.connect cancellation-safe to avoid FFI timeout panic Cancelling Room.connect while ConnectRequest is in flight previously stranded the FFI room-event handshake because ReadyForRoomEventRequest was never sent, resulting in an FFI panic 15 seconds later that killed the process with SIGTERM (issue #784). Additionally, the native connection remained active in the background, causing duplicate identity disconnects on reconnect (issue #804). - Await ConnectCallback uninterruptibly on CancelledError - Complete handshake with ReadyForRoomEventRequest and cleanly disconnect the native session - Add unit tests for connect cancellation, connect failure, timeout, and cleanup under setup exceptions --- livekit-rtc/livekit/rtc/room.py | 183 +++++++-- .../tests/test_ffi_handle_lifecycle.py | 30 +- .../tests/test_room_connect_cancellation.py | 370 ++++++++++++++++++ 3 files changed, 535 insertions(+), 48 deletions(-) create mode 100644 livekit-rtc/tests/test_room_connect_cancellation.py diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0601f9a4..69a0e970 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,7 @@ def __init__( self._token: str | None = None self._server_url: str | None = None + self._task: Optional[asyncio.Task] = None def __del__(self) -> None: if self._ffi_handle is not None: @@ -561,53 +562,164 @@ 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: + # 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 + self._loop.create_task( + self._cleanup_cancelled_connect(queue, resp.connect.async_id, self._ffi_queue) + ) + 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) + 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()) + + # 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: + if not ready_sent: + if self._task is not None and not self._task.done(): + self._task.cancel() + await self._cleanup_orphaned_room(self._ffi_handle) + self._ffi_handle = None + self._connection_state = ConnectionState.CONN_DISCONNECTED + self._remote_participants.clear() + self._local_participant = None + self._info = proto_room.RoomInfo() + if self._first_sid_future.done(): + self._first_sid_future = asyncio.Future[str]() + 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 asyncio.wait_for( + queue.wait_for(lambda e: e.connect.async_id == async_id), + timeout=30.0, + ) + except (asyncio.TimeoutError, asyncio.CancelledError): + return - # 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)) + 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(): @@ -705,7 +817,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_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..651f321c --- /dev/null +++ b/livekit-rtc/tests/test_room_connect_cancellation.py @@ -0,0 +1,370 @@ +# 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() From c55bcf380ed686ad38ae2bcabb34d26d6bdba073 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 23:40:51 +0500 Subject: [PATCH 2/8] fix(rtc): synchronize reconnect and disconnect with pending connect cleanup --- livekit-rtc/livekit/rtc/room.py | 19 ++++- .../tests/test_room_connect_cancellation.py | 83 +++++++++++++++++++ 2 files changed, 101 insertions(+), 1 deletion(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 69a0e970..52a493ec 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -199,6 +199,7 @@ 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 def __del__(self) -> None: if self._ffi_handle is not None: @@ -468,6 +469,12 @@ def on_participant_connected(participant): await room.connect("ws://localhost:7880", "your_token") ``` """ + if self._cleanup_task is not None and not self._cleanup_task.done(): + try: + await asyncio.shield(self._cleanup_task) + except Exception: + pass + self._server_url = url self._token = token req = proto_ffi.FfiRequest() @@ -575,7 +582,7 @@ def on_participant_connected(participant): # while guaranteeing that when ConnectCallback arrives the FFI handshake is # completed and cleanly disconnected (Issues #784, #804). queue_unsubscribed = True - self._loop.create_task( + self._cleanup_task = self._loop.create_task( self._cleanup_cancelled_connect(queue, resp.connect.async_id, self._ffi_queue) ) raise @@ -796,11 +803,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_task is not None and not self._cleanup_task.done(): + try: + await asyncio.shield(self._cleanup_task) + except Exception: + pass return await self._drain_rpc_invocation_tasks() diff --git a/livekit-rtc/tests/test_room_connect_cancellation.py b/livekit-rtc/tests/test_room_connect_cancellation.py index 651f321c..9d3969a1 100644 --- a/livekit-rtc/tests/test_room_connect_cancellation.py +++ b/livekit-rtc/tests/test_room_connect_cancellation.py @@ -368,3 +368,86 @@ def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: 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) + 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 From 0d32ef161a030e0c74284004147a0e2d739046d8 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 23:51:25 +0500 Subject: [PATCH 3/8] test(rtc): eliminate timing race when asserting second connect dispatch in retry test --- livekit-rtc/tests/test_room_connect_cancellation.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/livekit-rtc/tests/test_room_connect_cancellation.py b/livekit-rtc/tests/test_room_connect_cancellation.py index 9d3969a1..0704b059 100644 --- a/livekit-rtc/tests/test_room_connect_cancellation.py +++ b/livekit-rtc/tests/test_room_connect_cancellation.py @@ -444,6 +444,12 @@ def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: # 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"] From f110eef0ab44e4a8ff16653f25cb089a139be377 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:13:18 +0500 Subject: [PATCH 4/8] fix(rtc): retain cleanup tasks in set and fail pending sid waiters on cancel or error --- livekit-rtc/livekit/rtc/room.py | 29 +++++++++++----- .../tests/test_room_connect_cancellation.py | 34 +++++++++++++++++++ 2 files changed, 54 insertions(+), 9 deletions(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 52a493ec..d3d23574 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -200,6 +200,7 @@ def __init__( 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: @@ -226,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. @@ -469,9 +477,9 @@ def on_participant_connected(participant): await room.connect("ws://localhost:7880", "your_token") ``` """ - if self._cleanup_task is not None and not self._cleanup_task.done(): + if self._cleanup_tasks: try: - await asyncio.shield(self._cleanup_task) + await asyncio.shield(asyncio.gather(*self._cleanup_tasks)) except Exception: pass @@ -576,15 +584,19 @@ def on_participant_connected(participant): cb: proto_ffi.FfiEvent = await queue.wait_for( lambda e: e.connect.async_id == resp.connect.async_id ) - except asyncio.CancelledError: + 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 - self._cleanup_task = self._loop.create_task( + 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: if not queue_unsubscribed: @@ -631,7 +643,7 @@ def on_participant_connected(participant): ready_req.ready_for_room_event.room_handle = self._ffi_handle.handle FfiClient.instance.request(ready_req) ready_sent = True - except BaseException: + except BaseException as e: if not ready_sent: if self._task is not None and not self._task.done(): self._task.cancel() @@ -641,8 +653,7 @@ def on_participant_connected(participant): self._remote_participants.clear() self._local_participant = None self._info = proto_room.RoomInfo() - if self._first_sid_future.done(): - self._first_sid_future = asyncio.Future[str]() + self._fail_sid_waiters(e) raise @staticmethod @@ -813,9 +824,9 @@ async def disconnect( ) -> None: """Disconnects from the room.""" if not self.isconnected(): - if self._cleanup_task is not None and not self._cleanup_task.done(): + if self._cleanup_tasks: try: - await asyncio.shield(self._cleanup_task) + await asyncio.shield(asyncio.gather(*self._cleanup_tasks)) except Exception: pass return diff --git a/livekit-rtc/tests/test_room_connect_cancellation.py b/livekit-rtc/tests/test_room_connect_cancellation.py index 0704b059..bfd8f584 100644 --- a/livekit-rtc/tests/test_room_connect_cancellation.py +++ b/livekit-rtc/tests/test_room_connect_cancellation.py @@ -457,3 +457,37 @@ def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: 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() From e243b610e4591282a8a2e6dad7bc5027accfc96a Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:39:10 +0500 Subject: [PATCH 5/8] fix(rtc): shield orphaned room cleanup with state reset in finally and remove cleanup timeout --- livekit-rtc/livekit/rtc/room.py | 23 +++++++++++------------ livekit-rtc/tests/test_ffi_client.py | 2 +- 2 files changed, 12 insertions(+), 13 deletions(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index d3d23574..63ea916c 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -647,13 +647,15 @@ def on_participant_connected(participant): if not ready_sent: if self._task is not None and not self._task.done(): self._task.cancel() - await self._cleanup_orphaned_room(self._ffi_handle) - 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) + 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 @@ -664,11 +666,8 @@ async def _cleanup_cancelled_connect( ) -> None: try: try: - cb = await asyncio.wait_for( - queue.wait_for(lambda e: e.connect.async_id == async_id), - timeout=30.0, - ) - except (asyncio.TimeoutError, asyncio.CancelledError): + 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: 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 33b7d783d77cb730e135c7383accc6cd555de32f Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:47:47 +0500 Subject: [PATCH 6/8] 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 f3fafa1b34761e0e6dd9b8ac235b1d5f370ed99b Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 00:54:47 +0500 Subject: [PATCH 7/8] 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 5d5bd9e2a8af5bb8b030859302f75d51157e3e1b Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Wed, 21 Oct 2026 01:07:41 +0500 Subject: [PATCH 8/8] 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) """