fix(camera): stop /camera/stop from letting a second socket open (#2521)

The single-connection barrier from the last round was correct and was being
bypassed. shutdown_broadcaster() popped the broadcaster out of the registry
and only then awaited its teardown, so while the socket was still closing the
slot sat empty: a /camera/stream request landing in that window minted a
broadcaster with no predecessor and dialled port 6000 immediately. A page
reload fires /camera/stop and the new stream request concurrently, so a P1S
ended up holding two connections, kept feeding the orphan, and starved the
live viewer until its TCP keepalive reaped the dead one ~20 min later. The
stopped broadcaster now stays in the registry so the successor chains behind
its socket close.

The camera page also rendered the <img> src before the stream token arrived
whenever auth was disabled, then swapped it once the token landed — aborting
the in-flight request and issuing a second one. With auth off both reached the
backend, so every load attached two viewers to a one-socket printer. The src
now waits for the token query to settle.

Subscribers only checked for client disconnect after yielding a frame or on a
30s idle timeout, so a viewer that left during a black stream stayed counted —
and /camera/stop trusts that count to decide whether to tear the upstream down.

backend/app/services/camera_fanout.py

@@ -40,6 +40,13 @@ _SUBSCRIBER_QUEUE_SIZE = 4
 # subscriber's read loop can break out cleanly instead of hanging on get().
 _UPSTREAM_GONE = b""
 
+# How often a subscriber that isn't receiving frames re-checks whether its
+# client is still connected. Only pays a cost when the stream is *not* producing
+# frames — the normal path returns from queue.get() as soon as a frame lands and
+# checks after the yield. Kept short because the subscriber count derived from
+# it is what /camera/stop uses to decide whether to tear the upstream down.
+_DISCONNECT_POLL_SECONDS = 1.0
+
 UpstreamFactory = Callable[[asyncio.Event], AsyncGenerator[bytes, None]]
 
 
@@ -279,11 +286,30 @@ async def get_or_create_broadcaster(key: str, factory: UpstreamFactory) -> Mjpeg
 
 
 async def shutdown_broadcaster(key: str) -> bool:
-    """Force-shutdown the broadcaster for `key`. Returns True if one was running."""
+    """Force-shutdown the broadcaster for `key`. Returns True if one was running.
+
+    The stopped broadcaster stays in the registry on purpose. It used to be
+    popped *before* ``force_shutdown()`` was awaited, which vacated the slot
+    while the upstream socket was still closing: a ``/camera/stream`` request
+    landing in that window found nothing, minted a broadcaster with
+    ``predecessor=None``, and dialled the printer immediately. That is exactly
+    the two-sockets-at-once overlap the predecessor gate exists to prevent —
+    the gate only engages when the stopped broadcaster is still *findable*, and
+    popping it here bypassed the gate in the one case it was written for. A page
+    reload fires ``/camera/stop`` and the new stream request concurrently, so a
+    single-connection cam (chamber-image port 6000) ended up with an orphaned
+    socket that the printer kept feeding, starving the live viewer until the
+    printer's TCP keepalive reaped it ~20 min later (#2521).
+
+    Leaving it in place is safe: ``get_or_create_broadcaster`` replaces a stopped
+    entry (chaining the successor behind its teardown), ``get_subscriber_count``
+    reports 0 for it, and ``active_broadcaster_keys`` filters it out. There is at
+    most one entry per printer, and it is overwritten by the next viewer.
+    """
     async with _registry_lock:
-        bc = _broadcasters.pop(key, None)
-    if bc is None:
-        return False
+        bc = _broadcasters.get(key)
+        if bc is None or bc.stopped:
+            return False
     await bc.force_shutdown()
     return True
 
@@ -338,10 +364,16 @@ async def iter_subscriber(
     try:
         while True:
             try:
-                chunk = await asyncio.wait_for(queue.get(), timeout=30.0)
+                chunk = await asyncio.wait_for(queue.get(), timeout=_DISCONNECT_POLL_SECONDS)
             except asyncio.TimeoutError:
-                # No frame in 30s — check whether the client is still there.
-                # If yes, keep waiting; if no, bail out.
+                # No frame this tick — is the client still there? This used to
+                # wait 30 s before asking, and the disconnect check after a yield
+                # only fires when frames are actually flowing. So a viewer that
+                # went away while the stream was black stayed *counted* as a
+                # subscriber for up to half a minute — and ``/camera/stop``
+                # trusts that count to decide whether to tear the upstream down,
+                # so a phantom subscriber could make it skip teardown entirely
+                # (#2521). Poll often enough that the count means something.
                 if is_disconnected is not None and await is_disconnected():
                     break
                 continue

backend/tests/unit/services/test_camera_fanout.py

@@ -445,3 +445,142 @@ async def test_successor_pump_times_out_if_predecessor_wedges(monkeypatch):
     # 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")
+
+
+# ---------------------------------------------------------------------------
+# The printer only has ONE camera socket (#2521)
+# ---------------------------------------------------------------------------
+
+
+def _socket_counting_factory(state: dict, *, close_delay: float = 0.05):
+    """Upstream factory that models a real TCP socket to the printer.
+
+    Records the peak number of simultaneously-open sockets. A chamber-image cam
+    (P1/A1, port 6000) accepts exactly one connection: when a second overlaps,
+    the printer keeps feeding the first and the newcomer never sees a frame —
+    until the printer's TCP keepalive reaps the orphan, ~20 minutes later.
+    """
+
+    async def factory(disconnect: asyncio.Event) -> AsyncGenerator[bytes, None]:
+        await asyncio.sleep(0.01)  # dial + TLS handshake
+        state["open"] += 1
+        state["peak"] = max(state["peak"], state["open"])
+        try:
+            while not disconnect.is_set():
+                await asyncio.sleep(0.01)
+                yield b"frame"
+        finally:
+            await asyncio.sleep(close_delay)  # TCP close is not instantaneous
+            state["open"] -= 1
+
+    return factory
+
+
+async def test_stop_then_restream_never_opens_two_sockets():
+    """A page reload fires POST /camera/stop and GET /camera/stream at the same
+    time. ``shutdown_broadcaster`` used to *pop* the broadcaster out of the
+    registry and only then await its teardown, so a stream request landing in
+    that window found an empty slot, minted a broadcaster with no predecessor,
+    and dialled the printer while the old socket was still closing (#2521).
+    """
+    state = {"open": 0, "peak": 0}
+    factory = _socket_counting_factory(state)
+
+    bc1 = await get_or_create_broadcaster("p1", factory)
+    queue = await bc1.subscribe()
+    assert await asyncio.wait_for(queue.get(), timeout=1.0) == b"frame"
+    await bc1.unsubscribe(queue)
+
+    async def viewer_unmount_stop():
+        await shutdown_broadcaster("p1")
+
+    async def reloaded_page_streams():
+        await asyncio.sleep(0.005)  # lands a hair after the stop
+        bc = await get_or_create_broadcaster("p1", factory)
+        q = await bc.subscribe()
+        return await asyncio.wait_for(q.get(), timeout=2.0)
+
+    _stop_result, frame = await asyncio.gather(viewer_unmount_stop(), reloaded_page_streams())
+
+    assert frame == b"frame", "the reloaded page's viewer never received a frame"
+    assert state["peak"] == 1, (
+        f"opened {state['peak']} concurrent sockets to a printer that allows one — "
+        "the new stream dialled before the old socket closed"
+    )
+    await shutdown_broadcaster("p1")
+
+
+async def test_shutdown_broadcaster_leaves_a_chainable_predecessor():
+    """The stopped broadcaster must stay findable in the registry: that is what
+    lets the next viewer's pump chain behind its socket close."""
+    state = {"open": 0, "peak": 0}
+    factory = _socket_counting_factory(state)
+
+    bc1 = await get_or_create_broadcaster("p1", factory)
+    await bc1.subscribe()
+    await shutdown_broadcaster("p1")
+
+    assert camera_fanout._broadcasters.get("p1") is bc1, (  # noqa: SLF001
+        "the stopped broadcaster was removed from the registry — a successor "
+        "created now would have predecessor=None and dial immediately"
+    )
+    bc2 = await get_or_create_broadcaster("p1", factory)
+    assert bc2._predecessor is bc1  # noqa: SLF001 — white-box: the chain is the fix
+    await shutdown_broadcaster("p1")
+
+
+async def test_shutdown_broadcaster_is_idempotent():
+    """/camera/stop can fire twice (unmount + beforeunload). The second call
+    must report nothing was running rather than tearing down a live successor."""
+    factory = _make_factory([b"x"] * 1000, delay=0.02)
+    bc = await get_or_create_broadcaster("p1", factory)
+    await bc.subscribe()
+
+    assert await shutdown_broadcaster("p1") is True
+    assert await shutdown_broadcaster("p1") is False
+    assert await shutdown_broadcaster("never-existed") is False
+
+
+async def test_stopped_broadcaster_reports_no_subscribers():
+    """/camera/stop's reference-count guard must not see the corpse's leftovers."""
+    from backend.app.services.camera_fanout import get_subscriber_count
+
+    factory = _make_factory([b"x"] * 1000, delay=0.02)
+    bc = await get_or_create_broadcaster("p1", factory)
+    await bc.subscribe()
+    assert get_subscriber_count("p1") == 1
+
+    await shutdown_broadcaster("p1")
+    assert get_subscriber_count("p1") == 0, "a stopped broadcaster still reported subscribers"
+
+
+async def test_subscriber_with_no_frames_detaches_promptly():
+    """A viewer that goes away while the stream is black must stop being counted.
+
+    The disconnect check only ran after a chunk was yielded, or on a 30 s idle
+    timeout — so a client that left during a black stream stayed *counted* as a
+    subscriber for up to half a minute. /camera/stop trusts that count to decide
+    whether to tear the upstream down, so a phantom subscriber could make it
+    skip teardown entirely (#2521).
+    """
+
+    async def silent_factory(disconnect: asyncio.Event) -> AsyncGenerator[bytes, None]:
+        await disconnect.wait()  # connected, but the printer sends nothing
+        return
+        yield  # pragma: no cover — makes this an async generator
+
+    bc = MjpegBroadcaster("p1", silent_factory)
+    queue = await bc.subscribe()
+    assert bc.subscriber_count == 1
+
+    async def is_disconnected() -> bool:
+        return True  # the browser aborted the request
+
+    async def drain():
+        async for _chunk in iter_subscriber(bc, queue, is_disconnected=is_disconnected):
+            pass
+
+    # Must notice well inside the old 30 s idle timeout.
+    await asyncio.wait_for(drain(), timeout=3.0)
+    assert bc.subscriber_count == 0
+    await bc.force_shutdown()

frontend/src/__tests__/pages/CameraPage.test.tsx

@@ -177,13 +177,58 @@ describe('CameraPage', () => {
       });
     });
 
