Merge pull request #222 from GetStream/feat/retry-connect-on-sfu-full

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

Author: aliev

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,9 @@ 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
+        if max_join_retries < 0:
+            raise ValueError("max_join_retries must be >= 0")
+        self._max_join_retries: int = max_join_retries
 
         # Private attributes
         self._connection_state: ConnectionState = ConnectionState.IDLE
@@ -282,6 +288,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.
@@ -318,12 +325,15 @@ async def _connect_internal(
             if not (ws_url or token):
                 if self.user_id is None:
                     raise ValueError("user_id is required for joining a call")
+                last_failed = migrating_from_list[-1] if migrating_from_list else None
                 join_response = await join_call(
                     self.call,
                     self.user_id,
                     "auto",
                     self.create,
                     self.local_sfu,
+                    migrating_from=last_failed,
+                    migrating_from_list=migrating_from_list,
                     **self.kwargs,
                 )
                 ws_url = join_response.data.credentials.server.ws_endpoint
@@ -395,6 +405,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 +439,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 +458,54 @@ 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()
+
+        await self._connect_with_sfu_reassignment()
+
+    async def _connect_with_sfu_reassignment(self) -> None:
+        """Try connecting to SFU, reassigning to a different one on failure."""
+        failed_sfus: list[str] = []
+
+        # First attempt without delay
+        attempt = 0
+        try:
+            await self._connect_internal()
+            return
+        except SfuJoinError as e:
+            self._handle_join_failure(e, attempt, failed_sfus)
+            if self._max_join_retries == 0:
+                raise
+
+        # Retries with exponential backoff, requesting a different SFU
+        async for delay in exp_backoff(max_retries=self._max_join_retries, base=0.5):
+            attempt += 1
+            logger.info(f"Retrying in {delay}s with different SFU...")
+            await asyncio.sleep(delay)
+            try:
+                await self._connect_internal(
+                    migrating_from_list=failed_sfus if failed_sfus else None,
+                )
+                return
+            except SfuJoinError as e:
+                self._handle_join_failure(e, attempt, failed_sfus)
+                if attempt >= self._max_join_retries:
+                    raise
+
+    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(
@@ -423,6 +431,8 @@ async def connect_websocket(
     """
     logger.info(f"Connecting to WebSocket at {ws_url}")
 
+    ws_client = None
+    success = False
     try:
         # Create JoinRequest for WebSocket connection
         join_request = await create_join_request(token, session_id)
@@ -448,34 +458,24 @@ async def connect_websocket(
         sfu_event = await ws_client.connect()
 
         logger.debug("WebSocket connection established")
+        success = True
         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)
+    finally:
+        if ws_client and not success:
+            ws_client.close()

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"):

scripts/test_sfu_connect.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""
+Utility script for testing SFU connection and retry behavior.
+
+Connects to a call as a given user and logs each step of the connection
+process — useful for verifying SFU assignment, retry on transient errors
+(e.g. SFU_FULL), and reassignment via the coordinator.
+
+Environment variables
+---------------------
+STREAM_API_KEY      — Stream API key (required)
+STREAM_API_SECRET   — Stream API secret (required)
+STREAM_BASE_URL     — Coordinator URL (default: Stream cloud).
+                      Set to http://127.0.0.1:3030 for a local coordinator.
+USER_ID             — User ID to join as (default: "test-user").
+CALL_TYPE           — Call type (default: "default").
+CALL_ID             — Call ID. If not set, a random UUID is generated.
+
+Usage
+-----
+    # Connect via cloud coordinator
+    STREAM_API_KEY=... STREAM_API_SECRET=... \\
+        uv run --extra webrtc python scripts/test_sfu_connect.py
+
+    # Connect via local coordinator
+    STREAM_BASE_URL=http://127.0.0.1:3030 \\
+        uv run --extra webrtc python scripts/test_sfu_connect.py
+"""
+
+import asyncio
+import logging
+import os
+import uuid
+
+from dotenv import load_dotenv
+
+from getstream import AsyncStream
+from getstream.models import CallRequest
+from getstream.video.rtc import ConnectionManager
+
+load_dotenv()
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+async def run():
+    base_url = os.getenv("STREAM_BASE_URL")
+    user_id = os.getenv("USER_ID", "test-user")
+    call_type = os.getenv("CALL_TYPE", "default")
+    call_id = os.getenv("CALL_ID", str(uuid.uuid4()))
+
+    logger.info("Configuration:")
+    logger.info(f"  Coordinator: {base_url or 'cloud (default)'}")
+    logger.info(f"  User:        {user_id}")
+    logger.info(f"  Call:        {call_type}:{call_id}")
+
+    client_kwargs = {}
+    if base_url:
+        client_kwargs["base_url"] = base_url
+
+    client = AsyncStream(timeout=10.0, **client_kwargs)
+
+    call = client.video.call(call_type, call_id)
+    logger.info("Creating call...")
+    await call.get_or_create(data=CallRequest(created_by_id=user_id))
+    logger.info("Call created")
+
+    cm = ConnectionManager(
+        call=call,
+        user_id=user_id,
+        create=False,
+    )
+
+    logger.info("Connecting to SFU...")
+
+    async with cm:
+        join = cm.join_response
+        if join and join.credentials:
+            logger.info(f"Connected to SFU: {join.credentials.server.edge_name}")
+            logger.info(f"  WS endpoint:  {join.credentials.server.ws_endpoint}")
+        logger.info(f"  Session ID:   {cm.session_id}")
+
+        logger.info("Holding connection for 3s...")
+        await asyncio.sleep(3)
+
+        logger.info("Leaving call")
+
+    logger.info("Done")
+
+
+if __name__ == "__main__":
+    asyncio.run(run())