Return stale city panel cache while refreshing
This commit is contained in:
@@ -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
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user