fix(camera): drain a streaming ffmpeg's stderr continuously (#2707)

ffmpeg is spawned with stderr=PIPE and it was only read on the error
paths, so for the life of a working stream nobody read that pipe. ffmpeg
writes its banner, the input analysis, then a progress line at a steady
rate; a 64 KiB pipe fills eventually, ffmpeg blocks writing to it, frames
stop, and the stream's own 30s timeout fires -- logged as "RTSP read
timeout" with no hint that we starved it ourselves.

How long that takes is unmeasured and evidently long: one H2D upstream
ran 21m36s without stalling, and an earlier 512 B/s extrapolation of mine
was mostly the one-off startup banner. So this is a bounded resource being
treated as unbounded, not a fault anyone has reported.

_FfmpegStderrTail drains the pipe continuously and keeps a 16 KiB rolling
tail. That tail is what the error paths now report, which is better
material than before: it holds what ffmpeg said as things went wrong,
where the on-demand read returned whatever was printed first -- usually
the banner, which the summariser strips anyway.

Three readers wanted this one pipe, and asyncio rejects concurrent reads
on a StreamReader, so the collector is authoritative: it registers by pid,
_read_ffmpeg_stderr returns its tail when present and otherwise reads the
pipe unchanged, and _terminate_ffmpeg skips its own stderr drain when the
collector owns it (the collector keeps draining through teardown, which is
all wait() needs). The generator starts it after the immediate-failure
check, which reads the pipe directly because the process is already dead,
and releases it after _terminate_ffmpeg.

text() goes through _summarize_ffmpeg_stderr like every other stderr log
here, so the access code ffmpeg echoes in its input URL stays masked.
aclose() awaits the cancelled pump rather than firing and forgetting, so
no pending task survives into loop teardown.

backend/app/api/routes/camera.py

@@ -1,6 +1,7 @@
 """Camera streaming API endpoints for Bambu Lab printers."""
 
 import asyncio
+import contextlib
 import logging
 import os
 import subprocess
@@ -98,6 +99,14 @@ _disconnect_events: dict[str, asyncio.Event] = {}
 # Track last frame time per stream_id (not just per printer_id) for stale detection
 _stream_last_frame_times: dict[str, float] = {}
 
