Keep status dependency checks from blocking probes

src/server/config.py

@@ -57,6 +57,12 @@ class ServerSettings(BaseSettings):
     rate_limit_enabled: bool = Field(default=True)
     rate_limit_per_minute: int = Field(default=100)
 
+    # Dependency status checks are used by the dashboard-facing /system/status
+    # endpoint. Keep them bounded so slow external services cannot block a
+    # single-worker server and starve readiness/liveness probes.
+    dependency_check_timeout_seconds: float = Field(default=2.0, gt=0)
+    dependency_status_cache_ttl_seconds: float = Field(default=10.0, gt=0)
+
     # Logging settings
     log_level: str = Field(default="INFO")
     # Default follows .env.example.full (LOG_FORMAT=json): machine-parseable,

src/server/utils/health_check.py

@@ -1,15 +1,105 @@
-"""
-Health check utilities for system dependencies
-"""
+"""Health check utilities for system dependencies."""
 
+import asyncio
+import logging
 import time
-from typing import Dict
+from concurrent.futures import Future, ThreadPoolExecutor
 from datetime import datetime
+from typing import Callable, Dict, Tuple
 
 from ..models.response import DependencyStatus
+from ..config import config
 
 
-async def check_database() -> DependencyStatus:
+logger = logging.getLogger("server")
+
+_DependencyProbe = Callable[[], DependencyStatus]
+_DEPENDENCY_STATUS_CACHE: Dict[str, Tuple[float, DependencyStatus]] = {}
+_DEPENDENCY_STATUS_LOCKS: Dict[str, asyncio.Lock] = {}
+_DEPENDENCY_STATUS_IN_FLIGHT: Dict[str, Future[DependencyStatus]] = {}
+_DEPENDENCY_PROBE_EXECUTOR = ThreadPoolExecutor(
+    max_workers=2,
+    thread_name_prefix="powermem-dependency-probe",
+)
+
+
+def _get_dependency_lock(name: str) -> asyncio.Lock:
+    lock = _DEPENDENCY_STATUS_LOCKS.get(name)
+    if lock is None:
+        lock = asyncio.Lock()
+        _DEPENDENCY_STATUS_LOCKS[name] = lock
+    return lock
+
+
+def _timeout_status(name: str, timeout_seconds: float, latency_ms: float) -> DependencyStatus:
+    return DependencyStatus(
+        name=name,
+        status="degraded",
+        latency_ms=round(latency_ms, 2),
+        error_message=(
+            f"Dependency check timed out after {timeout_seconds:g}s; "
+            "status probe skipped to keep API server responsive"
+        ),
+        last_checked=datetime.utcnow(),
+    )
+
+
+async def _run_probe_with_timeout(
+    name: str,
+    probe: _DependencyProbe,
+    timeout_seconds: float,
+) -> DependencyStatus:
+    start_time = time.time()
+    future = _DEPENDENCY_STATUS_IN_FLIGHT.get(name)
+    if future is None or future.done():
+        future = _DEPENDENCY_PROBE_EXECUTOR.submit(probe)
+        _DEPENDENCY_STATUS_IN_FLIGHT[name] = future
+
+    try:
+        return await asyncio.wait_for(
+            asyncio.wrap_future(future),
+            timeout=timeout_seconds,
+        )
+    except asyncio.TimeoutError:
+        latency_ms = (time.time() - start_time) * 1000
+        logger.warning(
+            "Dependency health check timed out",
+            extra={
+                "dependency": name,
+                "timeout_seconds": timeout_seconds,
+                "latency_ms": round(latency_ms, 2),
+            },
+        )
+        return _timeout_status(name, timeout_seconds, latency_ms)
+    finally:
+        if future.done() and _DEPENDENCY_STATUS_IN_FLIGHT.get(name) is future:
+            _DEPENDENCY_STATUS_IN_FLIGHT.pop(name, None)
+
+
+async def _cached_probe(
+    name: str,
+    probe: _DependencyProbe,
+    *,
+    timeout_seconds: float,
+    cache_ttl_seconds: float,
+) -> DependencyStatus:
+    now = time.monotonic()
+    cached = _DEPENDENCY_STATUS_CACHE.get(name)
+    if cached is not None and now - cached[0] < cache_ttl_seconds:
+        return cached[1]
+
+    async with _get_dependency_lock(name):
+        now = time.monotonic()
+        cached = _DEPENDENCY_STATUS_CACHE.get(name)
+        if cached is not None and now - cached[0] < cache_ttl_seconds:
+            return cached[1]
+
+        status = await _run_probe_with_timeout(name, probe, timeout_seconds)
+        _DEPENDENCY_STATUS_CACHE[name] = (time.monotonic(), status)
+        return status
+
+
+def _check_database_sync() -> DependencyStatus:
     """
     Check database health and measure latency
 
@@ -52,7 +142,35 @@ async def check_database() -> DependencyStatus:
         )
 
 
-async def check_llm() -> DependencyStatus:
+async def check_database(
+    *,
+    timeout_seconds: float | None = None,
+    cache_ttl_seconds: float | None = None,
+) -> DependencyStatus:
+    """
+    Check database health without blocking the event loop.
+
+    The actual probe may initialize storage clients and perform blocking I/O,
+    so run it in a worker thread with a short timeout.
+    """
+
+    return await _cached_probe(
+        "database",
+        _check_database_sync,
+        timeout_seconds=(
+            config.dependency_check_timeout_seconds
+            if timeout_seconds is None
+            else timeout_seconds
+        ),
+        cache_ttl_seconds=(
+            config.dependency_status_cache_ttl_seconds
+            if cache_ttl_seconds is None
+            else cache_ttl_seconds
+        ),
+    )
+
+
+def _check_llm_sync() -> DependencyStatus:
     """
     Check LLM provider health and measure latency
 
