diff --git a/api/config.py b/api/config.py index b36b8156..c2d01bfc 100644 --- a/api/config.py +++ b/api/config.py @@ -6907,6 +6907,36 @@ def get_available_models(*, prefer_cache: bool = False, force_refresh: bool = Fa if force_refresh_started_at is not None and _available_models_cache_ts >= force_refresh_started_at: return cached + # A concurrent forced refresh may have started after this caller sampled + # should_wait but before it acquired the lock. Reuse that in-flight build + # instead of launching another one, and preserve this caller's budget. + if ( + force_refresh + and force_refresh_started_at is not None + and _cache_build_in_progress + ): + remaining_budget = 60.0 + if _LIVE_REBUILD_BUDGET_SECONDS > 0: + remaining_budget = max( + 0.0, + _LIVE_REBUILD_BUDGET_SECONDS - (time.monotonic() - force_refresh_started_at), + ) + if remaining_budget > 0: + _cache_build_cv.wait_for( + lambda: not _cache_build_in_progress, + timeout=remaining_budget, + ) + cached = _get_fresh_memory_models_cache(time.monotonic()) + if ( + cached is not None + and _available_models_cache_ts >= force_refresh_started_at + ): + return cached + if _cache_build_in_progress and _LIVE_REBUILD_BUDGET_SECONDS > 0: + if stale_disk_groups is not None: + return copy.deepcopy(stale_disk_groups) + return copy.deepcopy(_static_models_catalog_without_live_probes()) + # Cold path: disk cache hit — use it (fast, no lock contention) if disk_groups is not None and not force_refresh: _available_models_cache = disk_groups diff --git a/tests/test_issue4756_session_visit_model_refresh.py b/tests/test_issue4756_session_visit_model_refresh.py index 35c72fd5..77ec1401 100644 --- a/tests/test_issue4756_session_visit_model_refresh.py +++ b/tests/test_issue4756_session_visit_model_refresh.py @@ -176,6 +176,67 @@ def test_session_visit_overlapping_stale_calls_coalesce_to_single_live_rebuild(t assert rebuild_count == 1 +def test_session_visit_overlapping_stale_calls_do_not_duplicate_over_budget_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.01, 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 + time.sleep(0.05) + 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 == stale_catalog for result in results) + assert rebuild_count == 1 + time.sleep(0.1) + assert cfg._available_models_cache == rebuilt_catalog + assert cfg._cache_build_in_progress is False + + def test_session_visit_fresh_disk_hit_does_not_overwrite_newer_memory_cache(tmp_path, monkeypatch): import api.config as cfg