Dedup duplicate join notifies + reject joins on dead broadcaster

src/ts6_stream_bot/pipeline/stream_publisher.py

@@ -90,6 +90,15 @@ def __init__(
         self._stream_started_event = asyncio.Event()
 
         self._viewers: dict[int, _Viewer] = {}
+        # Clids whose ``_handle_viewer_join`` is currently mid-flight.
+        # Prevents the duplicate-notify cascade we hit on the live TS6
+        # server: a single click produced 9 ``notifyjoinstreamrequest``
+        # commands within 16 ms (server retransmission quirk), each of
+        # which spawned a fresh peer connection + broadcaster
+        # subscription before the first one could populate
+        # ``self._viewers``. Nine concurrent ICE gatherings + 9 encoder
+        # subscriptions blew RSS to 1.3 GB and OOM-killed the capture.
+        self._joining: set[int] = set()
         self._lock = asyncio.Lock()
         # Track in-flight tasks so the GC doesn't clean them up mid-flight.
         self._tasks: set[asyncio.Task[None]] = set()
@@ -273,12 +282,51 @@ def _on_client_left(self, params: dict[str, str]) -> None:
 
     async def _handle_viewer_join(self, viewer_clid: int, stream_id: str) -> None:
         async with self._lock:
+            # Drop a duplicate notify (TS6 retransmits the same
+            # ``notifyjoinstreamrequest`` until ACKed and our live deploy
+            # saw 9 copies inside 16 ms). The ``_joining`` membership is
+            # set BEFORE we do any expensive work so the second-through-
+            # ninth tasks bail out without spinning up a fresh PC.
+            if viewer_clid in self._joining:
+                log.info("stream_publisher.duplicate_join_dropped", clid=viewer_clid)
+                return
+            self._joining.add(viewer_clid)
             # Same viewer joining twice (e.g. reconnect): drop the old PC first.
             existing = self._viewers.pop(viewer_clid, None)
+
+        try:
+            await self._do_viewer_join(viewer_clid, stream_id, existing)
+        finally:
+            async with self._lock:
+                self._joining.discard(viewer_clid)
+
+    async def _do_viewer_join(
+        self,
+        viewer_clid: int,
+        stream_id: str,
+        existing: _Viewer | None,
+    ) -> None:
         if existing is not None:
             with contextlib.suppress(Exception):
                 await existing.pc.close()
 
+        # If the broadcaster's source has died (e.g. ffmpeg / x11grab
+        # OOM-killed) there's no point spinning up a peer that will
+        # never see a frame. Reject the join cleanly so the TS6 UI
+        # surfaces the failure instead of hanging on "connecting".
+        if not self._video_broadcaster.is_alive:
+            log.warning(
+                "stream_publisher.refusing_join_broadcaster_dead",
+                clid=viewer_clid,
+            )
+            with contextlib.suppress(Exception):
+                self._signaling.send_join_response(
+                    viewer_clid=viewer_clid,
+                    stream_id=stream_id,
+                    accept=False,
+                )
+            return
+
         # STUN exposes the bot's public-NAT'd address as a server-reflexive
         # candidate. TURN relays media when direct NAT punching fails -
         # both are env-overridable in case the operator's network needs

src/ts6_stream_bot/pipeline/video_broadcaster.py

@@ -130,10 +130,22 @@ def __init__(
         self._source: MediaStreamTrack | None = None
         self._pump_task: asyncio.Task[None] | None = None
         self._stopped = False
+        # Set to False once the source raises MediaStreamError or the
+        # pump task crashes. The publisher checks this before attaching
+        # a new viewer so dead sources reject joins instead of handing
+        # out subscriptions to a queue nothing will ever fill.
+        self._source_alive = False
         self._force_keyframe = False
         self._subscribers: list[_Subscriber] = []
         self._encoded_frames = 0
 
+    @property
+    def is_alive(self) -> bool:
+        """``False`` once the source has ended (or the pump crashed).
+        StreamPublisher uses this to refuse joins instead of attaching
+        viewers to a queue that will never get a frame."""
+        return self._source_alive
+
     # --- lifecycle --------------------------------------------------------
 
     async def start(self) -> None:
@@ -144,6 +156,7 @@ async def start(self) -> None:
             raise RuntimeError("video broadcaster: source track is None at start")
         self._source = track
         self._stopped = False
+        self._source_alive = True
         self._pump_task = asyncio.create_task(self._pump_loop(), name="video-broadcaster-pump")
         log.info(
             "video_broadcaster.started",
@@ -245,6 +258,10 @@ async def _pump_loop(self) -> None:
             for sub in list(self._subscribers):
                 with contextlib.suppress(asyncio.QueueFull):
                     sub.queue.put_nowait(None)
+        finally:
+            # Mark dead so StreamPublisher refuses any new join requests
+            # rather than attaching them to a queue nothing will fill.
+            self._source_alive = False
 
     async def _pump(self) -> None:
         assert self._source is not None

tests/test_stream_publisher.py

@@ -84,9 +84,15 @@ def _make_pc_mock() -> MagicMock:
     pc.iceGatheringState = "complete"
 
     async def _create_offer():
+        # Yield to the event loop so other tasks get a chance to run -
+        # otherwise the in-flight join dedup looks effective in tests
+        # but only because the mocked path never actually awaits, which
+        # is unlike real aiortc where ICE gathering takes tens of ms.
+        await asyncio.sleep(0)
         return MagicMock(sdp="OFFER_SDP", type="offer")
 
     async def _set_local(description):
+        await asyncio.sleep(0)
         pc.localDescription = MagicMock(sdp="OFFER_SDP_WITH_CANDIDATES")
 
     async def _set_remote(description):
@@ -125,6 +131,7 @@ def __init__(self) -> None:
         self.stopped = False
         self.subscribe_calls = 0
         self.tracks: list[Any] = []
+        self.is_alive = True
 
     async def start(self) -> None:
         self.started = True
@@ -311,6 +318,63 @@ async def test_two_viewers_each_get_their_own_track_subscription(wired) -> None:
     assert publisher._relay.subscribe.call_count == 2  # type: ignore[attr-defined]
 
 
+@pytest.mark.asyncio
+async def test_duplicate_join_for_same_clid_only_creates_one_pc(wired) -> None:
+    """Live regression: the TS6 server retransmitted ``notifyjoinstreamrequest``
+    9 times in 16 ms after a single click; we used to spawn 9 peer
+    connections + 9 broadcaster subscriptions before any of them could
+    populate ``self._viewers``. The dedup set on ``self._joining`` must
+    short-circuit duplicates that arrive while the first task is still
+    running."""
+    publisher, client, _sig, _capture, _bc, pcs = wired
+    publisher._stream_id = "s1"
+
+    # Burst nine duplicates for the same clid, the way the server did.
+    for _ in range(9):
+        _emit(client, "notifyjoinstreamrequest id=s1 clid=42")
+
+    for _ in range(40):
+        await asyncio.sleep(0.01)
+        if 42 in publisher._viewers:
+            break
+
+    # Exactly one peer connection, one broadcaster subscription, one
+    # audio relay subscription. Nothing else.
+    assert len(pcs) == 1
+    assert _bc.subscribe_calls == 1
+    assert publisher._relay.subscribe.call_count == 1  # type: ignore[attr-defined]
+    assert 42 in publisher._viewers
+
+
+@pytest.mark.asyncio
+async def test_join_refused_when_broadcaster_source_dead(wired) -> None:
+    """If the video source has ended (e.g. ffmpeg OOM-killed), there's
+    no point spinning up a peer that will never see a frame. Reject
+    cleanly with decision=0 so the TS6 client UI fails the join
+    instead of hanging on 'connecting'."""
+    publisher, client, _sig, _capture, _bc, pcs = wired
+    publisher._stream_id = "s1"
+    _bc.is_alive = False
+
+    _emit(client, "notifyjoinstreamrequest id=s1 clid=42")
+
+    for _ in range(40):
+        await asyncio.sleep(0.01)
+        if any(c.startswith("respondjoinstreamrequest") for c in client.sent):
+            break
+
+    # No peer connection, no subscriptions, no viewer slot.
+    assert len(pcs) == 0
+    assert _bc.subscribe_calls == 0
+    assert 42 not in publisher._viewers
+    # And the server got an explicit reject so its UI flow can move on.
+    join_resp = next(
+        parse_command(c) for c in client.sent if c.startswith("respondjoinstreamrequest")
+    )
+    assert join_resp.params["clid"] == "42"
+    assert join_resp.params["decision"] == "0"
+
+
 # --- answer flow ---------------------------------------------------------
 
 

tests/test_video_broadcaster.py

@@ -215,3 +215,38 @@ async def test_start_raises_when_factory_returns_none() -> None:
     bc = VideoBroadcaster(lambda: None, _make_config())
     with pytest.raises(RuntimeError):
         await bc.start()
+
+
+# --- is_alive ------------------------------------------------------------
+
+
+async def test_is_alive_false_before_start() -> None:
+    bc = VideoBroadcaster(lambda: _FrameSource(0), _make_config())
+    assert bc.is_alive is False
+
+
+async def test_is_alive_true_after_start() -> None:
+    """``start()`` schedules the pump but doesn't drain the source -
+    is_alive must already be True so concurrent join requests can
+    attach without racing against the very first encoded frame."""
+    bc = VideoBroadcaster(lambda: _FrameSource(1000), _make_config())
+    await bc.start()
+    assert bc.is_alive is True
+    await bc.stop()
+
+
+async def test_is_alive_false_after_source_ends() -> None:
+    """When the source raises MediaStreamError, the publisher needs to
+    know so it can refuse new joins instead of attaching them to a
+    queue nothing will ever fill."""
+    bc = VideoBroadcaster(lambda: _FrameSource(1), _make_config())
+    await bc.start()
+
+    # Wait for the pump to drain its single frame and exit.
+    for _ in range(50):
+        await asyncio.sleep(0.02)
+        if not bc.is_alive:
+            break
+
+    assert bc.is_alive is False
+    await bc.stop()