fix: Detect and handle watsonx.ai model provider rate limit exceptions

Issue

- #1599

Summary

- Added async retry logic with Full Jitter exponential backoff for all watsonx.ai HTTP calls to handle HTTP 429 rate limit responses gracefully
- Added a short-TTL in-process cache for provider health check responses to prevent concurrent identical watsonx.ai validation round-trips from amplifying rate limit exposure

New Utilities

- Added `src/utils/watsonx_retry.py`: async `request_with_retry` helper that retries only on HTTP 429, respects `Retry-After` headers (integer seconds and HTTP-date formats), applies Full Jitter backoff, and enforces both per-attempt and total wall-clock budgets
- Added `src/utils/provider_health_cache.py`: TTL-based in-process cache (default 10s, configurable via `OPENRAG_PROVIDER_HEALTH_TTL`) keyed on a SHA-256 fingerprint of provider config; API keys are hashed and never stored in plaintext

Retry Integration

- Replaced direct `client.post` / `client.get` calls with `request_with_retry` in `src/api/provider_validation.py` for all three watsonx validation paths (IAM token, completion with tools, embedding)
- Applied the same replacement in `src/services/models_service.py` for IAM token fetch and both model-listing requests

Provider Health Cache

- Updated `src/api/provider_health.py` to short-circuit repeated polled health checks using the new cache; only successful 200 responses on the default path are cached — explicit `?provider=` overrides and 503 error paths bypass the cache

Tests

- Added `tests/unit/utils/test_watsonx_retry.py` covering: immediate return on non-429, retry-then-succeed, `Retry-After` integer/HTTP-date header handling, cap clamping, non-retried status codes, max-attempt exhaustion, and total wall-clock budget short-circuit
- Added `tests/unit/utils/test_provider_health_cache.py` covering: cache miss, set/get round-trip, invalidation, key uniqueness per field, key stability, API key non-leakage in plaintext, and TTL expiry

Author: mpawlow

src/api/provider_health.py

@@ -6,6 +6,7 @@
 from fastapi import Depends
 from fastapi.responses import JSONResponse
 from utils.logging_config import get_logger
+from utils import provider_health_cache
 from config.settings import get_openrag_config
 from api.provider_validation import validate_provider_setup
 from dependencies import get_current_user
