diff --git a/CHANGELOG.md b/CHANGELOG.md index 14cf099ef..073d957ba 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,7 +3,9 @@ ## [Unreleased] -_No unreleased changes. Entries are moved into their version block when a release is tagged._ +### Fixed + +- **No more 57–70 s cold-startup stalls from the profile skills-stats thundering herd.** At container boot the frontend fires several profile-data requests at once; with `ThreadingHTTPServer` (one thread per request) they all missed the empty skills-stats cache simultaneously and each walked + parsed every profile's skill tree, stacking thousands of concurrent `stat()` calls under Docker's overlay2 filesystem. `_get_profile_skills_stats()` now serializes per-profile with double-checked locking (concurrent misses on one profile collapse to a single compute; independent profiles still compute in parallel), and `list_profiles_api()` single-flights the row build under `_LIST_PROFILES_CACHE_LOCK` so one thread builds while the rest wait for the cached result. The every-call cheap mtime probe (the #4783 out-of-band change-detection contract) is unchanged. (#5364) ## [v0.51.792] — 2026-07-01 diff --git a/api/profiles.py b/api/profiles.py index 7033a1845..20ddbbe5f 100644 --- a/api/profiles.py +++ b/api/profiles.py @@ -1584,6 +1584,31 @@ def switch_profile(name: str, *, process_wide: bool = True) -> dict: _SKILLS_STATS_CACHE: dict[Path, tuple[int, int, int, float]] = {} _SKILLS_STATS_CACHE_TTL = 300.0 # seconds — long because .clear() handles programmatic changes +# Per-profile compute locks (#5364). Without these, concurrent cold-startup +# requests (ThreadingHTTPServer runs one OS thread per request) all miss the +# unlocked _SKILLS_STATS_CACHE at once and each walks + parses the whole skill +# tree simultaneously — a thundering herd that stalled workers 57–70s under +# Docker overlay2. A per-profile lock lets independent profiles compute in +# parallel while collapsing concurrent misses on the SAME profile to a single +# shared compute (double-checked locking below). The lock registry is guarded by +# its own meta-lock and is bounded by the (small) number of profiles. +_SKILLS_STATS_LOCKS: dict[Path, threading.Lock] = {} +_SKILLS_STATS_LOCKS_GUARD = threading.Lock() + + +def _skills_stats_lock_for(profile_dir: Path) -> threading.Lock: + """Return (creating if needed) the per-profile compute lock for profile_dir. + + profile_dir must already be resolved so distinct spellings of the same + directory share one lock. + """ + with _SKILLS_STATS_LOCKS_GUARD: + lock = _SKILLS_STATS_LOCKS.get(profile_dir) + if lock is None: + lock = threading.Lock() + _SKILLS_STATS_LOCKS[profile_dir] = lock + return lock + def _skill_tree_max_mtime_ns(skills_dir: Path, config_path: Path) -> int: """Return the max st_mtime_ns across config.yaml, skill dirs, and SKILL.md files.""" @@ -1725,13 +1750,29 @@ def _get_profile_skills_stats(profile_dir: Path) -> tuple[int, int]: if current_mtime_ns == cached_mtime_ns and now < expiry: return enabled, compat - # Cache miss, mtime changed, or TTL expired — snapshot mtime BEFORE compute - # so any concurrent SKILL.md write during the compute window causes a mismatch - # on the next probe instead of silently serving stale data (TOCTOU). - new_mtime_ns = _skill_tree_max_mtime_ns(skills_dir, config_path) - res = _compute_profile_skills_stats(profile_dir) - _SKILLS_STATS_CACHE[profile_dir] = (res[0], res[1], new_mtime_ns, now + _SKILLS_STATS_CACHE_TTL) - return res + # Cache miss, mtime changed, or TTL expired — serialize per-profile so a + # burst of concurrent misses (cold startup) collapses to ONE compute instead + # of a thundering herd of simultaneous os.walk + SKILL.md parses (#5364). + lock = _skills_stats_lock_for(profile_dir) + with lock: + # Double-checked locking: another thread may have populated a fresh entry + # while we waited for the lock. Reuse it when the mtime we already probed + # still matches and the entry is within its TTL — no second compute. + cached = _SKILLS_STATS_CACHE.get(profile_dir) + if cached is not None: + enabled, compat, cached_mtime_ns, expiry = cached + if current_mtime_ns == cached_mtime_ns and time.time() < expiry: + return enabled, compat + + # Snapshot mtime BEFORE compute so any concurrent SKILL.md write during + # the compute window causes a mismatch on the next probe instead of + # silently serving stale data (TOCTOU). + new_mtime_ns = _skill_tree_max_mtime_ns(skills_dir, config_path) + res = _compute_profile_skills_stats(profile_dir) + _SKILLS_STATS_CACHE[profile_dir] = ( + res[0], res[1], new_mtime_ns, time.time() + _SKILLS_STATS_CACHE_TTL + ) + return res _LIST_PROFILES_CACHE: tuple[list, float] | None = None @@ -1890,14 +1931,21 @@ def list_profiles_api() -> list: 'total_skills': total_count, }] + # Single-flight the build (#5364): hold the cache lock across the row build + # so a cold-startup burst of concurrent requests collapses to ONE build while + # the others wait and then serve the freshly-cached rows — instead of every + # thread rebuilding (each walking all profiles' skill trees) at once. The + # per-profile skills locks taken inside _build_profile_rows_fast are always + # acquired AFTER this lock (never the reverse), so there is no deadlock. with _LIST_PROFILES_CACHE_LOCK: cached = _LIST_PROFILES_CACHE - if cached is not None and now - cached[1] < _LIST_PROFILES_CACHE_TTL: - active = get_active_profile_name() - # Return a fresh copy with is_active recomputed (cheap, per-request). - return [{**p, 'is_active': p['name'] == active} for p in cached[0]] + if cached is not None and now - cached[1] < _LIST_PROFILES_CACHE_TTL: + rows = cached[0] + else: + rows = _build_profile_rows_fast() + if rows is not None: + _LIST_PROFILES_CACHE = (rows, now) - rows = _build_profile_rows_fast() if rows is None: # Fallback: cheap helpers unavailable — use the original (slow) path, # or the default-only dict if hermes_cli isn't importable at all. @@ -1931,9 +1979,6 @@ def list_profiles_api() -> list: }) return result - with _LIST_PROFILES_CACHE_LOCK: - _LIST_PROFILES_CACHE = (rows, now) - active = get_active_profile_name() return [{**p, 'is_active': p['name'] == active} for p in rows] diff --git a/tests/test_issue5364_skills_stats_thundering_herd.py b/tests/test_issue5364_skills_stats_thundering_herd.py new file mode 100644 index 000000000..ea1e582d1 --- /dev/null +++ b/tests/test_issue5364_skills_stats_thundering_herd.py @@ -0,0 +1,298 @@ +"""Tests for the cold-startup thundering-herd fix in api.profiles (#5364). + +Background +---------- +The two-tier mtime cache from #4783 fixed the per-request SKILL.md rescan, but +left two concurrency holes that only bite at container cold start, when the +frontend fires several profile-data requests at once and the caches are empty: + + 1. ``_get_profile_skills_stats`` had NO lock, so concurrent misses on the same + profile each ran ``os.walk`` + parsed every SKILL.md simultaneously. + 2. ``_build_profile_rows_fast`` ran OUTSIDE ``_LIST_PROFILES_CACHE_LOCK`` in + ``list_profiles_api``, so every concurrent request rebuilt all rows (each + walking every profile's skill tree) instead of one building while the rest + waited. + +Under Docker overlay2 with 9 profiles this stacked ~45k concurrent ``stat`` +calls and stalled worker threads for 57–70 s (per the report). + +These tests prove: + * concurrent misses on one profile collapse to a SINGLE compute; + * the per-profile lock registry returns a stable lock per profile; + * a concurrent ``list_profiles_api`` burst builds the rows exactly ONCE; + * the #4783 contract is preserved — the cheap mtime probe still runs on every + call so out-of-band changes stay promptly visible. +""" +import sys +import threading +import time +import types +from pathlib import Path +from unittest.mock import MagicMock, patch + +import pytest + + +# --------------------------------------------------------------------------- +# Import harness (mirrors tests/test_issue4783_profile_skills_mtime_cache.py) +# --------------------------------------------------------------------------- + +def _make_profiles_module(): + """Import api.profiles with minimal stubs for heavy dependencies.""" + stubs = { + "flask": types.ModuleType("flask"), + "yaml": types.ModuleType("yaml"), + "agent": types.ModuleType("agent"), + "agent.skill_utils": types.ModuleType("agent.skill_utils"), + } + + flask_mod = stubs["flask"] + flask_mod.request = MagicMock() + flask_mod.g = MagicMock() + flask_mod.Blueprint = MagicMock(return_value=MagicMock()) + flask_mod.jsonify = MagicMock(side_effect=lambda x: x) + flask_mod.abort = MagicMock() + flask_mod.current_app = MagicMock() + + stubs["yaml"].safe_load = MagicMock(return_value=None) + + su = stubs["agent.skill_utils"] + su.iter_skill_index_files = lambda skills_dir, filename: skills_dir.rglob(filename) + su.parse_frontmatter = MagicMock(return_value=({}, "")) + su.skill_matches_platform = MagicMock(return_value=True) + + for name, mod in stubs.items(): + sys.modules.setdefault(name, mod) + + mod_name = "api.profiles" + if mod_name in sys.modules: + del sys.modules[mod_name] + + api_pkg = types.ModuleType("api") + sys.modules["api"] = api_pkg + + import importlib.util + spec_path = Path(__file__).parent.parent / "api" / "profiles.py" + spec = importlib.util.spec_from_file_location(mod_name, spec_path) + mod = importlib.util.module_from_spec(spec) + sys.modules[mod_name] = mod + try: + spec.loader.exec_module(mod) + except Exception: + pass + return mod + + +@pytest.fixture() +def mod(tmp_path): + try: + m = _make_profiles_module() + assert hasattr(m, "_get_profile_skills_stats") + assert hasattr(m, "_skills_stats_lock_for") + except Exception: + pytest.skip("api.profiles not importable in this environment") + # Reset both caches + the per-profile lock registry for isolation. + if hasattr(m, "_SKILLS_STATS_CACHE"): + m._SKILLS_STATS_CACHE.clear() + if hasattr(m, "_SKILLS_STATS_LOCKS"): + m._SKILLS_STATS_LOCKS.clear() + if hasattr(m, "_LIST_PROFILES_CACHE"): + m._LIST_PROFILES_CACHE = None + yield m + saved = {k: v for k, v in list(sys.modules.items()) + if k == "api" or k.startswith("api.")} + for k in list(saved): + sys.modules.pop(k, None) + + +# --------------------------------------------------------------------------- +# 1. Per-profile compute lock collapses a concurrent-miss herd to one compute +# --------------------------------------------------------------------------- + +class TestConcurrentMissComputesOnce: + def test_concurrent_cache_miss_computes_exactly_once(self, mod, tmp_path): + profile_dir = tmp_path / "p" + (profile_dir / "skills").mkdir(parents=True) + + compute_calls = [] + compute_lock = threading.Lock() + + def _slow_compute(_pdir): + with compute_lock: + compute_calls.append(1) + # Widen the race window so every thread would pile in without a lock. + time.sleep(0.15) + return (3, 5) + + fixed_mtime = 1_700_000_000_000_000_000 + results = [] + results_lock = threading.Lock() + + n_threads = 24 + barrier = threading.Barrier(n_threads) + + def _worker(): + barrier.wait() # release all threads simultaneously (cold-boot burst) + r = mod._get_profile_skills_stats(profile_dir) + with results_lock: + results.append(r) + + with ( + patch.object(mod, "_compute_profile_skills_stats", side_effect=_slow_compute), + patch.object(mod, "_skill_tree_max_mtime_ns", return_value=fixed_mtime), + ): + threads = [threading.Thread(target=_worker) for _ in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert sum(compute_calls) == 1, ( + "concurrent cache misses on ONE profile must collapse to a single " + f"_compute_profile_skills_stats call, got {sum(compute_calls)}" + ) + assert len(results) == n_threads + assert all(r == (3, 5) for r in results), \ + "every waiting thread must see the single shared computed result" + + def test_distinct_profiles_compute_in_parallel(self, mod, tmp_path): + """Different profiles must NOT serialize on each other (per-profile lock, + not a single global lock) — independent trees compute concurrently.""" + n = 4 + dirs = [] + for i in range(n): + d = tmp_path / f"p{i}" + (d / "skills").mkdir(parents=True) + dirs.append(d) + + in_compute = [] + max_concurrent = [0] + gate = threading.Lock() + + def _slow_compute(_pdir): + with gate: + in_compute.append(1) + max_concurrent[0] = max(max_concurrent[0], sum(in_compute)) + time.sleep(0.15) + with gate: + in_compute.pop() + return (1, 1) + + barrier = threading.Barrier(n) + + def _worker(pdir): + barrier.wait() + mod._get_profile_skills_stats(pdir) + + # A single patched probe returns per-path deterministic mtimes so each + # profile is a distinct cache key (distinct lock). + path_mtimes = {Path(d).resolve(): 1_700_000_000_000_000_000 + i + for i, d in enumerate(dirs)} + + def _probe(skills_dir, config_path): + return path_mtimes.get(Path(skills_dir).parent.resolve(), 0) + + with ( + patch.object(mod, "_compute_profile_skills_stats", side_effect=_slow_compute), + patch.object(mod, "_skill_tree_max_mtime_ns", side_effect=_probe), + ): + threads = [threading.Thread(target=_worker, args=(d,)) for d in dirs] + for t in threads: + t.start() + for t in threads: + t.join() + + assert max_concurrent[0] >= 2, ( + "distinct profiles must be able to compute concurrently — a single " + "global lock would force max_concurrent==1" + ) + + +# --------------------------------------------------------------------------- +# 2. Lock registry returns a stable per-profile lock +# --------------------------------------------------------------------------- + +class TestLockRegistry: + def test_same_profile_returns_same_lock(self, mod, tmp_path): + d = (tmp_path / "p").resolve() + l1 = mod._skills_stats_lock_for(d) + l2 = mod._skills_stats_lock_for(d) + assert l1 is l2 + + def test_distinct_profiles_get_distinct_locks(self, mod, tmp_path): + a = (tmp_path / "a").resolve() + b = (tmp_path / "b").resolve() + assert mod._skills_stats_lock_for(a) is not mod._skills_stats_lock_for(b) + + +# --------------------------------------------------------------------------- +# 3. list_profiles_api single-flights the row build under a concurrent burst +# --------------------------------------------------------------------------- + +class TestListProfilesSingleFlight: + def test_concurrent_list_profiles_builds_rows_once(self, mod): + build_calls = [] + build_lock = threading.Lock() + + def _slow_build(): + with build_lock: + build_calls.append(1) + time.sleep(0.15) + return [{"name": "default", "path": "/x"}] + + n_threads = 16 + barrier = threading.Barrier(n_threads) + results = [] + results_lock = threading.Lock() + + def _worker(): + barrier.wait() + r = mod.list_profiles_api() + with results_lock: + results.append(r) + + with ( + patch.object(mod, "_is_isolated_profile_mode", return_value=False), + patch.object(mod, "get_active_profile_name", return_value="default"), + patch.object(mod, "_build_profile_rows_fast", side_effect=_slow_build), + ): + mod._LIST_PROFILES_CACHE = None + threads = [threading.Thread(target=_worker) for _ in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert sum(build_calls) == 1, ( + "a concurrent cold-start burst must build the profile rows exactly " + f"once (single-flight), got {sum(build_calls)}" + ) + assert len(results) == n_threads + assert all(r and r[0]["name"] == "default" for r in results) + + +# --------------------------------------------------------------------------- +# 4. #4783 contract preserved: the cheap mtime probe still runs on every call +# --------------------------------------------------------------------------- + +class TestProbeStillRunsEveryCall: + def test_probe_runs_on_cache_hit(self, mod, tmp_path): + profile_dir = tmp_path / "p" + (profile_dir / "skills").mkdir(parents=True) + + with ( + patch.object(mod, "_compute_profile_skills_stats", + wraps=mod._compute_profile_skills_stats) as mock_compute, + patch.object(mod, "_skill_tree_max_mtime_ns", + wraps=mod._skill_tree_max_mtime_ns) as mock_probe, + ): + mod._get_profile_skills_stats(profile_dir) + compute_after_first = mock_compute.call_count + probe_after_first = mock_probe.call_count + + mod._get_profile_skills_stats(profile_dir) # within TTL, unchanged + + assert mock_compute.call_count == compute_after_first, \ + "expensive compute must be skipped within TTL when unchanged" + assert mock_probe.call_count > probe_after_first, \ + "cheap mtime probe MUST still run on every call (#4783 contract)"