fix: introduce asynchronous, revision-aware config snapshot saving to keep event loops responsive and prevent stale writes from overwriting newer state (#9300)

* fix(platform): offload Weixin account state writes

* fix(config): serialize async snapshot persistence

* fix(config): preserve valid snapshot on newer save failure

---------

Co-authored-by: Rhonin <rhonin@STARFORGE.localdomain>
Co-authored-by: Soulter <905617992@qq.com>

astrbot/core/config/astrbot_config.py

@@ -1,8 +1,11 @@
+import asyncio
+import copy
 import enum
 import json
 import logging
 import os
 import tempfile
+import threading
 
 from astrbot.core.utils.astrbot_path import get_astrbot_data_path
 from astrbot.core.utils.auth_password import (
@@ -49,6 +52,10 @@ class AstrBotConfig(dict):
         object.__setattr__(self, "config_path", config_path)
         object.__setattr__(self, "default_config", default_config)
         object.__setattr__(self, "schema", schema)
+        object.__setattr__(self, "_save_state_lock", threading.Lock())
+        object.__setattr__(self, "_save_commit_lock", threading.Lock())
+        object.__setattr__(self, "_save_revision", 0)
+        object.__setattr__(self, "_save_committed_revision", 0)
 
         if schema:
             default_config = self._config_schema_to_default_config(schema)
@@ -225,30 +232,95 @@ class AstrBotConfig(dict):
     def save_config(
         self, replace_config: dict | None = None, *, indent: int = 2
     ) -> None:
-        """将配置写入文件
+        """Persist the current configuration synchronously.
 
-        如果传入 replace_config，则将配置替换为 replace_config
+        Args:
+            replace_config: Values to merge into the configuration before saving.
+            indent: Number of spaces used to indent the JSON output.
+        """
+        snapshot, revision = self._prepare_config_snapshot(replace_config)
+        self._write_config_snapshot(snapshot, revision, indent)
+
+    async def save_config_async(
+        self, replace_config: dict | None = None, *, indent: int = 2
+    ) -> bool:
+        """Persist a stable configuration snapshot without blocking the event loop.
+
+        Args:
+            replace_config: Values to merge into the configuration before saving.
+            indent: Number of spaces used to indent the JSON output.
+
+        Returns:
+            Whether this snapshot was committed. A newer committed snapshot supersedes
+            an older snapshot.
+        """
+        snapshot, revision = self._prepare_config_snapshot(replace_config)
+        return await asyncio.to_thread(
+            self._write_config_snapshot,
+            snapshot,
+            revision,
+            indent,
+        )
+
+    def _prepare_config_snapshot(self, replace_config: dict | None) -> tuple[dict, int]:
+        """Create an isolated snapshot and allocate its save revision.
+
+        Args:
+            replace_config: Values to merge into the configuration before snapshotting.
+
+        Returns:
+            The isolated configuration snapshot and its monotonically increasing
+            revision.
+        """
+        with self._save_state_lock:
+            if replace_config:
+                self.update(replace_config)
+            snapshot = copy.deepcopy(dict(self))
+            revision = self._save_revision + 1
+            object.__setattr__(self, "_save_revision", revision)
+        return snapshot, revision
+
+    def _write_config_snapshot(
+        self, snapshot: dict, revision: int, indent: int
+    ) -> bool:
+        """Write and conditionally commit a prepared configuration snapshot.
+
+        Args:
+            snapshot: Isolated configuration data to serialize.
+            revision: Revision allocated when the snapshot was prepared.
+            indent: Number of spaces used to indent the JSON output.
+
+        Returns:
+            Whether the snapshot replaced the current configuration file.
         """
-        if replace_config:
-            self.update(replace_config)
         directory = os.path.dirname(os.path.abspath(self.config_path)) or "."
         fd, temp_path = tempfile.mkstemp(
             dir=directory,
             prefix=f".{os.path.basename(self.config_path)}.",
             suffix=".tmp",
         )
+        committed = False
         try:
             with os.fdopen(fd, "w", encoding="utf-8-sig") as f:
-                json.dump(self, f, indent=indent, ensure_ascii=False)
+                json.dump(snapshot, f, indent=indent, ensure_ascii=False)
                 f.flush()
                 os.fsync(f.fileno())
-            os.replace(temp_path, self.config_path)
-        except Exception:
-            try:
-                os.unlink(temp_path)
-            except FileNotFoundError:
-                pass
-            raise
+            with self._save_commit_lock:
+                if revision > self._save_committed_revision:
+                    os.replace(temp_path, self.config_path)
+                    object.__setattr__(
+                        self,
+                        "_save_committed_revision",
+                        revision,
+                    )
+                    committed = True
+        finally:
+            if not committed:
+                try:
+                    os.unlink(temp_path)
+                except FileNotFoundError:
+                    pass
+        return committed
 
     def __getattr__(self, item):
         try:

astrbot/core/platform/sources/weixin_oc/weixin_oc_adapter.py

@@ -157,6 +157,7 @@ class WeixinOCAdapter(Platform):
         self._qr_expired_count = 0
         self._context_tokens: dict[str, str] = {}
         self._context_tokens_dirty = False
+        self._context_tokens_revision = 0
         self._typing_states: dict[str, TypingSessionState] = {}
         self._last_inbound_error = ""
         self._recent_message_cache_size = self._get_int_config(
@@ -564,6 +565,7 @@ class WeixinOCAdapter(Platform):
 
     async def _save_account_state(self) -> None:
         normalized_context_tokens = self._normalize_context_tokens(self._context_tokens)
+        context_tokens_revision = self._context_tokens_revision
         self.config["weixin_oc_token"] = self.token or ""
         self.config["weixin_oc_account_id"] = self.account_id or ""
         self.config["weixin_oc_sync_buf"] = self._sync_buf
@@ -585,8 +587,9 @@ class WeixinOCAdapter(Platform):
             break
 
         self._sync_client_state()
-        astrbot_config.save_config()
-        self._context_tokens_dirty = False
+        committed = await astrbot_config.save_config_async()
+        if committed and context_tokens_revision == self._context_tokens_revision:
+            self._context_tokens_dirty = False
 
     def _is_login_session_valid(
         self, login_session: OpenClawLoginSession | None
@@ -980,6 +983,7 @@ class WeixinOCAdapter(Platform):
         self.account_id = None
         self._sync_buf = ""
         self._context_tokens = {}
+        self._context_tokens_revision += 1
         self._context_tokens_dirty = False
         self._login_session = None
         await self._save_account_state()
@@ -1507,6 +1511,7 @@ class WeixinOCAdapter(Platform):
             previous_context_token = self._context_tokens.get(from_user_id)
             if previous_context_token != context_token:
                 self._context_tokens[from_user_id] = context_token
+                self._context_tokens_revision += 1
                 self._context_tokens_dirty = True
 
         item_list = cast(list[dict[str, Any]], msg.get("item_list", []))

tests/test_weixin_oc_state_persistence.py

@@ -0,0 +1,153 @@
+import asyncio
+
+import pytest
+
+from astrbot.core.platform.sources.weixin_oc import weixin_oc_adapter
+from astrbot.core.platform.sources.weixin_oc.weixin_oc_adapter import WeixinOCAdapter
+
+
+class _Config(dict):
+    def __init__(self, *args, calls: list[str], **kwargs):
+        super().__init__(*args, **kwargs)
+        self._calls = calls
+
+    async def save_config_async(self) -> bool:
+        self._calls.append("save_config_async")
+        return True
+
+
+@pytest.mark.asyncio
+async def test_save_account_state_uses_async_config_persistence(monkeypatch):
+    calls: list[str] = []
+    config = _Config(
+        {
+            "platform": [
+                {
+                    "id": "weixin-test",
+                    "type": "weixin_oc",
+                }
+            ]
+        },
+        calls=calls,
+    )
+
+    monkeypatch.setattr(weixin_oc_adapter, "astrbot_config", config)
+
+    adapter = object.__new__(WeixinOCAdapter)
+    adapter.config = {"id": "weixin-test", "type": "weixin_oc"}
+    adapter.token = "token"
+    adapter.account_id = "account"
+    adapter._sync_buf = "sync-buffer"
+    adapter.base_url = "https://example.com"
+    adapter._context_tokens = {"user": "context-token"}
+    adapter._context_tokens_dirty = True
+    adapter._context_tokens_revision = 0
+    adapter._sync_client_state = lambda: None
+
+    await adapter._save_account_state()
+
+    assert calls == ["save_config_async"]
+    assert config["platform"][0]["weixin_oc_sync_buf"] == "sync-buffer"
+    assert adapter._context_tokens_dirty is False
+
+
+@pytest.mark.asyncio
+async def test_save_account_state_keeps_dirty_flag_for_new_context_token(monkeypatch):
+    save_started = asyncio.Event()
+    finish_save = asyncio.Event()
+
+    class BlockingConfig(dict):
+        async def save_config_async(self) -> bool:
+            save_started.set()
+            await finish_save.wait()
+            return True
+
+    config = BlockingConfig(
+        {
+            "platform": [
+                {
+                    "id": "weixin-test",
+                    "type": "weixin_oc",
+                }
+            ]
+        }
+    )
+    monkeypatch.setattr(weixin_oc_adapter, "astrbot_config", config)
+
+    adapter = object.__new__(WeixinOCAdapter)
+    adapter.config = {"id": "weixin-test", "type": "weixin_oc"}
+    adapter.token = "token"
+    adapter.account_id = "account"
+    adapter._sync_buf = "sync-buffer"
+    adapter.base_url = "https://example.com"
+    adapter._context_tokens = {"user": "old-context-token"}
+    adapter._context_tokens_dirty = True
+    adapter._context_tokens_revision = 0
+    adapter._sync_client_state = lambda: None
+
+    save_task = asyncio.create_task(adapter._save_account_state())
+    save_started_task = asyncio.create_task(save_started.wait())
+    done, _ = await asyncio.wait(
+        {save_task, save_started_task},
+        timeout=5,
+        return_when=asyncio.FIRST_COMPLETED,
+    )
+    if save_task in done:
+        await save_task
+    assert save_started_task in done
+    adapter._context_tokens["user"] = "new-context-token"
+    adapter._context_tokens_revision += 1
+    adapter._context_tokens_dirty = True
+    finish_save.set()
+    await save_task
+
+    assert adapter._context_tokens_dirty is True
+
+
+@pytest.mark.asyncio
+async def test_save_account_state_keeps_dirty_flag_after_context_token_aba(
+    monkeypatch,
+):
+    save_started = asyncio.Event()
+    finish_save = asyncio.Event()
+
+    class BlockingConfig(dict):
+        async def save_config_async(self) -> bool:
+            save_started.set()
+            await finish_save.wait()
+            return True
+
+    config = BlockingConfig(
+        {
+            "platform": [
+                {
+                    "id": "weixin-test",
+                    "type": "weixin_oc",
+                }
+            ]
+        }
+    )
+    monkeypatch.setattr(weixin_oc_adapter, "astrbot_config", config)
+
+    adapter = object.__new__(WeixinOCAdapter)
+    adapter.config = {"id": "weixin-test", "type": "weixin_oc"}
+    adapter.token = "token"
+    adapter.account_id = "account"
+    adapter._sync_buf = "sync-buffer"
+    adapter.base_url = "https://example.com"
+    adapter._context_tokens = {"user": "context-a"}
+    adapter._context_tokens_dirty = True
+    adapter._context_tokens_revision = 0
+    adapter._sync_client_state = lambda: None
+
+    save_task = asyncio.create_task(adapter._save_account_state())
+    await asyncio.wait_for(save_started.wait(), timeout=5)
+    adapter._context_tokens["user"] = "context-b"
+    adapter._context_tokens_revision += 1
+    adapter._context_tokens["user"] = "context-a"
+    adapter._context_tokens_revision += 1
+    adapter._context_tokens_dirty = True
+    finish_save.set()
+    await save_task
+
+    assert adapter._context_tokens_dirty is True

tests/unit/test_config.py

@@ -1,7 +1,9 @@
 """Tests for config module."""
 
+import asyncio
 import json
 import os
+import threading
 
 import pytest
 
@@ -557,6 +559,180 @@ class TestConfigHotReload:
 
         assert loaded_config["new_field"] == "new_value"
 
+    @pytest.mark.asyncio
+    async def test_save_config_async_keeps_event_loop_responsive(
+        self, temp_config_path, minimal_default_config, monkeypatch
+    ):
+        config = AstrBotConfig(
+            config_path=temp_config_path, default_config=minimal_default_config
+        )
+        write_started = threading.Event()
+        finish_write = threading.Event()
+        original_fsync = os.fsync
+
+        def blocking_fsync(fd):
+            write_started.set()
+            assert finish_write.wait(timeout=5)
+            original_fsync(fd)
+
+        monkeypatch.setattr(os, "fsync", blocking_fsync)
+        config["async_field"] = "saved"
+
+        save_task = asyncio.create_task(config.save_config_async())
+        assert await asyncio.to_thread(write_started.wait, 5)
+        await asyncio.sleep(0)
+        assert not save_task.done()
+
+        finish_write.set()
+        await save_task
+
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert json.load(f)["async_field"] == "saved"
+
+    @pytest.mark.asyncio
+    async def test_save_config_async_writes_stable_snapshot(
+        self, temp_config_path, minimal_default_config, monkeypatch
+    ):
+        config = AstrBotConfig(
+            config_path=temp_config_path, default_config=minimal_default_config
+        )
+        dump_started = threading.Event()
+        finish_dump = threading.Event()
+        original_dump = json.dump
+
+        def blocking_dump(snapshot, file_obj, **kwargs):
+            dump_started.set()
+            assert finish_dump.wait(timeout=5)
+            original_dump(snapshot, file_obj, **kwargs)
+
+        monkeypatch.setattr(json, "dump", blocking_dump)
+        config["snapshot_field"] = "captured"
+
+        save_task = asyncio.create_task(config.save_config_async())
+        assert await asyncio.to_thread(dump_started.wait, 5)
+        config["snapshot_field"] = "changed-after-save-started"
+        finish_dump.set()
+        await save_task
+
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert json.load(f)["snapshot_field"] == "captured"
+
+    @pytest.mark.asyncio
+    async def test_save_config_async_does_not_block_next_snapshot_during_replace(
+        self, temp_config_path, minimal_default_config, monkeypatch
+    ):
+        config = AstrBotConfig(
+            config_path=temp_config_path, default_config=minimal_default_config
+        )
+        first_replace_started = threading.Event()
+        finish_first_replace = threading.Event()
+        replace_call_count = 0
+        replace_call_lock = threading.Lock()
+        original_replace = os.replace
+
+        def blocking_replace(source, destination):
+            nonlocal replace_call_count
+            with replace_call_lock:
+                replace_call_count += 1
+                call_number = replace_call_count
+            if call_number == 1:
+                first_replace_started.set()
+                if not finish_first_replace.wait(timeout=2):
+                    raise TimeoutError("event loop could not prepare the next snapshot")
+            original_replace(source, destination)
+
+        monkeypatch.setattr(os, "replace", blocking_replace)
+        config["replace_order"] = "older"
+        older_save = asyncio.create_task(config.save_config_async())
+        assert await asyncio.to_thread(first_replace_started.wait, 5)
+
+        config["replace_order"] = "newer"
+        newer_save = asyncio.create_task(config.save_config_async())
+        await asyncio.sleep(0)
+        finish_first_replace.set()
+        await asyncio.gather(older_save, newer_save)
+
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert json.load(f)["replace_order"] == "newer"
+
+    @pytest.mark.asyncio
+    async def test_save_config_async_discards_older_late_write(
+        self, temp_config_path, minimal_default_config, monkeypatch
+    ):
+        config = AstrBotConfig(
+            config_path=temp_config_path, default_config=minimal_default_config
+        )
+        first_write_started = threading.Event()
+        finish_first_write = threading.Event()
+        fsync_call_count = 0
+        fsync_call_lock = threading.Lock()
+        original_fsync = os.fsync
+
+        def reorder_fsync(fd):
+            nonlocal fsync_call_count
+            with fsync_call_lock:
+                fsync_call_count += 1
+                call_number = fsync_call_count
+            if call_number == 1:
+                first_write_started.set()
+                assert finish_first_write.wait(timeout=5)
+            original_fsync(fd)
+
+        monkeypatch.setattr(os, "fsync", reorder_fsync)
+        config["save_order"] = "older"
+        older_save = asyncio.create_task(config.save_config_async())
+        assert await asyncio.to_thread(first_write_started.wait, 5)
+
+        config["save_order"] = "newer"
+        newer_save_committed = await config.save_config_async()
+        finish_first_write.set()
+        older_save_committed = await older_save
+
+        assert newer_save_committed is True
+        assert older_save_committed is False
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert json.load(f)["save_order"] == "newer"
+
+    @pytest.mark.asyncio
+    async def test_save_config_commits_older_snapshot_when_newer_write_fails(
+        self, temp_config_path, minimal_default_config, monkeypatch
+    ):
+        config = AstrBotConfig(
+            config_path=temp_config_path, default_config=minimal_default_config
+        )
+        first_write_started = threading.Event()
+        finish_first_write = threading.Event()
+        fsync_call_count = 0
+        fsync_call_lock = threading.Lock()
+        original_fsync = os.fsync
+
+        def fail_newer_fsync(fd):
+            nonlocal fsync_call_count
+            with fsync_call_lock:
+                fsync_call_count += 1
+                call_number = fsync_call_count
+            if call_number == 1:
+                first_write_started.set()
+                assert finish_first_write.wait(timeout=5)
+                original_fsync(fd)
+                return
+            raise OSError("simulated newer fsync failure")
+
+        monkeypatch.setattr(os, "fsync", fail_newer_fsync)
+        config["save_order"] = "older-valid"
+        older_save = asyncio.create_task(asyncio.to_thread(config.save_config))
+        assert await asyncio.to_thread(first_write_started.wait, 5)
+
+        config["save_order"] = "newer-failed"
+        with pytest.raises(OSError, match="simulated newer fsync failure"):
+            await config.save_config_async()
+
+        finish_first_write.set()
+        await older_save
+
+        with open(temp_config_path, encoding="utf-8-sig") as f:
+            assert json.load(f)["save_order"] == "older-valid"
+
     def test_save_config_with_replace(self, temp_config_path, minimal_default_config):
         """Test saving config with replacement."""
         config = AstrBotConfig(