fix(api): jitter the failover backoff, and retry a 503 only with Retry-After

livekit-api/livekit/api/_failover.py

@@ -14,15 +14,18 @@
 
 """Region failover for the Twirp API clients.
 
-On a retryable failure (any transport error or HTTP 5xx) the client discovers
+On a retryable failure (any transport error, or an HTTP 5xx other than a 503
+without Retry-After) the client discovers
 alternative LiveKit Cloud regions via ``/settings/regions`` and replays the
-request against the next region, with exponential backoff. 4xx responses are
+request against the next region, with jittered exponential backoff. 4xx responses are
 returned immediately.
 """
 
 from __future__ import annotations
 
+import random
 import time
+from email.utils import parsedate_to_datetime
 from dataclasses import dataclass
 from typing import Dict, List, Optional
 from urllib.parse import urlparse
@@ -66,6 +69,26 @@ def is_cloud_api(host: str) -> bool:
     return host.startswith("cloud-api.") and host.endswith(".livekit.io")
 
 
+def backoff(attempt: int, base: float) -> float:
+    """Full jitter: a uniform delay in [0, base * 2^attempt) so clients that
+    failed together don't retry in lockstep."""
+    return random.random() * base * (2**attempt)
+
+
+def parse_retry_after(header: Optional[str]) -> Optional[float]:
+    """Parses a Retry-After header (delay-seconds or an HTTP-date) to seconds."""
+    value = (header or "").strip()
+    if not value:
+        return None
+    if value.isdigit():
+        return float(value)
+    try:
+        date = parsedate_to_datetime(value)
+    except (TypeError, ValueError):
+        return None
+    return max(0.0, date.timestamp() - time.time())
+
+
 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

@@ -24,10 +24,12 @@
 from ._failover import (
     FAILOVER_BACKOFF_BASE,
     RegionCache,
+    backoff,
     failover_attempts,
     host_key,
     is_cloud_api,
     origin_of,
+    parse_retry_after,
     pick_next,
 )
 from .version import __version__
