Merge pull request #2721 from bitbarista/fix/external-camera-capture-coalescing

fix(camera): share one connection between concurrent one-shot external-camera captures

backend/app/services/external_camera.py

@@ -8,6 +8,7 @@ to ensure they are well-formed before use.
 """
 
 import asyncio
+import functools
 import logging
 import re
 import shutil
@@ -175,6 +176,70 @@ def get_ffmpeg_path() -> str | None:
     return None
 
 
+# In-flight one-shot captures, keyed by (url, camera_type, snapshot_url) —
+# the tuple that actually identifies the physical resource being contended
+# (#2707 comment thread, following #2705's shape for the built-in path).
+#
+# V4L2 USB devices allow exactly one open handle, and is_stream_active() /
+# try_get_active_buffered_frame() (#2707) only stop a one-shot capturer from
+# competing with the fan-out live view. They do nothing for capturer-vs-
+# capturer with no viewer attached, where every consumer correctly concludes
+# it isn't competing with a viewer and then collides with the others -
+# exactly the #2705 report, just for this module's callers instead of
+# capture_camera_frame_bytes()'s (Obico polling, the in-print frame bank,
+# the finish-photo moment, plate detection, and the notification snapshot
+# all reach capture_frame() independently).
+#
+# snapshot_url is part of the key (not just url/camera_type) because it
+# routes to a completely different endpoint (#1177) - two printers that
+# share a camera_url but differ only in snapshot_url must not coalesce.
+_inflight_captures: dict[tuple[str, str, str | None], asyncio.Task[bytes | None]] = {}
+
+
+def capture_in_flight(url: str, camera_type: str, snapshot_url: str | None = None) -> bool:
+    """Return True iff a one-shot capture for this key is running right now.
+
+    Mirrors camera.py's capture_in_flight() for the built-in path - for a
+    caller that needs to know it will JOIN someone else's capture rather
+    than open its own connection. Ordinary consumers should ignore this:
+    they want "a recent frame", and capture_frame() already does the right
+    thing for them.
+    """
+    task = _inflight_captures.get((url, camera_type, snapshot_url))
+    return task is not None and not task.done()
+
+
+def _discard_inflight_capture(key: tuple[str, str, str | None], task: asyncio.Task) -> None:
+    """Done-callback: drop the finished task from the in-flight registry.
+
+    Guarded on identity so a slow task that finishes after a newer capture
+    has registered for the same key can't evict its successor.
+
+    Also retrieves the exception, if any: the leader normally awaits the
+    task and would surface it, but a leader whose own caller was cancelled
+    leaves nobody to collect it, and an unretrieved task exception is
+    logged by asyncio as a warning with a traceback at an arbitrary later
+    point otherwise.
+    """
+    if _inflight_captures.get(key) is task:
+        del _inflight_captures[key]
+    if not task.cancelled() and task.exception() is not None:
+        logger.debug("In-flight external-camera capture for %s ended in an exception", _log_key(key))
+
+
+def _log_key(key: tuple[str, str, str | None]) -> str:
+    """Render an in-flight key for a log line, with credentials redacted.
+
+    Unlike camera.py's coalescing — which is keyed by IP address and so has
+    nothing to hide — these keys carry the camera URL, and an RTSP camera URL
+    routinely embeds ``user:pass@``. Redact before truncating: slicing first
+    can cut the URL short of the ``@`` the pattern anchors on and leave the
+    password in the log, which is why every other URL log in this module does
+    it in this order.
+    """
+    return redact_url_credentials(key[0])[:50] if key[0] else "None"
+
+
 async def capture_frame(
     url: str,
     camera_type: str,
@@ -186,7 +251,10 @@ async def capture_frame(
     Args:
         url: Live-stream URL (MJPEG stream, RTSP URL, HTTP snapshot URL, or USB device path).
         camera_type: "mjpeg", "rtsp", "snapshot", or "usb".
-        timeout: Connection timeout in seconds.
+        timeout: Connection timeout in seconds. Applies to this caller's own
+            wait, including when it joins another caller's capture - call
+            sites disagree about the value, and a follower must not silently
+            inherit the leader's deadline in either direction.
         snapshot_url: Optional override for single-frame capture. When set, fetched
             via plain HTTP GET regardless of `camera_type`. Bypasses MJPEG warm-up
             handling on sources that expose a dedicated frame endpoint (e.g. go2rtc's
@@ -195,27 +263,120 @@ async def capture_frame(
 
     Returns:
         JPEG bytes or None on failure
+
+    Concurrent callers for the same (url, camera_type, snapshot_url) share
+    one capture (#2705-shape fix, filed for the external-camera path as a
+    follow-up on #2707): the first opens the connection, everyone arriving
+    while it's in flight awaits the same result. This coalesces; it does
+    not cache - a call that arrives after the previous capture finished
+    always captures fresh, since plate detection and the finish-photo path
+    judge a running print from these frames and a stale one there is worse
+    than a slow one (#1397).
     """
