diff --git a/frontend/components/dashboard/ScanTerminalDashboard.tsx b/frontend/components/dashboard/ScanTerminalDashboard.tsx index 827cff53..133b8be6 100644 --- a/frontend/components/dashboard/ScanTerminalDashboard.tsx +++ b/frontend/components/dashboard/ScanTerminalDashboard.tsx @@ -77,6 +77,8 @@ const TrainingDashboard = dynamic( }, ); +const ONLINE_USERS_REFRESH_MS = 5 * 60_000; + function createEmptyAccess(loading = true): ProAccessState { return { loading, @@ -425,14 +427,22 @@ function PolyWeatherTerminal({ useEffect(() => { const fetchOnline = () => { + if (typeof document !== "undefined" && document.visibilityState === "hidden") return; fetch("/api/ops/online-users", { headers: { Accept: "application/json" } }) .then((r) => (r.ok ? r.json() : null)) .then((d) => { if (d?.online != null) setOnlineCount(d.online); }) .catch(() => {}); }; fetchOnline(); - const id = setInterval(fetchOnline, 60_000); - return () => clearInterval(id); + const handleVisibilityChange = () => { + if (document.visibilityState === "visible") fetchOnline(); + }; + document.addEventListener("visibilitychange", handleVisibilityChange); + const id = setInterval(fetchOnline, ONLINE_USERS_REFRESH_MS); + return () => { + document.removeEventListener("visibilitychange", handleVisibilityChange); + clearInterval(id); + }; }, []); const [gridCols, setGridCols] = useState(() => { diff --git a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx index bea1b0df..2644ffbd 100644 --- a/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx +++ b/frontend/components/dashboard/scan-terminal/LiveTemperatureThresholdChart.tsx @@ -101,6 +101,10 @@ function formatCityLocalDate(tzOffsetSeconds: number | null | undefined) { return `${y}-${m}-${d}`; } +function getLiveTempFromHourly(data: HourlyForecast) { + return validNumber(data?.airportCurrent?.temp) ?? validNumber(data?.airportPrimary?.temp) ?? null; +} + function getWundergroundDailyHigh(hourly: HourlyForecast) { return validNumber(hourly?.wundergroundCurrent?.max_so_far) ?? null; } @@ -365,6 +369,8 @@ export function LiveTemperatureThresholdChart({ .then((data) => { if (cancelled || !data) return; hasLoadedHourlyDetailRef.current = true; + const temp = getLiveTempFromHourly(data); + if (temp !== null) setLiveTemp(temp); setHourly(data); }) .catch(() => {}) @@ -377,15 +383,6 @@ export function LiveTemperatureThresholdChart({ if (typeof document !== "undefined" && document.visibilityState === "hidden") return; if (Date.now() - lastPatchAtRef.current < 2 * 60_000) return; - fetch(`/api/city/${encodeURIComponent(city)}/summary`) - .then((res) => (res.ok ? res.json() : null)) - .then((payload) => { - if (cancelled || !payload) return; - const temp = validNumber(payload?.current?.temp); - if (temp !== null) setLiveTemp(temp); - }) - .catch(() => {}); - refreshFullDetail(); }; @@ -403,19 +400,12 @@ export function LiveTemperatureThresholdChart({ const refreshForegroundFullDetail = () => { lastPatchAtRef.current = Date.now(); - fetch(`/api/city/${encodeURIComponent(city)}/summary`) - .then((res) => (res.ok ? res.json() : null)) - .then((payload) => { - if (cancelled || !payload) return; - const temp = validNumber(payload?.current?.temp); - if (temp !== null) setLiveTemp(temp); - }) - .catch(() => {}); - fetchHourlyForecastForCity(city, { ignoreCache: true, resolution: targetResolution }) .then((data) => { if (cancelled || !data) return; hasLoadedHourlyDetailRef.current = true; + const temp = getLiveTempFromHourly(data); + if (temp !== null) setLiveTemp(temp); setHourly(data); }) .catch(() => {}); diff --git a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts index f635bbf0..86c9f265 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts @@ -49,6 +49,10 @@ export async function runTests() { path.join(projectRoot, "components", "dashboard", "scan-terminal", "temperature-chart-logic.ts"), "utf8", ); + const dashboardSource = fs.readFileSync( + path.join(projectRoot, "components", "dashboard", "ScanTerminalDashboard.tsx"), + "utf8", + ); assert( querySource.includes("useSsePatchVersion") && @@ -91,10 +95,16 @@ export async function runTests() { chartLogicSource.includes("primeCityDetailCache"), "visible terminal chart detail fetches should be coalesced into one batch request and prime the shared chart cache", ); + const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\n}\n\nfunction fetchCityDetailWithTimeout/)?.[0] || ""; assert( - chartLogicSource.includes("options.ignoreCache\n ? runQueuedHourlyDetailRequest") && - chartLogicSource.includes(": queueCityDetailBatch(city, resParam)"), - "normal first-paint city detail requests should enter the batch queue before single-request concurrency limiting", + fetchHourlyBlock.includes("queueCityDetailBatch(city, resParam)") && + !fetchHourlyBlock.includes("runQueuedHourlyDetailRequest"), + "first-paint and background city detail refreshes should both enter the batch queue before falling back to single requests", + ); + assert( + dashboardSource.includes("ONLINE_USERS_REFRESH_MS = 5 * 60_000") && + dashboardSource.includes('document.visibilityState === "hidden"'), + "terminal online-user presence should refresh slowly and pause while the browser tab is hidden", ); assert( chartSource.includes("IntersectionObserver") && diff --git a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts index 2bac47fe..c835d3b4 100644 --- a/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts +++ b/frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts @@ -182,6 +182,10 @@ export function runTests() { !foregroundRefreshBlock.includes("setIsHourlyLoading(true)"), "foreground resume refresh should update full detail immediately in the background without showing the loading overlay", ); + assert( + !chart.includes("/api/city/${encodeURIComponent(city)}/summary"), + "visible chart fallback and foreground refresh should not issue a separate summary request after requesting full detail", + ); assert(chart.includes("viewMode"), "temperature chart must expose a view mode for DEB-peak auto view versus full-day view"); assert(chart.includes('useState<"auto" | "full">("full")'), "temperature chart must default every city panel to the all-day view"); assert( diff --git a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts index 0dc547d1..fecf4f41 100644 --- a/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts +++ b/frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts @@ -1098,7 +1098,10 @@ async function flushCityDetailBatch(resolution: string) { return; } try { - resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution)); + resolveBatchWaiters( + waiters, + await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)), + ); } catch (error) { rejectBatchWaiters(waiters, error); } @@ -1109,7 +1112,10 @@ async function flushCityDetailBatch(resolution: string) { cities.map(async (city) => { const waiters = queue.waiters.get(city); try { - resolveBatchWaiters(waiters, await fetchSingleHourlyForecastForCity(city, resolution)); + resolveBatchWaiters( + waiters, + await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)), + ); } catch (fallbackError) { rejectBatchWaiters(waiters, fallbackError || error); } @@ -1158,11 +1164,7 @@ async function fetchHourlyForecastForCity( const pending = _hourlyRequestCache.get(requestKey); if (pending) return pending; - const request = ( - options.ignoreCache - ? runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resParam)) - : queueCityDetailBatch(city, resParam) - ) + const request = queueCityDetailBatch(city, resParam) .finally(() => { _hourlyRequestCache.delete(requestKey); }); diff --git a/src/data_collection/weather_sources.py b/src/data_collection/weather_sources.py index bee76b32..6da550a5 100644 --- a/src/data_collection/weather_sources.py +++ b/src/data_collection/weather_sources.py @@ -946,6 +946,9 @@ class WeatherDataCollector(OpenMeteoCacheMixin, SettlementSourceMixin, MetarSour for key in list(self._wunderground_historical_cache.keys()): if key.startswith(prefix): self._wunderground_historical_cache.pop(key, None) + for key in list(getattr(self, "_wunderground_negative_cache", {}).keys()): + if key.startswith(prefix): + self._wunderground_negative_cache.pop(key, None) def _uses_fahrenheit(self, city_lower: str) -> bool: return city_lower in self.US_CITIES diff --git a/src/data_collection/wunderground_sources.py b/src/data_collection/wunderground_sources.py index 21726d5d..39692648 100644 --- a/src/data_collection/wunderground_sources.py +++ b/src/data_collection/wunderground_sources.py @@ -73,6 +73,13 @@ class WundergroundHistoricalMixin: self._wunderground_historical_cache = {} if not hasattr(self, "_wunderground_historical_cache_lock"): self._wunderground_historical_cache_lock = threading.Lock() + if not hasattr(self, "_wunderground_negative_cache"): + self._wunderground_negative_cache = {} + if not hasattr(self, "wunderground_negative_cache_ttl_sec"): + self.wunderground_negative_cache_ttl_sec = max( + 60, + int(os.getenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900")), + ) if not hasattr(self, "wunderground_historical_cache_ttl_sec"): self.wunderground_historical_cache_ttl_sec = max( 30, @@ -239,6 +246,15 @@ class WundergroundHistoricalMixin: cached = self._wunderground_historical_cache.get(cache_key) if cached and now_ts - float(cached.get("t", 0)) < self.wunderground_historical_cache_ttl_sec: return dict(cached["d"]) + negative_cached = self._wunderground_negative_cache.get(cache_key) + if negative_cached and now_ts - float(negative_cached.get("t", 0)) < self.wunderground_negative_cache_ttl_sec: + record_source_call( + "wunderground", + "historical", + "negative_cached", + (time.perf_counter() - started) * 1000.0, + ) + return None current_url = ( "https://api.weather.com/v1/location/" @@ -286,6 +302,16 @@ class WundergroundHistoricalMixin: location_id, exc, ) + if ( + isinstance(exc, httpx.HTTPStatusError) + and exc.response is not None + and 400 <= exc.response.status_code < 500 + ): + with self._wunderground_historical_cache_lock: + self._wunderground_negative_cache[cache_key] = { + "t": time.time(), + "status": exc.response.status_code, + } record_source_call( "wunderground", "historical", @@ -396,6 +422,7 @@ class WundergroundHistoricalMixin: "t": time.time(), "d": payload, } + self._wunderground_negative_cache.pop(cache_key, None) record_source_call( "wunderground", "historical", diff --git a/tests/test_web_observability.py b/tests/test_web_observability.py index 5653e562..7cb39344 100644 --- a/tests/test_web_observability.py +++ b/tests/test_web_observability.py @@ -564,7 +564,183 @@ def test_city_detail_batch_endpoint_builds_multiple_cached_details(monkeypatch): assert sorted(payload["details"]) == ["paris", "shanghai"] assert payload["details"]["shanghai"]["resolution"] == "10m" assert payload["details"]["paris"]["overlay_city"] == "paris" - assert calls == [("shanghai", "10m"), ("paris", "10m")] + assert sorted(calls) == [("paris", "10m"), ("shanghai", "10m")] + + +def test_city_detail_batch_endpoint_limits_backend_concurrency(monkeypatch): + import asyncio + + active = 0 + max_active = 0 + + async def fake_run_in_threadpool(fn, *args, **kwargs): + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + try: + await asyncio.sleep(0.01) + return fn(*args, **kwargs) + finally: + active -= 1 + + monkeypatch.setenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "2") + 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, "_city_cache_is_fresh", lambda entry, ttl: False) + + class FakeCache: + def get_city_cache(self, kind, city): + assert kind == "full" + return None + + def refresh_full(city, force_refresh): + return { + "city": city, + "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]}, + } + + def build_detail(data, market_slug, target_date, resolution): + return { + "city": data["city"], + "hourly": data["hourly"], + "resolution": resolution, + } + + monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache()) + monkeypatch.setattr(city_api.legacy_routes, "_refresh_city_full_cache", refresh_full) + monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail) + + response = client.get("/api/cities/detail-batch?cities=a,b,c,d,e&resolution=10m&limit=5") + + assert response.status_code == 200 + assert response.json()["cities"] == ["a", "b", "c", "d", "e"] + assert max_active <= 2 + + +def test_concurrent_city_detail_requests_share_same_full_cache_refresh(monkeypatch): + import asyncio + + refresh_calls = 0 + build_calls = 0 + + class FakeCache: + payload = None + + def get_city_cache(self, kind, city): + assert kind == "full" + if self.payload is None: + return None + return {"payload": self.payload} + + fake_cache = FakeCache() + + async def fake_run_in_threadpool(fn, *args, **kwargs): + if fn is city_api.legacy_routes._refresh_city_full_cache: + await asyncio.sleep(0.02) + return fn(*args, **kwargs) + + def refresh_full(city, force_refresh): + nonlocal refresh_calls + refresh_calls += 1 + fake_cache.payload = { + "city": city, + "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]}, + } + return fake_cache.payload + + def build_detail(data, market_slug, target_date, resolution): + nonlocal build_calls + build_calls += 1 + return { + "city": data["city"], + "hourly": data["hourly"], + "resolution": resolution, + } + + 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", fake_cache) + 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_two_requests(): + return await asyncio.gather( + city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"), + city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m"), + ) + + results = asyncio.run(run_two_requests()) + + assert [item["city"] for item in results] == ["paris", "paris"] + assert refresh_calls == 1 + assert build_calls == 1 + + +def test_force_refresh_invalidates_short_city_detail_payload_cache(monkeypatch): + import asyncio + + build_calls = 0 + refreshed_payloads = [ + { + "city": "paris", + "local_date": "2026-05-30", + "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]}, + }, + { + "city": "paris", + "local_date": "2026-05-30", + "hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [21.0]}, + }, + ] + + class FakeCache: + def get_city_cache(self, kind, city): + assert kind == "full" + return None + + async def fake_run_in_threadpool(fn, *args, **kwargs): + return fn(*args, **kwargs) + + def refresh_full(city, force_refresh): + assert city == "paris" + assert refreshed_payloads + return refreshed_payloads.pop(0) + + def build_detail(data, market_slug, target_date, resolution): + nonlocal build_calls + build_calls += 1 + return { + "city": data["city"], + "live_temp": data["hourly"]["temps"][0], + "resolution": resolution, + } + + city_api._CITY_DETAIL_PAYLOAD_CACHE.clear() + city_api._CITY_DETAIL_PAYLOAD_CACHE_TS.clear() + city_api._CITY_DETAIL_PAYLOAD_INFLIGHT.clear() + + 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, "_refresh_city_full_cache", refresh_full) + monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail) + + first = asyncio.run(city_api.get_city_detail_aggregate_payload(object(), "Paris", resolution="10m")) + second = asyncio.run( + city_api.get_city_detail_aggregate_payload( + object(), + "Paris", + resolution="10m", + force_refresh=True, + ), + ) + + assert first["live_temp"] == 20.0 + assert second["live_temp"] == 21.0 + assert build_calls == 2 def test_payment_runtime_endpoint_returns_shape(): diff --git a/tests/test_wunderground_historical.py b/tests/test_wunderground_historical.py index 871ed14f..f497614c 100644 --- a/tests/test_wunderground_historical.py +++ b/tests/test_wunderground_historical.py @@ -180,6 +180,55 @@ def test_fetch_wunderground_historical_keeps_fahrenheit_for_us_cities(monkeypatc assert payload["max_so_far"] == 75 +def test_fetch_wunderground_historical_negative_caches_client_errors(monkeypatch, tmp_path): + collector = _collector(monkeypatch, tmp_path) + monkeypatch.setenv("WUNDERGROUND_NEGATIVE_CACHE_TTL_SEC", "900") + calls = [] + + def fake_get(url: str, **_kwargs): + calls.append(url) + return httpx.Response( + 400, + json={"errors": [{"message": "invalid location"}]}, + request=httpx.Request("GET", url), + ) + + monkeypatch.setattr(collector, "_http_get", fake_get) + + first = collector.fetch_wunderground_historical( + "istanbul", + use_fahrenheit=False, + utc_offset=10800, + local_date="2026-05-31", + ) + second = collector.fetch_wunderground_historical( + "istanbul", + use_fahrenheit=False, + utc_offset=10800, + local_date="2026-05-31", + ) + + assert first is None + assert second is None + assert len(calls) == 2 + + collector._evict_city_caches( + city="istanbul", + lat=None, + lon=None, + use_fahrenheit=False, + ) + third = collector.fetch_wunderground_historical( + "istanbul", + use_fahrenheit=False, + utc_offset=10800, + local_date="2026-05-31", + ) + + assert third is None + assert len(calls) == 4 + + def test_fetch_all_sources_attaches_wunderground_historical(monkeypatch, tmp_path): collector = _collector(monkeypatch, tmp_path) called = {} diff --git a/web/services/city_api.py b/web/services/city_api.py index 1fe45777..36db7e09 100644 --- a/web/services/city_api.py +++ b/web/services/city_api.py @@ -22,6 +22,22 @@ _RECENT_DEB_CACHE_TTL_SEC = max( 60, int(os.getenv("POLYWEATHER_CITIES_DEB_RECENT_CACHE_TTL_SEC", "300") or "300"), ) +_CITY_FULL_REFRESH_INFLIGHT: Dict[str, "asyncio.Task[Dict[str, Any]]"] = {} +_CITY_FULL_REFRESH_LOCK = asyncio.Lock() +CityDetailPayloadCacheKey = Tuple[str, str, str, str, str, int] +_CITY_DETAIL_PAYLOAD_CACHE: Dict[CityDetailPayloadCacheKey, Dict[str, Any]] = {} +_CITY_DETAIL_PAYLOAD_CACHE_TS: Dict[CityDetailPayloadCacheKey, float] = {} +_CITY_DETAIL_PAYLOAD_INFLIGHT: Dict[CityDetailPayloadCacheKey, "asyncio.Task[Dict[str, Any]]"] = {} +_CITY_DETAIL_PAYLOAD_EPOCH: Dict[str, int] = {} +_CITY_DETAIL_PAYLOAD_LOCK = asyncio.Lock() + + +def _city_detail_payload_cache_ttl() -> float: + try: + value = float(os.getenv("POLYWEATHER_CITY_DETAIL_PAYLOAD_CACHE_TTL_SEC", "8") or "8") + except ValueError: + value = 8.0 + return max(0.0, min(30.0, value)) async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Dict[str, Any]: @@ -32,6 +48,134 @@ async def _overlay_cached_wunderground(city: str, payload: Dict[str, Any]) -> Di ) +async def _refresh_city_full_cache_singleflight(city: str, force_refresh: bool) -> Dict[str, Any]: + key = f"{city}:{bool(force_refresh)}" + async with _CITY_FULL_REFRESH_LOCK: + task = _CITY_FULL_REFRESH_INFLIGHT.get(key) + if task is None: + task = asyncio.create_task( + run_in_threadpool(legacy_routes._refresh_city_full_cache, city, force_refresh), + ) + _CITY_FULL_REFRESH_INFLIGHT[key] = task + try: + return await task + finally: + if task.done(): + async with _CITY_FULL_REFRESH_LOCK: + if _CITY_FULL_REFRESH_INFLIGHT.get(key) is task: + _CITY_FULL_REFRESH_INFLIGHT.pop(key, None) + + +async def _invalidate_city_detail_payload_cache(city: str) -> None: + normalized = str(city or "").strip().lower() + if not normalized: + return + async with _CITY_DETAIL_PAYLOAD_LOCK: + _CITY_DETAIL_PAYLOAD_EPOCH[normalized] = _CITY_DETAIL_PAYLOAD_EPOCH.get(normalized, 0) + 1 + old_keys = [key for key in _CITY_DETAIL_PAYLOAD_CACHE if key[0] == normalized] + for key in old_keys: + _CITY_DETAIL_PAYLOAD_CACHE.pop(key, None) + _CITY_DETAIL_PAYLOAD_CACHE_TS.pop(key, None) + + +async def _refresh_city_full_data(city: str, force_refresh: bool) -> Dict[str, Any]: + await _invalidate_city_detail_payload_cache(city) + return await _refresh_city_full_cache_singleflight(city, force_refresh) + + +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) + cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) + if cached_entry: + if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): + return await _refresh_city_full_data(city, False) + return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) + return await _refresh_city_full_data(city, False) + + +def _city_detail_payload_cache_key( + data: Dict[str, Any], + market_slug: Optional[str], + target_date: Optional[str], + resolution: Optional[str], +) -> CityDetailPayloadCacheKey: + city = str(data.get("city") or data.get("name") or "").strip().lower() + fingerprint = str( + data.get("updated_at_ts") + or data.get("updated_at") + or data.get("local_time") + or data.get("local_date") + or id(data) + ) + generation = _CITY_DETAIL_PAYLOAD_EPOCH.get(city, 0) + return ( + city, + str(resolution or "10m"), + str(market_slug or ""), + str(target_date or ""), + fingerprint, + generation, + ) + + +async def _build_city_detail_payload_cached( + data: Dict[str, Any], + market_slug: Optional[str], + target_date: Optional[str], + resolution: Optional[str], +) -> Dict[str, Any]: + ttl = _city_detail_payload_cache_ttl() + if ttl <= 0: + return await run_in_threadpool( + legacy_routes._build_city_detail_payload, + data, + market_slug, + target_date, + resolution, + ) + + key = _city_detail_payload_cache_key(data, market_slug, target_date, resolution) + now_ts = time.time() + async with _CITY_DETAIL_PAYLOAD_LOCK: + cached = _CITY_DETAIL_PAYLOAD_CACHE.get(key) + cached_ts = _CITY_DETAIL_PAYLOAD_CACHE_TS.get(key, 0.0) + if cached is not None and now_ts - cached_ts < ttl: + return cached + task = _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key) + if task is None: + task = asyncio.create_task( + run_in_threadpool( + legacy_routes._build_city_detail_payload, + data, + market_slug, + target_date, + resolution, + ), + ) + _CITY_DETAIL_PAYLOAD_INFLIGHT[key] = task + try: + payload = await task + finally: + if task.done(): + async with _CITY_DETAIL_PAYLOAD_LOCK: + if _CITY_DETAIL_PAYLOAD_INFLIGHT.get(key) is task: + _CITY_DETAIL_PAYLOAD_INFLIGHT.pop(key, None) + + async with _CITY_DETAIL_PAYLOAD_LOCK: + _CITY_DETAIL_PAYLOAD_CACHE[key] = payload + _CITY_DETAIL_PAYLOAD_CACHE_TS[key] = time.time() + if len(_CITY_DETAIL_PAYLOAD_CACHE) > 256: + oldest_keys = sorted( + _CITY_DETAIL_PAYLOAD_CACHE_TS, + key=lambda item: _CITY_DETAIL_PAYLOAD_CACHE_TS.get(item, 0.0), + )[:64] + for old_key in oldest_keys: + _CITY_DETAIL_PAYLOAD_CACHE.pop(old_key, None) + _CITY_DETAIL_PAYLOAD_CACHE_TS.pop(old_key, None) + return payload + + def _default_deb_recent() -> Dict[str, object]: return { "tier": "other", @@ -167,14 +311,7 @@ async def get_city_detail_payload( else: detail_mode = "panel" if detail_mode == "full": - if force_refresh: - return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True) - cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) - if cached_entry: - if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): - return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) - return await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - return await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) + 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) @@ -233,20 +370,9 @@ async def get_city_detail_aggregate_payload( ) -> Dict[str, Any]: legacy_routes._assert_entitlement(request) city = legacy_routes._normalize_city_or_404(name) - if force_refresh: - data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, True) - else: - cached_entry = await run_in_threadpool(legacy_routes._CACHE_DB.get_city_cache, "full", city) - if cached_entry: - if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC): - data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) - else: - data = await _overlay_cached_wunderground(city, cached_entry.get("payload") or {}) - else: - data = await run_in_threadpool(legacy_routes._refresh_city_full_cache, city, False) + data = await _get_city_full_data(city, force_refresh=force_refresh) - return await run_in_threadpool( - legacy_routes._build_city_detail_payload, + return await _build_city_detail_payload_cached( data, market_slug, target_date, @@ -271,7 +397,7 @@ def _parse_batch_city_names(raw_cities: str, *, limit: int) -> List[str]: return out -def _build_city_detail_batch_item( +async def _build_city_detail_batch_item_async( city: str, *, force_refresh: bool, @@ -279,22 +405,8 @@ def _build_city_detail_batch_item( target_date: Optional[str], resolution: Optional[str], ) -> Tuple[str, Dict[str, Any]]: - if force_refresh: - data = legacy_routes._refresh_city_full_cache(city, True) - else: - cached_entry = legacy_routes._CACHE_DB.get_city_cache("full", city) - if cached_entry and legacy_routes._city_cache_is_fresh( - cached_entry, - legacy_routes.CITY_FULL_CACHE_TTL_SEC, - ): - data = legacy_routes._overlay_latest_wunderground_current( - city, - cached_entry.get("payload") or {}, - ) - else: - data = legacy_routes._refresh_city_full_cache(city, False) - - detail = legacy_routes._build_city_detail_payload( + data = await _get_city_full_data(city, force_refresh=force_refresh) + detail = await _build_city_detail_payload_cached( data, market_slug, target_date, @@ -303,6 +415,14 @@ def _build_city_detail_batch_item( return city, detail +def _city_detail_batch_concurrency() -> int: + try: + value = int(os.getenv("POLYWEATHER_CITY_DETAIL_BATCH_CONCURRENCY", "3") or "3") + except ValueError: + value = 3 + return max(1, min(6, value)) + + async def get_city_detail_batch_payload( request: Request, *, @@ -318,15 +438,20 @@ async def get_city_detail_batch_payload( if not city_names: return {"cities": [], "details": {}, "errors": {}} + semaphore = asyncio.Semaphore(_city_detail_batch_concurrency()) + + async def _build_with_limit(city: str) -> Tuple[str, Dict[str, Any]]: + async with semaphore: + return await _build_city_detail_batch_item_async( + city, + force_refresh=force_refresh, + market_slug=market_slug, + target_date=target_date, + resolution=resolution, + ) + tasks = [ - run_in_threadpool( - _build_city_detail_batch_item, - city, - force_refresh=force_refresh, - market_slug=market_slug, - target_date=target_date, - resolution=resolution, - ) + _build_with_limit(city) for city in city_names ] results = await asyncio.gather(*tasks, return_exceptions=True) @@ -344,5 +469,3 @@ async def get_city_detail_batch_payload( "details": details, "errors": errors, } - -