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) <noreply@anthropic.com>
This commit is contained in:
Chris Coutinho
2026-06-10 21:28:19 +02:00
co-authored by Claude Opus 4.8
parent 86ed15f466
commit 6ef7786cec
6 changed files with 420 additions and 227 deletions
+18
View File
@@ -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.
+242 -211
View File
@@ -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(
+15
View File
@@ -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",
@@ -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()
)
+6 -16
View File
@@ -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()
+64
View File
@@ -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