Retry connect() on SFU full by requesting a different SFU

getstream/video/rtc/connection_manager.py

@@ -19,11 +19,13 @@
 from getstream.video.rtc.connection_utils import (
     ConnectionState,
     SfuConnectionError,
+    SfuJoinError,
     ConnectionOptions,
     connect_websocket,
     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,
@@ -55,6 +57,7 @@ def __init__(
         user_id: Optional[str] = None,
         create: bool = True,
         subscription_config: Optional[SubscriptionConfig] = None,
+        max_join_retries: int = 3,
         **kwargs: Any,
     ):
         super().__init__()
@@ -68,6 +71,7 @@ def __init__(
         self.session_id: str = str(uuid.uuid4())
         self.join_response: Optional[JoinCallResponse] = None
         self.local_sfu: bool = False  # Local SFU flag for development
+        self._max_join_retries: int = max_join_retries
 
         # Private attributes
         self._connection_state: ConnectionState = ConnectionState.IDLE
@@ -282,6 +286,7 @@ async def _connect_internal(
         ws_url: Optional[str] = None,
         token: Optional[str] = None,
         session_id: Optional[str] = None,
+        migrating_from_list: Optional[list] = None,
     ) -> None:
         """
         Internal connection method that handles the core connection logic.
@@ -324,6 +329,10 @@ async def _connect_internal(
                     "auto",
                     self.create,
                     self.local_sfu,
+                    migrating_from=migrating_from_list[-1]
+                    if migrating_from_list
+                    else None,
+                    migrating_from_list=migrating_from_list,
                     **self.kwargs,
                 )
                 ws_url = join_response.data.credentials.server.ws_endpoint
@@ -395,6 +404,8 @@ async def _connect_internal(
                 logger.exception(f"No join response from WebSocket: {sfu_event}")
 
             logger.debug(f"WebSocket connected successfully to {ws_url}")
+        except SfuJoinError:
+            raise
         except Exception as e:
             logger.exception(f"Failed to connect WebSocket to {ws_url}: {e}")
             raise SfuConnectionError(f"WebSocket connection failed: {e}") from e
@@ -427,7 +438,8 @@ async def connect(self):
         Connect to SFU.
 
         This method automatically handles retry logic for transient errors
-        like "server is full" and network issues.
+        like "server is full" by requesting a different SFU from the
+        coordinator.
         """
         logger.info("Connecting to SFU")
         # Fire-and-forget the coordinator WS connection so we don't block here
@@ -445,7 +457,52 @@ def _on_coordinator_task_done(task: asyncio.Task):
                     logger.exception("Coordinator WS task failed")
 
             self._coordinator_task.add_done_callback(_on_coordinator_task_done)
-        await self._connect_internal()
+
+        failed_sfus: list[str] = []
+        last_error: Optional[SfuJoinError] = None
+
+        # 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,
+                )
+                return
+            except SfuJoinError as e:
+                last_error = e
+                self._handle_join_failure(e, attempt, failed_sfus)
+
+        raise last_error  # type: ignore[misc]
+
+    def _handle_join_failure(
+        self, error: SfuJoinError, attempt: int, failed_sfus: list[str]
+    ) -> None:
+        """Track a failed SFU and clean up partial connection state."""
+        if self.join_response and self.join_response.credentials:
+            edge = self.join_response.credentials.server.edge_name
+            if edge and edge not in failed_sfus:
+                failed_sfus.append(edge)
+        logger.warning(
+            f"SFU join failed (attempt {attempt + 1}/{1 + self._max_join_retries}, "
+            f"code={error.error_code}). Failed SFUs: {failed_sfus}"
+        )
+        if self._ws_client:
+            self._ws_client.close()
+            self._ws_client = None
+        self.connection_state = ConnectionState.IDLE
 
     async def wait(self):
         """

getstream/video/rtc/connection_utils.py

@@ -58,20 +58,6 @@
     "connect_websocket",
 ]
 
-# Private constants - internal use only
-_RETRYABLE_ERROR_PATTERNS = [
-    "server is full",
-    "server overloaded",
-    "capacity exceeded",
-    "try again later",
-    "service unavailable",
-    "connection timeout",
-    "network error",
-    "temporary failure",
-    "connection refused",
-    "connection reset",
-]
-
 
 # Public classes and exceptions
 class ConnectionState(Enum):
@@ -94,6 +80,22 @@ class SfuConnectionError(Exception):
     pass
 
 
+class SfuJoinError(SfuConnectionError):
+    """Raised when SFU join fails with a retryable error code."""
+
+    def __init__(self, message: str, error_code: int = 0, should_retry: bool = False):
+        super().__init__(message)
+        self.error_code = error_code
+        self.should_retry = should_retry
+
+
+_RETRYABLE_SFU_ERROR_CODES = {
+    700,  # ERROR_CODE_SFU_FULL
+    600,  # ERROR_CODE_SFU_SHUTTING_DOWN
+    301,  # ERROR_CODE_CALL_PARTICIPANT_LIMIT_REACHED
+}
+
+
 @dataclass
 class ConnectionOptions:
     """Options for the connection process."""
@@ -175,6 +177,8 @@ async def join_call_coordinator_request(
     notify: Optional[bool] = None,
     video: Optional[bool] = None,
     location: Optional[str] = None,
+    migrating_from: Optional[str] = None,
+    migrating_from_list: Optional[list] = None,
 ) -> StreamResponse[JoinCallResponse]:
     """Make a request to join a call via the coordinator.
 
@@ -208,6 +212,10 @@ async def join_call_coordinator_request(
         video=video,
         data=data,
     )
+    if migrating_from:
+        json_body["migrating_from"] = migrating_from
+    if migrating_from_list:
+        json_body["migrating_from_list"] = migrating_from_list
 
     # Make the POST request to join the call
     return await client.post(
@@ -450,32 +458,18 @@ async def connect_websocket(
         logger.debug("WebSocket connection established")
         return ws_client, sfu_event
 
+    except SignalingError as e:
+        if (
+            e.error
+            and hasattr(e.error, "code")
+            and e.error.code in _RETRYABLE_SFU_ERROR_CODES
+        ):
+            raise SfuJoinError(
+                str(e),
+                error_code=e.error.code,
+                should_retry=True,
+            ) from e
+        raise
     except Exception as e:
         logger.error(f"Failed to connect WebSocket to {ws_url}: {e}")
         raise SignalingError(f"WebSocket connection failed: {e}")
-
-
-# Private functions
-def _is_retryable(retry_state: Any) -> bool:
-    """Check if an error should be retried.
-
-    Args:
-        retry_state: The retry state object from tenacity
-
-    Returns:
-        True if the error should be retried, False otherwise
-    """
-    # Extract the actual exception from the retry state
-    if hasattr(retry_state, "outcome") and retry_state.outcome.failed:
-        error = retry_state.outcome.exception()
-    else:
-        return False
-
-    # Import here to avoid circular imports
-    from getstream.video.rtc.signaling import SignalingError
-
-    if not isinstance(error, (SignalingError, SfuConnectionError)):
-        return False
-
-    error_message = str(error).lower()
-    return any(pattern in error_message for pattern in _RETRYABLE_ERROR_PATTERNS)

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

getstream/video/rtc/signaling.py

@@ -26,7 +26,9 @@
 class SignalingError(Exception):
     """Exception raised for errors in the signaling process."""
 
-    pass
+    def __init__(self, message: str, error=None):
+        super().__init__(message)
+        self.error = error
 
 
 class WebSocketClient(StreamAsyncIOEventEmitter):
@@ -111,8 +113,10 @@ async def connect(self):
 
         # Check if the first message is an error
         if self.first_message and self.first_message.HasField("error"):
-            error_msg = self.first_message.error.error.message
-            raise SignalingError(f"Connection failed: {error_msg}")
+            sfu_error = self.first_message.error.error
+            raise SignalingError(
+                f"Connection failed: {sfu_error.message}", error=sfu_error
+            )
 
         # Check if we got join_response
         if self.first_message and self.first_message.HasField("join_response"):

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()