fix(health): fail readiness on essential worker/scheduler errors (E07)

server/main.py

@@ -1064,8 +1064,12 @@ def _readiness_checks(request: Request):
     """Run synchronous drivers off the event loop with driver-level bounds.
 
     Postgres is required everywhere. Redis is required in strict deployed
-    runtimes, matching auth and checkout enforcement. Neo4j and background
-    runners remain optional: their failures report degraded with HTTP 200.
+    runtimes, matching auth and checkout enforcement. Essential background
+    units (ingestion worker; embedded scheduler when SCHEDULER_EMBEDDED is
+    on) fail readiness with HTTP 503 so a broken deploy cannot promote
+    (diligence E07). Neo4j and the rate limiter stay non-blocking: they
+    report in the payload but never alone flip the status code. /health
+    remains a separate liveness probe and does not consult these units.
     """
     from fastapi.responses import JSONResponse
 
@@ -1165,14 +1169,16 @@ def _readiness_checks(request: Request):
         except Exception as exc:
             checks["neo4j"] = {"status": "error", "error": type(exc).__name__}
 
-    # --- Background runners (#2518) ---
+    # --- Background runners (#2518 / E07) ---
     # The ingestion worker and embedded scheduler start in the lifespan inside
     # best-effort try/except blocks. Before #2518 a swallowed start failure left
     # app.state.* = None while /readiness stayed green, so stuck ingestion /
     # disabled recall alerting was invisible to deploy health checks. Report
-    # their start outcome here. These are non-fatal (degraded, never 503):
-    # mirror the optional-dependency pattern so a background failure surfaces in
-    # the payload without blocking a deploy.
+    # their start outcome and thread liveness here. Diligence E07 makes the
+    # essential units fail readiness (HTTP 503): the ingestion worker always,
+    # and the embedded scheduler only when SCHEDULER_EMBEDDED enables it.
+    # disabled / not_configured stay non-blocking. Neo4j and rate_limiter
+    # remain informational (never alone 503). /health stays separate.
     state = getattr(request.app, "state", None)
 
     # --- Ingestion worker ---
@@ -1200,18 +1206,20 @@ def _readiness_checks(request: Request):
     elif getattr(state, "embedded_scheduler", None) is None:
         checks["scheduler"] = {"status": "disabled"}
     elif not getattr(state.embedded_scheduler, "thread_alive", True):
-        # #2697: the daemon thread can die after a successful start (init
-        # failure inside the thread) — "started ok" alone must not report
-        # ok forever. Degrades the payload; never 503 (documented contract).
+        # #2697 / E07: the daemon thread can die after a successful start
+        # (init failure inside the thread) — "started ok" alone must not
+        # report ok forever. When the scheduler is essential this fails
+        # readiness (503); otherwise it only degrades the payload.
         checks["scheduler"] = {"status": "error", "error": "embedded scheduler thread is not alive"}
     else:
         checks["scheduler"] = {"status": "ok"}
 
-    # --- Rate-limiter storage (#2711) ---
+    # --- Rate-limiter storage (#2711 / E07) ---
     # slowapi decorators (1/day GDPR export, 10/min proof-verify, …) bind to
     # whatever storage existed at import. README asserts Redis-backed rate
     # limiting; production running on memory:// means every such budget is
-    # per-process (× workers × replicas). Degrades the payload; never 503.
+    # per-process (× workers × replicas). Reported in the payload; never
+    # alone flips readiness to 503 (diligence E07: rate limiter non-blocking).
     try:
         from shared.env import is_production as _is_prod_env
         from shared.rate_limit import slowapi_storage_backend
@@ -1232,14 +1240,21 @@ def _readiness_checks(request: Request):
     # Postgres is required for the consolidated API to serve production traffic.
     # Missing DATABASE_URL must fail readiness so deploy health checks do not
     # route requests to an instance that can only satisfy liveness.
+    # Diligence E07: essential background units also fail readiness so a
+    # release cannot promote while ingestion or (when embedded) scheduled
+    # compliance work is unavailable. Liveness (/health) stays separate.
     from shared.env import requires_strict_auth_runtime
 
+    from server.workers.embedded_scheduler import embedded_scheduler_enabled
+
     pg_ok = checks.get("postgres", {}).get("status") == "ok"
     redis_ok = checks.get("redis", {}).get("status") == "ok"
     redis_required = requires_strict_auth_runtime()
