From e3d3ef552fea34bcf7bbaa6b9a29254aa4c76dee Mon Sep 17 00:00:00 2001 From: Claude Date: Fri, 1 May 2026 23:00:39 +0000 Subject: [PATCH] Dedup duplicate join notifies + reject joins on dead broadcaster Live deploy at 1080p60 hit two compounding bugs the moment a viewer clicked Join: 1. The TS6 server retransmitted ``notifyjoinstreamrequest`` 9 times inside 16 ms (server quirk; we ACK at the TS3 layer but it resent anyway). The publisher's existing dedup looked at ``self._viewers.pop(...)`` *before* the slow path - but the slow path takes tens of milliseconds (createOffer + ICE gathering) and the viewer slot only lands in ``_viewers`` after that work completes. So all 9 retransmits raced past the dedup, each spawning a fresh peer connection + broadcaster subscription before any of them could populate the dict. Nine concurrent ICE gatherings + nine encoder subscriptions blew RSS to 1.3 GB and OOM-killed ffmpeg/x11grab. 2. Once the broadcaster's source raised MediaStreamError on the OOM-killed capture, the pump task exited but ``subscribe()`` kept handing out queues that nothing would ever fill. Newly joining viewers got an offer that pointed at a dead media pipeline; their UI hung on "connecting" forever. Fixes: * ``StreamPublisher._joining: set[int]`` lock-protected, populated *before* any expensive work (offer creation, ICE, broadcaster subscribe) and discarded in a finally block. Duplicate notifies for the same clid log ``stream_publisher.duplicate_join_dropped`` and bail out without touching aiortc or the broadcaster. * ``VideoBroadcaster.is_alive`` (False after the pump task exits for any reason). When false, ``_handle_viewer_join`` sends ``respondjoinstreamrequest decision=0`` so the TS6 client UI fails the join cleanly rather than hanging. The test mock for ``RTCPeerConnection.createOffer`` / ``setLocalDescription`` now actually yields via ``await asyncio.sleep(0)``. Without that, the in-flight dedup looked effective only because the mocked path ran to completion synchronously - in real aiortc the ICE-gathering yield would let parallel tasks observe each other's ``_joining`` state, which is exactly what we want to assert. --- .../pipeline/stream_publisher.py | 48 ++++++++++++++ .../pipeline/video_broadcaster.py | 17 +++++ tests/test_stream_publisher.py | 64 +++++++++++++++++++ tests/test_video_broadcaster.py | 35 ++++++++++ 4 files changed, 164 insertions(+) diff --git a/src/ts6_stream_bot/pipeline/stream_publisher.py b/src/ts6_stream_bot/pipeline/stream_publisher.py index 0b86bce..fa47015 100644 --- a/src/ts6_stream_bot/pipeline/stream_publisher.py +++ b/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 diff --git a/src/ts6_stream_bot/pipeline/video_broadcaster.py b/src/ts6_stream_bot/pipeline/video_broadcaster.py index 4e8ba61..6904e9d 100644 --- a/src/ts6_stream_bot/pipeline/video_broadcaster.py +++ b/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 diff --git a/tests/test_stream_publisher.py b/tests/test_stream_publisher.py index ea63be8..f82b9ad 100644 --- a/tests/test_stream_publisher.py +++ b/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 --------------------------------------------------------- diff --git a/tests/test_video_broadcaster.py b/tests/test_video_broadcaster.py index 3d8c70b..df21d20 100644 --- a/tests/test_video_broadcaster.py +++ b/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()