From 6bccf7d02c276c16a24ee7e72eb12dfa40235c2d Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Mon, 9 Nov 2026 20:11:32 +0500 Subject: [PATCH 1/4] 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. # Conflicts: # livekit-rtc/livekit/rtc/audio_stream.py --- livekit-rtc/livekit/rtc/video_stream.py | 58 ++++++++++++----------- livekit-rtc/tests/test_video_stream.py | 62 +++++++++++++++++++++++++ 2 files changed, 93 insertions(+), 27 deletions(-) create mode 100644 livekit-rtc/tests/test_video_stream.py diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py index 07938fe5..57932fa2 100644 --- a/livekit-rtc/livekit/rtc/video_stream.py +++ b/livekit-rtc/livekit/rtc/video_stream.py @@ -63,7 +63,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( @@ -115,7 +114,10 @@ def from_track( ) def __del__(self) -> None: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) + # _ffi_queue is unset if __init__ raised early or the instance was built via __new__ + ffi_queue = getattr(self, "_ffi_queue", None) + if ffi_queue is not None: + FfiClient.instance.queue.unsubscribe(ffi_queue) def _create_owned_stream(self) -> Any: assert self._track is not None @@ -144,31 +146,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 078aca109d783137959e99a66f226514710838c7 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 20:46:01 +0500 Subject: [PATCH 2/4] 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 e1c2328df9b0b397c74d6a9dce9e94ac0ea8426c Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Tue, 20 Oct 2026 20:55:07 +0500 Subject: [PATCH 3/4] 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)) From 604d7fe5f761289352450679ad83afabf4789d22 Mon Sep 17 00:00:00 2001 From: Nabi Bukhsh Date: Mon, 9 Nov 2026 20:24:29 +0500 Subject: [PATCH 4/4] test(rtc): exit cleanly in ctypes array garbage test to avoid SIGABRT on Linux --- livekit-rtc/tests/test_ffi_client.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/livekit-rtc/tests/test_ffi_client.py b/livekit-rtc/tests/test_ffi_client.py index c57927ce..bededd77 100644 --- a/livekit-rtc/tests/test_ffi_client.py +++ b/livekit-rtc/tests/test_ffi_client.py @@ -22,6 +22,8 @@ _COUNT_ARRAY_TYPES_IN_GARBAGE = """ import ctypes import gc +import os +import sys from livekit import rtc @@ -30,7 +32,10 @@ for _ in range(10): rtc.AudioProcessingModule() gc.collect() -print(sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage)) +count = sum(isinstance(o, type(ctypes.Array)) for o in gc.garbage) +sys.stdout.write(f"{count}\\n") +sys.stdout.flush() +os._exit(0) """