diff --git a/lending-poc/.env.example b/lending-poc/.env.example index 5419649..c1ea069 100644 --- a/lending-poc/.env.example +++ b/lending-poc/.env.example @@ -16,3 +16,10 @@ COMPOSE_PROFILES=cpu # docker compose exec ollama ollama pull OLLAMA_MODEL=gemma4:e4b-it-qat # OLLAMA_HOST=http://ollama:11434 + +# How long (seconds) the gateway waits on each backend before giving up. +# Defaults (300s) already match what the frontend budgets for these same +# calls — only override if you've changed the frontend's timeouts too. +# OCR_REQUEST_TIMEOUT_SECONDS=300 +# TRANSLATION_REQUEST_TIMEOUT_SECONDS=300 +# FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS=300 diff --git a/lending-poc/docker-compose.yml b/lending-poc/docker-compose.yml index 965aaea..a72a413 100644 --- a/lending-poc/docker-compose.yml +++ b/lending-poc/docker-compose.yml @@ -192,6 +192,13 @@ services: FIELD_MAPPING_BASE_URL: http://field_mapping:8002 OCR_BASE_URL: http://ocr:8010 TRANSLATION_BASE_URL: http://translation:8001 + # Each backend calls out to a local model and can legitimately run for + # minutes (OCR to Surya, translation/field-mapping to Ollama) — these + # match the timeouts the frontend already budgets for the same calls, + # so the gateway is never the first link in the chain to give up. + OCR_REQUEST_TIMEOUT_SECONDS: ${OCR_REQUEST_TIMEOUT_SECONDS:-300} + TRANSLATION_REQUEST_TIMEOUT_SECONDS: ${TRANSLATION_REQUEST_TIMEOUT_SECONDS:-300} + FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS: ${FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS:-300} volumes: - ./gateway:/app depends_on: diff --git a/lending-poc/document_processing/translation/api/models.py b/lending-poc/document_processing/translation/api/models.py index c7e45ed..952f4c5 100644 --- a/lending-poc/document_processing/translation/api/models.py +++ b/lending-poc/document_processing/translation/api/models.py @@ -11,6 +11,8 @@ # in config.py automatically updates the API validation too. DomainLiteral = Literal["legal", "banking"] +HealthStatus = Literal["ok", "initializing", "unreachable"] + # --------------------------------------------------------------------------- # Request models @@ -62,7 +64,14 @@ class FilesTranslateResponse(BaseModel): class HealthResponse(BaseModel): """Response for GET /health.""" - status: str = Field(description="'ok' if the model is reachable, 'degraded' otherwise.") + status: HealthStatus = Field( + description=( + "'unreachable' if Ollama isn't responding, 'initializing' if it's " + "reachable but the model hasn't responded yet (e.g. cold-loading), " + "'ok' if the model is loaded and responding." + ) + ) + detail: str = Field(description="Human-readable explanation of `status`.") model: str = Field(description="Model name currently configured.") adapter: str = Field(description="Adapter/backend currently configured.") domains: list[str] = Field(description="Supported translation domains.") diff --git a/lending-poc/document_processing/translation/api/routes.py b/lending-poc/document_processing/translation/api/routes.py index c1a3847..19d7514 100644 --- a/lending-poc/document_processing/translation/api/routes.py +++ b/lending-poc/document_processing/translation/api/routes.py @@ -54,6 +54,14 @@ def _get_service(request: Request, domain: str): # Health # --------------------------------------------------------------------------- +# Human-readable detail shown alongside HealthResponse.status. +STATUS_DETAIL = { + "unreachable": "Ollama is unreachable.", + "initializing": "Ollama reachable — waiting for the model to respond.", + "ok": "Model is loaded and responding.", +} + + @router.get( "/health", response_model=HealthResponse, @@ -62,9 +70,11 @@ def _get_service(request: Request, domain: str): ) async def health(request: Request): """Returns model reachability and KB sizes for all loaded domains.""" - # Check health via the default domain service (one model shared across all) + # Check health via the default domain service (one model shared across all). + # health_status() is an instant, non-blocking read of state tracked by a + # background monitor — it never calls Ollama itself (see ollama_adapter.py). default_service = request.app.state.services[DEFAULT_DOMAIN] - reachable = default_service.health_check() + status = default_service.health_status() kb_entries = { domain: svc.kb_size() @@ -72,7 +82,8 @@ async def health(request: Request): } return HealthResponse( - status="ok" if reachable else "degraded", + status=status, + detail=STATUS_DETAIL[status], model=MODEL_NAME, adapter=MODEL_ADAPTER, domains=SUPPORTED_DOMAINS, diff --git a/lending-poc/document_processing/translation/api_server.py b/lending-poc/document_processing/translation/api_server.py index 8c23875..55153ca 100644 --- a/lending-poc/document_processing/translation/api_server.py +++ b/lending-poc/document_processing/translation/api_server.py @@ -16,7 +16,13 @@ from fastapi import FastAPI from translation_service import TranslationService -from translation_service.config import SUPPORTED_DOMAINS, MODEL_NAME, MODEL_ADAPTER, MODEL_OPTIONS +from translation_service.config import ( + SUPPORTED_DOMAINS, + DEFAULT_DOMAIN, + MODEL_NAME, + MODEL_ADAPTER, + MODEL_OPTIONS, +) from api.routes import router @@ -33,9 +39,14 @@ async def lifespan(app: FastAPI): model_options=MODEL_OPTIONS, ) + # Only the default domain's service is monitored — /health checks it alone + # since all domains share one underlying model (see api/routes.py). + await app.state.services[DEFAULT_DOMAIN].start_health_monitor() + print(f"[startup] Ready — model={MODEL_NAME}, adapter={MODEL_ADAPTER}, " f"domains={SUPPORTED_DOMAINS}") yield + await app.state.services[DEFAULT_DOMAIN].stop_health_monitor() print("[shutdown] Translation service stopped.") diff --git a/lending-poc/document_processing/translation/translation_service/adapters/base.py b/lending-poc/document_processing/translation/translation_service/adapters/base.py index ba2208c..619d4d0 100644 --- a/lending-poc/document_processing/translation/translation_service/adapters/base.py +++ b/lending-poc/document_processing/translation/translation_service/adapters/base.py @@ -41,7 +41,31 @@ def health_check(self) -> bool: """ Return True if the model / backend service is reachable. - Used by run.py before starting a batch and by the future /health endpoint. - Should not raise — catch exceptions internally and return False. + Blocking — used by run.py before starting a batch. Should not raise — + catch exceptions internally and return False. """ ... + + def health_status(self) -> str: + """ + Return one of "ok", "initializing", "unreachable" for the FastAPI + /health endpoint. Non-blocking by contract — should be an instant read + of previously-observed state, not a fresh call to the backend. + + Default implementation falls back to a blocking health_check() call, + for adapters that don't track finer-grained state. Override alongside + start_monitoring()/stop_monitoring() to report real-time state instead. + """ + return "ok" if self.health_check() else "unreachable" + + async def start_monitoring(self) -> None: + """ + Optional hook: begin any background readiness tracking backing + health_status(). Default is a no-op — adapters that don't override + this just rely on health_status()'s default blocking fallback. + """ + return + + async def stop_monitoring(self) -> None: + """Optional hook: stop background tracking started by start_monitoring().""" + return diff --git a/lending-poc/document_processing/translation/translation_service/adapters/ollama_adapter.py b/lending-poc/document_processing/translation/translation_service/adapters/ollama_adapter.py index 9b6d34e..eabae7a 100644 --- a/lending-poc/document_processing/translation/translation_service/adapters/ollama_adapter.py +++ b/lending-poc/document_processing/translation/translation_service/adapters/ollama_adapter.py @@ -8,10 +8,28 @@ Swapping to a different Ollama model requires only a config.py change (MODEL_NAME). Swapping to a completely different backend requires a new adapter file; this file does not need to change. + +Incident notes — read before touching health_check()/health_status(): +A chat()/generate() call must NEVER be given a client-side timeout. Ollama +treats the client closing that connection as cancellation and aborts an +in-progress cold model load outright ("client connection closed before +llama-server finished loading, aborting load"). A short-timeout health probe +around chat() caused a load/timeout/abort loop that never let the model +finish loading. The fix is not to avoid chat() — it's to never race it against +a timeout and never call it synchronously inside a request handler. See +_ping() and _monitor_loop() below. """ +import asyncio + from ollama import chat, ChatResponse from .base import ModelAdapter +from ..config import ( + OLLAMA_HEALTH_RETRY_SECONDS, + OLLAMA_HEALTH_MAX_FAST_RETRIES, + OLLAMA_HEALTH_BACKOFF_SECONDS, + OLLAMA_HEALTH_RECHECK_SECONDS, +) class OllamaAdapter(ModelAdapter): @@ -26,6 +44,8 @@ class OllamaAdapter(ModelAdapter): def __init__(self, model_name: str, model_options: dict): self.model_name = model_name self.model_options = model_options + self._status = "initializing" + self._monitor_task: asyncio.Task | None = None def translate(self, prompt: str) -> str: """Send the prompt to Ollama and return the response text.""" @@ -36,18 +56,69 @@ def translate(self, prompt: str) -> str: ) return response["message"]["content"] + def _ping(self) -> None: + """ + Blocking probe shared by health_check() (CLI) and the background + monitor (server). Deliberately untimed — see module docstring. + """ + chat( + model=self.model_name, + messages=[{"role": "user", "content": "ping"}], + options={"num_predict": 1}, + ) + def health_check(self) -> bool: """ - Verify Ollama is running and the configured model is available. - Sends a minimal prompt to avoid false positives from API-only checks. + Blocking reachability check for the CLI (run.py). Waits out a cold + load rather than racing it — appropriate for a one-shot batch script. """ try: - response: ChatResponse = chat( - model=self.model_name, - messages=[{"role": "user", "content": "ping"}], - options={"num_predict": 1}, - ) - return bool(response["message"]["content"] is not None) + self._ping() + return True except Exception as exc: print(f"[OllamaAdapter] health_check failed: {exc}") return False + + def health_status(self) -> str: + """ + Instant, non-blocking read for the FastAPI /health route. Never calls + Ollama itself — reflects whatever the background monitor last observed. + See start_monitoring(). + """ + return self._status + + async def start_monitoring(self) -> None: + """Begin the background readiness monitor backing health_status().""" + self._monitor_task = asyncio.create_task(self._monitor_loop()) + + async def stop_monitoring(self) -> None: + """Stop the background monitor started by start_monitoring().""" + if self._monitor_task and not self._monitor_task.done(): + self._monitor_task.cancel() + + async def _monitor_loop(self) -> None: + """ + Background loop backing health_status(). Runs _ping() to completion on + a worker thread — never given a timeout, never cancelled mid-flight — + so it can never trigger the abort-on-close incident described in the + module docstring. Retries quickly at first, then backs off to a slow, + indefinite retry once a genuine outage looks sustained, and keeps + re-confirming "ok" so a later outage is eventually reflected too. + """ + consecutive_failures = 0 + while True: + self._status = "initializing" + try: + await asyncio.to_thread(self._ping) + self._status = "ok" + consecutive_failures = 0 + await asyncio.sleep(OLLAMA_HEALTH_RECHECK_SECONDS) + except Exception as exc: + consecutive_failures += 1 + self._status = "unreachable" + if consecutive_failures <= OLLAMA_HEALTH_MAX_FAST_RETRIES: + delay = OLLAMA_HEALTH_RETRY_SECONDS + else: + delay = OLLAMA_HEALTH_BACKOFF_SECONDS + print(f"[OllamaAdapter] monitor: {exc}; retry #{consecutive_failures} in {delay}s") + await asyncio.sleep(delay) diff --git a/lending-poc/document_processing/translation/translation_service/config.py b/lending-poc/document_processing/translation/translation_service/config.py index 9bc7be7..ecfc5f3 100644 --- a/lending-poc/document_processing/translation/translation_service/config.py +++ b/lending-poc/document_processing/translation/translation_service/config.py @@ -70,6 +70,22 @@ def get_kb_path(domain: str) -> Path: # Model identifier passed to the chosen adapter. MODEL_NAME = os.getenv("OLLAMA_MODEL", "gemma4:e4b-it-qat") +# Background /health monitor timing (see OllamaAdapter._monitor_loop). The +# monitor pings Ollama with a real chat() call, untimed, on a background task +# — never inline in a request — so these control retry/recheck cadence only, +# never a request timeout. +# +# Fast retry interval while consecutive failures are within the limit below. +OLLAMA_HEALTH_RETRY_SECONDS = float(os.getenv("OLLAMA_HEALTH_RETRY_SECONDS", "5")) +# How many consecutive failures before backing off to the slower interval. +OLLAMA_HEALTH_MAX_FAST_RETRIES = int(os.getenv("OLLAMA_HEALTH_MAX_FAST_RETRIES", "12")) +# Slow retry interval once the fast-retry budget is exhausted — keeps trying +# forever, just less aggressively, so the service self-heals without a restart. +OLLAMA_HEALTH_BACKOFF_SECONDS = float(os.getenv("OLLAMA_HEALTH_BACKOFF_SECONDS", "120")) +# Reconfirmation interval once status is "ok", so a later Ollama outage is +# eventually reflected again instead of leaving /health stuck on stale "ok". +OLLAMA_HEALTH_RECHECK_SECONDS = float(os.getenv("OLLAMA_HEALTH_RECHECK_SECONDS", "30")) + # --------------------------------------------------------------------------- # Model options (adapter-specific — passed through as-is) # --------------------------------------------------------------------------- diff --git a/lending-poc/document_processing/translation/translation_service/translator.py b/lending-poc/document_processing/translation/translation_service/translator.py index baef0b5..e11af9d 100644 --- a/lending-poc/document_processing/translation/translation_service/translator.py +++ b/lending-poc/document_processing/translation/translation_service/translator.py @@ -112,6 +112,18 @@ def health_check(self) -> bool: """Delegate health check to the underlying model adapter.""" return self._adapter.health_check() + def health_status(self) -> str: + """Delegate the non-blocking health status read to the underlying adapter.""" + return self._adapter.health_status() + + async def start_health_monitor(self) -> None: + """Delegate starting the background health monitor to the underlying adapter.""" + await self._adapter.start_monitoring() + + async def stop_health_monitor(self) -> None: + """Delegate stopping the background health monitor to the underlying adapter.""" + await self._adapter.stop_monitoring() + def kb_size(self) -> int: """Return the number of terminology entries loaded from the KB.""" return len(self._kb) diff --git a/lending-poc/gateway/main.py b/lending-poc/gateway/main.py index 6e7b551..671fb90 100644 --- a/lending-poc/gateway/main.py +++ b/lending-poc/gateway/main.py @@ -24,6 +24,15 @@ TRANSLATION_BASE_URL = os.environ.get("TRANSLATION_BASE_URL", "http://127.0.0.1:8001") FIELD_MAPPING_BASE_URL = os.environ.get("FIELD_MAPPING_BASE_URL", "http://127.0.0.1:8002") +# Per-service request timeouts. Each backend calls out to a local model (OCR +# to Surya, translation/field-mapping to Ollama) and can legitimately run for +# minutes — these match the timeouts the frontend already budgets for the +# same calls (see frontend/src/api/{extract,translation,fieldMapping}.ts), so +# the gateway is never the first link in the chain to give up. +OCR_REQUEST_TIMEOUT_SECONDS = float(os.environ.get("OCR_REQUEST_TIMEOUT_SECONDS", "300")) +TRANSLATION_REQUEST_TIMEOUT_SECONDS = float(os.environ.get("TRANSLATION_REQUEST_TIMEOUT_SECONDS", "300")) +FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS = float(os.environ.get("FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS", "300")) + # Headers that must not be forwarded as-is between hops (RFC 7230) plus a few # that httpx/Starlette will recompute themselves and that would otherwise # desync from the body we're actually sending/returning. @@ -53,7 +62,7 @@ async def lifespan(app: FastAPI): ) -async def _proxy(request: Request, base_url: str, path: str) -> Response: +async def _proxy(request: Request, base_url: str, path: str, timeout: float) -> Response: client: httpx.AsyncClient = request.app.state.http headers = {k: v for k, v in request.headers.items() if k.lower() not in REQUEST_STRIP_HEADERS} body = await request.body() @@ -64,6 +73,7 @@ async def _proxy(request: Request, base_url: str, path: str) -> Response: headers=headers, params=list(request.query_params.multi_items()), content=body, + timeout=timeout, ) except httpx.RequestError as exc: return JSONResponse( @@ -82,39 +92,43 @@ async def _proxy(request: Request, base_url: str, path: str) -> Response: @app.post("/extract") async def extract(request: Request) -> Response: - return await _proxy(request, OCR_BASE_URL, "/extract") + return await _proxy(request, OCR_BASE_URL, "/extract", timeout=OCR_REQUEST_TIMEOUT_SECONDS) @app.post("/translate/text") async def translate_text(request: Request) -> Response: - return await _proxy(request, TRANSLATION_BASE_URL, "/translate/text") + return await _proxy(request, TRANSLATION_BASE_URL, "/translate/text", timeout=TRANSLATION_REQUEST_TIMEOUT_SECONDS) @app.post("/translate/files") async def translate_files(request: Request) -> Response: - return await _proxy(request, TRANSLATION_BASE_URL, "/translate/files") + return await _proxy(request, TRANSLATION_BASE_URL, "/translate/files", timeout=TRANSLATION_REQUEST_TIMEOUT_SECONDS) @app.post("/map") async def map_fields(request: Request) -> Response: - return await _proxy(request, FIELD_MAPPING_BASE_URL, "/map") + return await _proxy(request, FIELD_MAPPING_BASE_URL, "/map", timeout=FIELD_MAPPING_REQUEST_TIMEOUT_SECONDS) # --- Per-service health (namespaced since all three modules define /health) --- +# Liveness probes, not business calls — kept short regardless of the +# per-service request timeouts above, matching the aggregate /health below. +HEALTH_PROXY_TIMEOUT_SECONDS = 5.0 + @app.get("/ocr/health") async def ocr_health(request: Request) -> Response: - return await _proxy(request, OCR_BASE_URL, "/health") + return await _proxy(request, OCR_BASE_URL, "/health", timeout=HEALTH_PROXY_TIMEOUT_SECONDS) @app.get("/translation/health") async def translation_health(request: Request) -> Response: - return await _proxy(request, TRANSLATION_BASE_URL, "/health") + return await _proxy(request, TRANSLATION_BASE_URL, "/health", timeout=HEALTH_PROXY_TIMEOUT_SECONDS) @app.get("/field-mapping/health") async def field_mapping_health(request: Request) -> Response: - return await _proxy(request, FIELD_MAPPING_BASE_URL, "/health") + return await _proxy(request, FIELD_MAPPING_BASE_URL, "/health", timeout=HEALTH_PROXY_TIMEOUT_SECONDS) @app.get("/health") @@ -127,7 +141,7 @@ async def health(request: Request) -> dict: ("field_mapping", FIELD_MAPPING_BASE_URL), ): try: - resp = await client.get(f"{base}/health", timeout=5.0) + resp = await client.get(f"{base}/health", timeout=HEALTH_PROXY_TIMEOUT_SECONDS) statuses[name] = "healthy" if resp.status_code == 200 else f"unhealthy ({resp.status_code})" except httpx.RequestError: statuses[name] = "unreachable"