From 7a57022125469bd9c8b717eb3c3406b28fdb1d93 Mon Sep 17 00:00:00 2001 From: Hyeonsang Cho Date: Wed, 14 Oct 2026 11:21:02 +0900 Subject: [PATCH] fix(channels): merge runtime channel config saves across Gateway workers (#6844) * fix(channels): merge runtime channel config saves across Gateway workers Each Gateway worker builds its own ChannelRuntimeConfigStore, loads runtime-config.json once, and rewrote the whole file from that copy on every save. With two workers, a bot saved on one worker was dropped when another bot was saved on the other, and a disconnected bot came back. Every save, disconnect and removal now rereads the file while holding the thread lock and an advisory sidecar file lock shared across processes, changes only its own provider, and saves. The served snapshot is refreshed only after a successful save, and an unreadable file still falls back to this process's copy as before. * docs(changelog): reference #6844 for the runtime channel config fix * fix(channels): normalize the runtime config lock path like extensions_config The sidecar lock was derived from the store path as given, so a worker configured with a symlinked spelling of runtime-config.json took a different lock file and no longer excluded peers using the real path. Resolve a symlinked path, and make any other path absolute, before deriving the sidecar, as extensions_config_file_lock already does. --------- Co-authored-by: Willem Jiang --- CHANGELOG.md | 8 + CHANGELOG_zh.md | 6 + backend/app/channels/runtime_config_store.py | 86 +++++++-- backend/docs/IM_CHANNEL_CONNECTIONS.md | 6 +- .../tests/test_channel_connections_router.py | 60 ++++++ ...test_channel_runtime_config_store_peers.py | 175 ++++++++++++++++++ 6 files changed, 327 insertions(+), 14 deletions(-) create mode 100644 backend/tests/test_channel_runtime_config_store_peers.py 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()