fix #5368

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
 

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]
 

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)"