feat(observability): diagnose blocked-loop vs pool-exhaustion on stalled /health

.env.example

@@ -238,6 +238,16 @@ HINDSIGHT_API_LOG_LEVEL=info
 # Expose async-operation queue + consolidation-backlog gauges on /metrics.
 # Runs periodic per-schema COUNT queries on a background task (disabled by default).
 # HINDSIGHT_API_METRICS_BACKLOG_ENABLED=true
+#
+# Runtime-stall observability (enabled by default). When a liveness probe fails,
+# these tell you WHY: a blocked event loop vs DB connection-pool exhaustion.
+# The loop watchdog logs the offending stack when the loop is unresponsive; the
+# DB-pool acquire timing logs (and exposes hindsight.db.pool.waiting) when
+# callers queue for a connection. Both are cheap; tune or disable if needed.
+# HINDSIGHT_API_LOOP_WATCHDOG_ENABLED=false
+# HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS=1000
+# HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS=250
+# HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS=1000
 
 # -----------------------------------------------------------------------------
 # Control Plane (Optional)

hindsight-api-slim/hindsight_api/api/http.py

@@ -3102,6 +3102,7 @@ async def lifespan(app: FastAPI):
         config = get_config()
         poller = None
         poller_task = None
+        loop_watchdog = None
 
         # Initialize OpenTelemetry metrics
         try:
@@ -3145,6 +3146,12 @@ async def lifespan(app: FastAPI):
                 metrics_collector.set_db_pool(memory._pool)
                 logging.info("DB pool metrics configured")
 
+        # Start the event-loop stall watchdog (logs the culprit stack if a task
+        # blocks the loop, so a failing /health can be told apart from pool exhaustion).
+        from ..loop_watchdog import start_loop_watchdog
+
+        loop_watchdog = start_loop_watchdog(asyncio.get_running_loop())
+
         # Start worker poller if the backend supports it.
         # All current backends (PostgreSQL, Oracle) support async worker/poller.
         if config.worker_enabled and memory._backend.supports_worker_poller:
@@ -3189,6 +3196,10 @@ async def lifespan(app: FastAPI):
 
         yield
 
+        # Stop the loop watchdog first so it doesn't fire during teardown.
+        if loop_watchdog is not None:
+            loop_watchdog.stop()
+
         # Shutdown worker poller if running
         if poller is not None:
             await poller.shutdown_graceful(timeout=30.0)

hindsight-api-slim/hindsight_api/config.py

@@ -493,6 +493,12 @@ def _resolve_operation_strict_schema(operation_env: str) -> bool:
 ENV_METRICS_INCLUDE_BANK_ID = "HINDSIGHT_API_METRICS_INCLUDE_BANK_ID"
 ENV_METRICS_BACKLOG_ENABLED = "HINDSIGHT_API_METRICS_BACKLOG_ENABLED"
 
+# Runtime-stall observability (loop watchdog + DB pool acquire instrumentation)
+ENV_LOOP_WATCHDOG_ENABLED = "HINDSIGHT_API_LOOP_WATCHDOG_ENABLED"
+ENV_LOOP_WATCHDOG_STALL_THRESHOLD_MS = "HINDSIGHT_API_LOOP_WATCHDOG_STALL_THRESHOLD_MS"
+ENV_LOOP_WATCHDOG_POLL_INTERVAL_MS = "HINDSIGHT_API_LOOP_WATCHDOG_POLL_INTERVAL_MS"
+ENV_DB_ACQUIRE_WARN_THRESHOLD_MS = "HINDSIGHT_API_DB_ACQUIRE_WARN_THRESHOLD_MS"
+
 # Vertex AI configuration
 ENV_LLM_VERTEXAI_PROJECT_ID = "HINDSIGHT_API_LLM_VERTEXAI_PROJECT_ID"
 ENV_LLM_VERTEXAI_REGION = "HINDSIGHT_API_LLM_VERTEXAI_REGION"
@@ -1158,6 +1164,16 @@ def _parse_strategy_boosts(raw: str | None) -> dict[str, str]:
 DEFAULT_METRICS_INCLUDE_BANK_ID = False  # Disabled by default to avoid high-cardinality OTel metric growth
 DEFAULT_METRICS_BACKLOG_ENABLED = False  # Disabled by default: runs periodic per-schema COUNT queries
 
