From 6e120fbf723e54491bfb2e007117996d8aa4150e Mon Sep 17 00:00:00 2001 From: Subomi-olagoke Date: Mon, 14 Sep 2026 10:54:12 +0100 Subject: [PATCH 1/2] rtc: close the room when Room.connect is cancelled The FFI server has no cancel path for an in-flight connect: it answers the connect request and then waits for ReadyForRoomEventRequest, which connect() sends as its last statement. A coroutine cancelled anywhere inside connect() never reaches it, the server times out after 15s and panics, and the panic handler sends SIGTERM to the process. Hand the room to a task that survives the cancellation, answer the pending ready request and disconnect. disconnect() waits for that task so callers can close deterministically, and the room no longer stays joined server-side, which is what evicts a retry using the same identity. Fixes #784 Refs #804 --- livekit-rtc/livekit/rtc/room.py | 72 ++++++++++- .../tests/test_connect_cancellation.py | 118 ++++++++++++++++++ 2 files changed, 185 insertions(+), 5 deletions(-) create mode 100644 livekit-rtc/tests/test_connect_cancellation.py diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 8f89e8a5..0f514ef1 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, task_done_logger from .e2ee import E2EEManager, E2EEOptions from .log import logger from .participant import ( @@ -184,6 +184,7 @@ def __init__( self._room_queue = BroadcastQueue[proto_ffi.FfiEvent]() self._info = proto_room.RoomInfo() self._rpc_invocation_tasks: set[asyncio.Task] = set() + self._aborted_connect_tasks: set[asyncio.Task] = set() self._remote_participants: Dict[str, RemoteParticipant] = {} self._connection_state = ConnectionState.CONN_DISCONNECTED @@ -554,13 +555,25 @@ def on_participant_connected(participant): self._ffi_queue = FfiClient.instance.queue.subscribe(self._loop) queue = FfiClient.instance.queue.subscribe() + aborted = 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 FFI server is already connecting and expects a ReadyForRoomEvent + # once it answers. leaving that unanswered panics it, and the panic + # handler terminates the process, so close the room from a task that + # outlives this cancellation. + aborted = True + FfiClient.instance.queue.unsubscribe(self._ffi_queue) + self._close_aborted_connect(resp.connect.async_id, queue) + raise finally: - FfiClient.instance.queue.unsubscribe(queue) + if not aborted: + FfiClient.instance.queue.unsubscribe(queue) if cb.connect.error: FfiClient.instance.queue.unsubscribe(self._ffi_queue) @@ -602,6 +615,50 @@ def on_participant_connected(participant): ready_req.ready_for_room_event.room_handle = self._ffi_handle.handle FfiClient.instance.request(ready_req) + def _close_aborted_connect(self, async_id: int, queue: Queue[proto_ffi.FfiEvent]) -> None: + """Close a room that connect() was cancelled before it could own. + + Takes ownership of `queue`. The FFI server has no cancel path for an in-flight + connect, so the room has to be created and then disconnected. Without this the + room also stays joined server-side and reconnecting with the same identity + evicts the new session as a duplicate. + """ + + async def _close() -> None: + try: + cb: proto_ffi.FfiEvent = await queue.wait_for( + lambda e: e.connect.async_id == async_id + ) + finally: + FfiClient.instance.queue.unsubscribe(queue) + + if cb.connect.error: + return + + ffi_handle = FfiHandle(cb.connect.result.room.handle.id) + + ready_req = proto_ffi.FfiRequest() + ready_req.ready_for_room_event.room_handle = ffi_handle.handle + FfiClient.instance.request(ready_req) + + close_req = proto_ffi.FfiRequest() + close_req.disconnect.room_handle = ffi_handle.handle + close_req.disconnect.reason = DisconnectReason.CLIENT_INITIATED + close_queue = FfiClient.instance.queue.subscribe() + try: + resp = FfiClient.instance.request(close_req) + await close_queue.wait_for( + lambda e: e.disconnect.async_id == resp.disconnect.async_id + ) + finally: + FfiClient.instance.queue.unsubscribe(close_queue) + + task = self._loop.create_task(_close()) + self._aborted_connect_tasks.add(task) + task.add_done_callback(self._aborted_connect_tasks.discard) + # a failure here still ends in an FFI panic, so it must not be swallowed + task.add_done_callback(task_done_logger) + async def get_rtc_stats(self) -> RtcStats: if not self.isconnected(): raise RuntimeError("the room isn't connected") @@ -681,6 +738,11 @@ async def disconnect( self, *, reason: DisconnectReason.ValueType = DisconnectReason.CLIENT_INITIATED ) -> None: """Disconnects from the room.""" + if self._aborted_connect_tasks: + # a cancelled connect may still be closing a room the FFI server opened. + # wait for it so disconnect() leaves nothing behind. + await asyncio.gather(*tuple(self._aborted_connect_tasks), return_exceptions=True) + if not self.isconnected(): return diff --git a/livekit-rtc/tests/test_connect_cancellation.py b/livekit-rtc/tests/test_connect_cancellation.py new file mode 100644 index 00000000..cb222eec --- /dev/null +++ b/livekit-rtc/tests/test_connect_cancellation.py @@ -0,0 +1,118 @@ +# 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 room as room_mod +from livekit.rtc._ffi_client import FfiClient +from livekit.rtc._proto import ffi_pb2 as proto_ffi +from utils import wait_until # type: ignore[import-not-found] + +CONNECT_ASYNC_ID = 101 +DISCONNECT_ASYNC_ID = 202 +ROOM_HANDLE = 7 + + +class _FakeHandle: + """Stand-in for FfiHandle so a made-up handle id is never dropped natively.""" + + def __init__(self, handle: int) -> None: + self.handle = handle + + +def _install_fake_ffi(monkeypatch: pytest.MonkeyPatch) -> list[proto_ffi.FfiRequest]: + """Record every FfiRequest and answer the ones the cancel path waits on.""" + requests: list[proto_ffi.FfiRequest] = [] + + def fake_request(req: proto_ffi.FfiRequest) -> proto_ffi.FfiResponse: + requests.append(req) + resp = proto_ffi.FfiResponse() + which = req.WhichOneof("message") + if which == "connect": + # the connect callback is delivered by the test, not here + resp.connect.async_id = CONNECT_ASYNC_ID + elif which == "disconnect": + resp.disconnect.async_id = DISCONNECT_ASYNC_ID + event = proto_ffi.FfiEvent() + event.disconnect.async_id = DISCONNECT_ASYNC_ID + FfiClient.instance.queue.put(event) + return resp + + monkeypatch.setattr(FfiClient.instance, "request", fake_request) + monkeypatch.setattr(room_mod, "FfiHandle", _FakeHandle) + return requests + + +def _deliver_connect_callback() -> None: + event = proto_ffi.FfiEvent() + event.connect.async_id = CONNECT_ASYNC_ID + event.connect.result.room.handle.id = ROOM_HANDLE + FfiClient.instance.queue.put(event) + + +async def test_cancelled_connect_answers_ready_and_closes_the_room( + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests = _install_fake_ffi(monkeypatch) + subscribers_before = len(FfiClient.instance.queue._subscribers) + + room = rtc.Room() + task = asyncio.create_task(room.connect("ws://localhost:7880", "token")) + await wait_until(lambda: bool(requests), message="connect request never issued") + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # the FFI server does not cancel an in-flight connect: it answers, then waits for + # ReadyForRoomEvent. an unanswered wait panics it and the panic kills the process. + _deliver_connect_callback() + await room.disconnect() + + assert [req.WhichOneof("message") for req in requests] == [ + "connect", + "ready_for_room_event", + "disconnect", + ] + assert requests[1].ready_for_room_event.room_handle == ROOM_HANDLE + assert requests[2].disconnect.room_handle == ROOM_HANDLE + assert len(FfiClient.instance.queue._subscribers) == subscribers_before + + +async def test_cancelled_connect_leaves_no_pending_work_when_the_server_errors( + monkeypatch: pytest.MonkeyPatch, +) -> None: + requests = _install_fake_ffi(monkeypatch) + subscribers_before = len(FfiClient.instance.queue._subscribers) + + room = rtc.Room() + task = asyncio.create_task(room.connect("ws://localhost:7880", "token")) + await wait_until(lambda: bool(requests), message="connect request never issued") + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + event = proto_ffi.FfiEvent() + event.connect.async_id = CONNECT_ASYNC_ID + event.connect.error = "could not connect" + FfiClient.instance.queue.put(event) + await room.disconnect() + + # there is no room to close, so nothing follows the connect + assert [req.WhichOneof("message") for req in requests] == ["connect"] + assert len(FfiClient.instance.queue._subscribers) == subscribers_before From 24d566422cc7559bad394b5b46020e05cf28006e Mon Sep 17 00:00:00 2001 From: Subomi-olagoke Date: Sat, 19 Sep 2026 15:10:46 +0100 Subject: [PATCH 2/2] rtc: shield the aborted-connect cleanup from a cancelled disconnect gather() cancels its children when it is cancelled, so a caller who bounds disconnect() with a timeout, or abandons it during shutdown, cancelled the cleanup task that answers the FFI's wait for ReadyForRoomEvent. The wait then timed out and panicked, and the panic handler kills the process, which is the failure this path was added to prevent. The test fails without the shield. --- livekit-rtc/livekit/rtc/room.py | 9 ++++- .../tests/test_connect_cancellation.py | 36 +++++++++++++++++++ 2 files changed, 44 insertions(+), 1 deletion(-) diff --git a/livekit-rtc/livekit/rtc/room.py b/livekit-rtc/livekit/rtc/room.py index 0f514ef1..7f00c50e 100644 --- a/livekit-rtc/livekit/rtc/room.py +++ b/livekit-rtc/livekit/rtc/room.py @@ -741,7 +741,14 @@ async def disconnect( if self._aborted_connect_tasks: # a cancelled connect may still be closing a room the FFI server opened. # wait for it so disconnect() leaves nothing behind. - await asyncio.gather(*tuple(self._aborted_connect_tasks), return_exceptions=True) + # + # shielded, because gather() cancels its children when it is cancelled. + # a caller who gives up on disconnect() would otherwise cancel the very + # cleanup that answers the FFI's wait, leaving it to time out and panic, + # which is the failure this path exists to prevent. + await asyncio.shield( + asyncio.gather(*tuple(self._aborted_connect_tasks), return_exceptions=True) + ) if not self.isconnected(): return diff --git a/livekit-rtc/tests/test_connect_cancellation.py b/livekit-rtc/tests/test_connect_cancellation.py index cb222eec..6bcb7808 100644 --- a/livekit-rtc/tests/test_connect_cancellation.py +++ b/livekit-rtc/tests/test_connect_cancellation.py @@ -116,3 +116,39 @@ async def test_cancelled_connect_leaves_no_pending_work_when_the_server_errors( # there is no room to close, so nothing follows the connect assert [req.WhichOneof("message") for req in requests] == ["connect"] assert len(FfiClient.instance.queue._subscribers) == subscribers_before + + +async def test_a_cancelled_disconnect_still_lets_the_cleanup_finish( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Giving up on disconnect() must not cancel the cleanup it is waiting on. + + gather() cancels its children when it is cancelled, so a caller who bounds + disconnect() with a timeout, or abandons it on shutdown, would cancel the + task that answers the FFI's wait for ReadyForRoomEvent. The wait then times + out, the FFI panics, and the panic handler kills the process: the exact + failure the rest of this file is about, reintroduced one layer up. + """ + requests = _install_fake_ffi(monkeypatch) + + room = rtc.Room() + task = asyncio.create_task(room.connect("ws://localhost:7880", "token")) + await wait_until(lambda: bool(requests), message="connect request never issued") + + task.cancel() + with pytest.raises(asyncio.CancelledError): + await task + + # the cleanup is now parked on the connect callback, which has not arrived + closing = asyncio.create_task(room.disconnect()) + await asyncio.sleep(0) + closing.cancel() + with pytest.raises(asyncio.CancelledError): + await closing + + _deliver_connect_callback() + await wait_until( + lambda: any(r.WhichOneof("message") == "ready_for_room_event" for r in requests), + message="the cancelled disconnect took the cleanup down with it", + ) + assert requests[1].ready_for_room_event.room_handle == ROOM_HANDLE