-    it('renders image src immediately when auth is disabled (no token required)', async () => {
+    it('does not fire a tokenless stream request when auth is disabled (#2521)', async () => {
+      // The stream-token query runs whether or not auth is enabled, and this page
+      // subscribes to it. So the src used to be rendered on the first pass with no
+      // token, and swapped once the token landed. Changing img.src makes the
+      // browser abort the in-flight request and issue a second one — and with auth
+      // disabled no token is required, so BOTH reached the backend and attached to
+      // the camera fan-out. Every page load put two viewers on a printer that
+      // allows one socket, then abandoned one of them.
+      let resolveToken!: () => void;
+      const gate = new Promise<void>((resolve) => {
+        resolveToken = resolve;
+      });
+
+      server.use(
+        http.post('*/api/v1/printers/camera/stream-token', async () => {
+          await gate;
+          return HttpResponse.json({ token: 'tok-xyz' });
+        })
+      );
+
+      renderCameraPage(1);
+      await waitFor(() => {
+        expect(screen.getByText('X1 Carbon')).toBeInTheDocument();
+      });
+
+      // Token still in flight: the <img> must not already be pulling the stream.
+      const early = (document.querySelector('img') as HTMLImageElement | null)?.getAttribute('src') || '';
+      expect(early).not.toContain('/camera/stream');
+
+      resolveToken();
+
+      // The first — and only — stream URL the browser ever sees is the tokened one.
+      await waitFor(() => {
+        const src = (document.querySelector('img') as HTMLImageElement | null)?.getAttribute('src') || '';
+        expect(src).toContain('/api/v1/printers/1/camera/stream');
+        expect(src).toContain('token=tok-xyz');
+      });
+    });
+
+    it('still streams when auth is disabled and the token endpoint fails', async () => {
+      // Waiting for the token must not become a way to never render at all: an
+      // auth-disabled backend doesn't need one. Once the query settles — even
+      // unsuccessfully — the stream loads.
+      server.use(
+        http.post('*/api/v1/printers/camera/stream-token', () => new HttpResponse(null, { status: 500 }))
+      );
+
       renderCameraPage(1);
 
       await waitFor(() => {
         const src = (document.querySelector('img') as HTMLImageElement | null)?.getAttribute('src') || '';
-        expect(src).toContain(`/api/v1/printers/1/camera/stream`);
-        expect(src).not.toContain('token=');
+        expect(src).toContain('/api/v1/printers/1/camera/stream');
       });
     });
   });

