github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-22 14:08:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(artifacts): offload blocking filesystem IO in artifact serving (#3551)

Browse Source 
get_artifact ran its filesystem work directly on the event loop: virtual-path
resolution (os.path.abspath via .resolve()), exists/is_file probes, MIME sniffing
(mimetypes lazily stats the system MIME DB on first use), full-file
read_text/read_bytes, is_text_file_by_content (open+read), and .skill ZIP
open+extract. So serving any artifact blocked the loop for the whole read;
`make detect-blocking-io` flagged it. Same class as #3457 / #3529.

Offload each branch's IO via asyncio.to_thread: one sync helper per branch
(_load_skill_archive_member, _read_artifact_payload) folds stat + MIME + read /
extract into a single worker hop and returns a small (kind, mime, payload) plan
the handler turns into the response on the loop. FileResponse (download / active
content) keeps streaming the file itself. Behavior, branching, error codes, and
security boundaries are unchanged.

Add tests/blocking_io/test_artifacts_router.py anchor (text / binary / .skill
member), verified red->green under the strict Blockbuster gate. The gate also
caught a blocking call the static scan missed: resolve_thread_virtual_path's
.resolve() (os.path.abspath), now offloaded too.

Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
ly-wang19 2026-06-24 10:44:46 +08:00 committed by GitHub
parent 7be73fcf19
commit 435edbd8c6
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 141 additions and 33 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

93
backend/app/gateway/routers/artifacts.py
View File

@ -1,3 +1,4 @@
import asyncio
import logging
import mimetypes
import zipfile
@ -96,6 +97,51 @@ def _extract_file_from_skill_archive(zip_path: Path, internal_path: str) -> byte
return None
def _load_skill_archive_member(actual_skill_path: Path, skill_file_path: str, internal_path: str) -> tuple[bytes, str | None]:
"""Worker-thread body for the ``.skill`` branch of ``get_artifact``.
The ``exists`` / ``is_file`` probes, the ZIP open+extract, and the MIME
sniff (``mimetypes`` lazily stats the system MIME database on first use) are
blocking filesystem IO and must stay off the event loop. Raised
``HTTPException``s propagate through ``asyncio.to_thread`` unchanged,
preserving status codes.
"""
if not actual_skill_path.exists():
raise HTTPException(status_code=404, detail=f"Skill file not found: {skill_file_path}")
if not actual_skill_path.is_file():
raise HTTPException(status_code=400, detail=f"Path is not a file: {skill_file_path}")
content = _extract_file_from_skill_archive(actual_skill_path, internal_path)
if content is None:
raise HTTPException(status_code=404, detail=f"File '{internal_path}' not found in skill archive")
mime_type, _ = mimetypes.guess_type(internal_path)
return content, mime_type
def _read_artifact_payload(actual_path: Path, path: str, download: bool) -> tuple[str, str | None, bytes | str | None]:
"""Worker-thread body for the regular branch of ``get_artifact``.
Stat probes, MIME sniffing (``mimetypes`` lazily stats the system MIME
database on first use), and full-file reads are all blocking filesystem IO.
Returns a ``(kind, mime_type, payload)`` plan the handler turns into a
response on the loop: ``("file", mime, None)`` (let ``FileResponse`` stream
it), ``("text", mime, str)``, or ``("bytes", mime, bytes)``. Behavior/error
codes match the previous inline logic.
"""
if not actual_path.exists():
raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")
if not actual_path.is_file():
raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")
mime_type, _ = mimetypes.guess_type(actual_path)
# Active content / explicit download is streamed by FileResponse — no read here.
if download or mime_type in ACTIVE_CONTENT_MIME_TYPES:
return ("file", mime_type, None)
if mime_type and mime_type.startswith("text/"):
return ("text", mime_type, actual_path.read_text(encoding="utf-8"))
if is_text_file_by_content(actual_path):
return ("text", mime_type, actual_path.read_text(encoding="utf-8"))
return ("bytes", mime_type, actual_path.read_bytes())
@router.get(
"/threads/{thread_id}/artifacts/{path:path}",
summary="Get Artifact File",
@ -143,21 +189,11 @@ async def get_artifact(thread_id: str, path: str, request: Request, download: bo
skill_file_path = path[: marker_pos + len(".skill")] # e.g., "mnt/user-data/outputs/my-skill.skill"
internal_path = path[marker_pos + len(skill_marker) :] # e.g., "SKILL.md"
actual_skill_path = resolve_thread_virtual_path(thread_id, skill_file_path)
actual_skill_path = await asyncio.to_thread(resolve_thread_virtual_path, thread_id, skill_file_path)
if not actual_skill_path.exists():
raise HTTPException(status_code=404, detail=f"Skill file not found: {skill_file_path}")
# Offload the stat probes + ZIP open/extract + MIME sniff (blocking filesystem IO).
content, mime_type = await asyncio.to_thread(_load_skill_archive_member, actual_skill_path, skill_file_path, internal_path)
if not actual_skill_path.is_file():
raise HTTPException(status_code=400, detail=f"Path is not a file: {skill_file_path}")
# Extract the file from the .skill archive
content = _extract_file_from_skill_archive(actual_skill_path, internal_path)
if content is None:
raise HTTPException(status_code=404, detail=f"File '{internal_path}' not found in skill archive")
# Determine MIME type based on the internal file
mime_type, _ = mimetypes.guess_type(internal_path)
# Add cache headers to avoid repeated ZIP extraction (cache for 5 minutes)
cache_headers = {"Cache-Control": "private, max-age=300"}
download_name = Path(internal_path).name or actual_skill_path.stem
@ -173,30 +209,21 @@ async def get_artifact(thread_id: str, path: str, request: Request, download: bo
except UnicodeDecodeError:
return Response(content=content, media_type=mime_type or "application/octet-stream", headers=cache_headers)
actual_path = resolve_thread_virtual_path(thread_id, path)
actual_path = await asyncio.to_thread(resolve_thread_virtual_path, thread_id, path)
logger.info(f"Resolving artifact path: thread_id={thread_id}, requested_path={path}, actual_path={actual_path}")
if not actual_path.exists():
raise HTTPException(status_code=404, detail=f"Artifact not found: {path}")
# Offload path stat + MIME sniff + file reads (all blocking filesystem IO).
# Active content and explicit downloads are streamed by FileResponse, so the
# worker only reports the kind; inline text/binary payloads are read in-thread.
kind, mime_type, payload = await asyncio.to_thread(_read_artifact_payload, actual_path, path, download)
if not actual_path.is_file():
raise HTTPException(status_code=400, detail=f"Path is not a file: {path}")
mime_type, _ = mimetypes.guess_type(actual_path)
if download:
if kind == "file":
# Always force download for active content types to prevent script
# execution in the application origin when users open generated artifacts.
return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))
# Always force download for active content types to prevent script execution
# in the application origin when users open generated artifacts.
if mime_type in ACTIVE_CONTENT_MIME_TYPES:
return FileResponse(path=actual_path, filename=actual_path.name, media_type=mime_type, headers=_build_attachment_headers(actual_path.name))
if kind == "text":
return PlainTextResponse(content=payload, media_type=mime_type)
if mime_type and mime_type.startswith("text/"):
return PlainTextResponse(content=actual_path.read_text(encoding="utf-8"), media_type=mime_type)
if is_text_file_by_content(actual_path):
return PlainTextResponse(content=actual_path.read_text(encoding="utf-8"), media_type=mime_type)
return Response(content=actual_path.read_bytes(), media_type=mime_type, headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)})
return Response(content=payload, media_type=mime_type, headers={"Content-Disposition": _build_content_disposition("inline", actual_path.name)})

