fix(rtc): mark room queue task done on publish and unpublish track errors

livekit-rtc/livekit/rtc/participant.py

@@ -1089,6 +1089,10 @@ async def publish_track(
             cb: proto_ffi.FfiEvent = await queue.wait_for(
                 lambda e: e.publish_track.async_id == resp.publish_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.publish_track.error:
                 raise PublishTrackError(cb.publish_track.error)
@@ -1098,7 +1102,6 @@ async def publish_track(
             track._info.sid = track_publication.sid
             self._track_publications[track_publication.sid] = track_publication
 
-            queue.task_done()
             return track_publication
         finally:
             self._room_queue.unsubscribe(queue)
@@ -1131,6 +1134,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)
@@ -1149,7 +1156,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)
 

livekit-rtc/tests/test_publish_track_deadlock.py

@@ -0,0 +1,227 @@
+# 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 tests for publish_track and unpublish_track room queue lifecycle.
+
+When the FFI reports a publish or unpublish error, publish_track/unpublish_track raises.
+The event returned by queue.wait_for() must be marked done immediately so that
+Room._listen_task's join() does not remain blocked on the unconsumed event.
+"""
+
+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, Queue
+from livekit.rtc.participant import PublishTrackError, UnpublishTrackError
+
+_ASYNC_ID = 54321
+
+
+@pytest.mark.asyncio
+async def test_publish_track_marks_event_done_before_raising_error() -> None:
+    """Verify that publish_track marks the event done immediately upon receipt,
+
+    even when the FFI event reports an error. It must not rely solely on
+    unsubscription release to account for the event.
+    """
+    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 = {}
+
+    track = MagicMock()
+    track._ffi_handle = MagicMock(handle=2)
+
+    resp = proto_ffi.FfiResponse()
+    resp.publish_track.async_id = _ASYNC_ID
+
+    error_event = proto_ffi.FfiEvent()
+    error_event.publish_track.async_id = _ASYNC_ID
+    error_event.publish_track.error = "publish track failed"
+
+    called_while_subscribed = False
+
+    orig_subscribe = room_queue.subscribe
+
+    def spy_subscribe() -> Queue[proto_ffi.FfiEvent]:
+        q = orig_subscribe()
+        orig_task_done = q.task_done
+
+        def spy_task_done() -> None:
+            nonlocal called_while_subscribed
+            if q in room_queue._subscribers:
+                called_while_subscribed = True
+            orig_task_done()
+
+        q.task_done = spy_task_done  # type: ignore[method-assign]
+        return q
+
+    room_queue.subscribe = spy_subscribe  # type: ignore[method-assign]
+
+    with patch.object(FfiClient.instance, "request", return_value=resp):
+        pub_task = asyncio.create_task(participant.publish_track(track))
+        await asyncio.sleep(0)  # let publish_track subscribe
+
+        room_queue.put_nowait(error_event)
+
+        with pytest.raises(PublishTrackError, match="publish track failed"):
+            await pub_task
+
+    assert called_while_subscribed, (
+        "queue.task_done() was not called for the event while the subscriber was active"
+    )
+
+
+@pytest.mark.asyncio
+async def test_publish_track_error_releases_room_queue_join() -> None:
+    """Verify that Room._listen_task's join() finishes cleanly when publish_track fails."""
+    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 = {}
+
+    track = MagicMock()
+    track._ffi_handle = MagicMock(handle=2)
+
+    resp = proto_ffi.FfiResponse()
+    resp.publish_track.async_id = _ASYNC_ID
+
+    error_event = proto_ffi.FfiEvent()
+    error_event.publish_track.async_id = _ASYNC_ID
+    error_event.publish_track.error = "publish track failed"
+
+    async def deliver_event_like_listen_task() -> None:
+        room_queue.put_nowait(error_event)
+        await room_queue.join()
+
+    with patch.object(FfiClient.instance, "request", return_value=resp):
+        pub_task = asyncio.create_task(participant.publish_track(track))
+        await asyncio.sleep(0)
+        listen_task = asyncio.create_task(deliver_event_like_listen_task())
+
+        try:
+            with pytest.raises(PublishTrackError):
+                await pub_task
+
+            try:
+                await asyncio.wait_for(listen_task, timeout=2.0)
+            except asyncio.TimeoutError:
+                pytest.fail("room queue join() deadlocked after publish_track error")
+        finally:
+            if not listen_task.done():
+                listen_task.cancel()
+
+
+@pytest.mark.asyncio
+async def test_unpublish_track_marks_event_done_before_raising_error() -> None:
+    """Verify that unpublish_track marks the event done immediately upon receipt,
+
+    even when the FFI event reports an error. It must not rely solely on
+    unsubscription release to account for the event.
+    """
+    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 track failed"
+
+    called_while_subscribed = False
+
+    orig_subscribe = room_queue.subscribe
+
+    def spy_subscribe() -> Queue[proto_ffi.FfiEvent]:
+        q = orig_subscribe()
+        orig_task_done = q.task_done
+
+        def spy_task_done() -> None:
+            nonlocal called_while_subscribed
+            if q in room_queue._subscribers:
+                called_while_subscribed = True
+            orig_task_done()
+
+        q.task_done = spy_task_done  # type: ignore[method-assign]
+        return q
+
+    room_queue.subscribe = spy_subscribe  # type: ignore[method-assign]
+
+    with patch.object(FfiClient.instance, "request", return_value=resp):
+        unpub_task = asyncio.create_task(participant.unpublish_track("TR_test"))
+        await asyncio.sleep(0)
+
+        room_queue.put_nowait(error_event)
+
+        with pytest.raises(UnpublishTrackError, match="unpublish track failed"):
+            await unpub_task
+
+    assert called_while_subscribed, (
+        "queue.task_done() was not called for the event while the subscriber was active"
+    )
+
+
+@pytest.mark.asyncio
+async def test_unpublish_track_error_releases_room_queue_join() -> None:
+    """Verify that Room._listen_task's join() finishes cleanly when unpublish_track fails."""
+    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 track failed"
+
+    async def deliver_event_like_listen_task() -> None:
+        room_queue.put_nowait(error_event)
+        await room_queue.join()
+
+    with patch.object(FfiClient.instance, "request", return_value=resp):
+        unpub_task = asyncio.create_task(participant.unpublish_track("TR_test"))
+        await asyncio.sleep(0)
+        listen_task = asyncio.create_task(deliver_event_like_listen_task())
+
+        try:
+            with pytest.raises(UnpublishTrackError):
+                await unpub_task
+
+            try:
+                await asyncio.wait_for(listen_task, timeout=2.0)
+            except asyncio.TimeoutError:
+                pytest.fail("room queue join() deadlocked after unpublish_track error")
+        finally:
+            if not listen_task.done():
+                listen_task.cancel()