diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 894f01ce..60b306ad 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -1339,6 +1339,59 @@ def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(monkeyp assert refresh_calls == 1 +def test_stale_panel_returns_cached_payload_while_refreshing(monkeypatch): + import asyncio + + refresh_calls = 0 + + class FakeCache: + def get_city_cache(self, kind, city): + assert kind == "panel" + assert city == "paris" + return { + "payload": { + "name": "paris", + "deb": {"prediction": 20.0}, + "from_cache": True, + }, + } + + async def fake_run_in_threadpool(fn, *args, **kwargs): + if fn is city_api.legacy_routes._refresh_city_panel_cache: + await asyncio.sleep(0.05) + return fn(*args, **kwargs) + + def refresh_panel(city, force_refresh): + nonlocal refresh_calls + refresh_calls += 1 + return {"name": city, "deb": {"prediction": 21.0}, "from_cache": False} + + city_api._CITY_STALE_REFRESH_TASKS.clear() + + monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool) + monkeypatch.setattr(city_api.legacy_routes, "_normalize_city_or_404", lambda name: name.strip().lower()) + monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache()) + monkeypatch.setattr(city_api.legacy_routes, "_city_cache_is_fresh", lambda entry, ttl: False) + monkeypatch.setattr(city_api.legacy_routes, "_overlay_latest_wunderground_current", lambda city, payload: payload) + monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_panel_cache", refresh_panel) + + async def run_request(): + payload = await city_api.get_city_detail_payload( + object(), + "Paris", + force_refresh=False, + depth="panel", + ) + await asyncio.sleep(0.06) + return payload + + result = asyncio.run(run_request()) + + assert result["from_cache"] is True + assert result["deb"]["prediction"] == 20.0 + assert refresh_calls == 1 + + def test_force_refresh_full_detail_returns_cached_payload_when_refresh_is_slow(monkeypatch): import asyncio diff --git a/web/services/city_api.py b/web/services/city_api.py index 35db29f1..691272cf 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -30,6 +30,7 @@ _CITY_FULL_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _CITY_FULL_REFRESH_LOCK = asyncio.Lock() _CITY_FORCE_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} _CITY_FORCE_REFRESH_LOCK = asyncio.Lock() +_CITY_STALE_REFRESH_TASKS: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int] CityChartDetailPayloadCacheKey = Tuple[str, str, str, int] CityDetailBatchResponseCacheKey = Tuple[Tuple[str, ...], bool, str, str, str, str] @@ -144,6 +145,37 @@ async def _refresh_city_cache_with_stale_timeout( ) +def _start_city_cache_stale_refresh( + city: str, + kind: str, + refresh_fn: Callable[[str, bool], Dict[str, Any]], +) -> None: + normalized = str(city or "").strip().lower() + cache_kind = str(kind or "").strip().lower() + if not normalized or not cache_kind: + return + key = f"{cache_kind}:{normalized}" + existing = _CITY_STALE_REFRESH_TASKS.get(key) + if existing is not None and not existing.done(): + return + + async def _run_refresh() -> Dict[str, Any]: + return await run_in_threadpool(refresh_fn, normalized, False) + + task = asyncio.create_task(_run_refresh()) + _CITY_STALE_REFRESH_TASKS[key] = task + + def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None: + if _CITY_STALE_REFRESH_TASKS.get(key) is done: + _CITY_STALE_REFRESH_TASKS.pop(key, None) + try: + done.result() + except Exception as exc: # pragma: no cover - defensive background guard + logger.warning("city stale refresh failed city={} kind={}: {}", normalized, cache_kind, exc) + + task.add_done_callback(_cleanup) + + async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: return await run_in_threadpool( legacy_routes._overlay_latest_wunderground_current, @@ -613,6 +645,10 @@ async def get_city_detail_payload( cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "panel", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_PANEL_CACHE_TTL_SEC): + payload = cached_entry.get("payload") or {} + if payload: + _start_city_cache_stale_refresh(city, "panel", legacy_routes._refresh_city_panel_cache) + return await _overlay_cached_wunderground(city, payload) return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False) @@ -626,6 +662,10 @@ async def get_city_detail_payload( cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "nearby", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_NEARBY_CACHE_TTL_SEC): + payload = cached_entry.get("payload") or {} + if payload: + _start_city_cache_stale_refresh(city, "nearby", legacy_routes._refresh_city_nearby_cache) + return await _overlay_cached_wunderground(city, payload) return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False) @@ -639,6 +679,10 @@ async def get_city_detail_payload( cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "market", city) if cached_entry: if not legacy_routes._market_analysis_cache_is_fresh(cached_entry): + payload = cached_entry.get("payload") or {} + if payload: + _start_city_cache_stale_refresh(city, "market", legacy_routes._refresh_city_market_cache) + return await _overlay_cached_wunderground(city, payload) return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, False) @@ -661,6 +705,10 @@ async def get_city_summary_payload( cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "summary", city) if cached_entry: if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_SUMMARY_CACHE_TTL_SEC): + payload = cached_entry.get("payload") or {} + if payload: + _start_city_cache_stale_refresh(city, "summary", legacy_routes._refresh_city_summary_cache) + return await _overlay_cached_wunderground(city, payload) return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False) return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, False)