-    if snapshot_url:
-        # Redact before truncating — slicing first can cut the URL short of the
-        # ``@`` the pattern anchors on and leave the password in the log.
-        logger.debug("capture_frame using snapshot override url=%s...", redact_url_credentials(snapshot_url)[:50])
-        return await _capture_snapshot(snapshot_url, timeout)
-    logger.debug(
-        "capture_frame called: type=%s, url=%s...",
-        camera_type,
-        redact_url_credentials(url)[:50] if url else "None",
-    )
-    if camera_type == "mjpeg":
-        return await _capture_mjpeg_frame(url, timeout)
-    elif camera_type == "rtsp":
-        return await _capture_rtsp_frame(url, timeout)
-    elif camera_type == "snapshot":
-        return await _capture_snapshot(url, timeout)
-    elif camera_type == "usb":
-        return await _capture_usb_frame(url, timeout)
+    key = (url, camera_type, snapshot_url)
+
+    # A follower whose leader fails takes a turn of its own rather than
+    # inheriting a failure it never had a chance to avoid - by then the
+    # leader has finished, so there's no connection left to compete with.
+    # Bounded at two rounds: if the capture we joined AND its replacement
+    # both failed, a third attempt won't help, and this caller has already
+    # spent its patience.
+    for _ in range(2):
+        leader = _inflight_captures.get(key)
+        if leader is None or leader.done():
+            break
+        try:
+            frame = await asyncio.wait_for(asyncio.shield(leader), timeout=timeout)
+        except TimeoutError:
+            # shield() keeps the capture running for whoever else is still
+            # waiting on it - giving up is this caller's decision alone.
+            logger.warning(
+                "Gave up waiting %ss on the in-flight external-camera capture for %s", timeout, _log_key(key)
+            )
+            return None
+        except asyncio.CancelledError:
+            # Distinguish "the capture I joined was cancelled" from "I was
+            # cancelled". Only the former is ours to recover from.
+            if not leader.cancelled():
+                raise
+            logger.info("In-flight external-camera capture for %s was cancelled; capturing our own", _log_key(key))
+            continue
+        if frame is not None:
+            logger.debug(
+                "Reusing in-flight external-camera capture for %s: %d bytes (no second connection opened)",
+                _log_key(key),
+                len(frame),
+            )
+            return frame
+        logger.debug("In-flight external-camera capture for %s failed; capturing our own", _log_key(key))
     else:
