From 5e2fde5792355f36ca1858a7a4acf9482f832e91 Mon Sep 17 00:00:00 2001 From: davd-gzl <60177543+davd-gzl@users.noreply.github.com> Date: Tue, 29 Sep 2026 14:25:26 +0900 Subject: [PATCH] =?UTF-8?q?=F0=9F=90=9B(backend)=20put=20a=20timeout=20on?= =?UTF-8?q?=20every=20LiveKit=20API=20call?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The SDK passes timeout=None to aiohttp on every call, so a LiveKit that stops answering held a worker until the kernel gave up. A client middleware now bounds each request, its answer included, and the client closes the session it is given, which fixes #1648. --- CHANGELOG.md | 1 + docs/installation/kubernetes.md | 1 + .../core/recording/event/notification.py | 7 +- .../tests/test_settings_livekit_timeout.py | 14 ++ src/backend/core/tests/test_utils.py | 158 +++++++++++++----- src/backend/core/utils.py | 33 +++- src/backend/meet/settings.py | 6 + 7 files changed, 161 insertions(+), 59 deletions(-) create mode 100644 src/backend/core/tests/test_settings_livekit_timeout.py diff --git a/CHANGELOG.md b/CHANGELOG.md index d6f8f7eeb..eb7091b34 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -20,6 +20,7 @@ and this project adheres to - 🔒️(frontend) upgrade libx11 to fix CVE-2026-88806 - 🔒️(backend) throttle the start recording endpoint +- 🐛(backend) put a timeout on every LiveKit API call ## [1.35.0] - 2026-10-13 diff --git a/docs/installation/kubernetes.md b/docs/installation/kubernetes.md index 48bf7264e..1f729e8ae 100644 --- a/docs/installation/kubernetes.md +++ b/docs/installation/kubernetes.md @@ -405,6 +405,7 @@ These are the environmental options available on meet backend. | LIVEKIT_API_SECRET | LiveKit API secret | | | LIVEKIT_API_URL | LiveKit API URL | | | LIVEKIT_VERIFY_SSL | Verify SSL for LiveKit connections | true | +| LIVEKIT_API_TIMEOUT_SECONDS | Seconds before a call to the LiveKit API gives up | 10 | | LIVEKIT_FORCE_WSS_PROTOCOL | Enables WSS protocol conversion for legacy browser compatibility (Firefox <124, Chrome <125, Edge <125) where HTTPS URLs fail in WebSocket() constructor. | false | | LIVEKIT_ENABLE_FIREFOX_PROXY_WORKAROUND | Firefox-only connection warmup: pre-calls WebSocket endpoint (expecting 401) to initialize cache, resolving proxy/network connectivity issues. | false | | RESOURCE_DEFAULT_ACCESS_LEVEL | Default resource access level for rooms | public | diff --git a/src/backend/core/recording/event/notification.py b/src/backend/core/recording/event/notification.py index 54f1b119c..c59525613 100644 --- a/src/backend/core/recording/event/notification.py +++ b/src/backend/core/recording/event/notification.py @@ -10,7 +10,6 @@ from django.conf import settings from django.utils.translation import get_language, gettext, gettext_noop, override from django.utils.translation import gettext_lazy as _ -import aiohttp import requests from asgiref.sync import async_to_sync from celery.exceptions import OperationalError as CeleryOperationalError @@ -138,11 +137,7 @@ class NotificationService: if not worker_id: return None, None - custom_configuration = { - **settings.LIVEKIT_CONFIGURATION, - "timeout": aiohttp.ClientTimeout(total=10), - } - lkapi = utils.create_livekit_client(custom_configuration=custom_configuration) + lkapi = utils.create_livekit_client() try: egress_list = await lkapi.egress.list_egress( livekit_api.ListEgressRequest(egress_id=worker_id) # pylint: disable=no-member diff --git a/src/backend/core/tests/test_settings_livekit_timeout.py b/src/backend/core/tests/test_settings_livekit_timeout.py new file mode 100644 index 000000000..c70619cc9 --- /dev/null +++ b/src/backend/core/tests/test_settings_livekit_timeout.py @@ -0,0 +1,14 @@ +"""Unit tests for the LIVEKIT_API_TIMEOUT_SECONDS setting.""" + +from unittest import mock + +import pytest + +from meet.settings import Test + + +def test_livekit_api_timeout_zero_is_refused(): + """A zero timeout would fail every LiveKit call, so it stops the settings loading.""" + with mock.patch.object(Test, "LIVEKIT_API_TIMEOUT_SECONDS", 0): + with pytest.raises(ValueError, match="LIVEKIT_API_TIMEOUT_SECONDS"): + Test.post_setup() diff --git a/src/backend/core/tests/test_utils.py b/src/backend/core/tests/test_utils.py index db4cdb38a..db085cc9b 100644 --- a/src/backend/core/tests/test_utils.py +++ b/src/backend/core/tests/test_utils.py @@ -4,6 +4,8 @@ Test utils functions # pylint: disable=W0621 import json +import socket +import threading from unittest import mock from django.conf import settings @@ -11,9 +13,11 @@ from django.contrib.auth.models import AnonymousUser import jwt import pytest +from asgiref.sync import async_to_sync from livekit.api import TwirpError from core.factories import UserFactory +from core.services.room_management import RoomManagement from core.utils import ( NotificationError, create_livekit_client, @@ -107,59 +111,121 @@ def test_anonymous_falls_back_to_anonymous_label(): assert claims["name"] == "Anonymous" -@mock.patch("asyncio.get_running_loop") -@mock.patch("core.utils.LiveKitAPI") -def test_create_livekit_client_ssl_enabled( - mock_livekit_api, mock_get_running_loop, settings -): - """Test LiveKitAPI client creation with SSL verification enabled.""" - mock_get_running_loop.return_value = mock.MagicMock() - settings.LIVEKIT_VERIFY_SSL = True +@pytest.fixture +def fake_livekit(request): + """A LiveKit address that takes the connection, sends `request.param`, then waits.""" + server = socket.create_server(("127.0.0.1", 0)) + held = [] - create_livekit_client() + def accept(): + while True: + try: + connection, _ = server.accept() + except OSError: + return + held.append(connection) + connection.sendall(request.param) - mock_livekit_api.assert_called_once_with( - **settings.LIVEKIT_CONFIGURATION, session=None - ) + threading.Thread(target=accept, daemon=True).start() + yield f"http://127.0.0.1:{server.getsockname()[1]:d}" + server.close() + for connection in held: + connection.close() -@mock.patch("core.utils.aiohttp.ClientSession") -@mock.patch("asyncio.get_running_loop") -@mock.patch("core.utils.LiveKitAPI") -def test_create_livekit_client_ssl_disabled( - mock_livekit_api, mock_get_running_loop, mock_client_session, settings -): - """Test LiveKitAPI client creation with SSL verification disabled.""" - mock_get_running_loop.return_value = mock.MagicMock() - mock_session_instance = mock.MagicMock() - mock_client_session.return_value = mock_session_instance - settings.LIVEKIT_VERIFY_SSL = False - - create_livekit_client() - - mock_livekit_api.assert_called_once_with( - **settings.LIVEKIT_CONFIGURATION, session=mock_session_instance - ) - - -@mock.patch("asyncio.get_running_loop") -@mock.patch("core.utils.LiveKitAPI") -def test_create_livekit_client_custom_configuration( - mock_livekit_api, mock_get_running_loop, settings -): - """Test LiveKitAPI client creation with custom configuration.""" - settings.LIVEKIT_VERIFY_SSL = True - - mock_get_running_loop.return_value = mock.MagicMock() - custom_configuration = { - "api_key": "mock_key", - "api_secret": "mock_secret", - "url": "http://mock-url.com", +@pytest.mark.parametrize( + "fake_livekit", + [b"", b"HTTP/1.1 200 OK\r\nContent-Length: 64\r\n\r\n"], + ids=["no answer", "answer cut short"], + indirect=True, +) +def test_create_livekit_client_gives_up_on_a_quiet_livekit(fake_livekit, settings): + """A LiveKit that stops answering fails the call after the timeout.""" + settings.LIVEKIT_CONFIGURATION = { + **settings.LIVEKIT_CONFIGURATION, + "url": fake_livekit, } + settings.LIVEKIT_API_TIMEOUT_SECONDS = 1 - create_livekit_client(custom_configuration) + raised = [] - mock_livekit_api.assert_called_once_with(**custom_configuration, session=None) + def delete_room(): + try: + RoomManagement.delete_room("room-abc") + except TimeoutError as error: + raised.append(error) + + # A thread, so a call that never gives up fails the test rather than hangs it. + call = threading.Thread(target=delete_room, daemon=True) + call.start() + call.join(5) + + assert not call.is_alive() + assert len(raised) == 1 + + +@pytest.mark.parametrize( + "fake_livekit", + [b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\n\r\n"], + ids=["answer"], + indirect=True, +) +def test_create_livekit_client_lets_an_answer_through(fake_livekit, settings): + """A LiveKit that answers within the timeout is not cut off.""" + settings.LIVEKIT_CONFIGURATION = { + **settings.LIVEKIT_CONFIGURATION, + "url": fake_livekit, + } + settings.LIVEKIT_API_TIMEOUT_SECONDS = 1 + + RoomManagement.delete_room("room-abc") + + +@pytest.mark.parametrize("verify_ssl", [True, False]) +def test_create_livekit_client_closes_its_session(verify_ssl, settings): + """Closing the client closes its session, whether SSL is verified or not.""" + settings.LIVEKIT_VERIFY_SSL = verify_ssl + + @async_to_sync + async def open_and_close(): + client = create_livekit_client() + await client.aclose() + return client._session # pylint: disable=protected-access + + session = open_and_close() + + assert session.closed + assert session.connector is None + + +@pytest.mark.parametrize("verify_ssl", [True, False]) +def test_create_livekit_client_ssl(verify_ssl, settings): + """SSL is verified unless the setting turns it off.""" + settings.LIVEKIT_VERIFY_SSL = verify_ssl + + @async_to_sync + async def ssl_of_new_client(): + client = create_livekit_client() + ssl = client._session.connector._ssl # pylint: disable=protected-access + await client.aclose() + return ssl + + assert (ssl_of_new_client() is not False) is verify_ssl + + +def test_create_livekit_client_custom_configuration(): + """A custom configuration picks the LiveKit the client talks to.""" + + @async_to_sync + async def url_of_new_client(): + client = create_livekit_client( + {"api_key": "key", "api_secret": "secret", "url": "http://mock-url.com"} + ) + url = client.room._client.host # pylint: disable=protected-access + await client.aclose() + return url + + assert url_of_new_client().startswith("http://mock-url.com") @mock.patch("core.utils.create_livekit_client") diff --git a/src/backend/core/utils.py b/src/backend/core/utils.py index 434328e8e..b8bb9be7b 100644 --- a/src/backend/core/utils.py +++ b/src/backend/core/utils.py @@ -5,6 +5,7 @@ Utils functions used in the core app # pylint: disable=R0913, R0917 # ruff: noqa:S311, PLR0913 +import asyncio import hashlib import json import logging @@ -218,19 +219,37 @@ def generate_s3_authorization_headers(key): return request +async def bound_livekit_request(request, handler): + """Give up on a LiveKit request, its answer included, after the set timeout. + + The SDK passes timeout=None to aiohttp on every call, which means no limit, + so a timeout set on the session never applies. + """ + async with asyncio.timeout(settings.LIVEKIT_API_TIMEOUT_SECONDS): + response = await handler(request) + await response.read() + return response + + +class LiveKitClient(LiveKitAPI): + """A LiveKit API client that closes the session it is given.""" + + async def aclose(self): + """Close the session, which LiveKitAPI leaves open when it did not create it.""" + await self._session.close() + + def create_livekit_client(custom_configuration=None): """Create and return a configured LiveKit API client.""" - - custom_session = None - - if not settings.LIVEKIT_VERIFY_SSL: - connector = aiohttp.TCPConnector(ssl=False) - custom_session = aiohttp.ClientSession(connector=connector) + connector = None if settings.LIVEKIT_VERIFY_SSL else aiohttp.TCPConnector(ssl=False) + session = aiohttp.ClientSession( + connector=connector, middlewares=(bound_livekit_request,) + ) # Use default configuration if none provided configuration = custom_configuration or settings.LIVEKIT_CONFIGURATION - return LiveKitAPI(session=custom_session, **configuration) + return LiveKitClient(session=session, **configuration) class NotificationError(Exception): diff --git a/src/backend/meet/settings.py b/src/backend/meet/settings.py index 1f29edbfc..66b2bfc2c 100755 --- a/src/backend/meet/settings.py +++ b/src/backend/meet/settings.py @@ -798,6 +798,9 @@ class Base(Configuration): LIVEKIT_VERIFY_SSL = values.BooleanValue( True, environ_name="LIVEKIT_VERIFY_SSL", environ_prefix=None ) + LIVEKIT_API_TIMEOUT_SECONDS = values.PositiveIntegerValue( + 10, environ_name="LIVEKIT_API_TIMEOUT_SECONDS", environ_prefix=None + ) # Regex to filter webhook events by room name. Only matching events are processed. LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX = values.Value( None, environ_name="LIVEKIT_WEBHOOK_EVENTS_FILTER_REGEX", environ_prefix=None @@ -1556,6 +1559,9 @@ class Base(Configuration): cls._check_recording_encoding_maps() cls._check_audit_log_data_stream_namespace() + if cls.LIVEKIT_API_TIMEOUT_SECONDS < 1: + raise ValueError("LIVEKIT_API_TIMEOUT_SECONDS must be at least 1") + if ( cls.SUMMARY_SERVICE_VERSION == 1 and cls.SUMMARY_SERVICE_ENDPOINT is not None