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

.env.example

@@ -135,6 +135,11 @@ OLLAMA_ENDPOINT=
 WATSONX_API_KEY=
 WATSONX_ENDPOINT=
 WATSONX_PROJECT_ID=
+# Tune the ibm-watsonx-ai SDK's native 429 retry used during ingestion embedding.
+# OPTIONAL: Max retries = retry attempts (default 10).
+# WATSONX_MAX_RETRIES=10
+# OPTIONAL: Delay time = base backoff seconds for the jittered exponential backoff (default 0.5).
+# WATSONX_DELAY_TIME=0.5
 
 # LLM Provider configuration. Providers can be "anthropic", "watsonx", "ibm" or "ollama".
 LLM_PROVIDER=

docker-compose.yml

@@ -166,6 +166,8 @@ services:
       - WATSONX_APIKEY=${WATSONX_API_KEY}
       - WATSONX_URL=${WATSONX_URL:-${WATSONX_ENDPOINT}}
       - WATSONX_PROJECT_ID=${WATSONX_PROJECT_ID}
+      - WATSONX_MAX_RETRIES=${WATSONX_MAX_RETRIES:-10}
+      - WATSONX_DELAY_TIME=${WATSONX_DELAY_TIME:-0.5}
       - OLLAMA_BASE_URL=${OLLAMA_ENDPOINT}
       - LANGFLOW_CONFIG_DIR=/app/langflow-data
       - LANGFLOW_DATABASE_URL=${LANGFLOW_DATABASE_URL:-sqlite:////app/langflow-data/langflow.db}

kubernetes/helm/openrag/templates/langflow/langflow-dotenv.yaml

@@ -82,6 +82,12 @@ stringData:
     {{- else }}
     WATSONX_PROJECT_ID="None"
     {{- end }}
+    {{- if .Values.llmProviders.watsonx.maxRetries }}
+    WATSONX_MAX_RETRIES={{ .Values.llmProviders.watsonx.maxRetries | quote }}
+    {{- end }}
+    {{- if .Values.llmProviders.watsonx.delayTime }}
+    WATSONX_DELAY_TIME={{ .Values.llmProviders.watsonx.delayTime | quote }}
+    {{- end }}
 
     # LLM / Embedding provider configuration
     {{- if .Values.appConfig.agent.llmProvider }}

kubernetes/helm/openrag/values.yaml

@@ -553,6 +553,8 @@ llmProviders:
     apiKey: ""                            # WatsonX API key (stored in secret)
     endpoint: "https://us-south.ml.cloud.ibm.com"
     projectId: ""                         # WatsonX project ID
+    maxRetries: ""                        # IBM SDK retry count on 429 (optional; blank = SDK default of 10)
+    delayTime: ""                         # IBM SDK backoff delay in seconds (optional; blank = SDK default of 0.5)
   ollama:
     enabled: false
     endpoint: ""                          # Ollama endpoint URL (e.g., "http://ollama:11434")

src/api/provider_health.py

@@ -1,21 +1,23 @@
 """Provider health check endpoint."""
 
 import asyncio
-from typing import Optional
+
 import httpx
 from fastapi import Depends
 from fastapi.responses import JSONResponse
-from utils.logging_config import get_logger
-from config.settings import get_openrag_config
+
 from api.provider_validation import validate_provider_setup
+from config.settings import get_openrag_config
 from dependencies import get_current_user
 from session_manager import User
+from utils import provider_health_cache
+from utils.logging_config import get_logger
 
 logger = get_logger(__name__)
 
 
 async def check_provider_health(
-    provider: Optional[str] = None,
+    provider: str | None = None,
     test_completion: bool = False,
     user: User = Depends(get_current_user),
 ):
@@ -33,6 +35,7 @@ async def check_provider_health(
         503: Provider validation failed
     """
     check_provider = provider
+    _health_leader_key: str | None = None  # set when this coroutine wins leader election
     try:
         # Get current config
         current_config = get_openrag_config()
@@ -66,8 +69,16 @@ async def check_provider_health(
                 project_id = getattr(provider_config, "project_id", None)
 
                 # Check if this provider is used for LLM or embedding
-                llm_model = current_config.agent.llm_model if provider == current_config.agent.llm_provider else None
-                embedding_model = current_config.knowledge.embedding_model if provider == current_config.knowledge.embedding_provider else None
+                llm_model = (
+                    current_config.agent.llm_model
+                    if provider == current_config.agent.llm_provider
+                    else None
+                )
+                embedding_model = (
+                    current_config.knowledge.embedding_model
+                    if provider == current_config.knowledge.embedding_provider
+                    else None
+                )
             except ValueError:
                 # Provider not found in configuration
                 return JSONResponse(
@@ -94,9 +105,47 @@ 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)
+
+            # Singleflight: if another coroutine is already validating this
+            # exact config, wait for it to finish rather than issuing a
+            # redundant upstream call.
+            while True:
+                is_leader = await provider_health_cache.acquire(health_cache_key)
+                if is_leader:
+                    _health_leader_key = health_cache_key
+                    break
+                # Woke up after an in-flight validation completed.
+                cached_payload = provider_health_cache.get(health_cache_key)
+                if cached_payload is not None:
+                    logger.debug("Returning cached provider-health response (waited for in-flight)")
+                    return JSONResponse(cached_payload, status_code=200)
+                # Leader's validation failed; retry leader election rather than
+                # all waiters fanning out to validate simultaneously.
+
         logger.info(f"Checking health for provider: {provider}")
-        
+
         # Validate provider setup
         if check_provider:
             # Validate specific provider
@@ -160,9 +209,11 @@ async def check_provider_health(
                 and embedding_provider == "watsonx"
                 and llm_error is None
             ):
-                logger.info("Waiting 2 seconds before WatsonX embedding test (after completion test)")
+                logger.info(
+                    "Waiting 2 seconds before WatsonX embedding test (after completion test)"
+                )
                 await asyncio.sleep(2)
-            
+
             try:
                 await validate_provider_setup(
                     provider=embedding_provider,
@@ -179,10 +230,14 @@ async def check_provider_health(
                     logger.info(f"Embedding provider ({embedding_provider}) appears busy: {str(e)}")
                 else:
                     embedding_error = str(e)
-                    logger.error(f"Embedding provider ({embedding_provider}) validation timed out: {embedding_error}")
+                    logger.error(
+                        f"Embedding provider ({embedding_provider}) validation timed out: {embedding_error}"
+                    )
             except Exception as e:
                 embedding_error = str(e)
-                logger.error(f"Embedding provider ({embedding_provider}) validation failed: {embedding_error}")
+                logger.error(
+                    f"Embedding provider ({embedding_provider}) validation failed: {embedding_error}"
+                )
 
             # Return combined status
             if llm_error or embedding_error:
@@ -192,6 +247,9 @@ async def check_provider_health(
                 if embedding_error:
                     errors.append(f"Embedding ({embedding_provider}): {embedding_error}")
 
+                if _health_leader_key:
+                    provider_health_cache.release_error(_health_leader_key)
+                    _health_leader_key = None
                 return JSONResponse(
                     {
                         "status": "unhealthy",
@@ -204,24 +262,30 @@ 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_and_release(health_cache_key, healthy_payload)
+            _health_leader_key = None
+            return JSONResponse(healthy_payload, status_code=200)
+
+    except asyncio.CancelledError:
+        if _health_leader_key:
+            provider_health_cache.release_error(_health_leader_key)
+        raise
     except Exception as e:
+        if _health_leader_key:
+            provider_health_cache.release_error(_health_leader_key)
         error_message = str(e)
         logger.error(f"Provider health check failed for {provider}: {error_message}")
-        
+
         return JSONResponse(
             {
                 "status": "unhealthy",
@@ -230,4 +294,3 @@ async def check_provider_health(
             },
             status_code=503,
         )
-

src/api/settings/endpoints.py

@@ -82,6 +82,7 @@
 )
 from services.docling_service import DoclingConfig, get_docling_preset_configs
 from session_manager import User
+from utils import provider_health_cache
 from utils.langflow_utils import LangflowNotReadyError, wait_for_langflow
 from utils.logging_config import get_logger
 from utils.telemetry import Category, MessageId, TelemetryClient
@@ -709,6 +710,8 @@ async def update_settings(
         if not config_manager.save_config_file(current_config):
             return JSONResponse({"error": "Failed to save configuration"}, status_code=500)
 
+        provider_health_cache.invalidate()
+
         # Refresh patched client immediately so subsequent requests pick up latest config.
         await clients.refresh_patched_client()
 
@@ -1064,6 +1067,8 @@ async def onboarding(
                             {"error": "Failed to save configuration"}, status_code=500
                         )
 
+                    provider_health_cache.invalidate()
+
                     ingestion_jwt = (
                         user.jwt_token if IBM_AUTH_ENABLED and user and user.jwt_token else None
                     )
@@ -1098,6 +1103,7 @@ async def onboarding(
                     )
 
         if config_manager.save_config_file(current_config):
+            provider_health_cache.invalidate()
             set_fields = [k for k, v in body.model_dump(exclude_unset=True).items()]
             logger.info(
                 "Onboarding configuration updated successfully",

src/config/settings.py

@@ -187,6 +187,13 @@ def _resolve_skip_os_security_default() -> str:
 # Each entry holds ~1 KB of claim data; 1024 entries ≈ 1 MB.
 JWT_CLAIMS_CACHE_MAX_SIZE = get_env_int("OPENRAG_JWT_CACHE_MAXSIZE", 1024)
 
+# TTL (seconds) for the in-process provider health-check response cache.
+# The banner polls GET /api/provider/health every 5-30 s per browser tab;
+# caching coalesces concurrent identical calls so watsonx round-trips are
+# not fanned out. Must be >= 1; non-positive values fall back to the default.
+_raw_phc_ttl = get_env_int("OPENRAG_PROVIDER_HEALTH_TTL", 10)
+PROVIDER_HEALTH_CACHE_TTL_SECONDS = _raw_phc_ttl if _raw_phc_ttl > 0 else 10
+
 # Docling service URL configuration
 # Priority:
 # 1. DOCLING_SERVE_URL environment variable