From 07d1f649fd935aa4101c0e55777bcd7428c5ffc0 Mon Sep 17 00:00:00 2001 From: styu12 Date: Mon, 13 Apr 2026 18:38:16 -0700 Subject: [PATCH] fix(adapters): tolerate EAGAIN/EINTR in UVC capture loop MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit AVFoundation (macOS) and V4L2 (Linux) raise BlockingIOError or OSError(EAGAIN) from the decode iterator during camera warmup and occasionally between frames. The previous for-in-decode loop treated every exception as fatal, killing the capture thread and surfacing as 'capture loop ended: BlockingIOError(35, ...)' in the viewer health panel within a second of connect() — despite the camera being fine. Drive the iterator manually via next() so EAGAIN can be caught per call. Catch all OSError variants and dispatch on errno: - 4 (EINTR), 11 (Linux EAGAIN), 35 (macOS EAGAIN), None: sleep 1ms and retry. These are 'not ready yet' signals, never fatal. - Other errnos (EIO, ENODEV, etc.): fall through to the fatal health-event path. Non-OSError exceptions keep the original fatal behavior. Adds two tests: - test_blocking_io_error_does_not_kill_loop — inject EAGAIN 3x, then real frames; verify frames arrive and no health event emitted. - test_fatal_os_error_still_ends_loop — inject ENODEV; verify the loop exits and emits a health event. Co-Authored-By: Claude Opus 4.6 (1M context) --- src/syncfield/adapters/uvc_webcam.py | 66 +++++++++++----- tests/unit/adapters/test_uvc_webcam.py | 101 +++++++++++++++++++++++++ 2 files changed, 149 insertions(+), 18 deletions(-) diff --git a/src/syncfield/adapters/uvc_webcam.py b/src/syncfield/adapters/uvc_webcam.py index 06ea2dd..bf7dd78 100644 --- a/src/syncfield/adapters/uvc_webcam.py +++ b/src/syncfield/adapters/uvc_webcam.py @@ -254,20 +254,29 @@ def _capture_loop(self) -> None: The loop exits when ``_stop_event`` fires or the input container is exhausted (device disconnect). + + Transient OS errors (EAGAIN / EINTR) raised by AVFoundation or + V4L2 during camera warmup are NOT fatal — the demuxer is just + saying "no frame ready yet, try again." We sleep briefly and + keep polling. Only genuinely fatal errors surface as a health + event and terminate the thread. """ assert self._input is not None - try: - for frame in self._input.decode(video=0): + # Errno values treated as transient "retry soon": + # macOS EAGAIN=35, Linux EAGAIN=11, EINTR=4. ``None`` is also + # treated as transient because PyAV sometimes omits errno on + # "not ready" conditions. + _TRANSIENT_ERRNOS = {4, 11, 35} + + frame_iter = iter(self._input.decode(video=0)) + while not self._stop_event.is_set(): + try: + frame = next(frame_iter) capture_ns = time.monotonic_ns() - # Jitter collection runs only during the recording window so preview - # intervals don't pollute the report and so the capture thread never - # mutates the list while stop_recording() is reading it. if self._recording: if self._prev_capture_ns is not None: self._intervals_ns.append(capture_ns - self._prev_capture_ns) self._prev_capture_ns = capture_ns - if self._stop_event.is_set(): - break frame_bgr = frame.to_ndarray(format="bgr24") @@ -288,18 +297,39 @@ def _capture_loop(self) -> None: capture_ns=capture_ns, ) ) - except Exception as exc: # noqa: BLE001 - PyAV surfaces diverse errors here - # Device disconnect or decode error — emit a health event so the - # orchestrator sees a first-class signal, then exit the thread. - self._emit_health( - HealthEvent( - stream_id=self.id, - kind=HealthEventKind.ERROR, - at_ns=time.monotonic_ns(), - detail=f"capture loop ended: {exc!r}", + except StopIteration: + # Device exhausted (explicit end-of-stream). + break + except OSError as exc: + # AVFoundation / V4L2 can surface "not ready" as: + # - av.error.BlockingIOError (subclass of OSError) + # - bare OSError(35) on macOS, OSError(11) on Linux + # - FFmpegError with errno=None + # All are transient and must not kill the capture loop. + if exc.errno in _TRANSIENT_ERRNOS or exc.errno is None: + time.sleep(0.001) + continue + # Real OSError (EIO, ENODEV, etc.) — treat as fatal. + self._emit_health( + HealthEvent( + stream_id=self.id, + kind=HealthEventKind.ERROR, + at_ns=time.monotonic_ns(), + detail=f"capture loop ended: {exc!r}", + ) ) - ) - return + return + except Exception as exc: # noqa: BLE001 - PyAV surfaces diverse errors here + # Genuine non-OS error — emit a health event and exit. + self._emit_health( + HealthEvent( + stream_id=self.id, + kind=HealthEventKind.ERROR, + at_ns=time.monotonic_ns(), + detail=f"capture loop ended: {exc!r}", + ) + ) + return # ------------------------------------------------------------------ # Live preview diff --git a/tests/unit/adapters/test_uvc_webcam.py b/tests/unit/adapters/test_uvc_webcam.py index 494eda7..de83f7a 100644 --- a/tests/unit/adapters/test_uvc_webcam.py +++ b/tests/unit/adapters/test_uvc_webcam.py @@ -185,3 +185,104 @@ def test_jitter_reported_when_enough_frames( # Sanity: jitter should be on the order of pace_seconds (1ms = 1_000_000 ns) # — allow generous bounds for CI load. assert 0 < report.jitter_p95_ns < 100_000_000 # < 100ms + + +class TestDecoderResilience: + """The capture loop must tolerate transient decoder errors. + + AVFoundation on macOS raises ``BlockingIOError`` (EAGAIN, errno 35) + during camera warmup and occasionally between frames. Linux V4L2 + surfaces the same under EAGAIN=11. Interrupted syscalls (EINTR=4) + fall in the same bucket. None should kill the capture thread. + """ + + def test_blocking_io_error_does_not_kill_loop( + self, mock_av_generous, tmp_path + ): + """Inject EAGAIN into the decode iterator; loop must keep going.""" + import numpy as np + from unittest.mock import MagicMock + + from syncfield.adapters.uvc_webcam import UVCWebcamStream + + # A generator that ``raise``s without ever yielding becomes + # dead after the first next(), so we use an iterator class + # that keeps state across next() calls: first 3 calls raise + # EAGAIN, subsequent 50 yield real BGR frames. + class FlakyIter: + def __init__(self) -> None: + self._eagain_left = 3 + self._i = 0 + + def __iter__(self): + return self + + def __next__(self): + if self._eagain_left > 0: + self._eagain_left -= 1 + raise BlockingIOError( + 35, "Resource temporarily unavailable", "0" + ) + if self._i >= 50: + raise StopIteration + time.sleep(0.001) + frame = MagicMock(name=f"Frame-{self._i}") + frame.to_ndarray = MagicMock( + return_value=np.full( + (48, 64, 3), self._i % 256, dtype=np.uint8 + ) + ) + self._i += 1 + return frame + + mock_av_generous.input_container.decode = MagicMock( + return_value=FlakyIter() + ) + + stream = UVCWebcamStream( + "cam", device_index=0, output_dir=tmp_path, fps=30.0 + ) + stream.prepare() + stream.connect() + stream.start_recording(_clock()) + time.sleep(0.1) + report = stream.stop_recording() + stream.disconnect() + + assert report.frame_count >= 1 + assert not any( + "BlockingIOError" in (h.detail or "") + for h in report.health_events + ) + + def test_fatal_os_error_still_ends_loop( + self, mock_av_generous, tmp_path + ): + """Non-transient OSError (e.g. ENODEV=19) still emits + exits.""" + from unittest.mock import MagicMock + + from syncfield.adapters.uvc_webcam import UVCWebcamStream + + class FatalIter: + def __iter__(self): + return self + + def __next__(self): + raise OSError(19, "No such device", "99") + + mock_av_generous.input_container.decode = MagicMock( + return_value=FatalIter() + ) + + stream = UVCWebcamStream( + "cam", device_index=99, output_dir=tmp_path, fps=30.0 + ) + stream.prepare() + stream.connect() + time.sleep(0.05) + stream.disconnect() + + collected = stream._collected_health # noqa: SLF001 + assert any( + "No such device" in (h.detail or "") for h in collected + ), f"expected fatal OSError in health events, got {collected!r}"