Fix P1/A1 camera black screen from fan-out churn on single-connection cams (#2521)

Chamber-image printers came up black on load and only recovered ~20 min
later. Two causes: (1) late fan-out subscribers got an empty queue and
waited for the next frame, so the browser never fired onLoad and the
stall-detector reconnect-looped, churning short-lived viewers; (2) the
churn reopened the port-6000 socket before the old one closed, so the
printer fed an orphaned socket until its TCP keepalive reaped it.

Prime late subscribers with the last pumped frame; make a replacement
broadcaster's pump wait for the predecessor's socket to close before
dialing (bounded 10s); require two consecutive stalled reads before the
frontend reconnects.

backend/app/services/camera_fanout.py

@@ -26,6 +26,11 @@ logger = logging.getLogger(__name__)
 # on some firmwares and is the very reconnect cost we are trying to avoid).
 _GRACE_SECONDS = 5.0
 
+# Upper bound on how long a new broadcaster waits for a displaced one to finish
+# tearing down before proceeding anyway (#2521). Teardown is normally sub-second
+# (cancel pump + close socket); the cap only guards a wedged upstream close.
+_TEARDOWN_WAIT_SECONDS = 10.0
+
 # Per-subscriber queue depth. Small on purpose: if a viewer can't keep up
 # with the printer's frame rate we drop frames for that viewer rather than
 # blocking the broadcaster. Live video — old frames have no value.
@@ -41,7 +46,7 @@ UpstreamFactory = Callable[[asyncio.Event], AsyncGenerator[bytes, None]]
 class MjpegBroadcaster:
     """Single upstream MJPEG stream, fanned out to N subscribers."""
 
-    def __init__(self, key: str, factory: UpstreamFactory) -> None:
+    def __init__(self, key: str, factory: UpstreamFactory, predecessor: MjpegBroadcaster | None = None) -> None:
         self._key = key
         self._factory = factory
         self._subscribers: list[asyncio.Queue[bytes]] = []
@@ -52,6 +57,22 @@ class MjpegBroadcaster:
         # stop reconnecting when the last subscriber leaves.
         self._upstream_disconnect = asyncio.Event()
         self._stopped = False
+        # Most recent chunk pumped to subscribers. New (late) subscribers are
+        # primed with it so the browser renders a frame immediately instead of
+        # waiting for the next upstream frame — critical on slow chamber-image
+        # cams where the wait looked like a permanent black screen (#2521).
+        self._last_chunk: bytes | None = None
+        # Set once teardown is fully complete (pump cancelled AND the upstream
+        # socket closed). A successor broadcaster waits on this before dialing
+        # so a single-connection printer never sees two sockets at once — the
+        # overlap stranded frames on an orphaned socket for the ~20 min it took
+        # the printer's TCP keepalive to reap it (#2521).
+        self._teardown_complete = asyncio.Event()
+        # The stopped broadcaster this one replaces, if any. The pump waits for
+        # its socket to close before opening ours. Guarding at the pump (not at
+        # get_or_create) keeps it correct when concurrent viewers race to
+        # replace the same stopped broadcaster — only the single pump dials.
+        self._predecessor = predecessor
 
     @property
     def key(self) -> str:
@@ -79,6 +100,15 @@ class MjpegBroadcaster:
             queue: asyncio.Queue[bytes] = asyncio.Queue(maxsize=_SUBSCRIBER_QUEUE_SIZE)
             self._subscribers.append(queue)
 
+            # Prime a late joiner with the last frame so it renders instantly
+            # (#2521). The very first subscriber has nothing to prime yet — it
+            # starts the pump below.
+            if self._last_chunk is not None:
+                try:
+                    queue.put_nowait(self._last_chunk)
+                except asyncio.QueueFull:  # pragma: no cover — fresh queue
+                    pass
+
             if self._pump_task is None or self._pump_task.done():
                 # Reset the disconnect signal in case a previous pump set it.
                 self._upstream_disconnect = asyncio.Event()
@@ -105,6 +135,18 @@ class MjpegBroadcaster:
         """Tear down immediately, kick all subscribers. Idempotent."""
         pump_task = await self._mark_stopped_locked(notify_subscribers=True)
         await self._await_pump_cancellation(pump_task)
+        # Upstream socket is now closed (pump's finally ran) — release anyone
+        # waiting to open a replacement broadcaster (#2521).
+        self._teardown_complete.set()
+
+    async def wait_until_torn_down(self) -> None:
+        """Block until this broadcaster's upstream socket has fully closed.
+
+        Only meaningful for a stopped broadcaster; on a live one this never
+        returns. get_or_create_broadcaster gates a replacement on it so the
+        old and new upstream sockets never overlap (#2521).
+        """
+        await self._teardown_complete.wait()
 
     async def _grace_then_stop(self) -> None:
         try:
@@ -123,6 +165,8 @@ class MjpegBroadcaster:
             self._grace_task = None
             self._stopped = True
         await self._await_pump_cancellation(pump_task)
+        # Upstream socket is now closed — release any pending replacement (#2521).
+        self._teardown_complete.set()
 
     async def _mark_stopped_locked(self, *, notify_subscribers: bool) -> asyncio.Task | None:
         """Mark the broadcaster stopped and detach the pump task.
@@ -165,10 +209,23 @@ class MjpegBroadcaster:
     async def _pump(self) -> None:
         """Drive the upstream generator and broadcast each chunk."""
         try:
+            # Don't dial the printer until the broadcaster we're replacing has
+            # closed its socket (#2521). Bounded so a wedged teardown degrades
+            # to the old overlap behaviour rather than never producing a frame.
+            predecessor = self._predecessor
+            self._predecessor = None
+            if predecessor is not None:
+                try:
+                    await asyncio.wait_for(predecessor.wait_until_torn_down(), timeout=_TEARDOWN_WAIT_SECONDS)
+                except asyncio.TimeoutError:
+                    logger.warning("Prior broadcaster %r didn't tear down in time; dialing anyway", self._key)
             async for chunk in self._factory(self._upstream_disconnect):
                 # Snapshot subscribers under lock so we don't iterate a list
                 # mutated by subscribe()/unsubscribe() while we are putting.
+                # Remember the frame under the same lock so subscribe() can
+                # prime a late joiner with a consistent last-chunk value (#2521).
                 async with self._lock:
+                    self._last_chunk = chunk
                     targets = list(self._subscribers)
                 for queue in targets:
                     try:
@@ -203,12 +260,20 @@ async def get_or_create_broadcaster(key: str, factory: UpstreamFactory) -> Mjpeg
 
     A broadcaster that has been stopped (force shutdown or grace timeout) is
     replaced with a fresh instance — the caller will subscribe to the new one.
+
+    When replacing a stopped broadcaster, the fresh instance is handed it as a
+    predecessor: its pump waits for the old socket to close before dialing, so
+    a single-connection cam (chamber-image port 6000) never sees two sockets at
+    once. Otherwise the printer keeps feeding the orphaned socket and starves
+    the new one until its TCP keepalive reaps it, ~20 min later (#2521).
     """
     async with _registry_lock:
         existing = _broadcasters.get(key)
         if existing is not None and not existing.stopped:
             return existing
-        new_bc = MjpegBroadcaster(key, factory)
+        # `existing` (if any) is stopped/tearing down — chain the new pump
+        # behind its socket close.
+        new_bc = MjpegBroadcaster(key, factory, predecessor=existing)
         _broadcasters[key] = new_bc
         return new_bc
 

backend/tests/unit/services/test_camera_fanout.py

@@ -111,6 +111,50 @@ async def test_multiple_subscribers_share_single_upstream():
     await bc.force_shutdown()
 
 
+# ---------------------------------------------------------------------------
+# Late subscribers are primed with the last frame (#2521)
+# ---------------------------------------------------------------------------
+
+
+async def test_late_subscriber_primed_with_last_frame():
+    """A viewer that joins after the stream is running must receive the most
+    recent frame immediately, not wait for the next upstream frame. On slow
+    chamber-image cams that wait looked like a permanent black screen (#2521).
+    """
+
+    async def factory(disconnect: asyncio.Event) -> AsyncGenerator[bytes, None]:
+        yield b"first"
+        await disconnect.wait()  # then hold the stream open, no further frames
+
+    bc = MjpegBroadcaster("p1", factory)
+    q1 = await bc.subscribe()
+    # First subscriber consumes the frame; this also guarantees the pump has
+    # recorded it as the last chunk.
+    assert await asyncio.wait_for(q1.get(), timeout=1.0) == b"first"
+
+    # Late joiner is handed that frame at once, even though no new frame is coming.
+    q2 = await bc.subscribe()
+    assert await asyncio.wait_for(q2.get(), timeout=0.2) == b"first"
+
+    await bc.force_shutdown()
+
+
+async def test_first_subscriber_not_primed():
+    """The very first subscriber has no prior frame to be primed with — its
+    queue starts empty and it triggers the upstream connect.
+    """
+
+    async def factory(disconnect: asyncio.Event) -> AsyncGenerator[bytes, None]:
+        await disconnect.wait()  # never produces a frame
+        yield b"never"  # pragma: no cover
+
+    bc = MjpegBroadcaster("p1", factory)
+    q1 = await bc.subscribe()
+    await asyncio.sleep(0)  # let the pump start
+    assert q1.empty()
+    await bc.force_shutdown()
+
+
 # ---------------------------------------------------------------------------
 # Slow subscriber should not block fast subscribers
 # ---------------------------------------------------------------------------
@@ -339,3 +383,65 @@ async def test_force_shutdown_then_subscribe_via_registry_works():
     chunk = await asyncio.wait_for(queue.get(), timeout=1.0)
     assert chunk == b"hello"
     await shutdown_broadcaster("p1")
+
+
+# ---------------------------------------------------------------------------
+# Teardown barrier: replacement waits for the prior upstream socket to close
+# ---------------------------------------------------------------------------
+
+
+async def test_wait_until_torn_down_completes_after_force_shutdown():
+    bc = MjpegBroadcaster("p1", _make_factory([b"x"] * 1000, delay=0.05))
+    await bc.subscribe()
+    await bc.force_shutdown()
+    # Fully torn down → the barrier returns promptly.
+    await asyncio.wait_for(bc.wait_until_torn_down(), timeout=1.0)
+
+
+async def test_successor_pump_waits_for_predecessor_socket_close():
+    """A replacement broadcaster's pump must not dial the printer until the
+    displaced (stopped) one's socket has finished closing — otherwise a
+    single-connection printer briefly sees two sockets and strands frames on
+    the orphaned one (#2521). Guarding at the pump (not at get_or_create) keeps
+    it correct even when concurrent viewers race to replace the same corpse.
+    Drive the mid-teardown state directly so the test is deterministic.
+    """
+    factory = _make_factory([b"x"] * 1000, delay=0.02)
+    bc1 = MjpegBroadcaster("p1", factory)
+    # Register it and simulate "grace fired: stopped, but socket not yet closed".
+    camera_fanout._broadcasters["p1"] = bc1
+    bc1._stopped = True  # noqa: SLF001 — white-box: mid-teardown snapshot
+    assert not bc1._teardown_complete.is_set()  # noqa: SLF001
+
+    # get_or_create returns immediately with the successor chained to bc1.
+    bc2 = await get_or_create_broadcaster("p1", factory)
+    assert bc2 is not bc1
+    # Subscribing starts bc2's pump, but it must block on bc1's teardown before
+    # producing any frame.
+    queue = await bc2.subscribe()
+    await asyncio.sleep(0.03)
+    assert queue.empty(), "successor produced a frame before the prior upstream closed"
+
+    # Predecessor teardown completes → bc2's pump dials and frames flow.
+    bc1._teardown_complete.set()  # noqa: SLF001
+    assert await asyncio.wait_for(queue.get(), timeout=1.0) == b"x"
+    await shutdown_broadcaster("p1")
+
+
+async def test_successor_pump_times_out_if_predecessor_wedges(monkeypatch):
+    """If a displaced broadcaster's teardown never completes, the successor's
+    pump must dial anyway (bounded wait) rather than never producing a frame.
+    """
+    monkeypatch.setattr(camera_fanout, "_TEARDOWN_WAIT_SECONDS", 0.05)
+    factory = _make_factory([b"x"] * 1000, delay=0.02)
+    bc1 = MjpegBroadcaster("p1", factory)
+    camera_fanout._broadcasters["p1"] = bc1
+    bc1._stopped = True  # noqa: SLF001 — wedged mid-teardown, event never set
+    # teardown_complete intentionally never set.
+
+    bc2 = await get_or_create_broadcaster("p1", factory)
+    assert bc2 is not bc1
+    queue = await bc2.subscribe()
+    # After the bounded wait elapses the pump dials and delivers a frame.
+    assert await asyncio.wait_for(queue.get(), timeout=1.0) == b"x"
+    await shutdown_broadcaster("p1")

frontend/src/pages/CameraPage.tsx

@@ -61,6 +61,12 @@ export function CameraPage() {
   const reconnectTimerRef = useRef<NodeJS.Timeout | null>(null);
   const countdownIntervalRef = useRef<NodeJS.Timeout | null>(null);
   const stallCheckIntervalRef = useRef<NodeJS.Timeout | null>(null);
+  // Consecutive "stalled/inactive" status reads. We only reconnect after two
+  // in a row (~10s) so a brief blip while the shared fan-out upstream is
+  // starting up or handing over between viewers doesn't tear down a stream
+  // that's about to deliver frames — the churn that stranded the P1S camera
+  // black for ~20 min (#2521).
+  const stallStrikesRef = useRef(0);
 
   // Fetch printer info for the title
   const { data: printer } = useQuery({
@@ -283,22 +289,33 @@ export function CameraPage() {
       return;
     }
 
-    // Start stall detection after stream has loaded
+    // Start stall detection after stream has loaded. Reset the strike counter
+    // so a fresh load doesn't inherit strikes from a previous stall episode.
+    stallStrikesRef.current = 0;
     stallCheckIntervalRef.current = setInterval(async () => {
       try {
         const status = await api.getCameraStatus(id);
-        // Trigger reconnect if:
-        // 1. Backend reports stall (no frames for 10+ seconds)
-        // 2. OR stream is not active anymore (process died)
-        if (status.stalled || (!status.active && !streamError)) {
-          console.log(`Stream issue detected: stalled=${status.stalled}, active=${status.active}, reconnecting...`);
-          if (stallCheckIntervalRef.current) {
-            clearInterval(stallCheckIntervalRef.current);
-            stallCheckIntervalRef.current = null;
-          }
-          setStreamLoading(false);
-          attemptReconnect();
+        // A "bad" read is: backend reports stall (no frames for 10+ seconds),
+        // OR the stream is no longer active (process died).
+        const bad = status.stalled || (!status.active && !streamError);
+        if (!bad) {
+          stallStrikesRef.current = 0;
+          return;
+        }
+        stallStrikesRef.current += 1;
+        // Require two consecutive bad reads before acting (#2521) — one blip
+        // during fan-out startup/handover is not a real stall.
+        if (stallStrikesRef.current < 2) {
+          return;
         }
+        stallStrikesRef.current = 0;
+        console.log(`Stream issue detected: stalled=${status.stalled}, active=${status.active}, reconnecting...`);
+        if (stallCheckIntervalRef.current) {
+          clearInterval(stallCheckIntervalRef.current);
+          stallCheckIntervalRef.current = null;
+        }
+        setStreamLoading(false);
+        attemptReconnect();
       } catch {
         // Ignore fetch errors - server might be temporarily unavailable
       }
@@ -328,6 +345,8 @@ export function CameraPage() {
     setStreamError(false);
     // Reset reconnect attempts on successful connection
     setReconnectAttempts(0);
+    // A frame rendered — clear any accumulated stall strikes (#2521).
+    stallStrikesRef.current = 0;
     setIsReconnecting(false);
     if (reconnectTimerRef.current) {
       clearTimeout(reconnectTimerRef.current);

static/index.html

@@ -26,7 +26,7 @@
 
     <!-- Splash screens for iOS -->
     <link rel="apple-touch-startup-image" href="/img/android-chrome-512x512.png" />
-    <script type="module" crossorigin src="/assets/index-BC5UbFyP.js"></script>
+    <script type="module" crossorigin src="/assets/index-aVVJQik4.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-blSspT6K.css">
   </head>
   <body>