From 3f1438d4a0037f4518b7f0f31ab103b7f253dd19 Mon Sep 17 00:00:00 2001 From: llb555 <3103086905@qq.com> Date: Fri, 30 Oct 2026 14:54:41 +0800 Subject: [PATCH] fix(channels): publish Discord and WeChat JSON state atomically (#7595) * fix(channels): publish Discord and WeChat JSON state atomically * fix(channels): adopt the shared atomic publisher for both state writers * test(channels): pin the cleanup-failure direction too * test(channels): keep a platform-independent guard on the publish call shape * style(tests): apply ruff format to the channel encoding test --- backend/app/channels/discord.py | 6 +- backend/app/channels/wechat.py | 6 +- .../test_channel_state_atomic_publish.py | 148 ++++++++++++++++++ backend/tests/test_channel_state_encoding.py | 66 +++++++- 4 files changed, 222 insertions(+), 4 deletions(-) create mode 100644 backend/tests/test_channel_state_atomic_publish.py diff --git a/backend/app/channels/discord.py b/backend/app/channels/discord.py index b0450e3a66..01ddff9941 100644 --- a/backend/app/channels/discord.py +++ b/backend/app/channels/discord.py @@ -28,6 +28,7 @@ from app.channels.message_bus import ( OutboundMessage, ResolvedAttachment, ) +from deerflow.utils.atomic_write import publish_bytes_atomically from deerflow.utils.file_io import await_drained logger = logging.getLogger(__name__) @@ -275,7 +276,10 @@ class DiscordChannel(Channel): try: snapshot = dict(self._active_threads) self._thread_store_path.parent.mkdir(parents=True, exist_ok=True) - self._thread_store_path.write_text(json.dumps(snapshot, indent=2), encoding="utf-8", newline="") + # Publish through the shared helper: it fsyncs the staged bytes + # before the rename (so a machine crash cannot leave an empty + # mapping behind) and re-applies an existing file's mode. + publish_bytes_atomically(self._thread_store_path, json.dumps(snapshot, indent=2).encode("utf-8")) except Exception: logger.exception("[Discord] failed to persist thread mappings") diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index 9200413f4f..09643fa748 100644 --- a/backend/app/channels/wechat.py +++ b/backend/app/channels/wechat.py @@ -32,6 +32,7 @@ from app.channels.commands import is_known_channel_command from app.channels.connection_identity import attach_connection_identity from app.channels.credential_staging import cleanup_stale_credential_staging, credential_staging_prefix from app.channels.message_bus import INBOUND_FILE_CLEANUP_KEY, InboundMessage, InboundMessageType, MessageBus, OutboundMessage, ResolvedAttachment, cleanup_inbound_files +from deerflow.utils.atomic_write import publish_bytes_atomically from deerflow.utils.file_io import await_drained logger = logging.getLogger(__name__) @@ -1707,7 +1708,10 @@ class WechatChannel(Channel): return try: self._cursor_path.parent.mkdir(parents=True, exist_ok=True) - self._cursor_path.write_text(json.dumps({"get_updates_buf": self._get_updates_buf}, ensure_ascii=False, indent=2), encoding="utf-8", newline="") + # Publish through the shared helper: it fsyncs the staged bytes + # before the rename (so a machine crash cannot leave a truncated + # cursor behind) and re-applies an existing file's mode. + publish_bytes_atomically(self._cursor_path, json.dumps({"get_updates_buf": self._get_updates_buf}, ensure_ascii=False, indent=2).encode("utf-8")) except OSError: logger.warning("[WeChat] failed to persist cursor state to %s", self._cursor_path) diff --git a/backend/tests/test_channel_state_atomic_publish.py b/backend/tests/test_channel_state_atomic_publish.py new file mode 100644 index 0000000000..c2e71ed32c --- /dev/null +++ b/backend/tests/test_channel_state_atomic_publish.py @@ -0,0 +1,148 @@ +"""Channel JSON state must be published whole, never truncated in place. + +``DiscordChannel._persist_thread_mappings`` and ``WechatChannel._save_state`` +write state a crash could otherwise truncate: the Discord mapping would fail to +load and the WeChat cursor buffer would be lost, re-reading from the start. Both +now stage a sibling and replace it, and a stage that cannot be discarded is +reported instead of replacing the outcome. +""" + +from __future__ import annotations + +import json +import threading +from pathlib import Path + +from app.channels.discord import DiscordChannel +from app.channels.wechat import WechatChannel + + +def _discord(tmp_path): + channel = DiscordChannel.__new__(DiscordChannel) + channel._thread_store_lock = threading.Lock() + channel._active_threads = {"a": {"thread_id": "t1"}} + channel._thread_store_path = tmp_path / "state" / "discord-threads.json" + return channel + + +def _wechat(tmp_path): + channel = WechatChannel.__new__(WechatChannel) + channel._cursor_path = tmp_path / "state" / "wechat-cursor.json" + channel._get_updates_buf = "cursor-1" + return channel + + +def _fail_staged_write(monkeypatch): + """Fail the staged write the way a full disk would, before any rename. + + With the shared helper this is the earliest injectable point: the bytes are + written to a sibling, fsync fails, and the rename never happens. The point of + the test is that the live file is untouched by that failure. + """ + + def failing_fsync(_fd): + raise OSError("disk full") + + monkeypatch.setattr("deerflow.utils.atomic_write.os.fsync", failing_fsync) + + +def _no_sidecar(target: Path) -> bool: + prefix = f".{target.name}." + return [p.name for p in target.parent.iterdir() if p.name.startswith(prefix)] == [] + + +def test_discord_state_is_replaced_whole(tmp_path): + channel = _discord(tmp_path) + target = tmp_path / "state" / "discord-threads.json" + + channel._persist_thread_mappings() + + assert json.loads(target.read_text(encoding="utf-8")) == {"a": {"thread_id": "t1"}} + assert _no_sidecar(target) + + +def test_discord_failed_staged_write_keeps_the_previous_mapping(tmp_path, monkeypatch): + channel = _discord(tmp_path) + target = tmp_path / "state" / "discord-threads.json" + target.parent.mkdir(parents=True) + target.write_text('{"kept": true}', encoding="utf-8") + _fail_staged_write(monkeypatch) + + channel._persist_thread_mappings() + + assert target.read_text(encoding="utf-8") == '{"kept": true}' + assert json.loads(target.read_text(encoding="utf-8")) == {"kept": True} + assert _no_sidecar(target) + + +def test_wechat_state_is_replaced_whole(tmp_path): + channel = _wechat(tmp_path) + target = tmp_path / "state" / "wechat-cursor.json" + + channel._save_state() + + assert json.loads(target.read_text(encoding="utf-8")) == {"get_updates_buf": "cursor-1"} + assert _no_sidecar(target) + + +def test_wechat_failed_staged_write_keeps_the_previous_cursor(tmp_path, monkeypatch): + channel = _wechat(tmp_path) + target = tmp_path / "state" / "wechat-cursor.json" + target.parent.mkdir(parents=True) + target.write_text('{"get_updates_buf": "kept"}', encoding="utf-8") + _fail_staged_write(monkeypatch) + + channel._save_state() + + assert json.loads(target.read_text(encoding="utf-8")) == {"get_updates_buf": "kept"} + assert _no_sidecar(target) + + +def _fail_staged_write_and_cleanup(monkeypatch): + """Fail the staged write, and fail discarding the staged sibling too.""" + + def failing_fsync(_fd): + raise OSError("disk full") + + real_unlink = Path.unlink + + def failing_unlink(self, *args, **kwargs): + if self.name.startswith(".") and self.name.endswith(".tmp"): + raise PermissionError("staging is still open") + return real_unlink(self, *args, **kwargs) + + monkeypatch.setattr("deerflow.utils.atomic_write.os.fsync", failing_fsync) + monkeypatch.setattr(Path, "unlink", failing_unlink) + + +def test_discord_failed_cleanup_still_reports_and_keeps_the_outcome(tmp_path, monkeypatch, caplog): + """A discard that fails must not escape, and both failures stay visible.""" + channel = _discord(tmp_path) + target = tmp_path / "state" / "discord-threads.json" + target.parent.mkdir(parents=True) + target.write_text('{"kept": true}', encoding="utf-8") + _fail_staged_write_and_cleanup(monkeypatch) + caplog.clear() + + with caplog.at_level("WARNING"): + channel._persist_thread_mappings() + + assert target.read_text(encoding="utf-8") == '{"kept": true}' + assert "failed to persist thread mappings" in caplog.text + assert "Failed to remove temporary output file" in caplog.text + + +def test_wechat_failed_cleanup_still_reports_and_keeps_the_outcome(tmp_path, monkeypatch, caplog): + channel = _wechat(tmp_path) + target = tmp_path / "state" / "wechat-cursor.json" + target.parent.mkdir(parents=True) + target.write_text('{"get_updates_buf": "kept"}', encoding="utf-8") + _fail_staged_write_and_cleanup(monkeypatch) + caplog.clear() + + with caplog.at_level("WARNING"): + channel._save_state() + + assert target.read_text(encoding="utf-8") == '{"get_updates_buf": "kept"}' + assert "failed to persist cursor state" in caplog.text + assert "Failed to remove temporary output file" in caplog.text diff --git a/backend/tests/test_channel_state_encoding.py b/backend/tests/test_channel_state_encoding.py index 062bd47202..e4ac1c9c05 100644 --- a/backend/tests/test_channel_state_encoding.py +++ b/backend/tests/test_channel_state_encoding.py @@ -22,10 +22,8 @@ def _state_io_calls(filename: str, class_name: str, method_name: str, call_name: ("filename", "class_name", "method_name", "call_name"), [ ("store.py", "JsonChannelStore", "_save", "NamedTemporaryFile"), - ("discord.py", "DiscordChannel", "_persist_thread_mappings", "write_text"), ("runtime_config_store.py", "ChannelRuntimeConfigStore", "_save", "NamedTemporaryFile"), ("buzz_seen_events.py", "BuzzSeenEventStore", "_write_snapshot", "NamedTemporaryFile"), - ("wechat.py", "WechatChannel", "_save_state", "write_text"), ("wechat.py", "WechatChannel", "_save_auth_state", "NamedTemporaryFile"), ], ) @@ -57,3 +55,67 @@ def test_discord_state_read_is_explicitly_utf8() -> None: keywords = {keyword.arg: keyword.value for keyword in call.keywords} encoding = keywords.get("encoding") assert isinstance(encoding, ast.Constant) and encoding.value == "utf-8" + + +def test_discord_thread_mappings_publish_utf8_lf_bytes(tmp_path: Path) -> None: + import threading + + from app.channels.discord import DiscordChannel + + channel = DiscordChannel.__new__(DiscordChannel) + channel._thread_store_lock = threading.Lock() + channel._active_threads = {"微信:资源": {"thread_id": "thread-讨论"}} + channel._thread_store_path = tmp_path / "discord-threads.json" + + channel._persist_thread_mappings() + + text = channel._thread_store_path.read_bytes().decode("utf-8") + assert "\r" not in text + assert json.loads(text)["微信:资源"]["thread_id"] == "thread-讨论" + + +def test_wechat_cursor_state_publishes_utf8_lf_bytes(tmp_path: Path) -> None: + from app.channels.wechat import WechatChannel + + channel = WechatChannel.__new__(WechatChannel) + channel._cursor_path = tmp_path / "wechat-cursor.json" + channel._get_updates_buf = "游标-1" + + channel._save_state() + + text = channel._cursor_path.read_bytes().decode("utf-8") + assert "\r" not in text + assert json.loads(text) == {"get_updates_buf": "游标-1"} + + +def _method_calls(filename: str, class_name: str, method_name: str) -> list[ast.Call]: + module = ast.parse((CHANNELS / filename).read_text(encoding="utf-8")) + owner = next(node for node in module.body if isinstance(node, ast.ClassDef) and node.name == class_name) + method = next(node for node in owner.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name == method_name) + return [node for node in ast.walk(method) if isinstance(node, ast.Call)] + + +@pytest.mark.parametrize( + ("filename", "class_name", "method_name"), + [ + ("discord.py", "DiscordChannel", "_persist_thread_mappings"), + ("wechat.py", "WechatChannel", "_save_state"), + ], +) +def test_json_channel_state_writers_publish_through_the_helper(filename: str, class_name: str, method_name: str) -> None: + """Platform-independent: the call shape, not just the bytes it happens to emit. + + CI runs with ``PYTHONUTF8=1``, so a regression to ``write_text(...)`` without + ``encoding=`` would still produce valid UTF-8 there; only a source-level check + catches it on every platform. + """ + calls = _method_calls(filename, class_name, method_name) + names = [call.func.attr if isinstance(call.func, ast.Attribute) else call.func.id for call in calls if isinstance(call.func, (ast.Attribute, ast.Name))] + assert "publish_bytes_atomically" in names, f"{filename}:{class_name}.{method_name} must publish through the helper" + + for call in calls: + if not isinstance(call.func, ast.Attribute) or call.func.attr not in {"write_text", "write_bytes"}: + continue + keywords = {keyword.arg: keyword.value for keyword in call.keywords} + encoding = keywords.get("encoding") + assert isinstance(encoding, ast.Constant) and encoding.value == "utf-8", f"{filename}:{call.lineno} writes text without an explicit utf-8 encoding"