Avoid blocking city force refresh

2026-06-08 00:33:57 +08:00
parent 139e5dbb06
commit 26775693e7
2 changed files with 222 additions and 6 deletions
@@ -1288,6 +1288,122 @@ def test_stale_city_detail_uses_cached_full_payload_while_refreshing(monkeypatch
    assert refresh_calls == 1


+def test_force_refresh_panel_returns_cached_payload_when_refresh_is_slow(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}
+
+    monkeypatch.setenv("POLYWEATHER_CITY_FORCE_REFRESH_TIMEOUT_SEC", "0.01")
+    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, "_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=True,
+            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
+
+    refresh_calls = 0
+    build_inputs = []
+
+    class FakeCache:
+        def get_city_cache(self, kind, city):
+            assert kind == "full"
+            assert city == "paris"
+            return {
+                "payload": {
+                    "city": "paris",
+                    "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
+                },
+            }
+
+    async def fake_run_in_threadpool(fn, *args, **kwargs):
+        if fn is city_api.legacy_routes._refresh_city_full_cache:
+            await asyncio.sleep(0.05)
+        return fn(*args, **kwargs)
+
+    def refresh_full(city, force_refresh):
+        nonlocal refresh_calls
+        refresh_calls += 1
+        return {
+            "city": city,
+            "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]},
+        }
+
+    def build_detail(data, market_slug, target_date, resolution):
+        build_inputs.append(data["hourly"]["temps"][0])
+        return {"city": data["city"], "live_temp": data["hourly"]["temps"][0]}
+
+    city_api._CITY_FULL_REFRESH_INFLIGHT.clear()
+    city_api._CITY_DETAIL_PAYLOAD_CACHE.clear()
+    city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear()
+    city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear()
+
+    monkeypatch.setenv("POLYWEATHER_CITY_FORCE_REFRESH_TIMEOUT_SEC", "0.01")
+    monkeypatch.setattr(city_api, "run_in_threadpool", fake_run_in_threadpool)
+    monkeypatch.setattr(city_api.legacy_routes, "_assert_entitlement", lambda request: None)
+    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, "_overlay_latest_wunderground_current", lambda city, payload: payload)
+    monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full)
+    monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
+
+    async def run_request():
+        payload = await city_api.get_city_detail_aggregate_payload(
+            object(),
+            "Paris",
+            force_refresh=True,
+            resolution="10m",
+        )
+        await asyncio.sleep(0.06)
+        return payload
+
+    result = asyncio.run(run_request())
+
+    assert result["live_temp"] == 20.0
+    assert build_inputs == [20.0]
+    assert refresh_calls == 1
+
+
 def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch):
    import asyncio

@@ -7,7 +7,7 @@ import asyncio
 import threading
 import time
 from copy import deepcopy
-from typing import Any, Dict, List, Optional, Tuple
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple

 from fastapi import HTTPException, Request
 from fastapi.concurrency import run_in_threadpool
@@ -28,6 +28,8 @@ _RECENT_DEB_CACHE_TTL_SEC = max(
 _CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {}
 _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()
 CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int]
 CityChartDetailPayloadCacheKey = Tuple[str, str, str, int]
 CityDetailBatchResponseCacheKey = Tuple[Tuple[str, ...], bool, str, str, str, str]
@@ -64,6 +66,84 @@ def _city_detail_batch_response_cache_ttl() -> float:
    return max(0.0, min(30.0, value))


