diff --git a/backend/routers/transcribe.py b/backend/routers/transcribe.py index b56379eb0aa..377abfdd8bc 100644 --- a/backend/routers/transcribe.py +++ b/backend/routers/transcribe.py @@ -26,6 +26,7 @@ ) from utils.other import endpoints as auth from utils.executors import db_executor, run_blocking +from utils.metrics import record_listen_ws_handshake_failure logger = logging.getLogger(__name__) @@ -157,6 +158,7 @@ async def _listen( try: await websocket.accept() except RuntimeError as error: + record_listen_ws_handshake_failure(surface='mobile') logger.error('listen accept failed uid=%s error_type=%s', uid, type(error).__name__) return await _stream_handler( @@ -253,6 +255,7 @@ async def web_listen_handler( try: await websocket.accept() except RuntimeError as error: + record_listen_ws_handshake_failure(surface='web') logger.error('web listen accept failed error_type=%s', type(error).__name__) return try: diff --git a/backend/tests/unit/test_listen_quota_handshake_metrics.py b/backend/tests/unit/test_listen_quota_handshake_metrics.py new file mode 100644 index 00000000000..f481a3d16f2 --- /dev/null +++ b/backend/tests/unit/test_listen_quota_handshake_metrics.py @@ -0,0 +1,268 @@ +"""Unit tests for listen WS handshake, API 429 rate-limit, and direct LLM 429/quota counters (#21461).""" + +from __future__ import annotations + +import asyncio +import json +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest +from fastapi import HTTPException + +from utils import metrics +from utils.rate_limit_config import RATE_POLICIES +from utils.llm import byok_errors +from utils.other import endpoints + + +class _FakeHTTPError(Exception): + def __init__(self, message: str, status_code: int): + super().__init__(message) + self.status_code = status_code + + +class RateLimitError(Exception): + pass + + +def _counter_value(counter, **labels) -> float: + return counter.labels(**labels)._value.get() + + +def test_metrics_have_zero_initialized_bounded_children(): + for surface in ('mobile', 'web'): + assert ( + _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface=surface, + reason='accept_runtime_error', + ) + >= 0.0 + ) + + for scope in ('redis_policy', 'ip_window'): + for family in metrics._API_RATE_LIMIT_POLICY_FAMILIES: + assert ( + _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope=scope, + policy_family=family, + ) + >= 0.0 + ) + + for provider in ('openai', 'anthropic', 'gemini', 'openrouter', 'groq', 'other'): + for source in ('platform', 'byok'): + for reason in ('quota', 'rate_limit'): + assert ( + _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider=provider, + source=source, + reason=reason, + ) + >= 0.0 + ) + + +def test_record_helpers_collapse_unbounded_labels(): + before_ws = _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='other', + reason='other', + ) + metrics.record_listen_ws_handshake_failure(surface='unexpected-surface', reason='unexpected-reason') + assert ( + _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='other', + reason='other', + ) + == before_ws + 1.0 + ) + + before_rl = _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='other', + policy_family='custom', + ) + metrics.record_api_rate_limit_throttled(scope='unknown_scope', policy='unbounded_user_id:12345') + assert ( + _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='other', + policy_family='custom', + ) + == before_rl + 1.0 + ) + + before_llm = _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='other', + source='other', + reason='rate_limit', + ) + metrics.record_direct_llm_rate_limit_error(provider='custom-vendor', source='invalid', reason='invalid') + assert ( + _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='other', + source='other', + reason='rate_limit', + ) + == before_llm + 1.0 + ) + + +def test_listen_and_web_listen_handshake_failures_increment_counter(): + from testing.import_isolation import stub_modules + + runtime_stub = SimpleNamespace(run_listen_session=AsyncMock()) + with stub_modules({'routers.listen.runtime': runtime_stub}): + from routers import transcribe + + ws = MagicMock() + ws.accept = AsyncMock(side_effect=RuntimeError('ASGI callable returned without sending handshake')) + + before_mobile = _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='mobile', + reason='accept_runtime_error', + ) + asyncio.run(transcribe._listen(ws, 'uid-test')) + assert ( + _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='mobile', + reason='accept_runtime_error', + ) + == before_mobile + 1.0 + ) + + before_web = _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='web', + reason='accept_runtime_error', + ) + asyncio.run(transcribe.web_listen_handler(ws)) + assert ( + _counter_value( + metrics.OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL, + surface='web', + reason='accept_runtime_error', + ) + == before_web + 1.0 + ) + + +def test_rate_limit_429_increments_throttled_counter(monkeypatch): + before_redis = _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='redis_policy', + policy_family='voice', + ) + monkeypatch.setattr(endpoints, 'RATE_LIMIT_SHADOW', False) + monkeypatch.setattr(endpoints, 'check_rate_limit', lambda *_args, **_kwargs: (False, 0, 12)) + + with pytest.raises(HTTPException) as exc_info: + endpoints._enforce_rate_limit('uid-1', 'voice:transcribe') + assert exc_info.value.status_code == 429 + assert ( + _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='redis_policy', + policy_family='voice', + ) + == before_redis + 1.0 + ) + + before_ip = _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='ip_window', + policy_family='custom', + ) + fake_request = SimpleNamespace(client=SimpleNamespace(host='127.0.0.1')) + endpoints.cached['rate_limit:phone_verify:127.0.0.1'] = json.dumps({'timestamp': 9999999999, 'remaining': 0}) + try: + with pytest.raises(HTTPException) as ip_exc: + endpoints.rate_limit_custom('phone_verify', fake_request, 5, 60) + assert ip_exc.value.status_code == 429 + finally: + endpoints.cached.pop('rate_limit:phone_verify:127.0.0.1', None) + + assert ( + _counter_value( + metrics.OMI_API_RATE_LIMIT_THROTTLED_TOTAL, + scope='ip_window', + policy_family='custom', + ) + == before_ip + 1.0 + ) + + +@patch('utils.llm.byok_errors.get_byok_uid', return_value='user-1') +@patch('utils.llm.byok_errors._send_byok_llm_error_notification') +def test_handle_llm_error_records_platform_and_byok_429_and_quota_metrics(mock_notify, _mock_uid): + before_platform_quota = _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='platform', + reason='quota', + ) + before_platform_rl = _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='platform', + reason='rate_limit', + ) + before_byok_quota = _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='byok', + reason='quota', + ) + + with patch('utils.llm.byok_errors.get_byok_key', return_value=None): + byok_errors.handle_llm_error( + _FakeHTTPError('You exceeded your current quota: insufficient_quota', 429), 'openai' + ) + byok_errors.handle_llm_error(RateLimitError('Rate limit reached for gpt-4o'), 'openai') + byok_errors.handle_llm_error(_FakeHTTPError('Internal server error', 500), 'openai') + + with patch('utils.llm.byok_errors.get_byok_key', return_value='sk-byok'): + byok_errors.handle_llm_error(_FakeHTTPError('insufficient_quota', 429), 'openai') + + assert ( + _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='platform', + reason='quota', + ) + == before_platform_quota + 1.0 + ) + assert ( + _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='platform', + reason='rate_limit', + ) + == before_platform_rl + 1.0 + ) + assert ( + _counter_value( + metrics.OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL, + provider='openai', + source='byok', + reason='quota', + ) + == before_byok_quota + 1.0 + ) + mock_notify.assert_called_once_with('user-1', 'openai', 'quota') + + +def test_api_rate_limit_policy_families_match_rate_policies(): + expected_families = {k.split(':', 1)[0] for k in RATE_POLICIES} | {'custom'} + assert metrics._API_RATE_LIMIT_POLICY_FAMILIES == expected_families diff --git a/backend/utils/llm/byok_errors.py b/backend/utils/llm/byok_errors.py index 81c5c01616c..1226320fa66 100644 --- a/backend/utils/llm/byok_errors.py +++ b/backend/utils/llm/byok_errors.py @@ -34,6 +34,7 @@ def release_byok_llm_error_notification_lock(uid: str, provider: str, reason: st from utils.byok import get_byok_key, get_byok_uid from utils.executors import storage_executor, submit_with_context from utils.log_sanitizer import sanitize +from utils.metrics import record_direct_llm_rate_limit_error logger = logging.getLogger(__name__) @@ -51,6 +52,21 @@ def get_llm_error_source(provider: Optional[str]) -> str: return 'platform' +def _classify_rate_limit_or_quota_error(error: Exception) -> Optional[str]: + """Return 'quota' or 'rate_limit' when an LLM SDK exception represents a 429/quota failure.""" + status_code = _get_status_code(error) + error_name = type(error).__name__ + error_text = sanitize(str(error)).lower() + + if status_code == 429 or error_name in _QUOTA_ERROR_NAMES: + if 'insufficient_quota' in error_text or 'quota' in error_text: + return 'quota' + return 'rate_limit' + if 'insufficient_quota' in error_text: + return 'quota' + return None + + def classify_byok_llm_error(error: Exception) -> Optional[str]: """Classify user-actionable BYOK failures for structured logging.""" status_code = _get_status_code(error) @@ -77,6 +93,9 @@ def handle_llm_error( """Log LLM failures with source context.""" source = get_llm_error_source(provider) reason = classify_byok_llm_error(error) if source == 'byok' else None + rate_limit_reason = _classify_rate_limit_or_quota_error(error) + if rate_limit_reason is not None: + record_direct_llm_rate_limit_error(provider=provider, source=source, reason=rate_limit_reason) uid = get_byok_uid() status_code = _get_status_code(error) diff --git a/backend/utils/metrics.py b/backend/utils/metrics.py index e32160d47d2..45681b494bc 100644 --- a/backend/utils/metrics.py +++ b/backend/utils/metrics.py @@ -1521,3 +1521,72 @@ def stop_metrics_sidecar_server() -> None: ['stage'], buckets=(1, 2, 3, 4, 6, 8, 12, 16, 25, 50, 100, 250, 1000, 2500), ) + +OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL = Counter( + 'omi_listen_ws_handshake_failures_total', + 'WebSocket accept/handshake failures on /v4/listen endpoints by bounded surface and reason', + ['surface', 'reason'], +) +_LISTEN_WS_HANDSHAKE_SURFACES = frozenset({'mobile', 'web'}) +_LISTEN_WS_HANDSHAKE_REASONS = frozenset({'accept_runtime_error'}) +for _ws_surface in _LISTEN_WS_HANDSHAKE_SURFACES: + OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL.labels(surface=_ws_surface, reason='accept_runtime_error') + + +def record_listen_ws_handshake_failure(*, surface: str, reason: str = 'accept_runtime_error') -> None: + if surface not in _LISTEN_WS_HANDSHAKE_SURFACES: + surface = 'other' + OMI_LISTEN_WS_HANDSHAKE_FAILURES_TOTAL.labels( + surface=surface, reason='accept_runtime_error' if reason == 'accept_runtime_error' else 'other' + ).inc() + + +OMI_API_RATE_LIMIT_THROTTLED_TOTAL = Counter( + 'omi_api_rate_limit_throttled_total', + 'HTTP 429 rate-limit/quota rejections returned by backend rate limiters by bounded scope and policy family', + ['scope', 'policy_family'], +) +import utils.rate_limit_config as _rate_limit_cfg + +_API_RATE_LIMIT_SCOPES = frozenset({'redis_policy', 'ip_window'}) +_API_RATE_LIMIT_POLICY_FAMILIES = frozenset( + {k.split(':', 1)[0] for k in getattr(_rate_limit_cfg, 'RATE_POLICIES', {})} | {'custom'} +) +for _rl_scope in _API_RATE_LIMIT_SCOPES: + for _rl_family in _API_RATE_LIMIT_POLICY_FAMILIES: + OMI_API_RATE_LIMIT_THROTTLED_TOTAL.labels(scope=_rl_scope, policy_family=_rl_family) + + +def record_api_rate_limit_throttled(*, scope: str, policy: str) -> None: + if scope not in _API_RATE_LIMIT_SCOPES: + scope = 'other' + family = (policy or '').split(':', 1)[0].strip().lower() + if family not in _API_RATE_LIMIT_POLICY_FAMILIES: + family = 'custom' + OMI_API_RATE_LIMIT_THROTTLED_TOTAL.labels(scope=scope, policy_family=family).inc() + + +OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL = Counter( + 'omi_direct_llm_rate_limit_errors_total', + 'Direct LLM SDK 429 and quota errors by bounded provider, credential source, and reason', + ['provider', 'source', 'reason'], +) +_DIRECT_LLM_PROVIDERS = frozenset({'openai', 'anthropic', 'gemini', 'openrouter', 'groq', 'other'}) +_DIRECT_LLM_SOURCES = frozenset({'platform', 'byok'}) +_DIRECT_LLM_REASONS = frozenset({'quota', 'rate_limit'}) +for _llm_provider in _DIRECT_LLM_PROVIDERS: + for _llm_source in _DIRECT_LLM_SOURCES: + for _llm_reason in _DIRECT_LLM_REASONS: + OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL.labels( + **{'provider': _llm_provider, 'source': _llm_source, 'reason': _llm_reason} + ) + + +def record_direct_llm_rate_limit_error(*, provider: str | None, source: str, reason: str) -> None: + norm_provider = (provider or '').strip().lower() + norm_provider = norm_provider if norm_provider in _DIRECT_LLM_PROVIDERS else 'other' + source = source if source in _DIRECT_LLM_SOURCES else 'other' + reason = reason if reason in _DIRECT_LLM_REASONS else 'rate_limit' + OMI_DIRECT_LLM_RATE_LIMIT_ERRORS_TOTAL.labels( + **{'provider': norm_provider, 'source': source, 'reason': reason} + ).inc() diff --git a/backend/utils/other/endpoints.py b/backend/utils/other/endpoints.py index 15ae21fca37..c0c27de17a3 100644 --- a/backend/utils/other/endpoints.py +++ b/backend/utils/other/endpoints.py @@ -33,6 +33,7 @@ validate_byok_websocket_keys, ) from utils.executors import ExecutorSaturatedError, critical_executor, db_executor, run_blocking +from utils.metrics import record_api_rate_limit_throttled from utils.rate_limit_config import RATE_POLICIES, RATE_LIMIT_SHADOW, get_effective_limit from database.fsr_read_cache import request_read_scope @@ -570,6 +571,7 @@ def rate_limit_custom(endpoint: str, request: Request, requests_per_window: int, remaining = requests_per_window timestamp = current_time elif remaining == 0: + record_api_rate_limit_throttled(scope='ip_window', policy=endpoint) raise HTTPException(status_code=429, detail="Too Many Requests") remaining -= 1 @@ -613,6 +615,7 @@ def _enforce_rate_limit(key: str, policy_name: str, *, fail_closed: bool = False if RATE_LIMIT_SHADOW: logger.warning(f"[shadow] rate_limit_exceeded policy={policy_name} key={key} retry_after={retry_after}") return + record_api_rate_limit_throttled(scope='redis_policy', policy=policy_name) raise HTTPException( status_code=429, detail=f"Rate limit exceeded. Try again in {retry_after}s.",