✨(backend) add connection-test API

Currently users have no way to reliably test their connection before
joining a room. To address this, we plan to build a connection-test
page.

The testing requires a dedicated LiveKit token, issued without going
through the room API, which is tied to registered meetings, lobby
rules, and longer-lived access tokens.

Introduce a new viewset for all diagnostics-related features. The
first route issues a token for diagnostics, even for anonymous
users. Each request creates a new dedicated room so users never
share the same LiveKit room during tests. Tokens are short-lived
(default 10 minutes) to limit reuse, and the endpoint is throttled
to prevent abuse.

A Celery worker also schedules a callback that deletes the room
after a certain delay, in every case.

src/backend/core/api/diagnostics.py

@@ -0,0 +1,77 @@
+"""Diagnostics API endpoints."""
+
+from datetime import timedelta
+from uuid import uuid4
+
+from django.conf import settings
+
+from rest_framework import decorators, permissions, viewsets
+from rest_framework import (
+    response as drf_response,
+)
+
+from core.api import throttling
+from core.tasks.connection_test import delete_connection_test_room
+from core.utils import generate_token
+
+CONNECTION_TEST_USERNAME = "Test connexion"
+
+
+class DiagnosticsViewSet(viewsets.ViewSet):
+    """Endpoints helping users and support diagnose connectivity issues.
+
+    Diagnostics are grouped behind a single prefix so upcoming checks
+    (rtcstats collection, ICE candidate reports, etc.) can be added as new
+    actions rather than new top-level routes.
+
+    They are open to anonymous users: someone who cannot join a room is
+    exactly who needs to run a test, and they may well not be logged in.
+    Each action therefore carries its own throttle scope.
+    """
+
+    permission_classes = [permissions.AllowAny]
+
+    @decorators.action(
+        detail=False,
+        methods=["get"],
+        url_path="connection",
+        url_name="connection",
+        throttle_classes=[
+            throttling.ConnectionTestUserRateThrottle,
+            throttling.ConnectionTestAnonRateThrottle,
+        ],
+    )
+    def connection(self, request):
+        """Return a short-lived LiveKit token for an ephemeral test room.
+
+        Going through the room API is not an option here: it is tied to
+        registered meetings, lobby rules and longer-lived tokens. Each call
+        gets its own room so two people testing at the same time never meet.
+        """
+        room = f"{settings.CONNECTION_TEST_ROOM_PREFIX}-{uuid4()}"
+        expires_in = settings.CONNECTION_TEST_TOKEN_TTL_SECONDS
+
+        # LiveKit refreshes tokens for connected clients, so JWT TTL alone does not
+        # eject someone who stays connected. Schedule a hard DeleteRoom when Celery
+        # is available.
+        if settings.CELERY_ENABLED:
+            delete_connection_test_room.apply_async(
+                args=[room],
+                countdown=settings.CONNECTION_TEST_ROOM_MAX_AGE_SECONDS,
+            )
+
+        return drf_response.Response(
+            {
+                "livekit": {
+                    "url": settings.LIVEKIT_CONFIGURATION["url"],
+                    "room": room,
+                    "token": generate_token(
+                        room=room,
+                        user=request.user,
+                        username=CONNECTION_TEST_USERNAME,
+                        ttl=timedelta(seconds=expires_in),
+                    ),
+                    "expires_in": expires_in,
+                },
+            }
+        )

src/backend/core/api/throttling.py

