mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-30 09:56:05 +00:00
fix(skills): offload blocking filesystem IO in update_skill and serialize writes (#3565)
* fix(skills): offload update_skill's blocking IO and share the config write lock Re-applied on top of main rather than merged: update_skill has since gained admin gating, user-scoped storage and a PUBLIC vs CUSTOM/LEGACY split, so the offload is applied per path. - PUBLIC: the extensions_config.json read-modify-write (path resolve, snapshot, merge, write, reload) moves to a worker thread via asyncio.to_thread. The payload is built from a snapshot so the cached singleton is never mutated in place while the write is still in flight. - CUSTOM/LEGACY: set_skill_enabled_state is offloaded; the non-user-scoped fallback takes the same shared-file RMW path as PUBLIC. - Both load_skills calls (and storage construction) are offloaded. The RMW lock now lives next to reload_extensions_config as get_extensions_config_write_lock() and is acquired by both the skills router and the MCP router, which performs the same read-modify-write on the same file. Previously each router held its own module-local lock, so once both sides offloaded, a PUT /api/mcp/config could run inside a skill toggle's read->write window and the later write would silently drop the other's change. The lock is keyed by the running event loop rather than being a module-level singleton: asyncio primitives bind to the first loop that awaits them, which makes a plain module-level lock unusable in a process that runs more than one loop. Adds a cross-router regression anchor asserting a skill toggle and an MCP update never overlap inside the RMW (max in-flight 1); it observes 2 when the routers use separate locks. * fix(config): own the extensions_config RMW lock from the worker thread An asyncio.Lock held around `await asyncio.to_thread(...)` protects only the awaiting task. If that task is cancelled the context manager releases the lock immediately while Python keeps running the worker thread, so a second skills or MCP writer could acquire it and operate on extensions_config.json concurrently with the first worker — reopening the lost-update window this was meant to close. The per-event-loop keying had a second hole: writers on different loops got different locks and so did not exclude each other at all. Replace it with a process-wide threading.Lock acquired *inside* the worker that performs the RMW (`_write_extensions_skill_state` and `_apply_mcp_config_update`), so ownership belongs to the thread doing the writing and is held until the write and reload actually finish, regardless of what happens to the caller. A threading.Lock also has no event-loop affinity. Adds a cancellation regression: the skills worker is paused inside the lock, its route task is cancelled, and the MCP writer is started — the MCP RMW must not enter until the skills worker is released. Against the previous asyncio-lock design this test fails with ['skills-enter', 'mcp-enter']. The two existing serialization tests instrumented by replacing the worker functions, which now bypasses the lock under test; they instead patch inside the real workers (each module's reload_extensions_config, the last step under the lock) so the production lock is exercised. --------- Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com> Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
ly-wang19
GitHub
parent
1b5220e35e
commit
4582a41347
@ -10,18 +10,12 @@ from fastapi import APIRouter, HTTPException, Request, status
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
|
||||
from app.gateway.deps import require_admin_user
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpRoutingConfig, McpToolOverride, get_extensions_config, reload_extensions_config
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpRoutingConfig, McpToolOverride, extensions_config_write_lock, get_extensions_config, reload_extensions_config
|
||||
from deerflow.mcp.cache import reset_mcp_tools_cache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
router = APIRouter(prefix="/api", tags=["mcp"])
|
||||
|
||||
# Serializes the read-modify-write of extensions_config.json within this worker
|
||||
# process. Offloading the RMW to a thread removed the implicit serialization the
|
||||
# single-threaded event loop used to provide, so two concurrent
|
||||
# PUT /api/mcp/config calls could otherwise interleave and clobber each other.
|
||||
# (Cross-process writers remain a separate, pre-existing concern.)
|
||||
_mcp_config_write_lock = asyncio.Lock()
|
||||
_ADMIN_REQUIRED_DETAIL = "Admin privileges required to manage MCP configuration."
|
||||
|
||||
|
||||
@ -346,59 +340,60 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
||||
lives here too so the whole read-modify-write is a single worker hop.
|
||||
Returns the reloaded MCP server configs for the response.
|
||||
"""
|
||||
# Get the current config path (or determine where to save it)
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
with extensions_config_write_lock:
|
||||
# Get the current config path (or determine where to save it)
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
|
||||
# If no config file exists, create one in the parent directory (project root)
|
||||
if config_path is None:
|
||||
config_path = Path.cwd().parent / "extensions_config.json"
|
||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||
# If no config file exists, create one in the parent directory (project root)
|
||||
if config_path is None:
|
||||
config_path = Path.cwd().parent / "extensions_config.json"
|
||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||
|
||||
# Load current config to preserve skills
|
||||
current_config = get_extensions_config()
|
||||
# Load current config to preserve skills
|
||||
current_config = get_extensions_config()
|
||||
|
||||
# Load raw (un-resolved) JSON from disk to use as the merge source.
|
||||
# This preserves $VAR placeholders in env values and top-level keys
|
||||
# like mcpInterceptors that would otherwise be lost.
|
||||
raw_servers: dict[str, dict] = {}
|
||||
raw_other_keys: dict = {}
|
||||
if config_path is not None and config_path.exists():
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
raw_data = json.load(f)
|
||||
raw_servers = raw_data.get("mcpServers", {})
|
||||
# Preserve any top-level keys beyond mcpServers/skills
|
||||
for key, value in raw_data.items():
|
||||
if key not in ("mcpServers", "skills"):
|
||||
raw_other_keys[key] = value
|
||||
# Load raw (un-resolved) JSON from disk to use as the merge source.
|
||||
# This preserves $VAR placeholders in env values and top-level keys
|
||||
# like mcpInterceptors that would otherwise be lost.
|
||||
raw_servers: dict[str, dict] = {}
|
||||
raw_other_keys: dict = {}
|
||||
if config_path is not None and config_path.exists():
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
raw_data = json.load(f)
|
||||
raw_servers = raw_data.get("mcpServers", {})
|
||||
# Preserve any top-level keys beyond mcpServers/skills
|
||||
for key, value in raw_data.items():
|
||||
if key not in ("mcpServers", "skills"):
|
||||
raw_other_keys[key] = value
|
||||
|
||||
# Merge incoming server configs with raw on-disk secrets
|
||||
merged_servers: dict[str, McpServerConfigResponse] = {}
|
||||
for name, incoming in body.mcp_servers.items():
|
||||
raw_server = raw_servers.get(name)
|
||||
if raw_server is not None:
|
||||
merged_servers[name] = _merge_preserving_secrets(
|
||||
incoming,
|
||||
McpServerConfigResponse(**raw_server),
|
||||
)
|
||||
else:
|
||||
merged_servers[name] = incoming
|
||||
# Merge incoming server configs with raw on-disk secrets
|
||||
merged_servers: dict[str, McpServerConfigResponse] = {}
|
||||
for name, incoming in body.mcp_servers.items():
|
||||
raw_server = raw_servers.get(name)
|
||||
if raw_server is not None:
|
||||
merged_servers[name] = _merge_preserving_secrets(
|
||||
incoming,
|
||||
McpServerConfigResponse(**raw_server),
|
||||
)
|
||||
else:
|
||||
merged_servers[name] = incoming
|
||||
|
||||
# Build config data preserving all top-level keys from the original file
|
||||
config_data = dict(raw_other_keys)
|
||||
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
|
||||
config_data["skills"] = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
|
||||
# Build config data preserving all top-level keys from the original file
|
||||
config_data = dict(raw_other_keys)
|
||||
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
|
||||
config_data["skills"] = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
|
||||
|
||||
# Write the configuration to file
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
# Write the configuration to file
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
|
||||
logger.info(f"MCP configuration updated and saved to: {config_path}")
|
||||
logger.info(f"MCP configuration updated and saved to: {config_path}")
|
||||
|
||||
# Reload the Gateway configuration and update the global cache. The
|
||||
# agent runtime lives in Gateway, so this keeps API reads and tool
|
||||
# execution aligned after extensions_config.json changes.
|
||||
reloaded_config = reload_extensions_config()
|
||||
return reloaded_config.mcp_servers
|
||||
# Reload the Gateway configuration and update the global cache. The
|
||||
# agent runtime lives in Gateway, so this keeps API reads and tool
|
||||
# execution aligned after extensions_config.json changes.
|
||||
reloaded_config = reload_extensions_config()
|
||||
return reloaded_config.mcp_servers
|
||||
|
||||
|
||||
@router.post(
|
||||
@ -466,10 +461,10 @@ async def update_mcp_configuration(request: Request, body: McpConfigUpdateReques
|
||||
|
||||
# Offload the blocking read-modify-write of extensions_config.json
|
||||
# (path resolve, existence probe, raw read, merged write, reload). The
|
||||
# lock serializes concurrent updates within this process so the RMW stays
|
||||
# atomic now that it no longer runs inline on the event loop.
|
||||
async with _mcp_config_write_lock:
|
||||
reloaded_servers = await asyncio.to_thread(_apply_mcp_config_update, body)
|
||||
# worker takes extensions_config_write_lock for the whole RMW, so it stays
|
||||
# atomic and serialized against the skills router (the other writer of
|
||||
# this file) even if this request is cancelled mid-write.
|
||||
reloaded_servers = await asyncio.to_thread(_apply_mcp_config_update, body)
|
||||
|
||||
servers = {name: _mask_server_config(McpServerConfigResponse(**server.model_dump())) for name, server in reloaded_servers.items()}
|
||||
reset_mcp_tools_cache()
|
||||
|
||||
@ -12,7 +12,7 @@ from app.gateway.deps import get_config, require_admin_user
|
||||
from app.gateway.path_utils import resolve_thread_virtual_path
|
||||
from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache, refresh_skills_system_prompt_cache_async, refresh_user_skills_system_prompt_cache_async
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, extensions_config_write_lock, get_extensions_config, reload_extensions_config
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
from deerflow.skills import Skill
|
||||
from deerflow.skills.installer import SkillAlreadyExistsError, SkillSecurityScanError
|
||||
@ -420,6 +420,38 @@ async def get_skill(skill_name: str, config: AppConfig = Depends(get_config)) ->
|
||||
raise HTTPException(status_code=500, detail=f"Failed to get skill: {str(e)}")
|
||||
|
||||
|
||||
def _write_extensions_skill_state(skill_name: str, enabled: bool) -> None:
|
||||
"""Read-modify-write a skill's enabled state in the shared extensions_config.json.
|
||||
|
||||
Blocking filesystem IO: always call this via ``asyncio.to_thread``. It takes
|
||||
``extensions_config_write_lock`` itself, so that this router and the MCP
|
||||
router (which performs the same RMW on the same file) cannot interleave and
|
||||
drop each other's change. The lock is held by the worker rather than by the
|
||||
awaiting task, so cancelling the request cannot release it mid-write.
|
||||
"""
|
||||
with extensions_config_write_lock:
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
if config_path is None:
|
||||
config_path = Path.cwd().parent / "extensions_config.json"
|
||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||
|
||||
# Work on a deep copy rather than the cached singleton: mutating the
|
||||
# singleton in place would publish the new state to readers before it is
|
||||
# durable on disk, and leave it applied even if the write below fails.
|
||||
# to_file_dict() serializes the full extensions_config.json shape (all
|
||||
# top-level keys), so no field is dropped from the file.
|
||||
extensions_config = get_extensions_config().model_copy(deep=True)
|
||||
extensions_config.skills[skill_name] = SkillStateConfig(enabled=enabled)
|
||||
|
||||
config_data = extensions_config.to_file_dict()
|
||||
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
|
||||
logger.info(f"Skills configuration updated and saved to: {config_path}")
|
||||
reload_extensions_config()
|
||||
|
||||
|
||||
@router.put(
|
||||
"/skills/{skill_name}",
|
||||
response_model=SkillResponse,
|
||||
@ -434,8 +466,14 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
|
||||
await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL)
|
||||
try:
|
||||
skill_name = skill_name.replace("\r\n", "").replace("\n", "")
|
||||
storage = _get_user_skill_storage(config)
|
||||
skills = storage.load_skills(enabled_only=False)
|
||||
|
||||
def _load_storage_and_skills() -> tuple[SkillStorage, list[Skill]]:
|
||||
# Worker thread: storage construction and skill enumeration both walk
|
||||
# the filesystem.
|
||||
storage = _get_user_skill_storage(config)
|
||||
return storage, storage.load_skills(enabled_only=False)
|
||||
|
||||
storage, skills = await asyncio.to_thread(_load_storage_and_skills)
|
||||
skill = next((s for s in skills if s.name == skill_name), None)
|
||||
|
||||
if skill is None:
|
||||
@ -445,38 +483,20 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
|
||||
# CUSTOM / LEGACY skills → per-user _skill_states.json (isolated state)
|
||||
# so that two users with same-named custom skills can toggle independently.
|
||||
if skill.category == SkillCategory.PUBLIC:
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
if config_path is None:
|
||||
config_path = Path.cwd().parent / "extensions_config.json"
|
||||
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
|
||||
|
||||
extensions_config = get_extensions_config()
|
||||
extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled)
|
||||
|
||||
config_data = extensions_config.to_file_dict()
|
||||
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
|
||||
logger.info(f"Skills configuration updated and saved to: {config_path}")
|
||||
reload_extensions_config()
|
||||
# Shared-file RMW. The worker takes extensions_config_write_lock for
|
||||
# the whole read→write window, so it stays serialized against the MCP
|
||||
# router even if this request is cancelled mid-write.
|
||||
await asyncio.to_thread(_write_extensions_skill_state, skill_name, body.enabled)
|
||||
else:
|
||||
# CUSTOM / LEGACY: write per-user state
|
||||
from deerflow.skills.storage.user_scoped_skill_storage import UserScopedSkillStorage
|
||||
|
||||
if isinstance(storage, UserScopedSkillStorage):
|
||||
storage.set_skill_enabled_state(skill_name, body.enabled)
|
||||
await asyncio.to_thread(storage.set_skill_enabled_state, skill_name, body.enabled)
|
||||
else:
|
||||
# Fallback for non-user-scoped storage (unlikely in practice)
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
if config_path is None:
|
||||
config_path = Path.cwd().parent / "extensions_config.json"
|
||||
extensions_config = get_extensions_config()
|
||||
extensions_config.skills[skill_name] = SkillStateConfig(enabled=body.enabled)
|
||||
config_data = extensions_config.to_file_dict()
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
reload_extensions_config()
|
||||
# Fallback for non-user-scoped storage (unlikely in practice):
|
||||
# same shared-file RMW as the PUBLIC branch, same lock.
|
||||
await asyncio.to_thread(_write_extensions_skill_state, skill_name, body.enabled)
|
||||
|
||||
# PUBLIC skill enabled state lives in the global extensions_config.json
|
||||
# and affects every user, so the prompt cache for ALL users must be
|
||||
@ -491,7 +511,10 @@ async def update_skill(skill_name: str, body: SkillUpdateRequest, request: Reque
|
||||
else:
|
||||
await refresh_user_skills_system_prompt_cache_async(get_effective_user_id())
|
||||
|
||||
skills = _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
def _reload_skills() -> list[Skill]:
|
||||
return _get_user_skill_storage(config).load_skills(enabled_only=False)
|
||||
|
||||
skills = await asyncio.to_thread(_reload_skills)
|
||||
updated_skill = next((s for s in skills if s.name == skill_name), None)
|
||||
|
||||
if updated_skill is None:
|
||||
|
||||
@ -3,6 +3,7 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
@ -337,6 +338,25 @@ def get_extensions_config() -> ExtensionsConfig:
|
||||
return _extensions_config
|
||||
|
||||
|
||||
#: Serializes read-modify-write cycles on ``extensions_config.json`` across every
|
||||
#: writer. Both the skills router (skill enable/disable) and the MCP router
|
||||
#: (server config updates) read this file, merge a change and write it back.
|
||||
#: While each RMW ran inline on the event loop they were implicitly serialized;
|
||||
#: once a writer offloads its RMW to a worker thread the loop is free to
|
||||
#: interleave the other writer inside the read->write window, and the second
|
||||
#: write silently drops the first one's change.
|
||||
#:
|
||||
#: This is a ``threading.Lock`` rather than an ``asyncio.Lock``, and it must be
|
||||
#: acquired *inside* the worker that performs the RMW. An asyncio lock held
|
||||
#: around ``await asyncio.to_thread(...)`` protects only the awaiting task: if
|
||||
#: that task is cancelled the context manager releases immediately while the
|
||||
#: worker thread keeps writing, letting a second writer in. Owning the lock from
|
||||
#: the worker keeps it held until the write and reload actually finish. It also
|
||||
#: has no event-loop affinity, so writers running on different loops still
|
||||
#: exclude each other.
|
||||
extensions_config_write_lock = threading.Lock()
|
||||
|
||||
|
||||
def reload_extensions_config(config_path: str | None = None) -> ExtensionsConfig:
|
||||
"""Reload the extensions config from file and update the cached instance.
|
||||
|
||||
|
||||
@ -17,6 +17,7 @@ import asyncio
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
@ -51,15 +52,24 @@ async def test_update_mcp_configuration_does_not_block_event_loop(tmp_path: Path
|
||||
assert await asyncio.to_thread(config_path.exists)
|
||||
|
||||
|
||||
async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None:
|
||||
async def test_concurrent_mcp_updates_are_serialized(tmp_path: Path, monkeypatch) -> None:
|
||||
"""The write lock keeps the offloaded read-modify-write atomic within the process.
|
||||
|
||||
Offloading the RMW to a worker thread dropped the implicit serialization the
|
||||
single-threaded event loop provided. ``_mcp_config_write_lock`` restores it:
|
||||
even with several concurrent ``PUT /api/mcp/config`` calls, only one
|
||||
``_apply_mcp_config_update`` runs at a time. (Without the lock the tracked
|
||||
max concurrency would exceed 1.)
|
||||
single-threaded event loop provided. ``extensions_config_write_lock`` restores
|
||||
it — and, being shared with the skills router (the other writer of this file),
|
||||
also serializes against skill toggles: even with several concurrent
|
||||
``PUT /api/mcp/config`` calls, only one RMW is inside the critical section at a
|
||||
time. (Without the lock the tracked max concurrency would exceed 1.)
|
||||
|
||||
The tracker is injected *inside* the real worker (via ``reload_extensions_config``,
|
||||
the last step under the lock) rather than replacing ``_apply_mcp_config_update``,
|
||||
because the lock now lives in the worker — stubbing the worker out would bypass
|
||||
the very thing under test.
|
||||
"""
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
|
||||
|
||||
async def _noop_admin(_request, **_kwargs) -> None:
|
||||
return None
|
||||
@ -68,20 +78,19 @@ async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None:
|
||||
monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None)
|
||||
|
||||
state_lock = threading.Lock()
|
||||
active = 0
|
||||
max_active = 0
|
||||
counters = {"active": 0, "max": 0}
|
||||
|
||||
def _tracking_apply(_body) -> dict:
|
||||
nonlocal active, max_active
|
||||
def _tracking_reload(*_args, **_kwargs):
|
||||
# Runs inside the real worker, under extensions_config_write_lock.
|
||||
with state_lock:
|
||||
active += 1
|
||||
max_active = max(max_active, active)
|
||||
time.sleep(0.02) # worker thread (off-loop): hold long enough to expose any overlap
|
||||
counters["active"] += 1
|
||||
counters["max"] = max(counters["max"], counters["active"])
|
||||
time.sleep(0.02) # worker thread (off-loop): hold long enough to expose overlap
|
||||
with state_lock:
|
||||
active -= 1
|
||||
return {}
|
||||
counters["active"] -= 1
|
||||
return SimpleNamespace(mcp_servers={})
|
||||
|
||||
monkeypatch.setattr(mcp_router, "_apply_mcp_config_update", _tracking_apply)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", _tracking_reload)
|
||||
|
||||
body = McpConfigUpdateRequest(
|
||||
mcp_servers={"s": McpServerConfigResponse(type="http", url="https://example.test/mcp")},
|
||||
@ -89,4 +98,4 @@ async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None:
|
||||
|
||||
await asyncio.gather(*[update_mcp_configuration(request=None, body=body) for _ in range(5)])
|
||||
|
||||
assert max_active == 1, f"config updates were not serialized (max concurrency {max_active})"
|
||||
assert counters["max"] == 1, f"config updates were not serialized (max concurrency {counters['max']})"
|
||||
|
||||
273
backend/tests/blocking_io/test_skills_update_router.py
Normal file
273
backend/tests/blocking_io/test_skills_update_router.py
Normal file
@ -0,0 +1,273 @@
|
||||
"""Regression anchors for ``update_skill``: no event-loop blocking + serialized writes.
|
||||
|
||||
``app.gateway.routers.skills.update_skill`` toggles a skill's enabled state. For a
|
||||
PUBLIC skill that rewrites the shared ``extensions_config.json``; the skill
|
||||
enumeration, the config read-modify-write, and the reload are blocking filesystem
|
||||
IO, so they are offloaded via ``asyncio.to_thread``. Offloading removes the
|
||||
implicit serialization the single-threaded event loop provided, so the RMW is
|
||||
guarded by ``extensions_config_write_lock`` — shared with the MCP router, which
|
||||
performs the same RMW on the same file.
|
||||
|
||||
- ``test_update_skill_does_not_block_event_loop``: the strict Blockbuster gate
|
||||
fails if the config write regresses back onto the loop (teeth: red pre-fix).
|
||||
- ``test_update_skill_writes_from_snapshot_without_mutating_singleton``: the write
|
||||
payload is built from a snapshot, so the cached ``extensions_config`` singleton
|
||||
is never mutated in place while the write is still in flight.
|
||||
- ``test_update_skill_serializes_concurrent_writes``: two concurrent calls observe
|
||||
a max in-flight RMW count of 1 — red if the lock is removed.
|
||||
- ``test_skill_and_mcp_config_writes_are_serialized``: a skill toggle and an MCP
|
||||
config update never overlap inside the shared-file RMW — red if the two routers
|
||||
go back to separate module-local locks.
|
||||
|
||||
Only the config-infra boundaries (storage / ``get_extensions_config`` / reload /
|
||||
path resolution) are stubbed; the real ``open(config_path, "w")`` write to a tmp
|
||||
file is exercised.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import threading
|
||||
import time
|
||||
from pathlib import Path
|
||||
from types import SimpleNamespace
|
||||
from uuid import UUID
|
||||
|
||||
import pytest
|
||||
|
||||
from app.gateway.routers import mcp as mcp_router
|
||||
from app.gateway.routers import skills as skills_router
|
||||
from app.gateway.routers.mcp import McpConfigUpdateRequest
|
||||
from app.gateway.routers.skills import SkillUpdateRequest, update_skill
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig
|
||||
from deerflow.skills import Skill
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
def _admin_request() -> SimpleNamespace:
|
||||
# ``require_admin_user`` reads ``request.state.user``; AuthMiddleware normally
|
||||
# stamps it. A SimpleNamespace is enough for the direct-call tests.
|
||||
user = SimpleNamespace(id=UUID("11111111-2222-3333-4444-555555555555"), system_role="admin")
|
||||
return SimpleNamespace(state=SimpleNamespace(user=user))
|
||||
|
||||
|
||||
def _make_skill(name: str, *, enabled: bool) -> Skill:
|
||||
skill_dir = Path(f"/tmp/{name}")
|
||||
return Skill(
|
||||
name=name,
|
||||
description=f"Description for {name}",
|
||||
license="MIT",
|
||||
skill_dir=skill_dir,
|
||||
skill_file=skill_dir / "SKILL.md",
|
||||
relative_path=Path(name),
|
||||
category="public",
|
||||
enabled=enabled,
|
||||
)
|
||||
|
||||
|
||||
def _patch_config_infra(monkeypatch, config_path: Path, *, reload_hook=None) -> ExtensionsConfig:
|
||||
mock_storage = SimpleNamespace(load_skills=lambda *, enabled_only: [_make_skill("demo-skill", enabled=True)])
|
||||
shared_config = ExtensionsConfig()
|
||||
monkeypatch.setattr("app.gateway.routers.skills._get_user_skill_storage", lambda _config: mock_storage)
|
||||
monkeypatch.setattr("app.gateway.routers.skills.get_extensions_config", lambda: shared_config)
|
||||
monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", reload_hook or (lambda: None))
|
||||
monkeypatch.setattr(skills_router.ExtensionsConfig, "resolve_config_path", staticmethod(lambda _path=None: config_path))
|
||||
# PUBLIC toggles drop every user's prompt cache; the handler offloads this
|
||||
# sync call, so a no-op keeps the test focused on the config write.
|
||||
monkeypatch.setattr("app.gateway.routers.skills.clear_skills_system_prompt_cache", lambda: None)
|
||||
return shared_config
|
||||
|
||||
|
||||
async def test_update_skill_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
_patch_config_infra(monkeypatch, config_path)
|
||||
|
||||
result = await update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace())
|
||||
|
||||
assert result.name == "demo-skill"
|
||||
# the real config write ran off the loop
|
||||
assert await asyncio.to_thread(config_path.exists)
|
||||
|
||||
|
||||
async def test_update_skill_writes_from_snapshot_without_mutating_singleton(tmp_path: Path, monkeypatch) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
mock_storage = SimpleNamespace(load_skills=lambda *, enabled_only: [_make_skill("demo-skill", enabled=True)])
|
||||
shared_config = ExtensionsConfig(skills={"existing-skill": SkillStateConfig(enabled=True)})
|
||||
|
||||
monkeypatch.setattr("app.gateway.routers.skills._get_user_skill_storage", lambda _config: mock_storage)
|
||||
monkeypatch.setattr("app.gateway.routers.skills.get_extensions_config", lambda: shared_config)
|
||||
monkeypatch.setattr("app.gateway.routers.skills.reload_extensions_config", lambda: None)
|
||||
monkeypatch.setattr(skills_router.ExtensionsConfig, "resolve_config_path", staticmethod(lambda _path=None: config_path))
|
||||
monkeypatch.setattr("app.gateway.routers.skills.clear_skills_system_prompt_cache", lambda: None)
|
||||
|
||||
result = await update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace())
|
||||
|
||||
assert result.name == "demo-skill"
|
||||
# The cached singleton must not have been mutated: the new skill only exists
|
||||
# in the deep copy that was serialized to disk.
|
||||
assert "demo-skill" not in shared_config.skills
|
||||
config_text = await asyncio.to_thread(config_path.read_text, encoding="utf-8")
|
||||
written = json.loads(config_text)
|
||||
assert written["skills"] == {
|
||||
"existing-skill": {"enabled": True},
|
||||
"demo-skill": {"enabled": False},
|
||||
}
|
||||
# to_file_dict() serializes the full shape, so unrelated top-level keys survive.
|
||||
assert written["mcpServers"] == {}
|
||||
assert "middlewares" in written
|
||||
|
||||
|
||||
@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization
|
||||
async def test_update_skill_serializes_concurrent_writes(tmp_path: Path, monkeypatch) -> None:
|
||||
state_lock = threading.Lock()
|
||||
counters = {"active": 0, "max": 0}
|
||||
|
||||
def _tracking_reload() -> None:
|
||||
# Runs inside the offloaded RMW worker (off the loop), so the sleep that
|
||||
# widens the overlap window is allowed under the gate.
|
||||
with state_lock:
|
||||
counters["active"] += 1
|
||||
counters["max"] = max(counters["max"], counters["active"])
|
||||
time.sleep(0.02)
|
||||
with state_lock:
|
||||
counters["active"] -= 1
|
||||
|
||||
_patch_config_infra(monkeypatch, tmp_path / "extensions_config.json", reload_hook=_tracking_reload)
|
||||
|
||||
await asyncio.gather(
|
||||
update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()),
|
||||
update_skill("demo-skill", SkillUpdateRequest(enabled=True), _admin_request(), SimpleNamespace()),
|
||||
)
|
||||
|
||||
# The shared asyncio.Lock must serialize the offloaded read-modify-write.
|
||||
assert counters["max"] == 1
|
||||
|
||||
|
||||
@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization
|
||||
async def test_skill_and_mcp_config_writes_are_serialized(tmp_path: Path, monkeypatch) -> None:
|
||||
"""A skill toggle and an MCP update must not interleave on extensions_config.json.
|
||||
|
||||
Both routers read-modify-write the same file from a worker thread. With
|
||||
separate locks the loop is free to run the MCP RMW inside the skills RMW's
|
||||
read→write window, and the later write silently drops the other's change. The
|
||||
shared ``extensions_config_write_lock`` closes that window.
|
||||
|
||||
Both sides are instrumented *inside* their real workers (via each module's
|
||||
``reload_extensions_config``, the last step under the lock) rather than by
|
||||
stubbing the workers out — the lock lives in the worker, so replacing it would
|
||||
bypass the thing under test.
|
||||
"""
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
|
||||
|
||||
state_lock = threading.Lock()
|
||||
counters = {"active": 0, "max": 0}
|
||||
|
||||
def _enter_rmw() -> None:
|
||||
with state_lock:
|
||||
counters["active"] += 1
|
||||
counters["max"] = max(counters["max"], counters["active"])
|
||||
time.sleep(0.02)
|
||||
with state_lock:
|
||||
counters["active"] -= 1
|
||||
|
||||
# Skills side: real _write_extensions_skill_state (and its lock) runs.
|
||||
_patch_config_infra(monkeypatch, config_path, reload_hook=_enter_rmw)
|
||||
|
||||
# MCP side: real _apply_mcp_config_update (and its lock) runs; only admin,
|
||||
# validation and the reload are stubbed.
|
||||
async def _noop_admin(_request, **_kwargs) -> None:
|
||||
return None
|
||||
|
||||
def _tracking_reload(*_args, **_kwargs):
|
||||
_enter_rmw()
|
||||
return SimpleNamespace(mcp_servers={})
|
||||
|
||||
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
|
||||
monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", _tracking_reload)
|
||||
|
||||
await asyncio.gather(
|
||||
update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()),
|
||||
mcp_router.update_mcp_configuration(_admin_request(), McpConfigUpdateRequest(mcp_servers={})),
|
||||
)
|
||||
|
||||
assert counters["max"] == 1
|
||||
|
||||
|
||||
@pytest.mark.allow_blocking_io # gate-exempt: needs real worker-thread overlap to observe serialization
|
||||
async def test_cancelled_writer_keeps_the_lock_until_its_worker_finishes(tmp_path: Path, monkeypatch) -> None:
|
||||
"""Cancelling the awaiting task must not release the RMW to another writer.
|
||||
|
||||
An ``asyncio.Lock`` held around ``await asyncio.to_thread(...)`` protects only
|
||||
the awaiting task: cancelling it releases the lock immediately while Python
|
||||
keeps running the worker thread, so a second writer could enter and operate on
|
||||
``extensions_config.json`` concurrently with the first. Owning a
|
||||
``threading.Lock`` from inside the worker keeps the section held until the
|
||||
write and reload actually finish.
|
||||
|
||||
Here the skills worker is paused after it has written and is inside the lock;
|
||||
its route task is then cancelled and the MCP writer is started. The MCP RMW
|
||||
must not enter until the skills worker is released.
|
||||
"""
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
|
||||
|
||||
order: list[str] = []
|
||||
order_lock = threading.Lock()
|
||||
skills_inside = threading.Event()
|
||||
release_skills = threading.Event()
|
||||
mcp_cache_reset = threading.Event()
|
||||
|
||||
def _skills_reload() -> None:
|
||||
# Inside the real _write_extensions_skill_state, under the lock, after the
|
||||
# config write has already been committed to disk.
|
||||
with order_lock:
|
||||
order.append("skills-enter")
|
||||
skills_inside.set()
|
||||
release_skills.wait(timeout=5)
|
||||
with order_lock:
|
||||
order.append("skills-exit")
|
||||
|
||||
def _mcp_reload(*_args, **_kwargs):
|
||||
with order_lock:
|
||||
order.append("mcp-enter")
|
||||
return SimpleNamespace(mcp_servers={})
|
||||
|
||||
async def _noop_admin(_request, **_kwargs) -> None:
|
||||
return None
|
||||
|
||||
_patch_config_infra(monkeypatch, config_path, reload_hook=_skills_reload)
|
||||
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
|
||||
monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", mcp_cache_reset.set)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", _mcp_reload)
|
||||
|
||||
skills_task = asyncio.create_task(update_skill("demo-skill", SkillUpdateRequest(enabled=False), _admin_request(), SimpleNamespace()))
|
||||
assert await asyncio.to_thread(skills_inside.wait, 5), "skills worker never entered the critical section"
|
||||
|
||||
# Cancel the awaiting task while its worker thread is still inside the RMW.
|
||||
skills_task.cancel()
|
||||
with pytest.raises(asyncio.CancelledError):
|
||||
await skills_task
|
||||
|
||||
mcp_task = asyncio.create_task(mcp_router.update_mcp_configuration(_admin_request(), McpConfigUpdateRequest(mcp_servers={})))
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# The cancelled request's worker still owns the section.
|
||||
with order_lock:
|
||||
assert "mcp-enter" not in order, f"MCP writer entered while the cancelled worker was still inside: {order}"
|
||||
|
||||
release_skills.set()
|
||||
await mcp_task
|
||||
|
||||
with order_lock:
|
||||
assert order == ["skills-enter", "skills-exit", "mcp-enter"], order
|
||||
|
||||
# The non-cancelled writer still completed its cache invalidation.
|
||||
assert mcp_cache_reset.is_set()
|
||||
Loading…
x
Reference in New Issue
Block a user