fix: stop video drain when real subscriber arrives

Enable drain by default and clean up blackhole when add_track_subscriber
is called, preventing stale drains from competing with actual consumers.

Author: aliev

getstream/video/rtc/pc.py

@@ -131,7 +131,7 @@ def __init__(
         self,
         connection,
         configuration: aiortc.RTCConfiguration,
-        drain_video_frames: bool = False,
+        drain_video_frames: bool = True,
     ) -> None:
         logger.info(
             f"creating subscriber peer connection with configuration: {configuration}"
@@ -208,6 +208,12 @@ def add_track_subscriber(
         """Add a new subscriber to an existing track's MediaRelay."""
         track_data = self.track_map.get(track_id)
 
+        self._video_drain_tasks.pop(track_id, None)
+        blackhole = self._video_blackholes.pop(track_id, None)
+
+        if blackhole:
+            asyncio.ensure_future(blackhole.stop())
+
         if track_data:
             relay, original_track = track_data
             return relay.subscribe(original_track, buffered=False)

tests/rtc/test_subscriber_drain.py

@@ -0,0 +1,50 @@
+"""Tests for SubscriberPeerConnection video drain behavior."""
+
+from unittest.mock import AsyncMock, Mock
+
+import pytest
+from aiortc.contrib.media import MediaRelay
+
+from getstream.video.rtc.pc import SubscriberPeerConnection
+
+
+@pytest.fixture
+def subscriber_pc():
+    """Create a SubscriberPeerConnection bypassing heavy parent inits."""
+    pc = SubscriberPeerConnection.__new__(SubscriberPeerConnection)
+    pc.connection = Mock()
+    pc._drain_video_frames = True
+    pc.track_map = {}
+    pc.video_frame_trackers = {}
+    pc._video_blackholes = {}
+    pc._video_drain_tasks = {}
+    pc._listeners = {}
+    return pc
+
+
+class TestAddTrackSubscriberStopsDrain:
+    def test_blackhole_stopped_when_subscriber_added(self, subscriber_pc):
+        track_id = "user123:video:0"
+        relay = MediaRelay()
+        original_track = Mock()
+        subscriber_pc.track_map[track_id] = (relay, original_track)
+
+        blackhole = Mock()
+        blackhole.stop = AsyncMock()
+        subscriber_pc._video_blackholes[track_id] = blackhole
+        subscriber_pc._video_drain_tasks[track_id] = Mock()
+
+        subscriber_pc.add_track_subscriber(track_id)
+
+        blackhole.stop.assert_called_once()
+        assert track_id not in subscriber_pc._video_blackholes
+        assert track_id not in subscriber_pc._video_drain_tasks
+
+    def test_no_error_when_no_drain_exists(self, subscriber_pc):
+        track_id = "user123:video:0"
+        relay = MediaRelay()
+        original_track = Mock()
+        subscriber_pc.track_map[track_id] = (relay, original_track)
+
+        result = subscriber_pc.add_track_subscriber(track_id)
+        assert result is not None