mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-25 14:06:18 +00:00
fix(gateway): keep custom-skill rollback filesystem IO off the event loop (#5751)
* fix(gateway): keep custom-skill rollback filesystem IO off the event loop Closes #5747. Three blocking sites remained on the Gateway event loop in the rollback route and its shared response reader, all offloaded through asyncio.to_thread following the #5729 pattern: - storage.validate_skill_markdown_content() creates a temp dir and writes SKILL.md into it (site 1) - the post-scan read of the file being replaced (site 2, named by #5729's test docstring) - _read_custom_skill_response(): load_skills() walks every skill directory and read_custom_skill() opens SKILL.md — O(installed skills) loop time paid per accepted rollback and per GET /skills/custom/{name} (site 3) A blocking_io regression drives the full accepted-rollback path (past the security scan) under the strict gate; the two pre-existing cases only covered the pre-scan branches. * fix(gateway): keep the skill element type in the response-parts closure Review nit on #5751: tuple[object, ...] erased the element type for the _skill_to_response call; Skill is already imported in this module.
This commit is contained in:
parent
1b2898c367
commit
736b5a4216
@ -505,12 +505,22 @@ async def get_custom_skill(skill_name: str, request: Request, config: AppConfig
|
|||||||
async def _read_custom_skill_response(skill_name: str, config: AppConfig) -> CustomSkillContentResponse:
|
async def _read_custom_skill_response(skill_name: str, config: AppConfig) -> CustomSkillContentResponse:
|
||||||
try:
|
try:
|
||||||
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
|
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
|
||||||
storage = _get_user_skill_storage(config)
|
|
||||||
skills = storage.load_skills(enabled_only=False)
|
def _load_response_parts() -> tuple[Skill | None, str | None]:
|
||||||
skill = next((s for s in skills if s.name == skill_name and s.category == SkillCategory.CUSTOM), None)
|
# Worker thread: load_skills walks every skill directory and
|
||||||
|
# read_custom_skill opens SKILL.md — blocking filesystem IO that
|
||||||
|
# scales with the number of installed skills (#5747).
|
||||||
|
storage = _get_user_skill_storage(config)
|
||||||
|
skills = storage.load_skills(enabled_only=False)
|
||||||
|
skill = next((s for s in skills if s.name == skill_name and s.category == SkillCategory.CUSTOM), None)
|
||||||
|
if skill is None:
|
||||||
|
return None, None
|
||||||
|
return skill, storage.read_custom_skill(skill_name)
|
||||||
|
|
||||||
|
skill, content = await asyncio.to_thread(_load_response_parts)
|
||||||
if skill is None:
|
if skill is None:
|
||||||
raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
|
raise HTTPException(status_code=404, detail=f"Custom skill '{skill_name}' not found")
|
||||||
return CustomSkillContentResponse(**_skill_to_response(skill).model_dump(), content=storage.read_custom_skill(skill_name))
|
return CustomSkillContentResponse(**_skill_to_response(skill).model_dump(), content=content)
|
||||||
except HTTPException:
|
except HTTPException:
|
||||||
raise
|
raise
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
@ -637,11 +647,17 @@ async def rollback_custom_skill(skill_name: str, body: SkillRollbackRequest, req
|
|||||||
target_content = record.get("prev_content")
|
target_content = record.get("prev_content")
|
||||||
if target_content is None:
|
if target_content is None:
|
||||||
raise HTTPException(status_code=400, detail="Selected history entry has no previous content to roll back to")
|
raise HTTPException(status_code=400, detail="Selected history entry has no previous content to roll back to")
|
||||||
storage.validate_skill_markdown_content(skill_name, target_content)
|
await asyncio.to_thread(storage.validate_skill_markdown_content, skill_name, target_content)
|
||||||
static_findings = await _scan_static_skill_markdown_or_raise(skill_name, target_content, app_config=config)
|
static_findings = await _scan_static_skill_markdown_or_raise(skill_name, target_content, app_config=config)
|
||||||
scan = await scan_skill_content(target_content, executable=False, location=f"{skill_name}/{SKILL_MD_FILE}", app_config=config, static_findings=static_findings)
|
scan = await scan_skill_content(target_content, executable=False, location=f"{skill_name}/{SKILL_MD_FILE}", app_config=config, static_findings=static_findings)
|
||||||
skill_file = storage.get_custom_skill_file(skill_name)
|
|
||||||
current_content = skill_file.read_text(encoding="utf-8") if skill_file.exists() else None
|
def _read_current_content() -> str | None:
|
||||||
|
# Worker thread: the post-scan read of the file being replaced is
|
||||||
|
# blocking filesystem IO (#5747), same rule as the history read.
|
||||||
|
skill_file = storage.get_custom_skill_file(skill_name)
|
||||||
|
return skill_file.read_text(encoding="utf-8") if skill_file.exists() else None
|
||||||
|
|
||||||
|
current_content = await asyncio.to_thread(_read_current_content)
|
||||||
history_entry = {
|
history_entry = {
|
||||||
"action": "rollback",
|
"action": "rollback",
|
||||||
"author": "human",
|
"author": "human",
|
||||||
|
|||||||
@ -19,6 +19,7 @@ import asyncio
|
|||||||
import json
|
import json
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
|
from unittest.mock import AsyncMock
|
||||||
from uuid import UUID
|
from uuid import UUID
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
@ -84,3 +85,23 @@ async def test_rollback_history_read_does_not_block_event_loop() -> None:
|
|||||||
|
|
||||||
assert excinfo.value.status_code == 400
|
assert excinfo.value.status_code == 400
|
||||||
assert "history_index is out of range" in str(excinfo.value.detail)
|
assert "history_index is out of range" in str(excinfo.value.detail)
|
||||||
|
|
||||||
|
|
||||||
|
async def test_rollback_accepted_path_does_not_block_event_loop(monkeypatch) -> None:
|
||||||
|
"""The accepted-rollback path (validate → scan → current-content read →
|
||||||
|
write → append → response read) must keep every filesystem operation off
|
||||||
|
the loop (#5747)."""
|
||||||
|
await asyncio.to_thread(_install_skill)
|
||||||
|
await asyncio.to_thread(
|
||||||
|
_write_history,
|
||||||
|
[{"action": "edit", "ts": 1, "prev_content": _SKILL_MD, "new_content": _SKILL_MD}],
|
||||||
|
)
|
||||||
|
monkeypatch.setattr(
|
||||||
|
"app.gateway.routers.skills.scan_skill_content",
|
||||||
|
AsyncMock(return_value=SimpleNamespace(decision="allow", reason="ok", static_findings=[])),
|
||||||
|
)
|
||||||
|
config = AppConfig.model_validate({"sandbox": {"use": "test"}})
|
||||||
|
|
||||||
|
response = await rollback_custom_skill(_SKILL_NAME, SkillRollbackRequest(history_index=0), _admin_request(), config)
|
||||||
|
|
||||||
|
assert response.content == _SKILL_MD
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user