mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
fix(gateway): load skills off the event loop in the listing and detail routes (#5945)
* fix(gateway): load skills off the event loop in the listing and detail routes
GET /api/skills, GET /api/skills/custom and GET /api/skills/{name} called
load_skills() inline. Resolving the caller's storage builds absolute
paths, and load_skills walks every public and custom skill directory and
parses each SKILL.md, so the work grows with the number of installed
skills. #5747 moved the same call off the loop for get_custom_skill and
documented why; these three routes were missed, and the strict
Blockbuster gate raises BlockingError on each of them.
Offload the load through one shared _load_user_skills helper (also used
by update_skill's reload), and add a gate module covering the three
routes including get_skill's 404 branch.
* docs(changelog): reference #5945 in the skills listing blocking I/O entries
This commit is contained in:
parent
2377a49217
commit
b72753c226
12
CHANGELOG.md
12
CHANGELOG.md
@ -323,6 +323,17 @@ This release closes that milestone with **181 merged pull requests**.
|
||||
|
||||
### Fixed
|
||||
|
||||
- **gateway:** `GET /api/skills`, `GET /api/skills/custom` and
|
||||
`GET /api/skills/{name}` no longer walk the skill directories on the event
|
||||
loop. Each called `load_skills()` inline, which resolves the caller's
|
||||
storage, scans every public and custom skill directory and parses each
|
||||
`SKILL.md` — filesystem work that grows with the number of installed
|
||||
skills. #5747 moved that same call off the loop for the custom-skill
|
||||
content route and documented why; these three routes were missed, so the
|
||||
strict Blockbuster gate raised `BlockingError` on them and, in production,
|
||||
a large or slow skills tree stalled every other request on the worker for
|
||||
the duration of the scan. All three now offload the load with
|
||||
`asyncio.to_thread` through one shared helper. ([#5945])
|
||||
- **gateway:** `GET` and `PUT /api/user-profile` no longer run their
|
||||
filesystem work on the event loop. Both handlers resolved the per-user
|
||||
`USER.md` path (which builds absolute paths on every call), stat'ed, read,
|
||||
@ -6207,4 +6218,5 @@ with **180 merged pull requests** since the first 2.0 milestone tag.
|
||||
[#5928]: https://github.com/bytedance/deer-flow/pull/5928
|
||||
[#5934]: https://github.com/bytedance/deer-flow/pull/5934
|
||||
[#5935]: https://github.com/bytedance/deer-flow/pull/5935
|
||||
[#5945]: https://github.com/bytedance/deer-flow/pull/5945
|
||||
|
||||
|
||||
@ -328,6 +328,12 @@
|
||||
|
||||
### 修复
|
||||
|
||||
- **网关:** `GET /api/skills`、`GET /api/skills/custom` 与 `GET /api/skills/{name}` 不再在事件循环上
|
||||
遍历技能目录。三者此前都内联调用 `load_skills()`:它会解析调用者的存储、扫描所有公共与自定义
|
||||
技能目录并解析每个 `SKILL.md`,工作量随已安装技能数增长。#5747 已经为自定义技能内容路由把同一
|
||||
调用移出事件循环并注明原因,但这三条路由被遗漏了,因此严格的 Blockbuster 门禁会在它们上抛出
|
||||
`BlockingError`;在生产环境中,技能树很大或磁盘很慢时,扫描期间该 worker 上的其他所有请求都会
|
||||
停顿。现在三者通过一个共享辅助函数用 `asyncio.to_thread` 卸载加载过程。([#5945])
|
||||
- **网关:** `GET` 与 `PUT /api/user-profile` 不再在事件循环上执行文件系统操作。这两个处理器
|
||||
此前在循环上直接解析按用户隔离的 `USER.md` 路径(每次调用都会构造绝对路径)、stat、读取、
|
||||
创建用户目录并写入文件,而自定义智能体路由里的其他所有处理器都通过 `asyncio.to_thread`
|
||||
@ -5264,3 +5270,4 @@ DeerFlow 2.0 是围绕"超级智能体"框架的彻底重写,核心包含子
|
||||
[#5928]: https://github.com/bytedance/deer-flow/pull/5928
|
||||
[#5934]: https://github.com/bytedance/deer-flow/pull/5934
|
||||
[#5935]: https://github.com/bytedance/deer-flow/pull/5935
|
||||
[#5945]: https://github.com/bytedance/deer-flow/pull/5945
|
||||
|
||||
@ -187,6 +187,18 @@ def _get_user_skill_storage(config: AppConfig) -> SkillStorage:
|
||||
return get_or_new_user_skill_storage(get_effective_user_id(), app_config=config)
|
||||
|
||||
|
||||
def _load_user_skills(config: AppConfig) -> list[Skill]:
|
||||
"""Load every skill visible to the caller's storage, enabled or not.
|
||||
|
||||
Blocking filesystem IO: always call this via ``asyncio.to_thread``.
|
||||
Resolving the user-scoped storage builds absolute paths, and
|
||||
``load_skills`` walks every public and custom skill directory and parses
|
||||
each ``SKILL.md`` -- work that scales with the number of installed skills
|
||||
(#5747 moved the same call off the loop for ``get_custom_skill``).
|
||||
"""
|
||||
return _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
|
||||
|
||||
def _copy_uploaded_skill_archive(source: BinaryIO) -> Path:
|
||||
"""Copy an uploaded archive to a bounded temporary file off the event loop."""
|
||||
destination: Path | None = None
|
||||
@ -327,7 +339,7 @@ async def list_skills(request: Request, config: AppConfig = Depends(get_config))
|
||||
yields an empty list (fail-closed) or all skills (fail-open).
|
||||
"""
|
||||
try:
|
||||
skills = _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
skills = await asyncio.to_thread(_load_user_skills, config)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to load skills: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to load skills: {str(e)}")
|
||||
@ -447,7 +459,7 @@ async def list_custom_skills(request: Request, config: AppConfig = Depends(get_c
|
||||
surface names the main listing hides.
|
||||
"""
|
||||
try:
|
||||
skills = [skill for skill in _get_user_skill_storage(config).load_skills(enabled_only=False) if skill.category == SkillCategory.CUSTOM]
|
||||
skills = [skill for skill in await asyncio.to_thread(_load_user_skills, config) if skill.category == SkillCategory.CUSTOM]
|
||||
visible_skills = await _filter_visible_skills(request, config, skills)
|
||||
return SkillsListResponse(skills=[_skill_to_response(skill) for skill in visible_skills])
|
||||
except Exception as e:
|
||||
@ -697,7 +709,7 @@ async def rollback_custom_skill(skill_name: str, body: SkillRollbackRequest, req
|
||||
async def get_skill(skill_name: str, request: Request, config: AppConfig = Depends(get_config)) -> SkillResponse:
|
||||
try:
|
||||
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
|
||||
skills = _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
skills = await asyncio.to_thread(_load_user_skills, config)
|
||||
skill = next((s for s in skills if s.name == skill_name), None)
|
||||
|
||||
if skill is None:
|
||||
@ -838,10 +850,7 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
|
||||
else:
|
||||
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
|
||||
|
||||
def _reload_skills() -> list[Skill]:
|
||||
return _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
|
||||
skills = await asyncio.to_thread(_reload_skills)
|
||||
skills = await asyncio.to_thread(_load_user_skills, config)
|
||||
updated_skill = next((s for s in skills if s.name == skill_name), None)
|
||||
|
||||
if updated_skill is None:
|
||||
|
||||
104
backend/tests/blocking_io/test_skills_listing_router.py
Normal file
104
backend/tests/blocking_io/test_skills_listing_router.py
Normal file
@ -0,0 +1,104 @@
|
||||
"""Regression anchors: the skill listing/detail routes must not block the event loop.
|
||||
|
||||
``list_skills``, ``list_custom_skills`` and ``get_skill`` all call
|
||||
``SkillStorage.load_skills()``, which walks every public and custom skill
|
||||
directory and parses each ``SKILL.md`` -- blocking filesystem IO that scales
|
||||
with the number of installed skills. #5747 moved the same call off the loop
|
||||
for ``get_custom_skill`` (and left a comment saying why) but these three
|
||||
routes kept calling it inline. Under the strict Blockbuster gate they raise
|
||||
``BlockingError``; with the fix they offload it via ``asyncio.to_thread``.
|
||||
|
||||
Seeding the custom skill on disk is itself offloaded so only the handlers'
|
||||
own filesystem access is exercised on the loop.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
from fastapi import HTTPException, Request
|
||||
|
||||
# ``_filter_visible_skills`` resolves the caller through ``deps`` helpers that
|
||||
# import ``app.gateway.auth`` lazily; import it here so that one-time module
|
||||
# load happens at collection time and only the handlers' own filesystem access
|
||||
# is exercised on the loop (same reason ``test_agents_router`` builds the app
|
||||
# at import time).
|
||||
import app.gateway.auth # noqa: F401
|
||||
import app.gateway.auth.errors # noqa: F401
|
||||
from app.gateway.routers.skills import get_skill, list_custom_skills, list_skills
|
||||
from deerflow.config.paths import get_paths
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
_SKILL_NAME = "loop-listing-skill"
|
||||
_SKILL_MD = f"---\nname: {_SKILL_NAME}\ndescription: Anchor fixture skill.\n---\n\n# {_SKILL_NAME}\n"
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _isolate_paths(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path))
|
||||
monkeypatch.setattr("deerflow.config.paths._paths", None)
|
||||
|
||||
|
||||
def _config(skills_root: Path) -> SimpleNamespace:
|
||||
return SimpleNamespace(
|
||||
skills=SimpleNamespace(
|
||||
get_skills_path=lambda: skills_root,
|
||||
container_path="/mnt/skills",
|
||||
use="deerflow.skills.storage.local_skill_storage:LocalSkillStorage",
|
||||
),
|
||||
authorization=SimpleNamespace(enabled=False, fail_closed=False),
|
||||
)
|
||||
|
||||
|
||||
def _anonymous_request() -> Request:
|
||||
# No auth state: ``_filter_visible_skills`` returns the input unfiltered,
|
||||
# so the only filesystem work on the loop is the handler's own.
|
||||
return Request({"type": "http", "headers": [], "state": {}})
|
||||
|
||||
|
||||
def _install_skill(skills_root: Path) -> None:
|
||||
(skills_root / "public").mkdir(parents=True, exist_ok=True)
|
||||
skill_dir = get_paths().user_custom_skills_dir(get_effective_user_id()) / _SKILL_NAME
|
||||
skill_dir.mkdir(parents=True, exist_ok=True)
|
||||
(skill_dir / "SKILL.md").write_text(_SKILL_MD, encoding="utf-8")
|
||||
|
||||
|
||||
async def _seeded(tmp_path: Path) -> tuple[SimpleNamespace, Request]:
|
||||
skills_root = tmp_path / "skills"
|
||||
config = _config(skills_root)
|
||||
await asyncio.to_thread(_install_skill, skills_root)
|
||||
request = await asyncio.to_thread(_anonymous_request)
|
||||
return config, request
|
||||
|
||||
|
||||
async def test_list_skills_does_not_block_event_loop(tmp_path: Path) -> None:
|
||||
config, request = await _seeded(tmp_path)
|
||||
|
||||
response = await list_skills(request, config)
|
||||
|
||||
assert [skill.name for skill in response.skills] == [_SKILL_NAME]
|
||||
|
||||
|
||||
async def test_list_custom_skills_does_not_block_event_loop(tmp_path: Path) -> None:
|
||||
config, request = await _seeded(tmp_path)
|
||||
|
||||
response = await list_custom_skills(request, config)
|
||||
|
||||
assert [skill.name for skill in response.skills] == [_SKILL_NAME]
|
||||
|
||||
|
||||
async def test_get_skill_does_not_block_event_loop(tmp_path: Path) -> None:
|
||||
config, request = await _seeded(tmp_path)
|
||||
|
||||
response = await get_skill(_SKILL_NAME, request, config)
|
||||
assert response.name == _SKILL_NAME
|
||||
|
||||
# The 404 branch walks the same directories before deciding.
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await get_skill("never-installed-skill", request, config)
|
||||
assert exc_info.value.status_code == 404
|
||||
Loading…
x
Reference in New Issue
Block a user