Fix unpublish_track deadlock on FFI error #752

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:

queue = self._room_queue.subscribe() at livekit-rtc/livekit/rtc/participant.py:816
await queue.wait_for(...) at livekit-rtc/livekit/rtc/participant.py:819 — returns an event that the caller must mark done
Error check at livekit-rtc/livekit/rtc/participant.py:823-824 raises PublishTrackError before task_done() at line 831
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)

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

Show comments View file Open in desktop

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")

Navigation Menu

Uh oh!

Fix unpublish_track deadlock on FFI error #752

Are you sure you want to change the base?

Uh oh!

Fix unpublish_track deadlock on FFI error #752

Filter by extension

Uh oh!

Uh oh!

Diff view

Diff view

There are no files selected for viewing

@devin-ai-integration devin-ai-integration Bot Jul 11, 2026

Choose a reason for hiding this comment

Uh oh!

Uh oh!