@@ -85,3 +85,15 @@ class RoomKitJoinRateThrottle(MonitoredUserRateThrottle):
     """
 
     scope = "roomkit_join"
+
+
+class ConnectionTestUserRateThrottle(MonitoredUserRateThrottle):
+    """Throttle authenticated users requesting connection test tokens."""
+
+    scope = "connection_test"
+
+
+class ConnectionTestAnonRateThrottle(MonitoredAnonRateThrottle):
+    """Throttle anonymous users requesting connection test tokens."""
+
+    scope = "connection_test"

src/backend/core/services/livekit_events.py

@@ -228,9 +228,21 @@ class LiveKitEventsService:
 
         # Silently ignoring EGRESS_ABORTED, EGRESS_FAILED
 
+    @staticmethod
+    def _is_connection_test_room(room_name: str) -> bool:
+        """Return True for ephemeral rooms created by the connection test endpoint."""
+        return room_name.startswith(settings.CONNECTION_TEST_ROOM_PREFIX)
+
     def _handle_room_started(self, data):
         """Handle 'room_started' event."""
 
+        if self._is_connection_test_room(data.room.name):
+            logger.info(
+                "Ignoring room_started event for connection test room '%s'.",
+                data.room.name,
+            )
+            return
+
         try:
             room_id = uuid.UUID(data.room.name)
         except ValueError as e:
@@ -256,6 +268,13 @@ class LiveKitEventsService:
     def _handle_room_finished(self, data):
         """Handle 'room_finished' event."""
 
+        if self._is_connection_test_room(data.room.name):
+            logger.info(
+                "Ignoring room_finished event for connection test room '%s'.",
+                data.room.name,
+            )
+            return
+
         try:
             room_id = uuid.UUID(data.room.name)
         except ValueError as e:

src/backend/core/tasks/__init__.py

@@ -0,0 +1,9 @@
+"""Celery tasks for the core app."""
+
+from core.tasks.connection_test import delete_connection_test_room
+from core.tasks.file import process_file_deletion
+
+__all__ = (
+    "delete_connection_test_room",
+    "process_file_deletion",
+)

src/backend/core/tasks/connection_test.py

@@ -0,0 +1,51 @@
+"""Tasks related to connection test rooms."""
+
+import logging
+
+from django.conf import settings
+
+from asgiref.sync import async_to_sync
+from livekit.api import (  # pylint: disable=no-name-in-module
+    DeleteRoomRequest,
+    TwirpError,
+)
+
+from core.tasks._task import task
+from core.utils import create_livekit_client
+
+logger = logging.getLogger(__name__)
+
+
+@task
+def delete_connection_test_room(room_name: str):
+    """Force-delete an ephemeral connection-test room.
+
+    Used as a hard cap so a participant cannot keep an auto-refreshed
+    LiveKit session open indefinitely after requesting a test token.
+    """
+    prefix = settings.CONNECTION_TEST_ROOM_PREFIX
+    if not room_name.startswith(prefix):
+        logger.error(
+            "Refusing to delete room '%s': expected prefix '%s'.",
+            room_name,
+            prefix,
+        )
+        return
+
+    async_to_sync(_delete_room)(room_name)
+
+
+async def _delete_room(room_name: str):
+    lkapi = create_livekit_client()
+    try:
+        await lkapi.room.delete_room(DeleteRoomRequest(room=room_name))
+        logger.info("Deleted connection test room '%s'.", room_name)
+    except TwirpError as exc:
+        # Room may already be gone after empty/departure timeout.
+        logger.info(
+            "Could not delete connection test room '%s': %s",
+            room_name,
+            exc,
+        )
+    finally:
+        await lkapi.aclose()

src/backend/core/tests/services/test_livekit_events.py

@@ -6,6 +6,8 @@ Test LiveKitEvents service.
 import uuid
 from unittest import mock
 
+from django.test.utils import override_settings
+
 import pytest
 from livekit.api import EgressStatus
 
@@ -575,6 +577,23 @@ def test_handle_room_finished_raises_error_when_telephony_deletion_fails(
     mock_clear_cache.assert_not_called()
 
 
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test")
+@mock.patch.object(LobbyService, "clear_room_cache")
+@mock.patch.object(SIPManagement, "delete_dispatch_rule")
+def test_handle_room_finished_ignores_connection_test_room(
+    mock_delete_dispatch_rule, mock_clear_cache, service, settings
+):
+    """Should ignore room_finished events for connection test rooms."""
+    settings.ROOM_TELEPHONY_ENABLED = True
+    mock_data = mock.MagicMock()
+    mock_data.room.name = f"{settings.CONNECTION_TEST_ROOM_PREFIX}{uuid.uuid4()}"
+
+    service._handle_room_finished(mock_data)
+
+    mock_delete_dispatch_rule.assert_not_called()
+    mock_clear_cache.assert_not_called()
+
+
 def test_handle_room_finished_raises_error_for_invalid_room_name(service):
     """Should raise ActionFailedError when room name format is invalid when room finishes."""
     mock_data = mock.MagicMock()
@@ -678,6 +697,15 @@ def test_handle_room_started_raises_error_for_invalid_room_name(service):
         service._handle_room_started(mock_data)
 
 
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test-")
+def test_handle_room_started_ignores_connection_test_room(service, settings):
+    """Should ignore room_started events for connection test rooms."""
+    mock_data = mock.MagicMock()
+    mock_data.room.name = f"{settings.CONNECTION_TEST_ROOM_PREFIX}{uuid.uuid4()}"
+
+    service._handle_room_started(mock_data)
+
+
 def test_handle_room_started_raises_error_for_nonexistent_room(service):
     """Should raise ActionFailedError when a room starts that doesn't exist in the database."""
     mock_data = mock.MagicMock()