+def _city_force_refresh_timeout_sec() -> float:
+    try:
+        value = float(os.getenv("POLYWEATHER_CITY_FORCE_REFRESH_TIMEOUT_SEC", "8") or "8")
+    except ValueError:
+        value = 8.0
+    return max(0.01, min(30.0, value))
+
+
+async def _get_cached_city_payload(city: str, kind: str) -> Dict[str, Any]:
+    cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, kind, city)
+    if not isinstance(cached_entry, dict):
+        return {}
+    payload = cached_entry.get("payload")
+    return payload if isinstance(payload, dict) else {}
+
+
+async def _get_or_start_city_force_refresh_task(
+    key: str,
+    refresh_factory: Callable[[], Awaitable[Dict[str, Any]]],
+) -> "asyncio.Task[Dict[str, Any]]":
+    async with _CITY_FORCE_REFRESH_LOCK:
+        task = _CITY_FORCE_REFRESH_INFLIGHT.get(key)
+        if task is None or task.done():
+            task = asyncio.create_task(refresh_factory())
+            _CITY_FORCE_REFRESH_INFLIGHT[key] = task
+
+            def _cleanup(done: "asyncio.Task[Dict[str, Any]]") -> None:
+                if _CITY_FORCE_REFRESH_INFLIGHT.get(key) is done:
+                    _CITY_FORCE_REFRESH_INFLIGHT.pop(key, None)
+                try:
+                    done.result()
+                except Exception as exc:  # pragma: no cover - defensive background guard
+                    logger.warning("city force refresh failed key={}: {}", key, exc)
+
+            task.add_done_callback(_cleanup)
+        return task
+
+
+async def _refresh_city_payload_with_stale_timeout(
+    city: str,
+    kind: str,
+    refresh_factory: Callable[[], Awaitable[Dict[str, Any]]],
+) -> Dict[str, Any]:
+    task = await _get_or_start_city_force_refresh_task(f"{kind}:{city}", refresh_factory)
+    timeout_sec = _city_force_refresh_timeout_sec()
+    try:
+        return await asyncio.wait_for(asyncio.shield(task), timeout=timeout_sec)
+    except asyncio.TimeoutError:
+        cached_payload = await _get_cached_city_payload(city, kind)
+        if cached_payload:
+            logger.warning(
+                "city force refresh timed out city={} kind={} timeout_sec={}; returning stale cache",
+                city,
+                kind,
+                timeout_sec,
+            )
+            return await _overlay_cached_wunderground(city, cached_payload)
+        return await task
+    except Exception:
+        cached_payload = await _get_cached_city_payload(city, kind)
+        if cached_payload:
+            logger.warning("city force refresh failed city={} kind={}; returning stale cache", city, kind)
+            return await _overlay_cached_wunderground(city, cached_payload)
+        raise
+
+
+async def _refresh_city_cache_with_stale_timeout(
+    city: str,
+    kind: str,
+    refresh_fn: Callable[[str, bool], Dict[str, Any]],
+) -> Dict[str, Any]:
+    return await _refresh_city_payload_with_stale_timeout(
+        city,
+        kind,
+        lambda: run_in_threadpool(refresh_fn, city, True),
+    )
+
+
 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,
@@ -203,7 +283,11 @@ def _start_city_full_stale_refresh(city: str) -> None:

 async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
    if force_refresh:
-        return await _refresh_city_full_data(city, True)
+        return await _refresh_city_payload_with_stale_timeout(
+            city,
+            "full",
+            lambda: _refresh_city_full_data(city, True),
+        )
    cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city)
    if cached_entry:
        payload = cached_entry.get("payload") or {}
@@ -521,7 +605,11 @@ async def get_city_detail_payload(
        return await _get_city_full_data(city, force_refresh=force_refresh)
    if detail_mode == "panel":
        if force_refresh:
-            return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, True)
+            return await _refresh_city_cache_with_stale_timeout(
+                city,
+                "panel",
+                legacy_routes._refresh_city_panel_cache,
+            )
        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):
@@ -530,7 +618,11 @@ async def get_city_detail_payload(
        return await run_in_threadpool(legacy_routes._refresh_city_panel_cache, city, False)
    if detail_mode == "nearby":
        if force_refresh:
-            return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, True)
+            return await _refresh_city_cache_with_stale_timeout(
+                city,
+                "nearby",
+                legacy_routes._refresh_city_nearby_cache,
+            )
        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):
@@ -539,7 +631,11 @@ async def get_city_detail_payload(
        return await run_in_threadpool(legacy_routes._refresh_city_nearby_cache, city, False)
    if detail_mode == "market":
        if force_refresh:
-            return await run_in_threadpool(legacy_routes._refresh_city_market_cache, city, True)
+            return await _refresh_city_cache_with_stale_timeout(
+                city,
+                "market",
+                legacy_routes._refresh_city_market_cache,
+            )
        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):
@@ -557,7 +653,11 @@ async def get_city_summary_payload(
 ) -> Dict[str, Any]:
    city = legacy_routes._normalize_city_or_404(name)
    if force_refresh:
-        return await run_in_threadpool(legacy_routes._refresh_city_summary_cache, city, True)
+        return await _refresh_city_cache_with_stale_timeout(
+            city,
+            "summary",
+            legacy_routes._refresh_city_summary_cache,
+        )
    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):