From 5e3c24fe94dde86e0743596d5700f0a621b6affe Mon Sep 17 00:00:00 2001 From: Nan Gao Date: Thu, 8 Oct 2026 06:56:33 +0800 Subject: [PATCH] fix(gateway): probe the Redis stream bridge and provisioner in /health/ready (#6447) * fix(gateway): probe the Redis stream bridge and provisioner in /health/ready GET /health/ready only probed the ORM engine and the LangGraph checkpointer. The Redis stream bridge client connects lazily and nothing ever pinged it, so a Gateway whose Redis was down started, answered 200 ready to the Kubernetes readinessProbe and Compose healthcheck, and kept receiving traffic while every run's first publish failed. Since #6328 the Redis bridge is mandatory for multi-instance deployments, so such a replica is not serviceable at all. - StreamBridge gains `async ping() -> bool`; the base returns True (no external backend), RedisStreamBridge round-trips PING and answers False on RedisError. - /health/ready adds a `stream_bridge` probe (not_configured for the memory bridge; unreachable flips the endpoint to 503; a missing bridge fails closed) and a report-only `provisioner` probe of `{sandbox.provisioner_url}/health` that never changes the status code, since every replica shares one provisioner. The provisioner URL is snapshotted on app.state at startup. - A probe that overruns the endpoint deadline is now the only one marked unreachable; finished verdicts are kept. - The public probe gate is one lock per probe kind, so one request's probes still run concurrently while a thundering herd opens at most one probe connection per kind per process. - Tests: bridge ping cases, stream-bridge/provisioner probe cases, verdict and deadline cases, and the first route-level /health/ready tests via TestClient. - Docs: gateway AGENTS.md, CONFIGURATION.md multi-instance notes, Helm README and probe comment, Compose healthcheck comment. * docs(changelog): note the readiness probe fix (#6447) * fix(gateway): apply the sandbox proxy policy to the provisioner probe Review follow-up for #6447. - The provisioner readiness probe built its httpx client with the default trust_env=True. httpx 0.28 treats CIDR entries in NO_PROXY as host patterns rather than subnet bypasses, so with HTTP_PROXY set and NO_PROXY=10.0.0.0/8 a private provisioner that the requests-based sandbox calls reach directly would have been probed through the proxy and reported unreachable on every check. The probe now reuses `sandbox_http_trust_env`: loopback, private, link-local and cluster-local addresses bypass the proxy, external hosts keep the environment's proxy settings. - Pin the liveness/readiness separation with a route test: `GET /health` stays 200 while the stream bridge and database are unreachable and `/health/ready` is 503, so a Redis blip can never become a restart loop. --- CHANGELOG.md | 15 + backend/app/gateway/AGENTS.md | 12 +- backend/app/gateway/app.py | 29 +- backend/app/gateway/deps.py | 15 +- backend/app/gateway/health.py | 278 +++++++--- backend/docs/CONFIGURATION.md | 1 + .../deerflow/runtime/stream_bridge/base.py | 12 + .../deerflow/runtime/stream_bridge/redis.py | 14 + backend/tests/test_gateway_health.py | 514 ++++++++++++++++-- backend/tests/test_stream_bridge.py | 44 ++ deploy/helm/deer-flow/README.md | 14 +- .../templates/gateway-deployment.yaml | 7 +- docker/docker-compose.yaml | 6 +- 13 files changed, 827 insertions(+), 134 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 3240e7539c..373e3b06cc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -804,6 +804,20 @@ This release closes that milestone with **439 merged pull requests**. ### Fixed +- **gateway:** `GET /health/ready` now reports unready while the Redis stream + bridge is unreachable. The bridge's Redis client connects lazily and nothing + pinged it, so a gateway whose Redis was down started, answered `200 ready` to + the Kubernetes readiness probe and the Compose healthcheck, and kept + receiving traffic while every run's first publish failed; with the + multi-instance gate making the Redis bridge mandatory, such a replica is not + serviceable at all. `StreamBridge` gains `ping()`, the response body gains a + `stream_bridge` verdict (`not_configured` for the memory bridge) that flips + the endpoint to 503 when unreachable, and a report-only `provisioner` verdict + from the sandbox provisioner's own `/health` when `sandbox.provisioner_url` + is set, which never changes the status code because every replica shares one + provisioner. A probe that overruns the endpoint deadline is now the only one + marked unreachable, and the public probe gate is one lock per probe kind. + ([#6447]) - **frontend:** Retrying a message after its attachment upload fails now keeps the context that was attached to it. The composer dropped its quotes, conversation references, staged project files and stored draft as soon as a @@ -9130,3 +9144,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6407]: https://github.com/bytedance/deer-flow/pull/6407 [#6412]: https://github.com/bytedance/deer-flow/pull/6412 [#6426]: https://github.com/bytedance/deer-flow/pull/6426 +[#6447]: https://github.com/bytedance/deer-flow/pull/6447 diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index c557b35134..fd96971e55 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -44,12 +44,12 @@ locations and keeps `merge_run_context_overrides` clear. Admission and predicate. FastAPI listens on port 8001. `GET /health` is liveness; `GET /health/ready` -concurrently probes the ORM engine and effective LangGraph checkpointer/Store -under one deadline and a strict connection-opening gate. It reads the startup -config snapshot: legacy `checkpointer:` if set, else `database:`. Unreachable or -unresolved backends return 503; process-local backends such as `memory` report -`not_configured`. `GATEWAY_ENABLE_DOCS=false` disables `/docs`, `/redoc`, and -`/openapi.json`. +concurrently probes the ORM engine, the checkpointer/Store and the stream +bridge backend (`StreamBridge.ping()`) under one deadline and per-kind gates, +from startup snapshots. Unreachable or unresolved gating backends return 503; +`memory` reports `not_configured`; the provisioner `/health` verdict is +report-only. See `health.py`. `GATEWAY_ENABLE_DOCS=false` disables `/docs`, +`/redoc`, and `/openapi.json`. `build_run_config()` resolves the default LangGraph super-step budget from the hot-reloaded top-level `recursion_limit` setting. A valid request-level value diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 2ae871cf3c..07e5349853 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -19,7 +19,7 @@ from app.gateway.browser_capability import ensure_browser_runtime_available from app.gateway.config import get_gateway_config from app.gateway.csrf_middleware import CORS_EXPOSED_HEADERS, CSRFMiddleware, get_configured_cors_origins from app.gateway.deps import langgraph_runtime -from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, readiness_payload +from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, READINESS_PROVISIONER_URL_ATTR, readiness_payload from app.gateway.routers import ( agents, artifacts, @@ -1307,19 +1307,26 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for @app.get("/health/ready", tags=["health"]) async def readiness_check(request: Request, response: Response) -> dict[str, str]: - """Readiness endpoint: 200 when the persistence backends are reachable. + """Readiness endpoint: 200 when the backends behind agent runs are reachable. - Probes the ORM engine behind ``database:`` and the effective LangGraph + Probes the ORM engine behind ``database:``, the effective LangGraph checkpointer/Store backend (legacy ``checkpointer:`` section, otherwise - derived from ``database:``) concurrently beneath one bounded deadline. - The checkpointer config comes from the startup snapshot recorded by - ``langgraph_runtime`` (never hot-reloaded config), so orchestrators can - gate on the gateway actually being ready rather than merely alive. - Returns 503 with ``status: degraded`` when either probe fails or the - startup backend cannot be resolved. + derived from ``database:``) and the stream bridge's Redis backend + concurrently beneath one bounded deadline, and reports the sandbox + provisioner's own ``/health`` without gating on it. Targets come from + the startup snapshot recorded by ``langgraph_runtime`` (never + hot-reloaded config), so orchestrators can gate on the gateway actually + being ready rather than merely alive. Returns 503 with + ``status: degraded`` when a gating probe fails or a startup target + cannot be resolved; the provisioner verdict is informational because + every replica shares one provisioner. """ - checkpointer_config = getattr(request.app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, None) - status_code, payload = await readiness_payload(checkpointer_config) + state = request.app.state + status_code, payload = await readiness_payload( + getattr(state, READINESS_CHECKPOINTER_CONFIG_ATTR, None), + stream_bridge=getattr(state, "stream_bridge", None), + provisioner_url=getattr(state, READINESS_PROVISIONER_URL_ATTR, None), + ) response.status_code = status_code return payload diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index efaa8bc846..5521d32457 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -580,14 +580,17 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen app.state.checkpointer = await stack.enter_async_context(make_checkpointer(config)) app.state.store = await stack.enter_async_context(make_store(config)) - # Record the checkpointer/Store backend selected from this startup - # snapshot so GET /health/ready probes what the running process - # actually uses. These singletons are restart-required by design and - # are never rebuilt on config.yaml hot reload, so the probe must not - # re-resolve process-wide configuration per request. - from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, resolve_checkpointer_config + # Record the checkpointer/Store backend and the provisioner endpoint + # selected from this startup snapshot so GET /health/ready probes what + # the running process actually uses. These singletons are + # restart-required by design and are never rebuilt on config.yaml hot + # reload, so the probe must not re-resolve process-wide configuration + # per request. The stream bridge needs no snapshot: the probe pings the + # singleton stored on app.state.stream_bridge above. + from app.gateway.health import READINESS_CHECKPOINTER_CONFIG_ATTR, READINESS_PROVISIONER_URL_ATTR, resolve_checkpointer_config, resolve_provisioner_url setattr(app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, resolve_checkpointer_config(config)) + setattr(app.state, READINESS_PROVISIONER_URL_ATTR, resolve_provisioner_url(config)) # Initialize repositories — one get_session_factory() call for all. sf = get_session_factory() diff --git a/backend/app/gateway/health.py b/backend/app/gateway/health.py index 5c4f524b25..3e850b5102 100644 --- a/backend/app/gateway/health.py +++ b/backend/app/gateway/health.py @@ -1,29 +1,40 @@ """Readiness probe helpers for the gateway health endpoints. ``GET /health`` stays a pure liveness signal: 200 whenever the process is up. -``GET /health/ready`` additionally probes the persistence the gateway actually +``GET /health/ready`` additionally probes the backends the gateway actually depends on, so orchestrators (Docker healthchecks, Kubernetes probes) treat the -gateway as ready only when the databases behind agent runs are reachable. Two -backends can be configured independently: +gateway as ready only when agent runs can be persisted and streamed. Three +probes gate the verdict and can be configured independently: -* the ORM engine behind ``database:`` (application repositories), and +* the ORM engine behind ``database:`` (application repositories), * the effective LangGraph checkpointer/Store backend - the legacy ``checkpointer:`` section when present, otherwise derived from ``database:`` - (memory/sqlite/postgres). + (memory/sqlite/postgres), and +* the stream bridge's external backend (Redis for ``stream_bridge.type: + redis``). Its client connects lazily, so without this ping a gateway whose + Redis is down starts, reports ready and then fails every run's publish. -Both probes run concurrently beneath a single endpoint-wide deadline +A fourth probe is report-only: when ``sandbox.provisioner_url`` is set, the +provisioner's own unauthenticated ``GET /health`` is checked and its verdict +is included in the body, but it never flips the endpoint to 503. Every +replica shares one provisioner, so a blip there must not drain the service. + +All probes run concurrently beneath a single endpoint-wide deadline (:data:`_READINESS_DEADLINE_SECONDS`), so normal probe work completes within -one probe window rather than the sum of both budgets. Connection teardown is -ownership-critical and is drained to completion after cancellation, so a -stalled close may outlive the probe/deadline budget. The checkpointer config -is resolved once at startup from the same snapshot ``langgraph_runtime`` builds -its resources from and is stored on ``app.state``; probing a hot-reloaded +one probe window rather than the sum of the budgets. A probe that overruns the +deadline fails alone; verdicts already collected are kept. Connection teardown +is ownership-critical and is drained to completion after cancellation, so a +stalled close may outlive the probe/deadline budget. Probe targets are +resolved once at startup from the same snapshot ``langgraph_runtime`` builds +its resources from and are stored on ``app.state``; probing a hot-reloaded config instead could check a backend the running process is not using. A -``backend=memory`` deployment has nothing to probe and is always considered -ready; a startup config that cannot be resolved fails closed as unreachable. -Connection-opening probes are serialized behind a strict per-process gate: the -route is public through the ``/health`` auth prefix, so unlimited concurrent -requests must never translate into unlimited new PostgreSQL connections. +process-local backend (``memory``) has nothing to probe and reports +``not_configured``; a startup target that cannot be resolved fails closed as +unreachable. Connection-opening probes are serialized behind strict +per-process gates, one per probe kind: the route is public through the +``/health`` auth prefix, so unlimited concurrent requests must never translate +into unlimited new PostgreSQL connections, Redis pool growth or a connection +flood at the provisioner. """ from __future__ import annotations @@ -32,8 +43,10 @@ import asyncio import logging import pathlib import weakref +from collections.abc import Awaitable from typing import TYPE_CHECKING +import httpx from sqlalchemy import text from deerflow.persistence.engine import get_engine @@ -42,6 +55,7 @@ from deerflow.utils.file_io import await_drained if TYPE_CHECKING: from deerflow.config.app_config import AppConfig from deerflow.config.checkpointer_config import CheckpointerConfig + from deerflow.runtime.stream_bridge import StreamBridge logger = logging.getLogger(__name__) @@ -64,46 +78,64 @@ _READINESS_DEADLINE_SECONDS = 3.0 # records the startup-bound checkpointer/Store config the probe targets. READINESS_CHECKPOINTER_CONFIG_ATTR = "checkpointer_config" -# One gate per running event loop (one per worker process in production; one -# per test loop in the suite). ``/health/ready`` is public and unauthenticated, -# so a thundering herd of probes - or an attacker - must never be able to open -# an unbounded number of new connections: every connection-opening probe below -# is serialized through this gate, bounding in-flight probe connections to one -# per process. Waiting requests are still shed by the endpoint-wide deadline. -_PROBE_GATES: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, asyncio.Lock] = weakref.WeakKeyDictionary() +# ``app.state`` attribute under which :func:`app.gateway.deps.langgraph_runtime` +# records the startup-bound provisioner base URL (``None`` when unset). The +# stream bridge needs no snapshot: the singleton itself lives on +# ``app.state.stream_bridge``. +READINESS_PROVISIONER_URL_ATTR = "readiness_provisioner_url" + +# Probes whose ``unreachable`` verdict flips the endpoint to 503. The +# provisioner is deliberately absent: it is shared by every replica, so its +# verdict is informational and must never drain the whole service. +_GATING_PROBES = ("database", "checkpointer", "stream_bridge") + +# One set of gates per running event loop (one per worker process in +# production; one per test loop in the suite). ``/health/ready`` is public and +# unauthenticated, so a thundering herd of probes - or an attacker - must never +# be able to open an unbounded number of new connections: every +# connection-opening probe below is serialized through the gate for its kind, +# bounding in-flight probe connections to one per kind per process. Gates are +# per kind rather than shared so the probes of one request still run +# concurrently and the endpoint deadline keeps holding for a healthy response. +# Waiting requests are still shed by the endpoint-wide deadline. +_PROBE_GATES: weakref.WeakKeyDictionary[asyncio.AbstractEventLoop, dict[str, asyncio.Lock]] = weakref.WeakKeyDictionary() -def _probe_gate() -> asyncio.Lock: - """Return the serialization gate bound to the currently running loop.""" +def _probe_gate(kind: str) -> asyncio.Lock: + """Return the serialization gate for *kind* bound to the running loop.""" loop = asyncio.get_running_loop() - gate = _PROBE_GATES.get(loop) + gates = _PROBE_GATES.get(loop) + if gates is None: + gates = {} + _PROBE_GATES[loop] = gates + gate = gates.get(kind) if gate is None: gate = asyncio.Lock() - _PROBE_GATES[loop] = gate + gates[kind] = gate return gate -# Result vocabulary for the database probe. -DATABASE_OK = "ok" -DATABASE_NOT_CONFIGURED = "not_configured" -DATABASE_UNREACHABLE = "unreachable" +# Result vocabulary shared by every probe. +PROBE_OK = "ok" +PROBE_NOT_CONFIGURED = "not_configured" +PROBE_UNREACHABLE = "unreachable" async def check_database_health() -> str: - """Probe the persistence engine; return one of the DATABASE_* values.""" + """Probe the persistence engine; return one of the PROBE_* values.""" engine = get_engine() if engine is None: # backend=memory, or the engine has not been initialized yet: there is # no database to probe. - return DATABASE_NOT_CONFIGURED + return PROBE_NOT_CONFIGURED try: async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS): async with engine.connect() as connection: await connection.execute(text("SELECT 1")) except Exception: logger.warning("Readiness database probe failed", exc_info=True) - return DATABASE_UNREACHABLE - return DATABASE_OK + return PROBE_UNREACHABLE + return PROBE_OK def resolve_checkpointer_config(startup_config: AppConfig) -> CheckpointerConfig | None: @@ -156,16 +188,16 @@ async def _probe_sqlite_backend(conn_string: str | None) -> str: import aiosqlite except ImportError: logger.error("Readiness probe: aiosqlite is not installed for the sqlite checkpointer backend") - return DATABASE_UNREACHABLE + return PROBE_UNREACHABLE from deerflow.runtime.store._sqlite_utils import resolve_sqlite_conn_str try: conn_str = resolve_sqlite_conn_str(conn_string or "store.db") except ValueError as exc: logger.error("Readiness probe: %s", exc) - return DATABASE_UNREACHABLE + return PROBE_UNREACHABLE if conn_str == ":memory:": - return DATABASE_NOT_CONFIGURED + return PROBE_NOT_CONFIGURED try: async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS): connection = await aiosqlite.connect(_sqlite_disk_uri(conn_str), uri=True) @@ -175,8 +207,8 @@ async def _probe_sqlite_backend(conn_string: str | None) -> str: await await_drained(connection.close()) except Exception: logger.warning("Readiness sqlite checkpointer probe failed", exc_info=True) - return DATABASE_UNREACHABLE - return DATABASE_OK + return PROBE_UNREACHABLE + return PROBE_OK async def _probe_postgres_backend(conn_string: str, schema: str) -> str: @@ -185,7 +217,7 @@ async def _probe_postgres_backend(conn_string: str, schema: str) -> str: from psycopg import AsyncConnection except ImportError: logger.error("Readiness probe: psycopg is not installed for the postgres checkpointer backend") - return DATABASE_UNREACHABLE + return PROBE_UNREACHABLE try: from deerflow.persistence.postgres_schema import dsn_with_search_path, normalize_libpq_dsn @@ -199,8 +231,8 @@ async def _probe_postgres_backend(conn_string: str, schema: str) -> str: await await_drained(connection.close()) except Exception: logger.warning("Readiness postgres checkpointer probe failed", exc_info=True) - return DATABASE_UNREACHABLE - return DATABASE_OK + return PROBE_UNREACHABLE + return PROBE_OK async def _probe_checkpointer_backend(config: CheckpointerConfig) -> str: @@ -213,33 +245,132 @@ async def _probe_checkpointer_backend(config: CheckpointerConfig) -> str: """ if config.type == "memory": # In-process backend: there is nothing external to probe. - return DATABASE_NOT_CONFIGURED + return PROBE_NOT_CONFIGURED if config.type not in ("sqlite", "postgres"): logger.warning("Readiness probe: unknown checkpointer backend %r", config.type) - return DATABASE_UNREACHABLE - async with _probe_gate(): + return PROBE_UNREACHABLE + async with _probe_gate("checkpointer"): if config.type == "sqlite": return await _probe_sqlite_backend(config.connection_string) if not config.connection_string: - return DATABASE_UNREACHABLE + return PROBE_UNREACHABLE return await _probe_postgres_backend(config.connection_string, config.postgres_schema) -async def readiness_payload(checkpointer_config: CheckpointerConfig | None = None) -> tuple[int, dict[str, str]]: +async def _probe_stream_bridge(bridge: StreamBridge | None) -> str: + """Probe the stream bridge's external backend (Redis for the redis bridge). + + ``None`` fails closed: without a bridge no run can publish, stream or be + cancelled across instances, so readiness must not be claimed. A + process-local bridge (``supports_cross_process`` False) has nothing + external to probe. Pings are serialized so unauthenticated probe traffic + cannot grow the Redis connection pool; the bridge answers ``False`` for a + backend-side failure and anything it raises counts the same way. + """ + if bridge is None: + logger.error("Readiness probe: no stream bridge recorded on app.state; failing closed") + return PROBE_UNREACHABLE + if not bridge.supports_cross_process: + return PROBE_NOT_CONFIGURED + async with _probe_gate("stream_bridge"): + try: + async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS): + reachable = await bridge.ping() + except Exception: + logger.warning("Readiness stream bridge probe failed", exc_info=True) + return PROBE_UNREACHABLE + if not reachable: + logger.warning("Readiness stream bridge probe: backend reported unreachable") + return PROBE_UNREACHABLE + return PROBE_OK + + +def resolve_provisioner_url(startup_config: AppConfig) -> str | None: + """Return the provisioner base URL from the startup snapshot, or ``None``. + + ``sandbox`` is a startup-only section (the provider singleton is built + once), so the probe target is pinned at startup like the checkpointer + snapshot. ``provisioner_url`` is an ``extra="allow"`` key on + ``SandboxConfig`` that only ``AioSandboxProvider`` consumes, not a declared + field, hence the ``getattr`` access. Blank values mean "no provisioner". + """ + sandbox = getattr(startup_config, "sandbox", None) + url = getattr(sandbox, "provisioner_url", None) if sandbox is not None else None + if not isinstance(url, str): + return None + normalized = url.strip().rstrip("/") + return normalized or None + + +def _make_provisioner_client(provisioner_url: str) -> httpx.AsyncClient: + """Build the short-lived client for one provisioner probe (replaced in tests). + + Proxy handling mirrors the provisioner-bound sandbox clients: a loopback, + private, link-local or cluster-local address (``provisioner``, + ``*.docker.internal``) is control-plane traffic and bypasses ``HTTP_PROXY``, + while an external host keeps the environment's proxy settings. Unlike the + requests-based sandbox calls, httpx does not honour CIDR entries in + ``NO_PROXY``, so without this a private provisioner behind a proxy-only + egress would be reported unreachable on every probe. + """ + # Lazy: importing the aio_sandbox package pulls in the whole provider stack. + from deerflow.community.aio_sandbox.backend import sandbox_http_trust_env + + return httpx.AsyncClient(timeout=_PROBE_TIMEOUT_SECONDS, trust_env=sandbox_http_trust_env(provisioner_url)) + + +async def _probe_provisioner(provisioner_url: str | None) -> str: + """Probe the sandbox provisioner's unauthenticated ``GET /health``. + + Report-only: :func:`readiness_payload` never degrades on this verdict, + because every gateway replica shares one provisioner and a blip there + must not drain the whole service. The probe is still serialized per + process so public probe traffic cannot become a connection flood at the + provisioner, and the per-probe budget bounds a hung connection. + """ + if not provisioner_url: + return PROBE_NOT_CONFIGURED + async with _probe_gate("provisioner"): + try: + async with asyncio.timeout(_PROBE_TIMEOUT_SECONDS): + async with _make_provisioner_client(provisioner_url) as client: + response = await client.get(f"{provisioner_url}/health") + except Exception: + logger.warning("Readiness provisioner probe failed", exc_info=True) + return PROBE_UNREACHABLE + if response.status_code != 200: + logger.warning("Readiness provisioner probe: HTTP %s from %s/health", response.status_code, provisioner_url) + return PROBE_UNREACHABLE + return PROBE_OK + + +async def readiness_payload( + checkpointer_config: CheckpointerConfig | None = None, + *, + stream_bridge: StreamBridge | None = None, + provisioner_url: str | None = None, +) -> tuple[int, dict[str, str]]: """Return the (status_code, body) pair served by ``GET /health/ready``. - Probes both persistence halves the gateway depends on: the ORM engine - behind ``database:`` (repositories) and the effective LangGraph - checkpointer/Store backend (the legacy ``checkpointer:`` section, otherwise - derived from ``database:``). The probes run concurrently beneath one + Probes the backends the gateway depends on: the ORM engine behind + ``database:`` (repositories), the effective LangGraph checkpointer/Store + backend (the legacy ``checkpointer:`` section, otherwise derived from + ``database:``) and the stream bridge's external backend, plus the + report-only provisioner check. The probes run concurrently beneath one endpoint-wide deadline, so normal probe work is bounded by the slowest - single probe rather than their sum. Connection teardown is drained after - cancellation to preserve ownership and can therefore extend the in-flight - request beyond that deadline if close itself stalls. ``checkpointer_config`` is the startup - snapshot recorded by ``langgraph_runtime``; None means no snapshot could be - resolved, which fails closed as an unreachable backend rather than - reporting ready. Either backend can be configured independently of the - other, so an unreachable probe on either degrades the endpoint. + single probe rather than their sum; a probe that overruns the deadline is + the only one marked unreachable, so a wedged report-only probe cannot + flip the endpoint through the deadline path. Connection teardown is + drained after cancellation to preserve ownership and can therefore extend + the in-flight request beyond that deadline if close itself stalls. + + ``checkpointer_config`` and ``provisioner_url`` are the startup snapshots + recorded by ``langgraph_runtime`` and ``stream_bridge`` is the singleton + it built; a ``None`` checkpointer snapshot or bridge fails closed as + unreachable rather than reporting ready, while a ``None`` provisioner URL + simply means none is configured. Each gating backend can be configured + independently of the others, so an unreachable verdict on any of them + degrades the endpoint. """ async def _probe_engine() -> str: @@ -250,23 +381,38 @@ async def readiness_payload(checkpointer_config: CheckpointerConfig | None = Non # Fail closed: without the startup-bound config we cannot know what # backend agent runs use, so readiness must not be claimed. logger.error("Readiness probe: no startup checkpointer config snapshot recorded; failing closed") - return DATABASE_UNREACHABLE + return PROBE_UNREACHABLE return await _probe_checkpointer_backend(checkpointer_config) + probes = { + "database": _probe_engine(), + "checkpointer": _probe_checkpointer(), + "stream_bridge": _probe_stream_bridge(stream_bridge), + "provisioner": _probe_provisioner(provisioner_url), + } + # Every verdict starts out unreachable and is overwritten as its probe + # finishes, so a probe cancelled by the deadline fails alone instead of + # erasing the answers the others already produced. + results = dict.fromkeys(probes, PROBE_UNREACHABLE) + pending = set(probes) + + async def _record(name: str, probe: Awaitable[str]) -> None: + results[name] = await probe + pending.discard(name) + try: async with asyncio.timeout(_READINESS_DEADLINE_SECONDS): - database, checkpointer = await asyncio.gather(_probe_engine(), _probe_checkpointer()) + await asyncio.gather(*(_record(name, probe) for name, probe in probes.items())) except TimeoutError: logger.error( - "Readiness probes exceeded the %.1fs endpoint deadline", + "Readiness probes exceeded the %.1fs endpoint deadline; failing %s", _READINESS_DEADLINE_SECONDS, + ", ".join(sorted(pending)), ) - database = checkpointer = DATABASE_UNREACHABLE - degraded = DATABASE_UNREACHABLE in (database, checkpointer) + degraded = any(results[name] == PROBE_UNREACHABLE for name in _GATING_PROBES) payload = { "status": "degraded" if degraded else "ready", "service": "deer-flow-gateway", - "database": database, - "checkpointer": checkpointer, + **results, } return (503 if degraded else 200, payload) diff --git a/backend/docs/CONFIGURATION.md b/backend/docs/CONFIGURATION.md index 23ff14880b..cc63b9d302 100644 --- a/backend/docs/CONFIGURATION.md +++ b/backend/docs/CONFIGURATION.md @@ -580,6 +580,7 @@ Notes: - `GATEWAY_WORKERS` / `WEB_CONCURRENCY` only count the uvicorn workers of one process tree. A Kubernetes Deployment with `replicas > 1` runs one worker per Pod, so every Pod reports a single worker and the multi-worker startup gate stays inert — while each Pod's startup orphan reconciliation still writes the other Pods' lease-less runs off as crashed on every rolling update. - Set `multi_instance: true` (or export `DEER_FLOW_MULTI_INSTANCE=1`, which lets deploy tooling such as a Helm chart set it from its replica count) on every instance that shares one database. Startup then enforces the same prerequisites as `GATEWAY_WORKERS > 1`: `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true`, and a Redis stream bridge (`stream_bridge.type: redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL`). It refuses an explicit `sandbox.ownership.type: memory`, process-local browser tools, and `scheduler.enabled: true` without `scheduler.multi_instance: true`. - `DEER_FLOW_MULTI_INSTANCE` treats blank, `0`, `false`, `no` and `off` as "not declared"; any other value declares a multi-instance deployment, so a typo fails closed. +- `GET /health/ready` pings the Redis stream bridge on every probe and answers 503 (`stream_bridge: unreachable`) while Redis is down, so the orchestrator drains that instance instead of routing it runs it cannot publish or stream; the memory bridge reports `not_configured`. When `sandbox.provisioner_url` is set, the body also carries the provisioner's `/health` verdict (`provisioner: ok|unreachable`), which never changes the status code because every instance shares one provisioner. That probe follows the sandbox clients' proxy policy: loopback, private, link-local and cluster-local provisioner addresses bypass `HTTP_PROXY`, external hosts keep the environment's proxy settings. - The declaration also drives the `agent_storage.backend: file` divergence warning, the inbound webhook dedupe warning, and the WeChat QR-login guard, which otherwise only look at the worker count. - Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it. diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py b/backend/packages/harness/deerflow/runtime/stream_bridge/base.py index 446fec87ef..e616442eed 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/base.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/base.py @@ -111,5 +111,17 @@ class StreamBridge(abc.ABC): giving late subscribers a chance to drain remaining events. """ + async def ping(self) -> bool: + """Return whether the bridge's external backend is reachable right now. + + Used by the gateway readiness probe. The default is for bridges with + no external backend: the in-process memory bridge has nothing to + probe and is always reachable. A bridge that sets + :attr:`supports_cross_process` talks to a shared backend and should + override this with a bounded round trip, answering ``False`` (not + raising) when the backend is down. + """ + return True + async def close(self) -> None: """Release backend resources. Default is a no-op.""" diff --git a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py index 64470c112a..011509fc2e 100644 --- a/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py +++ b/backend/packages/harness/deerflow/runtime/stream_bridge/redis.py @@ -190,6 +190,20 @@ class RedisStreamBridge(StreamBridge): """Return whether Redis still has retained stream data for *run_id*.""" return bool(await self._redis.exists(self._stream_key(run_id))) + async def ping(self) -> bool: + """Round-trip ``PING`` to the Redis backend for the readiness probe. + + The client connects lazily, so until something publishes nothing has + verified that Redis is reachable; this is the only call that does. + Redis-side failures answer ``False`` so the probe gets a verdict rather + than a stack of client internals. Callers bound the wait themselves. + """ + try: + return bool(await self._redis.ping()) + except RedisError: + logger.warning("Redis stream bridge ping failed", exc_info=True) + return False + async def _resolve_start_stream_id(self, key: str, last_event_id: str | None) -> str: if last_event_id is None: return "0-0" diff --git a/backend/tests/test_gateway_health.py b/backend/tests/test_gateway_health.py index 9c8b3d4d67..5923911281 100644 --- a/backend/tests/test_gateway_health.py +++ b/backend/tests/test_gateway_health.py @@ -6,20 +6,29 @@ import sqlite3 import sys import time from contextlib import asynccontextmanager +from types import SimpleNamespace +import httpx import pytest +from fastapi.testclient import TestClient import app.gateway.health as health_module from app.gateway.health import ( - DATABASE_NOT_CONFIGURED, - DATABASE_OK, - DATABASE_UNREACHABLE, + PROBE_NOT_CONFIGURED, + PROBE_OK, + PROBE_UNREACHABLE, + READINESS_CHECKPOINTER_CONFIG_ATTR, + READINESS_PROVISIONER_URL_ATTR, _probe_checkpointer_backend, + _probe_provisioner, + _probe_stream_bridge, check_database_health, readiness_payload, resolve_checkpointer_config, + resolve_provisioner_url, ) from deerflow.config.checkpointer_config import CheckpointerConfig +from deerflow.runtime import MemoryStreamBridge class _FakeConnection: @@ -47,50 +56,84 @@ def _create_sqlite_file(path: pathlib.Path) -> None: sqlite3.connect(str(path)).close() +class _FakeCrossProcessBridge(MemoryStreamBridge): + """Bridge with an external backend whose ``ping`` answer is scripted.""" + + supports_cross_process = True + + def __init__(self, *, result: bool = True, error: Exception | None = None, delay: float = 0.0) -> None: + super().__init__() + self._result = result + self._error = error + self._delay = delay + self.ping_calls = 0 + + async def ping(self) -> bool: + self.ping_calls += 1 + if self._delay: + await asyncio.sleep(self._delay) + if self._error is not None: + raise self._error + return self._result + + +def _provisioner_client_factory(handler): + """Return a ``_make_provisioner_client`` replacement backed by ``httpx.MockTransport``.""" + + def _factory(provisioner_url: str) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + return _factory + + +def _install_provisioner(monkeypatch, handler) -> None: + monkeypatch.setattr(health_module, "_make_provisioner_client", _provisioner_client_factory(handler)) + + @pytest.mark.anyio async def test_check_database_health_without_engine(monkeypatch): """backend=memory (no engine) must report not_configured, never unreachable.""" monkeypatch.setattr("app.gateway.health.get_engine", lambda: None) - assert await check_database_health() == DATABASE_NOT_CONFIGURED + assert await check_database_health() == PROBE_NOT_CONFIGURED @pytest.mark.anyio async def test_check_database_health_reachable(monkeypatch): monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) - assert await check_database_health() == DATABASE_OK + assert await check_database_health() == PROBE_OK @pytest.mark.anyio async def test_check_database_health_unreachable(monkeypatch): monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True)) - assert await check_database_health() == DATABASE_UNREACHABLE + assert await check_database_health() == PROBE_UNREACHABLE @pytest.mark.anyio async def test_readiness_payload_ready_when_database_ok_and_memory_backend(monkeypatch): monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) - status_code, payload = await readiness_payload(CheckpointerConfig(type="memory")) + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge()) assert status_code == 200 assert payload["status"] == "ready" - assert payload["database"] == DATABASE_OK - assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_NOT_CONFIGURED @pytest.mark.anyio async def test_readiness_payload_degraded_when_database_unreachable(monkeypatch): monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True)) - status_code, payload = await readiness_payload(CheckpointerConfig(type="memory")) + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge()) assert status_code == 503 assert payload["status"] == "degraded" - assert payload["database"] == DATABASE_UNREACHABLE - assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED + assert payload["database"] == PROBE_UNREACHABLE + assert payload["checkpointer"] == PROBE_NOT_CONFIGURED @pytest.mark.anyio @@ -98,12 +141,12 @@ async def test_readiness_payload_ready_when_nothing_configured(monkeypatch): """backend=memory end to end must stay ready with not_configured results.""" monkeypatch.setattr("app.gateway.health.get_engine", lambda: None) - status_code, payload = await readiness_payload(CheckpointerConfig(type="memory")) + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge()) assert status_code == 200 assert payload["status"] == "ready" - assert payload["database"] == DATABASE_NOT_CONFIGURED - assert payload["checkpointer"] == DATABASE_NOT_CONFIGURED + assert payload["database"] == PROBE_NOT_CONFIGURED + assert payload["checkpointer"] == PROBE_NOT_CONFIGURED @pytest.mark.anyio @@ -112,12 +155,12 @@ async def test_readiness_payload_degraded_when_checkpointer_unreachable_but_data monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) config = CheckpointerConfig(type="sqlite", connection_string=str(tmp_path / "missing" / "checkpoints.db")) - status_code, payload = await readiness_payload(config) + status_code, payload = await readiness_payload(config, stream_bridge=MemoryStreamBridge()) assert status_code == 503 assert payload["status"] == "degraded" - assert payload["database"] == DATABASE_OK - assert payload["checkpointer"] == DATABASE_UNREACHABLE + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_UNREACHABLE @pytest.mark.anyio @@ -125,12 +168,12 @@ async def test_readiness_payload_fails_closed_without_startup_snapshot(monkeypat """No startup config snapshot must degrade readiness, never report ready.""" monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) - status_code, payload = await readiness_payload(None) + status_code, payload = await readiness_payload(None, stream_bridge=MemoryStreamBridge()) assert status_code == 503 assert payload["status"] == "degraded" - assert payload["database"] == DATABASE_OK - assert payload["checkpointer"] == DATABASE_UNREACHABLE + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_UNREACHABLE @pytest.mark.anyio @@ -139,19 +182,26 @@ async def test_readiness_probes_run_concurrently(monkeypatch): async def _slow_ok(*args) -> str: await asyncio.sleep(0.35) - return DATABASE_OK + return PROBE_OK monkeypatch.setattr(health_module, "check_database_health", _slow_ok) monkeypatch.setattr(health_module, "_probe_checkpointer_backend", _slow_ok) + monkeypatch.setattr(health_module, "_probe_provisioner", _slow_ok) started = time.perf_counter() - status_code, payload = await readiness_payload(CheckpointerConfig(type="memory")) + status_code, payload = await readiness_payload( + CheckpointerConfig(type="memory"), + stream_bridge=_FakeCrossProcessBridge(delay=0.35), + provisioner_url="http://provisioner:8002", + ) elapsed = time.perf_counter() - started assert status_code == 200 - assert payload["database"] == DATABASE_OK - assert payload["checkpointer"] == DATABASE_OK - # Two sequential 0.35s probes would take ~0.7s; concurrent ones finish + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_OK + assert payload["stream_bridge"] == PROBE_OK + assert payload["provisioner"] == PROBE_OK + # Four sequential 0.35s probes would take ~1.4s; concurrent ones finish # within a single probe window. assert elapsed < 0.6 @@ -162,18 +212,18 @@ async def test_readiness_payload_enforces_endpoint_deadline(monkeypatch): async def _hanging(*args) -> str: await asyncio.sleep(30) - return DATABASE_OK + return PROBE_OK monkeypatch.setattr(health_module, "check_database_health", _hanging) monkeypatch.setattr(health_module, "_probe_checkpointer_backend", _hanging) monkeypatch.setattr(health_module, "_READINESS_DEADLINE_SECONDS", 0.05) - status_code, payload = await readiness_payload(CheckpointerConfig(type="memory")) + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=MemoryStreamBridge()) assert status_code == 503 assert payload["status"] == "degraded" - assert payload["database"] == DATABASE_UNREACHABLE - assert payload["checkpointer"] == DATABASE_UNREACHABLE + assert payload["database"] == PROBE_UNREACHABLE + assert payload["checkpointer"] == PROBE_UNREACHABLE @pytest.mark.anyio @@ -194,7 +244,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec max_active = max(max_active, active) try: await asyncio.sleep(0.05) - return DATABASE_OK + return PROBE_OK finally: active -= 1 @@ -202,7 +252,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec monkeypatch.setattr("app.gateway.health.get_engine", lambda: None) config = CheckpointerConfig(type="sqlite", connection_string=":memory:") - results = await asyncio.gather(*(readiness_payload(config) for _ in range(8))) + results = await asyncio.gather(*(readiness_payload(config, stream_bridge=MemoryStreamBridge()) for _ in range(8))) assert [status_code for status_code, _ in results] == [200] * 8 assert max_active == 1 @@ -210,7 +260,7 @@ async def test_concurrent_readiness_requests_do_not_open_concurrent_probe_connec @pytest.mark.anyio async def test_probe_checkpointer_memory_reports_not_configured(): - assert await _probe_checkpointer_backend(CheckpointerConfig(type="memory")) == DATABASE_NOT_CONFIGURED + assert await _probe_checkpointer_backend(CheckpointerConfig(type="memory")) == PROBE_NOT_CONFIGURED @pytest.mark.anyio @@ -220,7 +270,7 @@ async def test_probe_checkpointer_sqlite_reachable(tmp_path): result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(db_path))) - assert result == DATABASE_OK + assert result == PROBE_OK @pytest.mark.anyio @@ -235,7 +285,7 @@ async def test_probe_checkpointer_sqlite_uri_is_unreachable(conn_string, tmp_pat result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=conn_string)) - assert result == DATABASE_UNREACHABLE + assert result == PROBE_UNREACHABLE assert [path.name for path in tmp_path.iterdir()] == ["checkpoints.db"] @@ -246,7 +296,7 @@ async def test_probe_checkpointer_sqlite_missing_file_stays_missing_and_unreacha result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(missing))) - assert result == DATABASE_UNREACHABLE + assert result == PROBE_UNREACHABLE assert not missing.exists() @@ -256,7 +306,7 @@ async def test_probe_checkpointer_sqlite_unreachable_when_parent_missing(tmp_pat result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=str(missing_parent))) - assert result == DATABASE_UNREACHABLE + assert result == PROBE_UNREACHABLE @pytest.mark.anyio @@ -264,7 +314,7 @@ async def test_probe_checkpointer_sqlite_in_memory_is_not_configured(): """In-memory SQLite has no external state, mirroring the memory backend.""" result = await _probe_checkpointer_backend(CheckpointerConfig(type="sqlite", connection_string=":memory:")) - assert result == DATABASE_NOT_CONFIGURED + assert result == PROBE_NOT_CONFIGURED @pytest.mark.anyio @@ -278,7 +328,7 @@ async def test_probe_checkpointer_postgres_without_psycopg_is_unreachable(monkey ) ) - assert result == DATABASE_UNREACHABLE + assert result == PROBE_UNREACHABLE def test_resolve_checkpointer_config_passes_through_resolution(monkeypatch): @@ -399,3 +449,389 @@ async def test_postgres_probe_drains_connection_close_across_repeated_cancellati connection, "PostgreSQL", ) + + +# --------------------------------------------------------------------------- +# Stream bridge probe +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_probe_stream_bridge_memory_reports_not_configured(): + """The process-local memory bridge has no external backend to probe.""" + bridge = MemoryStreamBridge() + + assert await _probe_stream_bridge(bridge) == PROBE_NOT_CONFIGURED + + +@pytest.mark.anyio +async def test_probe_stream_bridge_reachable_backend_is_ok(): + bridge = _FakeCrossProcessBridge(result=True) + + assert await _probe_stream_bridge(bridge) == PROBE_OK + assert bridge.ping_calls == 1 + + +@pytest.mark.anyio +async def test_probe_stream_bridge_false_ping_is_unreachable(): + assert await _probe_stream_bridge(_FakeCrossProcessBridge(result=False)) == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_probe_stream_bridge_raising_ping_is_unreachable(): + assert await _probe_stream_bridge(_FakeCrossProcessBridge(error=RuntimeError("redis exploded"))) == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_probe_stream_bridge_enforces_probe_timeout(monkeypatch): + """A ping that never answers must be bounded by the per-probe budget.""" + monkeypatch.setattr(health_module, "_PROBE_TIMEOUT_SECONDS", 0.05) + + started = time.perf_counter() + result = await _probe_stream_bridge(_FakeCrossProcessBridge(delay=30)) + elapsed = time.perf_counter() - started + + assert result == PROBE_UNREACHABLE + assert elapsed < 1.0 + + +@pytest.mark.anyio +async def test_probe_stream_bridge_fails_closed_without_bridge(): + """No bridge on app.state means runs cannot stream: never report ready.""" + assert await _probe_stream_bridge(None) == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_concurrent_stream_bridge_probes_are_serialized(): + """Public /health/ready must not fan a thundering herd out onto the Redis pool.""" + active = 0 + max_active = 0 + + class _TrackingBridge(_FakeCrossProcessBridge): + async def ping(self) -> bool: + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + try: + await asyncio.sleep(0.02) + return True + finally: + active -= 1 + + bridge = _TrackingBridge() + results = await asyncio.gather(*(_probe_stream_bridge(bridge) for _ in range(6))) + + assert results == [PROBE_OK] * 6 + assert max_active == 1 + + +# --------------------------------------------------------------------------- +# Provisioner probe +# --------------------------------------------------------------------------- + + +def test_resolve_provisioner_url_normalises_configured_url(): + config = SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=" http://provisioner:8002/ ")) + + assert resolve_provisioner_url(config) == "http://provisioner:8002" + + +@pytest.mark.parametrize( + "config", + [ + SimpleNamespace(sandbox=SimpleNamespace()), + SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=None)), + SimpleNamespace(sandbox=SimpleNamespace(provisioner_url=" ")), + SimpleNamespace(sandbox=None), + SimpleNamespace(), + ], + ids=["no-field", "none", "blank", "no-sandbox", "no-sandbox-attr"], +) +def test_resolve_provisioner_url_absent(config): + assert resolve_provisioner_url(config) is None + + +@pytest.mark.anyio +async def test_probe_provisioner_not_configured_without_url(): + assert await _probe_provisioner(None) == PROBE_NOT_CONFIGURED + + +@pytest.mark.anyio +@pytest.mark.parametrize( + ("provisioner_url", "trust_env"), + [ + ("http://provisioner:8002", False), + ("http://10.1.2.3:8002", False), + ("http://127.0.0.1:8002", False), + ("http://provisioner.docker.internal:8002", False), + ("https://provisioner.example.com", True), + ("http://8.8.8.8:8002", True), + ], + ids=["service-name", "private-ip", "loopback", "docker-internal", "external-fqdn", "public-ip"], +) +async def test_make_provisioner_client_bypasses_proxy_for_control_plane_addresses(provisioner_url, trust_env): + """Control-plane targets must not be routed through HTTP_PROXY. + + httpx does not treat CIDR entries in NO_PROXY as subnet bypasses the way the + requests-based sandbox calls do, so the probe applies the same + target-aware policy those clients use instead of trusting the environment. + """ + async with health_module._make_provisioner_client(provisioner_url) as client: + assert client.trust_env is trust_env + + +@pytest.mark.anyio +async def test_probe_provisioner_calls_unauthenticated_health_route(monkeypatch): + """The probe targets the provisioner's own GET /health, which needs no API key.""" + seen: list[httpx.Request] = [] + + def _handler(request: httpx.Request) -> httpx.Response: + seen.append(request) + return httpx.Response(200, json={"status": "ok"}) + + _install_provisioner(monkeypatch, _handler) + + assert await _probe_provisioner("http://provisioner:8002") == PROBE_OK + assert [str(request.url) for request in seen] == ["http://provisioner:8002/health"] + assert "x-api-key" not in {name.lower() for name in seen[0].headers} + + +@pytest.mark.anyio +async def test_probe_provisioner_non_200_is_unreachable(monkeypatch): + _install_provisioner(monkeypatch, lambda request: httpx.Response(503, text="draining")) + + assert await _probe_provisioner("http://provisioner:8002") == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_probe_provisioner_connection_error_is_unreachable(monkeypatch): + def _handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused", request=request) + + _install_provisioner(monkeypatch, _handler) + + assert await _probe_provisioner("http://provisioner:8002") == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_probe_provisioner_enforces_probe_timeout(monkeypatch): + monkeypatch.setattr(health_module, "_PROBE_TIMEOUT_SECONDS", 0.05) + + async def _hanging_handler(request: httpx.Request) -> httpx.Response: + await asyncio.sleep(30) + return httpx.Response(200) + + _install_provisioner(monkeypatch, _hanging_handler) + + started = time.perf_counter() + result = await _probe_provisioner("http://provisioner:8002") + elapsed = time.perf_counter() - started + + assert result == PROBE_UNREACHABLE + assert elapsed < 1.0 + + +@pytest.mark.anyio +async def test_concurrent_provisioner_probes_are_serialized(monkeypatch): + """Public /health/ready must not amplify into a connection flood at the provisioner.""" + active = 0 + max_active = 0 + + async def _handler(request: httpx.Request) -> httpx.Response: + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + try: + await asyncio.sleep(0.02) + return httpx.Response(200) + finally: + active -= 1 + + _install_provisioner(monkeypatch, _handler) + + results = await asyncio.gather(*(_probe_provisioner("http://provisioner:8002") for _ in range(6))) + + assert results == [PROBE_OK] * 6 + assert max_active == 1 + + +# --------------------------------------------------------------------------- +# readiness_payload: stream bridge and provisioner verdicts +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_readiness_payload_degraded_when_stream_bridge_unreachable(monkeypatch): + """Redis down means no run can publish, stream or be cancelled: the pod is not ready.""" + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=_FakeCrossProcessBridge(result=False)) + + assert status_code == 503 + assert payload["status"] == "degraded" + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_NOT_CONFIGURED + assert payload["stream_bridge"] == PROBE_UNREACHABLE + assert payload["provisioner"] == PROBE_NOT_CONFIGURED + + +@pytest.mark.anyio +async def test_readiness_payload_ready_when_stream_bridge_reachable(monkeypatch): + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=_FakeCrossProcessBridge(result=True)) + + assert status_code == 200 + assert payload["status"] == "ready" + assert payload["stream_bridge"] == PROBE_OK + + +@pytest.mark.anyio +async def test_readiness_payload_fails_closed_without_stream_bridge(monkeypatch): + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + + status_code, payload = await readiness_payload(CheckpointerConfig(type="memory"), stream_bridge=None) + + assert status_code == 503 + assert payload["stream_bridge"] == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_readiness_payload_reports_unreachable_provisioner_without_degrading(monkeypatch): + """Every replica shares one provisioner; its blip must not take the whole service out.""" + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + _install_provisioner(monkeypatch, lambda request: httpx.Response(503)) + + status_code, payload = await readiness_payload( + CheckpointerConfig(type="memory"), + stream_bridge=MemoryStreamBridge(), + provisioner_url="http://provisioner:8002", + ) + + assert status_code == 200 + assert payload["status"] == "ready" + assert payload["provisioner"] == PROBE_UNREACHABLE + + +@pytest.mark.anyio +async def test_readiness_payload_reports_reachable_provisioner(monkeypatch): + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + _install_provisioner(monkeypatch, lambda request: httpx.Response(200, json={"status": "ok"})) + + status_code, payload = await readiness_payload( + CheckpointerConfig(type="memory"), + stream_bridge=MemoryStreamBridge(), + provisioner_url="http://provisioner:8002", + ) + + assert status_code == 200 + assert payload["provisioner"] == PROBE_OK + + +@pytest.mark.anyio +async def test_readiness_deadline_only_fails_the_probes_that_overran(monkeypatch): + """A probe that ignores its budget must not erase the verdicts of the probes that finished. + + Otherwise a wedged report-only provisioner probe would flip the whole + endpoint to 503 through the deadline path. + """ + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + + async def _hanging(*args) -> str: + await asyncio.sleep(30) + return PROBE_OK + + monkeypatch.setattr(health_module, "_probe_provisioner", _hanging) + monkeypatch.setattr(health_module, "_READINESS_DEADLINE_SECONDS", 0.1) + + status_code, payload = await readiness_payload( + CheckpointerConfig(type="memory"), + stream_bridge=MemoryStreamBridge(), + provisioner_url="http://provisioner:8002", + ) + + assert status_code == 200 + assert payload["status"] == "ready" + assert payload["database"] == PROBE_OK + assert payload["checkpointer"] == PROBE_NOT_CONFIGURED + assert payload["stream_bridge"] == PROBE_NOT_CONFIGURED + assert payload["provisioner"] == PROBE_UNREACHABLE + + +# --------------------------------------------------------------------------- +# HTTP route +# --------------------------------------------------------------------------- + + +def _ready_client(monkeypatch, *, stream_bridge, provisioner_url: str | None) -> TestClient: + """Build the real gateway app and wire the startup state the probe reads. + + The lifespan is deliberately not run: the test pins the ``app.state`` + contract the route depends on instead of booting persistence. + """ + from app.gateway.app import create_app + + app = create_app() + setattr(app.state, READINESS_CHECKPOINTER_CONFIG_ATTR, CheckpointerConfig(type="memory")) + setattr(app.state, READINESS_PROVISIONER_URL_ATTR, provisioner_url) + app.state.stream_bridge = stream_bridge + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine()) + return TestClient(app) + + +def test_health_ready_route_reports_every_dependency(monkeypatch): + _install_provisioner(monkeypatch, lambda request: httpx.Response(200, json={"status": "ok"})) + client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=True), provisioner_url="http://provisioner:8002") + + response = client.get("/health/ready") + + assert response.status_code == 200 + assert response.json() == { + "status": "ready", + "service": "deer-flow-gateway", + "database": PROBE_OK, + "checkpointer": PROBE_NOT_CONFIGURED, + "stream_bridge": PROBE_OK, + "provisioner": PROBE_OK, + } + + +def test_health_ready_route_returns_503_when_stream_bridge_unreachable(monkeypatch): + client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=False), provisioner_url=None) + + response = client.get("/health/ready") + + assert response.status_code == 503 + body = response.json() + assert body["status"] == "degraded" + assert body["stream_bridge"] == PROBE_UNREACHABLE + assert body["provisioner"] == PROBE_NOT_CONFIGURED + + +def test_health_ready_route_keeps_200_when_only_provisioner_is_down(monkeypatch): + def _handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused", request=request) + + _install_provisioner(monkeypatch, _handler) + client = _ready_client(monkeypatch, stream_bridge=MemoryStreamBridge(), provisioner_url="http://provisioner:8002") + + response = client.get("/health/ready") + + assert response.status_code == 200 + assert response.json()["provisioner"] == PROBE_UNREACHABLE + + +def test_health_liveness_route_stays_200_while_readiness_is_degraded(monkeypatch): + """``GET /health`` is liveness only: an unreachable bridge or database must never restart the pod. + + Readiness pulling the pod out of the Service is the intended reaction to a + Redis or database outage; the liveness probe moving with it would turn the + same outage into a restart loop. + """ + client = _ready_client(monkeypatch, stream_bridge=_FakeCrossProcessBridge(result=False), provisioner_url=None) + monkeypatch.setattr("app.gateway.health.get_engine", lambda: _FakeEngine(unreachable=True)) + + assert client.get("/health/ready").status_code == 503 + live = client.get("/health") + assert live.status_code == 200 + assert live.json() == {"status": "healthy", "service": "deer-flow-gateway"} diff --git a/backend/tests/test_stream_bridge.py b/backend/tests/test_stream_bridge.py index f0581361ab..e48edd963d 100644 --- a/backend/tests/test_stream_bridge.py +++ b/backend/tests/test_stream_bridge.py @@ -38,6 +38,14 @@ class _FakeRedis: self.deleted = [] self.expirations = [] self.closed = False + self.ping_calls = 0 + self.ping_error: Exception | None = None + + async def ping(self): + self.ping_calls += 1 + if self.ping_error is not None: + raise self.ping_error + return True async def xadd(self, name, fields, maxlen=None, approximate=True): self.counters[name] += 1 @@ -535,6 +543,42 @@ def redis_bridge() -> RedisStreamBridge: return RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=_FakeRedis()) +# --------------------------------------------------------------------------- +# Readiness ping +# --------------------------------------------------------------------------- + + +@pytest.mark.anyio +async def test_memory_bridge_ping_has_no_external_backend(): + """The in-process bridge has nothing external to probe and never reports unreachable.""" + bridge = MemoryStreamBridge() + + assert bridge.supports_cross_process is False + assert await bridge.ping() is True + + +@pytest.mark.anyio +async def test_redis_bridge_ping_reports_reachable_backend(): + fake = _FakeRedis() + bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=fake) + + assert bridge.supports_cross_process is True + assert await bridge.ping() is True + assert fake.ping_calls == 1 + + +@pytest.mark.anyio +async def test_redis_bridge_ping_reports_unreachable_backend_on_redis_error(): + """A Redis-side failure is a readiness answer (False), not an exception for the probe to unpack.""" + from redis.exceptions import ConnectionError as RedisConnectionError + + fake = _FakeRedis() + fake.ping_error = RedisConnectionError("connection refused") + bridge = RedisStreamBridge(redis_url="redis://fake", queue_maxsize=2, client=fake) + + assert await bridge.ping() is False + + @pytest.mark.anyio async def test_redis_publish_subscribe(redis_bridge: RedisStreamBridge): """Redis bridge should deliver events in order and terminate on end.""" diff --git a/deploy/helm/deer-flow/README.md b/deploy/helm/deer-flow/README.md index 464abfe829..f8e97f10b3 100644 --- a/deploy/helm/deer-flow/README.md +++ b/deploy/helm/deer-flow/README.md @@ -234,9 +234,16 @@ helm install deer-flow deploy/helm/deer-flow \ ```bash kubectl -n deer-flow get pods kubectl -n deer-flow port-forward svc/nginx 2026:2026 -curl http://localhost:2026/health # gateway health via nginx +curl http://localhost:2026/health # gateway liveness via nginx +curl http://localhost:2026/health/ready # readiness: database, checkpointer, stream_bridge, provisioner ``` +`/health/ready` is what the gateway `readinessProbe` hits. It answers 503 while +Postgres or the Redis stream bridge is unreachable (`stream_bridge: unreachable`), +so those pods leave the Service instead of accepting runs they cannot stream. +The `provisioner` field reports the provisioner's own `/health` but never +changes the status code: every gateway pod shares that one provisioner. + Hit the Ingress host (map it in `/etc/hosts` for local clusters) to load the UI. Provisioner sanity check (`/health` is unauthenticated; `/api/*` requires the @@ -354,7 +361,10 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002 `DEER_FLOW_STREAM_BRIDGE_REDIS_URL`; `config.yaml` sets `stream_bridge.type: redis` by default. No-auth by default (ClusterIP isolation, matching compose); set `redis.auth.password` to enable AUTH. For a managed Redis, disable the - bundled instance and point at it via `redis.external`. + bundled instance and point at it via `redis.external`. The gateway readiness + probe pings this Redis on every check and reports the pod unready while it + is unreachable, since without the bridge no run can publish, stream or be + cancelled from a peer pod. - **Persistence.** A PVC (`-home`) backs `/app/backend/.deer-flow` (sqlite DB, memory, custom agents, per-thread user-data). The gateway mounts it with `subPath: deer-flow` so the layout matches the provisioner's PVC diff --git a/deploy/helm/deer-flow/templates/gateway-deployment.yaml b/deploy/helm/deer-flow/templates/gateway-deployment.yaml index ce2d6a9eae..88bb92b54f 100644 --- a/deploy/helm/deer-flow/templates/gateway-deployment.yaml +++ b/deploy/helm/deer-flow/templates/gateway-deployment.yaml @@ -216,9 +216,12 @@ spec: port: http initialDelaySeconds: 5 periodSeconds: 10 - # /health/ready bounds both probes to a 3s endpoint deadline; + # /health/ready bounds all its probes (database, checkpointer, + # Redis stream bridge, provisioner) to a 3s endpoint deadline; # timeoutSeconds must stay above that bound or Kubernetes aborts - # slow-but-healthy responses at its 1s default. + # slow-but-healthy responses at its 1s default. An unreachable + # Redis makes the pod unready; the provisioner verdict is + # report-only. timeoutSeconds: 5 livenessProbe: httpGet: diff --git a/docker/docker-compose.yaml b/docker/docker-compose.yaml index eea51fc96b..f4367d28af 100644 --- a/docker/docker-compose.yaml +++ b/docker/docker-compose.yaml @@ -180,8 +180,10 @@ services: redis: condition: service_healthy healthcheck: - # /health/ready runs both probes concurrently within a 3s endpoint - # deadline; the client timeout must stay above that bound. + # /health/ready runs its probes (database, checkpointer, Redis stream + # bridge, provisioner) concurrently within a 3s endpoint deadline; the + # client timeout must stay above that bound. The gateway is unhealthy + # while Redis is unreachable; the provisioner verdict is report-only. test: ["CMD", "python", "-c", "import urllib.request; response = urllib.request.urlopen('http://127.0.0.1:8001/health/ready', timeout=5); raise SystemExit(0 if response.status == 200 else 1)"] interval: 5s timeout: 5s