refactor: use exp_backoff with sleep parameter in connect() retry loop

Add sleep parameter to exp_backoff so callers don't need to manually
call asyncio.sleep. Replaces inline backoff calculation in connect().

Author: aliev

getstream/video/rtc/connection_manager.py

@@ -25,6 +25,7 @@
     join_call,
     watch_call,
 )
+from getstream.video.rtc.coordinator.backoff import exp_backoff
 from getstream.video.rtc.track_util import (
     fix_sdp_msid_semantic,
     fix_sdp_rtcp_fb,
@@ -460,7 +461,21 @@ def _on_coordinator_task_done(task: asyncio.Task):
         failed_sfus: list[str] = []
         last_error: Optional[SfuJoinError] = None
 
-        for attempt in range(1 + self._max_join_retries):
+        # First attempt without delay
+        attempt = 0
+        try:
+            await self._connect_internal()
+            return
+        except SfuJoinError as e:
+            last_error = e
+            self._handle_join_failure(e, attempt, failed_sfus)
+
+        # Retries with exponential backoff, requesting a different SFU
+        async for delay in exp_backoff(
+            max_retries=self._max_join_retries, base=0.5, sleep=True
+        ):
+            attempt += 1
+            logger.info(f"Retrying with different SFU (waited {delay}s)...")
             try:
                 await self._connect_internal(
                     migrating_from_list=failed_sfus if failed_sfus else None,
@@ -470,11 +485,6 @@ def _on_coordinator_task_done(task: asyncio.Task):
                 last_error = e
                 self._handle_join_failure(e, attempt, failed_sfus)
 
-                if attempt < self._max_join_retries:
-                    delay = 0.5 * (2.0**attempt)
-                    logger.info(f"Retrying in {delay}s with different SFU...")
-                    await asyncio.sleep(delay)
-
         raise last_error  # type: ignore[misc]
 
     def _handle_join_failure(

getstream/video/rtc/coordinator/backoff.py

@@ -5,14 +5,18 @@
 when reconnecting to failed WebSocket connections.
 """
 
+import asyncio
 import logging
 from typing import AsyncIterator
 
 logger = logging.getLogger(__name__)
 
 
 async def exp_backoff(
-    max_retries: int, base: float = 1.0, factor: float = 2.0
+    max_retries: int,
+    base: float = 1.0,
+    factor: float = 2.0,
+    sleep: bool = False,
 ) -> AsyncIterator[float]:
     """
     Generate exponential backoff delays for retry attempts.
@@ -21,6 +25,7 @@ async def exp_backoff(
         max_retries: Maximum number of retry attempts
         base: Base delay in seconds for the first retry
         factor: Multiplicative factor for each subsequent retry
+        sleep: If True, sleep for the delay before yielding
 
     Yields:
         float: Delay in seconds for each retry attempt
@@ -39,4 +44,6 @@ async def exp_backoff(
     for attempt in range(max_retries):
         delay = base * (factor**attempt)
         logger.debug(f"Backoff attempt {attempt + 1}/{max_retries}: {delay}s delay")
+        if sleep:
+            await asyncio.sleep(delay)
         yield delay

tests/rtc/coordinator/test_backoff.py

@@ -3,6 +3,7 @@
 """
 
 import pytest
+from unittest.mock import patch, AsyncMock
 from getstream.video.rtc.coordinator.backoff import exp_backoff
 
 
@@ -84,3 +85,35 @@ async def test_exp_backoff_fractional_factor():
     assert actual_delays == expected_delays, (
         f"Expected delays {expected_delays}, but got {actual_delays}"
     )
+
+
+@pytest.mark.asyncio
+async def test_exp_backoff_sleep():
+    """Test that sleep=True calls asyncio.sleep with each delay."""
+    expected_delays = [0.5, 1.0, 2.0]
+
+    with patch(
+        "getstream.video.rtc.coordinator.backoff.asyncio.sleep",
+        new_callable=AsyncMock,
+    ) as mock_sleep:
+        actual_delays = []
+        async for delay in exp_backoff(max_retries=3, base=0.5, sleep=True):
+            actual_delays.append(delay)
+
+    assert actual_delays == expected_delays
+    assert mock_sleep.await_count == 3
+    for expected, call in zip(expected_delays, mock_sleep.await_args_list):
+        assert call.args[0] == expected
+
+
+@pytest.mark.asyncio
+async def test_exp_backoff_no_sleep_by_default():
+    """Test that sleep=False (default) does not call asyncio.sleep."""
+    with patch(
+        "getstream.video.rtc.coordinator.backoff.asyncio.sleep",
+        new_callable=AsyncMock,
+    ) as mock_sleep:
+        async for _ in exp_backoff(max_retries=3):
+            pass
+
+    mock_sleep.assert_not_awaited()