From 9f8624b77826f3d962d5f4735e7305d65cad2578 Mon Sep 17 00:00:00 2001 From: shijing xian Date: Fri, 6 Nov 2026 12:47:31 -0800 Subject: [PATCH] Add zero-copy video frames and surface dropped frames `VideoStream(..., zero_copy=True)` hands out frames that read the SDK's own pixel buffer instead of a copy of it, the way the Unity SDK already does. A 1628x1920 I420 frame is 4.5 MB, so the copy this removes is 134 MB/s at 30 fps, and it ran on the event loop holding the GIL -- which also delayed the FFI callback that feeds the stream. It is opt-in because it changes the frame's lifetime. A borrowed frame's buffer belongs to the SDK and is only valid until the frame closes, so `np.frombuffer(frame.data)` outliving the frame now reads freed memory where before it read a private copy. `VideoFrame.close()`, a context manager, and `is_borrowed` make the boundary explicit; `VideoStream` closes the previous borrowed frame when the next is requested, so a plain `async for` holds at most one buffer without the application doing anything. `close()` deliberately does not try to detect outstanding views. CPython only reports an export against the exact buffer object it came from, and a view derived from `frame.data` re-exports the underlying array instead, so a check would catch almost nothing while reading as a guarantee. A test pins that behaviour so the contract is not mistaken for a bug. `VideoFrameEvent.dropped_frames` reports frames the SDK decoded but never delivered. No WebRTC stat covers them: `framesDecoded` counts them and `framesDropped` does not. Reads as 0 against an FFI build without the counter. `RingQueue` gains `on_discard`, so a bounded stream releases the buffer of an evicted frame instead of waiting for the collector, and `get_nowait` to drain on close. Co-Authored-By: Claude Opus 5 (1M context) --- livekit-rtc/livekit/rtc/_utils.py | 20 +- livekit-rtc/livekit/rtc/video_frame.py | 97 +++++++ livekit-rtc/livekit/rtc/video_stream.py | 91 ++++++- tests/rtc/test_video_frame_zero_copy.py | 334 ++++++++++++++++++++++++ 4 files changed, 535 insertions(+), 7 deletions(-) create mode 100644 tests/rtc/test_video_frame_zero_copy.py diff --git a/livekit-rtc/livekit/rtc/_utils.py b/livekit-rtc/livekit/rtc/_utils.py index 91175815..e036d8a9 100644 --- a/livekit-rtc/livekit/rtc/_utils.py +++ b/livekit-rtc/livekit/rtc/_utils.py @@ -17,7 +17,7 @@ from collections import deque import ctypes import random -from typing import Any, Callable, Generator, Generic, List, TypeVar, Union +from typing import Any, Callable, Generator, Generic, List, Optional, TypeVar, Union logger = logging.getLogger("livekit") @@ -76,14 +76,24 @@ def get_address(data: Union[bytes, bytearray, memoryview]) -> int: class RingQueue(Generic[T]): - def __init__(self, capacity: int = 0) -> None: + def __init__(self, capacity: int = 0, on_discard: Optional[Callable[[T], None]] = None) -> None: + """ + Args: + capacity: Maximum queued items, or 0 for unbounded. + on_discard: Called with the oldest item when a put evicts it. + Lets an owner of out-of-band resources (an SDK buffer, say) + release them instead of waiting for the collector. + """ self._capacity = capacity self._queue: deque[T] = deque() self._event = asyncio.Event() + self._on_discard = on_discard def put(self, item: T) -> None: if self._capacity > 0 and len(self._queue) == self._capacity: - self._queue.popleft() + discarded = self._queue.popleft() + if self._on_discard is not None: + self._on_discard(discarded) self._queue.append(item) self._event.set() @@ -93,6 +103,10 @@ async def get(self) -> T: self._event.clear() return self._queue.popleft() + def get_nowait(self) -> Optional[T]: + """Returns the next item, or None when the queue is empty.""" + return self._queue.popleft() if self._queue else None + class Queue(asyncio.Queue[T]): """asyncio.Queue with utility functions.""" diff --git a/livekit-rtc/livekit/rtc/video_frame.py b/livekit-rtc/livekit/rtc/video_frame.py index 30255323..e4aad244 100644 --- a/livekit-rtc/livekit/rtc/video_frame.py +++ b/livekit-rtc/livekit/rtc/video_frame.py @@ -50,6 +50,11 @@ def __init__( """ data = _ensure_compatible_buffer(data) + # Set for a borrowed frame only; see `_borrow_owned_info`. An owned + # frame holds its own copy of the pixels and has nothing to release. + self._owner: Optional[FfiHandle] = None + self._closed = False + self._width = width self._height = height self._type = type @@ -92,11 +97,79 @@ def data(self) -> memoryview: Returns: memoryview: The raw pixel data of the video frame as a memoryview object. + + Raises: + RuntimeError: If this is a borrowed frame that has already been + closed. See :meth:`close`. """ + if self._closed: + raise RuntimeError( + "the video frame has been closed; its pixel data belonged to the SDK " + "and has been released. Copy what you need before the frame closes." + ) return memoryview(self._data) + @property + def is_borrowed(self) -> bool: + """ + Whether this frame reads the SDK's own pixel buffer rather than a copy. + + A borrowed frame's :attr:`data` is only valid until the frame is closed; + see :meth:`close`. Frames are owned unless the stream was created with + ``zero_copy=True``. + """ + return self._owner is not None + + def close(self) -> None: + """ + Releases a borrowed frame's pixel buffer back to the SDK. + + No-op for an owned frame, and safe to call more than once. + + After this returns, :attr:`data` raises. **Anything else still pointing + at the buffer -- a ``memoryview`` taken from :attr:`data`, a ``numpy`` + array built with ``np.frombuffer`` -- points at freed memory and must + not be read.** + + There is deliberately no check for that. CPython reports an outstanding + export only against the exact buffer object it was taken from, and a + view derived from :attr:`data` re-exports the underlying array instead, + so a check here would catch almost nothing while reading as a + guarantee. Use ``np.array(frame.data)`` rather than + ``np.frombuffer(frame.data)`` when the result has to outlive the frame. + + :class:`VideoStream` closes the previous borrowed frame before yielding + the next one, so a loop that does not hold onto frames needs no explicit + call. Use the frame as a context manager to release it sooner. + """ + if self._closed or self._owner is None: + return + + self._closed = True + if isinstance(self._data, memoryview): + # Invalidates the frame's own view, so a later read of a stale + # `frame.data` fails loudly instead of reading freed memory. + self._data.release() + self._owner.dispose() + self._owner = None + + def __enter__(self) -> "VideoFrame": + return self + + def __exit__(self, *_: Any) -> None: + self.close() + + def __del__(self) -> None: + # Backstop for a borrowed frame the application never closed and the + # stream never superseded. Best effort: __del__ must not raise. + try: + self.close() + except Exception: + pass + @staticmethod def _from_owned_info(owned_info: proto_video.OwnedVideoBuffer) -> "VideoFrame": + """Copies the pixels out of the SDK buffer and releases the handle.""" info = owned_info.info data_len = _get_plane_length(info.type, info.width, info.height) cdata = (ctypes.c_uint8 * data_len).from_address(info.data_ptr) @@ -110,6 +183,30 @@ def _from_owned_info(owned_info: proto_video.OwnedVideoBuffer) -> "VideoFrame": FfiHandle(owned_info.handle.id).dispose() return frame + @staticmethod + def _borrow_owned_info(owned_info: proto_video.OwnedVideoBuffer) -> "VideoFrame": + """ + Wraps the SDK's buffer in place, without copying it. + + The returned frame owns the FFI handle and keeps the buffer alive until + the frame is closed. This is what makes the frame's lifetime matter: + an owned frame is independent of the SDK, a borrowed one is not. + """ + info = owned_info.info + data_len = _get_plane_length(info.type, info.width, info.height) + cdata = (ctypes.c_uint8 * data_len).from_address(info.data_ptr) + frame = VideoFrame( + width=info.width, + height=info.height, + type=info.type, + # `cast` is not optional: a view of a ctypes array carries format + # " proto_video.VideoBufferInfo: info = proto_video.VideoBufferInfo() addr = get_address(self.data) diff --git a/livekit-rtc/livekit/rtc/video_stream.py b/livekit-rtc/livekit/rtc/video_stream.py index eeced550..3609682f 100644 --- a/livekit-rtc/livekit/rtc/video_stream.py +++ b/livekit-rtc/livekit/rtc/video_stream.py @@ -35,9 +35,50 @@ class VideoFrameEvent: rotation: proto_video_frame.VideoRotation metadata: proto_video_frame.FrameMetadata | None = None + dropped_frames: int = 0 + """ + Frames this stream decoded but never delivered, counted since the stream + was created. + + A frame is dropped when the SDK's internal queue overflows because this + application did not read frames fast enough. No WebRTC stat reports the + loss -- ``framesDecoded`` counts these frames and ``framesDropped`` does + not -- so this counter is the only way to observe it. + + The value is cumulative and monotonic: compare it against the previous + event's to see how many frames were lost in between. It stays at zero when + the SDK in use predates the counter. + """ + + +def _close_discarded(event: VideoFrameEvent | None) -> None: + if event is not None: + event.frame.close() + + +def _dropped_frames(frame_received: proto_video_frame.VideoFrameReceived) -> int: + """ + Reads the dropped-frame counter, tolerating an SDK that predates it. + + The field arrived with the native counter; an older `livekit_ffi` does not + define it, and reporting zero is the honest answer there -- we have no + information either way. + + `getattr` rather than attribute access so this also builds against the + generated stubs from before the field existed. Once the vendored + `rust-sdks` is bumped past it, this can become a plain read. + """ + return int(getattr(frame_received, "dropped_frames", 0)) + class VideoStream: - """VideoStream is a stream of video frames received from a RemoteTrack.""" + """VideoStream is a stream of video frames received from a RemoteTrack. + + With ``zero_copy=True`` each frame reads the SDK's own pixel buffer instead + of a copy of it, which removes a full frame-sized copy per frame. The + buffer belongs to the SDK and is only valid while the frame is open; see + :meth:`VideoFrame.close` and the ``zero_copy`` argument. + """ def __init__( self, @@ -45,16 +86,26 @@ def __init__( loop: Optional[asyncio.AbstractEventLoop] = None, capacity: int = 0, format: Optional[proto_video_frame.VideoBufferType.ValueType] = None, + zero_copy: bool = False, **kwargs: Any, ) -> None: self._loop = loop or asyncio.get_event_loop() + self._zero_copy = zero_copy + # The most recent borrowed frame handed to the application. Closed when + # the next one is requested, so a plain `async for` never leaks a + # buffer even if the application does not close frames itself. + self._borrowed: VideoFrame | None = None # Only subscribe to video_stream_event to avoid unnecessary memory allocations # from other event types (room_event, track_event, etc.) self._ffi_queue = FfiClient.instance.queue.subscribe( self._loop, filter_fn=lambda e: e.WhichOneof("message") == "video_stream_event", ) - self._queue: RingQueue[VideoFrameEvent | None] = RingQueue(capacity) + # A borrowed frame evicted by a bounded queue has to be closed, or its + # SDK buffer stays out until the collector gets to it. + self._queue: RingQueue[VideoFrameEvent | None] = RingQueue( + capacity, on_discard=_close_discarded if zero_copy else None + ) self._track: Track | None = track self._format = format self._capacity = capacity @@ -82,6 +133,7 @@ def from_participant( loop: Optional[asyncio.AbstractEventLoop] = None, format: Optional[proto_video_frame.VideoBufferType.ValueType] = None, capacity: int = 0, + zero_copy: bool = False, ) -> VideoStream: return VideoStream( participant=participant, @@ -89,6 +141,7 @@ def from_participant( loop=loop, capacity=capacity, format=format, + zero_copy=zero_copy, track=None, # type: ignore ) @@ -100,16 +153,24 @@ def from_track( loop: Optional[asyncio.AbstractEventLoop] = None, format: Optional[proto_video_frame.VideoBufferType.ValueType] = None, capacity: int = 0, + zero_copy: bool = False, ) -> VideoStream: return VideoStream( track=track, loop=loop, capacity=capacity, format=format, + zero_copy=zero_copy, ) def __del__(self) -> None: - FfiClient.instance.queue.unsubscribe(self._ffi_queue) + # Tolerate a half-built stream: if __init__ raised before subscribing + # (a failed FFI request, say), there is nothing to unsubscribe, and + # raising from __del__ only produces an unraisable-exception warning + # that hides the original error. + 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 @@ -145,7 +206,10 @@ async def _run(self) -> None: 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) + if self._zero_copy: + frame = VideoFrame._borrow_owned_info(owned_buffer_info) + else: + 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() @@ -156,6 +220,7 @@ async def _run(self) -> None: timestamp_us=frame_received.timestamp_us, rotation=frame_received.rotation, metadata=metadata, + dropped_frames=_dropped_frames(frame_received), ) self._queue.put(event) @@ -167,6 +232,16 @@ async def _run(self) -> None: async def aclose(self) -> None: self._ffi_handle.dispose() await self._task + self._release_borrowed() + # Frames still queued hold SDK buffers; release them with the stream. + while (queued := self._queue.get_nowait()) is not None: + queued.frame.close() + + def _release_borrowed(self) -> None: + """Closes the borrowed frame handed out last, if there is one.""" + if self._borrowed is not None: + self._borrowed.close() + self._borrowed = None def _is_event(self, e: proto_ffi.FfiEvent) -> bool: return e.video_stream_event.stream_handle == self._ffi_handle.handle @@ -175,6 +250,11 @@ def __aiter__(self) -> AsyncIterator[VideoFrameEvent]: return self async def __anext__(self) -> VideoFrameEvent: + # Asking for the next frame ends the previous one's borrow. This is + # what keeps a plain `async for` from accumulating SDK buffers when + # the application never closes frames itself. + self._release_borrowed() + if self._task.done(): raise StopAsyncIteration @@ -182,4 +262,7 @@ async def __anext__(self) -> VideoFrameEvent: if item is None: raise StopAsyncIteration + if self._zero_copy: + self._borrowed = item.frame + return item diff --git a/tests/rtc/test_video_frame_zero_copy.py b/tests/rtc/test_video_frame_zero_copy.py new file mode 100644 index 00000000..7ecbbd42 --- /dev/null +++ b/tests/rtc/test_video_frame_zero_copy.py @@ -0,0 +1,334 @@ +"""Borrowed video frames: the ones that read the SDK's buffer instead of a copy. + +These exercise `VideoFrame._borrow_owned_info` and the lifetime rules around +it without a live room, by building the `OwnedVideoBuffer` the FFI would send +over a buffer this process owns. +""" + +from __future__ import annotations + +import asyncio +import ctypes +import itertools +from types import SimpleNamespace +from typing import Any, Callable + +import pytest + +from livekit.rtc._proto import video_frame_pb2 as proto_video +from livekit.rtc._utils import RingQueue +from livekit.rtc.video_frame import VideoFrame +from livekit.rtc.video_stream import ( + VideoFrameEvent, + VideoStream, + _close_discarded, + _dropped_frames, +) + +WIDTH = 64 +HEIGHT = 32 +I420_LEN = WIDTH * HEIGHT + ((WIDTH + 1) // 2) * ((HEIGHT + 1) // 2) * 2 + + +class FakeNativeBuffer: + """Stands in for the buffer the FFI hands over, and records its release. + + The real handle is an id into the Rust handle map. Here it is a ctypes + array this process keeps alive, so the pointer stays valid for the whole + test no matter when the frame releases it. Each instance gets its own + handle id, so a test holding several can tell which one was released. + """ + + def __init__(self, handle_id: int, disposed_ids: set[int], fill: int = 0x7F) -> None: + self.handle_id = handle_id + self._disposed_ids = disposed_ids + self.storage = (ctypes.c_uint8 * I420_LEN)(*([fill] * I420_LEN)) + + @property + def address(self) -> int: + return ctypes.addressof(self.storage) + + @property + def disposed(self) -> bool: + return self.handle_id in self._disposed_ids + + def owned_info(self) -> proto_video.OwnedVideoBuffer: + owned = proto_video.OwnedVideoBuffer() + owned.handle.id = self.handle_id + owned.info.type = proto_video.VideoBufferType.I420 + owned.info.width = WIDTH + owned.info.height = HEIGHT + owned.info.data_ptr = self.address + return owned + + +@pytest.fixture +def new_native(monkeypatch: pytest.MonkeyPatch) -> Callable[[], FakeNativeBuffer]: + """Mints fake FFI buffers whose releases are tracked per handle id. + + VideoFrame builds a real FfiHandle; this stops it from calling into the + native library for handle ids that do not exist there. + """ + disposed_ids: set[int] = set() + ids = itertools.count(1000) + + def fake_dispose(self: Any) -> None: + disposed_ids.add(self.handle) + + monkeypatch.setattr("livekit.rtc._ffi_client.FfiHandle.dispose", fake_dispose) + return lambda: FakeNativeBuffer(next(ids), disposed_ids) + + +@pytest.fixture +def native(new_native: Callable[[], FakeNativeBuffer]) -> FakeNativeBuffer: + return new_native() + + +def borrow(native: FakeNativeBuffer) -> VideoFrame: + return VideoFrame._borrow_owned_info(native.owned_info()) + + +def test_a_borrowed_frame_reads_the_native_buffer_without_copying( + native: FakeNativeBuffer, +) -> None: + frame = borrow(native) + + assert frame.is_borrowed + assert frame.width == WIDTH and frame.height == HEIGHT + assert len(frame.data) == I420_LEN + assert bytes(frame.data[:4]) == b"\x7f\x7f\x7f\x7f" + + # The giveaway that nothing was copied: a write through the native buffer + # shows up in the frame. + native.storage[0] = 0x01 + assert frame.data[0] == 0x01 + + frame.close() + + +def test_an_owned_frame_copies_and_releases_the_handle_immediately( + native: FakeNativeBuffer, +) -> None: + frame = VideoFrame._from_owned_info(native.owned_info()) + + assert not frame.is_borrowed + assert native.disposed, "the copying path is done with the buffer right away" + + native.storage[0] = 0x01 + assert frame.data[0] == 0x7F, "the copy is independent of the native buffer" + + +def test_closing_releases_the_handle_and_invalidates_the_data( + native: FakeNativeBuffer, +) -> None: + frame = borrow(native) + assert not native.disposed, "the buffer stays out while the frame is open" + + frame.close() + + assert native.disposed + assert not frame.is_borrowed + with pytest.raises(RuntimeError, match="closed"): + frame.data + + +def test_closing_twice_is_harmless(native: FakeNativeBuffer) -> None: + frame = borrow(native) + frame.close() + frame.close() + + assert native.disposed + + +def test_closing_an_owned_frame_does_nothing(native: FakeNativeBuffer) -> None: + frame = VideoFrame._from_owned_info(native.owned_info()) + + frame.close() + + assert frame.data[0] == 0x7F, "an owned frame stays usable after close()" + + +def test_the_frame_works_as_a_context_manager(native: FakeNativeBuffer) -> None: + with borrow(native) as frame: + assert frame.data[0] == 0x7F + + assert native.disposed + + +def test_close_does_not_pretend_to_guard_outstanding_views( + native: FakeNativeBuffer, +) -> None: + """The documented hazard, pinned so nobody mistakes close() for a guard. + + A view built from `frame.data` re-exports the underlying array rather than + the frame's memoryview, so CPython reports no outstanding export. close() + releases the buffer regardless; this is the contract, not a bug. + """ + frame = borrow(native) + derived = frame.data + + frame.close() + + assert native.disposed, "close() releases the buffer unconditionally" + assert derived.obj is not None, "the caller is left holding a dangling view" + + +def test_ring_queue_reports_evicted_items() -> None: + discarded: list[int] = [] + queue: RingQueue[int] = RingQueue(capacity=2, on_discard=discarded.append) + + queue.put(1) + queue.put(2) + queue.put(3) + + assert discarded == [1], "the oldest item is handed back before it is dropped" + + +def test_ring_queue_get_nowait_drains_then_returns_none() -> None: + queue: RingQueue[int] = RingQueue() + queue.put(1) + queue.put(2) + + assert [queue.get_nowait(), queue.get_nowait(), queue.get_nowait()] == [1, 2, None] + + +def test_plane_info_is_unused_by_the_borrowed_path_but_still_resolves( + native: FakeNativeBuffer, +) -> None: + # get_plane() derives plane offsets from the frame's own buffer address, so + # it has to keep working for a frame backed by native memory. + frame = borrow(native) + + y = frame.get_plane(0) + + assert y is not None + assert len(y) == WIDTH * HEIGHT + frame.close() + + +def test_handle_is_released_when_a_borrowed_frame_is_garbage_collected( + native: FakeNativeBuffer, +) -> None: + borrow(native) # dropped immediately; __del__ is the backstop + + import gc + + gc.collect() + + assert native.disposed + + +# --- VideoStream's side of the borrow ---------------------------------------- +# +# These drive the stream object directly rather than through a room: __init__ +# talks to the FFI, so the tests build an instance without running it and then +# exercise the frame hand-off, which is what bounds a borrowed buffer's life. + + +def make_stream(zero_copy: bool, capacity: int = 0) -> VideoStream: + stream = object.__new__(VideoStream) + stream._zero_copy = zero_copy + stream._borrowed = None + stream._queue = RingQueue(capacity, on_discard=_close_discarded if zero_copy else None) + # __anext__ checks this before reading the queue. + stream._task = asyncio.get_event_loop().create_future() + return stream + + +def frame_event(frame: VideoFrame) -> VideoFrameEvent: + return VideoFrameEvent(frame=frame, timestamp_us=0, rotation=0, metadata=None) + + +@pytest.mark.asyncio +async def test_requesting_the_next_frame_closes_the_previous_borrow( + native: FakeNativeBuffer, new_native: Callable[[], FakeNativeBuffer] +) -> None: + second = new_native() + stream = make_stream(zero_copy=True) + stream._queue.put(frame_event(borrow(native))) + stream._queue.put(frame_event(VideoFrame._borrow_owned_info(second.owned_info()))) + + first_event = await stream.__anext__() + assert not native.disposed, "the frame just handed out is still open" + + await stream.__anext__() + + assert native.disposed, "asking for the next frame ended the previous borrow" + assert first_event.frame._closed + + +@pytest.mark.asyncio +async def test_an_owning_stream_never_closes_frames_behind_the_caller( + native: FakeNativeBuffer, +) -> None: + stream = make_stream(zero_copy=False) + stream._queue.put(frame_event(VideoFrame._from_owned_info(native.owned_info()))) + stream._queue.put(frame_event(VideoFrame._from_owned_info(native.owned_info()))) + + first_event = await stream.__anext__() + await stream.__anext__() + + assert not first_event.frame._closed, "an owned frame stays valid indefinitely" + assert first_event.frame.data[0] == 0x7F + + +@pytest.mark.asyncio +async def test_a_frame_evicted_by_a_full_queue_is_closed( + native: FakeNativeBuffer, new_native: Callable[[], FakeNativeBuffer] +) -> None: + stream = make_stream(zero_copy=True, capacity=1) + evicted = borrow(native) + stream._queue.put(frame_event(evicted)) + + stream._queue.put(frame_event(borrow(new_native()))) + + assert evicted._closed, "the dropped frame released its buffer immediately" + assert native.disposed + + +@pytest.mark.asyncio +async def test_closing_the_stream_releases_the_outstanding_and_queued_frames( + native: FakeNativeBuffer, new_native: Callable[[], FakeNativeBuffer] +) -> None: + queued_buffer = new_native() + stream = make_stream(zero_copy=True) + stream._queue.put(frame_event(borrow(native))) + stream._queue.put(frame_event(VideoFrame._borrow_owned_info(queued_buffer.owned_info()))) + + await stream.__anext__() # hands out the first, leaves the second queued + stream._task.set_result(None) + stream._ffi_handle = _NoopHandle() + await stream.aclose() + + assert native.disposed, "the frame the caller was holding" + assert queued_buffer.disposed, "the frame still sitting in the queue" + + +class _NoopHandle: + def dispose(self) -> None: + pass + + +# --- the dropped-frame counter ---------------------------------------------- + + +def test_dropped_frames_reads_zero_when_the_ffi_predates_the_field() -> None: + # The generated protos only gained `dropped_frames` alongside the native + # counter. Against an older livekit_ffi the field is absent, and zero is + # the honest answer: nothing was reported either way. + frame_received = proto_video.VideoFrameReceived() + + assert _dropped_frames(frame_received) == 0 + + +def test_dropped_frames_reads_the_field_when_the_ffi_provides_it() -> None: + # Stands in for the regenerated message, so this keeps passing before and + # after the rust-sdks bump. + assert _dropped_frames(SimpleNamespace(dropped_frames=7)) == 7 + + +def test_a_frame_event_defaults_to_no_dropped_frames() -> None: + assert ( + frame_event(VideoFrame(1, 1, proto_video.VideoBufferType.RGBA, b"\0\0\0\0")).dropped_frames + == 0 + )