From 5d8c4c22729f7c8cdc3706f150680322dd0275fc Mon Sep 17 00:00:00 2001 From: Ryker_Feng <90562015+18062706139fcz@users.noreply.github.com> Date: Sat, 1 Aug 2026 22:16:34 +0800 Subject: [PATCH] fix(feishu): keep file receive off the event loop (#4627) --- backend/AGENTS.md | 3 + backend/app/channels/feishu.py | 25 ++-- .../blocking_io/test_feishu_receive_file.py | 126 ++++++++++++++++++ backend/tests/test_feishu_parser.py | 6 +- 4 files changed, 146 insertions(+), 14 deletions(-) create mode 100644 backend/tests/blocking_io/test_feishu_receive_file.py diff --git a/backend/AGENTS.md b/backend/AGENTS.md index abe0d085e..5a6288ea4 100644 --- a/backend/AGENTS.md +++ b/backend/AGENTS.md @@ -264,6 +264,9 @@ Blocking-IO runtime gate (`tests/blocking_io/`): `test_uploads_router.py` (locks Gateway upload/list/delete endpoints offloading upload directory creation, staged writes, chmod/cleanup, directory scans/deletes, and remote sandbox sync off the event loop); + `test_feishu_receive_file.py` (locks Feishu attachment path preparation and + persistence plus remote sandbox acquisition/sync off the event loop, and + skips redundant sandbox sync when thread data is already mounted); `test_openviking_memory_backend.py` (locks the OpenViking backend's async add/context/search entrypoints offloading synchronous HTTP and watermark filesystem IO); and diff --git a/backend/app/channels/feishu.py b/backend/app/channels/feishu.py index 6feb4f729..60eef38f6 100644 --- a/backend/app/channels/feishu.py +++ b/backend/app/channels/feishu.py @@ -424,10 +424,8 @@ class FeishuChannel(Channel): logger.warning("[Feishu] empty resource content: resource_key=%s, type=%s", file_key, type) return f"Failed to obtain the [{type}]" - paths = get_paths() effective_user_id = user_id or get_effective_user_id() - paths.ensure_thread_dirs(thread_id, user_id=effective_user_id) - uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve() + paths = await asyncio.to_thread(get_paths) ext = "png" if type == "image" else "bin" raw_filename = getattr(response, "file_name", "") or f"feishu_{file_key[-12:]}.{ext}" @@ -439,30 +437,33 @@ class FeishuChannel(Channel): filename = f"{name_part}.{ext}" else: filename = re.sub(r"[./\\]", "_", raw_filename) - resolved_target = uploads_dir / filename - def down_load(): - # use thread_lock to avoid filename conflicts when writing + def _persist(): + paths.ensure_thread_dirs(thread_id, user_id=effective_user_id) + uploads_dir = paths.sandbox_uploads_dir(thread_id, user_id=effective_user_id).resolve() + resolved_target = uploads_dir / filename + # Use thread_lock to avoid filename conflicts when writing. with self._thread_lock: resolved_target.write_bytes(content) + return resolved_target try: - await asyncio.to_thread(down_load) + resolved_target = await asyncio.to_thread(_persist) except Exception: - logger.exception("[Feishu] failed to persist downloaded resource: %s, type=%s", resolved_target, type) + logger.exception("[Feishu] failed to persist downloaded resource: %s, type=%s", filename, type) return f"Failed to obtain the [{type}]" virtual_path = f"{VIRTUAL_PATH_PREFIX}/uploads/{resolved_target.name}" try: - sandbox_provider = get_sandbox_provider() - sandbox_id = sandbox_provider.acquire(thread_id, user_id=effective_user_id) - if sandbox_id != "local": + sandbox_provider = await asyncio.to_thread(get_sandbox_provider) + if not getattr(sandbox_provider, "uses_thread_data_mounts", False): + sandbox_id = await sandbox_provider.acquire_async(thread_id, user_id=effective_user_id) sandbox = sandbox_provider.get(sandbox_id) if sandbox is None: logger.warning("[Feishu] sandbox not found for thread_id=%s", thread_id) return f"Failed to obtain the [{type}]" - sandbox.update_file(virtual_path, content) + await asyncio.to_thread(sandbox.update_file, virtual_path, content) except Exception: logger.exception("[Feishu] failed to sync resource into non-local sandbox: %s", virtual_path) return f"Failed to obtain the [{type}]" diff --git a/backend/tests/blocking_io/test_feishu_receive_file.py b/backend/tests/blocking_io/test_feishu_receive_file.py new file mode 100644 index 000000000..59603284c --- /dev/null +++ b/backend/tests/blocking_io/test_feishu_receive_file.py @@ -0,0 +1,126 @@ +"""Regression anchors: Feishu ``receive_file`` must not block the event loop.""" + +from __future__ import annotations + +import asyncio +import threading +from io import BytesIO +from unittest.mock import MagicMock + +import pytest + +pytestmark = pytest.mark.asyncio + + +def _channel_with_file(content: bytes = b"DATA", filename: str = "report.pdf"): + from app.channels.feishu import FeishuChannel + from app.channels.message_bus import MessageBus + + channel = FeishuChannel(MessageBus(), {"app_id": "test", "app_secret": "test"}) + channel._GetMessageResourceRequest = MagicMock() + builder = MagicMock() + builder.message_id.return_value = builder + builder.file_key.return_value = builder + builder.type.return_value = builder + builder.build.return_value = object() + channel._GetMessageResourceRequest.builder.return_value = builder + + response = MagicMock() + response.success.return_value = True + response.file = BytesIO(content) + response.file_name = filename + channel._api_client = MagicMock() + channel._api_client.im.v1.message_resource.get.return_value = response + return channel + + +class _RemoteSandbox: + def __init__(self) -> None: + self.updates: list[tuple[str, bytes]] = [] + self.update_thread_id: int | None = None + + def update_file(self, path: str, content: bytes) -> None: + self.update_thread_id = threading.get_ident() + self.updates.append((path, content)) + + +class _RemoteProvider: + uses_thread_data_mounts = False + + def __init__(self) -> None: + self.sandbox = _RemoteSandbox() + self.acquire_async_calls: list[tuple[str | None, str | None]] = [] + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("Feishu receive_file must use acquire_async") + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + self.acquire_async_calls.append((thread_id, user_id)) + return "remote-sandbox" + + def get(self, sandbox_id: str): + return self.sandbox if sandbox_id == "remote-sandbox" else None + + +class _MountedProvider: + uses_thread_data_mounts = True + + def acquire(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("mounted uploads must not acquire a sandbox") + + async def acquire_async(self, thread_id: str | None = None, *, user_id: str | None = None) -> str: + raise AssertionError("mounted uploads must not acquire a sandbox") + + def get(self, sandbox_id: str): + raise AssertionError("mounted uploads must not look up a sandbox") + + +async def test_receive_file_remote_sandbox_does_not_block_event_loop(tmp_path, monkeypatch) -> None: + from deerflow.config.paths import Paths + + paths = await asyncio.to_thread(Paths, str(tmp_path)) + provider = _RemoteProvider() + provider_lookup_thread_id = None + + def _get_provider(): + nonlocal provider_lookup_thread_id + provider_lookup_thread_id = threading.get_ident() + return provider + + monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths) + monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", _get_provider) + + loop_thread_id = threading.get_ident() + result = await _channel_with_file()._receive_single_file( + "message-1", + "file-key", + "file", + "thread-1", + user_id="ou-user", + ) + + assert result == "/mnt/user-data/uploads/report.pdf" + assert provider.acquire_async_calls == [("thread-1", "ou-user")] + assert provider.sandbox.updates == [(result, b"DATA")] + assert provider_lookup_thread_id != loop_thread_id + assert provider.sandbox.update_thread_id != loop_thread_id + + +async def test_receive_file_mounted_sandbox_skips_redundant_sync(tmp_path, monkeypatch) -> None: + from deerflow.config.paths import Paths + + paths = await asyncio.to_thread(Paths, str(tmp_path)) + monkeypatch.setattr("app.channels.feishu.get_paths", lambda: paths) + monkeypatch.setattr("app.channels.feishu.get_sandbox_provider", lambda: _MountedProvider()) + + result = await _channel_with_file()._receive_single_file( + "message-1", + "file-key", + "file", + "thread-1", + user_id="ou-user", + ) + + assert result == "/mnt/user-data/uploads/report.pdf" + uploaded = tmp_path / "users" / "ou-user" / "threads" / "thread-1" / "user-data" / "uploads" / "report.pdf" + assert await asyncio.to_thread(uploaded.read_bytes) == b"DATA" diff --git a/backend/tests/test_feishu_parser.py b/backend/tests/test_feishu_parser.py index 36d45926b..8cf332f86 100644 --- a/backend/tests/test_feishu_parser.py +++ b/backend/tests/test_feishu_parser.py @@ -177,7 +177,9 @@ def test_feishu_receive_file_syncs_sandbox_with_explicit_user_id(tmp_path, monke channel._api_client.im.v1.message_resource.get.return_value = response provider = MagicMock() - provider.acquire.return_value = "aio-1" + provider.uses_thread_data_mounts = False + provider.acquire.side_effect = AssertionError("receive_file must use acquire_async") + provider.acquire_async = AsyncMock(return_value="aio-1") sandbox = MagicMock() provider.get.return_value = sandbox @@ -189,7 +191,7 @@ def test_feishu_receive_file_syncs_sandbox_with_explicit_user_id(tmp_path, monke assert virtual_path == "/mnt/user-data/uploads/report.md" assert (tmp_path / "users" / "ou-user" / "threads" / "thread-1" / "user-data" / "uploads" / "report.md").read_bytes() == b"file-bytes" - provider.acquire.assert_called_once_with("thread-1", user_id="ou-user") + provider.acquire_async.assert_awaited_once_with("thread-1", user_id="ou-user") sandbox.update_file.assert_called_once_with("/mnt/user-data/uploads/report.md", b"file-bytes") _run(go())