+# How much of a streaming ffmpeg's stderr to retain: enough for the input
+# analysis plus a burst of errors, capped so a long-running stream can't grow it.
+_FFMPEG_STDERR_TAIL_BYTES = 16384
+
+# Live stderr collectors by pid — see _FfmpegStderrTail. Present means "this
+# process's stderr already has a reader; do not open a second one".
+_stderr_tails: dict[int, "_FfmpegStderrTail"] = {}
+
 
 def get_buffered_frame(printer_id: int) -> bytes | None:
     """Get the last buffered frame for a printer from an active stream.
@@ -338,10 +347,13 @@ async def _terminate_ffmpeg(process: asyncio.subprocess.Process, stream_id: str
         _spawned_ffmpeg_pids.pop(process.pid, None)
         return  # Already dead
 
-    drainers = [
-        asyncio.create_task(_drain_pipe(process.stdout)),
-        asyncio.create_task(_drain_pipe(process.stderr)),
-    ]
+    drainers = [asyncio.create_task(_drain_pipe(process.stdout))]
+    # A streaming ffmpeg's stderr already has a reader (_FfmpegStderrTail), and
+    # it keeps draining right through teardown, which is all we need here. Adding
+    # a second reader would race it — asyncio rejects concurrent reads on one
+    # StreamReader — so only drain stderr when nobody else owns it.
+    if process.pid not in _stderr_tails:
+        drainers.append(asyncio.create_task(_drain_pipe(process.stderr)))
     try:
         process.terminate()
         try:
@@ -405,6 +417,82 @@ def _summarize_ffmpeg_stderr(text: str | None) -> str:
     return "\n".join(meaningful[-10:])
 
 
+class _FfmpegStderrTail:
+    """Owns a long-lived ffmpeg's stderr: drains it continuously, keeps the tail.
+
+    Reading stderr only when something has already gone wrong leaves a pipe
+    nobody reads for the whole life of the stream. ffmpeg writes its banner, the
+    input analysis and then a progress line at a steady rate, so a 64 KiB pipe
+    fills eventually and ffmpeg blocks writing to it — at which point it stops
+    producing frames, the stream's own read timeout fires, and the log says
+    "RTSP read timeout" with no hint that we starved it ourselves.
+
+    How long that takes is unmeasured and may be a long time: one H2D upstream
+    ran 21m36s continuously without stalling, so this is a bounded resource
+    being treated as unbounded rather than an observed failure. Draining removes
+    the ceiling either way, and the tail is *better* diagnostic material than
+    the old on-demand read: it holds ffmpeg's most recent output at the moment
+    things went wrong, where reading the buffered pipe returned whatever was
+    printed first (usually the startup banner, which the summariser then strips).
+
+    Registers itself in ``_stderr_tails`` so the two other readers of this pipe
+    can defer to it — asyncio raises if two coroutines read one StreamReader
+    concurrently. See ``_read_ffmpeg_stderr`` and ``_terminate_ffmpeg``.
+    """
+
+    def __init__(self, process: asyncio.subprocess.Process) -> None:
+        self._process = process
+        self._buffer = bytearray()
+        self._task: asyncio.Task | None = None
+        if process.stderr is None:
+            return
+        self._task = asyncio.create_task(self._pump())
+        _stderr_tails[process.pid] = self
+
+    async def _pump(self) -> None:
+        reader = self._process.stderr
+        try:
+            while True:
+                chunk = await reader.read(8192)
+                if not chunk:
+                    return  # EOF — ffmpeg has exited
+                self._buffer.extend(chunk)
+                excess = len(self._buffer) - _FFMPEG_STDERR_TAIL_BYTES
+                if excess > 0:
+                    del self._buffer[:excess]
+        except asyncio.CancelledError:
+            raise
+        except Exception:  # noqa: BLE001 — a broken pipe just ends the tail
+            return
+
+    def text(self) -> str | None:
+        """The retained tail, summarised. None when nothing was captured.
+
+        Goes through _summarize_ffmpeg_stderr like every other stderr log in
+        this module: ffmpeg echoes its input URL, which carries the access code.
+        """
+        if not self._buffer:
+            return None
+        return _summarize_ffmpeg_stderr(self._buffer.decode(errors="replace")) or None
+
+    async def aclose(self) -> None:
+        """Stop draining and release ownership of the pipe. Idempotent.
+
+        Awaits the cancelled pump rather than firing and forgetting, so the task
+        is finished before the caller moves on — an abandoned pending task
+        becomes an "unraisable exception" warning at an arbitrary later point,
+        usually during interpreter or loop teardown.
+        """
+        task, self._task = self._task, None
+        if _stderr_tails.get(self._process.pid) is self:
+            del _stderr_tails[self._process.pid]
+        if task is None:
+            return
+        task.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await task
+
+
 async def _read_ffmpeg_stderr(process: asyncio.subprocess.Process) -> str | None:
     """Read whatever ffmpeg has written to stderr so far (best-effort).
 
@@ -415,8 +503,18 @@ async def _read_ffmpeg_stderr(process: asyncio.subprocess.Process) -> str | None
     banner + stream-analysis lines ffmpeg already printed. Reading in bounded
     chunks returns the buffered output promptly whether or not ffmpeg has
     exited. Returns the content with ffmpeg's boilerplate banner stripped.