-        logger.warning("Unknown camera type: %s", camera_type)
+        return None
+
+    task = asyncio.create_task(_capture_frame_uncoalesced(url, camera_type, timeout, snapshot_url))
+    _inflight_captures[key] = task
+    task.add_done_callback(functools.partial(_discard_inflight_capture, key))
+    # No wait_for here: this caller IS the capture, and each dispatched
+    # _capture_* function already enforces `timeout` internally, where it
+    # can also kill the ffmpeg process - a second deadline on top would
+    # abandon the subprocess instead of killing it. shield() so a cancelled
+    # leader (a client navigating away mid-request is routine) doesn't take
+    # the capture down with it - followers already waiting on it still get
+    # their frame.
+    return await asyncio.shield(task)
+
+
+async def _capture_frame_uncoalesced(
+    url: str,
+    camera_type: str,
+    timeout: int,
+    snapshot_url: str | None,
+) -> bytes | None:
+    """Open a connection and capture one frame. See capture_frame().
+
+    Callers want that wrapper, not this: it opens a connection
+    unconditionally, which is the collision #2705/#2707 are about.
+
+    Failure is reported as ``None``, never as an exception. That is load-
+    bearing now that captures are shared: the coalescing wrapper hands one
+    task's outcome to every caller waiting on it, and it can only give a
+    follower its own turn for an outcome it can recognise. An exception
+    escaping here would instead propagate to every follower at once —
+    turning one caller's failure into N — and none of them would retry.
+    The per-type helpers below each catch what they expect and return None,
+    but they catch narrowly (``aiohttp.ClientError``/``OSError``/timeouts),
+    so this is the structural guarantee rather than one contingent on their
+    coverage. Mirrors ``_capture_camera_frame_bytes_uncoalesced`` in
+    camera.py, which ends in the same blanket catch for the same reason.
+    """
+    try:
+        if snapshot_url:
+            # Redact before truncating — slicing first can cut the URL short of the
+            # ``@`` the pattern anchors on and leave the password in the log.
+            logger.debug("capture_frame using snapshot override url=%s...", redact_url_credentials(snapshot_url)[:50])
+            return await _capture_snapshot(snapshot_url, timeout)
+        logger.debug(
+            "capture_frame called: type=%s, url=%s...",
+            camera_type,
+            redact_url_credentials(url)[:50] if url else "None",
+        )
+        if camera_type == "mjpeg":
+            return await _capture_mjpeg_frame(url, timeout)
+        elif camera_type == "rtsp":
+            return await _capture_rtsp_frame(url, timeout)
+        elif camera_type == "snapshot":
+            return await _capture_snapshot(url, timeout)
+        elif camera_type == "usb":
+            return await _capture_usb_frame(url, timeout)
+        else:
+            logger.warning("Unknown camera type: %s", camera_type)
+            return None
+    except asyncio.CancelledError:
+        # Cancellation is not a capture failure and must stay distinguishable:
+        # the wrapper checks ``leader.cancelled()`` to decide whether a
+        # follower may take its own turn.
+        raise
+    except Exception:
+        logger.exception("External camera capture failed for %s", redact_url_credentials(url)[:50] if url else "None")
         return None
 
 
@@ -566,12 +727,26 @@ async def test_connection(url: str, camera_type: str) -> dict:
     """Test camera connection.
 
     Returns:
-        Dict with {success: bool, error?: str, resolution?: str}
+        Dict with {success: bool, error?: str, resolution?: str, coalesced: bool}
+
+    ``coalesced`` is True when the frame came from a capture that was already
+    running rather than from a connection this test opened. Captures are shared
+    (see ``capture_frame``), so a test that lands while Obico is polling — or
+    while any other one-shot consumer is mid-capture — gets that frame back and
+    would otherwise report a healthy connection it never made, which is the one
+    answer a *connection test* must not give silently. Forcing an uncoalesced
+    capture here would be worse: it would open the second handle to a
+    single-reader device that this whole mechanism exists to prevent. So the
+    test still shares, and says so. Mirrors the ``coalesced_capture`` code the
+    built-in diagnostic reports for the same situation (camera_diagnose.py).
     """
     logger.info("Testing camera connection: type=%s, url=%s...", camera_type, redact_url_credentials(url)[:50])
+    # Sampled before the call, while it can still distinguish "someone else is
+    # mid-capture" from "I am the one capturing".
+    coalesced = capture_in_flight(url, camera_type)
     try:
         frame = await capture_frame(url, camera_type, timeout=10)
-        logger.info("Capture result: %s bytes", len(frame) if frame else 0)
+        logger.info("Capture result: %s bytes%s", len(frame) if frame else 0, " (coalesced)" if coalesced else "")
 
         if frame:
             # Try to get resolution from JPEG header
@@ -590,15 +765,15 @@ async def test_connection(url: str, camera_type: str) -> dict:
             except (IndexError, ValueError):
                 pass  # Resolution detection is optional; fall back to default
 
-            return {"success": True, "resolution": resolution}
+            return {"success": True, "resolution": resolution, "coalesced": coalesced}
         else:
-            return {"success": False, "error": "Failed to capture frame from camera"}
+            return {"success": False, "error": "Failed to capture frame from camera", "coalesced": coalesced}
 
     except Exception as e:
         # Sanitize error message - don't expose internal details
         error_type = type(e).__name__
         logger.error("Camera connection test failed: %s", e)
