Repository navigation
Conversation
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 livekit#784 Refs livekit#804
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.
| @@ -554,13 +555,25 @@ def on_participant_connected(participant): | |||
| self._ffi_queue = FfiClient.instance.queue.subscribe(self._loop) | |||
There was a problem hiding this comment.
connect() does not wait for _aborted_connect_tasks, so a retry on the same Room can race the still-in-flight join from the cancelled attempt.
Scenario: await asyncio.wait_for(room.connect(url, token), 0.2) in a retry loop, with no disconnect() between attempts. Attempt 2 joins, then attempt 1's FFI join completes with the same identity. The server evicts attempt 2 as DUPLICATE_IDENTITY, and the cleanup then closes attempt 1. The caller is left with a dead room, which is the #804 symptom again.
The retry numbers in the description were measured with a disconnect() between attempts, so this path is not covered. Awaiting the same shielded gather at the top of connect() closes the race. A test for retry-on-the-same-Room-without-disconnect would also be good.
There was a problem hiding this comment.
Good catch. connect() now awaits the same shielded gather before it joins (_wait_aborted_connects(), shared with disconnect()). Added test_a_retry_on_the_same_room_waits_for_the_cancelled_attempt, which fails on the previous head. Live against livekit-server 1.13.7: wait_for(connect, 0.02) then a retry on the same Room with no disconnect, 10 trials. Previous head: 3 of 10 retries never completed (10 s timeout). Now: 10 of 10 connected and still connected 3 s later.
| if cb.connect.error: | ||
| return | ||
|
|
||
| ffi_handle = FfiHandle(cb.connect.result.room.handle.id) |
There was a problem hiding this comment.
Only the room handle gets wrapped in FfiHandle here. The other owned handles in cb.connect.result are never wrapped, so they are never dropped: the local participant, the remote participants, and the publications. The same applies to owned handles carried by room events (participant_connected, track_published, ...) that the FFI emits between the ready request and the disconnect.
On the normal path, LocalParticipant, RemoteParticipant and RemoteTrackPublication own these handles and free them. Here, each cancelled connect into a populated room leaks handles. That adds up in a long-lived worker that times out connects repeatedly.
There was a problem hiding this comment.
Fixed. The cleanup now drops every owned handle in the connect result except the room (local participant, remote participants, publications), then after the disconnect drains the room's events until its EOS and drops the handles they carry. The room handle goes last. The walk is generic over the protobuf (any FfiOwnedHandle at any depth), so new event types are covered. The EOS wait is bounded at 5 s so a missing EOS can't hang disconnect(). Covered by test_a_cancelled_connect_into_a_populated_room_drops_every_handle.
| cb: proto_ffi.FfiEvent = await queue.wait_for( | ||
| lambda e: e.connect.async_id == resp.connect.async_id | ||
| ) | ||
| except asyncio.CancelledError: |
There was a problem hiding this comment.
Only CancelledError triggers the hand-off. Other BaseExceptions at this await take the finally path, which unsubscribes the queue and leaves the FFI waiting for ReadyForRoomEvent. Examples: GeneratorExit when the coroutine is closed during loop shutdown, or KeyboardInterrupt. The result is the same panic and SIGTERM 15s later.
Catching BaseException and re-raising would cover more exits. Agreed that nothing helps once the loop is gone.
There was a problem hiding this comment.
Done: the hand-off now catches BaseException and re-raises. If the loop is already closed when the hand-off tries to create its task, it logs a warning and unsubscribes instead of masking the original exception. Covered by test_an_exit_other_than_cancellation_still_closes_the_room (closes the coroutine at the await, i.e. GeneratorExit).
| self._ffi_queue = FfiClient.instance.queue.subscribe(self._loop) | ||
|
|
||
| queue = FfiClient.instance.queue.subscribe() | ||
| aborted = False |
There was a problem hiding this comment.
Nit: the aborted flag, the nested try, and the conditional unsubscribe in finally are hard to follow. A flat try: cb = await ... except CancelledError: hand off; raise with an explicit unsubscribe on the success path does the same job without the flag.
There was a problem hiding this comment.
Flattened as suggested: try: resp = request(...), then try: cb = await ... except BaseException: hand off; raise, and an explicit unsubscribe on the success path. The flag is gone.
| FfiClient.instance.request(ready_req) | ||
|
|
||
| close_req = proto_ffi.FfiRequest() | ||
| close_req.disconnect.room_handle = ffi_handle.handle |
There was a problem hiding this comment.
Nit: this subscribe, disconnect request, wait_for(async_id), unsubscribe block is a copy of the one in disconnect(). A small _request_disconnect(handle, reason) coroutine used in both places keeps the two from drifting apart.
There was a problem hiding this comment.
Extracted _request_disconnect(room_handle, reason); disconnect() and the cleanup both use it.
| assert len(FfiClient.instance.queue._subscribers) == subscribers_before | ||
|
|
||
|
|
||
| async def test_a_cancelled_disconnect_still_lets_the_cleanup_finish( |
There was a problem hiding this comment.
This test only asserts that ready_for_room_event was sent. It does not check that disconnect followed or that the subscriber count returned to baseline. It also returns without awaiting the cleanup task, so on a slow run a pending task and a close_queue subscriber bound to this test's loop can outlive the test.
Await asyncio.gather(*room._aborted_connect_tasks) at the end, and assert the full request sequence and subscriber count as the first test does.
There was a problem hiding this comment.
Updated: it now awaits asyncio.gather(*room._aborted_connect_tasks) and asserts the full connect / ready / disconnect sequence and the subscriber count. The fake FFI also emits the room's EOS after a disconnect now, like the real one.
There was a problem hiding this comment.
Thanks for the detailed write-up and repro numbers. The core approach (hand the in-flight connect to a task that answers ReadyForRoomEvent and then closes) looks right to me. Inline comments below. The first two are the ones I would like addressed before merge.
One higher-level note: the root cause is in livekit-ffi. A missing ReadyForRoomEvent becomes a panic, and the panic handler SIGTERMs the process. This PR is a reasonable Python-side mitigation, but the Node, Unity and other FFI clients keep the crash. The window where the loop stops before the cleanup task runs also remains. The FFI fix is to log, drop the room and emit a disconnect on a ready timeout rather than panic, plus ideally an abort for in-flight connects (#804).
From review on livekit#813: - connect() now waits for a cancelled earlier attempt's cleanup before it joins. A retry on the same Room without disconnect() could otherwise race the still-in-flight join, get evicted as a duplicate identity, and leave the caller a dead room. - The cleanup drops every owned handle the aborted connect returned (local participant, remote participants, publications) and the ones in room events sent before the disconnect, until the room's end of stream. - Any BaseException at the connect await hands the room off, not only CancelledError (GeneratorExit on loop shutdown, KeyboardInterrupt). - Flattened the hand-off and shared the disconnect request with disconnect() through _request_disconnect(). - Tests: retry on the same Room without disconnect, a populated room's handles, a non-cancellation exit, and the cancelled-disconnect test now awaits the cleanup and checks the full request sequence and subscribers.
|
Thanks for the review. All six are addressed in c249c49; replies inline. Full rtc test suite passes (63 passed, 14 skipped), ruff and mypy clean. Agreed on the root cause: this only mitigates the Python side. The ready-timeout panic in livekit-ffi still takes down the Node and Unity clients, and the window where the loop stops before the cleanup task runs is still there. Logging, dropping the room and emitting a disconnect on a ready timeout (plus an abort for in-flight connects, #804) is the real fix. |
Cancelling
room.connect()takes the process down with it, as #784 describes. I couldreproduce it exactly as filed, so here is a fix for the Python side of it.
connect()sendsReadyForRoomEventRequestas the last statement in its body, so acoroutine cancelled anywhere inside it never gets there. The FFI server has already
answered the connect request by then and is waiting for that ready request. After 15s it
gives up:
and
_ffi_client.pyanswers every panic withos.kill(os.getpid(), signal.SIGTERM).There is no way around it from the caller side.
RoomOptions.connect_timeoutisdocumented as per signal connection attempt rather than a bound on the join, and
disconnect()early returns whileisconnected()is false, so it does nothing after acancel.
The only cancellation point in
connect()is the await on the connect callback,everything after it is synchronous up to the ready request. So this catches
CancelledErrorthere and hands the room to a task that outlives the cancellation: waitfor the connect callback the server is going to send anyway, answer the ready request,
then disconnect.
disconnect()awaits that task, so a caller that cancels and thendisconnects is deterministic.
Measured against
livekit-server --dev1.13.7, macOS arm64, CPython 3.12, cancellingwith a 0.2s
wait_for:before
after: the process survives, and
disconnect()takes about 0.25s actually closing theroom.
That second part is the symptom in #804. Reconnecting with the same identity after a
cancelled connect:
wait_pc_connection timed outTwo things this deliberately does not do.
#804 asks for the FFI handle and the connect request to be decoupled so an abort can be
sent mid-connect. This does not do that. The connect still runs to completion inside the
FFI and is closed immediately afterwards, which removes the duplicate-identity symptom
but is not the same thing, so I have left that issue open rather than claiming it.
#785 asks for the unconditional SIGTERM on panic to be reconsidered. Left alone, it reads
like a policy call for you rather than something to slip into a bug fix. The reporter of
both frames it the same way.
One limitation worth stating: the cleanup runs as a task, so if the event loop stops
before it gets a turn (cancel, then exit immediately) the FFI can still time out.
Awaiting
disconnect()is what makes it deterministic.Tests are in
livekit-rtc/tests/test_connect_cancellation.py. They use the fake-FFI stylefrom
test_audio_stream_room_lifecycle.pyso they need no credentials: they assert therequest sequence is connect → ready_for_room_event → disconnect, that a connect error
sends nothing further, and that no queue subscription leaks. Both fail on main. Locally
the rtc suite is 59 passed / 14 skipped (the credentialed ones), ruff 0.15.4 and mypy
clean apart from the pre-existing
jupyter.pyIPython imports.Claude helped me write this. I can explain every line of it, and the numbers above are
from runs on my own machine.
Fixes #784
Refs #804