fix(camera): drain ffmpeg's pipes during teardown (#NNNN)

Closing a camera view logged "ffmpeg didn't terminate gracefully,
killing" followed by "ffmpeg did not exit within 2.0s of SIGKILL;
abandoning wait", on every single close. Both waits expired every time,
so teardown took a fixed 4.00s -- and since the firmware allows one
camera connection, that was 4s in which nothing else could use it.

ffmpeg is spawned with stdout and stderr as pipes and the teardown paths
have stopped reading them, so it sits blocked in write() on a full 64 KiB
pipe. SIGTERM cannot be acted on there: the handler only sets a flag that
the main loop polls, and the loop never gets back to the check. SIGKILL
does kill it, but asyncio resolves Process.wait()'s waiter through
_try_finish(), which requires every pipe transport to report
disconnected; paused, unread pipes never reach EOF, so wait() blocks with
returncode already set. A negative-control test shows returncode=-9 at
the instant the abandon fires.

Draining both pipes while stopping the process fixes both halves: 4.00s
becomes ~0.15s. The signal ladder and its bounds stay as backstops, so a
genuinely wedged process still cannot hang a stream, a Stop request or
the janitor.

This corrects _FFMPEG_KILL_TIMEOUT's premise and #2580's conclusion. That
12-hour hang was the unbounded form of this same self-inflicted stall, not
an ffmpeg stuck in uninterruptible I/O -- the process observed doing it
was in state S, which cannot survive a delivered SIGKILL. Bounding the
wait capped the symptom without removing the cause.

backend/app/api/routes/camera.py

@@ -46,12 +46,25 @@ from backend.app.services.camera_profiles import get_camera_profile
 logger = logging.getLogger(__name__)
 router = APIRouter(prefix="/printers", tags=["camera"])
 
-# Upper bound on waiting for a SIGKILLed ffmpeg to be reaped (#2580). A killed
-# ffmpeg stuck in uninterruptible I/O on a dead RTSP socket can take arbitrarily
-# long to exit — an unbounded post-kill wait() parked the fan-out stream
-# coroutine for 12 hours on a P2S, leaving every viewer attached to a stalled
-# broadcaster. Abandoning the wait is safe: cleanup_orphaned_streams' /proc scan
-# reaps any Bambu ffmpeg not attached to an active stream on its next pass.
+# Grace period for a SIGTERMed ffmpeg to shut down before we SIGKILL it. Only
+# reachable when ffmpeg genuinely ignores SIGTERM: _terminate_ffmpeg drains the
+# pipes first, and a drained ffmpeg exits in ~0.15s.
+_FFMPEG_TERM_TIMEOUT = 2.0
+
+# Upper bound on waiting for a SIGKILLed ffmpeg to be reaped (#2580).
+#
+# The original diagnosis — "a killed ffmpeg stuck in uninterruptible I/O on a
+# dead RTSP socket" — was wrong, and this bound was capping a deadlock of our
+# own making rather than waiting out a stuck process. A process that survives
+# SIGKILL would have to be in uninterruptible sleep (state D); the ffmpeg seen
+# doing this was in state S, and its returncode was already set to -9 while
+# wait() was still blocked. The real cause was undrained pipes (see
+# _terminate_ffmpeg), which made this timeout fire on *every* camera close.
+#
+# Kept as a backstop now that the cause is fixed: it should no longer be
+# reachable, and if it ever is, abandoning the wait is still safe because
+# cleanup_orphaned_streams' /proc scan reaps any Bambu ffmpeg not attached to
+# an active stream on its next pass.
 _FFMPEG_KILL_TIMEOUT = 2.0
 
 # Track active ffmpeg processes for cleanup
@@ -241,14 +254,63 @@ async def generate_chamber_mjpeg_stream(
         logger.info("Chamber image stream stopped for %s (stream_id=%s)", ip_address, stream_id)
 
 
+async def _drain_pipe(reader) -> None:
+    """Read a subprocess pipe to EOF and discard, so it can never block.
+
+    Best-effort by design: any read failure means we cannot drain further, and
+    the caller is tearing the process down regardless.
+    """
+    if reader is None:
+        return
+    try:
+        while await reader.read(65536):
+            pass
+    except asyncio.CancelledError:
+        raise
+    except Exception:  # noqa: BLE001 — teardown must not fail on a dying pipe
+        return
+
+
 async def _terminate_ffmpeg(process: asyncio.subprocess.Process, stream_id: str | None = None) -> None:
-    """Terminate an ffmpeg process gracefully, then kill if needed."""
+    """Terminate an ffmpeg process gracefully, then kill if needed.
+
+    Drains stdout/stderr throughout, which is load-bearing rather than hygiene.
+    ffmpeg is spawned with both as pipes, and every caller of this has already
+    stopped reading stdout — so by the time we get here ffmpeg is typically
+    blocked in write() on a full 64 KiB pipe. Two things then go wrong:
+
+    * SIGTERM cannot be acted on. ffmpeg's handler only sets a flag that its
+      main loop polls, and a loop blocked in write() never reaches the check,
+      so the whole grace period is dead time.
+    * SIGKILL does kill it, but wait() cannot observe that. asyncio resolves
+      Process.wait()'s waiter through BaseSubprocessTransport._try_finish(),
+      which requires every pipe transport to report disconnected; paused,
+      unread pipes never reach EOF, so wait() blocks with returncode already
+      set. That is what made the "did not exit within Ns of SIGKILL" error
+      fire on every single camera close, and unbounded it was the 12-hour
+      hang in #2580.
+
+    Draining fixes both: SIGTERM becomes actionable and the exit observable.
+    Measured on an H2D: 4.0s of dead time per close before, ~0.15s after —
+    which matters because the printer allows exactly one camera connection,
+    so every one of those seconds was a connection nobody could use.
+
+    Discarding what we drain is deliberate. The stream loop already reads
+    stderr on its error paths (_read_ffmpeg_stderr), and it does so before
+    calling this, so nothing diagnostic is lost.
+    """
     if process.returncode is not None:
+        _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)),
+    ]
     try:
         process.terminate()
         try:
