failover: enable retries for the LiveKit Cloud API hosts

Author: u9g

livekit-api/livekit/api/_failover.py

@@ -44,11 +44,11 @@ def failover_attempts(
     timeout: Optional[float] = None,
 ) -> int:
     """Total request attempts for a host; 1 means no failover. Failover only
-    engages when enabled, the host is a LiveKit Cloud domain, and the request
-    timeout is long enough to retry. ``force`` bypasses the cloud-host check and
-    is for internal testing only.
+    engages when enabled, the host is a LiveKit Cloud project or Cloud API
+    domain, and the request timeout is long enough to retry. ``force`` bypasses
+    the cloud-host check and is for internal testing only.
     """
-    if not (enabled and (force or (host is not None and is_cloud(host)))):
+    if not (enabled and (force or (host is not None and (is_cloud(host) or is_cloud_api(host))))):
         return 1
     if timeout is not None and 0 < timeout < MIN_FAILOVER_TIMEOUT:
         return 1
@@ -60,6 +60,12 @@ def is_cloud(host: str) -> bool:
     return host.endswith(".livekit.cloud")
 
 
+def is_cloud_api(host: str) -> bool:
+    # cloud-api.livekit.io or a cloud-api.<env>.livekit.io variant; hostnames are case-insensitive.
+    host = host.lower()
+    return host.startswith("cloud-api.") and host.endswith(".livekit.io")
+
+
 def to_http(url: str) -> str:
     """Normalizes a region URL to an http(s) scheme (ws -> http, wss -> https)."""
     if url.startswith("ws"):

livekit-api/livekit/api/twirp_client.py

@@ -26,6 +26,7 @@
     RegionCache,
     failover_attempts,
     host_key,
+    is_cloud_api,
     origin_of,
     pick_next,
 )
@@ -226,7 +227,8 @@ async def request(
             self._failover, host, self._failover_force, effective_timeout
         )
         attempted = {host_key(self._origin)}
-        region_origins: Optional[List[str]] = None
+        # A Cloud API host has a single origin; region discovery is never consulted.
+        region_origins: Optional[List[str]] = [] if host and is_cloud_api(host) else None
         current_origin = self._origin
 
         for attempt in range(max_attempts):

tests/api/test_failover_unit.py

@@ -17,24 +17,59 @@
 regions (``/settings/regions`` is 404, as it is for cloud-api)."""
 
 import asyncio
-from typing import Callable, List
+import socket
+from typing import Callable, List, Optional
 
 import aiohttp
+import pytest
 from aiohttp import web
 from aiohttp.test_utils import TestServer
 
 from livekit.api import CreateRoomRequest, Room
+from livekit.api._failover import FAILOVER_MAX_ATTEMPTS, failover_attempts
 from livekit.api.twirp_client import TwirpClient
 
 Handler = Callable[[int, web.Request], "web.StreamResponse | None"]
 
 
-async def _call_single_host(behave: Handler, attempts: List[int]) -> Room:
-    """Runs one CreateRoom against a server whose only origin is itself and
-    appends each attempt index to ``attempts``. ``behave(attempt, request)``
+class _StaticResolver(aiohttp.abc.AbstractResolver):
+    """Resolves every hostname to the loopback address so a test server can be
+    reached under an arbitrary name."""
+
+    async def resolve(self, host: str, port: int = 0, family: int = socket.AF_INET) -> list:
+        return [
+            {
+                "hostname": host,
+                "host": "127.0.0.1",
+                "port": port,
+                "family": socket.AF_INET,
+                "proto": 0,
+                "flags": 0,
+            }
+        ]
+
+    async def close(self) -> None:
+        pass
+
+
+async def _call_single_host(
+    behave: Handler,
+    attempts: List[int],
+    *,
+    host: str = "127.0.0.1",
+    discovery_hits: Optional[List[None]] = None,
+) -> Room:
+    """Runs one CreateRoom against a server, reached as ``host``, whose only
+    origin is itself; appends each attempt index to ``attempts`` and each
+    ``/settings/regions`` hit to ``discovery_hits``. ``behave(attempt, request)``
     returns a response, or None to drop the connection (a transport error with
     no HTTP response)."""
 
+    async def regions(request: web.Request) -> web.StreamResponse:
+        if discovery_hits is not None:
+            discovery_hits.append(None)
+        raise web.HTTPNotFound()
+
     async def twirp(request: web.Request) -> web.StreamResponse:
         attempt = len(attempts)
         attempts.append(attempt)
@@ -47,13 +82,15 @@ async def twirp(request: web.Request) -> web.StreamResponse:
 
     app = web.Application()
     app.router.add_post("/twirp/livekit.RoomService/CreateRoom", twirp)
+    app.router.add_get("/settings/regions", regions)
     async with TestServer(app) as server:
-        async with aiohttp.ClientSession() as session:
+        connector = aiohttp.TCPConnector(resolver=_StaticResolver())
+        async with aiohttp.ClientSession(connector=connector) as session:
             client = TwirpClient(
                 session,
-                str(server.make_url("")),
+                f"http://{host}:{server.port}",
                 "livekit",
-                _failover_force=True,
+                _failover_force=host == "127.0.0.1",
                 _failover_backoff=0.001,
             )
             return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), {}, Room)
@@ -75,6 +112,22 @@ def behave(attempt: int, request: web.Request):
     assert len(attempts) == 2
 
 
+def test_cloud_api_host_never_consults_region_discovery():
+    """A Cloud API host retries the same host without any /settings/regions request."""
+
+    def behave(attempt: int, request: web.Request):
+        return None if attempt == 0 else _ok(request)
+
+    attempts: List[int] = []
+    hits: List[None] = []
+    room = asyncio.run(
+        _call_single_host(behave, attempts, host="cloud-api.livekit.io", discovery_hits=hits)
+    )
+    assert room.name == "r"
+    assert len(attempts) == 2
+    assert hits == []
+
+
 def test_retries_same_host_on_5xx():
     """Without a fallback origin, a 5xx retries the same host."""
 
@@ -87,3 +140,23 @@ def behave(attempt: int, request: web.Request):
     room = asyncio.run(_call_single_host(behave, attempts))
     assert room.name == "r"
     assert len(attempts) == 2
+
+
+@pytest.mark.parametrize(
+    "host, expected",
+    [
+        ("myproject.livekit.cloud", FAILOVER_MAX_ATTEMPTS),
+        ("myproject.region.livekit.cloud", FAILOVER_MAX_ATTEMPTS),
+        ("myproject.livekit.io", 1),
+        # The LiveKit Cloud API hosts fail over too (same-host retry).
+        ("cloud-api.livekit.io", FAILOVER_MAX_ATTEMPTS),
+        ("cloud-api.staging.livekit.io", FAILOVER_MAX_ATTEMPTS),
+        ("CLOUD-API.LIVEKIT.IO", FAILOVER_MAX_ATTEMPTS),
+        ("cloud-api.example.com", 1),
+        ("example.com", 1),
+        ("127.0.0.1", 1),
+        ("notlivekit.cloud", 1),
+    ],
+)
+def test_failover_attempts(host: str, expected: int):
+    assert failover_attempts(True, host) == expected