mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 05:06:59 +00:00
d18d2937e7
* 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>
104 lines
4.8 KiB
Python
104 lines
4.8 KiB
Python
"""``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)) == []
|