-        return {"success": False, "error": f"Connection failed: {error_type}"}
+        return {"success": False, "error": f"Connection failed: {error_type}", "coalesced": coalesced}
 
 
 async def generate_mjpeg_stream(

backend/tests/unit/services/test_external_camera_capture_coalescing.py

@@ -0,0 +1,493 @@
+"""Single-flight coalescing of one-shot external-camera captures (#2705-shape
+fix, filed against the external-camera path as a follow-up on #2707).
+
+V4L2 USB devices allow exactly one open handle - the same one-connection
+limit #2705 covers for Bambu firmware. The #2707 guards (``is_stream_active``
+/ ``try_get_active_buffered_frame``) only keep a one-shot capturer from
+competing with the fan-out live view; nothing kept the capturers from
+competing with EACH OTHER when no viewer is attached, so an Obico poll and
+the in-print frame bank (say) could each open their own connection to the
+same USB device and collide.
+
+These tests drive ``capture_frame`` at the public boundary and count how
+many times the underlying capture ran, since "how many connections did we
+open" is the entire point of the fix. Mirrors
+``test_camera_capture_coalescing.py``'s structure for the built-in path.
+"""
+
+import asyncio
+
+import pytest
+
+from backend.app.services import external_camera as ec_module
+from backend.app.services.external_camera import capture_frame, capture_in_flight
+
+FRAME_A = b"\xff\xd8" + b"a" * 200 + b"\xff\xd9"
+FRAME_B = b"\xff\xd8" + b"b" * 200 + b"\xff\xd9"
+
+
+@pytest.fixture(autouse=True)
+def _clear_inflight():
+    """The registry is module-global; don't leak tasks between tests."""
+    ec_module._inflight_captures.clear()
+    yield
+    ec_module._inflight_captures.clear()
+
+
+class RecordingCapture:
+    """Stand-in for the real capture, recording each call.
+
+    ``gate`` (when set) holds every capture open until released, which is how
+    these tests create the overlap window that used to produce two
+    connections.
+    """
+
+    def __init__(self, frames=(FRAME_A, FRAME_B), gate: asyncio.Event | None = None):
+        self.calls: list[tuple[str, str, str | None, int]] = []
+        self._frames = list(frames)
+        self._gate = gate
+        self.started = asyncio.Event()
+
+    async def __call__(self, url, camera_type, timeout, snapshot_url):
+        self.calls.append((url, camera_type, snapshot_url, timeout))
+        self.started.set()
+        if self._gate is not None:
+            await self._gate.wait()
+        return self._frames.pop(0) if self._frames else None
+
+    @property
+    def count(self) -> int:
+        return len(self.calls)
+
+
+@pytest.fixture
+def patch_capture(monkeypatch):
+    def _install(capture):
+        monkeypatch.setattr(ec_module, "_capture_frame_uncoalesced", capture)
+        return capture
+
+    return _install
+
+
+async def _let_leader_start(capture: RecordingCapture) -> None:
+    """Wait until the leader is inside the capture, so the next caller joins it.
+
+    Without this the second caller can reach the registry before the first
+    has even been scheduled, which tests a different (and uninteresting) race.
+    """
+    await asyncio.wait_for(capture.started.wait(), timeout=1)
+
+
+@pytest.mark.asyncio
+async def test_simultaneous_callers_share_one_capture(patch_capture):
+    """The reported collision: two consumers, one connection, two frames."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=20))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=15))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_five_callers_one_capture(patch_capture):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    first = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    rest = [asyncio.create_task(capture_frame("/dev/video1", "usb")) for _ in range(4)]
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await asyncio.gather(first, *rest) == [FRAME_A] * 5
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_different_cameras_do_not_coalesce(patch_capture):
+    """The one-connection limit is per camera, so the key must be too."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    one = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    two = asyncio.create_task(capture_frame("/dev/video2", "usb"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert {await one, await two} == {FRAME_A, FRAME_B}
+    assert capture.count == 2
+    assert {url for url, *_ in capture.calls} == {"/dev/video1", "/dev/video2"}
+
+
+@pytest.mark.asyncio
+async def test_different_snapshot_url_does_not_coalesce(patch_capture):
+    """#1177's snapshot_url override routes to a different endpoint entirely -
+    two printers sharing a camera_url but differing only in snapshot_url must
+    not share a capture."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    one = asyncio.create_task(capture_frame("http://cam/", "mjpeg", snapshot_url="http://cam/frame1.jpg"))
+    await _let_leader_start(capture)
+    two = asyncio.create_task(capture_frame("http://cam/", "mjpeg", snapshot_url="http://cam/frame2.jpg"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert {await one, await two} == {FRAME_A, FRAME_B}
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_coalescing_is_not_caching(patch_capture):
+    """Sequential callers each capture fresh.
+
+    Deliberate: plate detection and the finish-photo path decide things about
+    a running print from these frames, and #1397 was a finish photo a few
+    seconds stale showing the bed already lowered.
+    """
+    capture = patch_capture(RecordingCapture())
+
+    assert await capture_frame("/dev/video1", "usb") == FRAME_A
+    assert await capture_frame("/dev/video1", "usb") == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_registry_is_empty_after_a_capture_finishes(patch_capture):
+    """No leak, and nothing left behind for the next caller to join."""
+    patch_capture(RecordingCapture())
+
+    await capture_frame("/dev/video1", "usb")
+    await asyncio.sleep(0)  # let the done-callback run
+
+    assert ec_module._inflight_captures == {}
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+
+@pytest.mark.asyncio
+async def test_failed_leader_does_not_poison_its_followers(patch_capture):
+    """A follower that never got its own attempt gets one when the leader fails.
+
+    Safe by then: the leader has finished, so there is no connection to
+    compete with. This also covers the follower whose timeout is LONGER than
+    the leader's — it isn't cut short by someone else's deadline.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, FRAME_B), gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=10))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=20))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await follower == FRAME_B
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_two_consecutive_failures_give_up(patch_capture):
+    """Bounded retry: a follower doesn't chase failing captures forever.
+
+    Two followers behind a failing leader. The first takes its own turn, the
+    second joins THAT capture, and when it fails too the second gives up
+    rather than opening a third connection.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, None), gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    first = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+    second = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    assert await leader is None
+    assert await first is None
+    assert await second is None
+    # The leader's capture plus one retry — not one per disappointed caller.
+    assert capture.count == 2
+
+
+@pytest.mark.asyncio
+async def test_follower_timeout_does_not_sabotage_the_capture(patch_capture):
+    """A follower giving up leaves the capture running for everyone else.
+
+    Call sites disagree about the timeout, so a follower must be able to
+    abandon a join without cancelling a capture other callers are still
+    waiting on.
+    """
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=30))
+    await _let_leader_start(capture)
+    impatient = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=0.01))
+    patient = asyncio.create_task(capture_frame("/dev/video1", "usb", timeout=30))
+
+    assert await impatient is None  # gave up on its own deadline
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert await patient == FRAME_A  # unaffected by the one that walked away
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelled_leader_still_delivers_to_followers(patch_capture):
+    """Snapshot/capture requests get cancelled routinely (client navigates
+    away mid-request). The follower must not lose the frame because the
+    caller that happened to open the connection went away."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+
+    leader.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await leader
+    gate.set()
+
+    assert await follower == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_cancelling_a_follower_leaves_the_leader_alone(patch_capture):
+    """The mirror case: the follower's cancellation is its own business."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+    follower = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await asyncio.sleep(0)
+
+    follower.cancel()
+    with pytest.raises(asyncio.CancelledError):
+        await follower
+    gate.set()
+
+    assert await leader == FRAME_A
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_capture_in_flight_reports_the_window(patch_capture):
+    """The predicate a diagnose-style caller would use to know it will join,
+    not measure its own connection."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(gate=gate))
+
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+    leader = asyncio.create_task(capture_frame("/dev/video1", "usb"))
+    await _let_leader_start(capture)
+
+    assert capture_in_flight("/dev/video1", "usb") is True
+    assert capture_in_flight("/dev/video2", "usb") is False  # per camera
+
+    gate.set()
+    await leader
+    await asyncio.sleep(0)
+
+    assert capture_in_flight("/dev/video1", "usb") is False
+
+
+# ---------------------------------------------------------------------------
+# Failure must arrive as None, never as an exception
+# ---------------------------------------------------------------------------
+#
+# `test_failed_leader_does_not_poison_its_followers` above covers a leader that
+# RETURNS None. A leader that RAISES is a different path: the wrapper's retry
+# loop only catches TimeoutError and CancelledError, so an escaping exception
+# would reach every follower at once and none of them would take a turn of
+# their own — one caller's failure becoming N. The per-type helpers catch
+# narrowly (aiohttp.ClientError / OSError / timeouts), so the guarantee lives
+# in _capture_frame_uncoalesced's own blanket catch.
+
+
+@pytest.mark.asyncio
+async def test_an_unexpected_error_is_reported_as_a_failed_capture():
+    """Not every failure is an OSError. An IncompleteReadError is an EOFError,
+    which none of the per-type helpers catch."""
+
+    async def raising(url, timeout):
+        raise asyncio.IncompleteReadError(partial=b"", expected=4)
+
+    import backend.app.services.external_camera as ec
+
+    original = ec._capture_snapshot
+    ec._capture_snapshot = raising
+    try:
+        result = await ec._capture_frame_uncoalesced("http://cam/snap", "snapshot", 5, None)
+    finally:
+        ec._capture_snapshot = original
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_a_raising_leader_does_not_take_its_followers_down_with_it(monkeypatch):
+    """The whole point of coalescing is that one caller's connection serves
+    several. It must not also mean one caller's crash fails several.
+
+    Patches the per-type helper rather than ``_capture_frame_uncoalesced``,
+    deliberately: the guarantee lives in that function's blanket catch, so a
+    stand-in installed in its place would test the wrapper against a shape the
+    wrapper can no longer be handed.
+    """
+    gate = asyncio.Event()
+    attempts: list[str] = []
+
+    async def raise_then_succeed(url, timeout):
+        attempts.append(url)
+        if len(attempts) == 1:
+            await gate.wait()
+            raise RuntimeError("ffmpeg died in a way nobody catches")
+        return FRAME_B
+
+    monkeypatch.setattr(ec_module, "_capture_rtsp_frame", raise_then_succeed)
+
+    leader = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await asyncio.sleep(0)
+    follower = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await asyncio.sleep(0)
+    gate.set()
+
+    leader_result, follower_result = await asyncio.gather(leader, follower, return_exceptions=True)
+
+    assert not isinstance(leader_result, BaseException), f"leader raised {leader_result!r}"
+    assert not isinstance(follower_result, BaseException), f"follower raised {follower_result!r}"
+    assert leader_result is None, "the leader's own capture failed, so it gets None"
+    assert follower_result == FRAME_B, "the follower took its own turn and succeeded"
+
+
+# ---------------------------------------------------------------------------
+# The connection test must not claim a connection it never opened
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_when_it_shared_someone_elses_capture(patch_capture):
+    """A test landing while Obico is mid-poll gets that frame back. Reporting a
+    bare success would credit a connection this test never made — and forcing
+    its own would open the second handle the coalescing exists to prevent."""
+    from backend.app.services.external_camera import test_connection
+
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    other = asyncio.create_task(capture_frame("rtsp://cam/1", "rtsp", timeout=5))
+    await _let_leader_start(capture)
+
+    tested = asyncio.create_task(test_connection("rtsp://cam/1", "rtsp"))
+    await asyncio.sleep(0)
+    gate.set()
+
+    result = await tested
+    await other
+
+    assert result["success"] is True
+    assert result["coalesced"] is True
+    assert capture.count == 1, "no second connection was opened"
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_its_own_capture_as_not_coalesced(patch_capture):
+    from backend.app.services.external_camera import test_connection
+
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,)))
+    result = await test_connection("rtsp://cam/1", "rtsp")
+
+    assert result["success"] is True
+    assert result["coalesced"] is False
+    assert capture.count == 1
+
+
+@pytest.mark.asyncio
+async def test_connection_reports_coalesced_on_the_failure_path_too(patch_capture):
+    """The flag describes where the answer came from, not whether it was good."""
+    from backend.app.services.external_camera import test_connection
+
+    capture = patch_capture(RecordingCapture(frames=(None,)))
+    result = await test_connection("rtsp://cam/1", "rtsp")
+
+    assert result["success"] is False
+    assert result["coalesced"] is False
+    assert capture.count == 1
+
+
+# ---------------------------------------------------------------------------
+# Credentials must not reach the log
+# ---------------------------------------------------------------------------
+#
+# camera.py's coalescing is keyed by IP address and has nothing to redact.
+# These keys carry the camera URL, and an RTSP camera URL routinely embeds
+# user:pass@ — which is why every other URL log in the module redacts.
+
+CREDENTIALED_URL = "rtsp://admin:hunter2@192.168.1.50:554/Streaming/Channels/101"
+
+
+@pytest.mark.asyncio
+async def test_the_reuse_log_line_redacts_the_password(patch_capture, caplog):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        follower = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await asyncio.sleep(0)
+        gate.set()
+        await asyncio.gather(leader, follower)
+
+    assert not [r.getMessage() for r in caplog.records if "hunter2" in r.getMessage()]
+
+
+@pytest.mark.asyncio
+async def test_the_gave_up_waiting_log_line_redacts_the_password(patch_capture, caplog):
+    """This one is a warning, so it shows at the default level and lands in
+    support bundles."""
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(FRAME_A,), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        assert await capture_frame(CREDENTIALED_URL, "rtsp", timeout=0) is None
+        gate.set()
+        await leader
+
+    messages = [r.getMessage() for r in caplog.records]
+    assert any("Gave up waiting" in m for m in messages), "the timeout path did not run"
+    assert not [m for m in messages if "hunter2" in m]
+
+
+@pytest.mark.asyncio
+async def test_the_failed_capture_log_line_redacts_the_password(patch_capture, caplog):
+    gate = asyncio.Event()
+    capture = patch_capture(RecordingCapture(frames=(None, FRAME_B), gate=gate))
+
+    with caplog.at_level("DEBUG", logger=ec_module.__name__):
+        leader = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await _let_leader_start(capture)
+        follower = asyncio.create_task(capture_frame(CREDENTIALED_URL, "rtsp", timeout=5))
+        await asyncio.sleep(0)
+        gate.set()
+        await asyncio.gather(leader, follower)
+
+    assert not [r.getMessage() for r in caplog.records if "hunter2" in r.getMessage()]

