-
Notifications
You must be signed in to change notification settings - Fork 122
Fix unpublish_track deadlock on FFI error #752
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Open
apoorva-01
wants to merge
1
commit into
livekit:main
Choose a base branch
from
apoorva-01:fix/565-unpublish-deadlock
base: main
Could not load branches
Branch not found: {{ refName }}
Loading
Could not load tags
Nothing to show
Loading
Are you sure you want to change the base?
Some commits from the old base branch may be removed from the timeline,
and old review comments may become outdated.
+75
−1
Open
Changes from all commits
Commits
File filter
Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,71 @@ | ||
| # 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. | ||
|
|
||
| """Regression test for unpublish_track deadlocking the room event loop. | ||
|
|
||
| When the FFI reports an unpublish error, unpublish_track raises. If it raises | ||
| before marking its room-queue event done, Room._listen_task's join() blocks on | ||
| that unconsumed event forever and the room stops processing events. | ||
| """ | ||
|
|
||
| from __future__ import annotations | ||
|
|
||
| import asyncio | ||
| from unittest.mock import MagicMock, patch | ||
|
|
||
| import pytest | ||
|
|
||
| from livekit import rtc | ||
| from livekit.rtc._ffi_client import FfiClient | ||
| from livekit.rtc._proto import ffi_pb2 as proto_ffi | ||
| from livekit.rtc._utils import BroadcastQueue | ||
| from livekit.rtc.participant import UnpublishTrackError | ||
|
|
||
| _ASYNC_ID = 4242 | ||
|
|
||
|
|
||
| async def test_unpublish_track_error_does_not_deadlock_room_queue() -> None: | ||
| room_queue: BroadcastQueue[proto_ffi.FfiEvent] = BroadcastQueue() | ||
|
|
||
| participant = rtc.LocalParticipant.__new__(rtc.LocalParticipant) | ||
| participant._room_queue = room_queue | ||
| participant._ffi_handle = MagicMock(handle=1) | ||
| participant._track_publications = {} | ||
|
|
||
| resp = proto_ffi.FfiResponse() | ||
| resp.unpublish_track.async_id = _ASYNC_ID | ||
|
|
||
| error_event = proto_ffi.FfiEvent() | ||
| error_event.unpublish_track.async_id = _ASYNC_ID | ||
| error_event.unpublish_track.error = "unpublish failed" | ||
|
|
||
| async def deliver_event_like_listen_task() -> None: | ||
| # Mirror Room._listen_task: hand the event to subscribers, then wait for | ||
| # them to finish before moving on to the next event. | ||
| room_queue.put_nowait(error_event) | ||
| await room_queue.join() | ||
|
|
||
| with patch.object(FfiClient.instance, "request", return_value=resp): | ||
| unpublish = asyncio.ensure_future(participant.unpublish_track("TR_test")) | ||
| await asyncio.sleep(0) # let unpublish subscribe before the event is delivered | ||
| listen = asyncio.ensure_future(deliver_event_like_listen_task()) | ||
|
|
||
| with pytest.raises(UnpublishTrackError): | ||
| await unpublish | ||
|
|
||
| try: | ||
| await asyncio.wait_for(listen, timeout=5) | ||
| except asyncio.TimeoutError: | ||
| listen.cancel() | ||
| pytest.fail("room queue join() deadlocked after unpublish_track error") |
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🔴 Track publishing errors stall the room event loop with the same deadlock fixed for unpublishing
The room queue event is not marked as done (
queue.task_done()atlivekit-rtc/livekit/rtc/participant.py:831) when a publish error is raised (livekit-rtc/livekit/rtc/participant.py:824), so the room event loop blocks forever waiting on the unconsumed event.Impact: When publishing a track fails, the room stops processing all subsequent events, effectively freezing the connection.
Same deadlock pattern as the unpublish_track fix in this PR
This PR fixes the deadlock in
unpublish_trackby movingqueue.task_done()before the error check. However,publish_trackhas the identical pattern:queue = self._room_queue.subscribe()atlivekit-rtc/livekit/rtc/participant.py:816await queue.wait_for(...)atlivekit-rtc/livekit/rtc/participant.py:819— returns an event that the caller must mark donelivekit-rtc/livekit/rtc/participant.py:823-824raisesPublishTrackErrorbeforetask_done()at line 831finallyblock at line 833-834 only callsunsubscribe, nottask_done()The room event loop at
livekit-rtc/livekit/rtc/room.py:714-715does:Since
join()waits for all subscribers to calltask_done(), the unconsumed event blocks the room event loop permanently.The fix should mirror the
unpublish_trackfix: movequeue.task_done()to immediately afterwait_forreturns, before the error check.(Refers to lines 823-831)
Was this helpful? React with 👍 or 👎 to provide feedback.