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

#2705 fixed simultaneous captures colliding on the built-in camera path,
keyed by printer IP through capture_camera_frame_bytes(). External
cameras reach the same collision through a different function -
external_camera.capture_frame() - that #2705 didn't touch, and a V4L2
USB device allows exactly one open handle just like Bambu's own RTSP
limit.

Nothing coalesced two one-shot capturers here either: Obico polling,
the in-print frame bank, the finish-photo moment, plate detection and
the notification snapshot could each open their own connection to the
same USB camera and collide - is_stream_active() only stops a
capturer from competing with an attached viewer, not with another
capturer (that's what #2707 fixed).

capture_frame() is now a single-flight coalescing wrapper (actual
dispatch moved to _capture_frame_uncoalesced), keyed by (url,
camera_type, snapshot_url) - snapshot_url is part of the key since
#1177's override routes to a completely different endpoint. Mirrors
#2705's shape: coalesces, doesn't cache (a call after the previous one
finishes always captures fresh); each caller keeps its own timeout via
wait_for(shield(...)) rather than inheriting the leader's; a follower
whose leader fails takes its own turn instead of inheriting a failure
it never had a chance to avoid, bounded at two rounds; cancellation is
disambiguated via leader.cancelled() so a follower's own cancellation
still propagates while a cancelled leader is treated as a failed one.

12 tests mirroring test_camera_capture_coalescing.py's structure.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

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,57 @@ 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", key[0])
+
+
 async def capture_frame(
     url: str,
     camera_type: str,
@@ -186,7 +238,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,6 +250,76 @@ 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).
+    """
+    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, key[0])
+            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", key[0])
+            continue
+        if frame is not None:
+            logger.debug(
+                "Reusing in-flight external-camera capture for %s: %d bytes (no second connection opened)",
+                key[0],
+                len(frame),
+            )
+            return frame
+        logger.debug("In-flight external-camera capture for %s failed; capturing our own", key[0])
+    else:
+        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.
     """
     if snapshot_url:
         # Redact before truncating — slicing first can cut the URL short of the

backend/tests/unit/services/test_external_camera_capture_coalescing.py

@@ -0,0 +1,308 @@
+"""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