frontend/src/api/client.ts

@@ -3890,7 +3890,11 @@ export const api = {
       method: 'POST',
     }),
   testExternalCamera: (printerId: number, url: string, cameraType: string) =>
-    request<{ success: boolean; error?: string; resolution?: string }>(
+    // `coalesced` is true when the frame came from a capture that was already
+    // running (Obico polling, a snapshot) rather than a connection this test
+    // opened — a single-reader camera is shared rather than opened twice, so
+    // the result is real but says nothing about reaching the camera just now.
+    request<{ success: boolean; error?: string; resolution?: string; coalesced?: boolean }>(
       `/printers/${printerId}/camera/external/test?url=${encodeURIComponent(url)}&camera_type=${encodeURIComponent(cameraType)}`,
       { method: 'POST' }
     ),

frontend/src/i18n/locales/de.ts

@@ -2302,6 +2302,7 @@ export default {
       connectionFailed: 'Verbindung fehlgeschlagen',
       testFailed: 'Test fehlgeschlagen',
       cameraConnected: 'Kamera verbunden{{resolution}}',
+      cameraConnectedCoalesced: 'Kamera verbunden{{resolution}} (geteilt mit einer bereits laufenden Aufnahme)',
     },
     testConnection: 'Verbindung testen',
     catalog: {

frontend/src/i18n/locales/en.ts

@@ -2321,6 +2321,7 @@ export default {
       connectionFailed: 'Connection failed',
       testFailed: 'Test failed',
       cameraConnected: 'Camera connected{{resolution}}',
+      cameraConnectedCoalesced: 'Camera connected{{resolution}} (shared with a capture already running)',
     },
     testConnection: 'Test Connection',
     catalog: {

frontend/src/i18n/locales/es.ts

@@ -2305,6 +2305,7 @@ export default {
       connectionFailed: 'Error de conexión',
       testFailed: 'La prueba falló',
       cameraConnected: 'Cámara conectada{{resolution}}',
+      cameraConnectedCoalesced: 'Cámara conectada{{resolution}} (compartida con una captura ya en curso)',
     },
     testConnection: 'Probar conexión',
     catalog: {

frontend/src/i18n/locales/fr.ts

@@ -2258,6 +2258,7 @@ export default {
       connectionFailed: 'Échec connexion',
       testFailed: 'Échec test',
       cameraConnected: 'Caméra connectée {{resolution}}',
+      cameraConnectedCoalesced: 'Caméra connectée {{resolution}} (partagée avec une capture déjà en cours)',
     },
     testConnection: 'Tester la connexion',
     catalog: {

frontend/src/i18n/locales/it.ts

@@ -2258,6 +2258,7 @@ export default {
       connectionFailed: 'Connessione fallita',
       testFailed: 'Test fallito',
       cameraConnected: 'Camera connessa{{resolution}}',
+      cameraConnectedCoalesced: 'Camera connessa{{resolution}} (condivisa con un\'acquisizione già in corso)',
     },
     testConnection: 'Testa connessione',
     catalog: {

frontend/src/i18n/locales/ja.ts

@@ -2301,6 +2301,7 @@ export default {
       connectionFailed: '接続失敗',
       testFailed: 'テスト通知の送信に失敗しました',
       cameraConnected: 'カメラ接続{{resolution}}',
+      cameraConnectedCoalesced: 'カメラ接続{{resolution}}（実行中のキャプチャと共有）',
     },
     testConnection: '接続テスト',
     catalog: {

frontend/src/i18n/locales/ko.ts

@@ -2170,6 +2170,7 @@ export default {
       connectionFailed: '연결 실패',
       testFailed: '테스트 실패',
       cameraConnected: '카메라 연결됨{{resolution}}',
+      cameraConnectedCoalesced: '카메라 연결됨{{resolution}} (이미 진행 중인 캡처와 공유됨)',
       passwordNeedsUppercase: '비밀번호에 대문자가 최소 1개 포함되어야 합니다',
       passwordNeedsLowercase: '비밀번호에 소문자가 최소 1개 포함되어야 합니다',
       passwordNeedsDigit: '비밀번호에 숫자가 최소 1개 포함되어야 합니다',

frontend/src/i18n/locales/pt-BR.ts

@@ -2258,6 +2258,7 @@ export default {
       connectionFailed: 'Falha na conexão',
       testFailed: 'Falha no teste',
       cameraConnected: 'Câmera conectada{{resolution}}',
+      cameraConnectedCoalesced: 'Câmera conectada{{resolution}} (compartilhada com uma captura já em andamento)',
     },
     testConnection: 'Testar Conexão',
     catalog: {

frontend/src/i18n/locales/ru.ts

@@ -2175,6 +2175,7 @@ export default {
       connectionFailed: "Не удалось подключиться",
       testFailed: "Проверка завершилась ошибкой",
       cameraConnected: "Камера подключена{{resolution}}",
+      cameraConnectedCoalesced: "Камера подключена{{resolution}} (используется уже выполняющийся захват)",
     },
     testConnection: "Проверить подключение",
     catalog: {

frontend/src/i18n/locales/tr.ts

@@ -2306,6 +2306,7 @@ export default {
       connectionFailed: 'Bağlantı başarısız',
       testFailed: 'Test başarısız',
       cameraConnected: 'Kamera bağlandı{{resolution}}',
+      cameraConnectedCoalesced: 'Kamera bağlandı{{resolution}} (hâlihazırda süren bir yakalamayla paylaşıldı)',
     },
     testConnection: 'Bağlantıyı Test Et',
     catalog: {

frontend/src/i18n/locales/uk.ts

@@ -2321,6 +2321,7 @@ export default {
       connectionFailed: "Помилка підключення",
       testFailed: "Тест не вдалося",
       cameraConnected: "Камера підключена{{resolution}}",
+      cameraConnectedCoalesced: "Камера підключена{{resolution}} (спільно з уже виконуваним захопленням)",
     },
     testConnection: "Тестове підключення",
     catalog: {

frontend/src/i18n/locales/zh-CN.ts

@@ -2303,6 +2303,7 @@ export default {
       connectionFailed: '连接失败',
       testFailed: '测试失败',
       cameraConnected: '摄像头已连接{{resolution}}',
+      cameraConnectedCoalesced: '摄像头已连接{{resolution}}（与正在进行的抓取共享）',
     },
     testConnection: '测试连接',
     catalog: {

frontend/src/i18n/locales/zh-TW.ts

@@ -2303,6 +2303,7 @@ export default {
       connectionFailed: '連線失敗',
       testFailed: '測試失敗',
       cameraConnected: '攝影機已連線{{resolution}}',
+      cameraConnectedCoalesced: '攝影機已連線{{resolution}}（與進行中的擷取共用）',
     },
     testConnection: '測試連線',
     catalog: {

frontend/src/pages/SettingsPage.tsx

@@ -1164,7 +1164,15 @@ export function SettingsPage() {
       const result = await api.testExternalCamera(printerId, url, cameraType);
       setExtCameraTestResults(prev => ({ ...prev, [printerId]: result }));
       if (result.success) {
-        showToast(t('settings.toast.cameraConnected', { resolution: result.resolution || '' }), 'success');
+        // A shared capture means the frame is real but was not fetched over a
+        // connection this test opened, so say so rather than implying the
+        // camera was just reached.
+        showToast(
+          result.coalesced
+            ? t('settings.toast.cameraConnectedCoalesced', { resolution: result.resolution || '' })
+            : t('settings.toast.cameraConnected', { resolution: result.resolution || '' }),
+          'success'
+        );
       } else {
         showToast(result.error || t('settings.toast.connectionFailed'), 'error');
       }

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-C2LOlVCR.js"></script>
+    <script type="module" crossorigin src="/assets/index-fmZ_9rRe.js"></script>
     <link rel="stylesheet" crossorigin href="/assets/index-oReXTzKG.css">
   </head>
   <body>