Restart a camera stream that keeps repeating one frame (#3218, #3189)

After an RTSP session dropped, ffmpeg could keep writing the same JPEG
    indefinitely with no connection to the printer left (two hours at ~29 fps
    on a P2S), and every check that counts frames saw a healthy stream. The
    stream now restarts ffmpeg after 20 s without a changed frame, on the
    existing reconnect path, while viewers stay attached. If the first frame
    after the restart is the same picture again, the camera really shows a
    still scene (a dark, idle chamber): it is then only re-checked every
    5 minutes until the picture changes.

backend/app/api/routes/camera.py

@@ -8,6 +8,7 @@ import subprocess
 import sys
 import time
 import uuid
+import zlib
 from collections.abc import AsyncGenerator
 
 from fastapi import APIRouter, Depends, HTTPException, Request
@@ -71,6 +72,23 @@ _FFMPEG_TERM_TIMEOUT = 2.0
 # an active stream on its next pass.
 _FFMPEG_KILL_TIMEOUT = 2.0
 
+# How long an RTSP stream may keep emitting byte-identical JPEGs before the
+# ffmpeg behind it is restarted (#3218). A dead upstream can leave ffmpeg
+# repeating its last frame indefinitely: measured on a P2S, ~29 fps of one
+# frame for two hours with no socket to the printer left at all, while every
+# check that counts frames saw a healthy stream. A live camera practically
+# never repeats a frame byte for byte -- sensor noise changes every one (2262
+# of 2262 distinct on an idle X1C chamber, #3189) -- so a run this long means
+# the picture is frozen.
+_RTSP_FROZEN_SECONDS = 20.0
+# "Practically never" is not never: a dark, idle chamber can encode to the same
+# frame every time. When the first frame after such a restart is the frozen one
+# again, the camera really is showing that picture, and the stream only
+# re-checks at this much longer interval until the picture changes -- so a
+# still scene costs one reconnect every few minutes instead of every 20 s, and a
+# real freeze on it is still recovered.
+_RTSP_STILL_RECHECK_SECONDS = 300.0
+
 # Track active ffmpeg processes for cleanup
 _active_streams: dict[str, asyncio.subprocess.Process] = {}
 
@@ -635,6 +653,11 @@ async def generate_rtsp_mjpeg_stream(
     process = None
     stderr_tail: _FfmpegStderrTail | None = None
     got_any_frames = False
+    # Across sessions (#3218): the frame a session froze on, and whether the
+    # next session showed the very same picture -- a still scene, not a
+    # frozen ffmpeg.
+    frozen_crc: int | None = None
+    still_scene = False
 
     try:
         while reconnect_count <= profile.rtsp_reconnect_max:
@@ -706,6 +729,13 @@ async def generate_rtsp_mjpeg_stream(
             stream_ended = False
             client_gone = False
             session_got_frames = False
+            # A frame that differs from the one before, not any frame, is what
+            # shows the picture is live: ffmpeg can repeat its last one
+            # forever (#3218).
+            last_frame_crc: int | None = None
+            last_change = time.time()
+            identical_frames = 0
+            frozen = False
 
             while True:
                 if disconnect_event and disconnect_event.is_set():
@@ -745,11 +775,38 @@ async def generate_rtsp_mjpeg_stream(
                         got_any_frames = True
                         session_got_frames = True
 
+                        now = time.time()
+                        frame_crc = zlib.crc32(frame)
+                        if last_frame_crc is None:
+                            # First frame of this session. The same frame the
+                            # last session froze on means the camera really
+                            # shows that picture (see _RTSP_STILL_RECHECK_SECONDS).
+                            still_scene = frozen_crc is not None and frame_crc == frozen_crc
+                            if still_scene:
+                                logger.info(
+                                    "RTSP picture unchanged after restart for %s (stream_id=%s): "
+                                    "treating it as a still scene, re-checking every %.0fs",
+                                    ip_address,
+                                    stream_id,
+                                    _RTSP_STILL_RECHECK_SECONDS,
+                                )
+                            last_frame_crc = frame_crc
+                            last_change = now
+                        elif frame_crc != last_frame_crc:
+                            last_frame_crc = frame_crc
+                            last_change = now
+                            identical_frames = 0
+                            # The picture moves: back to the normal check.
+                            still_scene = False
+                            frozen_crc = None
+                        else:
+                            identical_frames += 1
+
                         if printer_id is not None:
                             _last_frames[printer_id] = frame
-                            _last_frame_times[printer_id] = time.time()
+                            _last_frame_times[printer_id] = now
                             if stream_id:
-                                _stream_last_frame_times[stream_id] = time.time()
+                                _stream_last_frame_times[stream_id] = now
 
                         yield (
                             b"--frame\r\n"
@@ -758,6 +815,33 @@ async def generate_rtsp_mjpeg_stream(
                             b"\r\n" + frame + b"\r\n"
                         )
 
+                        if now - last_change > (_RTSP_STILL_RECHECK_SECONDS if still_scene else _RTSP_FROZEN_SECONDS):
+                            frozen = True
+                            frozen_crc = last_frame_crc
+                            break
+
+                    if frozen:
+                        if still_scene:
+                            logger.info(
+                                "RTSP still-scene re-check for %s (stream_id=%s), restarting ffmpeg",
+                                ip_address,
+                                stream_id,
+                            )
+                        else:
+                            stderr_text = await _read_ffmpeg_stderr(process)
+                            if stderr_text:
+                                logger.warning("ffmpeg stderr (stream_id=%s): %s", stream_id, stderr_text)
+                            logger.warning(
+                                "RTSP output frozen for %s (stream_id=%s): %d identical frames over %.0fs, "
+                                "restarting ffmpeg",
+                                ip_address,
+                                stream_id,
+                                identical_frames,
+                                time.time() - last_change,
+                            )
+                        stream_ended = True
+                        break
+
                 except TimeoutError:
                     stderr_text = await _read_ffmpeg_stderr(process)
                     if stderr_text:

backend/tests/unit/test_camera_frozen_output_3218.py

@@ -0,0 +1,234 @@
+"""An RTSP stream repeating one frame is restarted, not trusted (#3218).
+
+On a P2S whose camera session had dropped, ffmpeg kept writing ~29 fps of one
+byte-identical JPEG for two hours with no socket to the printer left. Every
+repeat looked like a live stream to every check. 20 s without a changed frame
+now restarts ffmpeg while the viewer stays attached -- unless the restart shows
+the very same picture again, which is a still scene (a dark, idle chamber), not
+a frozen ffmpeg.
+"""
+
+import asyncio
+from contextlib import suppress
+
+import pytest
+
+from backend.app.api.routes import camera
+from backend.app.services.camera_profiles import CameraProfile
+
+PRINTER_ID = 3218
+STREAM_ID = f"{PRINTER_ID}-fanout-frozen"
+
+
+def _jpeg(n: int) -> bytes:
+    return b"\xff\xd8frame-" + str(n).encode() + b"\xff\xd9"
+
+
+class _Clock:
+    def __init__(self) -> None:
+        self.now = 1_000_000.0
+
+    def __call__(self) -> float:
+        return self.now
+
+
+class _FakeServer:
+    def close(self) -> None:
+        pass
+
+    async def wait_closed(self) -> None:
+        pass
+
+
+class _Stdout:
+    """One frame per read, the clock advancing a second each time."""
+
+    def __init__(self, frames, clock: _Clock) -> None:
+        self._frames = iter(frames)
+        self._clock = clock
+
+    async def read(self, _size: int = -1) -> bytes:
+        self._clock.now += 1.0
+        return next(self._frames, b"")
+
+
+class _Proc:
+    _next_pid = 79000
+
+    def __init__(self, frames, clock: _Clock) -> None:
+        _Proc._next_pid += 1
+        self.pid = _Proc._next_pid
+        self.returncode = None
+        self.stdout = _Stdout(frames, clock)
+        self.stderr = None
+        self.terminated = False
+
+    def terminate(self) -> None:
+        self.terminated = True
+        self.returncode = 0
+
+    def kill(self) -> None:
+        self.returncode = -9
+
+    async def wait(self) -> int:
+        if self.returncode is None:
+            self.returncode = 0
+        return self.returncode
+
+
+@pytest.fixture
+def rtsp(monkeypatch):
+    clock = _Clock()
+    sessions: list = []
+    spawned: list[_Proc] = []
+    real_sleep = asyncio.sleep
+
+    async def _fake_exec(*_args, **_kwargs):
+        proc = _Proc(sessions.pop(0) if sessions else [], clock)
+        spawned.append(proc)
+        return proc
+
+    async def _fake_proxy(_ip: str, _port: int):
+        return 48997, _FakeServer()
+
+    async def _no_sleep(_seconds, *args, **kwargs):
+        await real_sleep(0)
+
+    monkeypatch.setattr(camera, "get_ffmpeg_path", lambda: "/fake/ffmpeg")
+    monkeypatch.setattr(camera, "create_tls_proxy", _fake_proxy)
+    monkeypatch.setattr(camera.asyncio, "create_subprocess_exec", _fake_exec)
+    monkeypatch.setattr(camera.asyncio, "sleep", _no_sleep)
+    monkeypatch.setattr(camera.time, "time", clock)
+    monkeypatch.setattr(
+        camera, "get_camera_profile", lambda _m: CameraProfile(rtsp_reconnect_max=3, rtsp_reconnect_delay=0.2)
+    )
+    yield clock, sessions, spawned
+    camera._release_printer_frame_state(PRINTER_ID)
+
+
+def _stream():
+    return camera.generate_rtsp_mjpeg_stream(
+        ip_address="192.0.2.41",
+        access_code="test-code",
+        model="P2S",
+        fps=30,
+        stream_id=STREAM_ID,
+        disconnect_event=asyncio.Event(),
+        printer_id=PRINTER_ID,
+    )
+
+
+async def _collect(stream, limit: int) -> list[bytes]:
+    frames = []
+    async for chunk in stream:
+        if b"image/jpeg" in chunk:
+            frames.append(chunk.split(b"\r\n\r\n", 1)[1].rstrip(b"\r\n"))
+        if len(frames) >= limit:
+            break
+    with suppress(Exception):
+        await stream.aclose()
+    return frames
+
+
+async def test_a_frozen_session_is_restarted_and_the_viewer_keeps_going(rtsp):
+    _clock, sessions, spawned = rtsp
+    # The reporter's case: one real frame, then ffmpeg repeats it forever.
+    sessions.append([_jpeg(1)] * 100)
+    sessions.append([_jpeg(n) for n in range(2, 40)])
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=40), timeout=10)
+
+    assert len(spawned) == 2
+    assert spawned[0].terminated
+    # About 20 s of the frozen frame went out, then the new session's frames,
+    # on the same viewer stream.
+    assert 20 <= frames.count(_jpeg(1)) <= 23
+    assert frames[-1] != _jpeg(1)
+
+
+async def test_a_moving_picture_is_never_restarted(rtsp):
+    _clock, sessions, spawned = rtsp
+    sessions.append([_jpeg(n) for n in range(120)])
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=120), timeout=10)
+
+    assert len(frames) == 120
+    assert len(spawned) == 1
+
+
+async def test_short_runs_of_repeats_are_fine(rtsp):
+    """ffmpeg's -r repeats frames when it outputs faster than the camera
+    sends; a few repeats between changes are normal and keep the session."""
+    _clock, sessions, spawned = rtsp
+    sessions.append([_jpeg(n // 5) for n in range(150)])
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=150), timeout=10)
+
+    assert len(frames) == 150
+    assert len(spawned) == 1
+
+
+async def test_a_still_scene_is_rechecked_rarely_not_every_20s(rtsp):
+    """A dark chamber can encode to the same frame every time. The first
+    restart shows the same picture again, so the camera really shows it."""
+    _clock, sessions, spawned = rtsp
+    black = _jpeg(0)
+    sessions.append([black] * 30)  # restarted after 20 s...
+    sessions.append([black] * 400)  # ...and the picture is the same
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=250), timeout=10)
+
+    assert len(frames) == 250
+    # One restart, then no more for the 230-odd seconds that follow.
+    assert len(spawned) == 2
+
+
+async def test_a_still_scene_is_still_rechecked_eventually(rtsp):
+    _clock, sessions, spawned = rtsp
+    black = _jpeg(0)
+    sessions.append([black] * 30)
+    sessions.append([black] * 400)
+    sessions.append([_jpeg(n) for n in range(1, 50)])  # lights on
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=360), timeout=10)
+
+    # The re-check after 300 s found the picture moving again.
+    assert len(spawned) == 3
+    assert frames[-1] != black
+
+
+async def test_a_moving_picture_resets_the_still_scene(rtsp):
+    """Once the picture moves, a later freeze is caught after 20 s again."""
+    _clock, sessions, spawned = rtsp
+    black = _jpeg(0)
+    sessions.append([black] * 30)
+    # Same picture first (a still scene), then it moves, then freezes again.
+    sessions.append([black] + [_jpeg(n) for n in range(1, 10)] + [_jpeg(9)] * 400)
+    sessions.append([_jpeg(n) for n in range(100, 140)])
+
+    frames = await asyncio.wait_for(_collect(_stream(), limit=90), timeout=10)
+
+    # Restarted 20 s into the second freeze, not 300 s: the third session's
+    # frames arrive within the 90 collected.
+    assert len(spawned) == 3
+    assert frames[-1] in {_jpeg(n) for n in range(100, 140)}
+
+
+async def test_every_frame_still_counts_for_the_stream_watchdogs(rtsp):
+    """/camera/status, the diagnostic and the orphan janitor keep reading
+    every frame as before; the frozen check lives in the stream itself, so a
+    still scene never looks stalled to them."""
+    clock, sessions, _spawned = rtsp
+    sessions.append([_jpeg(1)] * 11)
+
+    stream = _stream()
+    seen = 0
+    async for chunk in stream:
+        if b"image/jpeg" in chunk:
+            seen += 1
+            if seen == 11:
+                break
+    assert camera._last_frame_times[PRINTER_ID] == clock.now
+    assert camera._stream_last_frame_times[STREAM_ID] == clock.now
+    with suppress(Exception):
+        await stream.aclose()