+# Runtime-stall observability defaults. Both are cheap and on by default: the
+# watchdog is a single background thread pinging the loop; the DB-pool acquire
+# timing is a monotonic() delta per acquire. They turn a failing liveness probe
+# from "pod restarted, cause unknown" into a logged root cause (blocked loop vs
+# pool exhaustion).
+DEFAULT_LOOP_WATCHDOG_ENABLED = True
+DEFAULT_LOOP_WATCHDOG_STALL_THRESHOLD_MS = 1000  # log a stall once the loop is unresponsive this long
+DEFAULT_LOOP_WATCHDOG_POLL_INTERVAL_MS = 250  # how often the watchdog thread pings the loop
+DEFAULT_DB_ACQUIRE_WARN_THRESHOLD_MS = 1000  # log a warning when a pool acquire waits this long
+
 # Audit log defaults
 DEFAULT_AUDIT_LOG_ENABLED = False  # Disabled by default
 DEFAULT_AUDIT_LOG_ACTIONS = ""  # Empty = audit all eligible actions
@@ -2049,6 +2065,12 @@ class HindsightConfig:
     metrics_include_bank_id: bool
     metrics_backlog_enabled: bool
 
+    # Runtime-stall observability (static, server-level only)
+    loop_watchdog_enabled: bool
+    loop_watchdog_stall_threshold_ms: int
+    loop_watchdog_poll_interval_ms: int
+    db_acquire_warn_threshold_ms: int
+
     # Audit log configuration
     # audit_log_enabled is hierarchical (env -> tenant -> bank): a deployment can
     # audit some banks and not others. The actions allowlist and retention window
@@ -3168,6 +3190,18 @@ def from_env(cls) -> "HindsightConfig":
             in ("true", "1", "yes"),
             metrics_backlog_enabled=os.getenv(ENV_METRICS_BACKLOG_ENABLED, str(DEFAULT_METRICS_BACKLOG_ENABLED)).lower()
             in ("true", "1", "yes"),