-            await asyncio.wait_for(process.wait(), timeout=2.0)
+            await asyncio.wait_for(process.wait(), timeout=_FFMPEG_TERM_TIMEOUT)
         except TimeoutError:
             logger.warning("ffmpeg didn't terminate gracefully, killing (stream_id=%s)", stream_id)
             process.kill()
@@ -257,7 +319,8 @@ async def _terminate_ffmpeg(process: asyncio.subprocess.Process, stream_id: str
             except TimeoutError:
                 # Do NOT keep waiting (#2580): the caller is the stream
                 # generator, and blocking here pins the fan-out pump forever.
-                # The orphan janitor reaps the process later.
+                # The orphan janitor reaps the process later. With the pipes
+                # drained this should be unreachable — see _FFMPEG_KILL_TIMEOUT.
                 logger.error(
                     "ffmpeg did not exit within %.1fs of SIGKILL; abandoning wait (stream_id=%s)",
                     _FFMPEG_KILL_TIMEOUT,
@@ -267,7 +330,11 @@ async def _terminate_ffmpeg(process: asyncio.subprocess.Process, stream_id: str
         pass  # Already dead
     except OSError as e:
         logger.warning("Error terminating ffmpeg: %s", e)
-    _spawned_ffmpeg_pids.pop(process.pid, None)
+    finally:
+        for drainer in drainers:
+            drainer.cancel()
+        await asyncio.gather(*drainers, return_exceptions=True)
+        _spawned_ffmpeg_pids.pop(process.pid, None)
 
 
 def _summarize_ffmpeg_stderr(text: str | None) -> str:

backend/tests/unit/test_camera_ffmpeg_termination.py

@@ -1,20 +1,33 @@
-"""Bounded post-kill ffmpeg cleanup (#2580, fix shape from PR #2581 by @ronaldheft).
+"""ffmpeg teardown: draining the pipes, and the bounded waits behind it.
 
-A SIGKILLed ffmpeg stuck in uninterruptible I/O on a dead RTSP socket can take
-arbitrarily long to be reaped. The cleanup paths used to ``await process.wait()``
-unbounded after ``kill()`` — on a P2S RTSP read timeout this parked the fan-out
-stream coroutine for 12 hours, leaving every viewer attached to a stalled
-broadcaster while snapshots/diagnostics (fresh connections) kept working.
+Originally #2580 (fix shape from PR #2581 by @ronaldheft): the cleanup paths
+``await process.wait()``-ed unbounded after ``kill()``, which on a P2S RTSP read
+timeout parked the fan-out stream coroutine for 12 hours, leaving every viewer
+attached to a stalled broadcaster while snapshots (fresh connections) kept
+working. Three places had it, all bounded now:
 
-The same unbounded wait existed in THREE places, all bounded now:
 1. ``_terminate_ffmpeg`` — the stream generator's cleanup (the reported hang).
 2. ``stop_camera`` — hung the very request a user makes to recover.
 3. ``cleanup_orphaned_streams`` — hung the janitor that is the safety net.
+
+That diagnosis — "a SIGKILLed ffmpeg stuck in uninterruptible I/O" — turned out
+to be wrong, and the bound was capping a deadlock of our own making. ffmpeg was
+blocked writing to a stdout pipe nobody was reading, which makes SIGTERM
+unactionable, and ``wait()`` cannot observe an exit while a pipe transport is
+still undrained. So the abandon path fired on every camera close, costing 4s of
+the printer's single camera connection each time. The pipes are drained now; the
+bounds remain as backstops, and the tests for them stay valid.
+
+The draining tests below drive a REAL subprocess, because the failure is in
+asyncio's pipe/transport bookkeeping — a fake process object cannot reproduce
+it and would happily pass against the broken code.
 """
 
 from __future__ import annotations
 
 import asyncio
+import logging
+import sys
 import time
 from contextlib import suppress
 
@@ -24,6 +37,46 @@ from backend.app.api.routes import camera
 
 pytestmark = pytest.mark.asyncio
 
+# Stands in for ffmpeg: floods stdout, and handles SIGTERM the way ffmpeg does
+# — a handler that sets a flag which only the main loop checks, so a process
+# blocked in write() never acts on it until something drains the pipe.
+_FFMPEG_LIKE = """
+import signal, sys
+stop = False
+def _handler(*_a):
+    global stop
+    stop = True
+signal.signal(signal.SIGTERM, _handler)
+sys.stderr.write("x" * 4096)
+sys.stderr.flush()
+while not stop:
+    sys.stdout.buffer.write(b"x" * 65536)
+    sys.stdout.buffer.flush()
+"""
+
+# Same, but SIGTERM is ignored outright — forces the SIGKILL branch.
+_SIGTERM_PROOF = """
+import signal, sys
+signal.signal(signal.SIGTERM, signal.SIG_IGN)
+while True:
+    sys.stdout.buffer.write(b"x" * 65536)
+    sys.stdout.buffer.flush()
+"""
+
+
+async def _spawn(program: str) -> asyncio.subprocess.Process:
+    """Start the stand-in and let it fill its stdout pipe, as the cancel path
+    leaves a real ffmpeg."""
+    process = await asyncio.create_subprocess_exec(
+        sys.executable,
+        "-c",
+        program,
+        stdout=asyncio.subprocess.PIPE,
+        stderr=asyncio.subprocess.PIPE,
+    )
+    await asyncio.sleep(0.4)
+    return process
+
 
 class _FakeServer:
     def close(self) -> None:
@@ -106,6 +159,64 @@ class _FrameProcess:
         return self.returncode
 
 
+# ---------------------------------------------------------------------------
+# 0. _terminate_ffmpeg drains the pipes — against a real subprocess
+# ---------------------------------------------------------------------------
+
+
+async def test_terminate_drains_stdout_so_sigterm_works(caplog):
+    """A process blocked writing to a full pipe still shuts down on SIGTERM.
+
+    Undrained, this took the full grace period plus the SIGKILL bound (4s
+    measured) and ended in the abandon error. Drained, SIGTERM lands.
+    """
+    process = await _spawn(_FFMPEG_LIKE)
+    camera._spawned_ffmpeg_pids[process.pid] = time.time()
+
+    with caplog.at_level(logging.WARNING, logger=camera.logger.name):
+        started = time.monotonic()
+        await asyncio.wait_for(camera._terminate_ffmpeg(process, "test-drain"), timeout=5.0)
+        elapsed = time.monotonic() - started
+
+    assert process.returncode is not None, "wait() must observe the exit"
+    # Comfortably under the 2.0s grace period: proves SIGTERM was acted on
+    # rather than timing out into the kill branch.
+    assert elapsed < 1.5, f"teardown took {elapsed:.2f}s — pipes likely not drained"
+    assert "didn't terminate gracefully" not in caplog.text
+    assert "abandoning wait" not in caplog.text
+    assert process.pid not in camera._spawned_ffmpeg_pids
+
+
+async def test_terminate_observes_kill_of_a_sigterm_proof_process(monkeypatch, caplog):
+    """Even when SIGTERM is genuinely ignored, wait() must see the SIGKILL.
+
+    This is the case the abandon error was invented for. With the pipes drained
+    the exit is observable, so it must not fire.
+    """
+    monkeypatch.setattr(camera, "_FFMPEG_TERM_TIMEOUT", 0.3)
+    process = await _spawn(_SIGTERM_PROOF)
+    camera._spawned_ffmpeg_pids[process.pid] = time.time()
+
+    with caplog.at_level(logging.WARNING, logger=camera.logger.name):
+        await asyncio.wait_for(camera._terminate_ffmpeg(process, "test-kill"), timeout=5.0)
+
+    assert process.returncode == -9, "SIGKILLed exit must be observed, not abandoned"
+    assert "didn't terminate gracefully" in caplog.text  # SIGTERM really was ignored
+    assert "abandoning wait" not in caplog.text
+    assert process.pid not in camera._spawned_ffmpeg_pids
+
+
+async def test_terminate_is_a_noop_for_an_already_dead_process():
+    """The early return must still drop the pid from the tracking dict."""
+    process = await asyncio.create_subprocess_exec(sys.executable, "-c", "pass")
+    await process.wait()
+    camera._spawned_ffmpeg_pids[process.pid] = time.time()
+
+    await asyncio.wait_for(camera._terminate_ffmpeg(process, "test-dead"), timeout=2.0)
+
+    assert process.pid not in camera._spawned_ffmpeg_pids
+
+
 # ---------------------------------------------------------------------------
 # 1. _terminate_ffmpeg — the helper itself is bounded
 # ---------------------------------------------------------------------------

backend/tests/unit/test_camera_usb_stream_cleanup.py

@@ -33,6 +33,10 @@ class _CleanProc:
     def __init__(self, pid: int) -> None:
         self.pid = pid
         self.returncode = None
+        # Real Process objects always expose these (None when not piped), and
+        # _terminate_ffmpeg drains them so a full pipe can't wedge the exit.
+        self.stdout = None
+        self.stderr = None
 
     def terminate(self) -> None:
         self.returncode = 0