src/backend/core/tests/tasks/test_connection_test.py

@@ -0,0 +1,51 @@
+"""Tests for connection test Celery tasks."""
+
+from unittest import mock
+
+from django.test.utils import override_settings
+
+from livekit.api import TwirpError
+
+from core.tasks.connection_test import delete_connection_test_room
+
+
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test")
+@mock.patch("core.tasks.connection_test.create_livekit_client")
+def test_delete_connection_test_room_calls_livekit(mock_create_livekit_client):
+    """DeleteRoom is called for rooms with the connection-test prefix."""
+    mock_api = mock.MagicMock()
+    mock_api.room.delete_room = mock.AsyncMock()
+    mock_api.aclose = mock.AsyncMock()
+    mock_create_livekit_client.return_value = mock_api
+
+    delete_connection_test_room("connection-test-abc")
+
+    mock_api.room.delete_room.assert_awaited_once()
+    request = mock_api.room.delete_room.await_args.args[0]
+    assert request.room == "connection-test-abc"
+    mock_api.aclose.assert_awaited_once()
+
+
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test")
+@mock.patch("core.tasks.connection_test.create_livekit_client")
+def test_delete_connection_test_room_refuses_other_rooms(mock_create_livekit_client):
+    """Refuse to delete rooms outside the connection-test namespace."""
+    delete_connection_test_room("production-room")
+
+    mock_create_livekit_client.assert_not_called()
+
+
+@override_settings(CONNECTION_TEST_ROOM_PREFIX="connection-test")
+@mock.patch("core.tasks.connection_test.create_livekit_client")
+def test_delete_connection_test_room_ignores_missing_room(mock_create_livekit_client):
+    """Missing rooms are treated as already cleaned up."""
+    mock_api = mock.MagicMock()
+    mock_api.room.delete_room = mock.AsyncMock(
+        side_effect=TwirpError("not_found", "room not found", status=404)
+    )
+    mock_api.aclose = mock.AsyncMock()
+    mock_create_livekit_client.return_value = mock_api
+
+    delete_connection_test_room("connection-test-gone")
+
+    mock_api.aclose.assert_awaited_once()

src/backend/core/tests/test_api_diagnostics.py

