Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion livekit-rtc/livekit/rtc/participant.py

Copy link
Copy Markdown
Contributor

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() at livekit-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_track by moving queue.task_done() before the error check. However, publish_track has the identical pattern:

  1. queue = self._room_queue.subscribe() at livekit-rtc/livekit/rtc/participant.py:816
  2. await queue.wait_for(...) at livekit-rtc/livekit/rtc/participant.py:819 — returns an event that the caller must mark done
  3. Error check at livekit-rtc/livekit/rtc/participant.py:823-824 raises PublishTrackError before task_done() at line 831
  4. The finally block at line 833-834 only calls unsubscribe, not task_done()

The room event loop at livekit-rtc/livekit/rtc/room.py:714-715 does:

self._room_queue.put_nowait(event)
await self._room_queue.join()

Since join() waits for all subscribers to call task_done(), the unconsumed event blocks the room event loop permanently.

The fix should mirror the unpublish_track fix: move queue.task_done() to immediately after wait_for returns, before the error check.

(Refers to lines 823-831)

Open in Devin Review

Was this helpful? React with 👍 or 👎 to provide feedback.

Original file line number Diff line number Diff line change
Expand Up @@ -861,6 +861,10 @@ async def unpublish_track(self, track_sid: str) -> None:
cb: proto_ffi.FfiEvent = await queue.wait_for(
lambda e: e.unpublish_track.async_id == resp.unpublish_track.async_id
)
# wait_for hands back one event the caller owns; mark it done before the
# error check, otherwise raising here leaves the room queue's join()
# waiting on it forever and the room event loop stalls.
queue.task_done()

if cb.unpublish_track.error:
raise UnpublishTrackError(cb.unpublish_track.error)
Expand All @@ -879,7 +883,6 @@ async def unpublish_track(self, track_sid: str) -> None:
if publication._track is not None:
publication._track._set_room(None)
publication._track = None
queue.task_done()
finally:
self._room_queue.unsubscribe(queue)

Expand Down
71 changes: 71 additions & 0 deletions livekit-rtc/tests/test_unpublish_track_deadlock.py
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")
Loading