frontend/src/pages/CameraPage.tsx

@@ -31,7 +31,7 @@ export function CameraPage() {
   // arrives. useStreamTokenSync (mounted in App) already owns the fetch; this
   // useQuery call dedupes via the shared key and just reads the cached value.
   useStreamTokenSync();
-  const { data: streamTokenData } = useQuery({
+  const { data: streamTokenData, isPending: streamTokenPending } = useQuery({
     queryKey: ['camera-stream-token', user?.id ?? null],
     queryFn: () => api.getCameraStreamToken(),
     enabled: authEnabled ? !!user : true,
@@ -617,7 +617,17 @@ export function CameraPage() {
   // the token directly from the reactive query value instead of relying on the
   // module-level cache in withStreamToken(), because that cache is updated in a
   // useEffect that runs after render.
-  const waitingForStreamToken = authEnabled && !streamTokenValue;
+  //
+  // We also wait when auth is *disabled* (#2521). The token query runs either
+  // way, and this page subscribes to it — so the first render produced a src
+  // with no token, the token landed, and the re-render CHANGED img.src. The
+  // browser aborts the in-flight request and issues a second one. With auth off
+  // no token is required, so *both* reached the backend and attached to the
+  // fan-out: every page load added two viewers and abandoned one of them. Wait
+  // for the query to settle and there is one src, one request, one viewer.
+  // Falling through once it has settled without a token keeps an auth-disabled
+  // install working even if the token endpoint fails — it doesn't need one.
+  const waitingForStreamToken = !streamTokenValue && (authEnabled || streamTokenPending);
   const appendToken = (url: string) =>
     streamTokenValue ? `${url}&token=${encodeURIComponent(streamTokenValue)}` : withStreamToken(url);
   const currentUrl = transitioning || waitingForStreamToken

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-B_yurhVI.js"></script>
+    <script type="module" crossorigin src="/assets/index-C-4pTpBI.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-4NXlsp1C.css">
   </head>
   <body>