From 26775693e783ed5debfc68fb5b4c556dc11179eb Mon Sep 17 00:00:00 2001 From: "2569718930@qq.com" <2569718930@qq.com> Date: Mon, 8 Jun 2026 00:33:57 +0800 Subject: [PATCH] Avoid blocking city force refresh --- tests/test_web_observability.py | 116 ++++++++++++++++++++++++++++++++ web/services/city_api.py | 112 ++++++++++++++++++++++++++++-- 2 files changed, 222 insertions(+), 6 deletions(-) diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 607c71b4..894f01ce 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -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 diff --git a/web/services/city_api.py b/web/services/city_api.py index e8d19fe3..35db29f1 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -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):