81
backend/tests/blocking_io/test_artifacts_router.py Normal file
View File

@ -0,0 +1,81 @@
"""Regression anchor: serving artifacts must not block the event loop.
``get_artifact`` probes the artifact path (``exists`` / ``is_file``), reads
text/binary content (``read_text`` / ``read_bytes``), sniffs text-ness
(``is_text_file_by_content``), and extracts ``.skill`` archive members — all
blocking filesystem IO. The handler offloads each branch's IO via
``asyncio.to_thread``; if any regresses back onto the event loop, the strict
Blockbuster gate raises ``BlockingError`` and these tests fail.
The ``@require_permission`` decorator is bypassed via ``__wrapped__`` so the
anchor exercises the handler's own filesystem IO, not the authz layer. Imports
sit at module top so any import-time IO runs at collection, outside the gate.
"""
from __future__ import annotations
import asyncio
import zipfile
from pathlib import Path
import pytest
from app.gateway.path_utils import resolve_thread_virtual_path
from app.gateway.routers.artifacts import get_artifact
pytestmark = pytest.mark.asyncio
# The undecorated coroutine (``require_permission`` uses ``functools.wraps``).
_get_artifact = get_artifact.__wrapped__
async def _seed(tmp_path: Path, monkeypatch, thread_id: str, virtual_path: str) -> Path:
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
# Rebuild cached Paths against the tmp home so the artifact resolves under it.
import deerflow.config.paths as paths_mod
monkeypatch.setattr(paths_mod, "_paths", None)
# Test-side path resolution also touches the filesystem (`.resolve()`); offload
# it so this seeding helper doesn't itself trip the gate.
target = await asyncio.to_thread(resolve_thread_virtual_path, thread_id, virtual_path)
await asyncio.to_thread(target.parent.mkdir, parents=True, exist_ok=True)
return target
async def test_get_artifact_text_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
vpath = "mnt/user-data/outputs/notes.txt"
target = await _seed(tmp_path, monkeypatch, "t1", vpath)
await asyncio.to_thread(target.write_text, "hello world", encoding="utf-8")
resp = await _get_artifact("t1", vpath, request=None, download=False)
assert resp.status_code == 200
assert resp.body == b"hello world"
async def test_get_artifact_binary_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
vpath = "mnt/user-data/outputs/blob.bin"
target = await _seed(tmp_path, monkeypatch, "t1", vpath)
payload = b"\x00\x01\x02PNGDATA" # null byte -> binary branch (read_bytes)
await asyncio.to_thread(target.write_bytes, payload)
resp = await _get_artifact("t1", vpath, request=None, download=False)
assert resp.status_code == 200
assert resp.body == payload
async def test_get_artifact_skill_archive_member_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
skill_vpath = "mnt/user-data/outputs/demo.skill"
target = await _seed(tmp_path, monkeypatch, "t1", skill_vpath)
def _build_skill_zip() -> None:
with zipfile.ZipFile(target, "w") as zf:
zf.writestr("SKILL.md", "# demo skill\n")
await asyncio.to_thread(_build_skill_zip)
resp = await _get_artifact("t1", f"{skill_vpath}/SKILL.md", request=None, download=False)
assert resp.status_code == 200
assert b"# demo skill" in resp.body