From 2ff1df3d5620807c5da9656b451f0111789fd35d Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Fri, 11 Sep 2026 13:24:54 -0400 Subject: [PATCH 1/2] failover: retry the same host when no fallback origin exists A retryable failure with no untried origin used to be surfaced after a single attempt. Retry it against the same origin instead, bounded by the existing attempt count and backoff. This matches the cross-region path, which already retries both transport errors and 5xx responses. --- livekit-api/livekit/api/twirp_client.py | 9 ++- tests/api/test_failover_unit.py | 89 +++++++++++++++++++++++++ 2 files changed, 95 insertions(+), 3 deletions(-) create mode 100644 tests/api/test_failover_unit.py diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index 228c6d1d..21bb5c7f 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/livekit-api/livekit/api/twirp_client.py @@ -264,9 +264,12 @@ async def request( next_origin = pick_next(region_origins, attempted) if next_origin is None: - if transport_exc is not None: - raise transport_exc - raise self._server_error(error_data, retryable_status or 500) + if is_last: + if transport_exc is not None: + raise transport_exc + raise self._server_error(error_data, retryable_status or 500) + # With no fallback origin, a retryable failure is retried against the same host. + next_origin = current_origin reason = transport_exc if transport_exc is not None else f"status {retryable_status}" logger.warning( diff --git a/tests/api/test_failover_unit.py b/tests/api/test_failover_unit.py new file mode 100644 index 00000000..d4592d21 --- /dev/null +++ b/tests/api/test_failover_unit.py @@ -0,0 +1,89 @@ +# Copyright 2026 LiveKit, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Failover tests that need no external mock server: the attempts policy, and +the retry loop against an in-process aiohttp server that has no fallback +regions (``/settings/regions`` is 404, as it is for cloud-api).""" + +import asyncio +from typing import Callable, List + +import aiohttp +from aiohttp import web +from aiohttp.test_utils import TestServer + +from livekit.api import CreateRoomRequest, Room +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)`` + returns a response, or None to drop the connection (a transport error with + no HTTP response).""" + + async def twirp(request: web.Request) -> web.StreamResponse: + attempt = len(attempts) + attempts.append(attempt) + resp = behave(attempt, request) + if resp is None: + assert request.transport is not None + request.transport.close() + raise web.HTTPServiceUnavailable() + return resp + + app = web.Application() + app.router.add_post("/twirp/livekit.RoomService/CreateRoom", twirp) + async with TestServer(app) as server: + async with aiohttp.ClientSession() as session: + client = TwirpClient( + session, + str(server.make_url("")), + "livekit", + _failover_force=True, + _failover_backoff=0.001, + ) + return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), {}, Room) + + +def _ok(request: web.Request) -> web.Response: + return web.Response(body=Room(name="r").SerializeToString()) + + +def test_retries_same_host_on_transport_error(): + """Without a fallback origin, a transport error retries the same host.""" + + def behave(attempt: int, request: web.Request): + return None if attempt == 0 else _ok(request) + + attempts: List[int] = [] + room = asyncio.run(_call_single_host(behave, attempts)) + assert room.name == "r" + assert len(attempts) == 2 + + +def test_retries_same_host_on_5xx(): + """Without a fallback origin, a 5xx retries the same host.""" + + def behave(attempt: int, request: web.Request): + if attempt == 0: + return web.json_response({"code": "unavailable", "msg": "down"}, status=502) + return _ok(request) + + attempts: List[int] = [] + room = asyncio.run(_call_single_host(behave, attempts)) + assert room.name == "r" + assert len(attempts) == 2 From 8448cc30981f6aa035b185e48895d777432b32a3 Mon Sep 17 00:00:00 2001 From: Jason Lernerman Date: Fri, 11 Sep 2026 13:25:32 -0400 Subject: [PATCH 2/2] failover: enable retries for the LiveKit Cloud API hosts --- livekit-api/livekit/api/_failover.py | 14 ++-- livekit-api/livekit/api/twirp_client.py | 4 +- tests/api/test_failover_unit.py | 87 +++++++++++++++++++++++-- 3 files changed, 93 insertions(+), 12 deletions(-) diff --git a/livekit-api/livekit/api/_failover.py b/livekit-api/livekit/api/_failover.py index 4babb74c..bd72de24 100644 --- a/livekit-api/livekit/api/_failover.py +++ b/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..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"): diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index 21bb5c7f..f68325dc 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/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): diff --git a/tests/api/test_failover_unit.py b/tests/api/test_failover_unit.py index d4592d21..898e7cf0 100644 --- a/tests/api/test_failover_unit.py +++ b/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