@@ -225,10 +227,11 @@ async def request(
         timeout: Optional[aiohttp.ClientTimeout] = None,
     ) -> T:
         """Issues a Twirp request, failing over to alternative regions on
-        retryable errors. On any transport error or HTTP 5xx it discovers
+        retryable errors. On any transport error or HTTP 5xx (a 503 only with
+        Retry-After, which then sets the delay) it discovers
         regions via ``/settings/regions`` and replays the request — body and
-        headers intact — against the next untried region, with exponential
-        backoff. A 4xx is returned immediately."""
+        headers intact — against the next untried region, with jittered
+        exponential backoff. A 4xx is returned immediately."""
         path = f"{self.prefix}/{self.pkg}.{service}/{method}"
         headers = dict(headers)
         headers["User-Agent"] = _USER_AGENT
@@ -260,6 +263,7 @@ async def request(
 
             transport_exc: Optional[BaseException] = None
             retryable_status: Optional[int] = None
+            retry_after: Optional[float] = None
             error_data: Dict = {}
             try:
                 async with self._session.post(
@@ -272,8 +276,10 @@ async def request(
                         error_data = await resp.json()
                     except Exception:
                         error_data = {}
-                    if resp.status < 500:
-                        # 4xx is terminal.
+                    retry_after = parse_retry_after(resp.headers.get("Retry-After"))
+                    # A 4xx is terminal. A 503 is the server shedding load, so
+                    # it is retried only when Retry-After says when.
+                    if resp.status < 500 or (resp.status == 503 and retry_after is None):
                         raise self._server_error(error_data, resp.status)
                     retryable_status = resp.status
             except (aiohttp.ClientError, asyncio.TimeoutError) as e:
@@ -303,7 +309,9 @@ async def request(
                 reason,
                 next_origin,
             )
-            await asyncio.sleep(self._failover_backoff * (2**attempt))
+            await asyncio.sleep(
+                retry_after if retry_after is not None else backoff(attempt, self._failover_backoff)
+            )
             attempted.add(host_key(next_origin))
             current_origin = next_origin
             current_base = next_origin

tests/api/test_failover.py

@@ -74,7 +74,8 @@ async def _call(
         headers = {
             "authorization": "Bearer test-token",
             # These tests exercise failover, not authz; skip the mock's permission check.
-            "X-Lk-Mock": json.dumps({"skipAuth": True, **mock}),
+            # The mock fails with a 503 by default, which is retried only with Retry-After.
+            "X-Lk-Mock": json.dumps({"skipAuth": True, "failStatus": 502, **mock}),
             **(extra_headers or {}),
         }
         return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), headers, Room)

tests/api/test_failover_unit.py

@@ -26,8 +26,13 @@
 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
+from livekit.api._failover import (
+    FAILOVER_MAX_ATTEMPTS,
+    backoff,
+    failover_attempts,
+    parse_retry_after,
+)
+from livekit.api.twirp_client import ServerError, TwirpClient
 
 Handler = Callable[[int, web.Request], "web.StreamResponse | None"]
 
@@ -142,6 +147,46 @@ def behave(attempt: int, request: web.Request):
     assert len(attempts) == 2
 
 
+def test_503_without_retry_after_not_retried():
+    """A 503 without Retry-After is the server shedding load, so it is terminal."""
+
+    def behave(attempt: int, request: web.Request):
+        return web.json_response({"code": "unavailable", "msg": "busy"}, status=503)
+
+    attempts: List[int] = []
+    with pytest.raises(ServerError):
+        asyncio.run(_call_single_host(behave, attempts))
+    assert len(attempts) == 1
+
+
+def test_503_with_retry_after_retried():
+    """A 503 with Retry-After is retried."""
+
+    def behave(attempt: int, request: web.Request):
+        if attempt == 0:
+            return web.json_response(
+                {"code": "unavailable", "msg": "busy"}, status=503, headers={"Retry-After": "0"}
+            )
+        return _ok(request)
+
+    attempts: List[int] = []
+    room = asyncio.run(_call_single_host(behave, attempts))
+    assert room.name == "r"
+    assert len(attempts) == 2
+
+
+def test_parse_retry_after(monkeypatch: pytest.MonkeyPatch):
+    """Retry-After is delay-seconds or an HTTP-date; anything else is ignored."""
+    assert parse_retry_after(None) is None
+    assert parse_retry_after("") is None
+    assert parse_retry_after("soon") is None
+    assert parse_retry_after("0") == 0.0
+    assert parse_retry_after(" 2 ") == 2.0
+    monkeypatch.setattr("time.time", lambda: 1792567680.0)  # Wed, 21 Oct 2026 07:28:00 GMT
+    assert parse_retry_after("Wed, 21 Oct 2026 07:28:03 GMT") == pytest.approx(3.0)
+    assert parse_retry_after("Wed, 21 Oct 2026 07:27:00 GMT") == 0.0
+
+
 @pytest.mark.parametrize(
     "host, expected",
     [
@@ -160,3 +205,13 @@ def behave(attempt: int, request: web.Request):
 )
 def test_failover_attempts(host: str, expected: int):
     assert failover_attempts(True, host) == expected
+
+
+def test_backoff_is_full_jitter(monkeypatch: pytest.MonkeyPatch):
+    """The delay is drawn uniformly from [0, base * 2^attempt)."""
+    monkeypatch.setattr("random.random", lambda: 0.0)
+    assert backoff(2, 0.2) == 0.0
+    monkeypatch.setattr("random.random", lambda: 0.5)
+    assert backoff(0, 0.2) == pytest.approx(0.1)
+    assert backoff(1, 0.2) == pytest.approx(0.2)
+    assert backoff(2, 0.2) == pytest.approx(0.4)