+
+    When a _FfmpegStderrTail owns this process's stderr — every streaming
+    ffmpeg — its retained tail is returned instead. Reading the pipe here as
+    well would race that collector, and asyncio refuses two concurrent readers
+    on one StreamReader outright.
     """
-    if not process or not process.stderr:
+    if not process:
+        return None
+    tail = _stderr_tails.get(getattr(process, "pid", None))
+    if tail is not None:
+        return tail.text()
+    if not process.stderr:
         return None
     chunks: list[bytes] = []
     total = 0
@@ -537,6 +635,7 @@ async def generate_rtsp_mjpeg_stream(
     jpeg_end = b"\xff\xd9"
     reconnect_count = 0
     process = None
+    stderr_tail: _FfmpegStderrTail | None = None
     got_any_frames = False
 
     try:
@@ -589,6 +688,14 @@ async def generate_rtsp_mjpeg_stream(
                 reconnect_count += 1
                 continue
 
+            # Take ownership of stderr for the life of this process. Started
+            # only after the immediate-failure check above, which reads the pipe
+            # directly (correct there: the process is already dead, so
+            # read-to-EOF returns at once and cannot be raced by a collector).
+            # Nothing is lost by starting late — the banner ffmpeg printed in the
+            # meantime is still sitting in the pipe.
+            stderr_tail = _FfmpegStderrTail(process)
+
             # Read JPEG frames from ffmpeg stdout
             buffer = b""
             stream_ended = False
@@ -662,6 +769,12 @@ async def generate_rtsp_mjpeg_stream(
 
             # Clean up this ffmpeg process before reconnecting or exiting
             await _terminate_ffmpeg(process, stream_id)
+            # Released after teardown, not before: _terminate_ffmpeg deliberately
+            # leaves stderr to this collector, which has to keep draining while
+            # the process is stopped or wait() can't observe the exit.
+            if stderr_tail is not None:
+                await stderr_tail.aclose()
+                stderr_tail = None
             process = None
 
             if client_gone:
@@ -710,6 +823,10 @@ async def generate_rtsp_mjpeg_stream(
             await _terminate_ffmpeg(process, stream_id)
             logger.info("Camera stream stopped for %s (stream_id=%s)", ip_address, stream_id)
 
+        # Same order as in the loop: terminate first, then release stderr.
+        if stderr_tail is not None:
+            await stderr_tail.aclose()
+
         # Shut down the TLS proxy
         proxy_server.close()
         await proxy_server.wait_closed()

backend/tests/unit/test_camera_stderr_tail.py

@@ -0,0 +1,287 @@
+"""Continuous stderr draining for streaming ffmpeg (_FfmpegStderrTail).
+
+ffmpeg is spawned with stderr=PIPE, and stderr used to be read only when
+something had already gone wrong — so for the life of a stream nobody read that
+pipe. ffmpeg writes its banner, the input analysis and then a progress line at a
+steady rate, so a 64 KiB pipe fills eventually, ffmpeg blocks writing to it,
+frames stop, and the stream's own read timeout fires with nothing in the log
+explaining that we starved it.
+
+How long that takes is unmeasured and evidently long — one H2D upstream ran
+21m36s without stalling — so this is a bounded resource being treated as
+unbounded rather than an observed failure. These tests pin the four properties
+that matter: the pipe is always drained, the retained tail is bounded, the tail
+is what the error paths report, and it goes through the same redaction funnel as
+every other stderr log in this module.
+"""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from backend.app.api.routes import camera
+
+pytestmark = pytest.mark.asyncio
+
+
+class _Reader:
+    """Feeds queued chunks, then blocks like a live-but-quiet ffmpeg."""
+
+    def __init__(self, chunks: list[bytes], then_block: bool = True) -> None:
+        self._chunks = list(chunks)
+        self._then_block = then_block
+        self.reads = 0
+
+    async def read(self, _size: int = -1) -> bytes:
+        self.reads += 1
+        if self._chunks:
+            return self._chunks.pop(0)
+        if self._then_block:
+            await asyncio.Event().wait()  # never returns, never EOFs
+        return b""
+
+
+class _Proc:
+    def __init__(self, reader, pid: int = 88010) -> None:
+        self.pid = pid
+        self.returncode = None
+        self.stdout = None
+        self.stderr = reader
+
+
+@pytest.fixture(autouse=True)
+def _no_leaked_tails():
+    yield
+    # Last-resort teardown only — this fixture is sync, so it cancels without
+    # awaiting. Tests are expected to aclose() their own tails.
+    for tail in list(camera._stderr_tails.values()):
+        if tail._task is not None:
+            tail._task.cancel()
+    camera._stderr_tails.clear()
+
+
+async def _settle() -> None:
+    """Let the pump task run."""
+    for _ in range(5):
+        await asyncio.sleep(0)
+
+
+async def test_it_keeps_draining_a_stream_that_never_closes_stderr():
+    """The whole point: the pipe is read continuously, not on demand."""
+    reader = _Reader([b"first\n", b"second\n"])
+    tail = camera._FfmpegStderrTail(_Proc(reader))
+
+    await _settle()
+
+    assert reader.reads >= 3, "pump stopped reading instead of following the pipe"
+    assert "second" in (tail.text() or "")
+    await tail.aclose()
+
+
+async def test_the_retained_tail_is_bounded():
+    """A long-running stream must not turn the pipe into unbounded memory."""
+    oversized = b"x" * (camera._FFMPEG_STDERR_TAIL_BYTES * 3)
+    tail = camera._FfmpegStderrTail(_Proc(_Reader([oversized])))
+
+    await _settle()
+
+    assert len(tail._buffer) == camera._FFMPEG_STDERR_TAIL_BYTES
+    await tail.aclose()
+
+
+async def test_the_tail_keeps_the_newest_output():
+    """Recent output is what explains a failure; the banner gets stripped anyway."""
+    filler = b"stale-line\n" * 4000
+    tail = camera._FfmpegStderrTail(_Proc(_Reader([filler, b"Connection timed out\n"])))
+
+    await _settle()
+
+    text = tail.text() or ""
+    assert "Connection timed out" in text
+    assert len(tail._buffer) <= camera._FFMPEG_STDERR_TAIL_BYTES
+    await tail.aclose()
+
+
+async def test_read_ffmpeg_stderr_defers_to_the_collector():
+    """Two readers on one StreamReader raise, so the on-demand read must not
+    touch a pipe the collector owns."""
+    reader = _Reader([b"Server returned 401 Unauthorized\n"])
+    process = _Proc(reader)
+    tail = camera._FfmpegStderrTail(process)
+    await _settle()
+    reads_before = reader.reads
+
+    result = await camera._read_ffmpeg_stderr(process)
+
+    assert "401 Unauthorized" in (result or "")
+    assert reader.reads == reads_before, "on-demand read raced the collector"
+    await tail.aclose()
+
+
+async def test_read_ffmpeg_stderr_still_reads_the_pipe_without_a_collector():
+    """An immediately-failed ffmpeg has no collector; that path must still work."""
+    process = _Proc(_Reader([b"Server returned 404 Not Found\n"], then_block=False))
+
+    result = await camera._read_ffmpeg_stderr(process)
+
+    assert "404 Not Found" in (result or "")
+
+
+async def test_the_tail_redacts_the_access_code():
+    """ffmpeg echoes its input URL, which carries the printer's access code.
+
+    This is a new stderr-to-log path, so it gets the same guarantee as the rest:
+    everything goes through _summarize_ffmpeg_stderr.
+    """
+    secret = "12345678"
+    leaky = f"[rtsp @ 0x55] Failed to resolve rtsp://bblp:{secret}@127.0.0.1:8554/streaming/live/1\n"
+    tail = camera._FfmpegStderrTail(_Proc(_Reader([leaky.encode()])))
+
+    await _settle()
+    text = tail.text() or ""
+
+    assert secret not in text, "access code leaked into a log line"
+    # Assert the line SURVIVED with the credential masked, not that it was
+    # dropped — otherwise this passes whenever the summariser happens to filter
+    # the line out, and proves nothing about redaction.
+    assert "Failed to resolve" in text, "line was filtered, so redaction is untested"
+    assert "[REDACTED]" in text
+    await tail.aclose()
+
+
+async def test_close_releases_ownership_and_is_idempotent():
+    process = _Proc(_Reader([b"line\n"]))
+    tail = camera._FfmpegStderrTail(process)
+    await _settle()
+    assert camera._stderr_tails.get(process.pid) is tail
+
+    await tail.aclose()
+    await tail.aclose()  # must not raise
+
+    assert process.pid not in camera._stderr_tails
+
+
+async def test_a_process_without_stderr_is_handled():
+    """Fakes and some spawn paths pass stderr=None; must not register or crash."""
+    process = _Proc(None, pid=88099)
+
+    tail = camera._FfmpegStderrTail(process)
+
+    assert tail.text() is None
+    assert process.pid not in camera._stderr_tails
+    await tail.aclose()
+
+
+async def test_the_stream_generator_owns_then_releases_the_collector(monkeypatch):
+    """Lifecycle inside the real generator: registered while streaming, gone after.
+
+    The other generator tests use fakes with stderr=None, so they never build a
+    collector at all — this is the one that would catch a missing close() or a
+    reader race between the collector and teardown.
+    """
+    printer_id = 8842
+    stream_id = f"{printer_id}-fanout-stderrtail"
+
+    class _FrameThenBlock:
+        def __init__(self) -> None:
+            self._sent = False
+
+        async def read(self, _size: int = -1) -> bytes:
+            if self._sent:
+                await asyncio.Event().wait()  # stay alive, don't trigger reconnect
+            self._sent = True
+            return b"\xff\xd8frame\xff\xd9"
+
+    class _Proc2:
+        def __init__(self) -> None:
+            self.pid = 88042
+            self.returncode = None
+            self.stdout = _FrameThenBlock()
+            self.stderr = _Reader([b"Stream #0:0: Video: h264\n"])
+
+        def terminate(self) -> None:
+            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
+
+    process = _Proc2()
+
+    class _FakeServer:
+        def close(self) -> None:
+            pass
+
+        async def wait_closed(self) -> None:
+            pass
+
+    async def _fake_exec(*_a, **_kw):
+        return process
+
+    async def _fake_proxy(_ip, _port):
+        return 48777, _FakeServer()
+
+    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)
+
+    stream = camera.generate_rtsp_mjpeg_stream(
+        ip_address="192.0.2.44",
+        access_code="c",
+        model="P2S",
+        fps=15,
+        stream_id=stream_id,
+        disconnect_event=asyncio.Event(),
+        printer_id=printer_id,
+    )
+    try:
+        chunk = await asyncio.wait_for(anext(stream), timeout=5.0)
+        assert b"frame" in chunk
+        assert process.pid in camera._stderr_tails, "generator did not take stderr ownership"
+
+        await asyncio.wait_for(stream.aclose(), timeout=5.0)
+
+        assert process.pid not in camera._stderr_tails, "collector outlived its stream"
+    finally:
+        camera._active_streams.pop(stream_id, None)
+        camera._disconnect_events.pop(stream_id, None)
+        camera._stream_last_frame_times.pop(stream_id, None)
+        camera._last_frames.pop(printer_id, None)
+        camera._last_frame_times.pop(printer_id, None)
+        camera._stream_start_times.pop(printer_id, None)
+        camera._spawned_ffmpeg_pids.pop(process.pid, None)
+
+
+async def test_terminate_skips_stderr_while_a_collector_owns_it():
+    """_terminate_ffmpeg must not add a second reader to an owned pipe."""
+    reader = _Reader([b"tearing down\n"])
+
+    class _Killable(_Proc):
+        def terminate(self):
+            self.returncode = 0
+
+        def kill(self):
+            self.returncode = -9
+
+        async def wait(self):
+            if self.returncode is None:
+                self.returncode = 0
+            return self.returncode
+
+    process = _Killable(reader, pid=88020)
+    tail = camera._FfmpegStderrTail(process)
+    await _settle()
+    reads_before = reader.reads
+
+    await asyncio.wait_for(camera._terminate_ffmpeg(process, "88020-fanout-abcd"), timeout=2.0)
+
+    # The collector, not _terminate_ffmpeg, is the only reader that advanced.
+    assert reader.reads >= reads_before
+    assert tail.text() is not None
+    await tail.aclose()