From 0fc0b69031904186522aa3321feb400d1d06ae8e Mon Sep 17 00:00:00 2001 From: colombod Date: Wed, 9 Sep 2026 19:27:10 +0000 Subject: [PATCH 1/2] fix: crash recovery must not block the HTTP surface at startup Move all four crash recovery phases into an asyncio.create_task() so the HTTP surface (serving /version and /status) opens immediately, regardless of spool size. Production incident 2026-09-09: recovery phases walked ~1700 session keys across 5000+ files on Azure Files SMB, blocking uvicorn lifespan completion and the HTTP socket binding. TCP probes saw the port open but no requests were answered for 231s, causing APIM gateway timeouts and 1493 event drops. Changes: - All four recovery phases move into new _startup_recovery(app) via create_task. The sweep loop is awaited inside the same task; the task is cancelled before shutdown_workers() so recovery cannot spawn drainers into a quiescing registry. - _startup_recovery wraps _startup_recovery_body: re-raises CancelledError, records recovery_error on app.state, logs via logger.exception, and sets recovery_complete in finally (fail-loud, not silent garbage collection). - /status gains recovery_complete and recovery_error for observability. - Untagged-:Node boot guard removed (inert dead data; all node-creating statements use MERGE with :Node tag). Detection moved to operator diagnostic path. Verification performed: - pytest -m "not neo4j": 1984 passed, 7 skipped, 92 deselected - tests/test_main.py: 89 passed - 4 new guards + 5 reworked recovery tests verified RED against main (10 failures on old source, 89 passed on new) - ruff format --check: clean on all touched files Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- context_intelligence_server/main.py | 293 +++++++++++------- tests/neo4j/test_node_identity_migration.py | 49 +-- tests/test_main.py | 325 +++++++++++++++----- 3 files changed, 455 insertions(+), 212 deletions(-) diff --git a/context_intelligence_server/main.py b/context_intelligence_server/main.py index c2faf780..e37d3dca 100644 --- a/context_intelligence_server/main.py +++ b/context_intelligence_server/main.py @@ -42,7 +42,6 @@ ) from context_intelligence_server.neo4j_store import ( build_bounded_neo4j_driver, - count_untagged_nodes, ensure_neo4j_schema, mark_schema_ready, ) @@ -245,105 +244,41 @@ async def _crash_recovery_sweep_loop(interval: int, respawn_limit: int) -> None: 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.""" - setup_logging() - _admin = _settings.resolve_neo4j_admin() - _query = _settings.resolve_neo4j_query() - logger.info( - "lifespan_startup: creating Neo4j drivers admin_url=%s query_url=%s query_access_mode=%s", - _admin.url, - _query.url, - _query.access_mode, - ) - # Admin (read/write): schema init + all mutation paths. Keep the existing - # app.state.neo4j_driver NAME so nothing that reads it silently breaks. - # build_neo4j_driver() is the SAME helper doctor.run_doctor() uses, so the - # server and the doctor CLI can never construct this connection differently. - app.state.neo4j_driver = build_neo4j_driver(_admin) - # Cypher-query (read-intent): /cypher + dashboard reads. Bounded through the - # same helper as the admin driver so every process-wide pool shares one cap. - app.state.neo4j_query_driver = build_bounded_neo4j_driver( - _query, - max_connection_pool_size=_settings.neo4j_max_connection_pool_size, - ) - # Stash the resolved query access_mode so /cypher opens READ sessions without - # re-resolving settings on every request. - app.state.neo4j_query_access_mode = _query.access_mode - # Initialize schema (indexes + uniqueness constraints) BEFORE the server starts - # accepting requests. This ensures the Session uniqueness constraint is active - # before any concurrent flush() transactions execute MERGE, which prevents the - # duplicate-Session-node race condition observed under concurrent upload load. - logger.info( - "lifespan_startup: initializing Neo4j schema (indexes + uniqueness constraints)" - ) - # Cold start FAILS LOUD on schema/data corruption that requires - # `doctor --fix` -- an un-migrated graph (duplicate legacy nodes OR - # nodes lacking the universal :Node label). Nothing has been written yet - # at cold start, so refusing to boot loses no data: this is the safest - # possible moment to surface an impossible state as an un-missable - # signal rather than a log line someone greps for later. Contrast with - # the flush path (Neo4jGraphStore._ensure_schema), which must keep - # self-healing and never raise (Salil's blocker -- raising there would - # dead-letter real in-flight activity records). fail_on_data_conflict=True - # here mirrors run_repair's contract: a :Node constraint data conflict - # raises a RuntimeError naming `doctor --fix` instead of being logged - # and swallowed. - schema_fully_established = await ensure_neo4j_schema( - app.state.neo4j_driver, fail_on_data_conflict=True - ) - # Seed the PROCESS-wide schema latch, but ONLY on a fully-established pass. - # - # fail_on_data_conflict=True makes this call fail closed on a :Node - # constraint DATA conflict -- but a CONNECTIVITY failure on any individual - # index/constraint is deliberately swallowed and reported through the - # return value instead (see ensure_neo4j_schema's docstring). Latching - # unconditionally would therefore mark a HALF-BUILT schema as ready and - # permanently disable the per-flush self-heal for the whole process -- - # exactly the "constraint created once, never retried" gap - # Neo4jGraphStore._ensure_schema exists to close. - # - # On the happy path this seed is what stops every per-session store from - # re-running the same ~11-statement catalog pass on its first flush -- and, - # whenever that pass cannot complete, on EVERY subsequent flush -- competing - # for the very bolt pool it needs. See neo4j_store._SCHEMA_READY. - if schema_fully_established: - mark_schema_ready() - logger.info("lifespan_startup: Neo4j schema initialized") - else: - logger.warning( - "lifespan_startup: Neo4j schema NOT fully established (indexes or " - "constraints missing); leaving the process-wide latch unset so the " - "flush path retries schema init (rate-limited by " - "neo4j_store._SCHEMA_RETRY_BACKOFF_SECONDS)." - ) - # Fail-loud migration-health guard: duplicate nodes are already caught - # above by the :Node constraint (fail_on_data_conflict=True); this catches - # the OTHER un-migrated shape the constraint can't see on its own -- - # nodes that simply lack the :Node label altogether, which violate no - # constraint and so raise nothing by themselves. O(1) via the counts - # store (see count_untagged_nodes) -- this must never regress into the - # AllNodesScan stall PR #67 removed from the write path. - # - # A connectivity/probe failure here is NOT the same as "confirmed - # un-migrated" -- it means graph state could not be determined, not that - # it was determined to be bad -- so it is logged at DEBUG and swallowed - # rather than treated as a corruption finding; the flush path's - # self-heal still covers a genuinely dirty graph once it becomes - # reachable. +async def _startup_recovery(app: FastAPI) -> None: + """Crash-recovery pass + deferred-backlog sweep, off the startup path. + + Runs as a background task created by ``lifespan`` so that startup can + complete -- and the HTTP surface open -- in ~1s regardless of how large + the durable spool is. Sets ``app.state.recovery_complete`` when the + one-shot recovery pass is done, so ``/status`` can say so honestly + rather than reporting un-seeded counters as though they were settled. + + Thin wrapper around ``_startup_recovery_body``: as a BACKGROUND task, an + unhandled exception here would otherwise be swallowed until garbage + collection ("Task exception was never retrieved") -- silently leaving the + counters unseeded and no drainers respawned, with a server that looks + perfectly healthy. On the old inline path the same failure aborted boot + loudly. Log it, record it for ``/status``, and always set the event so + ``/status`` cannot report "recovery in progress" forever. + """ try: - untagged = await count_untagged_nodes(app.state.neo4j_driver) - except Exception as exc: # noqa: BLE001 - connectivity probe, not a confirmed bad state - _LOG_MSG = "startup migration-health probe skipped (graph unreachable?): %s" - logger.debug(_LOG_MSG, exc) - untagged = 0 - if untagged: - raise RuntimeError( - f"Neo4j graph has {untagged} node(s) lacking the :Node label " - "(un-migrated). Cold start refuses to boot to avoid duplicating " - "them on write. Run: context-intelligence-server doctor --fix" + await _startup_recovery_body(app) + except asyncio.CancelledError: + raise # shutdown -- not a failure, and must propagate + except Exception as exc: # noqa: BLE001 - background task boundary + app.state.recovery_error = repr(exc) + logger.exception( + "startup_recovery: FAILED -- conservation counters may be unseeded " + "and recovered drainers may not have respawned. The server is still " + "serving; a later boot's recover() reports the same sessions again, " + "and a new event for a session spawns its drainer via get_or_create()." ) + finally: + app.state.recovery_complete.set() + + +async def _startup_recovery_body(app: FastAPI) -> None: + """The actual recovery pass. See ``_startup_recovery`` for why it is split.""" # Crash recovery: on startup, respawn one drainer per # session that still has an undrained, complete line. The workspace is # parsed from that session's FIRST log line so the respawned worker is @@ -405,7 +340,7 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: # silent, un-discoverable fact. Names the exact counts and the # setting to raise. logger.warning( - "lifespan_startup: crash-recovery respawn cap reached " + "startup_recovery: 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 " @@ -416,36 +351,163 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: deferred_count, ) logger.info( - "lifespan_startup: crash recovery respawned %d/%d drainers", + "startup_recovery: crash recovery respawned %d/%d drainers", respawned, len(recovered), ) + # The one-shot pass is done: counters are seeded and this boot's drainers + # are up. /status can now report settled numbers rather than mid-recovery + # ones. Set BEFORE the sweep loop below, which never returns. (The wrapper + # also sets it in a finally, so a failure cannot pin /status at + # "in progress" forever -- but on the success path it must be set HERE, or + # the sweep loop would delay it indefinitely.) + app.state.recovery_complete.set() # 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 + # (respawn_limit is None) there is no deferred tail, so the loop is not + # entered -- 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 + # + # AWAITED rather than wrapped in its own create_task: this coroutine is + # already a background task that lifespan cancels on shutdown, so awaiting + # the sweep here means one task to cancel instead of two -- and no way for + # the sweep to outlive its parent. _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, ) + await _crash_recovery_sweep_loop(_sweep_interval, respawn_limit) + + +@asynccontextmanager +async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: + """Manage application lifespan: configure logging and create shared Neo4j driver.""" + setup_logging() + _admin = _settings.resolve_neo4j_admin() + _query = _settings.resolve_neo4j_query() + logger.info( + "lifespan_startup: creating Neo4j drivers admin_url=%s query_url=%s query_access_mode=%s", + _admin.url, + _query.url, + _query.access_mode, + ) + # Admin (read/write): schema init + all mutation paths. Keep the existing + # app.state.neo4j_driver NAME so nothing that reads it silently breaks. + # build_neo4j_driver() is the SAME helper doctor.run_doctor() uses, so the + # server and the doctor CLI can never construct this connection differently. + app.state.neo4j_driver = build_neo4j_driver(_admin) + # Cypher-query (read-intent): /cypher + dashboard reads. Bounded through the + # same helper as the admin driver so every process-wide pool shares one cap. + app.state.neo4j_query_driver = build_bounded_neo4j_driver( + _query, + max_connection_pool_size=_settings.neo4j_max_connection_pool_size, + ) + # Stash the resolved query access_mode so /cypher opens READ sessions without + # re-resolving settings on every request. + app.state.neo4j_query_access_mode = _query.access_mode + # Initialize schema (indexes + uniqueness constraints) BEFORE the server starts + # accepting requests. This ensures the Session uniqueness constraint is active + # before any concurrent flush() transactions execute MERGE, which prevents the + # duplicate-Session-node race condition observed under concurrent upload load. + logger.info( + "lifespan_startup: initializing Neo4j schema (indexes + uniqueness constraints)" + ) + # Cold start FAILS LOUD on schema/data corruption that requires + # `doctor --fix` -- an un-migrated graph (duplicate legacy nodes OR + # nodes lacking the universal :Node label). Nothing has been written yet + # at cold start, so refusing to boot loses no data: this is the safest + # possible moment to surface an impossible state as an un-missable + # signal rather than a log line someone greps for later. Contrast with + # the flush path (Neo4jGraphStore._ensure_schema), which must keep + # self-healing and never raise (Salil's blocker -- raising there would + # dead-letter real in-flight activity records). fail_on_data_conflict=True + # here mirrors run_repair's contract: a :Node constraint data conflict + # raises a RuntimeError naming `doctor --fix` instead of being logged + # and swallowed. + schema_fully_established = await ensure_neo4j_schema( + app.state.neo4j_driver, fail_on_data_conflict=True + ) + # Seed the PROCESS-wide schema latch, but ONLY on a fully-established pass. + # + # fail_on_data_conflict=True makes this call fail closed on a :Node + # constraint DATA conflict -- but a CONNECTIVITY failure on any individual + # index/constraint is deliberately swallowed and reported through the + # return value instead (see ensure_neo4j_schema's docstring). Latching + # unconditionally would therefore mark a HALF-BUILT schema as ready and + # permanently disable the per-flush self-heal for the whole process -- + # exactly the "constraint created once, never retried" gap + # Neo4jGraphStore._ensure_schema exists to close. + # + # On the happy path this seed is what stops every per-session store from + # re-running the same ~11-statement catalog pass on its first flush -- and, + # whenever that pass cannot complete, on EVERY subsequent flush -- competing + # for the very bolt pool it needs. See neo4j_store._SCHEMA_READY. + if schema_fully_established: + mark_schema_ready() + logger.info("lifespan_startup: Neo4j schema initialized") + else: + logger.warning( + "lifespan_startup: Neo4j schema NOT fully established (indexes or " + "constraints missing); leaving the process-wide latch unset so the " + "flush path retries schema init (rate-limited by " + "neo4j_store._SCHEMA_RETRY_BACKOFF_SECONDS)." + ) + # NOTE: the O(1) untagged-:Node boot guard that used to sit here has been + # REMOVED. It refused to boot when any node lacked the universal :Node + # label. Two things made it dead weight: + # + # 1. Nothing in this service can produce an untagged node. Every + # node-creating statement in neo4j_store is label-scoped -- the node + # MERGE, both edge-endpoint MERGEs, and the session write path all + # MERGE (n:Node {node_id, workspace}). + # 2. Since get_node()/get_edge() became :Node-scoped, an untagged node is + # also INVISIBLE to every read. Read and write paths now agree, so a + # stray untagged node is inert dead data, not a correctness hazard -- + # and refusing to serve over inert data is disproportionate. + # + # The capability is not gone: `context-intelligence-server doctor` still + # reports untagged nodes via diagnose()/count_untagged_nodes, and + # `doctor --fix` still repairs them. Detection moved to the operator path, + # where it belongs; it is no longer a boot gate. (Historically this check + # was itself an AllNodesScan and caused a 25-30s boot stall until PR #67 + # made it O(1) -- a second reason not to keep it on the startup path.) + + # Crash recovery runs as a BACKGROUND task -- it MUST NOT gate serving. + # + # uvicorn runs the ASGI lifespan to completion BEFORE it handles a single + # request (uvicorn/server.py: `await self.startup()` then + # `await self.main_loop()`), and gunicorn binds the socket before that. So + # any work done here is work during which the process ACCEPTS connections + # and answers none -- /version and /status included, however cheap they + # are. Incident 2026-09-09: a ~5000-file spool on Azure Files kept this + # loop busy for 9+ minutes; ACA's tcpSocket probes saw an open port and + # reported the replica Healthy while every request timed out at the APIM + # gateway (231s), and clients dropped events from full buffers. + # + # Moving it behind create_task lets startup complete in ~1s, so the HTTP + # surface is up regardless of spool size. This is safe because respawn is + # idempotent by construction -- see _crash_recovery_topup, which is already + # documented as safe to call repeatedly on a LIVE server, and which the + # sweep loop has always called against a serving process. + app.state.recovery_complete = asyncio.Event() + app.state.recovery_error = None + _recovery_task: asyncio.Task[None] = asyncio.create_task(_startup_recovery(app)) + try: yield finally: - if _sweep_task is not None: - _sweep_task.cancel() - with suppress(asyncio.CancelledError): - await _sweep_task + # Cancel background recovery FIRST: it spawns drainers, and shutdown + # below quiesces them. Letting it keep spawning into a shutting-down + # registry would race the quiesce and dead-letter healthy events. + _recovery_task.cancel() + with suppress(asyncio.CancelledError): + await _recovery_task # ORDER IS LOAD-BEARING. Quiesce the drainers FIRST. Every session's # graph store now shares ONE driver, so closing it under a live drainer # is no longer a per-session concern: the drainer's batch fails, it @@ -929,6 +991,15 @@ async def get_status(request: Request) -> dict[str, Any]: # above. Cheap by construction (stat-only, short-TTL cached); see # QueueManager.spool_stats() for why this holds even under a huge spool. response["spool"] = await registry.queue_manager.spool_stats() + # Crash recovery now runs in the BACKGROUND so the HTTP surface opens + # immediately (see _startup_recovery). While it is still running, the + # conservation counters above have not been seeded yet, so say so rather + # than letting a caller read mid-recovery numbers as settled ones. + _recovery_evt = getattr(request.app.state, "recovery_complete", None) + response["recovery_complete"] = bool(_recovery_evt and _recovery_evt.is_set()) + # "complete" alone would be a lie by omission when recovery FAILED: the + # event is set either way (see _startup_recovery's finally). + response["recovery_error"] = getattr(request.app.state, "recovery_error", None) # 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 diff --git a/tests/neo4j/test_node_identity_migration.py b/tests/neo4j/test_node_identity_migration.py index 2358d0d2..88501adf 100644 --- a/tests/neo4j/test_node_identity_migration.py +++ b/tests/neo4j/test_node_identity_migration.py @@ -289,8 +289,19 @@ async def _run_migration_assertions(neo4j_container: dict[str, Any]) -> None: async def test_cold_start_guard_detects_untagged_only_graph( neo4j_container: dict[str, Any], ) -> None: - """Reproduces main.py's lifespan cold-start guard, against a REAL Neo4j, - for the untagged-only shape the :Node constraint alone CANNOT see. + """Exercises ``count_untagged_nodes`` and ``ensure_neo4j_schema`` against + a REAL Neo4j, for the untagged-only shape the :Node constraint alone + CANNOT see. + + NOTE: main.py's ``lifespan()`` no longer calls ``count_untagged_nodes`` + at all -- the O(1) untagged-:Node boot guard this test used to describe + was REMOVED (nothing in this service can produce an untagged node, and + since get_node()/get_edge() became :Node-scoped, an untagged node is + inert dead data rather than a correctness hazard; detection moved + entirely to the `doctor` operator path). This test does not call + ``lifespan()`` and its assertions are unaffected by that removal -- it + verifies the two underlying primitives directly, still used by + ``doctor``/``run_repair``. Uses ``_seed_untagged_only_graph`` (NOT ``_seed_dirty_graph``, whose duplicate ``dup-1`` :Event nodes trip the separate, always fail-open @@ -298,22 +309,19 @@ async def test_cold_start_guard_detects_untagged_only_graph( ``test_run_repair_dedups_backfills_and_constrains``). Every node seeded here has a unique (node_id, workspace) and no label collision, so NO uniqueness constraint (Session/Event/Node) sees a conflict -- the ONLY - defect is the missing ``:Node`` label, which is exactly why the lifespan - guard needs its second, independent check (``count_untagged_nodes``): - the constraint step provides no signal for this case on its own. + defect is the missing ``:Node`` label, which is exactly why + ``count_untagged_nodes`` exists as a second, independent check: the + constraint step provides no signal for this case on its own. - Reproduces the lifespan's two ordered steps directly against the live - container (the guard logic is inline in ``main.py``'s ``lifespan()``, - not its own importable function): + Exercises the two primitives directly against the live container: 1. ``ensure_neo4j_schema(driver, fail_on_data_conflict=True)`` -- succeeds (``True``), no constraint conflict to raise on. 2. ``count_untagged_nodes(driver)`` -- reports > 0. - Together, (1) succeeding and (2) being > 0 is precisely the condition - under which ``lifespan()`` raises ``RuntimeError`` naming - ``doctor --fix`` -- i.e. the un-migrated (untagged-only) graph IS - detected and cold start WOULD refuse to boot. + Together, (1) succeeding and (2) being > 0 is the un-migrated + (untagged-only) graph signature ``doctor``/``run_repair`` uses to detect + and repair this shape -- no longer a boot-time condition. """ _wipe(neo4j_container) try: @@ -324,10 +332,10 @@ async def test_cold_start_guard_detects_untagged_only_graph( auth=(neo4j_container["user"], neo4j_container["password"]), ) try: - # Step 1 (lifespan): fail-loud schema init does NOT raise here -- - # none of the seeded nodes carry :Node (or collide under any - # OTHER constraint), so no constraint sees a conflict. This is - # the case the constraint check alone misses. + # Step 1: fail-loud schema init does NOT raise here -- none of + # the seeded nodes carry :Node (or collide under any OTHER + # constraint), so no constraint sees a conflict. This is the + # case the constraint check alone misses. established = await ensure_neo4j_schema(driver, fail_on_data_conflict=True) assert established is True, ( "ensure_neo4j_schema(fail_on_data_conflict=True) must succeed " @@ -335,13 +343,14 @@ async def test_cold_start_guard_detects_untagged_only_graph( "dirty graph -- there is no constraint conflict to raise on." ) - # Step 2 (lifespan): the O(1) untagged guard DOES catch it. + # Step 2: the O(1) untagged counter DOES catch it (no longer + # called by lifespan() -- see the doctor operator path). untagged = await count_untagged_nodes(driver) assert untagged > 0, ( "count_untagged_nodes must report the seeded untagged legacy " - "nodes (legacy-sess-only + legacy-bare) -- this is the exact " - "signal main.py's lifespan() uses to raise RuntimeError and " - "refuse to boot on an un-migrated graph." + "nodes (legacy-sess-only + legacy-bare) -- this is the " + "signal `doctor`/`run_repair` use to detect and repair an " + "un-migrated graph." ) finally: await driver.close() diff --git a/tests/test_main.py b/tests/test_main.py index a5ab80a5..bcc17cc3 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -797,6 +797,196 @@ def _patched_lifespan_deps() -> Any: return mock_driver +# --------------------------------------------------------------------------- +# Background recovery task: startup must not block HTTP (fix/startup-must- +# not-block-http) +# --------------------------------------------------------------------------- + + +async def test_status_and_version_respond_while_recovery_is_still_running( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """THE load-bearing regression test: HTTP must be answerable while the + background recovery task is still running. + + uvicorn runs the ASGI lifespan to completion BEFORE it starts + main_loop (i.e. before it accepts/answers a single request), so any + startup work that runs INLINE in lifespan means the process accepts TCP + connections and answers nothing until that work finishes -- this is + exactly the 2026-09-09 incident (a ~5000-file spool kept lifespan busy + for 9+ minutes while ACA's tcpSocket probe reported the replica healthy). + Moving recovery behind ``asyncio.create_task`` lets lifespan yield + almost immediately, so /version and /status must respond -- and report + recovery as still in progress -- while recovery is deliberately held + open by a test-controlled gate. + + Against the OLD inline-recovery code this test CANNOT pass. Verified + directly (git stash on context_intelligence_server/main.py only): the + pre-fix module has no ``_startup_recovery_body`` at all -- recovery ran + fully inline inside ``lifespan`` -- so the monkeypatch.setattr below + fails fast with AttributeError. Had that seam existed but still run + synchronously before the ``yield``, this test would instead hang + forever (lifespan never yields while the gate is unset) until the + suite's timeout fired. Either way, the old code cannot make this test + pass -- that failure IS the regression this test guards against. + """ + recovery_gate = asyncio.Event() + + async def _blocked_recovery(app: Any) -> None: + await recovery_gate.wait() + + monkeypatch.setattr( + main_module, + "_startup_recovery_body", + AsyncMock(side_effect=_blocked_recovery), + ) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + ): + async with lifespan(main_module.app): + # Recovery is genuinely still running: lifespan already yielded + # without waiting for it. + assert main_module.app.state.recovery_complete.is_set() is False + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=main_module.app), + base_url="http://test", + ) as c: + version_response = await c.get("/version") + assert version_response.status_code == 200 + assert "version" in version_response.json() + + status_response = await c.get("/status") + assert status_response.status_code == 200 + assert status_response.json()["recovery_complete"] is False + + # Release the gate and wait for the background pass to finish. + recovery_gate.set() + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) + + settled_response = await c.get("/status") + assert settled_response.status_code == 200 + settled_data = settled_response.json() + assert settled_data["recovery_complete"] is True + assert settled_data["recovery_error"] is None + + +async def test_recovery_failure_is_logged_and_does_not_take_down_the_server( + monkeypatch: pytest.MonkeyPatch, + caplog: pytest.LogCaptureFixture, +) -> None: + """A failure inside the background recovery pass must be logged and + recorded on app.state -- never swallowed. + + As a bare background task (no ``_startup_recovery`` wrapper) an + unhandled exception here would vanish until garbage collection ("Task + exception was never retrieved"), leaving the conservation counters + unseeded and no drainers respawned behind a server that otherwise looks + perfectly healthy. On the OLD inline path the same failure aborted boot + loudly instead. The wrapper must catch it, log it, and stash it on + ``app.state.recovery_error`` so /status can surface it too. + """ + boom = RuntimeError("boom-test-recovery-failure") + monkeypatch.setattr( + main_module, "_startup_recovery_body", AsyncMock(side_effect=boom) + ) + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + caplog.at_level(logging.ERROR, logger="context_intelligence_server"), + ): + async with lifespan(main_module.app): + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) + assert "boom-test-recovery-failure" in main_module.app.state.recovery_error + + async with httpx.AsyncClient( + transport=httpx.ASGITransport(app=main_module.app), + base_url="http://test", + ) as c: + response = await c.get("/status") + assert response.status_code == 200 + assert "boom-test-recovery-failure" in response.json()["recovery_error"] + + error_records = [r for r in caplog.records if r.levelno == logging.ERROR] + assert any("startup_recovery" in r.getMessage() for r in error_records), ( + f"expected an ERROR log mentioning startup_recovery; " + f"got {[r.getMessage() for r in error_records]}" + ) + + +async def test_recovery_task_is_cancelled_on_shutdown() -> None: + """Shutdown must cancel the background recovery task -- not leave it running. + + Recovery spawns drainers and shutdown quiesces them (see + ``test_lifespan_quiesces_drain_workers_before_closing_shared_driver``); a + recovery task that survives lifespan exit would keep racing that + quiesce, spawning new drainers into a registry that is being torn down + and risking dead-lettered healthy events. + """ + never_set = asyncio.Event() + + async def _blocked_forever(app: Any) -> None: + await never_set.wait() + + mock_driver = _patched_lifespan_deps() + with ( + patch("context_intelligence_server.main.setup_logging"), + patch( + "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", + return_value=mock_driver, + ), + patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), + patch( + "context_intelligence_server.main._startup_recovery_body", + new=AsyncMock(side_effect=_blocked_forever), + ), + ): + async with lifespan(main_module.app): + # Let the recovery task actually get scheduled and start blocking. + await asyncio.sleep(0) + assert main_module.app.state.recovery_complete.is_set() is False + + leftover = [ + t + for t in asyncio.all_tasks() + if not t.done() and "_startup_recovery" in repr(t) + ] + assert leftover == [], f"recovery task(s) still pending after shutdown: {leftover}" + + +async def test_status_reports_recovery_state_before_lifespan_has_run( + client: httpx.AsyncClient, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """/status must degrade gracefully -- not 500 -- when app.state has never + had recovery_complete/recovery_error set (lifespan hasn't run yet).""" + monkeypatch.delattr(main_module.app.state, "recovery_complete", raising=False) + monkeypatch.delattr(main_module.app.state, "recovery_error", raising=False) + + response = await client.get("/status") + assert response.status_code == 200 + data = response.json() + assert data["recovery_complete"] is False + assert data["recovery_error"] is None + + async def test_lifespan_recovers_and_respawns_drainers( monkeypatch: pytest.MonkeyPatch, ) -> None: @@ -833,7 +1023,11 @@ async def test_lifespan_recovers_and_respawns_drainers( ), ): async with lifespan(main_module.app): - pass + # Recovery now runs in a background task; wait for the one-shot + # pass to finish before asserting on its effects. + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert (sid, "/recovered-ws") in spawned @@ -925,7 +1119,9 @@ async def test_lifespan_default_respawns_all_recovered_sessions_unbounded( patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert {s for s, _w in spawned} == set(sids) @@ -959,7 +1155,9 @@ async def test_lifespan_respawn_cap_defers_remainder_and_logs_warning( caplog.at_level(logging.WARNING, logger="context_intelligence_server"), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) # Exactly the cap's worth of sessions were respawned -- never more. assert len(spawned) == 2 @@ -1004,7 +1202,9 @@ async def test_lifespan_deferred_sessions_untouched_and_recoverable_next_boot( patch("context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock()), ): async with lifespan(main_module.app): - pass + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert len(spawned_boot1) == 1 deferred_sids = set(sids) - {s for s, _w in spawned_boot1} @@ -1116,7 +1316,14 @@ async def test_lifespan_enables_sweep_under_finite_limit( 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 + # The sweep is awaited inside the same background task, entered + # right after recovery_complete is set -- waiting for the event + # is enough for the "enabled" log line to have already fired by + # the time we resume (it's emitted synchronously before the + # sweep loop's first suspending await). + await asyncio.wait_for( + main_module.app.state.recovery_complete.wait(), timeout=5 + ) assert any( "crash_recovery_sweep: enabled" in r.getMessage() for r in caplog.records @@ -1183,15 +1390,16 @@ async def test_lifespan_no_sweep_when_interval_zero( # --------------------------------------------------------------------------- # Cold start FAILS LOUD on schema/data corruption that requires `doctor # --fix` (design decision, reversing the lifespan half of f4d8bab): an -# un-migrated graph -- duplicate legacy nodes (caught by the :Node -# constraint via fail_on_data_conflict=True) OR nodes lacking the :Node -# label altogether (caught by the O(1) count_untagged_nodes guard) -- must -# refuse to boot. Nothing has been written yet at cold start, so refusing to -# boot loses no data. The migration itself still lives ONLY in `doctor -# --fix` (run_repair); the flush path still self-heals (see -# tests/neo4j/test_node_identity_migration.py). A connectivity/probe -# failure (graph unreachable) is NOT treated as "confirmed un-migrated" and -# must not crash boot. +# un-migrated graph with duplicate legacy nodes (caught by the :Node +# constraint via fail_on_data_conflict=True) must refuse to boot. Nothing +# has been written yet at cold start, so refusing to boot loses no data. +# The migration itself still lives ONLY in `doctor --fix` (run_repair); the +# flush path still self-heals (see tests/neo4j/test_node_identity_migration.py). +# +# The O(1) untagged-:Node boot guard that used to sit alongside this check +# has been REMOVED (see test_lifespan_does_not_gate_boot_on_untagged_nodes +# below for why) -- detection of untagged nodes moved entirely to the +# `doctor` operator path. # --------------------------------------------------------------------------- @@ -1212,10 +1420,6 @@ async def test_lifespan_calls_ensure_schema_with_fail_on_data_conflict() -> None "context_intelligence_server.main.ensure_neo4j_schema", new=mock_ensure_schema, ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=0), - ), ): async with lifespan(main_module.app): pass @@ -1254,68 +1458,32 @@ async def test_lifespan_raises_on_ensure_schema_data_conflict() -> None: pass -async def test_lifespan_raises_on_untagged_nodes() -> None: - """On an un-migrated graph (untagged :Node count > 0), startup MUST - raise a RuntimeError naming `doctor --fix` -- boot refuses to start - rather than silently risking write-path duplication.""" - mock_driver = _patched_lifespan_deps() - with ( - patch("context_intelligence_server.main.setup_logging"), - patch( - "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", - return_value=mock_driver, - ), - patch( - "context_intelligence_server.main.ensure_neo4j_schema", - new=AsyncMock(return_value=True), - ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=42), - ), - pytest.raises(RuntimeError, match="doctor --fix") as exc_info, - ): - async with lifespan(main_module.app): - pass +async def test_lifespan_does_not_gate_boot_on_untagged_nodes() -> None: + """The O(1) untagged-:Node boot guard has been REMOVED entirely -- + startup no longer counts or checks untagged nodes at all, and no longer + raises when they exist. - assert "42" in str(exc_info.value), ( - f"Expected the untagged count in the error message, got: {exc_info.value}" - ) + Two things made the guard dead weight: (1) nothing in this service can + produce an untagged node -- every node-creating statement in + neo4j_store is a label-scoped MERGE (n:Node {node_id, workspace}); and + (2) since get_node()/get_edge() became :Node-scoped, an untagged node is + also invisible to every read, so a stray one is inert dead data, not a + correctness hazard -- refusing to serve over it would be disproportionate. + The capability is not gone: `context-intelligence-server doctor` still + reports and repairs untagged nodes via diagnose()/count_untagged_nodes; + detection simply moved to the operator path and is no longer a boot gate. -async def test_lifespan_does_not_raise_on_clean_graph() -> None: - """On a fully-migrated graph (untagged count == 0, no constraint - conflict), startup does NOT raise -- the fail-loud guards are silent - when there is nothing to report.""" - mock_driver = _patched_lifespan_deps() - with ( - patch("context_intelligence_server.main.setup_logging"), - patch( - "context_intelligence_server.neo4j_store.AsyncGraphDatabase.driver", - return_value=mock_driver, - ), - patch( - "context_intelligence_server.main.ensure_neo4j_schema", - new=AsyncMock(return_value=True), - ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(return_value=0), - ) as mock_count, - ): - async with lifespan(main_module.app): - pass - - mock_count.assert_awaited_once() - + This test asserts the removal is genuine (not just untested): boot + succeeds with no untagged-count patching of any kind, and the function + main.py used to import for this guard no longer exists on the module. + """ + assert not hasattr(main_module, "count_untagged_nodes"), ( + "count_untagged_nodes must no longer be imported into main -- the " + "untagged-node boot guard was removed; detection lives only on the " + "doctor operator path now." + ) -async def test_lifespan_does_not_raise_when_health_check_itself_fails( - caplog: pytest.LogCaptureFixture, -) -> None: - """A health-check probe failure (e.g. count_untagged_nodes raising due to - a transient connectivity blip) must NOT be treated as a confirmed - un-migrated graph -- it is logged at DEBUG and swallowed, and boot - proceeds. Connectivity failure != confirmed data corruption.""" mock_driver = _patched_lifespan_deps() with ( patch("context_intelligence_server.main.setup_logging"), @@ -1327,11 +1495,6 @@ async def test_lifespan_does_not_raise_when_health_check_itself_fails( "context_intelligence_server.main.ensure_neo4j_schema", new=AsyncMock(return_value=True), ), - patch( - "context_intelligence_server.main.count_untagged_nodes", - new=AsyncMock(side_effect=RuntimeError("transient connectivity blip")), - ), - caplog.at_level(logging.DEBUG), ): async with lifespan(main_module.app): # MUST NOT raise pass From f5549efcbe83118c3217b3bdaa7bac59f4d10b8f Mon Sep 17 00:00:00 2001 From: colombod Date: Wed, 9 Sep 2026 19:34:35 +0000 Subject: [PATCH 2/2] =?UTF-8?q?chore(release):=20version=206.7.3=20?= =?UTF-8?q?=E2=80=94=20production=20outage=20fix=20identifiable=20via=20/v?= =?UTF-8?q?ersion=20and=20/status?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #99 fixes crash-recovery blocking ASGI startup, leaving /version, /status, and POST /events unreachable for over an hour. This version bump ensures deployed instances are identifiable — `GET /version` and `/status` both read importlib.metadata(pyproject.toml [project].version) via SERVER_VERSION, and tests/routers/test_version.py pins the entire chain. Verification: ✓ Editable package reinstalled; SERVER_VERSION=6.7.3 ✓ tests/routers/test_version.py (6 passed) ✓ pytest -m 'not neo4j' (1984 passed, 7 skipped, 92 deselected) ✓ pytest -m neo4j (92 passed) Generated with [Amplifier](https://github.com/microsoft/amplifier) Co-Authored-By: Amplifier <240397093+microsoft-amplifier@users.noreply.github.com> --- pyproject.toml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pyproject.toml b/pyproject.toml index 8dd6420b..ab011196 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "context-intelligence-server" -version = "6.7.2" +version = "6.7.3" description = "Context Intelligence Server for Amplifier" requires-python = ">=3.11" dependencies = [