feat: bound crash-recovery respawn, expose spool depth, make worker timeouts configurable

context_intelligence_server/config.py

@@ -15,7 +15,7 @@
 import re
 from functools import lru_cache
 from pathlib import Path
-from typing import Any, Literal, Tuple, Type
+from typing import Any, Literal
 
 import yaml
 from pydantic import BaseModel, field_validator, model_validator
@@ -164,7 +164,7 @@ class YamlConfigSettingsSource(PydanticBaseSettingsSource):
 
     def __init__(
         self,
-        settings_cls: Type[BaseSettings],
+        settings_cls: type[BaseSettings],
         yaml_file: Path | None = None,
     ) -> None:
         super().__init__(settings_cls)
@@ -289,6 +289,26 @@ class Settings(BaseSettings):
     server_host: str = "0.0.0.0"
     server_port: int = 8000
 
+    # Gunicorn worker timeouts (run() in main.py). Both were hardcoded until
+    # the incident below made that a problem: a durable-spool boot whose
+    # crash-recovery work is legitimately O(backlog size) (see
+    # crash_recovery_respawn_limit above) can take minutes on a large
+    # backlog, and gunicorn's own worker-timeout watchdog cannot distinguish
+    # "still doing legitimate startup work" from "hung" -- it just SIGKILLs
+    # the worker either way, which then gets restarted by systemd and repeats
+    # the same slow boot forever. Defaults (30s / 10s) are UNCHANGED from the
+    # previous hardcoded values, so this PR is a no-op unless an operator
+    # opts in to raise them for a deployment that expects a slow/large-backlog
+    # boot.
+    #
+    # gunicorn_worker_timeout: seconds gunicorn allows a worker to go silent
+    #   (no heartbeat) before killing it. See gunicorn's `timeout` setting.
+    # gunicorn_graceful_timeout: seconds gunicorn waits for a worker to finish
+    #   handling in-flight work after SIGTERM before force-killing it. See
+    #   gunicorn's `graceful_timeout` setting.
+    gunicorn_worker_timeout: int = 30
+    gunicorn_graceful_timeout: int = 10
+
     # -------------------------------------------------------------------------
     # Authentication
     # -------------------------------------------------------------------------
@@ -776,6 +796,71 @@ def resolve_neo4j_query(self) -> Neo4jClientConfig:
     # (Neo4j default) holds all write_semaphore permits and stalls the pipeline.
     # Set to 0 to disable (no per-transaction timeout).
 