+            # Runtime-stall observability (static, server-level only)
+            loop_watchdog_enabled=os.getenv(ENV_LOOP_WATCHDOG_ENABLED, str(DEFAULT_LOOP_WATCHDOG_ENABLED)).lower()
+            in ("true", "1", "yes"),
+            loop_watchdog_stall_threshold_ms=int(
+                os.getenv(ENV_LOOP_WATCHDOG_STALL_THRESHOLD_MS, str(DEFAULT_LOOP_WATCHDOG_STALL_THRESHOLD_MS))
+            ),
+            loop_watchdog_poll_interval_ms=int(
+                os.getenv(ENV_LOOP_WATCHDOG_POLL_INTERVAL_MS, str(DEFAULT_LOOP_WATCHDOG_POLL_INTERVAL_MS))
+            ),
+            db_acquire_warn_threshold_ms=int(
+                os.getenv(ENV_DB_ACQUIRE_WARN_THRESHOLD_MS, str(DEFAULT_DB_ACQUIRE_WARN_THRESHOLD_MS))
+            ),
             # Audit log configuration (static, server-level only)
             audit_log_enabled=os.getenv(ENV_AUDIT_LOG_ENABLED, str(DEFAULT_AUDIT_LOG_ENABLED)).lower() == "true",
             audit_log_actions=[

hindsight-api-slim/hindsight_api/engine/db/oracle.py

@@ -23,6 +23,8 @@
 from contextlib import asynccontextmanager
 from typing import Any, NamedTuple
 
+from .pool_instrumentation import PoolStats, acquire_conn
+
 
 class _OracleJSONEncoder(json.JSONEncoder):
     """JSON encoder that handles datetime and UUID objects."""
@@ -1246,6 +1248,7 @@ def __init__(self) -> None:
         # SESSION_USER so default-schema acquisitions can explicitly reset a
         # connection that was previously used for a tenant schema.
         self._default_schema: str | None = None
+        self._acquire_warn_threshold_s: float = 1.0
 
     async def initialize(
         self,
@@ -1261,6 +1264,10 @@ async def initialize(
         oracledb = _import_oracledb()
         self._oracledb = oracledb
 
+        from ...config import get_config
+
+        self._acquire_warn_threshold_s = get_config().db_acquire_warn_threshold_ms / 1000.0
+
         # Parse URL-format DSN (oracle://user:pass@host:port/service)
         from urllib.parse import urlparse
 
@@ -1322,10 +1329,23 @@ async def _set_session_schema(self, conn: Any) -> None:
             # expression" and aborts every acquire().
             cursor.close()
 
+    def _pool_stats(self) -> PoolStats | None:
+        """Snapshot for slow-acquire logs, from oracledb pool attributes."""
+        pool = self._pool
+        if pool is None:
+            return None
+        try:
+            busy = pool.busy
+            return PoolStats(in_use=busy, max=pool.max, idle=pool.opened - busy)
+        except Exception:
+            return None
+
     @asynccontextmanager
     async def acquire(self) -> AsyncIterator[OracleConnection]:
         pool = self._ensure_pool()
-        conn = await pool.acquire()
+        conn = await acquire_conn(
+            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+        )
         try:
             await self._set_session_schema(conn)
             yield OracleConnection(conn)
@@ -1341,7 +1361,9 @@ async def acquire(self) -> AsyncIterator[OracleConnection]:
     @asynccontextmanager
     async def transaction(self) -> AsyncIterator[OracleConnection]:
         pool = self._ensure_pool()
-        conn = await pool.acquire()
+        conn = await acquire_conn(
+            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+        )
         try:
             await self._set_session_schema(conn)
             yield OracleConnection(conn)

hindsight-api-slim/hindsight_api/engine/db/pool_instrumentation.py

@@ -0,0 +1,137 @@
+"""Instrumentation for database connection-pool acquisition.
+
+asyncpg exposes pool *size* and *idle* counts, but not how many callers are
+currently **queued waiting** for a connection — and that queue depth is the
+signal that actually distinguishes a saturated pool from a healthy one. When the
+pool is exhausted, ``/health`` (which itself acquires a connection to run
+``SELECT 1``) blocks in ``pool.acquire()`` until a connection frees or the acquire
+times out, so a liveness probe can fail **with the event loop completely idle**.
+
+This module tracks the process-wide count of in-flight acquisitions that have not
+yet obtained a connection, and times each acquire so a slow one logs with full
+pool stats. It is the DB-side counterpart to ``loop_watchdog`` (which covers loop
+stalls); together, a stuck ``/health`` can be attributed to either a blocked loop
+or pool exhaustion from the logs alone.
+
+The counter is a plain int mutated only from the event-loop thread (asyncpg
+acquisitions are awaited on the loop), so no lock is needed.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import AsyncIterator, Callable
+from contextlib import asynccontextmanager
+from dataclasses import dataclass
+from typing import Any
+
+logger = logging.getLogger("hindsight.db.pool")
+
+_waiting = 0  # callers currently blocked in pool.acquire(), process-wide
+
+
+@dataclass(frozen=True, slots=True)
+class PoolStats:
+    """Point-in-time connection-pool utilization snapshot."""
+
+    in_use: int
+    max: int
+    idle: int
+
+
+def waiting_count() -> int:
+    """Number of callers currently blocked waiting to acquire a pooled connection."""
+    return _waiting
+
+
+@asynccontextmanager
+async def instrument_acquire(
+    acquire_cm: Any,
+    *,
+    pool_stats: Callable[[], PoolStats | None] | None = None,
+    warn_threshold_s: float,
+) -> AsyncIterator[Any]:
+    """Wrap a pool's ``acquire()`` context manager with wait tracking + slow-acquire logging.
+
+    Args:
+        acquire_cm: an async context manager yielding a connection (e.g. the object
+            returned by ``asyncpg.Pool.acquire()``).
+        pool_stats: optional zero-arg callable returning a ``PoolStats`` snapshot for
+            the slow-acquire log line.
+        warn_threshold_s: log a warning when the acquire itself takes at least this long.
+
+    Yields:
+        The acquired connection.
+    """
+    global _waiting
+    _waiting += 1
+    start = time.monotonic()
+    acquired = False
+    try:
+        async with acquire_cm as conn:
+            acquired = True
+            _waiting -= 1
+            _record_acquire_wait(time.monotonic() - start, pool_stats, warn_threshold_s)
+            yield conn
+    finally:
+        # If __aenter__ raised (acquire timeout / cancellation), we never
+        # decremented above — do it here so the waiter count can't leak.
+        if not acquired:
+            _waiting -= 1
+
+
+async def acquire_conn(
+    acquire_awaitable: Any,
+    *,
+    pool_stats: Callable[[], PoolStats | None] | None = None,
+    warn_threshold_s: float,
+) -> Any:
+    """Await a pool acquire that returns a connection, with wait tracking + slow log.
+
+    For pools whose acquire is ``conn = await pool.acquire()`` (oracledb) rather than
+    an async context manager (asyncpg — use ``instrument_acquire`` for those). The
+    caller is responsible for releasing the returned connection.
+    """
+    global _waiting
+    _waiting += 1
+    start = time.monotonic()
+    try:
+        conn = await acquire_awaitable
+    finally:
+        _waiting -= 1
+    _record_acquire_wait(time.monotonic() - start, pool_stats, warn_threshold_s)
+    return conn
+
+
+def _record_acquire_wait(
+    wait_s: float,
+    pool_stats: Callable[[], PoolStats | None] | None,
+    warn_threshold_s: float,
+) -> None:
+    try:
+        from ...metrics import get_metrics_collector
+
+        get_metrics_collector().record_db_acquire_wait(wait_s)
+    except Exception:
+        pass
+
+    if wait_s < warn_threshold_s:
+        return
+
+    stats: PoolStats | None = None
+    if pool_stats is not None:
+        try:
+            stats = pool_stats()
+        except Exception:
+            stats = None
+    logger.warning(
+        "slow DB pool acquire: waited %.3fs for a connection "
+        "(in_use=%s max=%s idle=%s waiting=%s). The pool is likely saturated; "
+        "/health can stall on connection acquisition while the event loop is free.",
+        wait_s,
+        stats.in_use if stats else None,
+        stats.max if stats else None,
+        stats.idle if stats else None,
+        _waiting,
+    )

hindsight-api-slim/hindsight_api/engine/db/postgresql.py

@@ -15,6 +15,7 @@
 import asyncpg  # noqa: F401
 
 from .base import DatabaseBackend, DatabaseConnection
+from .pool_instrumentation import PoolStats, instrument_acquire
 
 logger = logging.getLogger(__name__)
 
@@ -76,6 +77,7 @@ def run_migrations(self, dsn: str, *, schema: str | None = None) -> None:
 
     def __init__(self) -> None:
         self._pool: asyncpg.Pool | None = None
+        self._acquire_warn_threshold_s: float = 1.0
 
     async def initialize(
         self,
@@ -88,6 +90,9 @@ async def initialize(
         statement_cache_size: int = 0,
         init_callback: Any | None = None,
     ) -> None:
+        from ...config import get_config
+
+        self._acquire_warn_threshold_s = get_config().db_acquire_warn_threshold_ms / 1000.0
         self._pool = await asyncpg.create_pool(
             dsn,
             min_size=min_size,
@@ -121,16 +126,28 @@ async def shutdown(self) -> None:
     def is_ready(self) -> bool:
         return self._pool is not None
 
+    def _pool_stats(self) -> PoolStats | None:
+        """Snapshot for slow-acquire logs. in_use = live connections minus idle ones."""
+        pool = self._pool
+        if pool is None:
+            return None
+        idle = pool.get_idle_size()
+        return PoolStats(in_use=pool.get_size() - idle, max=pool.get_max_size(), idle=idle)
+
     @asynccontextmanager
     async def acquire(self) -> AsyncIterator[PostgresConnection]:
         pool = self._ensure_pool()
-        async with pool.acquire() as conn:
+        async with instrument_acquire(
+            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+        ) as conn:
             yield PostgresConnection(conn)
 
     @asynccontextmanager
     async def transaction(self) -> AsyncIterator[PostgresConnection]:
         pool = self._ensure_pool()
-        async with pool.acquire() as conn:
+        async with instrument_acquire(
+            pool.acquire(), pool_stats=self._pool_stats, warn_threshold_s=self._acquire_warn_threshold_s
+        ) as conn:
             async with conn.transaction():
                 yield PostgresConnection(conn)