mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(gateway): gate declared multi-instance deployments and require the redis stream bridge (#6328)
* fix(gateway): gate declared multi-instance deployments and require the redis stream bridge GATEWAY_WORKERS / WEB_CONCURRENCY only count the uvicorn workers of one process tree, so a Kubernetes Deployment with replicas > 1 (one worker per Pod) passed every multi-worker startup check while each Pod's startup orphan reconciliation wrote the other Pods' lease-less runs off as crashed on every rolling update. Add `deployment.multi_instance` as an explicit declaration that runs the same gate, plus a DEER_FLOW_MULTI_INSTANCE env override so deploy tooling can set the declaration from its replica count without editing config.yaml (wiring the Helm chart to export it is a separate change). Both the worker count and the declaration now also require a Redis stream bridge and refuse an explicit memory sandbox ownership store, since the memory bridge and ownership store are process-local. The agent-storage divergence warning, the inbound webhook dedupe warning and the WeChat QR-login guard honor the same declaration. * docs(changelog): link the multi-instance gate entry to #6328 * fix(ci): bump chart config_version to 53 and keep AGENTS.md under the size limits config.example.yaml moved to config_version 53 for the new deployment section, so the Helm chart's embedded config and its README example follow. The multi-instance notes in backend/AGENTS.md and backend/app/gateway/AGENTS.md shrink to one line each: main already sits within 55 bytes of the per-file hard limit for the gateway guide and within 163 bytes of the chain hard limit under backend/, and the full contract lives in backend/docs/CONFIGURATION.md. * fix(gateway): name every active knob in the multi-process rollback When GATEWAY_WORKERS > 1 and a multi-instance declaration are both active, the refusal offered only "Set GATEWAY_WORKERS=1", so an operator who followed it bounced through a second refusal on DEER_FLOW_MULTI_INSTANCE / deployment.multi_instance at the next start. The rollback now withdraws both knobs, and the browser-tools refusal takes the same path instead of the worker-count-only message. multi_instance_declaration() returns a frozen MultiInstanceDeclaration (source, value) with knob/rollback properties instead of a display string, so deps.py no longer branches on message prose. Tests pin the combined rollback text and that following it actually clears the gate. * docs(gateway): keep the gateway AGENTS.md under the size hard limit after merging main main grew backend/app/gateway/AGENTS.md by 30 bytes, which left it 25 bytes under the 49152-byte hard limit; the one-sentence multi-instance note from this branch pushed the merged file to 49164. Shorten that sentence by 18 bytes so scripts/check_agent_guidance.py passes on the merge result. * ci: re-run checks after test_stop_cancels_a_hung_driver_poll flaked on shard 4 tests/test_mcp_task_service.py::test_stop_cancels_a_hung_driver_poll timed out once on backend-unit-tests (4) for 2c16206ac. The MCP task service and its test are identical to main, main's run for cfe2023df passed, and 15 consecutive local runs pass; the test gives service.stop() a 1s budget while cancellation drain allows 5s, so a loaded runner can exceed it. Fork contributors cannot re-run a job, hence this empty commit. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao
GitHub
parent
c2f31e440e
commit
5872d07e9a
16
CHANGELOG.md
16
CHANGELOG.md
@ -135,6 +135,21 @@ This release closes that milestone with **301 merged pull requests**.
|
||||
duplicate keys, and syntactically valid large numbers are allowed; a UTF-8 BOM
|
||||
is rejected. Syntax only — no schema or business-field checks; existing
|
||||
criteria and verdict semantics are unchanged. ([#5947])
|
||||
- **gateway:** The multi-process startup gate now fires on an explicit
|
||||
`deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`) as well
|
||||
as `GATEWAY_WORKERS > 1`. The worker-count variables only see one process
|
||||
tree, so a Kubernetes Deployment with several one-worker Pods passed every
|
||||
check while each Pod's startup orphan reconciliation wrote the other Pods'
|
||||
lease-less runs off as crashed on every rolling update. Both paths require
|
||||
Postgres, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true`
|
||||
and now also a Redis stream bridge, and refuse an explicit
|
||||
`sandbox.ownership.type: memory`, process-local browser tools and a
|
||||
scheduler without `scheduler.multi_instance`. The agent-storage divergence
|
||||
warning, the inbound webhook dedupe warning and the WeChat QR-login guard
|
||||
honor the same declaration. **Behavior change:** `GATEWAY_WORKERS > 1` with
|
||||
the memory stream bridge no longer starts; configure `stream_bridge.type:
|
||||
redis` or `DEER_FLOW_STREAM_BRIDGE_REDIS_URL` (docker-compose and the Helm
|
||||
chart already inject it). ([#6328])
|
||||
|
||||
#### Memory
|
||||
|
||||
@ -7799,3 +7814,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6313]: https://github.com/bytedance/deer-flow/pull/6313
|
||||
[#6319]: https://github.com/bytedance/deer-flow/pull/6319
|
||||
[#6326]: https://github.com/bytedance/deer-flow/pull/6326
|
||||
[#6328]: https://github.com/bytedance/deer-flow/pull/6328
|
||||
|
||||
@ -472,7 +472,7 @@ Browser login uses `HttpOnly` session cookies. The login page offers a "keep me
|
||||
DeerFlow still uses `Forwarded` / `X-Forwarded-*` headers to recover the browser-facing scheme and origin behind a proxy. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header. Configure the outer trusted proxy to replace or strip client-supplied forwarding headers before traffic reaches DeerFlow.
|
||||
|
||||
> [!IMPORTANT]
|
||||
> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
|
||||
> The Gateway still owns active run tasks in process, so production defaults to a single Gateway worker (`GATEWAY_WORKERS=1`). Multi-worker deployments require Postgres, the Redis stream bridge (`stream_bridge.type: redis`), `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`; process-local memory/JSONL event stores cannot enforce singleton delivery receipts across workers. Kubernetes replicas run one worker per Pod, which the worker count cannot see: declare them with `deployment.multi_instance: true` (or `DEER_FLOW_MULTI_INSTANCE=1`, which deploy tooling such as a Helm chart can set from its replica count) so the startup gate enforces the same prerequisites instead of staying inert. The bridge shares SSE delivery and bounded `Last-Event-ID` replay across workers. When a valid reconnect cursor has been trimmed, or a subscriber that already established an empty-stream wait falls behind before its first delivery, Memory and Redis emit a machine-readable SSE `gap` event instead of silently returning a partial replay; the Web UI reloads durable thread/event state and resumes from the retained tail. Lease reconciliation marks runs from dead workers as errors, persists their delivery receipts, publishes the terminal stream marker, schedules retained-stream cleanup, and updates the affected thread status. SSE, `/wait`, and internal stream consumers use `stream_bridge.heartbeat_interval_seconds` (default `15`) for idle liveness checks; changing it requires a Gateway restart. Malformed Redis reconnect IDs live-tail new events instead of replaying the retained buffer, and the rolling retained-buffer TTL (`stream_ttl_seconds`) remains a cleanup safety net rather than a run timeout. IM channel state and other process-local services still need their own multi-worker coordination.
|
||||
>
|
||||
> In single-process JSONL deployments, cancelling an admitted event-store mutation waits for its background file I/O, rollback, and bookkeeping to settle before releasing the thread write lock. This prevents an older cancelled write from recreating deleted records or rolling back a later successful write. Cancellation can therefore wait on slow storage; it does not stop an in-flight filesystem operation. Callers still waiting to acquire the lock can cancel without starting a mutation. A batch spanning multiple threads drains its current thread group before propagating cancellation; subsequent thread groups do not start.
|
||||
>
|
||||
|
||||
@ -18,6 +18,7 @@ The backend runs a LangGraph-based super agent with sandbox execution, persisten
|
||||
- Scheduled tasks dispatch through the normal Gateway run path. `launch_scheduled_thread_run` reads `get_app_config()` at dispatch and passes `scheduler.recursion_limit` (default 1000, matching the web UI; clamped by `max_recursion_limit`), so YAML changes apply on the next run without restarting Gateway.
|
||||
- Run-history `status` filters are occurrence states, not task states. `ScheduledTaskRunStatus` in `persistence/scheduled_tasks/model.py` is the shared API/repository vocabulary and must match the active and terminal occurrence-status sets. Keep owner lookup before reading history, and apply SQL task/status predicates before pagination; omitted status preserves the existing response.
|
||||
- The background scheduler is single-instance by default. `scheduler.multi_instance=true` opts into lease-aware recovery across Gateway instances and requires shared Postgres, `run_ownership.heartbeat_enabled=true`, and `run_events.backend=db`; otherwise startup rejects the configuration. Live scheduled runs are preserved when a peer starts; expired launch claims return to the durable queue, expired run leases are atomically taken over, stale launch writes are fenced by lease ownership, and the Postgres advisory-locked budget makes `max_concurrent_runs` a shared global cap for `launching`/`running` rows.
|
||||
- The multi-process startup gate also fires on `deployment.multi_instance: true`; contract in `docs/CONFIGURATION.md`.
|
||||
- Long-running MCP work uses a separate durable task runtime (`McpTaskService` + `mcp_tasks`, lease-based recovery) rather than keeping remote task IDs or status polling inside the Agent loop; only submit remains Agent-visible, the database is the source of truth, and `ThreadState` receives only a bounded current-thread projection. Full contract (leases, cancellation fencing, delivery idempotency, management-tool exposure): [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md).
|
||||
- MCP task notification retries, dead-lettering, and the cancel endpoint's worker-stopped 503 are part of that same contract — see [packages/harness/deerflow/mcp/AGENTS.md](packages/harness/deerflow/mcp/AGENTS.md).
|
||||
- Scheduled-task dispatch permits one active occurrence per task via `uq_scheduled_task_run_active` (`task_id WHERE status IN ('queued','launching','running')`). Durable `queued` rows survive restarts; only lease-fenced `launching` may call Gateway launch; `running` references the durable run. Stable admission idempotency keys reuse that run after recovery. Reused-thread `ConflictError` returns `launching` to `queued`; other launch errors become `failed`. Atomic queue claims enforce `max_concurrent_runs`, excluding waiting rows; the budget count and its UPDATE are separate statements, so writers must serialize before the count (Postgres advisory lock; SQLite `BEGIN IMMEDIATE`, whose deferred transaction otherwise reserves the writer only at the UPDATE) or claims on distinct rows overshoot the cap. Repeated triggers coalesce; same-thread FIFO blocks behind older active rows. Queue admission, PATCH/resume, pause and delete lock the parent before the occurrence, freezing active task definitions. Pause/delete atomically cancel `queued` work but reject `launching`/`running`; PATCH/resume reject all active states. Only queued conflicts offer pause cancellation. Manual triggers may queue/run while paused. Recovery locks task/run pairs in task-id/run-id order and restores `run_id`, `started_at` and live errors before releasing launch claims. Launch/failure/timeout updates use one parent-first transaction to prevent interleaved claims. Queue timeout fails the occurrence and advances scheduled work to prevent immediate requeue. Repository boundaries coerce serialized timestamps before SQL `DateTime` binding.
|
||||
|
||||
@ -17,6 +17,8 @@ from typing import Any, Protocol
|
||||
|
||||
from sqlalchemy import text
|
||||
|
||||
from deerflow.config.deployment_config import multi_instance_declaration
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
INBOUND_DEDUPE_TTL_SECONDS = 10 * 60
|
||||
@ -269,7 +271,11 @@ def make_inbound_dedupe_store(app_config: Any | None = None) -> InboundDedupeSto
|
||||
db_is_postgres = db_backend == "postgres"
|
||||
|
||||
workers, worker_env = _gateway_workers()
|
||||
multi_worker = workers > 1
|
||||
# The worker count only sees one process tree; Kubernetes replicas declare
|
||||
# themselves through deployment.multi_instance / DEER_FLOW_MULTI_INSTANCE.
|
||||
declaration = multi_instance_declaration(app_config)
|
||||
multi_worker = workers > 1 or declaration is not None
|
||||
topology = f"{worker_env}>1" if workers > 1 else getattr(declaration, "knob", None)
|
||||
|
||||
if backend == "postgres":
|
||||
if not db_is_postgres:
|
||||
@ -282,7 +288,7 @@ def make_inbound_dedupe_store(app_config: Any | None = None) -> InboundDedupeSto
|
||||
if backend == "memory":
|
||||
if multi_worker:
|
||||
logger.warning(
|
||||
f"dedupe_storage=memory with {worker_env}>1: inbound webhook dedupe "
|
||||
f"dedupe_storage=memory with {topology}: inbound webhook dedupe "
|
||||
"is per-pod and will NOT drop redeliveries routed to a different replica. "
|
||||
"Use dedupe_storage=postgres (or remove the setting to let 'auto' pick it) "
|
||||
"for multi-worker deployments. See issue #4120."
|
||||
|
||||
@ -180,8 +180,8 @@ without changed outputs keep ordinary chat behavior. Journal mechanics
|
||||
(callback attribution, receipt idempotency and retries, orphan recovery):
|
||||
`packages/harness/deerflow/runtime/AGENTS.md`. Multi-worker deployments
|
||||
require `run_events.backend: db` for shared, ordered delivery events; the
|
||||
startup gate rejects process-local memory and JSONL event stores when
|
||||
`GATEWAY_WORKERS > 1`.
|
||||
startup gate rejects memory/JSONL event stores when `GATEWAY_WORKERS > 1`
|
||||
or `deployment.multi_instance: true`.
|
||||
|
||||
**RunManager / RunStore contract**:
|
||||
- LangGraph-compatible run requests validate their supported subset before creating a run. `runtime/stream_modes.py` is the shared backend contract for public stream modes and the worker's `graph.astream` mapping; the public `messages-tuple` mode maps to LangGraph's internal `messages` mode, while public `messages`, `events`, and other unsupported modes are rejected instead of being dropped or replaced with `values`. `app/gateway/run_models.py::RunCreateRequest` is shared by HTTP and internal scheduled launch paths, retains only truthful compatibility defaults for unimplemented options (`if_not_exists="create"` plus `None` placeholders), returns 422 for unsupported values including `on_completion="complete"`, `on_completion="continue"`, and `multitask_strategy="enqueue"`, and forbids undeclared SDK options so fields such as `checkpoint_during` and `durability` cannot be silently discarded. A placeholder must still accept the stock SDK's own default: `langgraph_sdk` drops only `None` from its run payload, so `stream_resumable=False` reaches every request and means "non-resumable", which is what DeerFlow serves — rejecting it 422'd every IM channel run (#4466). `tests/test_run_request_validation.py::test_gateway_accepts_langgraph_sdk_default_payload` pins the real SDK payload against this boundary; channel tests mock the SDK client and cannot catch this class of drift.
|
||||
|
||||
@ -29,6 +29,7 @@ from langgraph.types import Checkpointer
|
||||
|
||||
from deerflow.community.browser_automation.session import browser_multi_worker_error
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.config.deployment_config import multi_instance_declaration
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
from deerflow.runtime import ORPHAN_RECOVERY_STOP_REASON, STARTUP_ORPHAN_RECOVERY_ERROR, RunContext, RunManager, StreamBridge
|
||||
from deerflow.runtime.events.store.base import RunEventStore
|
||||
@ -86,9 +87,55 @@ def _gateway_worker_count() -> tuple[int, str]:
|
||||
return 1, _WORKER_COUNT_ENV_VARS[0]
|
||||
|
||||
|
||||
def _multi_process_signal(config: AppConfig) -> tuple[str, str] | None:
|
||||
"""Return ``(reason, rollback)`` when this process must assume peer Gateway processes.
|
||||
|
||||
``reason`` names the knob that established the topology (``GATEWAY_WORKERS=2``,
|
||||
``DEER_FLOW_MULTI_INSTANCE=1`` or ``deployment.multi_instance=true``) and
|
||||
``rollback`` is the single-instance remediation a refusal message offers.
|
||||
When the worker count and a declaration are both active, ``rollback``
|
||||
withdraws both: resetting only the worker count would bounce the operator
|
||||
through a second refusal on the declaration at the next start.
|
||||
|
||||
The worker-count variables only see one process tree, so a Kubernetes
|
||||
Deployment with ``replicas > 1`` and one worker per Pod is invisible to
|
||||
them; the explicit declaration exists for exactly that topology.
|
||||
"""
|
||||
workers, worker_env = _gateway_worker_count()
|
||||
declaration = multi_instance_declaration(config)
|
||||
if workers > 1:
|
||||
rollback = f"Set {worker_env}=1"
|
||||
if declaration is not None:
|
||||
rollback = f"{rollback} and {declaration.rollback}"
|
||||
return f"{worker_env}={workers}", rollback
|
||||
if declaration is None:
|
||||
return None
|
||||
step = declaration.rollback
|
||||
return declaration.knob, f"{step[0].upper()}{step[1:]} and run a single Gateway instance"
|
||||
|
||||
|
||||
def _stream_bridge_is_cross_process(config: AppConfig) -> bool:
|
||||
"""Return whether live run events can reach SSE clients on every instance.
|
||||
|
||||
Mirrors ``runtime/stream_bridge/async_provider.py::_resolve_config``: the
|
||||
config.yaml section wins, and an omitted section falls back to the
|
||||
``DEER_FLOW_STREAM_BRIDGE_REDIS_URL`` variable the Docker and Helm
|
||||
deployments inject.
|
||||
"""
|
||||
bridge = getattr(config, "stream_bridge", None)
|
||||
if bridge is not None:
|
||||
return getattr(bridge, "type", None) == "redis"
|
||||
return bool((os.environ.get("DEER_FLOW_STREAM_BRIDGE_REDIS_URL") or "").strip())
|
||||
|
||||
|
||||
def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
|
||||
"""Refuse unsafe multi-process configurations before persistence starts.
|
||||
|
||||
A deployment counts as multi-process when ``GATEWAY_WORKERS`` /
|
||||
``WEB_CONCURRENCY`` is above 1, or when the operator declares peers with
|
||||
``deployment.multi_instance: true`` / ``DEER_FLOW_MULTI_INSTANCE=1`` (the
|
||||
worker count cannot see other Pods, and a Pod that starts next to a peer
|
||||
without these prerequisites writes the peer's live runs off as orphans).
|
||||
Multi-instance scheduler recovery also needs the durable run ownership
|
||||
contract even when each Pod runs a single Gateway worker.
|
||||
|
||||
@ -105,13 +152,16 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
|
||||
every run has a NULL lease, so reconciliation treats all inflight
|
||||
runs as orphans and Worker B would kill Worker A's live runs on
|
||||
every rolling update or scale-up.
|
||||
6. The stream bridge must be Redis. The memory bridge is process-local,
|
||||
so SSE streams, reconnects and ``/wait`` only work on the owner.
|
||||
7. ``sandbox.ownership.type`` must not be an explicit ``memory``: an
|
||||
in-process ownership store cannot see peers, so reconciliation would
|
||||
adopt and idle-destroy another instance's live containers (#4206).
|
||||
|
||||
This gate runs once at startup before any persistence engine is
|
||||
initialised so the error message is clear and the process exits
|
||||
immediately.
|
||||
"""
|
||||
workers, worker_env = _gateway_worker_count()
|
||||
|
||||
scheduler = getattr(config, "scheduler", None)
|
||||
multi_instance_requested = bool(getattr(scheduler, "multi_instance", False))
|
||||
multi_instance_scheduler = bool(getattr(scheduler, "enabled", False) and multi_instance_requested)
|
||||
@ -127,34 +177,54 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
|
||||
if multi_instance_requested and (run_ownership is None or not run_ownership.heartbeat_enabled):
|
||||
raise SystemExit("scheduler.multi_instance=true requires run_ownership.heartbeat_enabled=true so peer runs retain a valid lease. Set scheduler.multi_instance=false or enable run ownership heartbeats.")
|
||||
|
||||
if workers <= 1:
|
||||
signal = _multi_process_signal(config)
|
||||
if signal is None:
|
||||
return
|
||||
reason, rollback = signal
|
||||
|
||||
if config.scheduler.enabled and not multi_instance_scheduler:
|
||||
raise SystemExit(f"{worker_env}={workers} cannot run with scheduler.enabled=true because each worker starts its own scheduler. Set {worker_env}=1, scheduler.multi_instance=true, or scheduler.enabled=false.")
|
||||
raise SystemExit(f"{reason} cannot run with scheduler.enabled=true because each worker starts its own scheduler. {rollback}, scheduler.multi_instance=true, or scheduler.enabled=false.")
|
||||
|
||||
if _browser_tools_enabled_in_config(config):
|
||||
raise SystemExit(browser_multi_worker_error(workers))
|
||||
workers, _worker_env = _gateway_worker_count()
|
||||
if workers > 1 and multi_instance_declaration(config) is None:
|
||||
raise SystemExit(browser_multi_worker_error(workers))
|
||||
raise SystemExit(f"{reason} cannot enable agentic browser tools: browser sessions are process-local and a request can land on any instance. {rollback} or disable the browser_navigate tool.")
|
||||
|
||||
if backend != "postgres":
|
||||
raise SystemExit(f"{worker_env}={workers} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. Set {worker_env}=1 or switch to Postgres.")
|
||||
raise SystemExit(f"{reason} requires database.backend='postgres', but database.backend is '{backend}'. SQLite cannot support concurrent multi-process access. {rollback} or switch to Postgres.")
|
||||
|
||||
if run_events_backend != "db":
|
||||
raise SystemExit(
|
||||
f"{worker_env}={workers} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. "
|
||||
f"{reason} requires run_events.backend='db', but run_events.backend is '{run_events_backend}'. "
|
||||
"Memory and JSONL event stores are process-local, so delivery receipt singleton guarantees cannot hold across workers. "
|
||||
f"Set {worker_env}=1 or configure run_events.backend: db."
|
||||
f"{rollback} or configure run_events.backend: db."
|
||||
)
|
||||
|
||||
if run_ownership is None or not run_ownership.heartbeat_enabled:
|
||||
raise SystemExit(
|
||||
f"{worker_env}={workers} requires run_ownership.heartbeat_enabled=true. "
|
||||
f"{reason} requires run_ownership.heartbeat_enabled=true. "
|
||||
"Without heartbeat, every run has a NULL lease, so reconciliation "
|
||||
"treats all inflight runs as orphans — Worker B would kill Worker A's "
|
||||
"live runs on every rolling update or scale-up. "
|
||||
"treats all inflight runs as orphans — a starting instance would kill its "
|
||||
"peers' live runs on every rolling update or scale-up. "
|
||||
"Set run_ownership.heartbeat_enabled=true in config.yaml."
|
||||
)
|
||||
|
||||
if not _stream_bridge_is_cross_process(config):
|
||||
raise SystemExit(
|
||||
f"{reason} requires stream_bridge.type='redis' (or DEER_FLOW_STREAM_BRIDGE_REDIS_URL), "
|
||||
"but the stream bridge is the process-local memory bridge: SSE streams, reconnects and /wait "
|
||||
f"only work on the instance that owns the run. {rollback} or configure the redis stream bridge."
|
||||
)
|
||||
|
||||
ownership = getattr(getattr(config, "sandbox", None), "ownership", None)
|
||||
if ownership is not None and getattr(ownership, "type", None) == "memory":
|
||||
raise SystemExit(
|
||||
f"{reason} cannot run with sandbox.ownership.type='memory': an in-process ownership store cannot see "
|
||||
"peer instances, so startup reconciliation would adopt and idle-destroy containers another instance is "
|
||||
"using (#4206). Set sandbox.ownership.type='redis', or omit the section so it is inferred from the redis stream bridge."
|
||||
)
|
||||
|
||||
|
||||
def _validate_agent_storage(config: AppConfig) -> None:
|
||||
"""Fail fast on an agent-storage backend the database cannot support.
|
||||
@ -178,14 +248,13 @@ def _validate_agent_storage(config: AppConfig) -> None:
|
||||
f"but database.backend is '{db_backend}'. A 'memory' database is per-process and cannot "
|
||||
"share agent definitions across nodes. Set database.backend, or use agent_storage.backend='file'."
|
||||
)
|
||||
workers, worker_env = _gateway_worker_count()
|
||||
if workers > 1 and db_backend == "postgres" and backend == "file":
|
||||
signal = _multi_process_signal(config)
|
||||
if signal is not None and db_backend == "postgres" and backend == "file":
|
||||
logger.warning(
|
||||
"%s=%s with database.backend='postgres' but agent_storage.backend='file': "
|
||||
"%s with database.backend='postgres' but agent_storage.backend='file': "
|
||||
"custom agents and managed subagents are stored per-node on local disk and are not visible "
|
||||
"across workers/nodes. Set agent_storage.backend='db' to share them.",
|
||||
worker_env,
|
||||
workers,
|
||||
signal[0],
|
||||
)
|
||||
|
||||
|
||||
|
||||
@ -21,6 +21,7 @@ from app.channels.wechat_qr_login import QRLoginError, WechatQRLogin
|
||||
from app.gateway.deps import require_admin_user
|
||||
from app.gateway.persistent_writes import run_drained_write
|
||||
from deerflow.config.channel_connections_config import ChannelConnectionsConfig
|
||||
from deerflow.config.deployment_config import multi_instance_declaration
|
||||
from deerflow.persistence.channel_connections import ChannelConnectionRepository
|
||||
from deerflow.persistence.engine import get_session_factory
|
||||
from deerflow.utils.file_io import await_drained
|
||||
@ -772,10 +773,14 @@ async def _require_wechat_qr_login(request: Request) -> ChannelConnectionsConfig
|
||||
workers = int(os.environ.get("GATEWAY_WORKERS") or os.environ.get("WEB_CONCURRENCY") or "1")
|
||||
except ValueError:
|
||||
workers = 0
|
||||
if workers != 1:
|
||||
# Kubernetes replicas run one worker each; they declare their peers instead.
|
||||
if workers != 1 or multi_instance_declaration(_get_app_config()) is not None:
|
||||
raise HTTPException(
|
||||
status_code=503,
|
||||
detail="WeChat QR login requires a single Gateway worker. Set GATEWAY_WORKERS=1 (or WEB_CONCURRENCY=1 when using Uvicorn directly), or enter a bot token manually.",
|
||||
detail=(
|
||||
"WeChat QR login requires a single Gateway worker. Set GATEWAY_WORKERS=1 (or WEB_CONCURRENCY=1 when using Uvicorn directly), "
|
||||
"run one Gateway instance without deployment.multi_instance / DEER_FLOW_MULTI_INSTANCE, or enter a bot token manually."
|
||||
),
|
||||
)
|
||||
return config
|
||||
|
||||
|
||||
@ -518,7 +518,7 @@ Notes:
|
||||
- **Upgrade note:** before upgrading a deployment with `GATEWAY_WORKERS > 1` and `scheduler.enabled: true`, either run the scheduler on exactly one Gateway worker or enable `scheduler.multi_instance: true` with shared Postgres, `run_ownership.heartbeat_enabled: true`, and `run_events.backend: db`. The startup gate now rejects the unsafe combination instead of allowing it to start silently.
|
||||
- **Upgrade note:** in multi-instance mode, `max_concurrent_runs` is cluster-wide rather than per Pod and counts `launching`/`running` occurrences. Waiting `queued` rows remain outside the execution cap; capacity does not multiply with the replica count.
|
||||
- **Upgrade note:** `scheduler.multi_instance` and its related scheduler, ownership, and run-event settings are startup-only. Restart all Gateway Pods together after changing them; a ConfigMap update without a coordinated restart leaves the running service on its previous mode.
|
||||
- Multi-worker deployments (`GATEWAY_WORKERS > 1`) must use the Postgres database backend, enable run ownership heartbeats, and set `run_events.backend: db`. SQLite silently ignores row-level locks, while memory and JSONL run-event stores are process-local and cannot enforce singleton delivery receipts across workers; startup rejects these combinations. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep the Gateway to one worker process while `browser_navigate` is enabled — `GATEWAY_WORKERS=1`, plus `WEB_CONCURRENCY` unset or `1` on launches that pass uvicorn no worker count (`backend/Dockerfile`, `scripts/serve.sh`), where uvicorn takes the process count from it. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available.
|
||||
- Multi-worker deployments (`GATEWAY_WORKERS > 1`) and declared multi-instance deployments (`deployment.multi_instance: true` / `DEER_FLOW_MULTI_INSTANCE=1`, see [Deployment topology](#deployment-topology-multi-instance)) must use the Postgres database backend, enable run ownership heartbeats, set `run_events.backend: db`, and use the Redis stream bridge. SQLite silently ignores row-level locks, while memory and JSONL run-event stores are process-local and cannot enforce singleton delivery receipts across workers; startup rejects these combinations. The process-local agentic browser tool group is incompatible with multiple Gateway workers; keep the Gateway to one worker process while `browser_navigate` is enabled — `GATEWAY_WORKERS=1`, plus `WEB_CONCURRENCY` unset or `1` on launches that pass uvicorn no worker count (`backend/Dockerfile`, `scripts/serve.sh`), where uvicorn takes the process count from it. Browser control also requires the backend `browser` extra (`cd backend && uv sync --extra browser && uv run playwright install chromium`); startup detects enabled browser config and fails fast when Playwright is missing, and `/api/features` reports `browser_control.enabled=false` until the runtime is available.
|
||||
- The MVP supports thread reuse and fresh-thread-per-run execution modes.
|
||||
- Create/update accept optional `assistant_id` (`lead_agent` by default, or an existing custom agent for the task owner).
|
||||
- Create/update accept `once`, `cron`, and `interval`. Interval uses `schedule_spec.every_seconds` (UTC `now + N`, no missed-beat catch-up). N is at least `min_once_delay_seconds` (default 60) and at most 30 days.
|
||||
@ -526,6 +526,21 @@ Notes:
|
||||
- Scheduled task definitions and task-run history are persisted in the application database.
|
||||
- With `channel_connections.enabled: true`, the scheduler enqueues IM outcome notifications for the task owner's connected identities. A delivery worker (same poll cadence as the scheduler) pushes them. Scheduled runs that finish as success or failed notify. For goal tasks, an `unmet` scheduled occurrence sends a goal-unmet notice instead, and an automatic pause after three unmet occurrences sends an auto-pause notice; a stop requested by the agent adds no separate notice. Manual triggers and interrupts stay silent, and so do occurrences that end without a finished run (launch error, queue timeout, restart recovery). Channel/transport outages park rows without consuming the retry budget, for up to about a day; platform rejections exhaust ~15 minutes of counted retries then settle `failed`. The worker re-checks the binding right before sending: a target the owner has disconnected since enqueue is dropped as `failed`, never pushed. Only WeCom currently implements proactive `send_notification`.
|
||||
|
||||
### Deployment topology (multi-instance)
|
||||
|
||||
```yaml
|
||||
deployment:
|
||||
multi_instance: false
|
||||
```
|
||||
|
||||
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.
|
||||
- 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.
|
||||
|
||||
### Agent Storage
|
||||
|
||||
Custom agent **definitions** (`config.yaml` + `SOUL.md`) are stored per-user on
|
||||
|
||||
@ -20,6 +20,7 @@ from deerflow.config.channel_connections_config import ChannelConnectionsConfig
|
||||
from deerflow.config.checkpointer_config import CheckpointerConfig, load_checkpointer_config_from_dict
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.config.dedupe_storage_config import DedupeStorageConfig
|
||||
from deerflow.config.deployment_config import DeploymentConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig
|
||||
from deerflow.config.file_signature import ConfigSignature as _ConfigSignature
|
||||
from deerflow.config.file_signature import get_config_signature as _get_config_signature
|
||||
@ -386,6 +387,13 @@ class AppConfig(BaseModel):
|
||||
field_doc="Stream bridge connecting agent workers to SSE endpoints.",
|
||||
),
|
||||
)
|
||||
deployment: DeploymentConfig = Field(
|
||||
default_factory=DeploymentConfig,
|
||||
description=format_field_description(
|
||||
"deployment",
|
||||
field_doc="Deployment topology declaration: whether more than one Gateway instance shares this database (drives the multi-process startup safety gate).",
|
||||
),
|
||||
)
|
||||
run_ownership: RunOwnershipConfig = Field(
|
||||
default_factory=RunOwnershipConfig,
|
||||
description=format_field_description(
|
||||
|
||||
107
backend/packages/harness/deerflow/config/deployment_config.py
Normal file
107
backend/packages/harness/deerflow/config/deployment_config.py
Normal file
@ -0,0 +1,107 @@
|
||||
"""Deployment topology declaration for the Gateway startup safety gates.
|
||||
|
||||
``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-process safety gate in
|
||||
``app/gateway/deps.py`` stays inert -- while each Pod's startup orphan
|
||||
reconciliation still treats the other Pods' lease-less runs as crashed.
|
||||
|
||||
This module is the explicit declaration that closes that gap: an operator (or
|
||||
deploy tooling such as a Helm chart, which can export the env variable from
|
||||
its replica count) states that more than one Gateway instance shares the
|
||||
database, and the gate then enforces the same prerequisites it enforces for
|
||||
``GATEWAY_WORKERS > 1``.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
#: Environment override for ``deployment.multi_instance``, so deploy tooling can
|
||||
#: set the declaration from its replica count instead of editing config.yaml.
|
||||
MULTI_INSTANCE_ENV_VAR = "DEER_FLOW_MULTI_INSTANCE"
|
||||
|
||||
_FALSY = frozenset({"", "0", "false", "no", "off"})
|
||||
|
||||
|
||||
class DeploymentConfig(BaseModel):
|
||||
"""Operator-declared deployment topology."""
|
||||
|
||||
multi_instance: bool = Field(
|
||||
default=False,
|
||||
description=(
|
||||
"Declare that more than one Gateway instance (uvicorn worker, container, or Kubernetes Pod) shares this database. "
|
||||
"GATEWAY_WORKERS counts only one process tree, so replicas > 1 with one worker per Pod are invisible to the startup "
|
||||
"safety gate without this flag. When true, startup requires database.backend='postgres', run_events.backend='db', "
|
||||
"run_ownership.heartbeat_enabled=true, and a Redis stream bridge; it refuses an explicit memory sandbox ownership store, "
|
||||
"process-local browser tools, and an enabled scheduler without scheduler.multi_instance. "
|
||||
"DEER_FLOW_MULTI_INSTANCE=1 declares the same thing from the environment."
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class MultiInstanceDeclaration:
|
||||
"""An explicit multi-instance declaration and the knob that made it.
|
||||
|
||||
Callers branch on ``source`` and compose messages from ``knob`` and
|
||||
``rollback``; the display text is never the thing to inspect.
|
||||
"""
|
||||
|
||||
source: Literal["env", "config"]
|
||||
value: str
|
||||
|
||||
@property
|
||||
def knob(self) -> str:
|
||||
"""The setting as the operator spelled it, for refusal and warning text."""
|
||||
if self.source == "env":
|
||||
return f"{MULTI_INSTANCE_ENV_VAR}={self.value}"
|
||||
return "deployment.multi_instance=true"
|
||||
|
||||
@property
|
||||
def rollback(self) -> str:
|
||||
"""The one step that withdraws this declaration, as a lower-case imperative."""
|
||||
if self.source == "env":
|
||||
return f"unset {MULTI_INSTANCE_ENV_VAR}"
|
||||
return "set deployment.multi_instance=false"
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.knob
|
||||
|
||||
|
||||
def multi_instance_declared_by_env() -> str | None:
|
||||
"""Return the ``DEER_FLOW_MULTI_INSTANCE`` value when it declares a multi-instance deployment.
|
||||
|
||||
Blank, ``0``, ``false``, ``no`` and ``off`` mean "not declared", so a
|
||||
templated ``DEER_FLOW_MULTI_INSTANCE=false`` stays inert. Every other value
|
||||
is a declaration: a typo in this variable must fail closed (run the gate)
|
||||
rather than silently disable it.
|
||||
"""
|
||||
raw = os.environ.get(MULTI_INSTANCE_ENV_VAR)
|
||||
if raw is None:
|
||||
return None
|
||||
value = raw.strip()
|
||||
if value.lower() in _FALSY:
|
||||
return None
|
||||
return value
|
||||
|
||||
|
||||
def multi_instance_declaration(config: Any) -> MultiInstanceDeclaration | None:
|
||||
"""Return the explicit multi-instance declaration in effect, or ``None``.
|
||||
|
||||
The environment is consulted first because deploy tooling that sets it
|
||||
knows the actual topology; ``deployment.multi_instance`` in config.yaml is
|
||||
the operator's manual spelling. The result carries the knob that made the
|
||||
declaration so a refusal message tells the operator what to change.
|
||||
"""
|
||||
env_value = multi_instance_declared_by_env()
|
||||
if env_value is not None:
|
||||
return MultiInstanceDeclaration(source="env", value=env_value)
|
||||
deployment = getattr(config, "deployment", None)
|
||||
if bool(getattr(deployment, "multi_instance", False)):
|
||||
return MultiInstanceDeclaration(source="config", value="true")
|
||||
return None
|
||||
@ -82,6 +82,10 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
|
||||
),
|
||||
"subagent_runtime": ("the shared native-subagent admission controller and isolated execution loop are configured once during Gateway lifespan startup; changing process slots, queue policy, or queue bounds requires a restart."),
|
||||
"subagent_batches": ("the durable subagent batch service is constructed and started once during Gateway lifespan startup; scheduler limits, leases, and recovery behavior are captured by that service instance."),
|
||||
"deployment": (
|
||||
"langgraph_runtime() evaluates deployment.multi_instance once at startup inside the multi-process safety gate (_enforce_postgres_for_multi_worker); "
|
||||
"the gate never re-runs on config.yaml edits, so declaring or withdrawing a multi-instance topology needs a Gateway restart on every instance."
|
||||
),
|
||||
"run_ownership": (
|
||||
"RunOwnershipConfig is captured once into RunManager at langgraph_runtime() startup; the lease heartbeat background task is created and "
|
||||
"started there, and heartbeat_enabled / lease_seconds / grace_seconds are not re-read on config.yaml edits."
|
||||
|
||||
@ -26,6 +26,7 @@ def _stub_app_config(monkeypatch):
|
||||
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "0")
|
||||
monkeypatch.delenv("GATEWAY_WORKERS", raising=False)
|
||||
monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
|
||||
monkeypatch.delenv("DEER_FLOW_MULTI_INSTANCE", raising=False)
|
||||
set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}}))
|
||||
yield
|
||||
reset_app_config()
|
||||
@ -1452,6 +1453,8 @@ def test_wechat_qr_confirmation_preserves_bot_id_unless_replaced(monkeypatch, re
|
||||
{"GATEWAY_WORKERS": "invalid"},
|
||||
# A blank GATEWAY_WORKERS is unset, so WEB_CONCURRENCY still decides.
|
||||
{"GATEWAY_WORKERS": "", "WEB_CONCURRENCY": "2"},
|
||||
# Kubernetes replicas run one worker each and declare their peers instead.
|
||||
{"GATEWAY_WORKERS": "1", "DEER_FLOW_MULTI_INSTANCE": "1"},
|
||||
],
|
||||
)
|
||||
@pytest.mark.parametrize("method,suffix", [("POST", ""), ("POST", "/session/poll"), ("DELETE", "/session")])
|
||||
|
||||
@ -28,6 +28,7 @@ def isolated_worker_env(monkeypatch):
|
||||
"""
|
||||
monkeypatch.delenv("GATEWAY_WORKERS", raising=False)
|
||||
monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
|
||||
monkeypatch.delenv("DEER_FLOW_MULTI_INSTANCE", raising=False)
|
||||
|
||||
|
||||
class _FakeDedupe:
|
||||
@ -107,6 +108,29 @@ def test_factory_warns_when_memory_explicit_under_multi_worker(monkeypatch, capl
|
||||
assert any("dedupe_storage=memory with GATEWAY_WORKERS>1" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
def test_factory_warns_when_memory_explicit_under_a_declared_multi_instance_deployment(monkeypatch, caplog):
|
||||
"""Kubernetes replicas run one worker each and declare their peers instead of a worker count."""
|
||||
import logging
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_MULTI_INSTANCE", "1")
|
||||
app = _FakeApp("memory", "sqlite")
|
||||
with caplog.at_level(logging.WARNING):
|
||||
store = make_inbound_dedupe_store(app)
|
||||
assert isinstance(store, MemoryInboundDedupeStore)
|
||||
assert any("dedupe_storage=memory with DEER_FLOW_MULTI_INSTANCE=1" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
def test_factory_warns_when_auto_resolves_memory_under_a_declared_multi_instance_deployment(monkeypatch, caplog):
|
||||
import logging
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_MULTI_INSTANCE", "true")
|
||||
app = _FakeApp("auto", "sqlite")
|
||||
with caplog.at_level(logging.WARNING):
|
||||
store = make_inbound_dedupe_store(app)
|
||||
assert isinstance(store, MemoryInboundDedupeStore)
|
||||
assert any("Multi-worker deployment detected but dedupe_storage=auto" in r.message for r in caplog.records)
|
||||
|
||||
|
||||
def test_factory_warns_when_auto_resolves_memory_under_multi_worker(monkeypatch, caplog):
|
||||
import logging
|
||||
|
||||
|
||||
@ -21,7 +21,9 @@ from fastapi import FastAPI
|
||||
from app.gateway.deps import _enforce_postgres_for_multi_worker, _validate_agent_storage, langgraph_runtime
|
||||
from app.gateway.routers.browser import _browser_tools_enabled
|
||||
from deerflow.config.database_config import DatabaseConfig
|
||||
from deerflow.config.deployment_config import MULTI_INSTANCE_ENV_VAR, DeploymentConfig, multi_instance_declaration
|
||||
from deerflow.config.run_ownership_config import RunOwnershipConfig
|
||||
from deerflow.config.stream_bridge_config import StreamBridgeConfig
|
||||
|
||||
|
||||
def _config_with_backend(
|
||||
@ -32,18 +34,33 @@ def _config_with_backend(
|
||||
run_events_backend: str = "db",
|
||||
scheduler_enabled: bool = False,
|
||||
scheduler_multi_instance: bool = False,
|
||||
deployment_multi_instance: bool = False,
|
||||
stream_bridge_type: str | None = None,
|
||||
ownership_type: str | None = None,
|
||||
) -> SimpleNamespace:
|
||||
run_ownership = RunOwnershipConfig(heartbeat_enabled=heartbeat_enabled) if heartbeat_enabled is not None else None
|
||||
tools = [SimpleNamespace(name="browser_navigate")] if browser_enabled else []
|
||||
stream_bridge = StreamBridgeConfig(type=stream_bridge_type) if stream_bridge_type is not None else None
|
||||
ownership = SimpleNamespace(type=ownership_type) if ownership_type is not None else None
|
||||
return SimpleNamespace(
|
||||
database=DatabaseConfig(backend=backend),
|
||||
run_ownership=run_ownership,
|
||||
run_events=SimpleNamespace(backend=run_events_backend),
|
||||
scheduler=SimpleNamespace(enabled=scheduler_enabled, multi_instance=scheduler_multi_instance),
|
||||
tools=tools,
|
||||
deployment=DeploymentConfig(multi_instance=deployment_multi_instance),
|
||||
stream_bridge=stream_bridge,
|
||||
sandbox=SimpleNamespace(ownership=ownership),
|
||||
)
|
||||
|
||||
|
||||
def _cluster_ready(**overrides):
|
||||
"""A Postgres deployment with every multi-instance prerequisite satisfied."""
|
||||
kwargs = dict(heartbeat_enabled=True, stream_bridge_type="redis")
|
||||
kwargs.update(overrides)
|
||||
return _config_with_backend("postgres", **kwargs)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def isolated_worker_env(monkeypatch):
|
||||
"""Keep the suite independent of the invoking shell's worker count.
|
||||
@ -56,6 +73,8 @@ def isolated_worker_env(monkeypatch):
|
||||
"""
|
||||
monkeypatch.delenv("GATEWAY_WORKERS", raising=False)
|
||||
monkeypatch.delenv("WEB_CONCURRENCY", raising=False)
|
||||
monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR, raising=False)
|
||||
monkeypatch.delenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", raising=False)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
@ -181,7 +200,7 @@ def test_agent_storage_warning_names_the_variable_that_set_the_count(monkeypatch
|
||||
|
||||
def test_gate_allows_multi_worker_with_postgres_and_heartbeat(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True))
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True, stream_bridge_type="redis"))
|
||||
|
||||
|
||||
def test_gate_rejects_multi_worker_with_scheduler_enabled(monkeypatch):
|
||||
@ -227,6 +246,7 @@ def test_gate_allows_multi_instance_scheduler_with_multiple_workers(monkeypatch)
|
||||
heartbeat_enabled=True,
|
||||
scheduler_enabled=True,
|
||||
scheduler_multi_instance=True,
|
||||
stream_bridge_type="redis",
|
||||
)
|
||||
)
|
||||
|
||||
@ -458,3 +478,232 @@ async def test_langgraph_runtime_invokes_gate_before_persistence_setup(monkeypat
|
||||
init_engine_from_config.assert_not_called()
|
||||
make_stream_bridge.assert_not_called()
|
||||
make_store.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Cross-process stream bridge: multi-worker now needs it too
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_multi_worker_requires_a_cross_process_stream_bridge(monkeypatch):
|
||||
"""GATEWAY_WORKERS > 1 with the memory bridge used to start and then 409 every SSE join on a peer."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
for bridge in (None, "memory"):
|
||||
with pytest.raises(SystemExit, match="stream_bridge.type='redis'"):
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True, stream_bridge_type=bridge))
|
||||
|
||||
|
||||
def test_multi_worker_accepts_the_env_redis_stream_bridge(monkeypatch):
|
||||
"""docker-compose and the Helm chart inject DEER_FLOW_STREAM_BRIDGE_REDIS_URL instead of a config section."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://redis:6379/0")
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("postgres", heartbeat_enabled=True))
|
||||
|
||||
|
||||
def test_multi_worker_rejects_explicit_memory_sandbox_ownership(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
with pytest.raises(SystemExit, match="sandbox.ownership.type='memory'"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(ownership_type="memory"))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Explicit multi-instance declaration: Kubernetes replicas run one worker per Pod
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_declared_multi_instance_is_gated_with_a_single_worker(monkeypatch):
|
||||
"""replicas > 1 with GATEWAY_WORKERS=1 per Pod must not bypass the gate.
|
||||
|
||||
Without the declaration every Pod reports one worker, passes every check,
|
||||
and its startup orphan reconciliation writes the peers' lease-less runs off
|
||||
as crashed on every rolling update.
|
||||
"""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "1")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=True))
|
||||
msg = str(exc_info.value)
|
||||
assert "deployment.multi_instance=true" in msg
|
||||
assert "requires database.backend='postgres'" in msg
|
||||
assert "deployment.multi_instance=false" in msg, "must name the rollback knob"
|
||||
|
||||
|
||||
def test_declared_multi_instance_accepts_a_fully_shared_deployment(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "1")
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True))
|
||||
|
||||
|
||||
def test_declared_multi_instance_requires_heartbeat():
|
||||
with pytest.raises(SystemExit, match="heartbeat_enabled=true"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, heartbeat_enabled=False))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("run_events_backend", ["memory", "jsonl"])
|
||||
def test_declared_multi_instance_requires_db_run_events(run_events_backend):
|
||||
with pytest.raises(SystemExit, match="run_events.backend='db'"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, run_events_backend=run_events_backend))
|
||||
|
||||
|
||||
def test_declared_multi_instance_requires_a_redis_stream_bridge():
|
||||
"""The memory bridge is process-local, so a peer never sees the owner's SSE events."""
|
||||
for bridge in (None, "memory"):
|
||||
with pytest.raises(SystemExit, match="stream_bridge.type='redis'"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, stream_bridge_type=bridge))
|
||||
|
||||
|
||||
def test_declared_multi_instance_accepts_the_env_redis_stream_bridge(monkeypatch):
|
||||
monkeypatch.setenv("DEER_FLOW_STREAM_BRIDGE_REDIS_URL", "redis://redis:6379/0")
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, stream_bridge_type=None))
|
||||
|
||||
|
||||
def test_declared_multi_instance_rejects_explicit_memory_sandbox_ownership():
|
||||
with pytest.raises(SystemExit, match="sandbox.ownership.type='memory'"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, ownership_type="memory"))
|
||||
|
||||
|
||||
def test_declared_multi_instance_accepts_redis_or_inferred_sandbox_ownership():
|
||||
"""An omitted ownership section is inferred from the redis stream bridge (ownership/factory.py)."""
|
||||
for ownership in (None, "redis"):
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, ownership_type=ownership))
|
||||
|
||||
|
||||
def test_declared_multi_instance_rejects_single_instance_scheduler():
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, scheduler_enabled=True))
|
||||
msg = str(exc_info.value)
|
||||
assert "each worker starts its own scheduler" in msg
|
||||
assert "scheduler.multi_instance=true" in msg
|
||||
assert "deployment.multi_instance=false" in msg
|
||||
|
||||
|
||||
def test_declared_multi_instance_accepts_multi_instance_scheduler():
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, scheduler_enabled=True, scheduler_multi_instance=True))
|
||||
|
||||
|
||||
def test_declared_multi_instance_rejects_process_local_browser():
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(deployment_multi_instance=True, browser_enabled=True))
|
||||
msg = str(exc_info.value)
|
||||
assert "process-local" in msg
|
||||
assert "deployment.multi_instance=false" in msg
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["1", "true", "True", "yes", "on", "replicas"])
|
||||
def test_env_declaration_is_gated(monkeypatch, value):
|
||||
"""DEER_FLOW_MULTI_INSTANCE is what deploy tooling sets from the replica count.
|
||||
|
||||
Any spelling that is not an explicit "off" counts: a typo must fail closed
|
||||
(run the gate) rather than silently disable it.
|
||||
"""
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, value)
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite"))
|
||||
msg = str(exc_info.value)
|
||||
assert f"{MULTI_INSTANCE_ENV_VAR}={value}" in msg
|
||||
assert f"Unset {MULTI_INSTANCE_ENV_VAR}" in msg, "must name the rollback knob"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["", "0", "false", "False", "no", "off", " "])
|
||||
def test_falsy_env_declaration_stays_inert(monkeypatch, value):
|
||||
"""A templated DEER_FLOW_MULTI_INSTANCE=false (one replica) must not trip the gate."""
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, value)
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite", scheduler_enabled=True))
|
||||
|
||||
|
||||
def test_env_declaration_accepts_a_fully_shared_deployment(monkeypatch):
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready())
|
||||
|
||||
|
||||
def test_worker_count_is_reported_over_the_declaration(monkeypatch):
|
||||
"""When both apply, the refusal names the worker count: that is the knob uvicorn acts on."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite"))
|
||||
assert "GATEWAY_WORKERS=2" in str(exc_info.value)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("declared_by", "expected_rollback"),
|
||||
[
|
||||
("env", f"Set GATEWAY_WORKERS=1 and unset {MULTI_INSTANCE_ENV_VAR}"),
|
||||
("config", "Set GATEWAY_WORKERS=1 and set deployment.multi_instance=false"),
|
||||
],
|
||||
)
|
||||
def test_worker_count_rollback_also_withdraws_the_declaration(monkeypatch, declared_by, expected_rollback):
|
||||
"""With both knobs active, a rollback that only resets the worker count is not enough.
|
||||
|
||||
Following ``Set GATEWAY_WORKERS=1`` alone leaves the declaration tripping the
|
||||
gate at the next start, so the operator would bounce through a second
|
||||
refusal before learning about the other knob. The message names both.
|
||||
"""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
if declared_by == "env":
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
|
||||
config = _config_with_backend("sqlite", deployment_multi_instance=declared_by == "config")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
assert expected_rollback in str(exc_info.value)
|
||||
|
||||
|
||||
def test_offered_env_rollback_clears_the_gate(monkeypatch):
|
||||
"""The remediation a refusal offers must make the next start succeed."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
|
||||
sqlite = _config_with_backend("sqlite")
|
||||
with pytest.raises(SystemExit):
|
||||
_enforce_postgres_for_multi_worker(sqlite)
|
||||
|
||||
# Resetting only the worker count is the half-step the message must not suggest.
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "1")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(sqlite)
|
||||
assert f"{MULTI_INSTANCE_ENV_VAR}=1" in str(exc_info.value)
|
||||
|
||||
# The full rollback the message offered clears the gate.
|
||||
monkeypatch.delenv(MULTI_INSTANCE_ENV_VAR)
|
||||
_enforce_postgres_for_multi_worker(sqlite)
|
||||
|
||||
|
||||
def test_offered_config_rollback_clears_the_gate(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=True))
|
||||
assert "Set GATEWAY_WORKERS=1 and set deployment.multi_instance=false" in str(exc_info.value)
|
||||
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "1")
|
||||
_enforce_postgres_for_multi_worker(_config_with_backend("sqlite", deployment_multi_instance=False))
|
||||
|
||||
|
||||
def test_browser_refusal_names_the_declaration_when_both_knobs_are_active(monkeypatch):
|
||||
"""The browser refusal must not send the operator through the same double bounce."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "1")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready(browser_enabled=True))
|
||||
msg = str(exc_info.value)
|
||||
assert "browser" in msg
|
||||
assert f"Set GATEWAY_WORKERS=1 and unset {MULTI_INSTANCE_ENV_VAR}" in msg
|
||||
|
||||
|
||||
def test_agent_storage_warning_fires_for_a_declared_multi_instance_deployment(caplog):
|
||||
with caplog.at_level("WARNING"):
|
||||
_validate_agent_storage(_cluster_ready(deployment_multi_instance=True))
|
||||
messages = [r.message for r in caplog.records if "not visible across workers" in r.message]
|
||||
assert messages and "deployment.multi_instance=true" in messages[0]
|
||||
|
||||
|
||||
def test_deployment_declaration_helpers(monkeypatch):
|
||||
assert DeploymentConfig().multi_instance is False
|
||||
assert multi_instance_declaration(None) is None
|
||||
|
||||
by_config = multi_instance_declaration(SimpleNamespace(deployment=DeploymentConfig(multi_instance=True)))
|
||||
assert by_config is not None
|
||||
assert (by_config.source, by_config.knob, by_config.rollback) == ("config", "deployment.multi_instance=true", "set deployment.multi_instance=false")
|
||||
assert str(by_config) == by_config.knob
|
||||
|
||||
# The environment wins over config.yaml and keeps the operator's spelling.
|
||||
monkeypatch.setenv(MULTI_INSTANCE_ENV_VAR, "replicas")
|
||||
by_env = multi_instance_declaration(SimpleNamespace(deployment=DeploymentConfig(multi_instance=True)))
|
||||
assert by_env is not None
|
||||
assert (by_env.source, by_env.knob, by_env.rollback) == ("env", f"{MULTI_INSTANCE_ENV_VAR}=replicas", f"unset {MULTI_INSTANCE_ENV_VAR}")
|
||||
|
||||
@ -20,7 +20,7 @@
|
||||
# ============================================================================
|
||||
# Bump this number when the config schema changes.
|
||||
# Run `make config-upgrade` to merge new fields into your local config.yaml.
|
||||
config_version: 52
|
||||
config_version: 53
|
||||
|
||||
# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
|
||||
# lead_prompt_overlay:
|
||||
@ -2754,6 +2754,27 @@ mcp_tasks:
|
||||
max_result_bytes: 65536
|
||||
result_preview_max_chars: 2000
|
||||
|
||||
# ============================================================================
|
||||
# Deployment Topology
|
||||
# ============================================================================
|
||||
# Declare whether more than one Gateway instance shares this database.
|
||||
#
|
||||
# GATEWAY_WORKERS / WEB_CONCURRENCY only count the uvicorn workers of one
|
||||
# process tree, so a Kubernetes Deployment with replicas > 1 (one worker per
|
||||
# Pod) is invisible to the startup safety gate. Without this declaration each
|
||||
# Pod passes every check, and its startup orphan reconciliation writes the
|
||||
# other Pods' lease-less runs off as crashed on every rolling update.
|
||||
#
|
||||
# With multi_instance: true (or the environment variable
|
||||
# DEER_FLOW_MULTI_INSTANCE=1, which deploy tooling such as a Helm chart can
|
||||
# set from its replica count) startup requires database.backend: postgres, run_events.backend: db,
|
||||
# run_ownership.heartbeat_enabled: true and a Redis stream bridge, and refuses
|
||||
# an explicit sandbox.ownership.type: memory, process-local browser tools, and
|
||||
# scheduler.enabled: true without scheduler.multi_instance: true.
|
||||
# Restart required; apply the same value to every instance.
|
||||
deployment:
|
||||
multi_instance: false
|
||||
|
||||
# ============================================================================
|
||||
# Run Ownership Configuration
|
||||
# ============================================================================
|
||||
|
||||
@ -135,7 +135,7 @@ they resolve from the `secrets` map):
|
||||
|
||||
```yaml
|
||||
config: |
|
||||
config_version: 52
|
||||
config_version: 53
|
||||
models:
|
||||
- name: gpt-4
|
||||
use: langchain_openai:ChatOpenAI
|
||||
|
||||
@ -263,7 +263,7 @@ ingress:
|
||||
# -- DeerFlow config.yaml content. Secrets MUST stay as $VAR references — never
|
||||
# inline literal secret values here. The default enables provisioner sandbox.
|
||||
config: |
|
||||
config_version: 52
|
||||
config_version: 53
|
||||
log_level: info
|
||||
recursion_limit: 100
|
||||
max_recursion_limit: 1000
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user