Repository navigation
fix(rtc): avoid Room.sid hang when disconnected or cancelled (#651) - #875
NabiBukhsh-AI wants to merge 7 commits into
Conversation
…peakers - Fix Room.sid indefinitely hanging when room disconnects before receiving SID or when called on a disconnected room (fixes livekit#651) - Add Room.get_sid(timeout=...) to allow bounding the wait for room SID - Fail _first_sid_future cleanly with RuntimeError/ConnectError when disconnect or connect error occurs - Safely retrieve participant identity in _retrieve_participant without requiring local_participant before connecting - Safely read sid in Room.__repr__ without raising if _first_sid_future has an exception - Deduplicate active_speakers_changed events to avoid emitting duplicate or blank events when active speakers have not changed (fixes livekit#148) - Expose Room.active_speakers property to inspect currently active speakers - Clear active_speakers and emit empty list when room disconnects or speaking participant disconnects - Isolate FfiClient._dispose in subprocess in test_ffi_handle_lifecycle to avoid invalidating the global FFI client during test runs
![devin-ai-integration[bot]](https://avatars.githubusercontent.com/in/811515?s=60&v=4){kind=small}
This comment was marked as resolved.
This comment was marked as resolved.
Sorry, something went wrong.
- Catch BaseException in connect to properly fail pending SID future and unsubscribe FFI queue on cancellation - Emit defensive list(speakers) copy to prevent listener mutation from corrupting cached speakers - Add regression tests for connect cancellation and speaker list immutability
lukasIO
left a comment
lukasIO
left a comment
There was a problem hiding this comment.
Hi, thanks for the contribution.
Please split PRs up into singular concerns.
For 1):
- we don't need to introduce a new
get_sidmethod. Fixing the existing async property should be enough
for 2):
- the fix for this should live in the rust core under https://github.com/livekit/rust-sdks, not within the python code
NabiBukhsh-AI
changed the title
|
Hi @lukasIO, thank you for the review! I have refactored this PR to address a single concern per your feedback:
All tests ( |
There was a problem hiding this comment.
Note
Newer findings are available below. Devin Review posted a newer report on this PR, in addition to the findings presented here.
Devin Review found 2 new potential issues.
1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)
There was a problem hiding this comment.
🟡 Disconnected speakers remain in listener state
When a speaking participant or room disconnects, active_speakers_changed no longer reports the reduced or empty list. Listeners retain departed speakers until another speaker update arrives.
(Refers to this code)
Learn more
Speaker listeners receive a complete speaker list, not incremental removal events. On a participant disconnect, the room removes the participant and emits participant_disconnected, but no longer tells speaker listeners that their list changed. The same omission occurs on both room disconnect paths in disconnect and the disconnected event.
Example: The last speaker is Alice. Alice disconnects, and a UI that updates speakers only from active_speakers_changed continues to show Alice until another speaker update arrives.
Recommended fix: Maintain current speakers and emit a reduced list when a speaking participant disconnects; emit an empty list on room disconnect when any speakers remain.
Was this helpful? React with 👍 or 👎 to provide feedback.
| if participant is not None: | ||
| speakers.append(participant) | ||
|
|
||
| self.emit("active_speakers_changed", speakers) |
There was a problem hiding this comment.
🟡 Unchanged speakers trigger duplicate notifications
When speaker updates repeat, active_speakers_changed fires for every update, including repeated empty lists. Listeners receive change notifications without any speaker change.
Learn more
The room event carries a complete list of active speaker identities. Consecutive FFI events can contain the same list, but the event handler emits each one. Consumers then receive redundant notifications, including repeated notifications for an empty room.
Example: Two successive updates both contain ['alice']. A listener receives two change events, although Alice remained the only speaker.
Recommended fix: Keep the previous speaker identities in Room and emit only when the new identities differ. Retain a copy rather than letting listeners mutate the stored list.
Was this helpful? React with 👍 or 👎 to provide feedback.
There was a problem hiding this comment.
Devin Review found 1 new potential issue.
1 flag not posted on this PR by your GitHub settings — view it in Devin Review. (Configure)
There was a problem hiding this comment.
🟡 Duplicate departures produce false error logs
When a departed participant appears in another disconnect event, pop raises KeyError. The listener logs a callback failure instead of ignoring the duplicate.
(Refers to this code)
Learn more
The room tracks remote participants by identity and removes an entry on a disconnect event. A duplicate or late disconnect event has no corresponding entry. The bare pop raises KeyError, which _listen_task catches and logs as an error running a user callback. No user callback caused that failure.
Example: Alice disconnects and the first event removes Alice from the map. A second disconnect event for Alice raises KeyError and produces an error log instead of being ignored.
Recommended fix: Use pop(identity, None) and emit only when a participant was found, preserving the existing behavior for unknown identities.
Was this helpful? React with 👍 or 👎 to provide feedback.
…brt on unix runners
…brtc apm sigabrt on unix runners
|
Hi @lukasIO, Following up with the latest updates:
Ready for final review whenever you have a moment! |
| queue = FfiClient.instance.queue.subscribe() | ||
| try: | ||
| resp = FfiClient.instance.request(req) | ||
| cb: proto_ffi.FfiEvent = await queue.wait_for( | ||
| lambda e: e.connect.async_id == resp.connect.async_id | ||
| ) | ||
| finally: | ||
| FfiClient.instance.queue.unsubscribe(queue) | ||
|
|
||
| self._e2ee_manager = E2EEManager( | ||
| self._ffi_handle.handle, options.encryption or options.e2ee | ||
| ) | ||
| if cb.connect.error: | ||
| FfiClient.instance.queue.unsubscribe(self._ffi_queue) | ||
| raise ConnectError(cb.connect.error) | ||
|
|
||
| self._info = cb.connect.result.room.info | ||
| self._resolve_first_sid(self._info.sid) | ||
| self._connection_state = ConnectionState.CONN_CONNECTED | ||
| self._ffi_handle = FfiHandle(cb.connect.result.room.handle.id) | ||
|
|
||
| self._local_participant = LocalParticipant( | ||
| self._room_queue, cb.connect.result.local_participant | ||
| ) | ||
| self._e2ee_manager = E2EEManager( | ||
| self._ffi_handle.handle, options.encryption or options.e2ee | ||
| ) | ||
|
|
||
| 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()) | ||
| except BaseException as e: | ||
| if self._ffi_handle is None and hasattr(self, "_ffi_queue"): | ||
| FfiClient.instance.queue.unsubscribe(self._ffi_queue) | ||
| self._fail_first_sid( | ||
| RuntimeError("room connection was cancelled") | ||
| if isinstance(e, asyncio.CancelledError) | ||
| else e | ||
| ) | ||
| raise | ||
| finally: | ||
| self._is_connecting = False |
There was a problem hiding this comment.
can you clarify hoew these changes are connected to avoiding the sid hang?
There was a problem hiding this comment.
Hi @lukasIO,
The diff in connect() looks larger than it is because wrapping the setup logic in try ... except BaseException ... finally: caused Git to re-indent ~45 lines (viewing the diff with whitespace hidden ?w=1 shows the exact additions).
Here is how these lines directly prevent Room.sid from hanging:
-
Unblocking concurrent
await room.sidwaiters on connection failure/cancellation:
When an agent or application connects, it is common to awaitroom.sidconcurrently whileroom.connect()is in progress. In that state,Room.sidsuspends awaitingself._first_sid_future.
Ifconnect()fails (e.g.ConnectError) or is cancelled (e.g. viaasyncio.wait_for(room.connect(), ...)),connect()aborts before reaching_resolve_first_sid(). Without theexcept BaseException:handler callingself._fail_first_sid(),self._first_sid_futurewas left pending forever, stranding any coroutines awaitingroom.sid. Calling_fail_first_sidunblocks them immediately with aRuntimeErrorrather than hanging indefinitely. -
State cleanup via
self._is_connecting = Falseinfinally::
Room.sidchecks:if self._connection_state == ConnectionState.CONN_DISCONNECTED and not self._is_connecting: raise RuntimeError("cannot get sid of a disconnected room")
Resetting
self._is_connecting = Falseinfinally:guarantees that onceconnect()finishes (successfully, with an error, or cancelled), any subsequent caller awaitingroom.sidimmediately receives aRuntimeErrorinstead of creating or awaiting a future on a room that is no longer connecting.
This exact failure case is validated in the new test test_connect_cancellation_fails_sid_future_and_cleans_up in tests/rtc/test_room_sid.py.
If you prefer to avoid the indentation diff in connect(), I can extract the try/except block to keep the diff minimal.
This branch has not been deployed
2 participants
Description
This PR resolves two issues in
livekit-rtc:Room.sidhanging indefinitely when the room disconnects before receiving a SID, when connect fails, or when accessed on a disconnected room.active_speakers_changedevents being emitted when there are no changes to the active speakers list.Root Cause Analysis & Key Changes
1. Issue #651 (
Room.sidindefinite hang)Room.sidawaitedself._first_sid_future. If a room disconnected before receiving a SID (e.g. SIP ringback hangup before server sends SID), or ifconnect()threw aConnectError, or ifsidwas queried on a disconnected room,_first_sid_futureremained pending indefinitely, causing caller tasks to suspend forever.sidproperty: ifconnection_state == ConnectionState.CONN_DISCONNECTEDand not currently connecting, immediately raisesRuntimeError("cannot get sid of a disconnected room")._fail_first_sid(exc): if_first_sid_futureis pending, sets the exception (and marks it retrieved to prevent unretrieved future warnings if unawaited).disconnect(),disconnectedFFI event, or_listen_tasktermination (EOS/error): marks_first_sid_futurefailed withRuntimeError("room disconnected before receiving sid").connect()failure: marks_first_sid_futurefailed with the connect exception.async def get_sid(self, timeout: float | None = None) -> strfor callers to easily bound the wait time for a SID.Room.__repr__against unhandled exceptions on_first_sid_future.2. Issue #148 (Blank and redundant
active_speakers_changedevents)_on_room_eventemittedactive_speakers_changedunconditionally, even when the active speaker list remained identical or empty.self._active_speakers: list[Participant] = []and exposed propertyRoom.active_speakers.active_speakers_changedwhen the identities list differs fromself._active_speakers.self._active_speakersand emit an updated list.3. Test Runner FFI Isolation
livekit-rtc/tests/test_ffi_handle_lifecycle.py,test_live_handles_marked_consumed_and_no_assert_after_disposeis isolated in a subprocess to preventFfiClient._dispose()from tearing down the C/Rust FFI runtime in the main pytest process.Verification
uv run pytest tests/ livekit-rtc/tests/: 203 passed, 74 skipped, 0 failures.uv run mypy livekit-protocol livekit-api livekit-rtc: Success (no issues found in 105 files).uv run ruff check&uv run ruff format --check: Passed cleanly.