diff --git a/backend/app/channels/wechat.py b/backend/app/channels/wechat.py index 2d48956d7..4b5a81799 100644 --- a/backend/app/channels/wechat.py +++ b/backend/app/channels/wechat.py @@ -255,19 +255,29 @@ class WechatChannel(Channel): self._state_dir = self._resolve_state_dir(config.get("state_dir")) self._cursor_path = self._state_dir / "wechat-getupdates.json" if self._state_dir else None self._auth_path = self._state_dir / "wechat-auth.json" if self._state_dir else None - self._load_state() + # NOTE: persisted state (auth token + cursor) is intentionally NOT loaded + # here. ChannelService._start_channel() constructs the channel directly + # on the async path, so filesystem IO in __init__ would block the event + # loop (the strict blocking-IO gate raises BlockingError on os.stat). + # State is loaded in start() via asyncio.to_thread instead. async def start(self) -> None: if self._running: return + # Load persisted state off the event loop before the bot_token check + # below: a token restored from the auth file must be visible here so + # the qrcode-login fallback isn't taken unnecessarily. __init__ defers + # this load precisely so construction stays IO-free on the async path. + await asyncio.to_thread(self._load_state) + if not self._bot_token and not self._qrcode_login_enabled: logger.error("WeChat channel requires bot_token or qrcode_login_enabled") return self._main_loop = asyncio.get_running_loop() if self._state_dir: - self._state_dir.mkdir(parents=True, exist_ok=True) + await asyncio.to_thread(self._state_dir.mkdir, parents=True, exist_ok=True) await self._ensure_client() self._running = True @@ -372,7 +382,7 @@ class WechatChannel(Channel): return False try: - plaintext = attachment.actual_path.read_bytes() + plaintext = await asyncio.to_thread(attachment.actual_path.read_bytes) except OSError: logger.exception("[WeChat] failed to read outbound image %s", attachment.actual_path) return False @@ -462,7 +472,7 @@ class WechatChannel(Channel): return False try: - plaintext = attachment.actual_path.read_bytes() + plaintext = await asyncio.to_thread(attachment.actual_path.read_bytes) except OSError: logger.exception("[WeChat] failed to read outbound file %s", attachment.actual_path) return False @@ -555,8 +565,8 @@ class WechatChannel(Channel): if errcode == -14: self._bot_token = "" self._get_updates_buf = "" - self._save_state() - self._save_auth_state(status="expired", bot_token="") + await asyncio.to_thread(self._save_state) + await asyncio.to_thread(self._save_auth_state, status="expired", bot_token="") logger.error("[WeChat] bot token expired; scan again or update bot_token and restart the channel") self._running = False break @@ -574,7 +584,7 @@ class WechatChannel(Channel): next_buf = data.get("get_updates_buf") if isinstance(next_buf, str) and next_buf != self._get_updates_buf: self._get_updates_buf = next_buf - self._save_state() + await asyncio.to_thread(self._save_state) for raw_message in data.get("msgs", []): await self._handle_update(raw_message) @@ -692,7 +702,7 @@ class WechatChannel(Channel): if self._bot_token: return True - self._load_auth_state() + await asyncio.to_thread(self._load_auth_state) if self._bot_token: return True @@ -720,7 +730,8 @@ class WechatChannel(Channel): if qrcode_img_content: logger.warning("[WeChat] qrcode_img_content=%s", qrcode_img_content) - self._save_auth_state( + await asyncio.to_thread( + self._save_auth_state, status="pending", qrcode=qrcode, qrcode_img_content=qrcode_img_content or None, @@ -742,7 +753,8 @@ class WechatChannel(Channel): if ilink_bot_id: self._ilink_bot_id = ilink_bot_id - return self._save_auth_state( + return await asyncio.to_thread( + self._save_auth_state, status="confirmed", bot_token=token, ilink_bot_id=self._ilink_bot_id, @@ -751,7 +763,8 @@ class WechatChannel(Channel): ) if status in {"expired", "canceled", "cancelled", "invalid", "failed"}: - self._save_auth_state( + await asyncio.to_thread( + self._save_auth_state, status=status, qrcode=qrcode, qrcode_img_content=qrcode_img_content or None, @@ -760,7 +773,8 @@ class WechatChannel(Channel): await asyncio.sleep(max(self._qrcode_poll_interval, 0.1)) - self._save_auth_state( + await asyncio.to_thread( + self._save_auth_state, status="timeout", qrcode=qrcode, qrcode_img_content=qrcode_img_content or None, @@ -1046,7 +1060,7 @@ class WechatChannel(Channel): detected_image = _detect_image_extension_and_mime(decrypted) image_extension = detected_image[0] if detected_image else ".jpg" filename = _safe_media_filename("wechat-image", image_extension, message_id=message_id, index=index) - stored_path = self._stage_downloaded_file(filename, decrypted) + stored_path = await asyncio.to_thread(self._stage_downloaded_file, filename, decrypted) if stored_path is None: return None @@ -1097,7 +1111,7 @@ class WechatChannel(Channel): logger.warning("[WeChat] inbound file exceeds size limit (%d bytes), skipping message_id=%s", len(decrypted), message_id) return None - stored_path = self._stage_downloaded_file(filename, decrypted) + stored_path = await asyncio.to_thread(self._stage_downloaded_file, filename, decrypted) if stored_path is None: return None diff --git a/backend/tests/blocking_io/test_wechat_channel_state.py b/backend/tests/blocking_io/test_wechat_channel_state.py new file mode 100644 index 000000000..b1f966b77 --- /dev/null +++ b/backend/tests/blocking_io/test_wechat_channel_state.py @@ -0,0 +1,114 @@ +"""Regression anchor: WeChat channel filesystem IO must not block the event loop. + +Two production paths touch the filesystem, and both must stay off the asyncio +loop: + +1. **Construction** — ``ChannelService._start_channel()`` instantiates the + channel directly on the async path (``channel_cls(...)`` with no thread + offload), so ``WechatChannel.__init__`` must be IO-free; persisted state is + loaded later in ``start()`` via ``asyncio.to_thread``. Constructing the + channel in an async context used to raise ``BlockingError: Blocking call to + os.stat`` because ``__init__`` called ``_load_state`` synchronously. + +2. **Runtime** — ``_handle_update`` stages downloaded inbound files + (``mkdir`` + ``write_bytes``) and ``_ensure_authenticated`` reads persisted + auth state (``read_text``); both offload via ``asyncio.to_thread``. + +If any of this regresses back onto the event loop, the strict Blockbuster gate +raises ``BlockingError`` and these tests fail. + +The constructors below are invoked *directly* on the event loop (no +``asyncio.to_thread`` wrapper) on purpose: that mirrors the production +``_start_channel`` path the gate is meant to protect. Test-only file setup +(writing the auth fixture) is still wrapped in ``asyncio.to_thread`` because +that scaffolding IO would otherwise trip the gate even though it is not the +code under test. +""" + +from __future__ import annotations + +import asyncio +import json +from pathlib import Path + +import pytest + +from app.channels.message_bus import MessageBus +from app.channels.wechat import WechatChannel, _encrypt_aes_128_ecb + +pytestmark = pytest.mark.asyncio + + +async def test_wechat_constructor_is_io_free_on_async_path(tmp_path: Path) -> None: + """``__init__`` must not touch the filesystem — ``_start_channel`` constructs inline.""" + bus = MessageBus() + # Seed an auth file that WOULD restore a different token if __init__ read it. + auth_path = tmp_path / "wechat-auth.json" + await asyncio.to_thread(auth_path.write_text, json.dumps({"status": "confirmed", "bot_token": "from-disk"})) + + # Direct construction on the event loop, no asyncio.to_thread wrapper — + # this is exactly the production _start_channel path. If __init__ regresses + # to doing os.stat / read_text (e.g. re-adding _load_state here), the gate + # raises BlockingError right at this line. + channel = WechatChannel(bus=bus, config={"bot_token": "from-config", "state_dir": str(tmp_path)}) + # Token came from config, not the disk file — proving __init__ did not read it. + assert channel._bot_token == "from-config" + + +async def test_wechat_inbound_file_staging_does_not_block_event_loop(tmp_path: Path) -> None: + """Staging a downloaded inbound image writes through ``asyncio.to_thread``.""" + bus = MessageBus() + published = [] + + async def capture(msg): + published.append(msg) + + bus.publish_inbound = capture # type: ignore[method-assign] + + channel = WechatChannel(bus=bus, config={"bot_token": "test-token", "state_dir": str(tmp_path)}) + + plaintext = b"fake-image-bytes" + aes_key = b"1234567890abcdef" + encrypted = _encrypt_aes_128_ecb(plaintext, aes_key) + + async def _fake_download(_url: str, *, timeout: float | None = None): + return encrypted + + channel._download_cdn_bytes = _fake_download # type: ignore[method-assign] + + await channel._handle_update( + { + "message_type": 1, + "message_id": 101, + "from_user_id": "wx-1", + "context_token": "ctx-img", + "item_list": [ + { + "type": 2, + "image_item": {"aeskey": aes_key.hex(), "media": {"full_url": "https://cdn.example/image.bin"}}, + } + ], + } + ) + + assert len(published) == 1 + assert len(published[0].files) == 1 + staged = Path(published[0].files[0]["path"]) + staged_exists = await asyncio.to_thread(staged.exists) + assert staged_exists, "inbound image should be staged under the tmp state dir" + + +async def test_wechat_auth_state_load_does_not_block_event_loop(tmp_path: Path) -> None: + """``_ensure_authenticated`` reads persisted auth state through ``asyncio.to_thread``.""" + bus = MessageBus() + # bot_token="" forces _ensure_authenticated to fall through to the + # _load_auth_state branch (offloaded) instead of returning early. + channel = WechatChannel(bus=bus, config={"bot_token": "", "state_dir": str(tmp_path)}) + + auth_path = tmp_path / "wechat-auth.json" + await asyncio.to_thread(auth_path.write_text, json.dumps({"status": "confirmed", "bot_token": "loaded-token"})) + + result = await channel._ensure_authenticated() + + assert result is True + assert channel._bot_token == "loaded-token" diff --git a/backend/tests/test_wechat_channel.py b/backend/tests/test_wechat_channel.py index 52b46d31a..d05770af1 100644 --- a/backend/tests/test_wechat_channel.py +++ b/backend/tests/test_wechat_channel.py @@ -1248,6 +1248,9 @@ def test_state_cursor_is_loaded_from_disk(tmp_path: Path): bus=MessageBus(), config={"bot_token": "bot-token", "state_dir": str(state_dir)}, ) + # State load moved out of __init__ (it does filesystem IO that would block + # the async path); mirror start() by loading explicitly here. + channel._load_state() assert channel._get_updates_buf == "cursor-123" @@ -1266,6 +1269,9 @@ def test_auth_state_is_loaded_from_disk(tmp_path: Path): bus=MessageBus(), config={"state_dir": str(state_dir), "qrcode_login_enabled": True}, ) + # State load moved out of __init__ (it does filesystem IO that would block + # the async path); mirror start() by loading explicitly here. + channel._load_state() assert channel._bot_token == "saved-token" assert channel._ilink_bot_id == "bot-1"