@@ -0,0 +1,159 @@
+"""Test diagnostics API endpoints."""
+
+import uuid
+from unittest import mock
+
+from django.test.utils import override_settings
+from django.urls import reverse
+
+import jwt
+import pytest
+from rest_framework.test import APIClient
+
+from core.api.diagnostics import CONNECTION_TEST_USERNAME
+from core.api.throttling import (
+    ConnectionTestAnonRateThrottle,
+    ConnectionTestUserRateThrottle,
+)
+from core.factories import UserFactory
+
+pytestmark = pytest.mark.django_db
+
+CONNECTION_URL = "/api/v1.0/diagnostics/connection/"
+
+
+def test_api_diagnostics_connection_url():
+    """The connection check is exposed under the diagnostics namespace."""
+    assert reverse("diagnostics-connection") == CONNECTION_URL
+
+
+def test_api_diagnostics_connection_rejects_post():
+    """Only GET is exposed, the endpoint has no side effect to trigger."""
+    client = APIClient()
+    response = client.post(CONNECTION_URL)
+
+    assert response.status_code == 405
+
+
+def test_api_diagnostics_connection_returns_ephemeral_livekit_config(settings, client):
+    """Each request gets a dedicated room and a short-lived token."""
+
+    settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 600
+    settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
+
+    response_a = client.get(CONNECTION_URL)
+    response_b = client.get(CONNECTION_URL)
+
+    assert response_a.status_code == 200
+    assert response_b.status_code == 200
+
+    data_a = response_a.json()
+    data_b = response_b.json()
+
+    room_a = data_a["livekit"]["room"]
+    room_b = data_b["livekit"]["room"]
+
+    assert room_a.startswith("connection-test-")
+    assert room_b.startswith("connection-test-")
+    uuid.UUID(room_a.removeprefix("connection-test-"))
+    uuid.UUID(room_b.removeprefix("connection-test-"))
+    assert room_a != room_b
+    assert data_a["livekit"]["url"]
+    assert data_a["livekit"]["token"]
+    assert data_a["livekit"]["expires_in"] == 600
+    assert data_a["livekit"]["token"] != data_b["livekit"]["token"]
+
+
+def test_api_diagnostics_connection_token_is_short_lived_for_user(settings, client):
+    """Connection test tokens expire quickly for users."""
+
+    settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 300
+
+    client = APIClient()
+    response = client.get(CONNECTION_URL)
+
+    assert response.status_code == 200
+
+    config = response.json()["livekit"]
+    payload = jwt.decode(
+        config["token"],
+        settings.LIVEKIT_CONFIGURATION["api_secret"],
+        algorithms=["HS256"],
+        options={"verify_exp": False},
+    )
+
+    assert config["expires_in"] == 300
+    assert payload["video"]["room"] == config["room"]
+    assert payload["name"] == CONNECTION_TEST_USERNAME
+    assert payload["video"]["roomAdmin"] is False
+    assert payload["exp"] - payload["nbf"] == 300
+
+
+@override_settings()
+def test_api_diagnostics_connection_token_for_authenticated_user(settings, client):
+    """Logged-in users get a token bound to their own identity."""
+
+    settings.CONNECTION_TEST_TOKEN_TTL_SECONDS = 300
+
+    user = UserFactory()
+    client.force_login(user)
+
+    response = client.get(CONNECTION_URL)
+
+    assert response.status_code == 200
+
+    payload = jwt.decode(
+        response.json()["livekit"]["token"],
+        settings.LIVEKIT_CONFIGURATION["api_secret"],
+        algorithms=["HS256"],
+        options={"verify_exp": False},
+    )
+
+    assert payload["sub"] == str(user.sub)
+    assert payload["video"]["roomAdmin"] is False
+    assert payload["exp"] - payload["nbf"] == 300
+
+
+@mock.patch("core.api.diagnostics.delete_connection_test_room.apply_async")
+def test_api_diagnostics_connection_schedules_room_deletion(
+    mock_apply_async, settings, client
+):
+    """When Celery is enabled, schedule a hard room delete after max age."""
+
+    settings.CELERY_ENABLED = True
+    settings.CONNECTION_TEST_ROOM_MAX_AGE_SECONDS = 300
+    settings.CONNECTION_TEST_ROOM_PREFIX = "connection-test"
+
+    response = client.get(CONNECTION_URL)
+
+    assert response.status_code == 200
+    room = response.json()["livekit"]["room"]
+    mock_apply_async.assert_called_once_with(args=[room], countdown=300)
+
+
+@mock.patch("core.api.diagnostics.delete_connection_test_room.apply_async")
+def test_api_diagnostics_connection_skips_room_deletion_without_celery(
+    mock_apply_async, settings, client
+):
+    """Without Celery, do not schedule deletion (apply_async would run immediately)."""
+
+    settings.CELERY_ENABLED = False
+    response = client.get(CONNECTION_URL)
+
+    assert response.status_code == 200
+    mock_apply_async.assert_not_called()
+
+
+@pytest.mark.parametrize(
+    "throttle_class",
+    [ConnectionTestAnonRateThrottle, ConnectionTestUserRateThrottle],
+)
+def test_api_diagnostics_connection_is_throttled(throttle_class, client):
+    """Both throttles stay wired to the action once routed through the viewset."""
+    with (
+        mock.patch.object(throttle_class, "allow_request", return_value=False),
+        mock.patch.object(throttle_class, "wait", return_value=42),
+    ):
+        response = client.get(CONNECTION_URL)
+
+    assert response.status_code == 429

