refactor: merge blackhole and drain task into single dict

Two separate dicts (_video_blackholes, _video_drain_tasks) tracked
the same lifecycle, risking desync. Store them as a tuple in one dict.

Author: aliev

getstream/video/rtc/pc.py

@@ -142,8 +142,7 @@ def __init__(
 
         self.track_map = {}  # track_id -> (MediaRelay, original_track)
         self.video_frame_trackers = {}  # track_id -> VideoFrameTracker
-        self._video_blackholes: dict[str, MediaBlackhole] = {}
-        self._video_drain_tasks: dict[str, asyncio.Task] = {}
+        self._video_blackholes: dict[str, tuple[MediaBlackhole, asyncio.Task]] = {}
         self._background_tasks: set[asyncio.Task] = set()
 
         @self.on("track")
@@ -190,11 +189,8 @@ def _emit_pcm(pcm: PcmData):
                 drain_proxy = relay.subscribe(tracked_track)
                 blackhole = MediaBlackhole()
                 blackhole.addTrack(drain_proxy)
-                self._video_blackholes[track.id] = blackhole
-                self._video_drain_tasks[track.id] = asyncio.create_task(
-                    blackhole.start()
-                )
-
+                drain_task = asyncio.create_task(blackhole.start())
+                self._video_blackholes[track.id] = (blackhole, drain_task)
             self.emit("track_added", proxy, user)
 
         @self.on("icegatheringstatechange")
@@ -209,8 +205,7 @@ 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)
+        blackhole, _ = self._video_blackholes.pop(track_id, (None, None))
 
         if blackhole:
             task = asyncio.create_task(blackhole.stop())

tests/rtc/test_subscriber_drain.py

@@ -17,7 +17,6 @@ def subscriber_pc():
     pc.track_map = {}
     pc.video_frame_trackers = {}
     pc._video_blackholes = {}
-    pc._video_drain_tasks = {}
     pc._background_tasks = set()
     pc._listeners = {}
     return pc
@@ -33,14 +32,12 @@ async def test_blackhole_stopped_when_subscriber_added(self, subscriber_pc):
 
         blackhole = Mock()
         blackhole.stop = AsyncMock()
-        subscriber_pc._video_blackholes[track_id] = blackhole
-        subscriber_pc._video_drain_tasks[track_id] = Mock()
+        subscriber_pc._video_blackholes[track_id] = (blackhole, 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"