+    # Crash-recovery respawn ceiling (incident: a 38 GB / 583-file durable
+    # spool made cold start respawn 94/94 drainers before the server could
+    # accept a single request; startup took ~4 minutes and RSS peaked at
+    # 43.9 GB, tripping the kernel OOM killer -- which systemd then restarted,
+    # repeating the same unbounded respawn and never letting the backlog
+    # shrink). This mirrors write_concurrency's role as a hard ceiling on a
+    # startup-time resource cost, but bounds the RESPAWN LOOP in
+    # lifespan() (main.py) rather than write-flush concurrency: at most this
+    # many sessions from the recovered backlog get a drainer respawned on
+    # THIS boot; the remainder are DEFERRED -- left completely untouched on
+    # disk (still durable, still recoverable on a later boot, or instantly
+    # via get_or_create() the moment a new event for that session arrives
+    # through POST /events). A deferred backlog is never silent: lifespan()
+    # logs a WARNING naming the exact respawned/deferred counts and this
+    # setting, and /status's spool block (pending_sessions, spool_bytes_total)
+    # makes the backlog observable continuously, not just at boot.
+    #
+    # None (the default) preserves TODAY'S BEHAVIOUR EXACTLY: unbounded,
+    # every recovered session is respawned on this boot, matching every
+    # existing deployment -- this PR is a no-op unless an operator opts in
+    # by setting a finite ceiling.
+    crash_recovery_respawn_limit: int | None = None
+
+    @field_validator("crash_recovery_respawn_limit")
+    @classmethod
+    def _validate_crash_recovery_respawn_limit(cls, v: int | None) -> int | None:
+        """Fail loud on a negative ceiling; None (unbounded) and 0 are valid."""
+        if v is not None and v < 0:
+            raise ValueError(
+                "crash_recovery_respawn_limit must be a non-negative integer "
+                f"or null (unbounded), got {v}"
+            )
+        return v
+
+    # Crash-recovery deferred-backlog SWEEP interval (seconds). Only relevant
+    # when crash_recovery_respawn_limit is FINITE. Without this, a finite cap
+    # would drain the head of the backlog on boot and leave the deferred tail
+    # untouched until either a restart or a NEW event for that exact session
+    # arrives -- so a backlog of already-COMPLETED sessions (the incident's
+    # shape) would never drain at all, and a cap of 0 would strand EVERYTHING
+    # permanently. This sweep periodically re-runs recover() and tops the
+    # drainer pool back up to the ceiling: because respawn is idempotent
+    # (get_or_create) and recover() drops sessions the moment they finish, the
+    # number of live recovered drainers stays <= the ceiling while the deferred
+    # tail advances in deterministic sorted order as head sessions drain.
+    #
+    # Default 300s applies ONLY when a finite ceiling is set; with the default
+    # crash_recovery_respawn_limit=None (unbounded) there is no deferred tail
+    # and NO sweep task is ever started -- so every existing deployment is
+    # completely unaffected. Set to 0 to DISABLE the sweep even under a finite
+    # ceiling (the deferred tail then drains only on restart or a new event --
+    # an explicit, documented choice, not a silent surprise).
+    crash_recovery_sweep_interval_seconds: int = 300
+
+    @field_validator("crash_recovery_sweep_interval_seconds")
+    @classmethod
+    def _validate_crash_recovery_sweep_interval(cls, v: int) -> int:
+        """Fail loud on a negative interval; 0 (disabled) and positive are valid."""
+        if v < 0:
+            raise ValueError(
+                "crash_recovery_sweep_interval_seconds must be a non-negative "
+                f"integer (0 disables the sweep), got {v}"
+            )
+        return v
+
     # -------------------------------------------------------------------------
     # Logging
     # -------------------------------------------------------------------------