src/backend/core/urls.py

@@ -7,7 +7,7 @@ from lasuite.oidc_login.urls import urlpatterns as oidc_urls
 from rest_framework.routers import DefaultRouter, SimpleRouter
 
 from core.addons import viewsets as addons_viewsets
-from core.api import get_frontend_configuration, viewsets
+from core.api import diagnostics, get_frontend_configuration, viewsets
 from core.external_api import viewsets as external_viewsets
 from core.roomkit import viewsets as roomkit_viewsets
 
@@ -30,6 +30,11 @@ router.register(
     addons_viewsets.SessionViewSet,
     basename="addons_sessions",
 )
+router.register(
+    "diagnostics",
+    diagnostics.DiagnosticsViewSet,
+    basename="diagnostics",
+)
 
 # - External API
 external_router = SimpleRouter()

src/backend/core/utils.py

@@ -12,6 +12,7 @@ import mimetypes
 import random
 import secrets
 import string
+from datetime import timedelta
 from functools import lru_cache
 from typing import List, Optional
 from uuid import uuid4
@@ -67,6 +68,7 @@ def generate_token(  # noqa: PLR0917
     sources: Optional[List[str]] = None,
     role: Optional[str] = None,
     participant_id: Optional[str] = None,
+    ttl: Optional[timedelta] = None,
 ) -> str:
     """Generate a LiveKit access token for a user in a specific room.
 
@@ -82,6 +84,7 @@ def generate_token(  # noqa: PLR0917
         role (Optional[str]): Room's access role if any
         participant_id (Optional[str]): Stable identifier for anonymous users;
                          used as identity when user.is_anonymous.
+        ttl (Optional[timedelta]): Token validity duration. Defaults to LiveKit SDK default.
 
     Returns:
         str: The LiveKit JWT access token.
@@ -135,6 +138,8 @@ def generate_token(  # noqa: PLR0917
             }
         )
     )
+    if ttl is not None:
+        token = token.with_ttl(ttl)
 
     return token.to_jwt()
 

src/backend/meet/settings.py

@@ -354,6 +354,11 @@ class Base(Configuration):
                 environ_name="ROOMKIT_JOIN_THROTTLE_RATES",
                 environ_prefix=None,
             ),
+            "connection_test": values.Value(
+                default="30/minute",
+                environ_name="CONNECTION_TEST_THROTTLE_RATES",
+                environ_prefix=None,
+            ),
         },
     }
     MONITORED_THROTTLE_FAILURE_CALLBACK = (
@@ -660,6 +665,21 @@ class Base(Configuration):
         environ_prefix=None,
         default=False,
     )
+    CONNECTION_TEST_TOKEN_TTL_SECONDS = values.PositiveIntegerValue(
+        300,
+        environ_name="CONNECTION_TEST_TOKEN_TTL_SECONDS",
+        environ_prefix=None,
+    )
+    CONNECTION_TEST_ROOM_MAX_AGE_SECONDS = values.PositiveIntegerValue(
+        300,
+        environ_name="CONNECTION_TEST_ROOM_MAX_AGE_SECONDS",
+        environ_prefix=None,
+    )
+    CONNECTION_TEST_ROOM_PREFIX = values.Value(
+        "connection-test",
+        environ_name="CONNECTION_TEST_ROOM_PREFIX",
+        environ_prefix=None,
+    )
     LIVEKIT_VERIFY_SSL = values.BooleanValue(
         True, environ_name="LIVEKIT_VERIFY_SSL", environ_prefix=None
     )