[NabiBukhsh-AI]

Summary

Cancelling Room.connect() while ConnectRequest is in flight previously stranded the FFI room-event handshake because ReadyForRoomEventRequest was never dispatched. When the 15-second ROOM_EVENT_READY_TIMEOUT expired in native Rust FFI, it raised an FFI panic that caused Python's FfiClient to kill the process with SIGTERM (#784). Furthermore, the native connection remained open on the LiveKit server in the background, causing subsequent reconnect attempts using the same token to disconnect due to duplicated identity (#804).

This PR makes Room.connect() cancellation-safe and ensures that cancellation or initialization exceptions reliably complete the FFI handshake, cleanly disconnect the session, and dispose of the native room handle. Additionally, reconnects and disconnects coordinate with any in-flight background cleanup so immediate retry attempts never collide with an unreleased native session.

---

Root Cause Analysis

1. In Room.connect(), cb = await queue.wait_for(...) is the only await point during the connection handshake.
2. When a timeout or task cancellation occurs while waiting for ConnectCallback, asyncio.CancelledError aborts connect().
3. In native Rust FFI, the connection finishes and produces a ConnectCallback, then waits up to 15 seconds for ReadyForRoomEventRequest from the Python SDK.
4. Because Python had aborted, ReadyForRoomEventRequest was never sent. When the 15-second timer expired, Rust FFI reported a panic, causing _ffi_client.py to terminate the process with os.kill(os.getpid(), signal.SIGTERM).
5. Simultaneously, the server still saw an active participant, preventing immediate reconnection due to identity collision.

---

Key Changes

1. Non-blocking Cancellation with Supervised Background Cleanup:
   • In Room.connect(), catch asyncio.CancelledError on queue.wait_for().
   • Transfer in-flight request cleanup to a background task self._loop.create_task(self._cleanup_cancelled_connect(...)) and re-raise CancelledError immediately so caller timeouts and cancellations are unblocked promptly.
2. Clean Disconnection and Handshake Completion:
   • If the connection attempt succeeded:
     • Immediately dispatch ReadyForRoomEventRequest with the allocated handle, satisfying the FFI 15-second timer and preventing the SIGTERM panic.
     • Send DisconnectRequest to cleanly tear down the participant session on the server, avoiding duplicate identity errors on reconnect.
     • Await the disconnect callback (bounded by a 10s timeout), dispose of the FfiHandle, and unsubscribe self._ffi_queue.
   • If the connection failed with an error, unsubscribe self._ffi_queue and finish cleanly.
3. Reconnect and Teardown Synchronization Barrier:
   • Track self._cleanup_task on Room.
   • At the beginning of connect() and in disconnect(), check if self._cleanup_task is not None and not self._cleanup_task.done(): await asyncio.shield(self._cleanup_task).
   • This prevents immediate reconnect attempts on the same Room instance from colliding with a session whose native teardown is still in flight.
4. Failure Guard During Room Setup:
   • If an exception occurs during room setup after ConnectCallback has arrived but before ReadyForRoomEventRequest is sent, run the cleanup sequence to prevent stranding the FFI room, and clear room state so subsequent retry attempts start fresh.
5. Disconnect Task Safety:
   • Guard await self._task in disconnect() with if self._task is not None:.

---

Tests Added

Added unit test suite in livekit-rtc/tests/test_room_connect_cancellation.py:

• test_room_connect_cancelled_during_handshake_cleans_up: Verifies that cancelling an in-flight connect() sends ready_for_room_event, triggers disconnect, disposes of the handle, and raises CancelledError.
• test_room_connect_cancelled_when_connect_fails: Verifies that cancelling when the connection fails cleanly unsubscribes the queue without dispatching disconnect requests.
• test_room_connect_wait_for_timeout_cancellation: Verifies asyncio.wait_for(room.connect(...), timeout=...) timeout cancellation cleans up properly.
• test_room_connect_setup_failure_cleans_up: Verifies that an exception during room setup before ready_for_room_event completes the handshake and closes the room session.
• test_room_connect_repeated_cancellation_during_cleanup: Verifies that repeated cancellations while waiting for cleanup callbacks do not interrupt the teardown sequence.
• test_room_connect_subsequent_connect_waits_for_cleanup: Verifies that retrying connect() on a room with an in-flight background cleanup waits for the previous teardown to complete before issuing the new connection request.

Fixes #784
Fixes #804

[xianshijing-lk]

should we hold the reference of cleanup_task ? like

self._cleanup_tasks: set[asyncio.Task] = set() # in init
...

t = self._loop.create_task(self._cleanup_cancelled_connect(...))
self._cleanup_tasks.add(t)
t.add_done_callback(self._cleanup_tasks.discard)

otherwise GC can just drop the task and the original bug could come back non-deterministically

[NabiBukhsh-AI]

Good call! Added self._cleanup_tasks: set[asyncio.Task[None]] = set() in f110eef, tracking all active cleanup tasks and awaiting �syncio.shield(asyncio.gather(*self._cleanup_tasks)) in connect() and disconnect() to prevent garbage collection and eliminate race conditions.

[xianshijing-lk]

wonder if it is better to do

def _fail_sid_waiters(self, err: BaseException) -> None:
    fut = self._first_sid_future
    if not fut.done():            # required: avoids InvalidStateError on the common path
        fut.set_exception(err)
        fut.exception()           # required: suppresses never-retrieved noise
    self._first_sid_future = asyncio.Future[str]()

and call this in both except asyncio.CancelledError (Path A) and except BaseException

[NabiBukhsh-AI]

Implemented in f110eef: added _fail_sid_waiters exactly as suggested and invoked in both CancelledError and BaseException teardowns, resetting _first_sid_future cleanly with regression verification in est_room_connect_cancellation_fails_sid_waiters.

[NabiBukhsh-AI]

Hi @xianshijing-lk,

All requested updates and synchronization improvements are in place:

1. Tracking background cleanups in self._cleanup_tasks: set[asyncio.Task[None]] and awaiting asyncio.shield(asyncio.gather(*self._cleanup_tasks)) across connect() and disconnect() to prevent garbage collection and eliminate reconnect race conditions.
2. Handled connect cancellation safety so pending SID waiters fail immediately on error/cancellation rather than hanging.
3. All 36 CI matrix jobs (Python 3.9–3.14 on Linux, macOS, and Windows) and Devin Review checks are passing 100% green. Ready for final review.

[github-actions]

👋 This PR has no changeset. If it should be released, add a markdown file under .changeset/ (any name, e.g. .changeset/fix-token-parsing.md):

---
"livekit-rtc": patch
---

Short description of the change, used in the release notes.

• Packages: livekit-rtc, livekit-api, livekit-protocol (one line per package)
• Types: patch, minor, major, patch-rc, minor-rc, major-rc, next-rc, promote

If this change doesn't need a release (docs, CI, tests), you can ignore this comment.

[NabiBukhsh-AI]

Hi @xianshijing-lk @cloudwebrtc,

Following up on the earlier review feedback:

• Rebased cleanly onto latest main.
• Implemented _fail_sid_waiters as suggested by @xianshijing-lk in both cancellation and connection-failure paths.
• Ensured background cleanup and supervisor tasks are isolated per event loop to avoid cross-loop task pollution.
• All 14 cancellation unit tests in test_room_connect_cancellation.py and the full test suite are passing.

All 44 CI checks and Devin review have completed with SUCCESS. Ready for your review!