From 6ef7786cecc9d0e5b2b8c2a62b7caf043351aa16 Mon Sep 17 00:00:00 2001 From: Chris Coutinho Date: Wed, 10 Jun 2026 21:28:19 +0200 Subject: [PATCH] fix(health): non-gating readiness probe; shared-task-group lifespan; settings migration Fixes MCP reconnect timeouts on tenant servers (Deck #302). Three changes: - /health/ready now gates only on local config. Nextcloud/Qdrant health is refreshed by a background loop, cached, and reported but NON-gating, so a single-replica tenant Pod is no longer pulled from its Service on a transient dependency blip (which dropped every MCP streamable-HTTP session and caused reconnect timeouts). The probe path performs no external I/O. - Refactor starlette_lifespan: collapse the four near-identical per-mode task-group + session + yield + teardown skeletons into one shared task group that also runs the readiness refresh loop; each mode contributes a (start, teardown) pair. eviction_task_group is now always present. - Migrate app.py off os.getenv: all config is read through dynaconf Settings (adds health_ready_refresh_interval, oidc_token_type, oidc_scopes, port). Inline/dynamic defaults preserved at each call site. Co-Authored-By: Claude Opus 4.8 (1M context) --- docs/configuration.md | 18 + nextcloud_mcp_server/app.py | 453 ++++++++++-------- nextcloud_mcp_server/config.py | 15 + .../observability/readiness.py | 75 +++ tests/unit/test_hybrid_auth_setup.py | 22 +- tests/unit/test_readiness_cache.py | 64 +++ 6 files changed, 420 insertions(+), 227 deletions(-) create mode 100644 nextcloud_mcp_server/observability/readiness.py create mode 100644 tests/unit/test_readiness_cache.py diff --git a/docs/configuration.md b/docs/configuration.md index 064c9995..18223d71 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -237,6 +237,24 @@ They do **not** affect connections to internal services (Ollama, Qdrant, Unstruc --- +## Health & Readiness Probes + +The server exposes two Kubernetes probe endpoints: + +- `GET /health/live` — liveness. Returns `200` whenever the process is running. It does **not** check external dependencies, so it never restarts the Pod on an upstream blip. +- `GET /health/ready` — readiness. Gates **only** on local configuration (`NEXTCLOUD_HOST` set, auth mode configured). External-dependency reachability (Nextcloud `status.php`, Qdrant `/readyz`) is reported in the response body for observability but is **non-gating**. + +> **Why non-gating (Deck #302):** the server typically runs as a single replica per tenant. If readiness failed whenever Nextcloud or Qdrant had a transient blip, the only Pod would be pulled from its Service, leaving the gateway with no upstream — turning a *degraded* dependency into a *total* outage and dropping every MCP client's streamable-HTTP session. Dependency health is instead refreshed by a background loop and cached, so the probe path performs no external I/O. + +```dotenv +# Cadence (seconds) for the background dependency-health refresh loop (default: 15) +HEALTH_READY_REFRESH_INTERVAL=15 +``` + +The probe reports each dependency under `checks` (`ok` / `embedded` / `pending` / `error: ...`); a non-`ok` dependency no longer flips the overall `status` to `not_ready`. + +--- + ## Semantic Search Configuration (Optional) **New in v0.58.0:** Simplified semantic search configuration with automatic dependency resolution. diff --git a/nextcloud_mcp_server/app.py b/nextcloud_mcp_server/app.py index 838480f2..d8630843 100644 --- a/nextcloud_mcp_server/app.py +++ b/nextcloud_mcp_server/app.py @@ -119,6 +119,7 @@ from nextcloud_mcp_server.observability.metrics import ( record_dependency_check, set_dependency_health, ) +from nextcloud_mcp_server.observability.readiness import ReadinessCache from nextcloud_mcp_server.server import ( AVAILABLE_APPS, configure_semantic_tools, @@ -417,6 +418,100 @@ def _clear_vector_sync_state() -> None: _vector_sync_state.scanner_wake_event = None +# ============================================================================= +# Readiness dependency health (Deck #302) +# ============================================================================= +# +# Readiness must reflect "this process is up and configured to serve", NOT the +# live reachability of shared external dependencies. The MCP server typically +# runs as a single replica per tenant; failing readiness when Nextcloud or +# Qdrant blips would pull the only Pod out of its Service, leaving the gateway +# with no upstream and turning a degraded dependency into a total outage plus +# an MCP reconnect storm. So external dependency health is refreshed by a +# background loop, cached, and *reported but non-gating* — and the probe path +# never performs external I/O. + +_readiness_cache = ReadinessCache(ttl_seconds=30.0) + + +async def _check_nextcloud_health() -> None: + """Probe Nextcloud ``status.php`` and record the result in the cache. + + Catches everything: the refresh loop must never crash on a dependency + error, and a failed check is just an unhealthy status, not an exception. + """ + host = get_settings().nextcloud_host + if not host: + return + start = time.time() + try: + async with nextcloud_httpx_client(timeout=2.0) as client: + response = await client.get(f"{host}/status.php") + healthy = response.status_code == 200 + detail = "ok" if healthy else f"error: status {response.status_code}" + except Exception as e: # noqa: BLE001 - any failure is "unhealthy" + healthy = False + detail = f"error: {e}" + _readiness_cache.update("nextcloud_reachable", healthy, detail) + set_dependency_health("nextcloud", healthy) + record_dependency_check("nextcloud", time.time() - start) + + +async def _check_qdrant_health() -> None: + """Probe Qdrant ``/readyz`` (network mode) and record the result. + + Qdrant Cloud's auth gateway 403s unauthenticated requests, so forward the + same api-key the configured client uses (see vector/qdrant_client.py). + """ + settings = get_settings() + qdrant_url = settings.qdrant_url + if not qdrant_url: + return + headers = {"api-key": settings.qdrant_api_key} if settings.qdrant_api_key else {} + start = time.time() + try: + async with httpx.AsyncClient(timeout=2.0) as client: + response = await client.get(f"{qdrant_url}/readyz", headers=headers) + healthy = response.status_code == 200 + detail = "ok" if healthy else f"error: status {response.status_code}" + except Exception as e: # noqa: BLE001 - any failure is "unhealthy" + healthy = False + detail = f"error: {e}" + _readiness_cache.update("qdrant", healthy, detail) + set_dependency_health("qdrant", healthy) + record_dependency_check("qdrant", time.time() - start) + + +async def _refresh_dependency_health() -> None: + """Refresh all external dependency statuses concurrently (one pass).""" + settings = get_settings() + async with anyio.create_task_group() as tg: + tg.start_soon(_check_nextcloud_health) + if settings.vector_sync_enabled and settings.qdrant_url: + tg.start_soon(_check_qdrant_health) + elif settings.vector_sync_enabled: + # Embedded Qdrant (memory/persistent mode) — no external service. + _readiness_cache.update("qdrant", True, "embedded") + set_dependency_health("qdrant", True) + + +async def _readiness_refresh_loop() -> None: + """Background loop that keeps ``_readiness_cache`` warm off the probe path. + + Started in the shared lifespan task group; cancelled on shutdown. + """ + interval = get_settings().health_ready_refresh_interval + logger.info( + "Readiness dependency-health refresh loop started (every %ss)", interval + ) + while True: + try: + await _refresh_dependency_health() + except Exception: # noqa: BLE001 - never let the loop die + logger.debug("Readiness dependency refresh iteration failed", exc_info=True) + await anyio.sleep(interval) + + @dataclass class AppContext: """Application context for BasicAuth mode.""" @@ -536,8 +631,8 @@ async def load_oauth_client_credentials( ValueError: If credentials cannot be obtained """ # Try environment variables first - client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID") - client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET") + client_id = get_settings().oidc_client_id + client_secret = get_settings().oidc_client_secret if client_id and client_secret: logger.info("Using pre-configured OAuth client credentials from environment") @@ -561,7 +656,9 @@ async def load_oauth_client_credentials( # Try dynamic registration if available if registration_endpoint: logger.info("Dynamic client registration available") - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") + mcp_server_url = ( + get_settings().nextcloud_mcp_server_url or "http://localhost:8000" + ) redirect_uris = [ f"{mcp_server_url}/oauth/callback", # Unified callback (flow determined by query param) ] @@ -600,7 +697,7 @@ async def load_oauth_client_credentials( # Get token type from environment (Bearer or jwt) # Note: Must be lowercase "jwt" to match OIDC app's check - token_type = os.getenv("NEXTCLOUD_OIDC_TOKEN_TYPE", "Bearer").lower() + token_type = get_settings().oidc_token_type.lower() # Special case: "bearer" should remain capitalized for compatibility if token_type != "jwt": token_type = "Bearer" @@ -728,7 +825,7 @@ async def setup_oauth_config(): # and ENABLE_OFFLINE_ACCESS environment variables) settings = get_settings() - nextcloud_host = os.getenv("NEXTCLOUD_HOST") + nextcloud_host = settings.nextcloud_host if not nextcloud_host: raise ValueError( "NEXTCLOUD_HOST environment variable is required for OAuth mode" @@ -737,8 +834,9 @@ async def setup_oauth_config(): nextcloud_host = nextcloud_host.rstrip("/") # Get OIDC discovery URL (defaults to Nextcloud integrated mode) - discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", f"{nextcloud_host}/.well-known/openid-configuration" + discovery_url = ( + settings.oidc_discovery_url + or f"{nextcloud_host}/.well-known/openid-configuration" ) logger.info("Performing OIDC discovery: %s", discovery_url) @@ -811,7 +909,7 @@ async def setup_oauth_config(): if enable_offline_access: try: # Validate encryption key before initializing - encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY") + encryption_key = settings.token_encryption_key if not encryption_key: logger.warning( "ENABLE_OFFLINE_ACCESS=true but TOKEN_ENCRYPTION_KEY not set. " @@ -831,8 +929,8 @@ async def setup_oauth_config(): ) # Load client credentials (static or dynamic registration) - client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID") - client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET") + client_id = settings.oidc_client_id + client_secret = settings.oidc_client_secret if client_id and client_secret: logger.info("Using static OIDC client credentials: %s", client_id) @@ -856,16 +954,16 @@ async def setup_oauth_config(): public_issuer_url = settings.nextcloud_public_issuer_url client_issuer = public_issuer_url if public_issuer_url else issuer # Get MCP server URL for audience validation - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") - nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host) + mcp_server_url = settings.nextcloud_mcp_server_url or "http://localhost:8000" + nextcloud_resource_uri = settings.nextcloud_resource_uri or nextcloud_host # Warn if resource URIs are not configured (required for ADR-005 compliance) - if not os.getenv("NEXTCLOUD_MCP_SERVER_URL"): + if not settings.nextcloud_mcp_server_url: logger.warning( "NEXTCLOUD_MCP_SERVER_URL not set, defaulting to: %s. This should be set explicitly for proper audience validation.", mcp_server_url, ) - if not os.getenv("NEXTCLOUD_RESOURCE_URI"): + if not settings.nextcloud_resource_uri: logger.warning( "NEXTCLOUD_RESOURCE_URI not set, defaulting to: %s. This should be set explicitly for proper audience validation.", nextcloud_resource_uri, @@ -908,7 +1006,7 @@ async def setup_oauth_config(): logger.info("✓ JWT signature verification enabled (JWKS)") # Progressive Consent mode (for offline access / background jobs) - encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY") + encryption_key = settings.token_encryption_key if enable_offline_access and encryption_key and refresh_token_storage: logger.info("✓ Progressive Consent mode enabled - offline access available") @@ -920,7 +1018,7 @@ async def setup_oauth_config(): oauth_client = None # Create auth settings - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") + mcp_server_url = settings.nextcloud_mcp_server_url or "http://localhost:8000" # Note: We don't set required_scopes here anymore. # Scopes are now advertised via PRM endpoint and enforced per-tool. @@ -985,9 +1083,9 @@ async def setup_oauth_config_for_multi_user_basic( nextcloud_host = nextcloud_host.rstrip("/") # Get OIDC discovery URL (always Nextcloud integrated mode for multi-user BasicAuth) - discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - f"{nextcloud_host}/.well-known/openid-configuration", + discovery_url = ( + settings.oidc_discovery_url + or f"{nextcloud_host}/.well-known/openid-configuration" ) logger.info( "Performing OIDC discovery for multi-user BasicAuth hybrid mode: %s", @@ -1041,8 +1139,8 @@ async def setup_oauth_config_for_multi_user_basic( logger.info(" Introspection: %s", introspection_uri) # Get MCP server URL for audience validation - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000") - nextcloud_resource_uri = os.getenv("NEXTCLOUD_RESOURCE_URI", nextcloud_host) + mcp_server_url = settings.nextcloud_mcp_server_url or "http://localhost:8000" + nextcloud_resource_uri = settings.nextcloud_resource_uri or nextcloud_host # Use public issuer URL for JWT validation if set (handles Docker internal/external URL mismatch) # Tokens are issued with the public URL, but OIDC discovery returns internal URL @@ -1080,7 +1178,7 @@ async def setup_oauth_config_for_multi_user_basic( refresh_token_storage = None if settings.enable_offline_access: try: - encryption_key = os.getenv("TOKEN_ENCRYPTION_KEY") + encryption_key = settings.token_encryption_key if not encryption_key: logger.warning( "ENABLE_OFFLINE_ACCESS=true but TOKEN_ENCRYPTION_KEY not set. " @@ -1182,8 +1280,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = ) # Check for static credentials first - static_client_id = os.getenv("NEXTCLOUD_OIDC_CLIENT_ID") - static_client_secret = os.getenv("NEXTCLOUD_OIDC_CLIENT_SECRET") + static_client_id = settings.oidc_client_id + static_client_secret = settings.oidc_client_secret if static_client_id and static_client_secret: logger.info("Using static OAuth credentials for background operations") @@ -1565,17 +1663,17 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = if not nextcloud_host_for_context: raise ValueError("NEXTCLOUD_HOST is required for OAuth mode") - mcp_server_url = os.getenv( - "NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000" + mcp_server_url = ( + settings.nextcloud_mcp_server_url or "http://localhost:8000" ) - nextcloud_resource_uri = os.getenv( - "NEXTCLOUD_RESOURCE_URI", nextcloud_host_for_context + nextcloud_resource_uri = ( + settings.nextcloud_resource_uri or nextcloud_host_for_context ) - discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - f"{nextcloud_host_for_context}/.well-known/openid-configuration", + discovery_url = ( + settings.oidc_discovery_url + or f"{nextcloud_host_for_context}/.well-known/openid-configuration" ) - scopes = os.getenv("NEXTCLOUD_OIDC_SCOPES", "") + scopes = settings.oidc_scopes oauth_context_dict = { "storage": refresh_token_storage, @@ -1630,12 +1728,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Create oauth_context for management API authentication nextcloud_host_for_context = settings.nextcloud_host - mcp_server_url = os.getenv( - "NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000" + mcp_server_url = ( + settings.nextcloud_mcp_server_url or "http://localhost:8000" ) - discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - f"{nextcloud_host_for_context}/.well-known/openid-configuration", + discovery_url = ( + settings.oidc_discovery_url + or f"{nextcloud_host_for_context}/.well-known/openid-configuration" ) oauth_context_dict = { @@ -1698,8 +1796,21 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # Note: enable_offline_access_for_sync, encryption_key, and refresh_token_storage # are already defined in outer scope before mode split - # Multi-user BasicAuth uses OAuth-style background sync (with app passwords) - # So skip single-user BasicAuth vector sync if in multi-user mode + # Each deployment mode contributes its background-sync work as a + # (start, teardown) pair; the shared task group further below runs that + # work alongside the readiness health-refresh loop and yields once + # through the MCP session manager — collapsing four near-identical + # task-group + session + yield + teardown skeletons into one. + async def _noop_start(tg: TaskGroup) -> None: + return + + async def _noop_teardown() -> None: + return + + start, teardown = _noop_start, _noop_teardown + + # Multi-user BasicAuth uses OAuth-style background sync (with app + # passwords); single-user BasicAuth sync is skipped in multi-user mode. if ( settings.vector_sync_enabled and not oauth_enabled @@ -1708,8 +1819,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # BasicAuth mode - single user sync logger.info("Starting background vector sync tasks for BasicAuth mode") - # Get username from environment - username = os.getenv("NEXTCLOUD_USERNAME") + # Get username from settings + username = settings.nextcloud_username if not username: raise ValueError( "NEXTCLOUD_USERNAME required for vector sync in BasicAuth mode" @@ -1752,9 +1863,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = app, ingest_transport, shutdown_event, scanner_wake_event ) - # Start background tasks using anyio TaskGroup - async with anyio.create_task_group() as tg: - # Start scanner task (publishes to the transport's producer) + # Background-sync work for this mode; the shared runner starts it. + async def _single_user_start(tg: TaskGroup) -> None: + # Scanner publishes to the transport's producer. await tg.start( scanner_task, ingest_transport.producer, @@ -1764,13 +1875,12 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = username, ) - # Start the in-process consumer pool. ``run_consumers`` is a - # no-op for the distributed (postgres) backend — its consumer is - # the out-of-process ``worker`` role. The closure binds this - # mode's shared client+username and forwards anyio's injected - # ``task_status`` so ``tg.start`` observes each worker's - # readiness. One shared receive stream + N workers ⇒ a single - # multiplexed queue processed with N-way parallelism (ADR-028). + # In-process consumer pool. ``run_consumers`` is a no-op for the + # distributed (postgres) backend — the out-of-process ``worker`` + # role consumes there. The closure binds this mode's shared + # client+username and forwards anyio's injected ``task_status``. + # One shared receive stream + N workers ⇒ a single multiplexed + # queue processed with N-way parallelism (ADR-028). async def spawn_worker( worker_id, receive_stream, *, task_status=anyio.TASK_STATUS_IGNORED ): @@ -1787,7 +1897,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, spawn_worker, settings.vector_sync_processor_workers ) - # Publish outstanding-work + corpus gauges on a fixed cadence, + # Outstanding-work + corpus gauges on a fixed cadence, # independent of the consumer path and queue backend (fixes the # gauge reading 0 on the multi-user path; see metrics_publisher). # receive_stream is None in postgres mode — get_ingest_pending @@ -1799,37 +1909,23 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event, ) - # Expose this long-lived task group to request-path code that - # wants to spawn background work (e.g. ADR-019 verify-on-read - # eviction). Eviction coroutines have their own try/except, so - # they cannot panic the parent group. - _vector_sync_state.eviction_task_group = tg - logger.info( "Background sync tasks started: 1 scanner + %s processors (queue=%s)", ingest_transport.active_consumer_count, ingest_transport.backend_name, ) - # Run MCP session manager and yield - async with _mcp_session_with_login_flow(app): - try: - yield - finally: - # Shutdown signal - logger.info("Shutting down background sync tasks") - shutdown_event.set() - # Request path must not spawn into a cancelling group. - _vector_sync_state.eviction_task_group = None - # Tear down backend-owned resources (closes the - # procrastinate connector pool in postgres mode; no-op - # for the memory stream, which task-group cancellation - # closes). - await ingest_transport.aclose() - # Drop stale singleton refs to the now-closed transport. - _clear_vector_sync_state() - await client.close() - # TaskGroup automatically cancels all tasks on exit + async def _single_user_teardown() -> None: + shutdown_event.set() + # Tear down backend-owned resources (closes the procrastinate + # connector pool in postgres mode; no-op for the memory stream, + # which task-group cancellation closes). + await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() + await client.close() + + start, teardown = _single_user_start, _single_user_teardown elif ( settings.vector_sync_enabled @@ -1847,9 +1943,9 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = raise ValueError("NEXTCLOUD_HOST required for vector sync") # Get OIDC discovery URL (same as used for OAuth setup) - discovery_url = os.getenv( - "OIDC_DISCOVERY_URL", - f"{nextcloud_host_for_sync}/.well-known/openid-configuration", + discovery_url = ( + settings.oidc_discovery_url + or f"{nextcloud_host_for_sync}/.well-known/openid-configuration" ) # Get client credentials - these were obtained before uvicorn started @@ -1965,10 +2061,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = # never reachable from any supported deployment mode. The # `token_broker` constructed above is still used by the # management API revoke endpoint (via app.state.oauth_context). - async with anyio.create_task_group() as tg: - # Start user manager task (supervises per-user scanners). - # Each per-user scanner clones the producer; for the bus - # producer clone() returns the shared connection. + async def _multi_user_start(tg: TaskGroup) -> None: + # User manager supervises per-user scanners. Each per-user + # scanner clones the producer; for the bus producer clone() + # returns the shared connection. await tg.start( user_manager_task, ingest_transport.producer, @@ -1980,14 +2076,14 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, ) - # Start the in-process consumer pool. ``run_consumers`` is a - # no-op for the distributed (postgres) backend — the - # out-of-process ``worker`` role consumes there. The closure - # binds this mode's nextcloud_host (per-document credential - # resolution) and forwards anyio's injected ``task_status``. - # One shared receive stream + N workers ⇒ a single - # multiplexed queue draining every user's documents with - # N-way parallelism, never one user at a time (ADR-028). + # In-process consumer pool. ``run_consumers`` is a no-op for + # the distributed (postgres) backend — the out-of-process + # ``worker`` role consumes there. The closure binds this + # mode's nextcloud_host (per-document credential resolution) + # and forwards anyio's injected ``task_status``. One shared + # receive stream + N workers ⇒ a single multiplexed queue + # draining every user's documents with N-way parallelism, + # never one user at a time (ADR-028). async def spawn_worker( worker_id, receive_stream, @@ -2006,8 +2102,8 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = tg, spawn_worker, settings.vector_sync_processor_workers ) - # Publish outstanding-work + corpus gauges on a fixed - # cadence. Critical on this multi-user path: the consumer is + # Outstanding-work + corpus gauges on a fixed cadence. + # Critical on this multi-user path: the consumer is # oauth_processor_task, which never updated the queue gauge, # so without this the gauge read 0 while the buffer held # thousands of pending docs (see metrics_publisher). @@ -2020,38 +2116,24 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = shutdown_event, ) - # Expose this long-lived task group to request-path code - # that wants to spawn background work (e.g. ADR-019 - # verify-on-read eviction). Eviction coroutines have their - # own try/except, so they cannot panic the parent group. - _vector_sync_state.eviction_task_group = tg - logger.info( "Background sync tasks started: 1 user manager + %s processors (queue=%s)", ingest_transport.active_consumer_count, ingest_transport.backend_name, ) - # Run MCP session manager and yield - async with _mcp_session_with_login_flow(app): - try: - yield - finally: - # Shutdown signal - logger.info("Shutting down background sync tasks") - shutdown_event.set() - # Request path must not spawn into a cancelling group. - _vector_sync_state.eviction_task_group = None - # Tear down backend-owned resources (closes the - # procrastinate connector pool in postgres mode; - # no-op for the memory stream). - await ingest_transport.aclose() - # Drop stale singleton refs to the now-closed transport. - _clear_vector_sync_state() - # Close token broker HTTP client - if token_broker._http_client: - await token_broker._http_client.aclose() - # TaskGroup automatically cancels all tasks on exit + async def _multi_user_teardown() -> None: + shutdown_event.set() + # Tear down backend-owned resources (closes the procrastinate + # connector pool in postgres mode; no-op for the memory stream). + await ingest_transport.aclose() + # Drop stale singleton refs to the now-closed transport. + _clear_vector_sync_state() + # Close token broker HTTP client + if token_broker._http_client: + await token_broker._http_client.aclose() + + start, teardown = _multi_user_start, _multi_user_teardown else: # No OAuth credentials available for background sync logger.warning( @@ -2059,9 +2141,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = "Multi-user BasicAuth mode will run without semantic search background operations. " "To enable, set NEXTCLOUD_OIDC_CLIENT_ID and NEXTCLOUD_OIDC_CLIENT_SECRET." ) - # Just run MCP session manager without vector sync - async with _mcp_session_with_login_flow(app): - yield + # start/teardown stay no-op; the shared runner yields below. else: # No vector sync - just run MCP session manager @@ -2076,12 +2156,28 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = logger.warning( "Vector sync enabled but refresh token storage not available" ) - elif oauth_enabled and not os.getenv("TOKEN_ENCRYPTION_KEY"): + elif oauth_enabled and not settings.token_encryption_key: logger.warning( "Vector sync enabled but TOKEN_ENCRYPTION_KEY not set" ) + # start/teardown stay no-op; the shared runner yields below. + + # One shared task group runs this mode's background tasks plus the + # readiness health-refresh loop, then yields through the MCP session + # manager. The group is exposed for request-path background work + # (ADR-019 verify-on-read eviction) and cancels every task on exit. + async with anyio.create_task_group() as tg: + await start(tg) + tg.start_soon(_readiness_refresh_loop) + _vector_sync_state.eviction_task_group = tg async with _mcp_session_with_login_flow(app): - yield + try: + yield + finally: + logger.info("Shutting down background tasks") + # Request path must not spawn into a cancelling group. + _vector_sync_state.eviction_task_group = None + await teardown() # Health check endpoints for Kubernetes probes def health_live(request): @@ -2100,46 +2196,25 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = async def health_ready(request): """Readiness probe endpoint. - Returns 200 OK if the application is ready to serve traffic. - Checks that required configuration is present and Qdrant if vector sync enabled. + Gates **only** on local, cheap configuration checks (that the process is + up and configured to serve). External dependency reachability (Nextcloud, + Qdrant) is reported for observability but is intentionally *non-gating* + and served from a background-refreshed cache, so the probe performs no + external I/O and a shared-dependency blip cannot pull a single-replica + Pod out of its Service (Deck #302). """ - checks = {} + checks: dict[str, object] = {} is_ready = True + settings = get_settings() - # Check Nextcloud host configuration and connectivity - nextcloud_host = os.getenv("NEXTCLOUD_HOST") - if nextcloud_host: + # --- Local, hard gates ------------------------------------------------ + if settings.nextcloud_host: checks["nextcloud_configured"] = "ok" - # Try to connect to Nextcloud - start_time = time.time() - try: - async with nextcloud_httpx_client(timeout=2.0) as client: - response = await client.get(f"{nextcloud_host}/status.php") - duration = time.time() - start_time - if response.status_code == 200: - checks["nextcloud_reachable"] = "ok" - set_dependency_health("nextcloud", True) - else: - checks["nextcloud_reachable"] = ( - f"error: status {response.status_code}" - ) - set_dependency_health("nextcloud", False) - is_ready = False - record_dependency_check("nextcloud", duration) - except Exception as e: - duration = time.time() - start_time - checks["nextcloud_reachable"] = f"error: {str(e)}" - set_dependency_health("nextcloud", False) - record_dependency_check("nextcloud", duration) - is_ready = False else: checks["nextcloud_configured"] = "error: NEXTCLOUD_HOST not set" - set_dependency_health("nextcloud", False) is_ready = False - # Check authentication configuration - # Report the deployment mode, not just whether OAuth is enabled - # This helps clients (like Astrolabe) determine which auth flow to use + # Report the deployment mode (helps clients pick the auth flow). if mode == AuthMode.LOGIN_FLOW: checks["auth_mode"] = "oauth" checks["auth_configured"] = "ok" @@ -2147,61 +2222,19 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = checks["auth_mode"] = "multi_user_basic" checks["auth_configured"] = "ok" # Indicate if app passwords are supported (when offline_access enabled) - checks["supports_app_passwords"] = get_settings().enable_offline_access + checks["supports_app_passwords"] = settings.enable_offline_access elif mode == AuthMode.SINGLE_USER_BASIC: - username = os.getenv("NEXTCLOUD_USERNAME") - password = os.getenv("NEXTCLOUD_PASSWORD") - if username and password: - checks["auth_mode"] = "basic" + checks["auth_mode"] = "basic" + if settings.nextcloud_username and settings.nextcloud_password: checks["auth_configured"] = "ok" else: - checks["auth_mode"] = "basic" checks["auth_configured"] = "error: credentials not set" is_ready = False - # Check Qdrant status if using network mode (external Qdrant service) - # In-memory and persistent modes use embedded Qdrant, no external service to check - # Note: get_settings() supports both ENABLE_SEMANTIC_SEARCH and VECTOR_SYNC_ENABLED - settings = get_settings() - vector_sync_enabled = settings.vector_sync_enabled - qdrant_url = os.getenv("QDRANT_URL") # Only set in network mode - - if vector_sync_enabled and qdrant_url: - start_time = time.time() - # Self-hosted Qdrant exposes /readyz unauthenticated, but - # Qdrant Cloud's auth gateway returns 403 for any - # unauthenticated request — so we have to forward the same - # api-key the configured AsyncQdrantClient uses (see - # vector/qdrant_client.py). Without this header, every - # readiness probe against a Cloud cluster returns 503, - # blocking the Pod from reaching Ready. - qdrant_headers = ( - {"api-key": settings.qdrant_api_key} if settings.qdrant_api_key else {} - ) - try: - async with httpx.AsyncClient(timeout=2.0) as client: - response = await client.get( - f"{qdrant_url}/readyz", headers=qdrant_headers - ) - duration = time.time() - start_time - if response.status_code == 200: - checks["qdrant"] = "ok" - set_dependency_health("qdrant", True) - else: - checks["qdrant"] = f"error: status {response.status_code}" - set_dependency_health("qdrant", False) - is_ready = False - record_dependency_check("qdrant", duration) - except Exception as e: - duration = time.time() - start_time - checks["qdrant"] = f"error: {str(e)}" - set_dependency_health("qdrant", False) - record_dependency_check("qdrant", duration) - is_ready = False - elif vector_sync_enabled: - # Using embedded Qdrant (memory or persistent mode) - checks["qdrant"] = "embedded" - set_dependency_health("qdrant", True) + # --- External dependencies: reported, NON-gating ---------------------- + # Read the background-refreshed snapshot; never do I/O on the probe path. + for name, status in _readiness_cache.snapshot().items(): + checks[name] = status.detail status_code = 200 if is_ready else 503 return JSONResponse( @@ -2372,10 +2405,10 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = """ # RFC 9728 requires resource to be a URL (not a client ID) # Use the MCP server's public URL - mcp_server_url = os.getenv("NEXTCLOUD_MCP_SERVER_URL") + mcp_server_url = settings.nextcloud_mcp_server_url if not mcp_server_url: # Fallback to constructing from host and port - mcp_server_url = f"http://localhost:{os.getenv('PORT', '8000')}" + mcp_server_url = f"http://localhost:{settings.port}" # Dynamically discover all scopes from registered tools # This provides a single source of truth based on @require_scopes decorators @@ -2699,9 +2732,7 @@ def get_app(transport: str = "streamable-http", enabled_apps: list[str] | None = @app.exception_handler(InsufficientScopeError) async def handle_insufficient_scope(request, exc: InsufficientScopeError): """Return 403 with WWW-Authenticate header for scope challenges.""" - resource_url = os.getenv( - "NEXTCLOUD_MCP_SERVER_URL", "http://localhost:8000" - ) + resource_url = settings.nextcloud_mcp_server_url or "http://localhost:8000" scope_str = " ".join(exc.missing_scopes) return JSONResponse( diff --git a/nextcloud_mcp_server/config.py b/nextcloud_mcp_server/config.py index b891df78..1392fe19 100644 --- a/nextcloud_mcp_server/config.py +++ b/nextcloud_mcp_server/config.py @@ -37,6 +37,9 @@ _DEFAULTS: dict[str, Any] = { "cookie_secure": None, # OAuth/OIDC "oidc_discovery_url": None, + "oidc_token_type": "Bearer", + "oidc_scopes": "", + "port": 8000, "nextcloud_oidc_client_id": None, "nextcloud_oidc_client_secret": None, "oidc_issuer": None, @@ -90,6 +93,7 @@ _DEFAULTS: dict[str, Any] = { "vector_sync_queue_max_size": 10000, "vector_sync_metrics_refresh_interval": 20, "vector_sync_user_poll_interval": 60, + "health_ready_refresh_interval": 15, # Orphan-sweep at Pod startup (card #101). When True, delete any # placeholders carrying a different / absent ``instance_id`` before # the scanner's first cycle, so a Pod restart mid-batch doesn't @@ -600,6 +604,9 @@ class Settings: oidc_client_secret: str | None = None oidc_issuer: str | None = None oidc_resource_server_id: str | None = None + oidc_token_type: str = "Bearer" # NEXTCLOUD_OIDC_TOKEN_TYPE + oidc_scopes: str = "" # NEXTCLOUD_OIDC_SCOPES (space-separated) + port: int = 8000 # Server port (PORT); used to build fallback URLs # Nextcloud settings nextcloud_host: str | None = None @@ -698,6 +705,10 @@ class Settings: vector_sync_metrics_refresh_interval: int = 20 # seconds vector_sync_user_poll_interval: int = 60 # seconds - OAuth mode user discovery vector_sync_orphan_sweep_enabled: bool = True # card #101 + # Cadence for the background readiness dependency-health refresh loop + # (app.py): keeps the Nextcloud/Qdrant snapshot warm off the probe path so + # /health/ready never does external I/O (Deck #302). + health_ready_refresh_interval: int = 15 # seconds # System tag marking files for vector indexing. The scanner indexes files # carrying this tag and verify-on-read gates results on current membership # (ADR-019), so an untagged file drops out of search immediately. @@ -1336,6 +1347,9 @@ def get_settings() -> Settings: "oidc_client_secret": "NEXTCLOUD_OIDC_CLIENT_SECRET", "oidc_issuer": "OIDC_ISSUER", "oidc_resource_server_id": "OIDC_RESOURCE_SERVER_ID", + "oidc_token_type": "NEXTCLOUD_OIDC_TOKEN_TYPE", + "oidc_scopes": "NEXTCLOUD_OIDC_SCOPES", + "port": "PORT", # Nextcloud settings "nextcloud_host": "NEXTCLOUD_HOST", "nextcloud_username": "NEXTCLOUD_USERNAME", @@ -1376,6 +1390,7 @@ def get_settings() -> Settings: "vector_sync_metrics_refresh_interval": "VECTOR_SYNC_METRICS_REFRESH_INTERVAL", "vector_sync_user_poll_interval": "VECTOR_SYNC_USER_POLL_INTERVAL", "vector_sync_orphan_sweep_enabled": "VECTOR_SYNC_ORPHAN_SWEEP_ENABLED", + "health_ready_refresh_interval": "HEALTH_READY_REFRESH_INTERVAL", "vector_sync_pdf_tag": "VECTOR_SYNC_PDF_TAG", # Verify-on-read (ADR-019) "verification_concurrency": "VERIFICATION_CONCURRENCY", diff --git a/nextcloud_mcp_server/observability/readiness.py b/nextcloud_mcp_server/observability/readiness.py new file mode 100644 index 00000000..1306108d --- /dev/null +++ b/nextcloud_mcp_server/observability/readiness.py @@ -0,0 +1,75 @@ +"""Non-blocking readiness dependency-health cache. + +Kubernetes readiness probes must be cheap and must not gate a (typically +single-replica) tenant Pod out of its Service on transient external-dependency +latency. Doing so converts a *degraded* shared dependency (Nextcloud, Qdrant) +into a *total* outage: the only Pod is removed from the Service, the gateway +has no upstream, and connected MCP clients see their streamable-HTTP sessions +drop and fail to reconnect (Deck #302). + +A background loop refreshes this snapshot off the probe path; the readiness +handler only ever reads ``snapshot()`` (no I/O), so probe latency is decoupled +from upstream latency. Dependency results are reported for observability but are +intentionally *non-gating*. +""" + +from __future__ import annotations + +import time + +from pydantic import BaseModel, Field + + +class DependencyStatus(BaseModel): + """Last observed health of a single external dependency. + + ``healthy`` is ``None`` until the first check completes; ``detail`` carries + the human-readable string the readiness handler reports verbatim + (``"ok"`` / ``"embedded"`` / ``"pending"`` / ``"error: ..."``). + """ + + name: str + healthy: bool | None = None + detail: str = "pending" + checked_at: float = 0.0 + + +class ReadinessCache(BaseModel): + """Time-bounded snapshot of external dependency health. + + Written only by the background refresh loop and read only by the readiness + handler. No locking: assignment into ``statuses`` is atomic under the GIL, + and a reader tolerates seeing the previous value for one entry. + """ + + ttl_seconds: float = 30.0 + statuses: dict[str, DependencyStatus] = Field(default_factory=dict) + + def snapshot(self) -> dict[str, DependencyStatus]: + """Return a shallow copy of the current per-dependency statuses.""" + return dict(self.statuses) + + def update( + self, name: str, healthy: bool, detail: str, *, now: float | None = None + ) -> None: + """Record the outcome of a dependency check.""" + self.statuses[name] = DependencyStatus( + name=name, + healthy=healthy, + detail=detail, + checked_at=time.monotonic() if now is None else now, + ) + + def is_stale(self, *, now: float | None = None) -> bool: + """True when there is no data yet or any entry is older than the TTL. + + Exposed for observability/diagnostics; the refresh loop runs on a fixed + cadence rather than polling this. + """ + if not self.statuses: + return True + current = time.monotonic() if now is None else now + return any( + current - status.checked_at >= self.ttl_seconds + for status in self.statuses.values() + ) diff --git a/tests/unit/test_hybrid_auth_setup.py b/tests/unit/test_hybrid_auth_setup.py index 1b6652c8..d33e9731 100644 --- a/tests/unit/test_hybrid_auth_setup.py +++ b/tests/unit/test_hybrid_auth_setup.py @@ -123,14 +123,9 @@ class TestSetupOAuthConfigForMultiUserBasic: valid_fernet_key = Fernet.generate_key().decode() - # Mock TOKEN_ENCRYPTION_KEY environment variable - mocker.patch( - "os.getenv", - side_effect=lambda k, default=None: { - "TOKEN_ENCRYPTION_KEY": valid_fernet_key, - "NEXTCLOUD_MCP_SERVER_URL": "http://localhost:8000", - }.get(k, default), - ) + # Provide the encryption key via settings: the function reads from the + # injected Settings, not os.getenv, after the env->settings migration. + hybrid_auth_settings.token_encryption_key = valid_fernet_key # Mock httpx.AsyncClient mock_response = MagicMock() @@ -273,17 +268,12 @@ class TestSetupOAuthConfigForMultiUserBasic: self, hybrid_auth_settings, oidc_discovery_response, mocker ): """Test using custom OIDC discovery URL.""" - # Mock OIDC_DISCOVERY_URL environment variable + # Provide the custom discovery URL via settings: the function reads from + # the injected Settings, not os.getenv, after the env->settings migration. custom_discovery_url = ( "https://custom.idp.example.com/.well-known/openid-configuration" ) - mocker.patch( - "os.getenv", - side_effect=lambda k, default=None: { - "OIDC_DISCOVERY_URL": custom_discovery_url, - "NEXTCLOUD_MCP_SERVER_URL": "http://localhost:8000", - }.get(k, default), - ) + hybrid_auth_settings.oidc_discovery_url = custom_discovery_url # Mock httpx.AsyncClient mock_response = MagicMock() diff --git a/tests/unit/test_readiness_cache.py b/tests/unit/test_readiness_cache.py new file mode 100644 index 00000000..5394ac30 --- /dev/null +++ b/tests/unit/test_readiness_cache.py @@ -0,0 +1,64 @@ +"""Unit tests for the readiness dependency-health cache (Deck #302). + +The readiness probe reads this snapshot without performing any I/O; these tests +pin the small amount of logic it relies on (update/snapshot/staleness). +""" + +import pytest + +from nextcloud_mcp_server.observability.readiness import ( + DependencyStatus, + ReadinessCache, +) + +pytestmark = pytest.mark.unit + + +def test_dependency_status_defaults(): + status = DependencyStatus(name="nextcloud") + assert status.healthy is None # not yet checked + assert status.detail == "pending" + assert status.checked_at == 0.0 + + +def test_update_and_snapshot_round_trip(): + cache = ReadinessCache() + cache.update("nextcloud_reachable", True, "ok", now=100.0) + cache.update("qdrant", False, "error: status 503", now=100.0) + + snap = cache.snapshot() + assert snap["nextcloud_reachable"].healthy is True + assert snap["nextcloud_reachable"].detail == "ok" + assert snap["qdrant"].healthy is False + assert snap["qdrant"].detail == "error: status 503" + + +def test_snapshot_is_a_copy(): + cache = ReadinessCache() + cache.update("nextcloud_reachable", True, "ok", now=1.0) + snap = cache.snapshot() + # Mutating the returned mapping must not affect the cache. + snap.clear() + assert "nextcloud_reachable" in cache.snapshot() + + +def test_is_stale_when_empty(): + assert ReadinessCache(ttl_seconds=30.0).is_stale(now=0.0) is True + + +def test_is_stale_respects_ttl(): + cache = ReadinessCache(ttl_seconds=30.0) + cache.update("nextcloud_reachable", True, "ok", now=100.0) + + # Within the TTL window the snapshot is fresh. + assert cache.is_stale(now=120.0) is False + # At/after the TTL boundary it is stale. + assert cache.is_stale(now=130.0) is True + + +def test_is_stale_if_any_entry_is_old(): + cache = ReadinessCache(ttl_seconds=30.0) + cache.update("nextcloud_reachable", True, "ok", now=100.0) + cache.update("qdrant", True, "ok", now=200.0) + # nextcloud_reachable is well past the TTL even though qdrant is fresh. + assert cache.is_stale(now=205.0) is True