@@ -791,12 +876,12 @@ def resolve_neo4j_query(self) -> Neo4jClientConfig:
     @classmethod
     def settings_customise_sources(
         cls,
-        settings_cls: Type[BaseSettings],
+        settings_cls: type[BaseSettings],
         init_settings: PydanticBaseSettingsSource,
         env_settings: PydanticBaseSettingsSource,
         dotenv_settings: PydanticBaseSettingsSource,
         file_secret_settings: PydanticBaseSettingsSource,
-    ) -> Tuple[PydanticBaseSettingsSource, ...]:
+    ) -> tuple[PydanticBaseSettingsSource, ...]:
         # Priority: programmatic > env vars > YAML file > defaults
         return (
             init_settings,

context_intelligence_server/main.py

@@ -9,7 +9,7 @@
 import sys
 import time
 from collections.abc import AsyncGenerator
-from contextlib import asynccontextmanager
+from contextlib import asynccontextmanager, suppress
 from datetime import datetime
 from pathlib import Path
 from typing import Any
@@ -143,6 +143,67 @@ def _recover_one_session(
     return True
 
 
+async def _crash_recovery_topup(respawn_limit: int | None) -> int:
+    """One bounded crash-recovery pass: respawn drainers for up to
+    ``respawn_limit`` recovered sessions (all of them when ``None``).
+
+    This is the shared body of the boot-time recovery and the periodic sweep.
+    It is SAFE to call repeatedly on a live server because respawn is
+    idempotent -- ``registry.get_or_create`` returns the existing worker for a
+    session that already has a live drainer (no duplicate drainer, no reset).
+    And because ``recover()`` reports only sessions that still have undrained
+    data, a session drops out the moment it finishes, so the number of live
+    RECOVERED drainers stays <= ``respawn_limit`` while the deferred tail
+    advances in deterministic sorted order as head sessions drain.
+
+    Returns the number of sessions DISPATCHED to get_or_create on this pass --
+    an upper bound on newly-spawned drainers, since get_or_create is a no-op
+    for a session that already has a live drainer (see NOTE in the loop).
+    """
+    recovered = await registry.queue_manager.recover()
+    to_process = recovered if respawn_limit is None else recovered[:respawn_limit]
+    respawned = 0
+    for sid in to_process:
+        batch = await registry.queue_manager.read_batch(sid, max_items=1)
+        if not batch.lines:
+            continue
+        # NOTE: _recover_one_session returns True whenever it dispatched to
+        # get_or_create, whether or not a drainer already existed (get_or_create
+        # is idempotent). So this count is "sessions dispatched this pass", an
+        # upper bound on newly-spawned drainers -- fine for an INFO log.
+        if _recover_one_session(sid, batch.lines[0], registry.get_or_create):
+            respawned += 1
+    return respawned
+
+
+async def _crash_recovery_sweep_loop(interval: int, respawn_limit: int) -> None:
+    """Periodically top the recovered-drainer pool back up to the ceiling so a
+    finite ``crash_recovery_respawn_limit`` cannot permanently strand the
+    deferred backlog (the tail only advances as head sessions finish draining).
+
+    Started by ``lifespan`` ONLY when a finite ceiling is configured and the
+    interval is > 0; with the default unbounded ceiling there is no deferred
+    tail and this loop never runs. A single failed tick must never kill the
+    loop, so the body is guarded (CancelledError propagates for clean
+    shutdown; everything else is logged and the loop continues).
+    """
+    while True:
+        try:
+            await asyncio.sleep(interval)
+            respawned = await _crash_recovery_topup(respawn_limit)
+            if respawned:
+                logger.info(
+                    "crash_recovery_sweep: dispatched %d recovered session(s) "
+                    "(ceiling=%d) -- draining deferred backlog",
+                    respawned,
+                    respawn_limit,
+                )
+        except asyncio.CancelledError:
+            raise
+        except Exception as exc:  # noqa: BLE001 - a sweep tick must never kill the loop
+            logger.warning("crash_recovery_sweep: tick failed, will retry: %s", exc)
+
+
 @asynccontextmanager
 async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     """Manage application lifespan: configure logging and create shared Neo4j driver."""
@@ -230,21 +291,83 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     _accepted_seed, _written_seed = await registry.queue_manager.recovery_seed_counts()
     registry.seed_counters(_accepted_seed, _written_seed)
     recovered = await registry.queue_manager.recover()
+    # Bound how many drainers this boot respawns (incident: an unbounded
+    # backlog respawned 94/94 drainers before the server could serve a
+    # single request, driving a ~4 minute boot and 43.9 GB RSS that tripped
+    # the OOM killer -- which then never let the backlog shrink because
+    # every restart repeated the same unbounded respawn). None (the default)
+    # preserves today's behaviour exactly: every recovered session is
+    # processed on this boot, unbounded. `recovered` is already sorted
+    # (QueueManager.recover()), so which sessions are processed this boot
+    # vs. deferred is deterministic across restarts of the same backlog.
+    #
+    # Deferred sessions are NOT touched in any way here -- no read, no
+    # write, no drainer -- so they remain exactly as durable and
+    # recoverable as they were before this boot: a later boot's recover()
+    # call reports them again, and a new event for that session arriving
+    # via POST /events spawns its drainer immediately via get_or_create(),
+    # independent of this startup loop.
+    respawn_limit = _settings.crash_recovery_respawn_limit
+    if respawn_limit is not None and len(recovered) > respawn_limit:
+        to_process = recovered[:respawn_limit]
+        deferred_count = len(recovered) - respawn_limit
+    else:
+        to_process = recovered
+        deferred_count = 0
     respawned = 0
-    for sid in recovered:
+    for sid in to_process:
         batch = await registry.queue_manager.read_batch(sid, max_items=1)
         if not batch.lines:
             continue
         if _recover_one_session(sid, batch.lines[0], registry.get_or_create):
             respawned += 1
+    if deferred_count:
+        # Loud on purpose (WARNING, not INFO): a deferred backlog must never
+        # be a silent, un-discoverable fact -- that silence is exactly what
+        # let the 38 GB spool go unnoticed for two days in the incident this
+        # guards against. Names the exact counts and the setting to raise.
+        logger.warning(
+            "lifespan_startup: crash-recovery respawn cap reached "
+            "(crash_recovery_respawn_limit=%d): %d/%d respawned this boot, "
+            "%d session(s) deferred to a later boot (untouched on disk, "
+            "still fully recoverable). Raise crash_recovery_respawn_limit "
+            "to respawn more per boot.",
+            respawn_limit,
+            respawned,
+            len(to_process),
+            deferred_count,
+        )
     logger.info(
         "lifespan_startup: crash recovery respawned %d/%d drainers",
         respawned,
         len(recovered),
     )
+    # Periodic deferred-backlog sweep: only meaningful under a FINITE ceiling
+    # (a deferred tail can exist). With the default unbounded ceiling
+    # (respawn_limit is None) there is no deferred tail, so NO background task
+    # is started -- existing deployments are completely unaffected. When a
+    # finite ceiling IS set, this drains the deferred tail over time instead of
+    # stranding it until a restart or a new event (see _crash_recovery_sweep_loop
+    # and config.crash_recovery_sweep_interval_seconds).
+    _sweep_task: asyncio.Task[None] | None = None
+    _sweep_interval = _settings.crash_recovery_sweep_interval_seconds
+    if respawn_limit is not None and _sweep_interval > 0:
+        _sweep_task = asyncio.create_task(
+            _crash_recovery_sweep_loop(_sweep_interval, respawn_limit)
+        )
+        logger.info(
+            "crash_recovery_sweep: enabled (interval=%ds, ceiling=%d) -- "
+            "deferred backlog will drain progressively, not just on restart",
+            _sweep_interval,
+            respawn_limit,
+        )
     try:
         yield
     finally:
+        if _sweep_task is not None:
+            _sweep_task.cancel()
+            with suppress(asyncio.CancelledError):
+                await _sweep_task
         logger.info("lifespan_shutdown: closing Neo4j drivers")
         await app.state.neo4j_driver.close()
         await app.state.neo4j_query_driver.close()
@@ -635,6 +758,14 @@ async def get_status(request: Request) -> dict[str, Any]:
     # unauthenticated, so this block must NOT carry the per-key table or the
     # dead-letter listing — both are authenticated-only.
     response["metrics"] = await registry.pipeline_metrics()
+    # Additive, aggregate-only spool footprint (incident: a 38 GB / 583-file
+    # durable spool grew completely unnoticed -- the only symptom was a graph
+    # that had silently stopped updating). Same /status contract as `metrics`
+    # above: two aggregate integers only, no session ids, no workspace names,
+    # no per-key table. Cheap by construction (stat-only, short-TTL cached) --
+    # see QueueManager.spool_stats() for why this is safe on every poll even
+    # with a huge spool.
+    response["spool"] = await registry.queue_manager.spool_stats()
     # T5 (E): surface auth mode and admin-API capability so operators can
     # confirm admin is enabled without tailing startup logs.  /status is
     # unauthenticated — only config-level boolean flags are exposed here
@@ -879,8 +1010,8 @@ def load_config(self) -> None:
                 "bind": f"{_settings.server_host}:{_settings.server_port}",
                 "workers": workers,
                 "worker_class": "uvicorn.workers.UvicornWorker",
-                "timeout": 30,
-                "graceful_timeout": 10,
+                "timeout": _settings.gunicorn_worker_timeout,
+                "graceful_timeout": _settings.gunicorn_graceful_timeout,
                 "loglevel": _settings.log_level.lower(),
             }.items():
                 self.cfg.set(key, value)