From 2ceac5a50f2a47c9cbb079e68e18bfb26899c224 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 20:37:47 +0500 Subject: [PATCH 1/3] fix(rtc): signal eos on VideoStream queue and guard stream finalizers In VideoStream._run(), push None to self._queue when receiving an eos event, matching AudioStream. Without this terminal sentinel, consumers awaiting __anext__() remain suspended indefinitely on stream completion or aclose(). Wrap _run() in try/finally to guarantee unsubscribing from the FFI event queue even on cancellation or error. Remove duplicate self._format assignment in VideoStream.__init__. Guard __del__ in VideoStream and AudioStream with hasattr(self, '_ffi_queue') to prevent AttributeError during garbage collection of partially initialized instances. --- livekit-rtc/livekit/rtc/audio_stream.py | 3 +- livekit-rtc/livekit/rtc/video_stream.py | 56 +++++++++++----------- livekit-rtc/tests/test_video_stream.py | 62 +++++++++++++++++++++++++ 3 files changed, 93 insertions(+), 28 deletions(-) create mode 100644 livekit-rtc/tests/test_video_stream.py diff --git a/livekit-rtc/livekit/rtc/audio_stream.py b/livekit-rtc/livekit/rtc/audio_stream.py index 97772f4f..f4f880b6 100644 --- a/livekit-rtc/livekit/rtc/audio_stream.py +++ b/livekit-rtc/livekit/rtc/audio_stream.py @@ -248,7 +248,8 @@ def from_track( ) def __del__(self) -> None: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) + if hasattr(self, "_ffi_queue"): + FfiClient.instance.queue.unsubscribe(self._ffi_queue) def _create_owned_stream(self) -> Any: assert self._track is not None diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py index eeced550..987a5d84 100644 --- a/livekit-rtc/livekit/rtc/video_stream.py +++ b/livekit-rtc/livekit/rtc/video_stream.py @@ -58,7 +58,6 @@ def __init__( self._track: Track | None = track self._format = format self._capacity = capacity - self._format = format stream: Any = None if "participant" in kwargs: stream = self._create_owned_stream_from_participant( @@ -109,7 +108,8 @@ def from_track( ) def __del__(self) -> None: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) + if hasattr(self, "_ffi_queue"): + FfiClient.instance.queue.unsubscribe(self._ffi_queue) def _create_owned_stream(self) -> Any: assert self._track is not None @@ -138,31 +138,33 @@ def _create_owned_stream_from_participant( return resp.video_stream_from_participant.stream async def _run(self) -> None: - while True: - event = await self._ffi_queue.wait_for(self._is_event) - video_event = event.video_stream_event - - if video_event.HasField("frame_received"): - frame_received = video_event.frame_received - owned_buffer_info = frame_received.buffer - frame = VideoFrame._from_owned_info(owned_buffer_info) - metadata: proto_video_frame.FrameMetadata | None = None - if frame_received.HasField("metadata"): - metadata = proto_video_frame.FrameMetadata() - metadata.CopyFrom(frame_received.metadata) - - event = VideoFrameEvent( - frame=frame, - timestamp_us=frame_received.timestamp_us, - rotation=frame_received.rotation, - metadata=metadata, - ) - - self._queue.put(event) - elif video_event.HasField("eos"): - break - - FfiClient.instance.queue.unsubscribe(self._ffi_queue) + try: + while True: + event = await self._ffi_queue.wait_for(self._is_event) + video_event = event.video_stream_event + + if video_event.HasField("frame_received"): + frame_received = video_event.frame_received + owned_buffer_info = frame_received.buffer + frame = VideoFrame._from_owned_info(owned_buffer_info) + metadata: proto_video_frame.FrameMetadata | None = None + if frame_received.HasField("metadata"): + metadata = proto_video_frame.FrameMetadata() + metadata.CopyFrom(frame_received.metadata) + + event = VideoFrameEvent( + frame=frame, + timestamp_us=frame_received.timestamp_us, + rotation=frame_received.rotation, + metadata=metadata, + ) + + self._queue.put(event) + elif video_event.HasField("eos"): + self._queue.put(None) + break + finally: + FfiClient.instance.queue.unsubscribe(self._ffi_queue) async def aclose(self) -> None: self._ffi_handle.dispose() diff --git a/livekit-rtc/tests/test_video_stream.py b/livekit-rtc/tests/test_video_stream.py new file mode 100644 index 00000000..adf83b52 --- /dev/null +++ b/livekit-rtc/tests/test_video_stream.py @@ -0,0 +1,62 @@ +# 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. + +from __future__ import annotations + +import asyncio +from unittest.mock import MagicMock, patch + +from livekit import rtc +from livekit.rtc._ffi_client import FfiClient +from livekit.rtc._proto import ffi_pb2 as proto_ffi +from livekit.rtc._proto import video_frame_pb2 as proto_video_frame + + +def test_video_stream_del_without_ffi_queue() -> None: + # A partially constructed VideoStream collected during error in __init__ + # must not raise AttributeError when __del__ is called. + stream = rtc.VideoStream.__new__(rtc.VideoStream) + stream.__del__() + + +async def test_video_stream_eos_terminates_iteration() -> None: + track = MagicMock(spec=rtc.Track) + stream_info = MagicMock() + stream_info.handle.id = 12345 + stream_info.info = MagicMock() + + with patch.object(rtc.VideoStream, "_create_owned_stream", return_value=stream_info): + video_stream = rtc.VideoStream(track) + # Mark mock handle disposed so GC does not attempt to drop non-existent handle + video_stream._ffi_handle._disposed = True + + async def consume() -> list[rtc.VideoFrameEvent]: + frames: list[rtc.VideoFrameEvent] = [] + async for frame in video_stream: + frames.append(frame) + return frames + + consume_task = asyncio.create_task(consume()) + + # Send EOS event for this stream handle + event = proto_ffi.FfiEvent() + event.video_stream_event.stream_handle = 12345 + event.video_stream_event.eos.CopyFrom(proto_video_frame.VideoStreamEOS()) + FfiClient.instance.queue.put(event) + + # Iteration must complete cleanly without hanging + frames = await asyncio.wait_for(consume_task, timeout=2.0) + assert frames == [] + + await video_stream.aclose() From cfc66b0427c3119592f61470ea14ef96cc0f892b Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 20:46:01 +0500 Subject: [PATCH 2/3] fix(rtc): do not evict buffered frames when enqueuing terminal sentinel in RingQueue When an EOS sentinel (None) is enqueued to RingQueue, do not evict the oldest buffered frame when at capacity. This ensures that the final frame received before EOS is preserved and consumed by the awaiting iterator rather than discarded. --- livekit-rtc/livekit/rtc/_utils.py | 5 ++-- livekit-rtc/tests/test_video_stream.py | 39 ++++++++++++++++++++++++++ 2 files changed, 42 insertions(+), 2 deletions(-) diff --git a/livekit-rtc/livekit/rtc/_utils.py b/livekit-rtc/livekit/rtc/_utils.py index 088fb0e1..5f23cb3a 100644 --- a/livekit-rtc/livekit/rtc/_utils.py +++ b/livekit-rtc/livekit/rtc/_utils.py @@ -82,8 +82,9 @@ def __init__(self, capacity: int = 0) -> None: self._event = asyncio.Event() def put(self, item: T) -> None: - if self._capacity > 0 and len(self._queue) == self._capacity: - self._queue.popleft() + if item is not None and self._capacity > 0: + while len(self._queue) >= self._capacity: + self._queue.popleft() self._queue.append(item) self._event.set() diff --git a/livekit-rtc/tests/test_video_stream.py b/livekit-rtc/tests/test_video_stream.py index adf83b52..0d7b8a66 100644 --- a/livekit-rtc/tests/test_video_stream.py +++ b/livekit-rtc/tests/test_video_stream.py @@ -60,3 +60,42 @@ async def consume() -> list[rtc.VideoFrameEvent]: assert frames == [] await video_stream.aclose() + + +async def test_video_stream_eos_does_not_evict_last_frame_at_capacity_one() -> None: + track = MagicMock(spec=rtc.Track) + stream_info = MagicMock() + stream_info.handle.id = 54321 + stream_info.info = MagicMock() + + with patch.object(rtc.VideoStream, "_create_owned_stream", return_value=stream_info): + video_stream = rtc.VideoStream(track, capacity=1) + video_stream._ffi_handle._disposed = True + + mock_frame = MagicMock(spec=rtc.VideoFrame) + with patch.object(rtc.VideoFrame, "_from_owned_info", return_value=mock_frame): + # Enqueue frame and EOS before consumer reads + frame_event = proto_ffi.FfiEvent() + frame_event.video_stream_event.stream_handle = 54321 + frame_event.video_stream_event.frame_received.timestamp_us = 1000 + frame_event.video_stream_event.frame_received.rotation = ( + proto_video_frame.VideoRotation.VIDEO_ROTATION_0 + ) + FfiClient.instance.queue.put(frame_event) + + eos_event = proto_ffi.FfiEvent() + eos_event.video_stream_event.stream_handle = 54321 + eos_event.video_stream_event.eos.CopyFrom(proto_video_frame.VideoStreamEOS()) + FfiClient.instance.queue.put(eos_event) + + # Consume stream + frames: list[rtc.VideoFrameEvent] = [] + async for frame in video_stream: + frames.append(frame) + + # The final frame must not be evicted by EOS sentinel + assert len(frames) == 1 + assert frames[0].frame is mock_frame + assert frames[0].timestamp_us == 1000 + + await video_stream.aclose() From 4e2ee875689beb4271b820dd4101485464da6ff4 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 20:55:07 +0500 Subject: [PATCH 3/3] test(rtc): reduce apm iteration count in ctypes array garbage test Reduce the iteration count from 50 to 10 when checking for ctypes array leaks in gc.garbage. Setting DEBUG_SAVEALL prevents Python from collecting cyclic garbage and keeps APMs and their internal threads alive, which can exhaust system threads and cause SIGABRT on constrained CI runners like Ubuntu Python 3.10. --- livekit-rtc/tests/test_ffi_client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index 53873884..c57927ce 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -27,7 +27,7 @@ gc.collect() gc.set_debug(gc.DEBUG_SAVEALL) -for _ in range(50): +for _ in range(10): rtc.AudioProcessingModule() gc.collect() print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage))