From d812086d2b045949ec8b19dc237052cdfaac9883 Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Thu, 15 Oct 2026 23:23:34 -0400 Subject: [PATCH 1/2] fix(api): add full jitter to the failover retry backoff Every client that failed together retried at exactly base * 2^attempt, so a shared 5xx event produced synchronized retry waves. Draw each delay uniformly from [0, base * 2^attempt) instead. --- livekit-api/livekit/api/_failover.py | 9 ++++++++- livekit-api/livekit/api/twirp_client.py | 7 ++++--- tests/api/test_failover_unit.py | 12 +++++++++++- 3 files changed, 23 insertions(+), 5 deletions(-) diff --git a/livekit-api/livekit/api/_failover.py b/livekit-api/livekit/api/_failover.py index bd72de24..2bda197d 100644 --- a/livekit-api/livekit/api/_failover.py +++ b/livekit-api/livekit/api/_failover.py @@ -16,12 +16,13 @@ On a retryable failure (any transport error or HTTP 5xx) 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 dataclasses import dataclass from typing import Dict, List, Optional @@ -66,6 +67,12 @@ 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 to_http(url: str) -> str: """Normalizes a region URL to an http(s) scheme (ws -> http, wss -> https).""" if url.startswith("ws"): diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index 4fdb2735..680a417f 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/livekit-api/livekit/api/twirp_client.py @@ -24,6 +24,7 @@ from ._failover import ( FAILOVER_BACKOFF_BASE, RegionCache, + backoff, failover_attempts, host_key, is_cloud_api, @@ -227,8 +228,8 @@ async def request( """Issues a Twirp request, failing over to alternative regions on retryable errors. On any transport error or HTTP 5xx 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 @@ -303,7 +304,7 @@ async def request( reason, next_origin, ) - await asyncio.sleep(self._failover_backoff * (2**attempt)) + await asyncio.sleep(backoff(attempt, self._failover_backoff)) attempted.add(host_key(next_origin)) current_origin = next_origin current_base = next_origin diff --git a/tests/api/test_failover_unit.py b/tests/api/test_failover_unit.py index 898e7cf0..f44410aa 100644 --- a/tests/api/test_failover_unit.py +++ b/tests/api/test_failover_unit.py @@ -26,7 +26,7 @@ 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._failover import FAILOVER_MAX_ATTEMPTS, backoff, failover_attempts from livekit.api.twirp_client import TwirpClient Handler = Callable[[int, web.Request], "web.StreamResponse | None"] @@ -160,3 +160,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) From 92404c0974a08380d2cc26d72199f37505b6eed1 Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Thu, 15 Oct 2026 23:24:17 -0400 Subject: [PATCH 2/2] fix(api): retry a 503 only with Retry-After, and honor Retry-After A 503 means the server is shedding load, so blind retries add to the overload. Retry it only when the server says when via Retry-After. Any retried response that carries Retry-After waits that long instead of the jittered backoff. --- livekit-api/livekit/api/_failover.py | 18 ++++++++- livekit-api/livekit/api/twirp_client.py | 15 ++++++-- tests/api/test_failover.py | 3 +- tests/api/test_failover_unit.py | 49 ++++++++++++++++++++++++- 4 files changed, 77 insertions(+), 8 deletions(-) diff --git a/livekit-api/livekit/api/_failover.py b/livekit-api/livekit/api/_failover.py index 2bda197d..857aea52 100644 --- a/livekit-api/livekit/api/_failover.py +++ b/livekit-api/livekit/api/_failover.py @@ -14,7 +14,8 @@ """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 jittered exponential backoff. 4xx responses are returned immediately. @@ -24,6 +25,7 @@ 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 @@ -73,6 +75,20 @@ def backoff(attempt: int, base: float) -> float: 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"): diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index 680a417f..8148aa93 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/livekit-api/livekit/api/twirp_client.py @@ -29,6 +29,7 @@ host_key, is_cloud_api, origin_of, + parse_retry_after, pick_next, ) from .version import __version__ @@ -226,7 +227,8 @@ 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 jittered exponential backoff. A 4xx is returned immediately.""" @@ -261,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( @@ -273,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: @@ -304,7 +309,9 @@ async def request( reason, next_origin, ) - await asyncio.sleep(backoff(attempt, self._failover_backoff)) + 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 diff --git a/tests/api/test_failover.py b/tests/api/test_failover.py index 598ab6d4..f8340800 100644 --- a/tests/api/test_failover.py +++ b/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) diff --git a/tests/api/test_failover_unit.py b/tests/api/test_failover_unit.py index f44410aa..6e217a90 100644 --- a/tests/api/test_failover_unit.py +++ b/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, backoff, 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", [