mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-08-14 00:38:42 +00:00
4582a41347
* 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>
102 lines
4.5 KiB
Python
102 lines
4.5 KiB
Python
"""Regression anchor: updating MCP config must not block the event loop.
|
|
|
|
``update_mcp_configuration`` resolves the extensions config path, probes its
|
|
existence, reads the raw JSON, writes the merged config, and reloads it — all
|
|
blocking filesystem IO. The handler offloads the whole read-modify-write via
|
|
``asyncio.to_thread``; if it regresses back onto the event loop, the strict
|
|
Blockbuster gate raises ``BlockingError`` and this test fails.
|
|
|
|
The admin check is patched to a no-op 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 threading
|
|
import time
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from app.gateway.routers import mcp as mcp_router
|
|
from app.gateway.routers.mcp import McpConfigUpdateRequest, McpServerConfigResponse, update_mcp_configuration
|
|
|
|
pytestmark = pytest.mark.asyncio
|
|
|
|
|
|
async def test_update_mcp_configuration_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
|
|
config_path = tmp_path / "extensions_config.json"
|
|
# resolve_config_path() requires the env-pointed file to exist; seed a minimal one.
|
|
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
|
|
|
|
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
|
|
|
|
# An http transport skips the stdio command allowlist check, so the anchor
|
|
# stays focused on the filesystem offload rather than command validation.
|
|
body = McpConfigUpdateRequest(
|
|
mcp_servers={"test-server": McpServerConfigResponse(type="http", url="https://example.test/mcp", description="anchor")},
|
|
)
|
|
|
|
resp = await update_mcp_configuration(request=None, body=body)
|
|
|
|
assert "test-server" in resp.mcp_servers
|
|
# The merged config was actually written to the env-pointed path (offload the
|
|
# stat so the assertion itself doesn't trip the gate).
|
|
assert await asyncio.to_thread(config_path.exists)
|
|
|
|
|
|
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. ``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
|
|
|
|
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
|
|
monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None)
|
|
|
|
state_lock = threading.Lock()
|
|
counters = {"active": 0, "max": 0}
|
|
|
|
def _tracking_reload(*_args, **_kwargs):
|
|
# Runs inside the real worker, under extensions_config_write_lock.
|
|
with state_lock:
|
|
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:
|
|
counters["active"] -= 1
|
|
return SimpleNamespace(mcp_servers={})
|
|
|
|
monkeypatch.setattr(mcp_router, "reload_extensions_config", _tracking_reload)
|
|
|
|
body = McpConfigUpdateRequest(
|
|
mcp_servers={"s": McpServerConfigResponse(type="http", url="https://example.test/mcp")},
|
|
)
|
|
|
|
await asyncio.gather(*[update_mcp_configuration(request=None, body=body) for _ in range(5)])
|
|
|
|
assert counters["max"] == 1, f"config updates were not serialized (max concurrency {counters['max']})"
|