+    worker_essential_error = checks.get("worker", {}).get("status") == "error"
+    scheduler_essential_error = checks.get("scheduler", {}).get("status") == "error" and embedded_scheduler_enabled()
     all_ok = all(c.get("status") != "error" for c in checks.values())
 
-    if not pg_ok or (redis_required and not redis_ok):
+    if not pg_ok or (redis_required and not redis_ok) or worker_essential_error or scheduler_essential_error:
         return JSONResponse(
             status_code=503,
             content={

tests/test_server_readiness_probe.py

@@ -5,6 +5,11 @@
 
 * Postgres down or unconfigured -> 503 (fatal; instance cannot serve traffic).
 * Redis down or unconfigured in strict deployed runtimes -> 503.
+* Diligence E07: ingestion worker error (startup failure or dead thread) ->
+  503. Embedded scheduler error -> 503 only when SCHEDULER_EMBEDDED enables
+  it; ``disabled`` / ``not_configured`` stay non-blocking.
+* Neo4j and rate-limiter errors stay non-blocking (reported; never alone
+  flip the status code). /health remains a separate liveness probe.
 * A stalled dependency probe returns 503 within its overall deadline while
   the event loop remains available for liveness requests.
 * Neo4j down (graph subsystem active) -> 200 with overall status ``degraded``
@@ -237,10 +242,8 @@ def test_runners_ok_when_started_and_alive(monkeypatch, _postgres_ok, _redis_uns
     assert body["checks"]["scheduler"]["status"] == "ok"
 
 
-def test_worker_start_failure_degrades_but_stays_200(monkeypatch, _postgres_ok, _redis_unset):
-    """A swallowed ingestion-worker start failure surfaces as worker=error and
-    degrades the payload — but /readiness stays 200 (not 503), so a background
-    failure does not block deploys the way postgres-down does (#2518)."""
+def test_worker_start_failure_returns_503(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: swallowed ingestion-worker start failure fails readiness (503)."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
     client = _client_with_state(
         ingestion_worker_started=False,
@@ -251,16 +254,15 @@ def test_worker_start_failure_degrades_but_stays_200(monkeypatch, _postgres_ok,
 
     resp = client.get("/readiness")
 
-    assert resp.status_code == 200
+    assert resp.status_code == 503
     body = resp.json()
-    assert body["status"] == "degraded"
+    assert body["status"] == "not_ready"
     assert body["checks"]["worker"]["status"] == "error"
-    # postgres-down is the only 503 path; confirm worker-down is NOT one.
     assert body["checks"]["postgres"]["status"] == "ok"
 
 
-def test_worker_dead_thread_is_error(monkeypatch, _postgres_ok, _redis_unset):
-    """Started but the loop thread died -> worker=error/degraded."""
+def test_worker_dead_thread_returns_503(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: started but the loop thread died -> worker=error and 503."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
     client = _client_with_state(
         ingestion_worker_started=True,
@@ -271,16 +273,18 @@ def test_worker_dead_thread_is_error(monkeypatch, _postgres_ok, _redis_unset):
 
     resp = client.get("/readiness")
 
-    assert resp.status_code == 200
-    assert resp.json()["status"] == "degraded"
-    assert resp.json()["checks"]["worker"]["status"] == "error"
+    assert resp.status_code == 503
+    body = resp.json()
+    assert body["status"] == "not_ready"
+    assert body["checks"]["worker"]["status"] == "error"
 
 
 def test_scheduler_disabled_is_healthy(monkeypatch, _postgres_ok, _redis_unset):
     """A None scheduler runner after a successful start is the *intended*
     disabled state (SCHEDULER_EMBEDDED=0), not a failure -> scheduler=disabled,
-    overall stays ready."""
+    overall stays ready (scheduler is not essential when disabled)."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "0")
     client = _client_with_state(
         ingestion_worker_started=True,
         ingestion_worker_thread=_LiveThread(),
@@ -296,9 +300,29 @@ def test_scheduler_disabled_is_healthy(monkeypatch, _postgres_ok, _redis_unset):
     assert body["checks"]["scheduler"]["status"] == "disabled"
 
 
-def test_scheduler_start_failure_degrades(monkeypatch, _postgres_ok, _redis_unset):
-    """A raised embedded-scheduler start -> scheduler=error/degraded (200)."""
+def test_scheduler_start_failure_returns_503_when_embedded(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: raised embedded-scheduler start -> 503 when SCHEDULER_EMBEDDED=1."""
+    monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "1")
+    client = _client_with_state(
+        ingestion_worker_started=True,
+        ingestion_worker_thread=_LiveThread(),
+        embedded_scheduler_started=False,
+        embedded_scheduler=None,
+    )
+
+    resp = client.get("/readiness")
+
+    assert resp.status_code == 503
+    body = resp.json()
+    assert body["status"] == "not_ready"
+    assert body["checks"]["scheduler"]["status"] == "error"
+
+
+def test_scheduler_start_failure_degrades_when_not_embedded(monkeypatch, _postgres_ok, _redis_unset):
+    """Scheduler error is non-essential when SCHEDULER_EMBEDDED=0 -> 200/degraded."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "0")
     client = _client_with_state(
         ingestion_worker_started=True,
         ingestion_worker_thread=_LiveThread(),
@@ -314,9 +338,8 @@ def test_scheduler_start_failure_degrades(monkeypatch, _postgres_ok, _redis_unse
     assert body["checks"]["scheduler"]["status"] == "error"
 
 
-def test_worker_down_distinguished_from_postgres_down(monkeypatch, _redis_unset):
-    """The required-vs-optional contract: postgres-down is 503, worker-down is
-    200/degraded. Same probe, different severity (#2518)."""
+def test_worker_and_postgres_down_both_reported_as_503(monkeypatch, _redis_unset):
+    """Postgres-down and worker-down both fail readiness; both appear in checks."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
     # Configured-but-unreachable (env_defaults deliberately sets no URL).
     url = "postgresql://probe:probe@localhost:5432/readiness_probe"  # pragma: allowlist secret
@@ -335,7 +358,6 @@ def _pg_down(*args, **kwargs):
 
     resp = client.get("/readiness")
 
-    # postgres-down dominates -> 503 regardless of the worker signal.
     assert resp.status_code == 503
     assert resp.json()["checks"]["postgres"]["status"] == "error"
     assert resp.json()["checks"]["worker"]["status"] == "error"
@@ -356,10 +378,10 @@ def test_minimal_app_without_state_reports_not_configured(monkeypatch, _postgres
     assert body["checks"]["scheduler"]["status"] == "not_configured"
 
 
-def test_scheduler_dead_thread_is_error(monkeypatch, _postgres_ok, _redis_unset):
-    """#2697: a scheduler runner whose daemon thread died after a successful
-    start must degrade the payload instead of reporting ok forever."""
+def test_scheduler_dead_thread_returns_503_when_embedded(monkeypatch, _postgres_ok, _redis_unset):
+    """#2697 / E07: dead scheduler thread fails readiness when embedded is on."""
     monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "1")
 
     class _DeadRunner:
         thread_alive = False
@@ -373,10 +395,33 @@ class _DeadRunner:
 
     resp = client.get("/readiness")
 
-    assert resp.status_code == 200  # degraded, never 503 (documented contract)
+    assert resp.status_code == 503
     body = resp.json()
     assert body["checks"]["scheduler"]["status"] == "error"
     assert "not alive" in body["checks"]["scheduler"]["error"]
+    assert body["status"] == "not_ready"
+
+
+def test_scheduler_dead_thread_degrades_when_not_embedded(monkeypatch, _postgres_ok, _redis_unset):
+    """Dead scheduler thread only degrades when SCHEDULER_EMBEDDED=0."""
+    monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "0")
+
+    class _DeadRunner:
+        thread_alive = False
+
+    client = _client_with_state(
+        ingestion_worker_started=True,
+        ingestion_worker_thread=_LiveThread(),
+        embedded_scheduler_started=True,
+        embedded_scheduler=_DeadRunner(),
+    )
+
+    resp = client.get("/readiness")
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["checks"]["scheduler"]["status"] == "error"
     assert body["status"] == "degraded"
 
 
@@ -621,3 +666,110 @@ async def exercise():
     with ThreadPoolExecutor(max_workers=2) as executor:
         monkeypatch.setattr(main, "_READINESS_EXECUTOR", executor)
         asyncio.run(exercise())
+
+
+def test_e07_neo4j_error_stays_non_blocking(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: Neo4j stays non-blocking even when essential units are healthy."""
+    monkeypatch.setenv("NEO4J_URI", "bolt://neo4j.invalid:7687")
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "1")
+    monkeypatch.setattr(
+        neo4j.GraphDatabase,
+        "driver",
+        lambda *a, **k: (_ for _ in ()).throw(OSError("connection refused")),
+    )
+    client = _client_with_state(
+        ingestion_worker_started=True,
+        ingestion_worker_thread=_LiveThread(),
+        embedded_scheduler_started=True,
+        embedded_scheduler=object(),
+    )
+
+    resp = client.get("/readiness")
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["status"] == "degraded"
+    assert body["checks"]["neo4j"]["status"] == "error"
+    assert body["checks"]["worker"]["status"] == "ok"
+
+
+def test_e07_rate_limiter_error_alone_stays_non_blocking(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: rate_limiter error alone never flips readiness to 503."""
+    monkeypatch.delenv("NEO4J_URI", raising=False)
+    import shared.env as env_mod
+    import shared.rate_limit as rl_mod
+
+    monkeypatch.setattr(rl_mod, "slowapi_storage_backend", lambda: "memory")
+    # Force the rate-limiter error path without making Redis required.
+    monkeypatch.setattr(env_mod, "is_production", lambda: True)
+    monkeypatch.setattr(env_mod, "requires_strict_auth_runtime", lambda: False)
+
+    client = _client_with_state(
+        ingestion_worker_started=True,
+        ingestion_worker_thread=_LiveThread(),
+        embedded_scheduler_started=True,
+        embedded_scheduler=object(),
+    )
+    resp = client.get("/readiness")
+
+    assert resp.status_code == 200
+    body = resp.json()
+    assert body["checks"]["rate_limiter"]["status"] == "error"
+    assert body["status"] == "degraded"
+
+
+def test_e07_readiness_recovers_to_200_after_restart(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: after a failed essential unit, a healthy restart returns 200."""
+    monkeypatch.delenv("NEO4J_URI", raising=False)
+    monkeypatch.setenv("SCHEDULER_EMBEDDED", "1")
+
+    failed = _client_with_state(
+        ingestion_worker_started=False,
+        ingestion_worker_thread=None,
+        embedded_scheduler_started=False,
+        embedded_scheduler=None,
+    )
+    failed_resp = failed.get("/readiness")
+    assert failed_resp.status_code == 503
+    assert failed_resp.json()["status"] == "not_ready"
+
+    class _LiveRunner:
+        thread_alive = True
+
+    recovered = _client_with_state(
+        ingestion_worker_started=True,
+        ingestion_worker_thread=_LiveThread(),
+        embedded_scheduler_started=True,
+        embedded_scheduler=_LiveRunner(),
+    )
+    recovered_resp = recovered.get("/readiness")
+    assert recovered_resp.status_code == 200
+    body = recovered_resp.json()
+    assert body["status"] == "ready"
+    assert body["checks"]["worker"]["status"] == "ok"
+    assert body["checks"]["scheduler"]["status"] == "ok"
+
+
+def test_e07_liveness_unaffected_when_essential_unit_fails(monkeypatch, _postgres_ok, _redis_unset):
+    """E07: /health stays 200 while /readiness is 503 for an essential failure.
+
+    Keeps Railway from coupling liveness to background-unit death (no restart
+    loop driven by a combined probe).
+    """
+    monkeypatch.delenv("NEO4J_URI", raising=False)
+    app = FastAPI()
+    app.state.ingestion_worker_started = False
+    app.state.ingestion_worker_thread = None
+    app.state.embedded_scheduler_started = True
+    app.state.embedded_scheduler = object()
+    app.add_api_route("/readiness", main.readiness, methods=["GET"])
+
+    @app.get("/health")
+    async def health():
+        return {"status": "alive"}
+
+    client = TestClient(app)
+    assert client.get("/readiness").status_code == 503
+    live = client.get("/health")
+    assert live.status_code == 200
+    assert live.json()["status"] == "alive"