rtc: close the room when Room.connect is cancelled

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,18 @@ 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.
+            #
+            # 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
 

livekit-rtc/tests/test_connect_cancellation.py

@@ -0,0 +1,154 @@
+# 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
+
+
+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