From 4c92664506031b1e39741d8cf079380526cec5b1 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Sat, 22 Aug 2026 22:07:55 +0800 Subject: [PATCH 1/2] Generate request-id for api call relate to https://github.com/livekit/server-sdk-go/pull/954 --- livekit-api/livekit/api/twirp_client.py | 8 ++ tests/api/test_request_id.py | 141 ++++++++++++++++++++++++ 2 files changed, 149 insertions(+) create mode 100644 tests/api/test_request_id.py diff --git a/livekit-api/livekit/api/twirp_client.py b/livekit-api/livekit/api/twirp_client.py index a4c71e9f..1d258ec9 100644 --- a/livekit-api/livekit/api/twirp_client.py +++ b/livekit-api/livekit/api/twirp_client.py @@ -14,6 +14,7 @@ import asyncio import logging +import uuid from typing import Dict, List, Optional, Type, TypeVar import aiohttp @@ -37,6 +38,11 @@ # Identifies the SDK and version to the server on every request. _USER_AGENT = f"livekit-server-sdk-python/{__version__}" +# Carries a per-request idempotency key. The SDK's auto-retries (see _failover) +# keep the same key across attempts, so the server can identify and deduplicate +# repeated requests. +REQUEST_ID_HEADER = "X-Livekit-Request-Id" + # Shared across all clients in the process so the region list is fetched once. _REGION_CACHE = RegionCache() @@ -207,6 +213,8 @@ async def request( headers["User-Agent"] = _USER_AGENT forward_headers = dict(headers) # for the discovery fetch (no content-type yet) headers["Content-Type"] = "application/protobuf" + if not any(h.lower() == REQUEST_ID_HEADER.lower() for h in headers): + headers[REQUEST_ID_HEADER] = str(uuid.uuid4()) serialized_data = data.SerializeToString() # The effective per-attempt timeout is the per-call override, or the diff --git a/tests/api/test_request_id.py b/tests/api/test_request_id.py new file mode 100644 index 00000000..db5e41a9 --- /dev/null +++ b/tests/api/test_request_id.py @@ -0,0 +1,141 @@ +# 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. + +"""Tests for the per-request idempotency key the client stamps on every API +request. These drive TwirpClient.request() against a fake aiohttp session so no +server is needed, and use the internal test-only failover knobs +(_failover_force/_failover_backoff) to exercise the retry path. +""" + +from __future__ import annotations + +import pytest + +from livekit.api import CreateRoomRequest, Room +from livekit.api.twirp_client import REQUEST_ID_HEADER, TwirpClient + +HOST = "https://primary.example.livekit.cloud" + + +class _FakeResponse: + def __init__( + self, + status: int, + *, + body: bytes = b"", + json_data: dict | None = None, + headers: dict[str, str] | None = None, + ) -> None: + self.status = status + self._body = body + self._json = json_data if json_data is not None else {} + self.headers = headers or {} + + async def read(self) -> bytes: + return self._body + + async def json(self) -> dict: + return self._json + + async def __aenter__(self) -> _FakeResponse: + return self + + async def __aexit__(self, *exc) -> None: + return None + + +class _FakeSession: + """Records the headers of every request; replays ``statuses`` in order for + the Twirp POSTs and serves ``regions`` from /settings/regions.""" + + timeout = None + + def __init__(self, statuses: list[int], regions: list[str] | None = None) -> None: + self.post_headers: list[dict[str, str]] = [] + self._statuses = list(statuses) + self._regions = regions or [] + + def post(self, url, headers=None, data=None, timeout=None) -> _FakeResponse: + self.post_headers.append(dict(headers or {})) + status = self._statuses.pop(0) if self._statuses else 200 + return _FakeResponse(status) + + def get(self, url, headers=None, timeout=None) -> _FakeResponse: + return _FakeResponse( + 200, + json_data={"regions": [{"url": u} for u in self._regions]}, + # Never cache, so each test discovers its own region list. + headers={"Cache-Control": "max-age=0"}, + ) + + +async def _call(client: TwirpClient, headers: dict[str, str]) -> Room: + return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), headers, Room) + + +def _request_ids(session: _FakeSession) -> list[str | None]: + return [h.get(REQUEST_ID_HEADER) for h in session.post_headers] + + +# The header lets the server dedup a request that the SDK replayed. +async def test_stamps_a_request_id(): + session = _FakeSession([200, 200]) + client = TwirpClient(session, HOST, "livekit", failover=False) # type: ignore[arg-type] + + await _call(client, {}) + await _call(client, {}) + + ids = _request_ids(session) + assert all(ids) + # A new logical call is a new request, so it gets its own id. + assert ids[0] != ids[1] + + +async def test_preserves_caller_request_id(): + session = _FakeSession([200]) + client = TwirpClient(session, HOST, "livekit", failover=False) # type: ignore[arg-type] + + # Matched case-insensitively, as HTTP header names are. + await _call(client, {"x-livekit-request-id": "caller-123"}) + + assert session.post_headers[0]["x-livekit-request-id"] == "caller-123" + assert REQUEST_ID_HEADER not in session.post_headers[0] + + +# The id is generated once per logical call, so every failover attempt must +# carry the same value. +async def test_same_request_id_across_failover_attempts(): + session = _FakeSession( + [503, 503, 200], + regions=["wss://r1.example.livekit.cloud", "wss://r2.example.livekit.cloud"], + ) + # _failover_force bypasses the cloud-host check; a zero backoff keeps it fast. + client = TwirpClient( + session, # type: ignore[arg-type] + HOST, + "livekit", + _failover_force=True, + _failover_backoff=0, + ) + + await _call(client, {}) + + ids = _request_ids(session) + assert len(ids) == 3 + assert ids[0] + assert len(set(ids)) == 1 + + +if __name__ == "__main__": + pytest.main([__file__]) From db1cecf588a9b72d43207cb0d77d7e69ec84cadd Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Sat, 22 Aug 2026 22:20:13 +0800 Subject: [PATCH 2/2] merge test --- tests/api/test_failover.py | 50 ++++++++++++- tests/api/test_request_id.py | 141 ----------------------------------- 2 files changed, 47 insertions(+), 144 deletions(-) delete mode 100644 tests/api/test_request_id.py diff --git a/tests/api/test_failover.py b/tests/api/test_failover.py index f9521c44..598ab6d4 100644 --- a/tests/api/test_failover.py +++ b/tests/api/test_failover.py @@ -28,12 +28,13 @@ import json import os import urllib.request +from typing import List, Optional import aiohttp import pytest from livekit.api import CreateRoomRequest, Room, ServerError -from livekit.api.twirp_client import TwirpClient +from livekit.api.twirp_client import REQUEST_ID_HEADER, TwirpClient BASE = os.getenv("LK_TEST_SERVER_URL", "http://127.0.0.1:9999") @@ -53,8 +54,15 @@ def _server_up() -> bool: # _failover_force bypasses the cloud-host check (the mock is on 127.0.0.1) and a # tiny backoff keeps the tests fast — both are internal, test-only knobs. -async def _call(mock: dict, *, failover: bool = True, force: bool = True) -> Room: - async with aiohttp.ClientSession() as session: +async def _call( + mock: dict, + *, + failover: bool = True, + force: bool = True, + extra_headers: Optional[dict] = None, + trace_configs: Optional[List[aiohttp.TraceConfig]] = None, +) -> Room: + async with aiohttp.ClientSession(trace_configs=trace_configs) as session: client = TwirpClient( session, BASE, @@ -67,6 +75,7 @@ async def _call(mock: dict, *, failover: bool = True, force: bool = True) -> Roo "authorization": "Bearer test-token", # These tests exercise failover, not authz; skip the mock's permission check. "X-Lk-Mock": json.dumps({"skipAuth": True, **mock}), + **(extra_headers or {}), } return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), headers, Room) @@ -113,3 +122,38 @@ def test_disabled(): # failover=False disables failover entirely. with pytest.raises(ServerError): asyncio.run(_call({"failRegions": [0]}, failover=False)) + + +# Records the request id header(s) the SDK put on the wire for each Twirp +# attempt. Region discovery is a separate request, so it is not recorded. +def _request_id_recorder(seen: List[List[str]]) -> aiohttp.TraceConfig: + trace = aiohttp.TraceConfig() + + async def on_request_start(_session, _ctx, params) -> None: + if not params.url.path.endswith("/settings/regions"): + seen.append(list(params.headers.getall(REQUEST_ID_HEADER, []))) + + trace.on_request_start.append(on_request_start) + return trace + + +def test_request_id_stable_across_attempts(): + # The id is generated once per logical call, so a replayed request carries + # the same idempotency key on every attempt and the server can dedup it. + seen: List[List[str]] = [] + asyncio.run(_call({"failRegions": [0, 1]}, trace_configs=[_request_id_recorder(seen)])) + assert len(seen) == 3 # primary + two fallbacks + assert all(len(ids) == 1 for ids in seen) # never duplicated + assert seen[0][0] + assert len({ids[0] for ids in seen}) == 1 + + +def test_request_id_unique_per_call(): + # A new logical call is a new request, so it gets its own id. + seen: List[List[str]] = [] + recorder = _request_id_recorder(seen) + asyncio.run(_call({}, trace_configs=[recorder])) + asyncio.run(_call({}, trace_configs=[recorder])) + assert len(seen) == 2 + assert seen[0][0] and seen[1][0] + assert seen[0][0] != seen[1][0] diff --git a/tests/api/test_request_id.py b/tests/api/test_request_id.py deleted file mode 100644 index db5e41a9..00000000 --- a/tests/api/test_request_id.py +++ /dev/null @@ -1,141 +0,0 @@ -# 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. - -"""Tests for the per-request idempotency key the client stamps on every API -request. These drive TwirpClient.request() against a fake aiohttp session so no -server is needed, and use the internal test-only failover knobs -(_failover_force/_failover_backoff) to exercise the retry path. -""" - -from __future__ import annotations - -import pytest - -from livekit.api import CreateRoomRequest, Room -from livekit.api.twirp_client import REQUEST_ID_HEADER, TwirpClient - -HOST = "https://primary.example.livekit.cloud" - - -class _FakeResponse: - def __init__( - self, - status: int, - *, - body: bytes = b"", - json_data: dict | None = None, - headers: dict[str, str] | None = None, - ) -> None: - self.status = status - self._body = body - self._json = json_data if json_data is not None else {} - self.headers = headers or {} - - async def read(self) -> bytes: - return self._body - - async def json(self) -> dict: - return self._json - - async def __aenter__(self) -> _FakeResponse: - return self - - async def __aexit__(self, *exc) -> None: - return None - - -class _FakeSession: - """Records the headers of every request; replays ``statuses`` in order for - the Twirp POSTs and serves ``regions`` from /settings/regions.""" - - timeout = None - - def __init__(self, statuses: list[int], regions: list[str] | None = None) -> None: - self.post_headers: list[dict[str, str]] = [] - self._statuses = list(statuses) - self._regions = regions or [] - - def post(self, url, headers=None, data=None, timeout=None) -> _FakeResponse: - self.post_headers.append(dict(headers or {})) - status = self._statuses.pop(0) if self._statuses else 200 - return _FakeResponse(status) - - def get(self, url, headers=None, timeout=None) -> _FakeResponse: - return _FakeResponse( - 200, - json_data={"regions": [{"url": u} for u in self._regions]}, - # Never cache, so each test discovers its own region list. - headers={"Cache-Control": "max-age=0"}, - ) - - -async def _call(client: TwirpClient, headers: dict[str, str]) -> Room: - return await client.request("RoomService", "CreateRoom", CreateRoomRequest(), headers, Room) - - -def _request_ids(session: _FakeSession) -> list[str | None]: - return [h.get(REQUEST_ID_HEADER) for h in session.post_headers] - - -# The header lets the server dedup a request that the SDK replayed. -async def test_stamps_a_request_id(): - session = _FakeSession([200, 200]) - client = TwirpClient(session, HOST, "livekit", failover=False) # type: ignore[arg-type] - - await _call(client, {}) - await _call(client, {}) - - ids = _request_ids(session) - assert all(ids) - # A new logical call is a new request, so it gets its own id. - assert ids[0] != ids[1] - - -async def test_preserves_caller_request_id(): - session = _FakeSession([200]) - client = TwirpClient(session, HOST, "livekit", failover=False) # type: ignore[arg-type] - - # Matched case-insensitively, as HTTP header names are. - await _call(client, {"x-livekit-request-id": "caller-123"}) - - assert session.post_headers[0]["x-livekit-request-id"] == "caller-123" - assert REQUEST_ID_HEADER not in session.post_headers[0] - - -# The id is generated once per logical call, so every failover attempt must -# carry the same value. -async def test_same_request_id_across_failover_attempts(): - session = _FakeSession( - [503, 503, 200], - regions=["wss://r1.example.livekit.cloud", "wss://r2.example.livekit.cloud"], - ) - # _failover_force bypasses the cloud-host check; a zero backoff keeps it fast. - client = TwirpClient( - session, # type: ignore[arg-type] - HOST, - "livekit", - _failover_force=True, - _failover_backoff=0, - ) - - await _call(client, {}) - - ids = _request_ids(session) - assert len(ids) == 3 - assert ids[0] - assert len(set(ids)) == 1 - - -if __name__ == "__main__": - pytest.main([__file__])