@@ -94,7 +95,29 @@ async def check_provider_health(
             embedding_endpoint = getattr(embedding_provider_config, "endpoint", None)
             embedding_project_id = getattr(embedding_provider_config, "project_id", None)
             embedding_model = current_config.knowledge.embedding_model
-        
+
+            # Short-circuit identical concurrent polls from the provider-health
+            # banner so we don't fan out N watsonx round-trips per poll cycle.
+            # Only the polled (no `check_provider`) success path is cached; the
+            # 503 branch and the specific-provider branch always re-validate.
+            health_cache_key = provider_health_cache.cache_key(
+                provider=provider,
+                embedding_provider=embedding_provider,
+                test_completion=test_completion,
+                llm_model=llm_model,
+                embedding_model=embedding_model,
+                endpoint=endpoint,
+                project_id=project_id,
+                api_key=api_key,
+                embedding_api_key=embedding_api_key,
+                embedding_endpoint=embedding_endpoint,
+                embedding_project_id=embedding_project_id,
+            )
+            cached_payload = provider_health_cache.get(health_cache_key)
+            if cached_payload is not None:
+                logger.debug("Returning cached provider-health response")
+                return JSONResponse(cached_payload, status_code=200)
+
         logger.info(f"Checking health for provider: {provider}")
 
         # Validate provider setup
@@ -204,19 +227,18 @@ async def check_provider_health(
                     status_code=503,
                 )
 
-            return JSONResponse(
-                {
-                    "status": "healthy",
-                    "message": "Both providers properly configured and validated",
-                    "llm_provider": provider,
-                    "embedding_provider": embedding_provider,
-                    "details": {
-                        "llm_model": llm_model,
-                        "embedding_model": embedding_model,
-                    },
+            healthy_payload = {
+                "status": "healthy",
+                "message": "Both providers properly configured and validated",
+                "llm_provider": provider,
+                "embedding_provider": embedding_provider,
+                "details": {
+                    "llm_model": llm_model,
+                    "embedding_model": embedding_model,
                 },
-                status_code=200,
-            )
+            }
+            provider_health_cache.set_(health_cache_key, healthy_payload)
+            return JSONResponse(healthy_payload, status_code=200)
 
     except Exception as e:
         error_message = str(e)

src/api/provider_validation.py

@@ -4,6 +4,7 @@
 import httpx
 from utils.container_utils import transform_localhost_url
 from utils.logging_config import get_logger
+from utils.watsonx_retry import request_with_retry
 
 logger = get_logger(__name__)
 def _parse_json_error_message(error_text: str) -> str:
@@ -392,7 +393,9 @@ async def _test_watsonx_lightweight_health(
     try:
         # Get bearer token from IBM IAM - this validates the API key without consuming credits
         async with httpx.AsyncClient() as client:
-            token_response = await client.post(
+            token_response = await request_with_retry(
+                client,
+                "POST",
                 "https://iam.cloud.ibm.com/identity/token",
                 headers={"Content-Type": "application/x-www-form-urlencoded"},
                 data={
@@ -428,7 +431,9 @@ async def _test_watsonx_completion_with_tools(
     try:
         # Get bearer token from IBM IAM
         async with httpx.AsyncClient() as client:
-            token_response = await client.post(
+            token_response = await request_with_retry(
+                client,
+                "POST",
                 "https://iam.cloud.ibm.com/identity/token",
                 headers={"Content-Type": "application/x-www-form-urlencoded"},
                 data={
@@ -484,7 +489,9 @@ async def _test_watsonx_completion_with_tools(
         }
 
         async with httpx.AsyncClient() as client:
-            response = await client.post(
+            response = await request_with_retry(
+                client,
+                "POST",
                 url,
                 headers=headers,
                 params=params,
@@ -523,7 +530,9 @@ async def _test_watsonx_embedding(
     try:
         # Get bearer token from IBM IAM
         async with httpx.AsyncClient() as client:
-            token_response = await client.post(
+            token_response = await request_with_retry(
+                client,
+                "POST",
                 "https://iam.cloud.ibm.com/identity/token",
                 headers={"Content-Type": "application/x-www-form-urlencoded"},
                 data={
@@ -557,7 +566,9 @@ async def _test_watsonx_embedding(
         }
 
         async with httpx.AsyncClient() as client:
-            response = await client.post(
+            response = await request_with_retry(
+                client,
+                "POST",
                 url,
                 headers=headers,
                 params=params,

src/services/models_service.py

@@ -11,6 +11,7 @@
 from config.embedding_constants import OPENAI_DEFAULT_EMBEDDING_MODEL, OPENAI_EMBEDDING_MODEL_PREFIX
 from utils.container_utils import transform_localhost_url
 from utils.logging_config import get_logger
+from utils.watsonx_retry import request_with_retry
 
 logger = get_logger(__name__)
 
@@ -456,7 +457,9 @@ async def get_ibm_models(
             bearer_token = None
             if api_key:
                 async with httpx.AsyncClient() as client:
-                    token_response = await client.post(
+                    token_response = await request_with_retry(
+                        client,
+                        "POST",
                         "https://iam.cloud.ibm.com/identity/token",
                         headers={"Content-Type": "application/x-www-form-urlencoded"},
                         data={
@@ -501,8 +504,8 @@ async def get_ibm_models(
                 if project_id:
                     text_params["project_id"] = project_id
 
-                text_response = await client.get(
-                    models_url, params=text_params, headers=headers, timeout=10.0
+                text_response = await request_with_retry(
+                    client, "GET", models_url, params=text_params, headers=headers, timeout=10.0
                 )
 
                 if text_response.status_code == 200:
@@ -538,8 +541,8 @@ async def get_ibm_models(
                 if project_id:
                     embed_params["project_id"] = project_id
 
-                embed_response = await client.get(
-                    models_url, params=embed_params, headers=headers, timeout=10.0
+                embed_response = await request_with_retry(
+                    client, "GET", models_url, params=embed_params, headers=headers, timeout=10.0
                 )
 
                 if embed_response.status_code == 200:

src/utils/provider_health_cache.py

@@ -0,0 +1,84 @@
+"""Short-TTL cache for ``GET /api/provider/health`` responses.
+
+The provider-health banner mounts on every page and polls the endpoint every
+5-30 seconds (per browser tab). With multiple tabs open, this fans out to many
+identical watsonx.ai validation calls and significantly raises the chance of
+hitting watsonx rate limits. A small in-process cache coalesces concurrent
+identical health checks so a single watsonx round-trip serves all of them.
+
+Only successful (200) responses on the default polled path are cached. The
+explicit ``?provider=`` query bypass and the 503 error path are not cached, so
+real outages and on-demand checks are never masked.
+"""
+
+import hashlib
+import os
+
+from cachetools import TTLCache
+
+_DEFAULT_TTL_S = 10
+
+
+def _ttl_seconds() -> int:
+    raw = os.environ.get("OPENRAG_PROVIDER_HEALTH_TTL")
+    if not raw:
+        return _DEFAULT_TTL_S
+    try:
+        value = int(raw)
+    except ValueError:
+        return _DEFAULT_TTL_S
+    return value if value > 0 else _DEFAULT_TTL_S
+
+
+_HEALTH_CACHE: TTLCache[str, dict] = TTLCache(maxsize=64, ttl=_ttl_seconds())
+
+
+def _fingerprint(value: str | None) -> str:
+    return hashlib.sha256((value or "").encode()).hexdigest()[:16]
+
+
+def cache_key(
+    provider: str | None,
+    embedding_provider: str | None,
+    test_completion: bool,
+    llm_model: str | None,
+    embedding_model: str | None,
+    endpoint: str | None,
+    project_id: str | None,
+    api_key: str | None,
+    embedding_api_key: str | None = None,
+    embedding_endpoint: str | None = None,
+    embedding_project_id: str | None = None,
+) -> str:
+    """Build the cache key for a polled health-check call.
+
+    The API keys are hashed (never stored in plaintext); rotating a key busts
+    the cache automatically because the fingerprint changes.
+    """
+    parts = [
+        provider or "",
+        embedding_provider or "",
+        "1" if test_completion else "0",
+        llm_model or "",
+        embedding_model or "",
+        endpoint or "",
+        project_id or "",
+        _fingerprint(api_key),
+        embedding_endpoint or "",
+        embedding_project_id or "",
+        _fingerprint(embedding_api_key),
+    ]
+    return hashlib.sha256("|".join(parts).encode()).hexdigest()
+
+
+def get(key: str) -> dict | None:
+    return _HEALTH_CACHE.get(key)
+
+
+def set_(key: str, value: dict) -> None:
+    _HEALTH_CACHE[key] = value
+
+
+def invalidate() -> None:
+    """Clear the entire cache. Intended for settings-save flows and tests."""
+    _HEALTH_CACHE.clear()

src/utils/watsonx_retry.py

@@ -0,0 +1,94 @@
+"""Async retry helper for watsonx.ai HTTP calls.
+
+Retries only on HTTP 429 using AWS-style "Full Jitter" exponential backoff,
+honors the ``Retry-After`` response header when present, and bounds both the
+attempt count and the total wall-clock budget so a caller (e.g. a health-check
+poll) cannot hang waiting on a throttled upstream.
+
+See https://aws.amazon.com/blogs/architecture/exponential-backoff-and-jitter/.
+"""
+
+import asyncio
+import random
+import time
+from datetime import UTC, datetime
+from email.utils import parsedate_to_datetime
+
+import httpx
+
+from utils.logging_config import get_logger
+
+logger = get_logger(__name__)
+
+
+def _retry_after_seconds(resp: httpx.Response, cap: float) -> float | None:
+    """Parse the ``Retry-After`` header (int seconds or HTTP-date), clamped at ``cap``.
+
+    Returns ``None`` when the header is absent or unparseable, so the caller
+    falls back to jittered exponential backoff.
+    """
+    raw = resp.headers.get("Retry-After")
+    if not raw:
+        return None
+    try:
+        return min(max(0.0, float(raw)), cap)
+    except (TypeError, ValueError):
+        pass
+    try:
+        dt = parsedate_to_datetime(raw)
+        if dt is None:
+            return None
+        if dt.tzinfo is None:
+            dt = dt.replace(tzinfo=UTC)
+        delta = (dt - datetime.now(UTC)).total_seconds()
+        return max(0.0, min(delta, cap))
+    except (TypeError, ValueError):
+        return None
+
+
+async def request_with_retry(
+    client: httpx.AsyncClient,
+    method: str,
+    url: str,
+    *,
+    max_attempts: int = 5,
+    base: float = 1.0,
+    cap: float = 15.0,
+    total_cap_s: float = 30.0,
+    **kwargs,
+) -> httpx.Response:
+    """Issue an httpx request, retrying only on HTTP 429.
+
+    Non-429 responses (including 4xx auth failures and 5xx server errors) are
+    returned to the caller on the first attempt so existing error-shaping logic
+    runs unchanged.
+    """
+    start = time.monotonic()
+    resp: httpx.Response | None = None
+    for attempt in range(1, max_attempts + 1):
+        resp = await client.request(method, url, **kwargs)
+        if resp.status_code != 429:
+            return resp
+        if attempt == max_attempts:
+            break
+        elapsed = time.monotonic() - start
+        remaining = total_cap_s - elapsed
+        if remaining <= 0:
+            break
+        ra = _retry_after_seconds(resp, cap)
+        if ra is not None:
+            sleep = ra
+        else:
+            # Full Jitter: random.uniform(0, min(cap, base * 2 ** (attempt - 1)))
+            sleep = random.uniform(0, min(cap, base * (2 ** (attempt - 1))))
+        sleep = min(sleep, remaining)
+        logger.warning(
+            "watsonx 429 on %s %s (attempt %d/%d), sleeping %.2fs",
+            method,
+            url,
+            attempt,
+            max_attempts,
+            sleep,
+        )
+        await asyncio.sleep(sleep)
+    return resp  # type: ignore[return-value]  # loop runs at least once