mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-27 13:40:37 +00:00
feat(channels): add channels.leader_lease and gate an explicitly disabled lease under multi-process signals (#6746)
* feat(channels): add channels.leader_lease and gate it under multi-process signals IM channels other than GitHub connect outbound to their platform. With more than one Gateway instance each one opens that connection with the same bot credentials: Telegram rejects the second poller with 409, Discord answers every message twice and per-process ChannelManager state diverges. The multi-process startup gate said nothing about channels; the only guard was a comment in the Helm values. Add the `channels.leader_lease` section (`enabled: auto|true|false`, `lease_seconds`, `grace_seconds`; startup-only, registered in the reload boundary) that the per-provider leader lease will read, validated by a Pydantic model from the non-schema `channels` mapping. The gate now refuses `enabled: false` under a multi-process signal while a non-GitHub provider is enabled under `channels.*` or connectable through `channel_connections`, naming the providers and the rollback; a malformed block is refused on every topology. `auto` follows the multi-process signal, so a single instance is unchanged. ChannelService strips the block before its per-provider loops and exposes it as `leader_lease`; `ensure_ready_channels` only visits registered providers. The two-Gateway harness overlay leaves the block alone. config_version 60 in the example, the Helm chart and its README; docs replace the comment-only "keep 1 replica" guidance with the gate. * docs(changelog): link #6746 * fix(channels): stop recommending the unimplemented leader lease as a remedy Review on #6746: the refusal told operators to switch channels.leader_lease from false to auto/true "so one instance owns each provider's connection", but the transport arbitration ships separately, so that change only bypassed the clause and every worker still started the same bot. The refusal now offers only the remedies that work today (a single instance, or disabling the channels) and states that auto/true are reserved for the lease that ships separately. A regression helper in the gate tests asserts the remedies start with the rollback and that auto/true are never offered. The config example, Helm values/README, CONFIGURATION.md, README, the deployment guide (en/zh), the model docstrings and both changelogs describe auto/true as reserved settings rather than an ownership guarantee. * fix(channels): warn on an enabled unknown channel key; reword the changelog entry Review on #6746: the registry guard in ensure_ready_channels skipped a misspelled provider key (channels.telegam) silently, where it previously reached _start_channel and logged 'Unknown channel type'. An enabled dict block that is neither a registered provider nor leader_lease now logs a warning that lists the registry. The changelog entry claimed the chart and docs drop the "keep 1 replica" guidance; they keep it (auto/true do not make several replicas safe yet) and only stop relying on the comment alone. * fix(dev): disable UI channel providers in multi-instance overlay --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
Nan Gao
GitHub
parent
f4d10ef42d
commit
d18d2937e7
17
CHANGELOG.md
17
CHANGELOG.md
@ -946,6 +946,22 @@ This release closes that milestone with **621 merged pull requests**.
|
||||
|
||||
#### Channels
|
||||
|
||||
- **channels:** New `channels.leader_lease` section (`enabled: auto|true|false`,
|
||||
`lease_seconds`, `grace_seconds`; startup-only), reserved for the per-provider
|
||||
leader lease that will give one Gateway instance each IM platform connection
|
||||
(the transport ownership itself ships separately; until then keep one
|
||||
instance while channels are enabled). The
|
||||
multi-process startup gate now refuses `enabled: false` while an IM channel
|
||||
other than GitHub is enabled under `channels.*` or connectable through
|
||||
`channel_connections`, naming the providers and only the remedies that work
|
||||
today, and a malformed
|
||||
block is refused on every topology. `auto` follows the multi-process signal, so
|
||||
a single instance is unchanged. `ChannelService` no longer treats the block as
|
||||
a provider, and the two-Gateway harness leaves it alone. `config_version` 60;
|
||||
the Helm chart and docs keep the "keep 1 replica while channels are enabled"
|
||||
guidance but no longer rely on that comment alone: the gate now enforces the
|
||||
disabled-lease case. ([#6746])
|
||||
|
||||
- **channels:** WeChat can be connected by scanning a QR code from the web UI.
|
||||
The connection dialog in the sidebar and Settings gains QR login with
|
||||
progress, pairing-code input, expiry and retry states, and a "Scan again"
|
||||
@ -11591,3 +11607,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#6692]: https://github.com/bytedance/deer-flow/pull/6692
|
||||
[#6702]: https://github.com/bytedance/deer-flow/pull/6702
|
||||
[#6706]: https://github.com/bytedance/deer-flow/pull/6706
|
||||
[#6746]: https://github.com/bytedance/deer-flow/pull/6746
|
||||
|
||||
@ -729,6 +729,17 @@
|
||||
|
||||
#### 渠道
|
||||
|
||||
- **channels:** 新增 `channels.leader_lease` 配置段（`enabled: auto|true|false`、
|
||||
`lease_seconds`、`grace_seconds`；仅启动时读取），为按 provider 的 leader 租约预留，
|
||||
落地后每个 IM 平台连接只由一个 Gateway 实例持有（传输归属本身另行提交，在此之前
|
||||
启用通道时仍需保持单实例）。多进程启动门禁现在会在
|
||||
`channels.*` 启用了 GitHub 以外的 IM 通道、或 `channel_connections` 允许从
|
||||
网页连接时拒绝 `enabled: false`，错误信息列出相关 provider 和当前真正可用的补救方式；格式
|
||||
错误的配置段在任何拓扑下都会拒绝启动。`auto` 跟随多进程信号，单实例行为不变。
|
||||
`ChannelService` 不再把该配置段当作 provider，双 Gateway 联调脚本也不再改动它。
|
||||
`config_version` 升至 60；Helm chart 与文档保留"启用通道时保持 1 副本"的提示，但不再只靠
|
||||
这条注释：门禁现在会对显式关闭租约的情况强制拒绝。 ([#6746])
|
||||
|
||||
- **渠道：** 现在可以在 Web UI 中扫码连接微信。
|
||||
侧边栏与设置中的连接对话框新增二维码登录，带进度、配对码输入、
|
||||
过期与重试状态，以及一个保留未过期绑定指令的“再次扫描”选项；
|
||||
@ -9284,3 +9295,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子
|
||||
[#6692]: https://github.com/bytedance/deer-flow/pull/6692
|
||||
[#6702]: https://github.com/bytedance/deer-flow/pull/6702
|
||||
[#6706]: https://github.com/bytedance/deer-flow/pull/6706
|
||||
[#6746]: https://github.com/bytedance/deer-flow/pull/6746
|
||||
|
||||
@ -541,7 +541,9 @@ Administrators can suspend and restore accounts from **Settings → Users** (adm
|
||||
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. 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. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); the remaining IM channel state (each instance's platform connections, follow-up buffers) still needs its own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness).
|
||||
> 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. Declared instances that store credentials (`channel_connections`) must also share one at-rest `DEER_FLOW_CREDENTIALS_KEY` (the Helm chart and `make up` generate it; back it up, see [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#credentials-encryption-key-deer_flow_credentials_key)). 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. Failed-login counters and lockouts for `POST /api/v1/auth/login/local` (`auth.local.max_login_attempts` / `lockout_seconds`) are kept in the shared `login_throttle` table whenever the application database is SQLite or Postgres (`auth.local.throttle_storage: auto`, the default), so every replica enforces one lockout per client IP; `memory` keeps the historical per-process counter, which under N replicas hands an attacker N × `max_login_attempts` guesses and logs a startup warning. IM chat-to-thread bindings live in the shared `channel_thread_bindings` table whenever the application database is SQLite or Postgres (an existing `channels/store.json` is imported once at startup and renamed `store.json.migrated`); each instance's platform connections still need one owner per provider — `channels.leader_lease` (default `auto`) is reserved for that per-provider leader lease, and until it ships the setting only drives the startup gate, which refuses `enabled: false` while an IM channel other than GitHub is enabled; keep one instance while channels are enabled (see [IM channel leader lease](backend/docs/CONFIGURATION.md#im-channel-leader-lease-channelsleader_lease)); follow-up buffers still need their own multi-worker coordination. To try this topology on one machine, `make dev-multi` starts two Gateways on shared Postgres, Redis and `DEER_FLOW_HOME`, and `make dev-multi-check` runs the cross-instance checks; see the local two-Gateway harness in [backend/docs/CONFIGURATION.md](backend/docs/CONFIGURATION.md#local-two-gateway-harness).
|
||||
>
|
||||
> The local two-Gateway harness disables IM providers under both `channels` and `channel_connections` by default. `DEERFLOW_MI_KEEP_CHANNELS=1` preserves them; `channels.leader_lease` stays as configured.
|
||||
>
|
||||
> During SSE replay-gap recovery, the Web UI preserves a reconnect pointer claimed by another run while durable state is loading. An older recovery cannot overwrite that pointer when it rejoins its own stream. Even a stale foreign pointer is preserved, so the next reload may first check that run before active-run recovery rejoins, adding a reconnect cycle.
|
||||
>
|
||||
|
||||
@ -6,7 +6,7 @@ import asyncio
|
||||
import logging
|
||||
import math
|
||||
import os
|
||||
from collections.abc import Callable
|
||||
from collections.abc import Callable, Mapping
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
@ -15,6 +15,7 @@ from app.channels.manager import DEFAULT_CHANNEL_MAX_CONCURRENCY, DEFAULT_CHANNE
|
||||
from app.channels.message_bus import DEFAULT_INBOUND_QUEUE_MAXSIZE, MessageBus
|
||||
from app.channels.runtime_config_store import merge_runtime_channel_configs
|
||||
from app.channels.store import ChannelStore, SqlChannelStore, resolve_channel_store
|
||||
from deerflow.config.channel_leader_lease_config import CHANNEL_LEADER_LEASE_KEY, ChannelLeaderLeaseConfig, channel_leader_lease_config
|
||||
from deerflow.utils.file_io import await_drained
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -60,6 +61,35 @@ def _channel_has_credentials(name: str, channel_config: dict[str, Any]) -> bool:
|
||||
return any(not isinstance(channel_config.get(key), bool) and channel_config.get(key) is not None and str(channel_config[key]).strip() for key in cred_keys)
|
||||
|
||||
|
||||
def enabled_transport_providers(app_config: Any) -> list[str]:
|
||||
"""Return the providers whose platform connection this deployment may open.
|
||||
|
||||
These are the channels the leader lease arbitrates: every registered
|
||||
provider that is ``enabled`` under ``channels.*`` in config.yaml, plus every
|
||||
provider a user may connect from the web UI (``channel_connections.<p>.enabled``
|
||||
with ``channel_connections.enabled``), whose credentials arrive at runtime
|
||||
and start the transport without a restart. GitHub is excluded: it receives
|
||||
webhooks pushed to whichever instance the load balancer picks and opens no
|
||||
platform connection of its own. ``model_extra`` / ``channel_connections``
|
||||
may be absent on config doubles; both then contribute nothing.
|
||||
"""
|
||||
names: set[str] = set()
|
||||
extra = getattr(app_config, "model_extra", None) or {}
|
||||
channels = extra.get("channels") if isinstance(extra, Mapping) else None
|
||||
if isinstance(channels, Mapping):
|
||||
for name, block in channels.items():
|
||||
if name in _CHANNEL_REGISTRY and isinstance(block, Mapping) and block.get("enabled", False):
|
||||
names.add(name)
|
||||
connections = getattr(app_config, "channel_connections", None)
|
||||
if connections is not None and getattr(connections, "enabled", False):
|
||||
for name in _CHANNEL_REGISTRY:
|
||||
provider = getattr(connections, name, None)
|
||||
if provider is not None and getattr(provider, "enabled", False):
|
||||
names.add(name)
|
||||
names.discard("github")
|
||||
return sorted(names)
|
||||
|
||||
|
||||
def _resolve_service_url(config: dict[str, Any], config_key: str, env_key: str, default: str) -> str:
|
||||
value = config.pop(config_key, None)
|
||||
if isinstance(value, str) and value.strip():
|
||||
@ -132,6 +162,11 @@ class ChannelService:
|
||||
store: ChannelStore | None = None,
|
||||
) -> None:
|
||||
config = dict(channels_config or {})
|
||||
# ``channels.leader_lease`` is a settings block, not a provider: take it
|
||||
# out before the per-provider loops below and in ensure_ready_channels
|
||||
# see it. The startup gate has already validated it; a direct caller
|
||||
# (tests, embedded use) gets the same ValueError on a bad block.
|
||||
self.leader_lease: ChannelLeaderLeaseConfig = channel_leader_lease_config({CHANNEL_LEADER_LEASE_KEY: config.pop(CHANNEL_LEADER_LEASE_KEY, None)})
|
||||
inbound_queue_maxsize = _resolve_positive_int(config, "inbound_queue_maxsize", DEFAULT_INBOUND_QUEUE_MAXSIZE)
|
||||
max_concurrency = _resolve_positive_int(config, "max_concurrency", DEFAULT_CHANNEL_MAX_CONCURRENCY)
|
||||
shutdown_grace_period_seconds = _resolve_non_negative_float(config, "shutdown_grace_period_seconds", DEFAULT_CHANNEL_SHUTDOWN_GRACE_PERIOD_SECONDS)
|
||||
@ -250,6 +285,13 @@ class ChannelService:
|
||||
for name, channel_config in self._config.items():
|
||||
if not isinstance(channel_config, dict):
|
||||
continue
|
||||
if name not in _CHANNEL_REGISTRY:
|
||||
# __init__ already popped the settings blocks (session, leader_lease),
|
||||
# so an enabled dict here is almost certainly a misspelled provider
|
||||
# key; say so instead of skipping it silently.
|
||||
if name != CHANNEL_LEADER_LEASE_KEY and channel_config.get("enabled", False):
|
||||
logger.warning("Ignoring channels.%s: it is enabled but is not a registered provider (%s). Check the key spelling in config.yaml.", name, ", ".join(sorted(_CHANNEL_REGISTRY)))
|
||||
continue
|
||||
if not channel_config.get("enabled", False):
|
||||
if _channel_has_credentials(name, channel_config):
|
||||
logger.warning(
|
||||
|
||||
@ -31,6 +31,7 @@ from langgraph.types import Checkpointer
|
||||
from app.gateway.auth.errors import AuthErrorCode
|
||||
from deerflow.community.browser_automation.session import browser_multi_worker_error
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.config.channel_leader_lease_config import CHANNEL_LEADER_LEASE_FIELD, ChannelLeaderLeaseConfig, ChannelLeaderLeaseConfigError, channel_leader_lease_from_app_config
|
||||
from deerflow.config.credentials_key import CREDENTIALS_KEY_ENV_VAR, CREDENTIALS_KEY_FILENAME, GENERATE_KEY_COMMAND, CredentialsKeyError, parse_credentials_keys
|
||||
from deerflow.config.deployment_config import multi_instance_declaration
|
||||
from deerflow.persistence.feedback import FeedbackRepository
|
||||
@ -160,6 +161,15 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
|
||||
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).
|
||||
8. ``channels.leader_lease.enabled`` must not be ``false`` while an IM
|
||||
channel other than GitHub is enabled: every instance would open the
|
||||
platform connection with the same bot credentials (Telegram rejects
|
||||
the second poller with 409, Discord answers every message twice).
|
||||
``auto`` (the default) and ``true`` pass the clause, but they are reserved
|
||||
configuration for the lease-backed transport ownership that ships
|
||||
separately: in this version they do not stop the duplicate connections,
|
||||
so the refusal offers only the single-instance and channel-disable
|
||||
remedies.
|
||||
|
||||
This gate runs once at startup before any persistence engine is
|
||||
initialised so the error message is clear and the process exits
|
||||
@ -180,6 +190,11 @@ 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.")
|
||||
|
||||
# A malformed channels.leader_lease block is refused on every topology:
|
||||
# ChannelService would otherwise raise the same error later in the
|
||||
# lifespan, after the persistence engine is already up.
|
||||
leader_lease = _channel_leader_lease(config)
|
||||
|
||||
signal = _multi_process_signal(config)
|
||||
if signal is None:
|
||||
return
|
||||
@ -228,6 +243,37 @@ def _enforce_postgres_for_multi_worker(config: AppConfig) -> None:
|
||||
"using (#4206). Set sandbox.ownership.type='redis', or omit the section so it is inferred from the redis stream bridge."
|
||||
)
|
||||
|
||||
if leader_lease.explicitly_disabled:
|
||||
providers = _enabled_channel_transport_providers(config)
|
||||
if providers:
|
||||
raise SystemExit(
|
||||
f"{reason} cannot run with {CHANNEL_LEADER_LEASE_FIELD}.enabled=false while IM channels {', '.join(providers)} are enabled: "
|
||||
"every instance would open the platform connection with the same bot credentials (Telegram rejects the second poller "
|
||||
"with 409 Conflict, Discord answers every message twice, and per-process channel state diverges). "
|
||||
f"{rollback}, or disable those channels (channels.<provider>.enabled=false and channel_connections.<provider>.enabled=false). "
|
||||
f"{CHANNEL_LEADER_LEASE_FIELD}.enabled=auto|true is reserved for the per-provider leader lease that ships separately and does not "
|
||||
"yet stop these duplicate connections, so it is not a remedy here."
|
||||
)
|
||||
|
||||
|
||||
def _channel_leader_lease(config: AppConfig) -> ChannelLeaderLeaseConfig:
|
||||
"""Parse ``channels.leader_lease``; a malformed block exits with the dotted path in the message."""
|
||||
try:
|
||||
return channel_leader_lease_from_app_config(config)
|
||||
except ChannelLeaderLeaseConfigError as exc:
|
||||
raise SystemExit(str(exc)) from None
|
||||
|
||||
|
||||
def _enabled_channel_transport_providers(config: AppConfig) -> list[str]:
|
||||
"""Non-GitHub IM providers whose platform connection this deployment may open.
|
||||
|
||||
Imported lazily: ``app.channels.service`` pulls in every channel module and
|
||||
the gate runs before the lifespan has any reason to load them.
|
||||
"""
|
||||
from app.channels.service import enabled_transport_providers
|
||||
|
||||
return enabled_transport_providers(config)
|
||||
|
||||
|
||||
def credentials_key_consumers(config: AppConfig) -> list[str]:
|
||||
"""Return the enabled features that store data under ``DEER_FLOW_CREDENTIALS_KEY``.
|
||||
|
||||
@ -593,6 +593,7 @@ Notes:
|
||||
- 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.
|
||||
- With the default DeerMem backend, the declaration also warns when the derived SQLite retrieval index sits inside the shared `storage_path` (the default `{storage_path}/.retrieval`). Set `memory.backend_config.retrieval_index_path` to an instance-local directory (relative values resolve against `storage_path`): SQLite WAL is unsupported on network filesystems, and the index is rebuilt from the Markdown facts at startup and re-synced per user scope after a peer writes.
|
||||
- A declared multi-instance deployment that enables a feature storing credentials (today `channel_connections.enabled: true`) also needs one shared `DEER_FLOW_CREDENTIALS_KEY`; see [Credentials encryption key](#credentials-encryption-key-deer_flow_credentials_key).
|
||||
- IM channels: the gate refuses `channels.leader_lease.enabled: false` while a provider other than GitHub is enabled (under `channels.*`, or connectable from the web UI through `channel_connections.<provider>.enabled`), because every instance would otherwise open the platform connection with the same bot credentials. `auto`/`true` pass the clause but are reserved for the lease-backed transport ownership that ships separately, so keep one instance while channels are enabled; see [IM channel leader lease](#im-channel-leader-lease-channelsleader_lease).
|
||||
- Restart-required: the gate runs once at startup. Restart all Gateway instances together after changing it.
|
||||
|
||||
#### Local two-Gateway harness
|
||||
@ -600,7 +601,7 @@ Notes:
|
||||
`scripts/dev_multi_instance.sh` (also `make dev-multi`, `make dev-multi-check`, `make dev-multi-down`) runs this topology on one machine, so cross-instance behavior can be checked without a cluster:
|
||||
|
||||
- Throwaway `postgres:17-alpine` and `redis:7-alpine` containers (`deerflow-mi-postgres`, `deerflow-mi-redis`) on `127.0.0.1:55432` and `127.0.0.1:56379`, Gateway A on `127.0.0.1:8001` and Gateway B on `127.0.0.1:8011` (8002 stays free for the provisioner). When nginx is installed, `http://localhost:2027` round-robins `/api` over both Gateways, names the Gateway that answered in an `X-DeerFlow-Upstream` response header, and forwards `/` to a frontend on port 3000 if you start one. Ports, container names and images are overridable; see `--help`.
|
||||
- The generated `config.yaml` is your own `config.yaml` (otherwise `config.example.yaml`; `DEERFLOW_MI_BASE_CONFIG` picks another file) plus `deployment.multi_instance: true`, `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true` and `stream_bridge.type: redis`. Settings the startup gate refuses are changed and reported: process-local browser tools are removed and an enabled scheduler gets `scheduler.multi_instance: true`. Enabled IM channels are disabled because both Gateways would connect the same bot; `DEERFLOW_MI_KEEP_CHANNELS=1` keeps them.
|
||||
- The generated `config.yaml` is your own `config.yaml` (otherwise `config.example.yaml`; `DEERFLOW_MI_BASE_CONFIG` picks another file) plus `deployment.multi_instance: true`, `database.backend: postgres`, `run_events.backend: db`, `run_ownership.heartbeat_enabled: true` and `stream_bridge.type: redis`. Settings the startup gate refuses are changed and reported: process-local browser tools are removed and an enabled scheduler gets `scheduler.multi_instance: true`. Enabled IM providers under both `channels.*` and the UI-connectable `channel_connections.*` are disabled because both Gateways would connect the same bot; `DEERFLOW_MI_KEEP_CHANNELS=1` keeps them. The `channels.leader_lease` block and connection metadata stay unchanged.
|
||||
- Everything the pair writes stays in the harness, never in your own deployment: any `checkpointer` section is removed (it would win over `database` for checkpoints and the Store), an explicit `sandbox.ownership` and a Redis `database.checkpoint_cache` point at the harness Redis (ownership timing settings are kept), a DeerMem `storage_path` (either spelling) and `blob_storage.backend_config.root` are dropped so their data lands in the harness home, and a local AIO sandbox gets a `container_prefix` derived from the state dir, so the harness never adopts or reaps sandbox containers it did not start. A non-DeerMem memory backend is used as configured, and the harness says so.
|
||||
- Both processes share one `DEER_FLOW_HOME`, one copy of `extensions_config.json` (MCP and skill edits never reach your own file), and generated `AUTH_JWT_SECRET`, `DEER_FLOW_INTERNAL_AUTH_TOKEN` and `DEER_FLOW_CREDENTIALS_KEY` values. Each process keeps its own DeerMem `retrieval_index_path`, which the shared config reads from `$DEERFLOW_MI_RETRIEVAL_INDEX_PATH`.
|
||||
- Gateway B starts only after A answers `/health`. Alembic migrations are serialized by an advisory lock, but the LangGraph checkpointer and store `setup()` that runs next is not, so two cold starts against an empty database could race.
|
||||
@ -631,6 +632,24 @@ These still need a manual pass:
|
||||
- edit a skill on A: B's system-prompt skills section changes within 30 s;
|
||||
- the IM channel gates (one leader per platform, failover after killing the leader, runtime-config changes), once channel leader election lands; run them with `DEERFLOW_MI_KEEP_CHANNELS=1`.
|
||||
|
||||
### IM channel leader lease (`channels.leader_lease`)
|
||||
|
||||
```yaml
|
||||
channels:
|
||||
leader_lease:
|
||||
enabled: auto # auto | true | false
|
||||
lease_seconds: 30
|
||||
grace_seconds: 10
|
||||
```
|
||||
|
||||
Every IM channel except GitHub connects outbound to its platform (Telegram long polling, Discord / Slack / Feishu / DingTalk WebSockets). With more than one Gateway instance each one would open that connection with the same bot credentials: Telegram rejects the second poller with `409 Conflict`, Discord delivers every message to every session so the user is answered twice, and the per-process `ChannelManager` state (the `/model` cache, GitHub follow-up buffers) diverges. The leader lease will arbitrate this: one row per provider in the shared database, held by one instance and renewed on a heartbeat, so only the holder starts that provider's transport. GitHub is webhook-driven and needs no lease. **In this version the section only drives the startup gate**: the transport ownership ships separately, so `auto`/`true` are reserved settings, not an ownership guarantee, and a deployment with channels enabled must still run one instance.
|
||||
|
||||
- `enabled: auto` (default) follows the multi-process signal: on with `GATEWAY_WORKERS` / `WEB_CONCURRENCY > 1`, `deployment.multi_instance: true` or `DEER_FLOW_MULTI_INSTANCE=1`, off on a single instance, which then never writes lease rows and behaves as before. `true` always turns it on; `false` never does, and the multi-process startup gate refuses `false` while an IM channel other than GitHub is enabled. The refusal names the providers and offers only the remedies that work today (a single instance, or disabling those channels); it does not recommend `auto`/`true`, which would merely bypass the clause until the lease ships.
|
||||
- `lease_seconds` (≥ 3, default 30) is how long a held lease stays valid without renewal; the holder renews every `lease_seconds // 3` seconds. `grace_seconds` (≥ 0, default 10) is the extra time after expiry before a peer may take the lease, absorbing a slow renewal. Worst-case failover after a crash is `lease_seconds + grace_seconds`; a cleanly stopped holder releases immediately, so a peer takes over within one heartbeat.
|
||||
- The block is validated on every topology: an unknown key, a non-mapping value, `enabled` outside `auto | true | false`, `lease_seconds < 3` or `grace_seconds < 0` stop startup with the dotted path in the message.
|
||||
- Restart-required (`channels.leader_lease` in the reload-boundary registry): the gate evaluates it once and `ChannelService` captures it at construction. Change it on every instance and restart them together.
|
||||
- The lease-backed transport ownership itself (the `channel_leases` table, the supervising loop and the leader reported by `GET /api/channels/providers`) ships separately; until it lands, `auto`/`true` do not stop every instance from connecting, so keep one instance while channels are enabled. This section fixes the configuration contract and the gate so that rollout needs no further config change.
|
||||
|
||||
### Credentials encryption key (`DEER_FLOW_CREDENTIALS_KEY`)
|
||||
|
||||
DeerFlow encrypts credentials it stores at rest (today: per-connection IM channel credentials in the `channel_credentials` table) with a deployment key read **only** from the environment — there is no `config.yaml` key, and it is never derived from `AUTH_JWT_SECRET`.
|
||||
|
||||
@ -63,7 +63,7 @@ YAML loading and both middleware tool-call paths in `test_tool_output_config_lim
|
||||
|
||||
**Config Hot-Reload Boundary**: Gateway dependencies route through `get_app_config()` on every request, so per-run fields like `models[*].max_tokens`, `summarization.*`, `title.*`, `memory.*`, `subagents.*`, `verification.*`, `tools[*]`, and the agent system prompt pick up `config.yaml` edits on the next message. `AppConfig` is intentionally **not** cached on `app.state` — `lifespan()` keeps a local `startup_config` variable for one-shot bootstrap work and passes it to `langgraph_runtime(app, startup_config)`.
|
||||
|
||||
Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig` or an explicitly registered nested config model, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `plugins`, `database`, `checkpointer`, `run_events`, `agent_storage`, `stream_bridge`, `sandbox`, `skills.container_path`, `log_level`, `logging`, `channels`, `channel_connections`, `scheduler`, `mcp_tasks`, `subagent_runtime`, `subagent_batches`, `run_ownership`, `dedupe_storage`, `auth.local.throttle_storage` (the login throttle store is installed once by `langgraph_runtime`; `max_login_attempts` / `lockout_seconds` stay live-read). Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. `scheduler.recursion_limit` is the exception inside that section: it is read from `get_app_config()` at each scheduled dispatch, so a YAML edit applies to the next run without restarting the poller.
|
||||
Infrastructure fields are **restart-required**. The authoritative list lives in `packages/harness/deerflow/config/reload_boundary.py::STARTUP_ONLY_FIELDS` and is mirrored by the standardised `"startup-only:"` prefix on the corresponding `Field(description=...)` in `AppConfig` or an explicitly registered nested config model, so IDE hover on those fields surfaces the reason inline (no need to context-switch into this table). Currently registered: `plugins`, `database`, `checkpointer`, `run_events`, `agent_storage`, `stream_bridge`, `sandbox`, `skills.container_path`, `log_level`, `logging`, `channels`, `channels.leader_lease` (validated by `channel_leader_lease_config.py` from the non-schema `channels` mapping; the multi-process gate refuses `enabled: false` while a non-GitHub channel is enabled), `channel_connections`, `scheduler`, `mcp_tasks`, `subagent_runtime`, `subagent_batches`, `run_ownership`, `dedupe_storage`, `auth.local.throttle_storage` (the login throttle store is installed once by `langgraph_runtime`; `max_login_attempts` / `lockout_seconds` stay live-read). Adding a new restart-required field requires updating the registry; drift is pinned by `tests/test_reload_boundary.py`. `scheduler.recursion_limit` is the exception inside that section: it is read from `get_app_config()` at each scheduled dispatch, so a YAML edit applies to the next run without restarting the poller.
|
||||
|
||||
**Persistence backend resolution**: the unified `database` section selects the
|
||||
Gateway's LangGraph checkpointer, LangGraph Store, and DeerFlow SQL repositories.
|
||||
|
||||
@ -0,0 +1,112 @@
|
||||
"""``channels.leader_lease``: one Gateway instance per IM platform connection.
|
||||
|
||||
Every IM channel connects outbound to its platform (Telegram long polling,
|
||||
Discord / Slack / Feishu / DingTalk WebSockets). With several Gateway
|
||||
instances each one would open that connection with the same bot credentials:
|
||||
Telegram rejects the second poller with ``409 Conflict``, Discord delivers
|
||||
every message to every session so the user is answered twice, and the
|
||||
per-process ``ChannelManager`` state diverges between instances.
|
||||
|
||||
The leader lease will be the arbitration for that: a row per provider in the
|
||||
shared database, held by one instance and renewed on a heartbeat, so only the
|
||||
holder starts the provider's transport. This module owns the configuration
|
||||
section and the startup gate consumes it; the transport arbitration itself
|
||||
ships separately with ``ChannelService``, so ``auto`` / ``true`` are reserved
|
||||
settings until then and must not be presented as an ownership guarantee.
|
||||
|
||||
``channels`` is not part of the ``AppConfig`` Pydantic schema (its provider
|
||||
blocks carry raw credentials), so this model is validated from the
|
||||
``channels.leader_lease`` mapping by :func:`channel_leader_lease_config`.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from collections.abc import Mapping
|
||||
from typing import Any, Literal
|
||||
|
||||
from pydantic import BaseModel, ConfigDict, Field, ValidationError
|
||||
|
||||
#: The key under ``channels`` that holds this section.
|
||||
CHANNEL_LEADER_LEASE_KEY = "leader_lease"
|
||||
|
||||
#: Dotted path used in refusal text and the reload-boundary registry.
|
||||
CHANNEL_LEADER_LEASE_FIELD = f"channels.{CHANNEL_LEADER_LEASE_KEY}"
|
||||
|
||||
|
||||
class ChannelLeaderLeaseConfig(BaseModel):
|
||||
"""Operator settings for the per-provider IM transport leader lease."""
|
||||
|
||||
# ``extra="forbid"``: a misspelled key here silently keeps every instance
|
||||
# connecting, which is exactly what the section exists to prevent.
|
||||
model_config = ConfigDict(extra="forbid")
|
||||
|
||||
enabled: bool | Literal["auto"] = Field(
|
||||
default="auto",
|
||||
description=(
|
||||
"auto (default) turns the lease on whenever the Gateway sees a multi-process signal (GATEWAY_WORKERS / WEB_CONCURRENCY > 1, "
|
||||
"deployment.multi_instance: true or DEER_FLOW_MULTI_INSTANCE=1) and leaves a single instance untouched; true always turns it on; "
|
||||
"false never does. A multi-process deployment with IM channels enabled refuses to start when this is false. Until the lease-backed "
|
||||
"transport ownership ships, auto and true are reserved settings: they do not yet stop every instance from connecting."
|
||||
),
|
||||
)
|
||||
lease_seconds: int = Field(
|
||||
default=30,
|
||||
ge=3,
|
||||
description="How long a held lease stays valid without renewal; the holder renews every lease_seconds // 3 seconds. Bounds failover after a crash together with grace_seconds.",
|
||||
)
|
||||
grace_seconds: int = Field(
|
||||
default=10,
|
||||
ge=0,
|
||||
description="Extra time after lease_expires_at before a peer may take an expired lease, absorbing a slow renewal. Worst-case failover is lease_seconds + grace_seconds.",
|
||||
)
|
||||
|
||||
@property
|
||||
def explicitly_disabled(self) -> bool:
|
||||
"""True only for a literal ``enabled: false``; ``auto`` is never disabled."""
|
||||
return self.enabled is False
|
||||
|
||||
def resolve(self, *, multi_process: bool) -> bool:
|
||||
"""Return whether the lease is active for this process.
|
||||
|
||||
``auto`` follows the multi-process signal, so a single-instance
|
||||
deployment never writes lease rows and behaves exactly as before.
|
||||
"""
|
||||
if self.enabled == "auto":
|
||||
return multi_process
|
||||
return bool(self.enabled)
|
||||
|
||||
|
||||
class ChannelLeaderLeaseConfigError(ValueError):
|
||||
"""``channels.leader_lease`` is present but not a valid section."""
|
||||
|
||||
|
||||
def channel_leader_lease_config(channels_section: Any) -> ChannelLeaderLeaseConfig:
|
||||
"""Validate the ``leader_lease`` block of a ``channels`` mapping.
|
||||
|
||||
A missing ``channels`` section, a missing ``leader_lease`` key, or an empty
|
||||
block all mean the defaults. Anything else must validate, and the error
|
||||
names the dotted path so an operator can find it in config.yaml.
|
||||
"""
|
||||
raw: Any = None
|
||||
if isinstance(channels_section, Mapping):
|
||||
raw = channels_section.get(CHANNEL_LEADER_LEASE_KEY)
|
||||
if raw is None:
|
||||
return ChannelLeaderLeaseConfig()
|
||||
if not isinstance(raw, Mapping):
|
||||
raise ChannelLeaderLeaseConfigError(f"{CHANNEL_LEADER_LEASE_FIELD} must be a mapping with enabled / lease_seconds / grace_seconds, got {type(raw).__name__}")
|
||||
try:
|
||||
return ChannelLeaderLeaseConfig.model_validate(dict(raw))
|
||||
except ValidationError as exc:
|
||||
problems = "; ".join(f"{'.'.join(str(part) for part in error['loc']) or CHANNEL_LEADER_LEASE_KEY}: {error['msg']}" for error in exc.errors())
|
||||
raise ChannelLeaderLeaseConfigError(f"{CHANNEL_LEADER_LEASE_FIELD} is invalid: {problems}") from None
|
||||
|
||||
|
||||
def channel_leader_lease_from_app_config(app_config: Any) -> ChannelLeaderLeaseConfig:
|
||||
"""Read ``channels.leader_lease`` from an ``AppConfig`` (or a test double).
|
||||
|
||||
``channels`` lives in ``AppConfig.model_extra``; objects without it (the
|
||||
gate tests use ``SimpleNamespace``) get the defaults.
|
||||
"""
|
||||
extra = getattr(app_config, "model_extra", None) or {}
|
||||
channels = extra.get("channels") if isinstance(extra, Mapping) else None
|
||||
return channel_leader_lease_config(channels)
|
||||
@ -67,6 +67,10 @@ STARTUP_ONLY_FIELDS: dict[str, str] = {
|
||||
# startup and the live channel clients are not rebuilt on
|
||||
# config.yaml edits.
|
||||
"channels": ("start_channel_service() is invoked once during startup; the live IM channel clients (Feishu, Slack, Telegram, DingTalk) are not rebuilt when channels.* changes."),
|
||||
"channels.leader_lease": (
|
||||
"the multi-process startup gate (_enforce_postgres_for_multi_worker) evaluates channels.leader_lease once at startup, and ChannelService captures enabled / lease_seconds / grace_seconds "
|
||||
"when it is constructed by start_channel_service(); the lease is not re-read on config.yaml edits, so changing it needs a restart of every Gateway instance together."
|
||||
),
|
||||
"channel_connections": (
|
||||
"start_channel_service() wires the connection repository and channel workers once at startup, and the channel-connections router caches the merged provider config on app.state; "
|
||||
"the scheduled-task notification delivery worker captures channel_connections.notification_locale when it starts. channel_connections.* edits need a restart."
|
||||
|
||||
100
backend/tests/test_channel_leader_lease_config.py
Normal file
100
backend/tests/test_channel_leader_lease_config.py
Normal file
@ -0,0 +1,100 @@
|
||||
"""``channels.leader_lease`` configuration model and its reload-boundary registration."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.config.channel_leader_lease_config import (
|
||||
CHANNEL_LEADER_LEASE_FIELD,
|
||||
ChannelLeaderLeaseConfig,
|
||||
ChannelLeaderLeaseConfigError,
|
||||
channel_leader_lease_config,
|
||||
channel_leader_lease_from_app_config,
|
||||
)
|
||||
from deerflow.config.reload_boundary import STARTUP_ONLY_FIELDS, is_startup_only_field
|
||||
|
||||
|
||||
def test_defaults_are_auto_30_10():
|
||||
config = ChannelLeaderLeaseConfig()
|
||||
assert config.enabled == "auto"
|
||||
assert config.lease_seconds == 30
|
||||
assert config.grace_seconds == 10
|
||||
assert config.explicitly_disabled is False
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("enabled", "multi_process", "expected"),
|
||||
[
|
||||
("auto", False, False),
|
||||
("auto", True, True),
|
||||
(True, False, True),
|
||||
(True, True, True),
|
||||
(False, False, False),
|
||||
(False, True, False),
|
||||
],
|
||||
)
|
||||
def test_auto_follows_the_multi_process_signal(enabled, multi_process, expected):
|
||||
config = ChannelLeaderLeaseConfig(enabled=enabled)
|
||||
assert config.resolve(multi_process=multi_process) is expected
|
||||
|
||||
|
||||
def test_only_a_literal_false_counts_as_explicitly_disabled():
|
||||
assert ChannelLeaderLeaseConfig(enabled=False).explicitly_disabled is True
|
||||
assert ChannelLeaderLeaseConfig(enabled=True).explicitly_disabled is False
|
||||
assert ChannelLeaderLeaseConfig(enabled="auto").explicitly_disabled is False
|
||||
|
||||
|
||||
def test_missing_section_and_empty_block_mean_defaults():
|
||||
assert channel_leader_lease_config(None) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_config({}) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_config({"telegram": {"enabled": True}}) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_config({"leader_lease": None}) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_config({"leader_lease": {}}) == ChannelLeaderLeaseConfig()
|
||||
|
||||
|
||||
def test_block_values_are_read():
|
||||
config = channel_leader_lease_config({"leader_lease": {"enabled": False, "lease_seconds": 15, "grace_seconds": 0}})
|
||||
assert config.enabled is False
|
||||
assert config.lease_seconds == 15
|
||||
assert config.grace_seconds == 0
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("block", "fragment"),
|
||||
[
|
||||
({"enabled": "maybe"}, "enabled"),
|
||||
({"lease_seconds": 0}, "lease_seconds"),
|
||||
({"lease_seconds": 2}, "lease_seconds"),
|
||||
({"grace_seconds": -1}, "grace_seconds"),
|
||||
({"enable": False}, "enable"), # typo: extra keys are refused, not ignored
|
||||
],
|
||||
)
|
||||
def test_invalid_blocks_name_the_dotted_path_and_the_field(block, fragment):
|
||||
with pytest.raises(ChannelLeaderLeaseConfigError) as exc_info:
|
||||
channel_leader_lease_config({"leader_lease": block})
|
||||
message = str(exc_info.value)
|
||||
assert CHANNEL_LEADER_LEASE_FIELD in message
|
||||
assert fragment in message
|
||||
|
||||
|
||||
def test_non_mapping_block_is_refused():
|
||||
with pytest.raises(ChannelLeaderLeaseConfigError, match="must be a mapping"):
|
||||
channel_leader_lease_config({"leader_lease": "auto"})
|
||||
|
||||
|
||||
def test_from_app_config_reads_model_extra_and_tolerates_doubles():
|
||||
app_config = SimpleNamespace(model_extra={"channels": {"leader_lease": {"enabled": True}}})
|
||||
assert channel_leader_lease_from_app_config(app_config).enabled is True
|
||||
# SimpleNamespace without model_extra, and an AppConfig-like object with no channels section.
|
||||
assert channel_leader_lease_from_app_config(SimpleNamespace()) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_from_app_config(SimpleNamespace(model_extra={})) == ChannelLeaderLeaseConfig()
|
||||
assert channel_leader_lease_from_app_config(SimpleNamespace(model_extra=None)) == ChannelLeaderLeaseConfig()
|
||||
|
||||
|
||||
def test_leader_lease_is_registered_as_startup_only():
|
||||
assert is_startup_only_field(CHANNEL_LEADER_LEASE_FIELD)
|
||||
reason = STARTUP_ONLY_FIELDS[CHANNEL_LEADER_LEASE_FIELD]
|
||||
assert "_enforce_postgres_for_multi_worker" in reason
|
||||
assert "ChannelService" in reason
|
||||
103
backend/tests/test_channel_leader_lease_service.py
Normal file
103
backend/tests/test_channel_leader_lease_service.py
Normal file
@ -0,0 +1,103 @@
|
||||
"""``ChannelService`` treats ``channels.leader_lease`` as settings, not as a provider."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from app.channels.service import ChannelService, enabled_transport_providers
|
||||
from deerflow.config.channel_connections_config import ChannelConnectionsConfig
|
||||
from deerflow.config.channel_leader_lease_config import ChannelLeaderLeaseConfig, ChannelLeaderLeaseConfigError
|
||||
|
||||
|
||||
def test_leader_lease_block_is_stripped_and_exposed():
|
||||
service = ChannelService(channels_config={"leader_lease": {"enabled": True, "lease_seconds": 12, "grace_seconds": 3}, "telegram": {"enabled": False}})
|
||||
|
||||
assert service.leader_lease == ChannelLeaderLeaseConfig(enabled=True, lease_seconds=12, grace_seconds=3)
|
||||
assert service._config == {"telegram": {"enabled": False}}
|
||||
assert "leader_lease" not in service.get_status()["channels"]
|
||||
|
||||
|
||||
def test_missing_block_gives_defaults():
|
||||
service = ChannelService(channels_config={})
|
||||
assert service.leader_lease == ChannelLeaderLeaseConfig()
|
||||
assert service.leader_lease.enabled == "auto"
|
||||
|
||||
|
||||
def test_malformed_block_raises_the_config_error():
|
||||
with pytest.raises(ChannelLeaderLeaseConfigError, match="channels.leader_lease"):
|
||||
ChannelService(channels_config={"leader_lease": {"lease_seconds": 0}})
|
||||
|
||||
|
||||
def test_from_app_config_strips_the_block(monkeypatch, tmp_path):
|
||||
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
||||
app_config = SimpleNamespace(model_extra={"channels": {"leader_lease": {"enabled": "auto"}, "telegram": {"enabled": False}}})
|
||||
|
||||
service = ChannelService.from_app_config(app_config)
|
||||
|
||||
assert service._config == {"telegram": {"enabled": False}}
|
||||
assert service.leader_lease.enabled == "auto"
|
||||
|
||||
|
||||
def test_ensure_ready_channels_ignores_non_provider_keys(caplog):
|
||||
"""A settings block whose ``enabled`` is truthy must never be started as a channel.
|
||||
|
||||
A dict block that is neither a registered provider nor ``leader_lease`` is
|
||||
almost certainly a misspelled provider key, so it is skipped with a warning
|
||||
(before the registry guard it reached ``_start_channel`` and logged
|
||||
'Unknown channel type'; silence would hide the typo).
|
||||
"""
|
||||
service = ChannelService(channels_config={"telegram": {"enabled": False}})
|
||||
# Simulate a later configure/overlay that put non-provider dicts back into the live config.
|
||||
service._config["leader_lease"] = {"enabled": "auto"}
|
||||
service._config["telegam"] = {"enabled": True, "bot_token": "x"}
|
||||
service._config["disabled_typo"] = {"enabled": False}
|
||||
service._running = True
|
||||
try:
|
||||
with caplog.at_level("WARNING", logger="app.channels.service"):
|
||||
ready = asyncio.run(service.ensure_ready_channels())
|
||||
finally:
|
||||
service._running = False
|
||||
assert ready == {}
|
||||
warnings = [record.message for record in caplog.records if record.levelname == "WARNING"]
|
||||
assert len(warnings) == 1, warnings
|
||||
assert warnings[0].startswith("Ignoring channels.telegam:")
|
||||
assert "telegram" in warnings[0] # the registry is listed so the operator sees the right spelling
|
||||
assert "leader_lease" not in "\n".join(warnings)
|
||||
|
||||
|
||||
def _app_config(*, channels=None, connections=None):
|
||||
config = SimpleNamespace(model_extra={"channels": channels} if channels is not None else {})
|
||||
if connections is not None:
|
||||
config.channel_connections = ChannelConnectionsConfig.model_validate(connections)
|
||||
return config
|
||||
|
||||
|
||||
def test_enabled_transport_providers_reads_config_yaml_channels():
|
||||
channels = {
|
||||
"langgraph_url": "http://gw:8001/api",
|
||||
"session": {"assistant_id": "lead_agent"},
|
||||
"leader_lease": {"enabled": "auto"},
|
||||
"telegram": {"enabled": True, "bot_token": "x"},
|
||||
"slack": {"enabled": False, "bot_token": "y"},
|
||||
"discord": {"enabled": True},
|
||||
"github": {"enabled": True},
|
||||
}
|
||||
assert enabled_transport_providers(_app_config(channels=channels)) == ["discord", "telegram"]
|
||||
|
||||
|
||||
def test_enabled_transport_providers_includes_ui_connectable_providers():
|
||||
connections = {"enabled": True, "telegram": {"enabled": True, "bot_username": "bot"}, "feishu": {"enabled": True}, "slack": {"enabled": False}}
|
||||
assert enabled_transport_providers(_app_config(connections=connections)) == ["feishu", "telegram"]
|
||||
# The master switch off: nothing can be connected from the UI.
|
||||
connections["enabled"] = False
|
||||
assert enabled_transport_providers(_app_config(connections=connections)) == []
|
||||
|
||||
|
||||
def test_enabled_transport_providers_merges_both_sources_and_tolerates_doubles():
|
||||
config = _app_config(channels={"discord": {"enabled": True}}, connections={"enabled": True, "telegram": {"enabled": True}})
|
||||
assert enabled_transport_providers(config) == ["discord", "telegram"]
|
||||
assert enabled_transport_providers(SimpleNamespace()) == []
|
||||
assert enabled_transport_providers(SimpleNamespace(model_extra=None)) == []
|
||||
@ -141,6 +141,39 @@ def test_overlay_keeps_channels_when_asked():
|
||||
assert config["channels"]["telegram"]["enabled"] is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize("keep_channels", [False, True])
|
||||
def test_overlay_handles_ui_connectable_providers_without_mutating_settings(keep_channels):
|
||||
providers = ("slack", "telegram", "discord", "feishu", "dingtalk", "wechat", "wecom", "buzz", "qq")
|
||||
base = {
|
||||
"channels": {"leader_lease": {"enabled": False, "lease_seconds": 15}},
|
||||
"channel_connections": {
|
||||
"enabled": True,
|
||||
"require_bound_identity": False,
|
||||
"notification_locale": "zh-CN",
|
||||
**{provider: {"enabled": True} for provider in providers},
|
||||
},
|
||||
}
|
||||
base["channel_connections"]["telegram"]["bot_username"] = "test_bot"
|
||||
snapshot = yaml.safe_dump(base)
|
||||
|
||||
config, notes = _overlay(base, keep_channels=keep_channels)
|
||||
|
||||
assert yaml.safe_dump(base) == snapshot
|
||||
assert config["channels"]["leader_lease"] == base["channels"]["leader_lease"]
|
||||
connections = config["channel_connections"]
|
||||
assert connections["enabled"] is True
|
||||
assert connections["require_bound_identity"] is False
|
||||
assert connections["notification_locale"] == "zh-CN"
|
||||
assert connections["telegram"]["bot_username"] == "test_bot"
|
||||
assert all(connections[provider]["enabled"] is keep_channels for provider in providers)
|
||||
channel_notes = [note for note in notes if "channel_connections" in note]
|
||||
if keep_channels:
|
||||
assert channel_notes == []
|
||||
else:
|
||||
assert len(channel_notes) == 1
|
||||
assert all(provider in channel_notes[0] for provider in providers)
|
||||
|
||||
|
||||
def test_overlay_drops_a_postgres_checkpointer_pointing_at_the_developers_database():
|
||||
# An explicit checkpointer section wins over database.postgres_url for LangGraph
|
||||
# checkpoints and the Store, so keeping it would write into the real database.
|
||||
@ -287,6 +320,35 @@ def test_generated_config_from_a_hostile_base_passes_the_real_startup_gate(tmp_p
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("keep_channels", [False, True])
|
||||
@pytest.mark.parametrize("yaml_channel_enabled", [False, True])
|
||||
def test_generated_config_handles_ui_channels_with_disabled_lease(tmp_path, monkeypatch, keep_channels, yaml_channel_enabled):
|
||||
base = {
|
||||
"config_version": 1,
|
||||
"models": [],
|
||||
"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"},
|
||||
"channels": {"leader_lease": {"enabled": False}, "telegram": {"enabled": yaml_channel_enabled, "bot_token": "test-token"}},
|
||||
"channel_connections": {"enabled": True, "telegram": {"enabled": True, "bot_username": "test_bot"}},
|
||||
}
|
||||
base_path = tmp_path / "base.yaml"
|
||||
base_path.write_text(yaml.safe_dump(base), encoding="utf-8")
|
||||
_enforce_postgres_for_multi_worker(AppConfig.from_file(str(base_path)))
|
||||
out = tmp_path / "generated.yaml"
|
||||
monkeypatch.setenv(mi.RETRIEVAL_INDEX_ENV_VAR, str(tmp_path / "index"))
|
||||
|
||||
mi.render_config(base_path, out, database_url=DATABASE_URL, redis_url=REDIS_URL, state_dir=tmp_path / "state", keep_channels=keep_channels)
|
||||
config = AppConfig.from_file(str(out))
|
||||
|
||||
assert config.model_extra["channels"]["leader_lease"] == {"enabled": False}
|
||||
if keep_channels:
|
||||
assert config.channel_connections.telegram.enabled is True
|
||||
with pytest.raises(SystemExit, match="channels.leader_lease.enabled=false while IM channels telegram"):
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
else:
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
assert config.channel_connections.telegram.enabled is False
|
||||
|
||||
|
||||
def test_each_process_resolves_its_own_retrieval_index(tmp_path, monkeypatch):
|
||||
out = tmp_path / "config.yaml"
|
||||
mi.render_config(REPO_ROOT / "config.example.yaml", out, database_url=DATABASE_URL, redis_url=REDIS_URL, state_dir=tmp_path)
|
||||
@ -703,3 +765,14 @@ def test_up_replaces_its_own_stale_containers_with_their_volumes(stubbed_harness
|
||||
assert "docker rm -f -v deerflow-mi-redis" in log[:first_run]
|
||||
labelled = [line for line in log if line.startswith("docker run")]
|
||||
assert all(f"--label deerflow.harness.state-dir={state}" in line for line in labelled), labelled
|
||||
|
||||
|
||||
def test_overlay_leaves_the_channel_leader_lease_block_alone():
|
||||
"""``channels.leader_lease`` is settings (``enabled: auto``), not a bot to disable."""
|
||||
config, notes = _overlay({"channels": {"leader_lease": {"enabled": "auto", "lease_seconds": 15}, "telegram": {"enabled": True}}})
|
||||
|
||||
assert config["channels"]["leader_lease"] == {"enabled": "auto", "lease_seconds": 15}
|
||||
assert config["channels"]["telegram"]["enabled"] is False
|
||||
joined = "\n".join(notes)
|
||||
assert "telegram" in joined
|
||||
assert "leader_lease" not in joined
|
||||
|
||||
@ -953,3 +953,124 @@ async def test_langgraph_runtime_enforces_the_credentials_key_before_persistence
|
||||
|
||||
init_engine_from_config.assert_not_called()
|
||||
make_stream_bridge.assert_not_called()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IM channel leader lease clause (channels.leader_lease)
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _channels_cluster(*, leader_lease=None, channels=None, connections=None, **overrides):
|
||||
"""A fully shared Postgres deployment plus the channel-facing config a gate double needs.
|
||||
|
||||
``channels`` is the raw ``channels`` mapping that lives in ``AppConfig.model_extra``;
|
||||
``connections`` is the ``channel_connections`` section.
|
||||
"""
|
||||
from deerflow.config.channel_connections_config import ChannelConnectionsConfig
|
||||
|
||||
config = _cluster_ready(**overrides)
|
||||
section = dict(channels or {})
|
||||
if leader_lease is not None:
|
||||
section["leader_lease"] = leader_lease
|
||||
config.model_extra = {"channels": section} if section else {}
|
||||
config.channel_connections = ChannelConnectionsConfig.model_validate(connections) if connections is not None else ChannelConnectionsConfig()
|
||||
return config
|
||||
|
||||
|
||||
def test_multi_worker_rejects_disabled_leader_lease_with_an_enabled_channel(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
config = _channels_cluster(leader_lease={"enabled": False}, channels={"telegram": {"enabled": True, "bot_token": "x"}, "discord": {"enabled": True}, "slack": {"enabled": False}})
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
msg = str(exc_info.value)
|
||||
assert "GATEWAY_WORKERS=2" in msg
|
||||
assert "channels.leader_lease.enabled=false" in msg
|
||||
assert "discord, telegram" in msg # sorted, only the enabled ones
|
||||
assert "slack" not in msg
|
||||
assert "409" in msg
|
||||
assert "Set GATEWAY_WORKERS=1" in msg
|
||||
_assert_lease_is_not_offered_as_a_remedy(msg)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("leader_lease", [None, {}, {"enabled": "auto"}, {"enabled": True}, {"enabled": True, "lease_seconds": 15, "grace_seconds": 5}])
|
||||
def test_multi_worker_accepts_auto_or_enabled_leader_lease_with_channels(monkeypatch, leader_lease):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease=leader_lease, channels={"telegram": {"enabled": True}}))
|
||||
|
||||
|
||||
def test_multi_worker_accepts_disabled_leader_lease_without_enabled_channels(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease={"enabled": False}))
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease={"enabled": False}, channels={"telegram": {"enabled": False, "bot_token": "x"}, "langgraph_url": "http://gw:8001/api", "session": {"assistant_id": "lead_agent"}}))
|
||||
|
||||
|
||||
def test_github_is_webhook_driven_and_needs_no_leader_lease(monkeypatch):
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease={"enabled": False}, channels={"github": {"enabled": True}}))
|
||||
|
||||
|
||||
def test_declared_multi_instance_rejects_disabled_leader_lease_for_ui_connectable_providers():
|
||||
"""channel_connections providers get credentials at runtime, so they count even without a channels block."""
|
||||
config = _channels_cluster(
|
||||
deployment_multi_instance=True,
|
||||
leader_lease={"enabled": False},
|
||||
connections={"enabled": True, "telegram": {"enabled": True, "bot_username": "bot"}, "slack": {"enabled": False}},
|
||||
)
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
msg = str(exc_info.value)
|
||||
assert "deployment.multi_instance=true" in msg
|
||||
assert "IM channels telegram are enabled" in msg
|
||||
assert "deployment.multi_instance=false and run a single Gateway instance" in msg
|
||||
assert "channel_connections.<provider>.enabled=false" in msg
|
||||
_assert_lease_is_not_offered_as_a_remedy(msg)
|
||||
|
||||
|
||||
def _assert_lease_is_not_offered_as_a_remedy(msg: str) -> None:
|
||||
"""Transport arbitration ships separately: until then auto/true only bypass the refusal.
|
||||
|
||||
The remedies the refusal may offer are the ones that work today -- a single
|
||||
instance or disabled channels. Recommending ``auto``/``true`` would send an
|
||||
operator straight back into the duplicate connections the clause exists to
|
||||
prevent (review on #6746).
|
||||
"""
|
||||
assert "Set channels.leader_lease.enabled to auto" not in msg
|
||||
assert "so one instance owns" not in msg
|
||||
assert "reserved" in msg and "not a remedy" in msg
|
||||
remedies = msg.split("diverges). ", 1)[1]
|
||||
assert remedies.startswith(("Set GATEWAY_WORKERS=1", "Set deployment.multi_instance=false", "Unset DEER_FLOW_MULTI_INSTANCE")), remedies
|
||||
|
||||
|
||||
def test_disabled_channel_connections_master_switch_does_not_count_providers():
|
||||
config = _channels_cluster(deployment_multi_instance=True, leader_lease={"enabled": False}, connections={"enabled": False, "telegram": {"enabled": True}})
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
|
||||
|
||||
def test_single_instance_ignores_a_disabled_leader_lease():
|
||||
"""Without a multi-process signal the clause is inert, like every other clause of the gate."""
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease={"enabled": False}, channels={"telegram": {"enabled": True}}))
|
||||
|
||||
|
||||
@pytest.mark.parametrize("block", [{"enabled": "maybe"}, {"lease_seconds": 0}, {"grace_seconds": -1}, {"enable": False}, "auto"])
|
||||
def test_malformed_leader_lease_is_refused_on_every_topology(block):
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(_channels_cluster(leader_lease=block))
|
||||
assert "channels.leader_lease" in str(exc_info.value)
|
||||
|
||||
|
||||
def test_leader_lease_clause_runs_after_the_postgres_clause(monkeypatch):
|
||||
"""The lease needs the shared database; a SQLite deployment is told about Postgres first."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
config = _channels_cluster(leader_lease={"enabled": False}, channels={"telegram": {"enabled": True}})
|
||||
config.database = DatabaseConfig(backend="sqlite")
|
||||
with pytest.raises(SystemExit) as exc_info:
|
||||
_enforce_postgres_for_multi_worker(config)
|
||||
msg = str(exc_info.value)
|
||||
assert "postgres" in msg.lower()
|
||||
assert "leader_lease" not in msg
|
||||
|
||||
|
||||
def test_gate_doubles_without_channel_attributes_still_pass(monkeypatch):
|
||||
"""The existing SimpleNamespace doubles carry neither model_extra nor channel_connections."""
|
||||
monkeypatch.setenv("GATEWAY_WORKERS", "2")
|
||||
_enforce_postgres_for_multi_worker(_cluster_ready())
|
||||
|
||||
@ -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: 59
|
||||
config_version: 60
|
||||
|
||||
# Optional operator-owned literal prompt extensions (see CONFIGURATION.md):
|
||||
# lead_prompt_overlay:
|
||||
@ -2886,8 +2886,10 @@ mcp_tasks:
|
||||
# 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.
|
||||
# an explicit sandbox.ownership.type: memory, process-local browser tools,
|
||||
# scheduler.enabled: true without scheduler.multi_instance: true, and
|
||||
# channels.leader_lease.enabled: false while an IM channel other than GitHub
|
||||
# is enabled (see the IM Channels section below).
|
||||
# Restart required; apply the same value to every instance.
|
||||
deployment:
|
||||
multi_instance: false
|
||||
@ -3044,6 +3046,26 @@ run_ownership:
|
||||
# # outer shutdown timeout remains the process-level bound for incomplete cleanup.
|
||||
# shutdown_grace_period_seconds: 3
|
||||
# #
|
||||
# # Leader lease: with more than one Gateway instance, only one instance may
|
||||
# # hold a provider's platform connection (Telegram rejects a second poller
|
||||
# # with 409, Discord answers every message twice). The lease will be a row
|
||||
# # per provider in the shared database, held by one instance and renewed on
|
||||
# # a heartbeat; a crashed holder is replaced after lease_seconds +
|
||||
# # grace_seconds, a cleanly stopped one within a heartbeat. The transport
|
||||
# # ownership itself ships separately: in this version the section only drives
|
||||
# # the startup gate, so keep ONE instance while channels are enabled.
|
||||
# # Restart-required.
|
||||
# leader_lease:
|
||||
# # auto (default): on whenever the Gateway sees a multi-process signal
|
||||
# # (GATEWAY_WORKERS / WEB_CONCURRENCY > 1, deployment.multi_instance: true
|
||||
# # or DEER_FLOW_MULTI_INSTANCE=1), off on a single instance. true: always
|
||||
# # on. false: never -- a multi-process deployment with an IM channel other
|
||||
# # than GitHub enabled then refuses to start. auto/true are reserved until
|
||||
# # the lease ships; they do not yet stop every instance from connecting.
|
||||
# enabled: auto
|
||||
# lease_seconds: 30 # validity of a held lease; renewed every lease_seconds // 3
|
||||
# grace_seconds: 10 # extra time before a peer may take an expired lease
|
||||
# #
|
||||
# # Docker Compose note:
|
||||
# # If channels run inside the gateway container, use container DNS names instead
|
||||
# # of localhost, for example:
|
||||
|
||||
@ -135,7 +135,7 @@ they resolve from the `secrets` map):
|
||||
|
||||
```yaml
|
||||
config: |
|
||||
config_version: 59
|
||||
config_version: 60
|
||||
models:
|
||||
- name: gpt-4
|
||||
use: langchain_openai:ChatOpenAI
|
||||
@ -303,8 +303,13 @@ kubectl -n deer-flow exec deploy/deer-flow-provisioner -- curl -s localhost:8002
|
||||
to drain nodes first. `gateway.replicas: 1` stays the
|
||||
default because the following are still single-instance: **IM channels**
|
||||
(every Pod would connect to every platform — Telegram polling conflicts and
|
||||
Discord double-processes; keep 1 replica while channels are enabled), the
|
||||
WeChat QR login, and browser tools. Before raising replicas, also set
|
||||
Discord double-processes; keep 1 replica while channels are enabled).
|
||||
`channels.leader_lease` in `config` (default `auto`) is reserved for the
|
||||
per-provider leader lease that will give one Pod each platform connection;
|
||||
until that ships the setting only drives the startup gate, which refuses
|
||||
`enabled: false` while channels are enabled, and `auto`/`true` do not yet
|
||||
make several replicas safe. Also single-instance: the WeChat QR login and
|
||||
browser tools. Before raising replicas, also set
|
||||
`persistence.home.accessMode: ReadWriteMany` on multi-node clusters (thread
|
||||
uploads, outputs, memory and `extensions_config.json` live on that volume)
|
||||
and `agent_storage.backend: db` so custom agents are visible on every Pod.
|
||||
|
||||
@ -35,6 +35,10 @@ gateway:
|
||||
# inert (its worker count only sees one Pod). Still single-instance: IM
|
||||
# channels (every Pod connects to every platform -- keep replicas at 1
|
||||
# while channels are enabled), the WeChat QR login, and browser tools.
|
||||
# `channels.leader_lease` in `config` (default `auto`) is reserved for the
|
||||
# per-provider leader lease that will give one Pod each platform connection;
|
||||
# until it ships the setting only drives the startup gate, which refuses
|
||||
# `enabled: false` while channels are enabled (see the chart README).
|
||||
# Multi-node clusters also need `persistence.home.accessMode: ReadWriteMany`,
|
||||
# and `agent_storage.backend: db` keeps custom agents visible on every Pod.
|
||||
# Memory stays on that shared volume, but DeerMem's SQLite retrieval index is
|
||||
@ -364,7 +368,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: 59
|
||||
config_version: 60
|
||||
log_level: info
|
||||
recursion_limit: 100
|
||||
max_recursion_limit: 1000
|
||||
|
||||
@ -156,7 +156,7 @@ The Helm chart ships `gateway.replicas: 1`. Raising it is supported for the web
|
||||
|
||||
The chart exports `DEER_FLOW_MULTI_INSTANCE=true` when `gateway.replicas > 1`, and the Gateway refuses to start if any of these is missing. If you scale with `kubectl scale` or an HPA instead of `gateway.replicas`, set `gateway.multiInstance: true` first so the declaration, the PodDisruptionBudget and the required shared `AUTH_JWT_SECRET` do not keep their single-replica rendering (the PodDisruptionBudget then blocks node drains of the single Pod until the scale-up happens; disable it with `gateway.podDisruptionBudget.enabled: false` for that interim if needed). On multi-node clusters also set `persistence.home.accessMode: ReadWriteMany` (uploads, outputs, memory and `extensions_config.json` live on that volume; MCP and skill changes made through one Pod are then visible to the others on their next request) and `agent_storage.backend: db` so custom agents are visible on every Pod.
|
||||
|
||||
Still single-instance: IM channels (every Pod would connect to every platform), the WeChat QR login, and browser tools. Keep one replica while those are enabled.
|
||||
Still single-instance: IM channels (every Pod would connect to every platform), the WeChat QR login, and browser tools. Keep one replica while those are enabled. `channels.leader_lease` in `config` (default `auto`) is reserved for the per-provider leader lease that will give one Pod each platform connection; until it ships the setting only drives the startup gate, which refuses `enabled: false` while channels are enabled, and `auto`/`true` do not yet make several replicas safe.
|
||||
|
||||
### K8s Provisioner setup
|
||||
|
||||
|
||||
@ -153,7 +153,7 @@ Helm chart 默认 `gateway.replicas: 1`。当所有副本共享同一个 Postgre
|
||||
|
||||
当 `gateway.replicas > 1` 时，chart 会导出 `DEER_FLOW_MULTI_INSTANCE=true`，缺少上述任一项时 Gateway 会拒绝启动。如果你通过 `kubectl scale` 或 HPA 而不是 `gateway.replicas` 扩容，请先设置 `gateway.multiInstance: true`：它们在不执行 Helm upgrade 的情况下改变副本数，否则该声明、PodDisruptionBudget 和必需的共享 `AUTH_JWT_SECRET` 都会停留在单副本的渲染结果上（在真正扩容之前，PodDisruptionBudget 会阻止驱逐唯一的 Pod 和排空节点；如有需要，可在此期间设置 `gateway.podDisruptionBudget.enabled: false`）。多节点集群还需要把 `persistence.home.accessMode` 设为 `ReadWriteMany`（上传、产物、记忆与 `extensions_config.json` 都在该卷上；在一个 Pod 上修改的 MCP 与技能配置会在其他 Pod 的下一次请求中生效），并设置 `agent_storage.backend: db` 让自定义 agent 在每个 Pod 上可见。
|
||||
|
||||
仍然只能单实例运行的部分：IM 通道（每个 Pod 都会连接所有平台）、WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。
|
||||
仍然只能单实例运行的部分：IM 通道（每个 Pod 都会连接所有平台）、WeChat 扫码登录、浏览器工具。启用这些功能时请保持 1 个副本。`config` 中的 `channels.leader_lease`（默认 `auto`）是为按 provider 的 leader 租约预留的配置，落地后每个平台的连接只由一个 Pod 持有；在此之前它只驱动启动门禁（启用通道时拒绝 `enabled: false`），`auto`/`true` 目前并不能让多副本安全运行。
|
||||
|
||||
### K8s Provisioner 配置
|
||||
|
||||
|
||||
@ -174,6 +174,9 @@ Only "no config anywhere" creates `<checkout>/config.yaml` from the example.
|
||||
`dev_multi_instance.{sh,py}` must keep the rendered config passing the real
|
||||
multi-instance gate and every endpoint/data root inside its state dir
|
||||
(`backend/tests/test_dev_multi_instance_script.py`).
|
||||
Unless `keep_channels` is requested, disable provider flags in both `channels`
|
||||
and `channel_connections`; UI credentials can otherwise start a duplicate
|
||||
transport later. Preserve `channels.leader_lease` and connection metadata.
|
||||
|
||||
## Shell Script Invocation Contract
|
||||
|
||||
|
||||
@ -158,13 +158,19 @@ def build_multi_instance_config(
|
||||
sandbox["container_prefix"] = sandbox_container_prefix
|
||||
notes.append(f"set sandbox.container_prefix: {sandbox_container_prefix} (was {previous_prefix!r}) so the harness never adopts or reaps other sandbox containers")
|
||||
|
||||
channels = config.get("channels")
|
||||
if not keep_channels and isinstance(channels, dict):
|
||||
disabled = sorted(name for name, value in channels.items() if isinstance(value, dict) and value.get("enabled"))
|
||||
for name in disabled:
|
||||
channels[name]["enabled"] = False
|
||||
if disabled:
|
||||
notes.append(f"disabled IM channels {', '.join(disabled)}: both Gateways would connect the same bot (set DEERFLOW_MI_KEEP_CHANNELS=1 to keep them)")
|
||||
if not keep_channels:
|
||||
# UI-connectable providers can start transports after credentials arrive,
|
||||
# so disable them as well as the static YAML providers. ``leader_lease``
|
||||
# and the scalar connection settings are not providers.
|
||||
for section_name in ("channels", "channel_connections"):
|
||||
channels = config.get(section_name)
|
||||
if not isinstance(channels, dict):
|
||||
continue
|
||||
disabled = sorted(name for name, value in channels.items() if name != "leader_lease" and isinstance(value, dict) and value.get("enabled"))
|
||||
for name in disabled:
|
||||
channels[name]["enabled"] = False
|
||||
if disabled:
|
||||
notes.append(f"disabled IM channels under {section_name}: {', '.join(disabled)}: both Gateways would connect the same bot (set DEERFLOW_MI_KEEP_CHANNELS=1 to keep them)")
|
||||
|
||||
memory = _section(config, "memory")
|
||||
manager_class = memory.get("manager_class")
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user