mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
4bab280250
* fix(skills): share skill cache invalidation across Gateway replicas SkillStorage.load_skills() rescans disk and re-reads extensions_config.json on every call, but the prompt layer in deerflow/agents/lead_agent/prompt.py caches the enabled-skills list, the per-(app_config, user_id) entries and the rendered <skill_system> section per process, and every skill mutation in routers/skills.py only refreshed the process that handled it. With several uvicorn workers or several Pods sharing one home volume, a skill installed, edited, deleted, rolled back or toggled through replica A kept being offered as the old skill set by replica B until B restarted, and POST /api/skills/reload had to be sent to every Pod separately. Reuse the MCP cache-reset mechanism. A new generic helper, deerflow.config.shared_reset_marker, owns a hidden JSON marker beside the resolved extensions config (.<config name>.<suffix>.json): publish() replaces it atomically with a random generation under extensions_config_write_lock + extensions_config_file_lock, chaining previous_generation and an optional user_id; SharedResetMarkerTracker polls it at most once per second (monotonic clock, never blocking behind a concurrent poller), adopts the current state silently on its first poll, and classifies a change as user-scoped only when exactly one publication carrying that user happened since the last poll. deerflow/mcp/cache.py now builds its .<config>.mcp-cache-reset.json marker through the same helper with no behavior change. prompt.py publishes .<config>.skills-cache-reset.json via publish_skills_cache_reset(user_id=...) and checks the marker on every cache lookup (global cache, per-config cache, prompt-section LRU, user scoped entries), calling _invalidate_enabled_skills_cache() or invalidate_user_skill_cache(). The publishing process remembers its own generation so it does not retire the caches it just rebuilt. Every mutation tail in routers/skills.py (install/upload, edit, delete, rollback, enable/disable, reload) refreshes locally and then publishes the marker off the event loop; /skills/reload answers scope=shared_config, or scope=process when no extensions config path resolves, and a marker publication failure is a 500 so a 200 never means only one process was refreshed. Docs: README, backend/docs/API.md, skills/config/mcp/gateway AGENTS.md, and frontend skills.mdx (en/zh; also fixes the reference to a non-existent skills/loader.py) describe the shared invalidation and drop the per-Pod guidance. CHANGELOG entry under Unreleased / Fixed. * docs(changelog): link the skills cache reset entry to #6495 * fix(skills): drain the reload tail and keep marker failures out of 404 Two review follow-ups on the shared skills cache reset (#6495). POST /skills/reload awaited the local refresh and the marker publication unshielded. CancelledError is a BaseException, so a caller that disconnected between the two slipped past the `except Exception -> 500` mapping, surfaced no error, and left every peer replica serving the old skill set -- the one outcome the operator hook for external mount writes exists to prevent. The refresh and the publication now run as one `await_drained(_drain_skill_mutation("reload", ...))` tail, the same discipline the install/edit/delete/rollback/toggle tails already use; the exception -> 500 mapping and the scope response are unchanged. Publication errors could surface as 404. resolve_shared_config_path() only swallows the resolution-time FileNotFoundError; a config directory that vanishes between resolution and the lock file / atomic replace still raised FileNotFoundError from the marker write, and the install, edit, delete and rollback handlers mapped that to HTTP 404 with the raw OS error text for a skill that had in fact been applied. publish_skills_cache_reset() now re-raises OSError (and a misconfigured project root's ValueError) as SkillCacheResetPublishError, a RuntimeError that no handler maps to 404/400, so every handler reports it through its generic 500 branch; the message names only the exception type, never a server path. The install handler gains an explicit clause that pins that contract beside its 404 mapping. Tests: cancelling reload after the refresh still publishes the marker; a FileNotFoundError from the marker write yields 500 (not 404) for install, edit, delete and rollback, 500 for the public toggle and the generic reload error; the wrapper type is not an OSError.
289 lines
10 KiB
Python
289 lines
10 KiB
Python
"""Shared cache-reset markers for Gateway processes that share one config directory.
|
|
|
|
``deerflow.config.shared_reset_marker`` is the generic helper behind the MCP
|
|
tools-cache reset (``.<extensions config>.mcp-cache-reset.json``) and the skills
|
|
prompt-cache reset (``.<extensions config>.skills-cache-reset.json``). A writer
|
|
publishes a fresh random generation with an atomic replace; every other process
|
|
compares the marker's ``(mtime, size, sha256)`` signature on its next lookup.
|
|
These tests drive two trackers side by side to simulate two Gateway processes
|
|
sharing one marker file.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
import deerflow.config.shared_reset_marker as shared_reset_marker_module
|
|
from deerflow.config.extensions_config import ExtensionsConfig
|
|
from deerflow.config.file_signature import get_config_signature
|
|
from deerflow.config.shared_reset_marker import (
|
|
SharedResetChange,
|
|
SharedResetMarker,
|
|
SharedResetMarkerTracker,
|
|
resolve_shared_config_path,
|
|
)
|
|
|
|
|
|
class _Clock:
|
|
def __init__(self, start: float = 1000.0) -> None:
|
|
self.now = start
|
|
|
|
def __call__(self) -> float:
|
|
return self.now
|
|
|
|
def advance(self, seconds: float) -> None:
|
|
self.now += seconds
|
|
|
|
|
|
@pytest.fixture()
|
|
def config_path(tmp_path: Path) -> Path:
|
|
path = tmp_path / "extensions_config.json"
|
|
path.write_text(json.dumps({"mcpServers": {}, "skills": {}}), encoding="utf-8")
|
|
return path
|
|
|
|
|
|
def _tracker(marker: SharedResetMarker, config_path: Path | None, clock: _Clock, **kwargs) -> SharedResetMarkerTracker:
|
|
return SharedResetMarkerTracker(marker, resolve_config_path=lambda: config_path, clock=clock, **kwargs)
|
|
|
|
|
|
def test_marker_path_is_hidden_sibling_of_the_config(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
|
|
assert marker.path_for(config_path) == config_path.parent / ".extensions_config.json.skills-cache-reset.json"
|
|
|
|
|
|
def test_marker_path_follows_a_symlinked_config(tmp_path: Path, config_path: Path) -> None:
|
|
link = tmp_path / "linked" / "extensions_config.json"
|
|
link.parent.mkdir()
|
|
link.symlink_to(config_path)
|
|
|
|
assert SharedResetMarker("skills-cache-reset").path_for(link) == config_path.parent / ".extensions_config.json.skills-cache-reset.json"
|
|
|
|
|
|
@pytest.mark.parametrize("suffix", ["", "has/slash", "has\\backslash", ".."])
|
|
def test_marker_rejects_unsafe_suffixes(suffix: str) -> None:
|
|
with pytest.raises(ValueError):
|
|
SharedResetMarker(suffix)
|
|
|
|
|
|
def test_publish_writes_an_atomic_json_marker(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
|
|
generation = marker.publish(config_path, user_id="alice")
|
|
|
|
marker_path = marker.path_for(config_path)
|
|
payload = json.loads(marker_path.read_text(encoding="utf-8"))
|
|
assert payload["version"] == 1
|
|
assert payload["generation"] == generation
|
|
assert payload["previous_generation"] is None
|
|
assert payload["user_id"] == "alice"
|
|
assert payload["published_at"]
|
|
assert not list(config_path.parent.glob(f".{marker_path.name}.*.tmp"))
|
|
|
|
|
|
def test_publish_chains_the_previous_generation_and_omits_user_id_for_global_resets(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
|
|
first = marker.publish(config_path, user_id="alice")
|
|
second = marker.publish(config_path)
|
|
|
|
payload = json.loads(marker.path_for(config_path).read_text(encoding="utf-8"))
|
|
assert first != second
|
|
assert payload["generation"] == second
|
|
assert payload["previous_generation"] == first
|
|
assert "user_id" not in payload
|
|
|
|
|
|
def test_read_tolerates_missing_and_malformed_markers(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
|
|
assert marker.read(config_path) == (None, None)
|
|
assert marker.current_signature(config_path) is None
|
|
assert marker.current_signature(None) is None
|
|
|
|
marker_path = marker.path_for(config_path)
|
|
marker_path.write_text("{not json", encoding="utf-8")
|
|
payload, signature = marker.read(config_path)
|
|
assert payload is None
|
|
assert signature == get_config_signature(marker_path)
|
|
|
|
|
|
def test_two_processes_share_one_marker(config_path: Path) -> None:
|
|
"""Process A publishes; process B sees the invalidation on its next poll."""
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
process_a = _tracker(marker, config_path, clock)
|
|
process_b = _tracker(marker, config_path, clock)
|
|
|
|
# Both processes adopt the current state (no marker yet) without a reset.
|
|
assert process_a.poll() is None
|
|
assert process_b.poll() is None
|
|
|
|
generation = marker.publish(config_path)
|
|
process_a.note_own_publication(generation)
|
|
clock.advance(1.0)
|
|
|
|
assert process_a.poll() is None, "the publishing process already refreshed itself"
|
|
change = process_b.poll()
|
|
assert change == SharedResetChange(user_ids=None, generation=generation)
|
|
clock.advance(1.0)
|
|
assert process_b.poll() is None, "an observed generation is not reported twice"
|
|
|
|
|
|
def test_user_scoped_publication_is_reported_with_its_user(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
process_b = _tracker(marker, config_path, clock)
|
|
assert process_b.poll() is None
|
|
|
|
generation = marker.publish(config_path, user_id="alice")
|
|
clock.advance(1.0)
|
|
|
|
assert process_b.poll() == SharedResetChange(user_ids=frozenset({"alice"}), generation=generation)
|
|
|
|
|
|
def test_multiple_publications_between_polls_fall_back_to_a_global_reset(config_path: Path) -> None:
|
|
"""Only the latest marker survives, so a skipped generation widens the reset."""
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
process_b = _tracker(marker, config_path, clock)
|
|
assert process_b.poll() is None
|
|
|
|
marker.publish(config_path, user_id="alice")
|
|
generation = marker.publish(config_path, user_id="bob")
|
|
clock.advance(1.0)
|
|
|
|
assert process_b.poll() == SharedResetChange(user_ids=None, generation=generation)
|
|
|
|
|
|
def test_own_publication_with_an_interleaved_foreign_write_still_resets(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
process_a = _tracker(marker, config_path, clock)
|
|
assert process_a.poll() is None
|
|
|
|
process_a.note_own_publication(marker.publish(config_path, user_id="alice"))
|
|
foreign = marker.publish(config_path, user_id="bob")
|
|
clock.advance(1.0)
|
|
|
|
assert process_a.poll() == SharedResetChange(user_ids=None, generation=foreign)
|
|
|
|
|
|
def test_poll_is_throttled_to_one_stat_per_interval(config_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
reads: list[Path] = []
|
|
original = shared_reset_marker_module.read_config_with_signature
|
|
|
|
def _counting_read(path: Path):
|
|
reads.append(path)
|
|
return original(path)
|
|
|
|
marker.publish(config_path)
|
|
monkeypatch.setattr(shared_reset_marker_module, "read_config_with_signature", _counting_read)
|
|
tracker = _tracker(marker, config_path, clock, poll_interval_seconds=1.0)
|
|
|
|
assert tracker.poll() is None
|
|
assert len(reads) == 1
|
|
clock.advance(0.5)
|
|
tracker.poll()
|
|
tracker.poll()
|
|
assert len(reads) == 1, "no repeated stat inside the throttle window"
|
|
clock.advance(0.5)
|
|
tracker.poll()
|
|
assert len(reads) == 2
|
|
|
|
|
|
def test_change_inside_the_throttle_window_is_seen_on_the_next_interval(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
tracker = _tracker(marker, config_path, clock)
|
|
assert tracker.poll() is None
|
|
|
|
generation = marker.publish(config_path)
|
|
clock.advance(0.2)
|
|
assert tracker.poll() is None, "throttled: the stat is skipped"
|
|
clock.advance(0.8)
|
|
assert tracker.poll() == SharedResetChange(user_ids=None, generation=generation)
|
|
|
|
|
|
def test_tracker_without_a_config_path_never_reports_a_change(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
tracker = _tracker(marker, None, clock)
|
|
|
|
assert tracker.poll() is None
|
|
marker.publish(config_path)
|
|
clock.advance(1.0)
|
|
assert tracker.poll() is None
|
|
|
|
|
|
def test_marker_deleted_after_adoption_is_a_global_reset(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
marker.publish(config_path)
|
|
tracker = _tracker(marker, config_path, clock)
|
|
assert tracker.poll() is None
|
|
|
|
marker.path_for(config_path).unlink()
|
|
clock.advance(1.0)
|
|
|
|
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
|
|
|
|
|
|
def test_malformed_marker_is_a_global_reset(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
tracker = _tracker(marker, config_path, clock)
|
|
assert tracker.poll() is None
|
|
|
|
marker.path_for(config_path).write_text("{not json", encoding="utf-8")
|
|
clock.advance(1.0)
|
|
|
|
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
|
|
|
|
|
|
def test_config_path_switch_after_adoption_is_a_global_reset(tmp_path: Path, config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
current = {"path": None}
|
|
tracker = SharedResetMarkerTracker(marker, resolve_config_path=lambda: current["path"], clock=clock)
|
|
assert tracker.poll() is None
|
|
|
|
current["path"] = config_path
|
|
clock.advance(1.0)
|
|
|
|
assert tracker.poll() == SharedResetChange(user_ids=None, generation=None)
|
|
|
|
|
|
def test_reset_forgets_the_observed_state(config_path: Path) -> None:
|
|
marker = SharedResetMarker("skills-cache-reset")
|
|
clock = _Clock()
|
|
tracker = _tracker(marker, config_path, clock)
|
|
assert tracker.poll() is None
|
|
marker.publish(config_path)
|
|
|
|
tracker.reset()
|
|
|
|
assert tracker.poll() is None, "after a reset the first poll adopts silently"
|
|
clock.advance(1.0)
|
|
assert tracker.poll() is None
|
|
|
|
|
|
def test_resolve_shared_config_path_treats_a_missing_explicit_file_as_unconfigured(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
missing = tmp_path / "missing.json"
|
|
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(missing))
|
|
|
|
with pytest.raises(FileNotFoundError):
|
|
ExtensionsConfig.resolve_config_path()
|
|
assert resolve_shared_config_path() is None
|
|
|
|
|
|
def test_resolve_shared_config_path_returns_the_explicit_file(config_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
|
|
|
|
assert resolve_shared_config_path() == config_path
|