diff --git a/CHANGELOG.md b/CHANGELOG.md index e27e4d2bc8..c9f2bd58b8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5871,6 +5871,13 @@ This release closes that milestone with **621 merged pull requests**. With the default `max_result_chars`, one 100-item page held about 34 MB and blocked the event loop for about 70 ms (170 MB and 300 ms at `limit=500`); it now holds under 1 MB. The response is unchanged. ([#6781]) +- **channels:** Saving or disconnecting a bot from the channel setup page no + longer overwrites bots saved through another Gateway worker. Each worker kept + the `runtime-config.json` contents it loaded at startup and rewrote the whole + file from that copy, so with two workers a Telegram bot saved on one worker + was dropped when a Slack bot was saved on the other, and a disconnected bot + came back. Every save now rereads the file under a lock shared across + processes and changes only its own provider. ([#6844]) ### Security @@ -11651,3 +11658,4 @@ with **180 merged pull requests** since the first 2.0 milestone tag. [#6781]: https://github.com/bytedance/deer-flow/pull/6781 [#6785]: https://github.com/bytedance/deer-flow/pull/6785 [#6798]: https://github.com/bytedance/deer-flow/pull/6798 +[#6844]: https://github.com/bytedance/deer-flow/pull/6844 diff --git a/CHANGELOG_zh.md b/CHANGELOG_zh.md index 05f715664e..3cbb57b292 100644 --- a/CHANGELOG_zh.md +++ b/CHANGELOG_zh.md @@ -4469,6 +4469,11 @@ 每 2 秒轮询一次）不再为了只返回预览而载入每个条目的完整报告、证据 JSON 和提示词。 默认 `max_result_chars` 下，一页 100 个条目约占用 34 MB，并阻塞事件循环约 70 ms （`limit=500` 时为 170 MB 和 300 ms）；现在不到 1 MB。响应内容不变。([#6781]) +- **渠道：** 在渠道设置页保存或断开机器人时，不再覆盖经由其他 Gateway worker + 保存的机器人。每个 worker 都保留启动时读入的 `runtime-config.json` 内容，并用这份 + 副本重写整个文件，因此在两个 worker 下，一个 worker 上保存的 Telegram 机器人会在 + 另一个 worker 保存 Slack 机器人时丢失，已断开的机器人也会重新出现。现在每次保存 + 都会在跨进程共享的锁下重新读取文件，只修改自己的 provider。([#6844]) ### 安全 @@ -9309,3 +9314,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写，核心包含子 [#6746]: https://github.com/bytedance/deer-flow/pull/6746 [#6781]: https://github.com/bytedance/deer-flow/pull/6781 [#6785]: https://github.com/bytedance/deer-flow/pull/6785 +[#6844]: https://github.com/bytedance/deer-flow/pull/6844 diff --git a/backend/app/channels/runtime_config_store.py b/backend/app/channels/runtime_config_store.py index 175eefeb6e..e03db79109 100644 --- a/backend/app/channels/runtime_config_store.py +++ b/backend/app/channels/runtime_config_store.py @@ -4,8 +4,11 @@ from __future__ import annotations import json import logging +import os import tempfile import threading +from collections.abc import Iterator +from contextlib import contextmanager from pathlib import Path from typing import Any @@ -14,6 +17,42 @@ logger = logging.getLogger(__name__) RUNTIME_CHANNEL_DISABLED_FLAG = "_runtime_disabled" +@contextmanager +def _runtime_config_file_lock(path: Path) -> Iterator[None]: + """Exclude read-modify-write cycles of this file in other Gateway processes. + + The lock is an advisory sidecar next to the store file, so it covers every + worker and embedded client that resolves the same ``DEER_FLOW_HOME``. The + path is normalized first, as in ``extensions_config_file_lock``, so a + symlinked spelling of the file shares the lock of the file it points to. + """ + target_path = path.resolve(strict=False) if path.is_symlink() else path.absolute() + lock_path = target_path.parent / f".{target_path.name}.lock" + with open(lock_path, "a+b") as lock_file: + if os.name == "nt": + import msvcrt + + lock_file.seek(0, os.SEEK_END) + if lock_file.tell() == 0: + lock_file.write(b"\0") + lock_file.flush() + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_LOCK, 1) + else: + import fcntl + + fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX) + + try: + yield + finally: + if os.name == "nt": + lock_file.seek(0) + msvcrt.locking(lock_file.fileno(), msvcrt.LK_UNLCK, 1) + else: + fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN) + + class ChannelRuntimeConfigStore: """JSON-backed store for channel credentials entered from the UI. @@ -32,18 +71,39 @@ class ChannelRuntimeConfigStore: self._data: dict[str, dict[str, Any]] = self._load() self._lock = threading.Lock() - def _load(self) -> dict[str, dict[str, Any]]: + def _load(self, fallback: dict[str, dict[str, Any]] | None = None) -> dict[str, dict[str, Any]]: if self._path.exists(): try: raw = json.loads(self._path.read_text(encoding="utf-8")) except (json.JSONDecodeError, OSError): + if fallback is not None: + logger.warning("Corrupt channel runtime config store at %s, rewriting it from this process's copy", self._path) + return {name: dict(config) for name, config in fallback.items()} logger.warning("Corrupt channel runtime config store at %s, starting fresh", self._path) return {} if isinstance(raw, dict): return {str(name): dict(value) for name, value in raw.items() if isinstance(value, dict)} return {} - def _save(self) -> None: + @contextmanager + def _mutation(self) -> Iterator[dict[str, dict[str, Any]]]: + """Yield the file's current providers for one read-modify-write cycle. + + Each Gateway worker keeps its own store, loaded when that worker first + needed it. Saving that snapshot would overwrite providers a peer worker + saved since, so every mutation reloads the file while holding the + thread lock and the cross-process file lock, changes one provider, and + saves. An unreadable file falls back to this process's copy rather + than an empty one. The snapshot this process serves is refreshed only + after the cycle completes; a failed save leaves the previous one in + place. + """ + with self._lock, _runtime_config_file_lock(self._path): + data = self._load(fallback=self._data) + yield data + self._data = data + + def _save(self, data: dict[str, dict[str, Any]]) -> None: fd = tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", @@ -56,7 +116,7 @@ class ChannelRuntimeConfigStore: Path(fd.name).chmod(0o600) except OSError: logger.debug("Unable to chmod temporary channel runtime config store at %s", fd.name, exc_info=True) - json.dump(self._data, fd, indent=2, ensure_ascii=False) + json.dump(data, fd, indent=2, ensure_ascii=False) fd.close() Path(fd.name).replace(self._path) try: @@ -78,24 +138,24 @@ class ChannelRuntimeConfigStore: return dict(config) if isinstance(config, dict) else None def set_provider_config(self, provider: str, config: dict[str, Any]) -> None: - with self._lock: - self._data[provider] = dict(config) - self._save() + with self._mutation() as data: + data[provider] = dict(config) + self._save(data) def set_provider_disconnected(self, provider: str) -> None: - with self._lock: - self._data[provider] = { + with self._mutation() as data: + data[provider] = { "enabled": False, RUNTIME_CHANNEL_DISABLED_FLAG: True, } - self._save() + self._save(data) def remove_provider_config(self, provider: str) -> bool: - with self._lock: - if provider not in self._data: + with self._mutation() as data: + if provider not in data: return False - del self._data[provider] - self._save() + del data[provider] + self._save(data) return True diff --git a/backend/docs/IM_CHANNEL_CONNECTIONS.md b/backend/docs/IM_CHANNEL_CONNECTIONS.md index 80125d6e0a..2c7fbe062b 100644 --- a/backend/docs/IM_CHANNEL_CONNECTIONS.md +++ b/backend/docs/IM_CHANNEL_CONNECTIONS.md @@ -617,7 +617,11 @@ from the browser runtime setup flow, which persists them through `ChannelRuntimeConfigStore` so local/private deployments can configure bots without editing YAML. The runtime store is a local plaintext JSON fallback with owner-only file permissions (`0600`); use it only where the DeerFlow data -directory is already trusted as secret storage. WeChat QR login auth state +directory is already trusted as secret storage. Every Gateway worker keeps its own copy +of the file; a save or disconnect rereads it under an advisory sidecar lock +(`.runtime-config.json.lock`) and changes only that provider, so workers sharing +`DEER_FLOW_HOME` do not overwrite each other's bots. Reads still serve the copy +the worker last loaded or wrote. WeChat QR login auth state follows the same local-runtime model and may persist a QR-derived bot token in the channel state directory. diff --git a/backend/tests/test_channel_connections_router.py b/backend/tests/test_channel_connections_router.py index c0683d9639..f6fac5c0ae 100644 --- a/backend/tests/test_channel_connections_router.py +++ b/backend/tests/test_channel_connections_router.py @@ -836,6 +836,66 @@ def test_configure_provider_runtime_credentials_survive_local_restart(tmp_path): anyio.run(repo.close) +def test_runtime_config_saved_on_peer_workers_both_survive_restart(tmp_path): + """Each Gateway worker keeps its own store; a save must not drop a peer's provider.""" + import anyio + + repo = anyio.run(_make_repo, tmp_path) + config = _enabled_connections_config() + runtime_config_path = tmp_path / "channels" / "runtime-config.json" + # Both workers load the (empty) file before either one saves. + worker_a = _make_app(config, repo, {}, runtime_config_store=ChannelRuntimeConfigStore(runtime_config_path)) + worker_b = _make_app(config, repo, {}, runtime_config_store=ChannelRuntimeConfigStore(runtime_config_path)) + + with TestClient(worker_a) as client: + telegram_response = client.post( + "/api/channels/telegram/runtime-config", + json={"values": {"bot_token": "tg-token", "bot_username": "ui_bot"}}, + ) + with TestClient(worker_b) as client: + slack_response = client.post( + "/api/channels/slack/runtime-config", + json={"values": {"bot_token": "xoxb-ui", "app_token": "xapp-ui"}}, + ) + + assert telegram_response.status_code == 200 + assert slack_response.status_code == 200 + assert ChannelRuntimeConfigStore(runtime_config_path).load_all() == { + "telegram": {"enabled": True, "bot_token": "tg-token", "bot_username": "ui_bot"}, + "slack": {"enabled": True, "bot_token": "xoxb-ui", "app_token": "xapp-ui"}, + } + + anyio.run(repo.close) + + +def test_runtime_disconnect_on_one_worker_is_not_undone_by_a_peer_save(tmp_path): + import anyio + + repo = anyio.run(_make_repo, tmp_path) + config = _enabled_connections_config() + runtime_config_path = tmp_path / "channels" / "runtime-config.json" + slack_config = {"enabled": True, "bot_token": "xoxb-ui", "app_token": "xapp-ui"} + ChannelRuntimeConfigStore(runtime_config_path).set_provider_config("slack", slack_config) + worker_a = _make_app(config, repo, {"slack": dict(slack_config)}, runtime_config_store=ChannelRuntimeConfigStore(runtime_config_path)) + worker_b = _make_app(config, repo, {"slack": dict(slack_config)}, runtime_config_store=ChannelRuntimeConfigStore(runtime_config_path)) + + with TestClient(worker_a) as client: + disconnect_response = client.delete("/api/channels/slack/runtime-config") + with TestClient(worker_b) as client: + telegram_response = client.post( + "/api/channels/telegram/runtime-config", + json={"values": {"bot_token": "tg-token", "bot_username": "ui_bot"}}, + ) + + assert disconnect_response.status_code == 200 + assert telegram_response.status_code == 200 + persisted = ChannelRuntimeConfigStore(runtime_config_path).load_all() + assert persisted["slack"] == {"enabled": False, "_runtime_disabled": True} + assert persisted["telegram"]["bot_token"] == "tg-token" + + anyio.run(repo.close) + + def test_configure_provider_runtime_credentials_preserves_masked_secrets(tmp_path): import anyio diff --git a/backend/tests/test_channel_runtime_config_store_peers.py b/backend/tests/test_channel_runtime_config_store_peers.py new file mode 100644 index 0000000000..81c1ebe565 --- /dev/null +++ b/backend/tests/test_channel_runtime_config_store_peers.py @@ -0,0 +1,175 @@ +"""Runtime channel config saves must merge with peer Gateway workers' saves. + +Every Gateway worker builds its own ``ChannelRuntimeConfigStore`` over the same +``runtime-config.json``. Two store instances on one path stand in for two +workers here; the advisory file lock is per open file, so it also excludes the +two instances inside one test process. +""" + +from __future__ import annotations + +import os +import threading + +import pytest + +from app.channels.runtime_config_store import ChannelRuntimeConfigStore + +TELEGRAM = {"enabled": True, "bot_token": "tg-token"} +SLACK = {"enabled": True, "bot_token": "xoxb-ui", "app_token": "xapp-ui"} + + +def _path(tmp_path): + return tmp_path / "channels" / "runtime-config.json" + + +def test_save_keeps_a_provider_a_peer_saved_after_this_store_loaded(tmp_path): + worker_a = ChannelRuntimeConfigStore(_path(tmp_path)) + worker_b = ChannelRuntimeConfigStore(_path(tmp_path)) + + worker_a.set_provider_config("telegram", TELEGRAM) + worker_b.set_provider_config("slack", SLACK) + + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == {"telegram": TELEGRAM, "slack": SLACK} + # The saving worker's own snapshot also picks up the peer's provider. + assert worker_b.get_provider_config("telegram") == TELEGRAM + + +def test_disconnect_survives_a_later_peer_save(tmp_path): + ChannelRuntimeConfigStore(_path(tmp_path)).set_provider_config("slack", SLACK) + worker_a = ChannelRuntimeConfigStore(_path(tmp_path)) + worker_b = ChannelRuntimeConfigStore(_path(tmp_path)) + + worker_a.set_provider_disconnected("slack") + worker_b.set_provider_config("telegram", TELEGRAM) + + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == { + "slack": {"enabled": False, "_runtime_disabled": True}, + "telegram": TELEGRAM, + } + + +def test_remove_sees_a_provider_a_peer_saved_after_this_store_loaded(tmp_path): + worker_a = ChannelRuntimeConfigStore(_path(tmp_path)) + worker_b = ChannelRuntimeConfigStore(_path(tmp_path)) + worker_b.set_provider_config("slack", SLACK) + + assert worker_a.remove_provider_config("slack") is True + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == {} + + +def test_save_over_a_corrupt_file_keeps_this_process_copy(tmp_path): + store = ChannelRuntimeConfigStore(_path(tmp_path)) + store.set_provider_config("telegram", TELEGRAM) + _path(tmp_path).write_text("{not json", encoding="utf-8") + + store.set_provider_config("slack", SLACK) + + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == {"telegram": TELEGRAM, "slack": SLACK} + + +def test_failed_save_leaves_the_served_snapshot_and_file_unchanged(tmp_path, monkeypatch): + store = ChannelRuntimeConfigStore(_path(tmp_path)) + store.set_provider_config("telegram", TELEGRAM) + + def failing_save(*_args): + raise OSError("disk full") + + monkeypatch.setattr(store, "_save", failing_save) + with pytest.raises(OSError, match="disk full"): + store.set_provider_config("slack", SLACK) + + assert store.load_all() == {"telegram": TELEGRAM} + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == {"telegram": TELEGRAM} + + +def test_peer_mutation_waits_for_the_file_lock_before_reading(tmp_path): + """The reload must sit inside the cross-process lock, not just before the write. + + Worker A stalls between reading and saving. Worker B's mutation must not + read the file until A has saved; reading earlier would let A's save + overwrite B's provider, or B's overwrite A's. + """ + worker_a = ChannelRuntimeConfigStore(_path(tmp_path)) + worker_b = ChannelRuntimeConfigStore(_path(tmp_path)) + a_saving = threading.Event() + release_a = threading.Event() + b_loaded = threading.Event() + + real_save_a = worker_a._save + + def stalled_save(data): + a_saving.set() + assert release_a.wait(timeout=5) + real_save_a(data) + + real_load_b = worker_b._load + + def observed_load(*args, **kwargs): + b_loaded.set() + return real_load_b(*args, **kwargs) + + worker_a._save = stalled_save + worker_b._load = observed_load + + thread_a = threading.Thread(target=worker_a.set_provider_config, args=("telegram", TELEGRAM)) + thread_b = threading.Thread(target=worker_b.set_provider_config, args=("slack", SLACK)) + thread_a.start() + try: + assert a_saving.wait(timeout=5) + thread_b.start() + assert not b_loaded.wait(timeout=0.5) + finally: + release_a.set() + thread_a.join(timeout=5) + if thread_b.ident is not None: + thread_b.join(timeout=5) + + assert b_loaded.is_set() + assert ChannelRuntimeConfigStore(_path(tmp_path)).load_all() == {"telegram": TELEGRAM, "slack": SLACK} + + +@pytest.mark.skipif(os.name == "nt", reason="creating symlinks needs extra privileges on Windows") +def test_symlinked_store_path_shares_the_lock_of_its_target(tmp_path): + """A worker configured with a symlinked spelling of the file must still exclude peers.""" + target = _path(tmp_path) + target.parent.mkdir(parents=True) + target.write_text("{}", encoding="utf-8") + link = target.parent / "runtime-config-link.json" + link.symlink_to(target) + via_link = ChannelRuntimeConfigStore(link) + via_target = ChannelRuntimeConfigStore(target) + link_saving = threading.Event() + release_link = threading.Event() + target_loaded = threading.Event() + + real_save = via_link._save + + def stalled_save(data): + link_saving.set() + assert release_link.wait(timeout=5) + real_save(data) + + real_load = via_target._load + + def observed_load(*args, **kwargs): + target_loaded.set() + return real_load(*args, **kwargs) + + via_link._save = stalled_save + via_target._load = observed_load + + thread_link = threading.Thread(target=via_link.set_provider_config, args=("telegram", TELEGRAM)) + thread_target = threading.Thread(target=via_target.set_provider_config, args=("slack", SLACK)) + thread_link.start() + try: + assert link_saving.wait(timeout=5) + thread_target.start() + assert not target_loaded.wait(timeout=0.5) + finally: + release_link.set() + thread_link.join(timeout=5) + if thread_target.ident is not None: + thread_target.join(timeout=5) + + assert target_loaded.is_set()