diff --git a/context_intelligence_server/config.py b/context_intelligence_server/config.py index 3ea5fd17..179557a4 100644 --- a/context_intelligence_server/config.py +++ b/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, diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index d787e21e..5081f855 100644 --- a/context_intelligence_server/main.py +++ b/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) diff --git a/context_intelligence_server/queue_manager.py b/context_intelligence_server/queue_manager.py index 00ab2742..6fba186e 100644 --- a/context_intelligence_server/queue_manager.py +++ b/context_intelligence_server/queue_manager.py @@ -33,6 +33,13 @@ from pathlib import Path from typing import Any +# Fixed buffer size for streaming scans over a session ``.log`` (last-newline +# search and newline counting). Bounds boot-time and /status memory to O(chunk) +# instead of O(file): a durable log can be multi-GB (4.9 GB in the incident), +# and loading one into RAM just to count newlines is what drove ~44 GB RSS at +# startup. 1 MiB balances syscall count against per-scan memory. +_SCAN_CHUNK_BYTES = 1 << 20 + @dataclass(frozen=True) class Batch: @@ -62,6 +69,15 @@ def __init__(self, queues_dir: Path): self._stats_cache: dict[str, Any] | None = None self._stats_cache_at: float = 0.0 self._stats_cache_ttl: float = 1.0 + # Separate cache for spool_stats() (Change 2 / /status spool block). + # A longer TTL than _stats_cache_ttl is fine here: spool_stats() is an + # operator-facing "is the backlog growing" signal, not a + # correctness-sensitive value, so a few extra seconds of staleness is + # an acceptable trade for fewer directory scans under frequent + # /status polling. + self._spool_cache: dict[str, int] | None = None + self._spool_cache_at: float = 0.0 + self._spool_cache_ttl: float = 5.0 def _log_path(self, session_id: str) -> Path: return self._dir / f"{session_id}.log" @@ -86,13 +102,74 @@ def _complete_data_end(self, session_id: str) -> int: A torn trailing line (bytes after the final newline) is ignored: the returned offset is one past the last ``\\n``, or 0 when the log is missing or contains no complete line. + + Streams BACKWARD from EOF in fixed chunks to find the last ``\\n`` -- + O(tail) memory and I/O, never O(file). This log can be multi-GB (the + durable spool grew to a 4.9 GB single file in the incident); reading + the whole thing into RAM just to find the final newline is exactly the + boot-time memory blowup this avoids. + """ + path = self._log_path(session_id) + try: + with open(path, "rb") as f: + f.seek(0, os.SEEK_END) + pos = f.tell() + while pos > 0: + read_size = min(_SCAN_CHUNK_BYTES, pos) + pos -= read_size + f.seek(pos) + buf = f.read(read_size) + idx = buf.rfind(b"\n") + if idx != -1: + return pos + idx + 1 + return 0 + except FileNotFoundError: + return 0 + + @staticmethod + def _stream_newlines(path: Path, start: int = 0, end: int | None = None) -> int: + """Count ``\\n`` bytes in ``path``'s byte range ``[start, end)`` -- streamed. + + ``end=None`` counts to EOF. Reads the range in fixed-size chunks + (O(chunk) memory) instead of materialising the whole file (or a slice + copy of it) in RAM, which is what ``read_bytes()`` + + ``data[a:b].count(b"\\n")`` did on multi-GB spool files. Numerically + identical to that slice-count for any range; a missing file counts 0. + + Path-based (not session-id-based) so it serves both ``.log`` scans + (``_count_newlines``) and the whole-file ``.dead.jsonl`` count + (``_count_dead``). """ + if end is not None and end <= start: + return 0 try: - data = self._log_path(session_id).read_bytes() + with open(path, "rb") as f: + f.seek(start) + remaining = None if end is None else end - start + count = 0 + while True: + to_read = ( + _SCAN_CHUNK_BYTES + if remaining is None + else min(_SCAN_CHUNK_BYTES, remaining) + ) + if to_read <= 0: + break + buf = f.read(to_read) + if not buf: + break + count += buf.count(b"\n") + if remaining is not None: + remaining -= len(buf) + return count except FileNotFoundError: return 0 - last_nl = data.rfind(b"\n") - return last_nl + 1 if last_nl != -1 else 0 + + def _count_newlines( + self, session_id: str, start: int = 0, end: int | None = None + ) -> int: + """Streamed newline count over a session ``.log``'s ``[start, end)``.""" + return self._stream_newlines(self._log_path(session_id), start, end) @staticmethod def _validate_session_id(session_id: str) -> None: @@ -270,12 +347,13 @@ def _count_dead(self, worker_key: str) -> int: Returns 0 when no dead-letter file exists. Dead-letter records are always written newline-terminated, so counting newlines yields the number of complete records. + + Streamed (bounded memory), not ``read_bytes()``: a .dead.jsonl is + usually small but is NOT bounded -- a systematically-failing session + dead-letters every line -- and this is called on the same boot and + polled-/status paths as the .log scans. """ - try: - data = self._dead_path(worker_key).read_bytes() - except FileNotFoundError: - return 0 - return data.count(b"\n") + return self._stream_newlines(self._dead_path(worker_key)) def _all_worker_keys(self) -> list[str]: """Return the sorted union of ``.log`` and ``.dead.jsonl`` stems. @@ -321,17 +399,26 @@ def _all() -> dict[str, Any]: in_queue_total = 0 dead_total = 0 for worker_key in self._all_worker_keys(): - committed = self._read_committed_offset(worker_key) - in_queue = 0 try: - with open(self._log_path(worker_key), "rb") as f: - f.seek(committed) - data = f.read() - last_nl = data.rfind(b"\n") - if last_nl != -1: - in_queue = data[: last_nl + 1].count(b"\n") - except FileNotFoundError: - in_queue = 0 + committed = self._read_committed_offset(worker_key) + except (OSError, ValueError): + # /status calls this (via pipeline_metrics); a corrupt or + # transiently-unreadable .offset must NOT 500 the health + # probe. Degrade to 0 for this key's stats -- mirroring the + # existing missing-file->0 convention in + # _read_committed_offset, and tending the conservation + # residual negative (benign, never a false `degraded`). + # Deliberately NO logging here: /status is polled, and a + # per-scan warning on a persistently-corrupt offset would + # flood the hot path. The visibility signal is the aggregate + # `spool.corrupt_offsets` field (see spool_stats()). + committed = 0 + # Streamed count of complete lines from committed -> EOF. + # Equivalent to the old f.read() + data[:last_nl+1].count(b"\n") + # (every b"\n" lies at or before the last one), but without + # materialising the undrained tail -- which can be gigabytes + # under a large backlog on this (polled) /status path. + in_queue = self._count_newlines(worker_key, committed) dead = self._count_dead(worker_key) per_key.append( {"worker_key": worker_key, "in_queue": in_queue, "dead": dead} @@ -349,6 +436,129 @@ def _all() -> dict[str, Any]: self._stats_cache_at = now return stats + async def spool_stats(self) -> dict[str, int]: + """Cheap, aggregate-only spool footprint for the unauthenticated /status. + + Incident context: a durable spool silently grew to 38 GB across 583 + files (largest single file 4.9 GB) with ZERO signal anywhere that it + was happening -- the only symptom was a graph that had stopped + updating. This method exists so that number is always one field away. + + Returns exactly two aggregate integers: + + - ``pending_sessions``: count of worker keys with a ``.log`` file + whose committed offset is strictly less than the file's size, i.e. + there is unconsumed data (mirrors ``active_sessions()``'s + definition, but via ``stat()`` instead of a full scan-and-compare + pass, so it is safe to call on every /status hit). + - ``spool_bytes_total``: total bytes on disk across EVERY file in the + queue directory (``.log`` + ``.offset`` + ``.dead.jsonl``) -- the + same number an operator would get from ``du`` on the spool + directory, without shelling out. + + CHEAP BY CONSTRUCTION: this walks the directory and calls ``stat()`` + on each entry -- O(file count), NEVER O(file bytes). No file content + is read (unlike ``derive_all_stats()``, which tail-reads each log to + count pending lines). This is deliberately how a 38 GB spool can be + sized on every /status poll without walking 38 GB of content. + On top of that, results are cached for ``_spool_cache_ttl`` seconds + (monotonic clock) so a deployment with a very large number of spool + files (thousands of sessions) still does not pay a full directory + scan on every request. + + Per the /status aggregate-only contract (D3): NO session ids, NO + workspace names, and NO per-key table are returned or computable from + this result -- two integers only. + + HEALTH-ENDPOINT SAFE: /status is the unauthenticated health probe (the + ACA liveness surface). This method therefore MUST NOT be able to raise + out to the /status handler -- an uncaught exception there becomes a 500, + a failed health probe, and a container restart loop. Two degradation + rules make that impossible: + + - A directory-level failure (the queue dir missing/unavailable -- e.g. + an Azure Files SMB remount -- or any transient OS error while + scanning) returns the degraded sentinel ``{-1, -1}`` instead of + raising. Unlike every sibling reader, which uses ``glob()`` (empty on + a missing dir), this scan uses ``iterdir()`` (raises on a missing + dir), so the guard is mandatory, not cosmetic. The sentinel is NOT + cached, so the very next poll re-scans and recovers the real numbers + the moment the filesystem is healthy again. + - A per-file failure (a raced delete, or a corrupt/unreadable + ``.offset``) skips just that entry rather than failing the whole + aggregate. + + A ``-1`` in either field is the operator-visible "spool footprint + temporarily unavailable" signal -- distinct from a real ``0`` -- and + never leaks any identifier. + """ + now = time.monotonic() + if ( + self._spool_cache is not None + and (now - self._spool_cache_at) < self._spool_cache_ttl + ): + return self._spool_cache + + def _scan() -> dict[str, int]: + spool_bytes_total = 0 + pending_sessions = 0 + corrupt_offsets = 0 + for entry in self._dir.iterdir(): + if not entry.is_file(): + continue + try: + size = entry.stat().st_size + except FileNotFoundError: + # Raced with a concurrent delete_drained()/purge; the + # entry no longer exists -- simply exclude it, don't fail + # a cheap, best-effort aggregate over a live directory. + continue + spool_bytes_total += size + if entry.suffix == ".log": + try: + committed = self._read_committed_offset(entry.stem) + except ValueError: + # The .offset exists but is not a valid integer -- a + # GENUINELY corrupt offset. This is the one visibility + # signal for it (no logging anywhere, to avoid flooding + # the polled health path): surface it as an aggregate + # count on /status so `spool.corrupt_offsets > 0` is the + # operator's alarm. Count this file's bytes; skip its + # pending calc. + corrupt_offsets += 1 + continue + except OSError: + # A transient/racing FS error reading the offset (NOT + # corruption): count bytes, skip pending calc, and do + # NOT inflate corrupt_offsets with a non-corruption cause. + continue + if committed < size: + pending_sessions += 1 + return { + "pending_sessions": pending_sessions, + "spool_bytes_total": spool_bytes_total, + "corrupt_offsets": corrupt_offsets, + } + + try: + stats = await asyncio.to_thread(_scan) + except (OSError, ValueError): + # Queue dir missing/unavailable (e.g. Azure Files SMB remount) or a + # transient FS error mid-scan. /status is the health probe and MUST + # return 200 -- degrade to a sentinel and DO NOT cache it, so the + # next poll retries immediately once the filesystem recovers. All + # three fields are -1 = "temporarily unavailable" (distinct from a + # real 0, and from a real corrupt_offsets count). + return { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + } + + self._spool_cache = stats + self._spool_cache_at = now + return stats + async def dead_letter_keys(self) -> list[str]: """Return sorted worker keys that have a ``.dead.jsonl`` file. @@ -426,12 +636,12 @@ def _seed() -> tuple[int, int]: committed = self._read_committed_offset(key) complete_end = self._complete_data_end(key) dead = self._count_dead(key) - try: - data = self._log_path(key).read_bytes() - except FileNotFoundError: - data = b"" - before = data[:committed].count(b"\n") - pending = data[committed:complete_end].count(b"\n") + # Streamed newline counts over byte ranges -- numerically + # identical to the old data[:committed].count(b"\n") / + # data[committed:complete_end].count(b"\n"), but without loading + # the whole (possibly multi-GB) log or its slice copies at boot. + before = self._count_newlines(key, 0, committed) + pending = self._count_newlines(key, committed, complete_end) written_seed = max(0, before - dead) accepted += written_seed + pending + dead written += written_seed diff --git a/server-config.example.yaml b/server-config.example.yaml index 81b3560c..af00dcff 100644 --- a/server-config.example.yaml +++ b/server-config.example.yaml @@ -149,6 +149,42 @@ queues_path: /data/queues # Durable per-session append-logs; mirrors bl write_concurrency: 8 # Max concurrent Neo4j write flushes across all session drainers (starvation guard) max_delivery_attempts: 5 # Flush retries for one batch before its offending line is dead-lettered +# --- Crash-recovery of the durable queue on startup --- +# On boot the server respawns one drainer per session that still has undrained +# data. A very large backlog (measured: 38 GB / 583 files) can respawn so many +# drainers that boot takes minutes and RSS spikes hard enough to be OOM-killed, +# which then restart-loops and never lets the backlog shrink. +# +# crash_recovery_respawn_limit: ceiling on how many drainers a single boot +# respawns. null (DEFAULT) = unbounded = respawn EVERY recovered session on +# this boot (the original, pre-hardening behaviour). A finite value caps the +# respawn to that many sessions per pass; the remainder are DEFERRED. +# WARNING: 0 means "never auto-respawn recovered sessions". Combined with a +# disabled sweep (below), a 0 or finite cap would strand the deferred +# backlog until a restart or a NEW event for that session arrives -- for a +# backlog of already-completed sessions that never happens. Leave the sweep +# enabled (its default) so a finite cap drains safely over time. +crash_recovery_respawn_limit: null + +# crash_recovery_sweep_interval_seconds: how often (seconds) a background sweep +# re-runs recovery and tops the drainer pool back up to the ceiling, so a +# FINITE crash_recovery_respawn_limit drains its deferred tail progressively +# instead of stranding it. Only runs when a finite ceiling is set; with the +# default null ceiling there is no deferred tail and NO sweep task starts +# (existing deployments are unaffected). 0 disables the sweep even under a +# finite ceiling (explicit opt-out: the tail then drains only on restart or a +# new event). +crash_recovery_sweep_interval_seconds: 300 + +# --- Gunicorn worker watchdog timeouts (seconds) --- +# gunicorn_worker_timeout: how long gunicorn lets a worker go silent before it +# SIGKILLs it. A legitimately slow, large-backlog boot can exceed the 30s +# default and be killed mid-startup; raise it for deployments that expect a +# slow boot. gunicorn_graceful_timeout: shutdown grace window after SIGTERM. +# Both defaults are UNCHANGED from the previously-hardcoded values. +gunicorn_worker_timeout: 30 +gunicorn_graceful_timeout: 10 + # ----------------------------------------------------------------------------- # Server bind address # These two values are combined into the HTTP URL the server binds and listens on. diff --git a/tests/test_config.py b/tests/test_config.py index da8a2e9e..957d71a8 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -71,6 +71,114 @@ def test_settings_has_durable_queue_defaults(): assert s.max_delivery_attempts == 5 +# --------------------------------------------------------------------------- +# crash_recovery_respawn_limit (Change 1) +# --------------------------------------------------------------------------- + + +def test_crash_recovery_respawn_limit_defaults_to_unbounded(): + """Default MUST preserve today's behaviour exactly: unbounded (None).""" + from context_intelligence_server.config import Settings + + s = Settings() + assert s.crash_recovery_respawn_limit is None + + +def test_crash_recovery_sweep_interval_defaults_to_300(): + """A finite ceiling drains its deferred tail via a periodic sweep; the + default interval must be a sane positive value so a finite cap is safe + out of the box (not silently stranded).""" + from context_intelligence_server.config import Settings + + assert Settings().crash_recovery_sweep_interval_seconds == 300 + + +def test_crash_recovery_sweep_interval_accepts_zero_and_positive(): + from context_intelligence_server.config import Settings + + assert ( + Settings( + crash_recovery_sweep_interval_seconds=0 + ).crash_recovery_sweep_interval_seconds + == 0 + ) + assert ( + Settings( + crash_recovery_sweep_interval_seconds=60 + ).crash_recovery_sweep_interval_seconds + == 60 + ) + + +def test_crash_recovery_sweep_interval_rejects_negative(): + import pytest + + from context_intelligence_server.config import Settings + + with pytest.raises(ValueError): + Settings(crash_recovery_sweep_interval_seconds=-1) + + +def test_crash_recovery_respawn_limit_accepts_zero_and_positive(): + from context_intelligence_server.config import Settings + + assert Settings(crash_recovery_respawn_limit=0).crash_recovery_respawn_limit == 0 + assert Settings(crash_recovery_respawn_limit=25).crash_recovery_respawn_limit == 25 + + +def test_crash_recovery_respawn_limit_rejects_negative(): + from context_intelligence_server.config import Settings + + with pytest.raises(ValueError, match="crash_recovery_respawn_limit"): + Settings(crash_recovery_respawn_limit=-1) + + +def test_crash_recovery_respawn_limit_env_override(monkeypatch): + monkeypatch.setenv( + "AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_CRASH_RECOVERY_RESPAWN_LIMIT", "10" + ) + from context_intelligence_server.config import Settings + + assert Settings().crash_recovery_respawn_limit == 10 + + +# --------------------------------------------------------------------------- +# gunicorn_worker_timeout / gunicorn_graceful_timeout (Change 3) +# --------------------------------------------------------------------------- + + +def test_gunicorn_timeout_defaults_match_previous_hardcoded_values(): + """Defaults MUST equal the values that used to be hardcoded in run() + (main.py) so an operator who sets nothing sees no behaviour change.""" + from context_intelligence_server.config import Settings + + s = Settings() + assert s.gunicorn_worker_timeout == 30 + assert s.gunicorn_graceful_timeout == 10 + + +def test_gunicorn_timeout_overridable(): + from context_intelligence_server.config import Settings + + s = Settings(gunicorn_worker_timeout=300, gunicorn_graceful_timeout=60) + assert s.gunicorn_worker_timeout == 300 + assert s.gunicorn_graceful_timeout == 60 + + +def test_gunicorn_timeout_env_override(monkeypatch): + monkeypatch.setenv( + "AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_GUNICORN_WORKER_TIMEOUT", "120" + ) + monkeypatch.setenv( + "AMPLIFIER_CONTEXT_INTELLIGENCE_SERVER_GUNICORN_GRACEFUL_TIMEOUT", "20" + ) + from context_intelligence_server.config import Settings + + s = Settings() + assert s.gunicorn_worker_timeout == 120 + assert s.gunicorn_graceful_timeout == 20 + + # --------------------------------------------------------------------------- # YAML config file tests # --------------------------------------------------------------------------- diff --git a/tests/test_main.py b/tests/test_main.py index f7530e90..302fa1bf 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -836,6 +836,304 @@ async def test_lifespan_skips_recovery_for_empty_workspace( assert spawned == [] +# --------------------------------------------------------------------------- +# Bounded crash-recovery respawn (Change 1): crash_recovery_respawn_limit +# --------------------------------------------------------------------------- + + +async def _seed_recoverable_session(qm: Any, sid: str, workspace: str) -> None: + body = json.dumps( + { + "event": "tool_use", + "workspace": workspace, + "data": {"session_id": sid}, + } + ).encode("utf-8") + await qm.append(sid, body) + + +async def test_lifespan_default_respawns_all_recovered_sessions_unbounded( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Default (crash_recovery_respawn_limit=None) MUST preserve today's + behaviour exactly: every recovered session is respawned on this boot, + no matter how many there are.""" + qm = registry.queue_manager + sids = [f"sess-unbounded-{i}" for i in range(10)] + for sid in sids: + await _seed_recoverable_session(qm, sid, "/ws") + + spawned: list[tuple] = [] + monkeypatch.setattr( + registry, "get_or_create", lambda s, w, **kw: spawned.append((s, w)) + ) + assert main_module._settings.crash_recovery_respawn_limit is None + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + ): + async with lifespan(main_module.app): + pass + + assert {s for s, _w in spawned} == set(sids) + + +async def test_lifespan_respawn_cap_defers_remainder_and_logs_warning( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """With a finite crash_recovery_respawn_limit, only that many sessions + are respawned THIS boot; the remainder are deferred and a WARNING names + the exact respawned/deferred counts + the setting to raise.""" + qm = registry.queue_manager + sids = [f"sess-cap-{i}" for i in range(5)] + for sid in sids: + await _seed_recoverable_session(qm, sid, "/ws") + + spawned: list[tuple] = [] + monkeypatch.setattr( + registry, "get_or_create", lambda s, w, **kw: spawned.append((s, w)) + ) + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", 2) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.WARNING, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + pass + + # Exactly the cap's worth of sessions were respawned -- never more. + assert len(spawned) == 2 + # The un-respawned sessions were NEVER passed to get_or_create at all. + assert {s for s, _w in spawned}.issubset(set(sids)) + + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert any("crash-recovery respawn cap reached" in r.getMessage() for r in warnings) + cap_warning = next( + r for r in warnings if "crash-recovery respawn cap reached" in r.getMessage() + ) + msg = cap_warning.getMessage() + assert "2/2 respawned" in msg # respawned/attempted this boot + assert "3 session(s) deferred" in msg # 5 recovered - 2 processed = 3 + assert "crash_recovery_respawn_limit" in msg + + +async def test_lifespan_deferred_sessions_untouched_and_recoverable_next_boot( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Deferred sessions are left completely untouched on disk -- no read, no + write -- so a SUBSEQUENT boot's recover() call reports them again and can + respawn them (no data loss, no corruption).""" + qm = registry.queue_manager + sids = [f"sess-defer-{i}" for i in range(4)] + for sid in sids: + await _seed_recoverable_session(qm, sid, "/ws") + + spawned_boot1: list[tuple] = [] + monkeypatch.setattr( + registry, "get_or_create", lambda s, w, **kw: spawned_boot1.append((s, w)) + ) + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", 1) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + ): + async with lifespan(main_module.app): + pass + + assert len(spawned_boot1) == 1 + deferred_sids = set(sids) - {s for s, _w in spawned_boot1} + assert len(deferred_sids) == 3 + + # The deferred sessions' queue lines are STILL fully intact and + # recoverable: recover() (a fresh scan, same on-disk state) reports them + # again, exactly as before this boot ran. + recovered_again = await qm.recover() + assert deferred_sids <= set(recovered_again) + for sid in deferred_sids: + batch = await qm.read_batch(sid, max_items=1) + assert batch.lines != [] # data neither dropped nor corrupted + + +async def test_lifespan_respawn_cap_zero_defers_everything( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A cap of 0 is a valid opt-in (never respawn automatically at boot); + every recovered session is deferred, none is touched.""" + qm = registry.queue_manager + sid = "sess-cap-zero" + await _seed_recoverable_session(qm, sid, "/ws") + + spawned: list[tuple] = [] + monkeypatch.setattr( + registry, "get_or_create", lambda s, w, **kw: spawned.append((s, w)) + ) + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", 0) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + ): + async with lifespan(main_module.app): + pass + + assert spawned == [] + recovered_again = await qm.recover() + assert sid in recovered_again + + +# --------------------------------------------------------------------------- +# Deferred-backlog sweep (ci_pr73-rt7): a finite crash_recovery_respawn_limit +# must NOT permanently strand the deferred tail. The periodic sweep re-runs +# recovery and tops the drainer pool up to the ceiling as head sessions drain. +# --------------------------------------------------------------------------- + + +async def test_crash_recovery_topup_drains_deferred_tail_across_passes( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The deferred tail is not stranded: with ceiling=2, pass 1 dispatches the + 2 head sessions; once those finish draining (drop out of recover()), pass 2 + dispatches the previously-deferred 2. Live recovered drainers never exceed + the ceiling.""" + qm = registry.queue_manager + sids = sorted(f"sess-sweep-{i}" for i in range(4)) + for sid in sids: + await _seed_recoverable_session(qm, sid, "/ws") + + spawned: list[str] = [] + monkeypatch.setattr(registry, "get_or_create", lambda s, w, **kw: spawned.append(s)) + + # Pass 1: only the ceiling's worth (2) are dispatched; the tail is deferred. + dispatched = await main_module._crash_recovery_topup(2) + assert dispatched == 2 + assert set(spawned) == set(sids[:2]) + + # The 2 head sessions finish draining -> commit them to EOF so recover() + # stops reporting them (exactly what a real drainer does on completion). + for sid in sids[:2]: + batch = await qm.read_batch(sid, max_items=10) + await qm.commit(sid, batch.end_offset) + + # Pass 2: the previously-DEFERRED tail is now dispatched -- not stranded. + spawned.clear() + dispatched = await main_module._crash_recovery_topup(2) + assert dispatched == 2 + assert set(spawned) == set(sids[2:]) + + +async def test_lifespan_enables_sweep_under_finite_limit( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A finite ceiling with a positive interval starts the background sweep + (logged), so the deferred tail drains progressively rather than only on + restart.""" + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", 2) + monkeypatch.setattr( + main_module._settings, "crash_recovery_sweep_interval_seconds", 300 + ) + monkeypatch.setattr(registry, "get_or_create", lambda *a, **kw: MagicMock()) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.INFO, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + pass # task is created on entry and cancelled cleanly on exit + + assert any( + "crash_recovery_sweep: enabled" in r.getMessage() for r in caplog.records + ) + + +async def test_lifespan_no_sweep_when_limit_unbounded( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """Default (unbounded) ceiling: there is no deferred tail, so NO sweep task + is started -- existing deployments are completely unaffected.""" + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", None) + monkeypatch.setattr(registry, "get_or_create", lambda *a, **kw: MagicMock()) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.INFO, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + pass + + assert not any( + "crash_recovery_sweep: enabled" in r.getMessage() for r in caplog.records + ) + + +async def test_lifespan_no_sweep_when_interval_zero( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """interval=0 is an explicit opt-out: even under a finite ceiling the sweep + is not started (documented tradeoff: tail drains only on restart/new event).""" + monkeypatch.setattr(main_module._settings, "crash_recovery_respawn_limit", 2) + monkeypatch.setattr( + main_module._settings, "crash_recovery_sweep_interval_seconds", 0 + ) + monkeypatch.setattr(registry, "get_or_create", lambda *a, **kw: MagicMock()) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.main.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.INFO, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + pass + + assert not any( + "crash_recovery_sweep: enabled" in r.getMessage() for r in caplog.records + ) + + # --------------------------------------------------------------------------- # Cold start FAILS LOUD on schema/data corruption that requires `doctor # --fix` (design decision, reversing the lifespan half of f4d8bab): an @@ -1209,6 +1507,124 @@ async def test_status_includes_metrics_block(client: httpx.AsyncClient) -> None: assert "dead_letters" not in data +# --------------------------------------------------------------------------- +# /status spool block (Change 2): pending_sessions + spool_bytes_total +# --------------------------------------------------------------------------- + + +async def test_status_includes_spool_block(client: httpx.AsyncClient) -> None: + """/status carries an additive, aggregate-only spool block so a growing + on-disk backlog is never invisible (the 38 GB / two-day incident this + guards against had zero signal anywhere).""" + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + + assert "spool" in data + spool = data["spool"] + assert set(spool.keys()) == { + "pending_sessions", + "spool_bytes_total", + "corrupt_offsets", + } + assert isinstance(spool["pending_sessions"], int) + assert isinstance(spool["spool_bytes_total"], int) + assert isinstance(spool["corrupt_offsets"], int) + + +async def test_status_spool_block_reflects_real_backlog( + client: httpx.AsyncClient, +) -> None: + """The spool block's numbers move when there's real undrained data on + disk -- not a hardcoded placeholder.""" + qm = registry.queue_manager + body = json.dumps( + { + "event": "tool_use", + "workspace": "/ws", + "data": {"session_id": "sess-spool-visible"}, + } + ).encode("utf-8") + await qm.append("sess-spool-visible", body) + # get_or_create is bypassed here (raw append only) so this line stays + # undrained -- exactly the "pending" shape spool_stats() measures. + + response = await client.get("/status") + data = response.json() + + assert data["spool"]["pending_sessions"] >= 1 + assert data["spool"]["spool_bytes_total"] > 0 + + +async def test_status_spool_block_never_leaks_session_identifiers( + client: httpx.AsyncClient, +) -> None: + """/status is unauthenticated: the spool block must never carry a + session id, workspace name, or any per-key table (aggregate-only).""" + qm = registry.queue_manager + secret_sid = "super-secret-session-id-should-not-leak" + body = json.dumps( + { + "event": "tool_use", + "workspace": "/very/private/workspace", + "data": {"session_id": secret_sid}, + } + ).encode("utf-8") + await qm.append(secret_sid, body) + + response = await client.get("/status") + raw_text = response.text + + assert secret_sid not in raw_text + assert "/very/private/workspace" not in raw_text + assert "per_key" not in response.json()["spool"] + + +async def test_status_corrupt_offset_returns_200_and_surfaces_count( + client: httpx.AsyncClient, +) -> None: + """Regression (ci_pr73-267): a corrupt .offset previously 500'd /status via + derive_all_stats() (which runs before spool_stats in get_status). /status + must now stay 200 AND surface the corruption as spool.corrupt_offsets -- the + only signal (no logging, so the polled health path is never flooded).""" + qm = registry.queue_manager + await qm.append("sess-corrupt-offset", b"a") + qm._offset_path("sess-corrupt-offset").write_text("not-a-number", encoding="utf-8") + qm._spool_cache = None # bypass TTL cache so the corruption is seen now + + response = await client.get("/status") + + assert response.status_code == 200 # was 500 before the fix + assert response.json()["spool"]["corrupt_offsets"] >= 1 + + +async def test_status_returns_200_when_spool_dir_unavailable( + client: httpx.AsyncClient, + tmp_path: Path, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Regression (ci_pr73-ueh): /status is the ACA health probe. spool_stats() + scans the queue dir with iterdir() (raises on a missing dir), unlike the + glob()-based sibling readers. A transiently-unavailable queue dir (e.g. an + Azure Files SMB remount) must NOT turn /status into a 500 -> failed probe + -> container restart loop. It must return 200 with a degraded sentinel.""" + qm = registry.queue_manager + # Point the scan at a directory that does not exist so iterdir() raises, + # exactly as it would during an SMB mount drop. Bypass the TTL cache so the + # scan actually runs on this call. + monkeypatch.setattr(qm, "_dir", tmp_path / "gone") + monkeypatch.setattr(qm, "_spool_cache", None) + + response = await client.get("/status") + + assert response.status_code == 200 + assert response.json()["spool"] == { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + } + + # --------------------------------------------------------------------------- # /queues/* data endpoints stay authenticated (C1; survives the C2 page removal) # --------------------------------------------------------------------------- diff --git a/tests/test_queue_manager.py b/tests/test_queue_manager.py index 54c2deab..bd270601 100644 --- a/tests/test_queue_manager.py +++ b/tests/test_queue_manager.py @@ -465,3 +465,323 @@ async def test_recovery_seed_counts_replay_window_residual_zero(qm): stats = await qm.derive_all_stats() residual = accepted - written - stats["in_queue_total"] - stats["dead_total"] assert residual == 0 + + +# --------------------------------------------------------------------------- +# spool_stats (Change 2): cheap, aggregate-only spool footprint for /status +# --------------------------------------------------------------------------- + + +async def test_spool_stats_counts_pending_session_and_bytes(qm, tmp_path): + """A session with unconsumed log data counts as pending; total bytes + reflects every file on disk (.log + .offset + .dead.jsonl).""" + await qm.append("s1", b"a") + await qm.append("s1", b"b") + + stats = await qm.spool_stats() + + assert stats["pending_sessions"] == 1 + queues_dir = tmp_path / "queues" + expected_bytes = sum(p.stat().st_size for p in queues_dir.iterdir()) + assert stats["spool_bytes_total"] == expected_bytes + assert expected_bytes > 0 + + +async def test_spool_stats_fully_committed_session_not_pending(qm): + """A session whose committed offset reaches EOF is NOT counted as + pending, even though its .log/.offset files still occupy disk space + (spool_bytes_total still reflects them).""" + await qm.append("s1", b"a") + line = b"a\n" + await qm.commit("s1", len(line)) + + stats = await qm.spool_stats() + + assert stats["pending_sessions"] == 0 + assert stats["spool_bytes_total"] > 0 # log + offset files still on disk + + +async def test_spool_stats_dead_letter_only_session_not_pending(qm): + """A dead-letter-only key (no .log) contributes bytes but is never + counted as a pending session -- pending_sessions is defined purely over + .log files with unconsumed data.""" + await qm.dead_letter("s-dead", b"poison", error="boom") + + stats = await qm.spool_stats() + + assert stats["pending_sessions"] == 0 + assert stats["spool_bytes_total"] > 0 + + +async def test_spool_stats_multiple_sessions_aggregate(qm): + """pending_sessions counts sessions independently; bytes sum across all.""" + await qm.append("s1", b"a") # pending + await qm.append("s2", b"b") + line = b"b\n" + await qm.commit("s2", len(line)) # fully committed, not pending + await qm.append("s3", b"c") # pending + + stats = await qm.spool_stats() + + assert stats["pending_sessions"] == 2 + + +async def test_spool_stats_returns_only_aggregate_keys_no_identifiers(qm): + """/status is unauthenticated: spool_stats() must return ONLY the two + aggregate integers -- no session ids, workspace names, or per-key table + of any kind, so there's nothing to accidentally leak through /status.""" + await qm.append("my-secret-session-id", b"a") + await qm.dead_letter("another-session-id", b"poison", error="boom") + + stats = await qm.spool_stats() + + assert set(stats.keys()) == { + "pending_sessions", + "spool_bytes_total", + "corrupt_offsets", + } + serialized = repr(stats) + assert "my-secret-session-id" not in serialized + assert "another-session-id" not in serialized + + +async def test_spool_stats_caches_within_ttl(qm, monkeypatch): + """Repeated calls within the TTL window are served from cache -- the + directory is not re-scanned on every /status poll.""" + import pathlib + + await qm.append("s1", b"a") + + calls = {"n": 0} + real_iterdir = pathlib.Path.iterdir + + # pathlib.Path instances use __slots__, so the target Path (qm._dir) + # cannot be monkeypatched directly -- patch the class method instead, + # counting only calls made against qm._dir (this codebase's only other + # .iterdir() caller checked clean at write time; see grep before this + # test was added). + def counting_iterdir(self: pathlib.Path): + if self == qm._dir: + calls["n"] += 1 + return real_iterdir(self) + + monkeypatch.setattr(pathlib.Path, "iterdir", counting_iterdir) + + await qm.spool_stats() + await qm.spool_stats() # within TTL -> served from cache + assert calls["n"] == 1 + + # Age the cache past the TTL; the next call must recompute. + qm._spool_cache_at = time.monotonic() - (qm._spool_cache_ttl + 1.0) + await qm.spool_stats() + assert calls["n"] == 2 + + +# --------------------------------------------------------------------------- +# Streamed boot/stats scans (ci_pr73-xq2): _complete_data_end and +# _count_newlines must be bounded-memory AND numerically identical to the old +# read_bytes() + slice-count implementation, including at chunk boundaries. +# --------------------------------------------------------------------------- + + +def _naive_complete_data_end(data: bytes) -> int: + last_nl = data.rfind(b"\n") + return last_nl + 1 if last_nl != -1 else 0 + + +def test_complete_data_end_matches_naive_and_handles_edges(qm, tmp_path): + """Backward-scan _complete_data_end == old rfind(b'\\n')+1 for every shape: + empty, no-newline (torn only), trailing newline, torn tail after data.""" + log = tmp_path / "queues" / "s1.log" + log.parent.mkdir(parents=True, exist_ok=True) + for payload in ( + b"", # empty + b"no-newline-yet", # single torn line, no complete data + b"a\n", # one complete line + b"a\nb\n", # two complete lines + b"a\nb\ntorn-tail", # complete data + torn trailing line + ): + log.write_bytes(payload) + assert qm._complete_data_end("s1") == _naive_complete_data_end(payload) + + +def test_complete_data_end_missing_log_is_zero(qm): + assert qm._complete_data_end("nope") == 0 + + +def test_complete_data_end_newline_on_chunk_boundary(qm, tmp_path, monkeypatch): + """The backward scan reads fixed non-overlapping windows; a newline landing + exactly on a chunk boundary must still be found (regression guard for the + streaming rewrite).""" + import context_intelligence_server.queue_manager as qm_mod + + monkeypatch.setattr(qm_mod, "_SCAN_CHUNK_BYTES", 8) + log = tmp_path / "queues" / "s1.log" + log.parent.mkdir(parents=True, exist_ok=True) + # Last newline sits at index 8 (exactly one chunk from the start), followed + # by a torn tail so complete_data_end must be 9, spanning a chunk boundary. + data = b"01234567\ntail" # '\n' at index 8 + log.write_bytes(data) + assert qm._complete_data_end("s1") == _naive_complete_data_end(data) == 9 + + +def test_count_newlines_matches_naive_across_ranges(qm, tmp_path, monkeypatch): + """_count_newlines(start,end) == data[start:end].count(b'\\n') for arbitrary + ranges, including across a small chunk size (multi-chunk streaming).""" + import context_intelligence_server.queue_manager as qm_mod + + monkeypatch.setattr(qm_mod, "_SCAN_CHUNK_BYTES", 4) + log = tmp_path / "queues" / "s1.log" + log.parent.mkdir(parents=True, exist_ok=True) + data = b"aa\nbbbb\nc\n\nddddddd\n" + log.write_bytes(data) + + n = len(data) + for start in range(n + 1): + # to-EOF form + assert qm._count_newlines("s1", start) == data[start:].count(b"\n") + for end in range(start, n + 1): + assert qm._count_newlines("s1", start, end) == data[start:end].count(b"\n") + + +def test_count_newlines_missing_and_empty_range(qm): + assert qm._count_newlines("missing") == 0 + assert qm._count_newlines("missing", 0, 0) == 0 + + +def test_count_dead_matches_naive_and_streams(qm, tmp_path, monkeypatch): + """_count_dead == old data.count(b'\\n') for empty / multi-record / missing, + including a newline on a chunk boundary (streamed, not read_bytes).""" + import context_intelligence_server.queue_manager as qm_mod + + monkeypatch.setattr(qm_mod, "_SCAN_CHUNK_BYTES", 8) + dead = tmp_path / "queues" / "s1.dead.jsonl" + dead.parent.mkdir(parents=True, exist_ok=True) + + assert qm._count_dead("missing") == 0 + + for payload in ( + b"", # empty -> 0 + b'{"a":1}\n', # one record + b'{"a":1}\n{"b":2}\n{"c":3}\n', # three records + b"01234567\n8\n", # newline at index 8 == chunk boundary, 2 records + ): + dead.write_bytes(payload) + assert qm._count_dead("s1") == payload.count(b"\n") + + +async def test_recovery_seed_counts_unchanged_under_streaming(qm): + """End-to-end: the streamed recovery_seed_counts yields the same + (accepted, written) baseline as the semantics it replaced.""" + # Two complete lines appended, one committed. + await qm.append("s1", b"a") + await qm.append("s1", b"bb") + line1 = b"a\n" + await qm.commit("s1", len(line1)) # 1 written, 1 still pending + + accepted, written = await qm.recovery_seed_counts() + + assert written == 1 # one committed line, no dead + assert accepted == 2 # one written + one pending + + +async def test_spool_stats_empty_directory(qm): + """An empty spool directory reports zero for both aggregates.""" + stats = await qm.spool_stats() + assert stats == { + "pending_sessions": 0, + "spool_bytes_total": 0, + "corrupt_offsets": 0, + } + + +# --------------------------------------------------------------------------- +# spool_stats health-endpoint safety (regression, ci_pr73-ueh): +# /status is the unauthenticated ACA health probe and calls spool_stats() +# unconditionally. spool_stats() uses iterdir() (raises on a missing dir), +# unlike every sibling reader which uses glob() (empty on a missing dir), so +# a transiently-unavailable queue dir or a corrupt .offset MUST degrade to a +# sentinel, never raise -- an escape becomes a 500 -> failed probe -> restart. +# --------------------------------------------------------------------------- + + +async def test_spool_stats_missing_directory_returns_sentinel(qm, tmp_path): + """A missing queue dir makes iterdir() raise FileNotFoundError; spool_stats + must return the degraded sentinel {-1, -1} rather than propagate (which + would 500 the /status health probe -- e.g. during an Azure Files remount).""" + import shutil + + shutil.rmtree(tmp_path / "queues") + qm._spool_cache = None # bypass the TTL cache so the scan actually runs + + stats = await qm.spool_stats() + + assert stats == { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + } + + +async def test_spool_stats_sentinel_is_not_cached(qm, tmp_path): + """The degraded sentinel is NOT cached: once the directory is healthy + again, the very next call recovers the real aggregate numbers.""" + import shutil + + queues_dir = tmp_path / "queues" + shutil.rmtree(queues_dir) + qm._spool_cache = None + assert await qm.spool_stats() == { + "pending_sessions": -1, + "spool_bytes_total": -1, + "corrupt_offsets": -1, + } + + # Filesystem recovers; no manual cache reset -- the sentinel was never stored. + queues_dir.mkdir(parents=True, exist_ok=True) + await qm.append("s1", b"a") + + stats = await qm.spool_stats() + assert stats["pending_sessions"] == 1 + assert stats["spool_bytes_total"] > 0 + + +async def test_spool_stats_corrupt_offset_does_not_sink_scan(qm): + """A corrupt/unreadable .offset for one session must not fail the whole + scan (which would 500 /status): that file's bytes still count, only its + pending calc is skipped.""" + await qm.append("s1", b"a") + qm._offset_path("s1").write_text("not-a-number", encoding="utf-8") + qm._spool_cache = None + + stats = await qm.spool_stats() + + assert stats["spool_bytes_total"] > 0 + assert isinstance(stats["pending_sessions"], int) + + +async def test_spool_stats_counts_corrupt_offsets(qm): + """A non-numeric .offset is surfaced as an aggregate corrupt_offsets count + (the ONLY visibility signal -- no logging). A healthy session contributes 0.""" + await qm.append("s-good", b"a") # valid: no .offset yet -> committed 0 + await qm.append("s-bad", b"a") + qm._offset_path("s-bad").write_text("not-a-number", encoding="utf-8") + qm._spool_cache = None + + stats = await qm.spool_stats() + + assert stats["corrupt_offsets"] == 1 + assert stats["spool_bytes_total"] > 0 # corrupt file's bytes still counted + + +async def test_spool_stats_healthy_offsets_report_zero_corrupt(qm): + """corrupt_offsets is 0 when every .offset is a valid integer (it must not + fire on the normal committed-offset path).""" + await qm.append("s1", b"a") + line = b"a\n" + await qm.commit("s1", len(line)) # writes a valid numeric .offset + qm._spool_cache = None + + stats = await qm.spool_stats() + + assert stats["corrupt_offsets"] == 0 diff --git a/tests/test_run_entrypoint.py b/tests/test_run_entrypoint.py index ff902dd5..07042b0b 100644 --- a/tests/test_run_entrypoint.py +++ b/tests/test_run_entrypoint.py @@ -2,11 +2,12 @@ from unittest.mock import patch -from gunicorn.app.base import BaseApplication -from uvicorn.workers import UvicornWorker - +import context_intelligence_server.main as main_module +import pytest from context_intelligence_server.config import get_settings from context_intelligence_server.main import run +from gunicorn.app.base import BaseApplication +from uvicorn.workers import UvicornWorker def test_run_uses_gunicorn_with_settings() -> None: @@ -27,3 +28,48 @@ def _capture(self: BaseApplication) -> None: assert cfg.graceful_timeout == 10 assert cfg.worker_class is UvicornWorker assert cfg.timeout == 30 + + +# --------------------------------------------------------------------------- +# Change 3: configurable gunicorn worker timeout / graceful_timeout +# --------------------------------------------------------------------------- + + +def test_run_gunicorn_timeouts_default_to_previous_hardcoded_values() -> None: + """No config set -> gunicorn sees the SAME 30s/10s that used to be + hardcoded (no-op default, verified end-to-end through run()).""" + instances: list[BaseApplication] = [] + + def _capture(self: BaseApplication) -> None: + instances.append(self) + + with patch.object(BaseApplication, "run", _capture): + run() + + cfg = instances[0].cfg + assert cfg.timeout == 30 + assert cfg.graceful_timeout == 10 + + +def test_run_gunicorn_timeouts_respect_settings_override( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A configured gunicorn_worker_timeout / gunicorn_graceful_timeout flows + through run() into the actual gunicorn config -- an operator can + accommodate a legitimately slow, large-backlog boot (see + crash_recovery_respawn_limit) without gunicorn's own watchdog SIGKILLing + the worker mid-startup.""" + monkeypatch.setattr(main_module._settings, "gunicorn_worker_timeout", 300) + monkeypatch.setattr(main_module._settings, "gunicorn_graceful_timeout", 45) + + instances: list[BaseApplication] = [] + + def _capture(self: BaseApplication) -> None: + instances.append(self) + + with patch.object(BaseApplication, "run", _capture): + run() + + cfg = instances[0].cfg + assert cfg.timeout == 300 + assert cfg.graceful_timeout == 45