5e5a41e79f
* fix(profiles): start list cache TTL after row build Slow row builds currently consume the cache TTL before the result is stored. Timestamp the cache at completion so every successful build receives the full TTL. Refs #5364 Assisted-by: Hermes Agent:gpt-5.6-sol Assisted-by: Codex:gpt-5.6-luna * Release exp-v0.52.176: start profile list-cache TTL after row build (#6846) --------- Co-authored-by: starship-s <45587122+starship-s@users.noreply.github.com> Co-authored-by: nesquena-hermes <nesquena-hermes@users.noreply.github.com>
354 lines
14 KiB
Python
354 lines
14 KiB
Python
"""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.
|
||
|
||
Returns ``(mod, saved_modules, injected_names)`` so the caller can fully
|
||
restore ``sys.modules`` on teardown. This test intentionally manipulates
|
||
``sys.modules`` (stubbing flask/yaml/agent and re-importing api.profiles
|
||
in isolation); if those mutations leak, later tests re-import api.config /
|
||
api.routes against the stub ``yaml`` (whose ``safe_load`` returns None) and
|
||
a half-populated ``api`` package, which silently breaks unrelated suites
|
||
(e.g. MCP/provider/config tests whose ``get_config`` patch no longer sees
|
||
real config). So we snapshot every key we touch and restore it exactly.
|
||
"""
|
||
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)
|
||
|
||
# Snapshot every sys.modules key we are about to mutate so teardown can
|
||
# restore the real modules (not just delete them). Keys absent now are
|
||
# recorded as sentinel so teardown removes any stub we introduced.
|
||
_ABSENT = object()
|
||
touched = set(stubs) | {"api", "api.profiles"}
|
||
saved_modules = {k: sys.modules.get(k, _ABSENT) for k in touched}
|
||
|
||
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, saved_modules, _ABSENT
|
||
|
||
|
||
@pytest.fixture()
|
||
def mod(tmp_path):
|
||
saved_modules = {}
|
||
_ABSENT = object()
|
||
try:
|
||
m, saved_modules, _ABSENT = _make_profiles_module()
|
||
assert hasattr(m, "_get_profile_skills_stats")
|
||
assert hasattr(m, "_skills_stats_lock_for")
|
||
except Exception:
|
||
# Restore anything we already mutated before skipping.
|
||
for k, v in saved_modules.items():
|
||
if v is _ABSENT:
|
||
sys.modules.pop(k, None)
|
||
else:
|
||
sys.modules[k] = v
|
||
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
|
||
try:
|
||
yield m
|
||
finally:
|
||
# Fully restore sys.modules: put back the real modules we evicted and
|
||
# drop any stub we introduced, so the manipulation cannot leak into
|
||
# subsequent tests (they re-import the real api.config/api.routes).
|
||
for k, v in saved_modules.items():
|
||
if v is _ABSENT:
|
||
sys.modules.pop(k, None)
|
||
else:
|
||
sys.modules[k] = v
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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)
|
||
|
||
def test_slow_build_gets_full_cache_ttl(self, mod, monkeypatch):
|
||
build_calls = []
|
||
clock = [0.0]
|
||
|
||
def _slow_build():
|
||
build_calls.append(1)
|
||
clock[0] += 0.08
|
||
return [{"name": "default", "path": "/x"}]
|
||
|
||
with (
|
||
patch.object(time, "time", side_effect=lambda: clock[0]),
|
||
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),
|
||
):
|
||
monkeypatch.setattr(mod, "_LIST_PROFILES_CACHE_TTL", 0.05)
|
||
mod._LIST_PROFILES_CACHE = None
|
||
mod.list_profiles_api()
|
||
mod.list_profiles_api()
|
||
|
||
assert len(build_calls) == 1, (
|
||
"cache age must start after a slow build completes, not before it starts"
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 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)"
|