@@ -117,15 +235,50 @@ async def check_llm() -> DependencyStatus:
         )
 
 
-async def check_all_dependencies() -> Dict[str, DependencyStatus]:
+async def check_llm(
+    *,
+    timeout_seconds: float | None = None,
+    cache_ttl_seconds: float | None = None,
+) -> DependencyStatus:
+    """Check LLM provider health without blocking the event loop."""
+
+    return await _cached_probe(
+        "llm",
+        _check_llm_sync,
+        timeout_seconds=(
+            config.dependency_check_timeout_seconds
+            if timeout_seconds is None
+            else timeout_seconds
+        ),
+        cache_ttl_seconds=(
+            config.dependency_status_cache_ttl_seconds
+            if cache_ttl_seconds is None
+            else cache_ttl_seconds
+        ),
+    )
+
+
+async def check_all_dependencies(
+    *,
+    timeout_seconds: float | None = None,
+    cache_ttl_seconds: float | None = None,
+) -> Dict[str, DependencyStatus]:
     """
     Check all system dependencies
 
     Returns:
         Dictionary mapping dependency name to status
     """
-    database_status = await check_database()
-    llm_status = await check_llm()
+    database_status, llm_status = await asyncio.gather(
+        check_database(
+            timeout_seconds=timeout_seconds,
+            cache_ttl_seconds=cache_ttl_seconds,
+        ),
+        check_llm(
+            timeout_seconds=timeout_seconds,
+            cache_ttl_seconds=cache_ttl_seconds,
+        ),
+    )
 
     return {
         "database": database_status,

tests/unit/server/test_health_check.py

@@ -0,0 +1,138 @@
+import asyncio
+import threading
+import time
+
+import pytest
+
+from server.models.response import DependencyStatus
+from server.utils import health_check
+
+
+@pytest.fixture(autouse=True)
+def clear_dependency_probe_state():
+    health_check._DEPENDENCY_STATUS_CACHE.clear()
+    health_check._DEPENDENCY_STATUS_LOCKS.clear()
+    health_check._DEPENDENCY_STATUS_IN_FLIGHT.clear()
+    yield
+    health_check._DEPENDENCY_STATUS_CACHE.clear()
+    health_check._DEPENDENCY_STATUS_LOCKS.clear()
+    health_check._DEPENDENCY_STATUS_IN_FLIGHT.clear()
+
+
+@pytest.mark.asyncio
+async def test_dependency_status_timeout_does_not_block_event_loop(monkeypatch):
+    health_check._DEPENDENCY_STATUS_CACHE.clear()
+    health_check._DEPENDENCY_STATUS_LOCKS.clear()
+
+    def slow_database_probe():
+        time.sleep(0.2)
+        return DependencyStatus(name="database", status="healthy")
+
+    def fast_llm_probe():
+        return DependencyStatus(name="llm", status="disabled")
+
+    monkeypatch.setattr(health_check, "_check_database_sync", slow_database_probe)
+    monkeypatch.setattr(health_check, "_check_llm_sync", fast_llm_probe)
+
+    start = time.monotonic()
+    dependencies = await health_check.check_all_dependencies(
+        timeout_seconds=0.05,
+        cache_ttl_seconds=0.0,
+    )
+    elapsed = time.monotonic() - start
+
+    assert elapsed < 0.15
+    assert dependencies["database"].status == "degraded"
+    assert "timed out" in dependencies["database"].error_message
+    assert dependencies["llm"].status == "disabled"
+
+
+@pytest.mark.asyncio
+async def test_timed_out_dependency_reuses_in_flight_worker(monkeypatch):
+    calls = {"database": 0}
+    started = threading.Event()
+    release = threading.Event()
+    finished = threading.Event()
+
+    def blocked_database_probe():
+        calls["database"] += 1
+        started.set()
+        release.wait(timeout=1.0)
+        finished.set()
+        return DependencyStatus(name="database", status="healthy")
+
+    monkeypatch.setattr(health_check, "_check_database_sync", blocked_database_probe)
+
+    first = await health_check.check_database(
+        timeout_seconds=0.01,
+        cache_ttl_seconds=0.0,
+    )
+    assert started.wait(timeout=1.0)
+
+    second = await health_check.check_database(
+        timeout_seconds=0.01,
+        cache_ttl_seconds=0.0,
+    )
+
+    release.set()
+    assert finished.wait(timeout=1.0)
+
+    assert first.status == "degraded"
+    assert second.status == "degraded"
+    assert calls == {"database": 1}
+
+
+@pytest.mark.asyncio
+async def test_dependency_status_uses_short_ttl_cache(monkeypatch):
+    health_check._DEPENDENCY_STATUS_CACHE.clear()
+    health_check._DEPENDENCY_STATUS_LOCKS.clear()
+    calls = {"database": 0, "llm": 0}
+
+    def database_probe():
+        calls["database"] += 1
+        return DependencyStatus(name="database", status="healthy")
+
+    def llm_probe():
+        calls["llm"] += 1
+        return DependencyStatus(name="llm", status="disabled")
+
+    monkeypatch.setattr(health_check, "_check_database_sync", database_probe)
+    monkeypatch.setattr(health_check, "_check_llm_sync", llm_probe)
+
+    await health_check.check_all_dependencies(
+        timeout_seconds=0.1,
+        cache_ttl_seconds=60.0,
+    )
+    await health_check.check_all_dependencies(
+        timeout_seconds=0.1,
+        cache_ttl_seconds=60.0,
+    )
+
+    assert calls == {"database": 1, "llm": 1}
+
+
+@pytest.mark.asyncio
+async def test_dependency_status_coalesces_concurrent_same_dependency(monkeypatch):
+    health_check._DEPENDENCY_STATUS_CACHE.clear()
+    health_check._DEPENDENCY_STATUS_LOCKS.clear()
+    calls = {"database": 0}
+
+    def database_probe():
+        calls["database"] += 1
+        time.sleep(0.03)
+        return DependencyStatus(name="database", status="healthy")
+
+    monkeypatch.setattr(health_check, "_check_database_sync", database_probe)
+
+    results = await asyncio.gather(
+        *[
+            health_check.check_database(
+                timeout_seconds=0.2,
+                cache_ttl_seconds=60.0,
+            )
+            for _ in range(10)
+        ]
+    )
+
+    assert calls == {"database": 1}
+    assert {result.status for result in results} == {"healthy"}

tests/unit/server/test_settings.py

@@ -1,3 +1,7 @@
+import pytest
+from pydantic import ValidationError
+
+
 def test_server_settings_parsing(monkeypatch):
     monkeypatch.setenv("POWERMEM_SERVER_AUTH_ENABLED", "false")
     monkeypatch.setenv("POWERMEM_SERVER_LOG_FILE", "")
@@ -18,3 +22,19 @@ def test_server_settings_parsing(monkeypatch):
         "https://a.example",
         "https://b.example",
     ]
+
+
+@pytest.mark.parametrize(
+    "env_name",
+    [
+        "POWERMEM_SERVER_DEPENDENCY_CHECK_TIMEOUT_SECONDS",
+        "POWERMEM_SERVER_DEPENDENCY_STATUS_CACHE_TTL_SECONDS",
+    ],
+)
+def test_server_dependency_probe_settings_must_be_positive(monkeypatch, env_name):
+    from server.config import ServerSettings
+
+    monkeypatch.setenv(env_name, "0")
+
+    with pytest.raises(ValidationError, match="greater than 0"):
+        ServerSettings(_env_file=None)