fix(#5047): coalesce session-visit refresh followers

This commit is contained in:
Rod Boev 2026-06-27 06:13:16 -04:00 committed by nesquena-hermes
parent db15233c4f
commit 53d44d3795
2 changed files with 63 additions and 3 deletions

View file

@ -6855,6 +6855,7 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa
# If another thread has already started the cold path, we will wait for
# its result rather than running the cold path concurrently.
should_wait = _cache_build_in_progress
force_refresh_started_at = time.monotonic() if force_refresh else None
# Check config mtime OUTSIDE the lock so this cheap check doesn't serialize
# concurrent requests. Must come before any config reads in the cold path.
@ -6899,9 +6900,11 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa
# Serve from memory cache if fresh
now = time.monotonic()
if not force_refresh:
cached = _get_fresh_memory_models_cache(now)
if cached is not None:
cached = _get_fresh_memory_models_cache(now)
if cached is not None:
if not force_refresh:
return cached
if force_refresh_started_at is not None and _available_models_cache_ts >= force_refresh_started_at:
return cached
# Cold path: disk cache hit — use it (fast, no lock contention)

View file

@ -119,6 +119,63 @@ def test_session_visit_stale_profile_cache_revalidates_with_live_rebuild(tmp_pat
assert calls == [{"force_refresh": True}]
def test_session_visit_overlapping_stale_calls_coalesce_to_single_live_rebuild(tmp_path, monkeypatch):
import api.config as cfg
import threading
from concurrent.futures import ThreadPoolExecutor
_reset_models_memory_cache(monkeypatch)
stale_catalog = _catalog("stale-model")
rebuilt_catalog = _catalog("rebuilt-model")
config_path = tmp_path / "config.yaml"
config_path.write_text("{}", encoding="utf-8")
cache_path = tmp_path / "models_cache.profile.json"
cache_path.write_text("{}", encoding="utf-8")
old = time.time() - 600.0
os.utime(cache_path, (old, old))
fingerprint = {"profile": "demo"}
stale_load_counts = {}
stale_load_lock = threading.Lock()
second_load_gate = threading.Barrier(2)
rebuild_count = 0
rebuild_lock = threading.Lock()
monkeypatch.setattr(cfg, "_SESSION_VISIT_MODELS_FRESHNESS_SECONDS", 300.0, raising=False)
monkeypatch.setattr(cfg, "_LIVE_REBUILD_BUDGET_SECONDS", 0.0, raising=False)
monkeypatch.setattr(cfg, "_get_config_path", lambda: config_path)
monkeypatch.setattr(cfg, "_cfg_path", config_path, raising=False)
monkeypatch.setattr(cfg, "_cfg_mtime", config_path.stat().st_mtime, raising=False)
monkeypatch.setattr(cfg, "_get_models_cache_path", lambda: cache_path)
monkeypatch.setattr(cfg, "_load_models_cache_from_disk", lambda: None)
monkeypatch.setattr(cfg, "_models_cache_source_fingerprint", lambda: fingerprint)
monkeypatch.setattr(cfg, "_save_models_cache_to_disk", lambda _cache: None)
def _load_stale_models_cache_from_disk():
ident = threading.get_ident()
with stale_load_lock:
count = stale_load_counts.get(ident, 0) + 1
stale_load_counts[ident] = count
if count == 2:
second_load_gate.wait(timeout=5)
return stale_catalog
def _invoke_models_rebuild(_builder):
nonlocal rebuild_count
with rebuild_lock:
rebuild_count += 1
return rebuilt_catalog
monkeypatch.setattr(cfg, "_load_stale_models_cache_from_disk", _load_stale_models_cache_from_disk)
monkeypatch.setattr(cfg, "_invoke_models_rebuild", _invoke_models_rebuild)
with ThreadPoolExecutor(max_workers=2) as executor:
futures = [executor.submit(cfg.get_available_models_for_session_visit) for _ in range(2)]
results = [future.result(timeout=10) for future in futures]
assert all(result == rebuilt_catalog for result in results)
assert rebuild_count == 1
def test_session_visit_fresh_disk_hit_does_not_overwrite_newer_memory_cache(tmp_path, monkeypatch):
import api.config as cfg