gavindiaz/PolyWeather
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Avoid terminal full-detail fallback overload

Browse Source
This commit is contained in:
2569718930@qq.com
2026-06-01 12:06:46 +08:00
parent fee402145e
commit fb6c354317
10 changed files with 269 additions and 89 deletions
Download Patch File Download Diff File Expand all files Collapse all files
frontend/components/dashboard/scan-terminal/__tests__/refreshCadencePolicy.test.ts
+5 -5
View File
@@ -142,15 +142,15 @@ export async function runTests() {
);
assert(
flushCityDetailBatchBlock.includes("if (!payload)") &&
flushCityDetailBatchBlock.includes("resolveCityDetailBatchWithSingleFallback") &&
flushCityDetailBatchBlock.includes("resolveBatchWaiters(waiters, null)"),
"single-city detail fallback should be reserved for whole-batch failures rather than successful batch misses",
flushCityDetailBatchBlock.includes("resolveAllBatchWaitersAsNull") &&
!flushCityDetailBatchBlock.includes("resolveCityDetailBatchWithSingleFallback"),
"whole-batch failures should stop at the chart batch layer instead of fanning out into single-city full-detail requests",
);
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction fetchCityDetailWithTimeout/)?.[0] || "";
const fetchHourlyBlock = chartLogicSource.match(/async function fetchHourlyForecastForCity[\s\S]*?\r?\n}\r?\n\r?\nfunction shouldPollLiveChart/)?.[0] || "";
assert(
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",
"first-paint and background city detail refreshes should both enter the batch queue without falling back to single requests",
);
assert(
dashboardSource.includes("ONLINE_USERS_REFRESH_MS = 5 * 60_000") &&
frontend/components/dashboard/scan-terminal/__tests__/ssePatchArchitecture.test.ts
+1 -1
View File
@@ -259,7 +259,7 @@ export function runTests() {
);
assert(
chartLogic.includes("HOURLY_DETAIL_REQUEST_TIMEOUT_MS = 12_000") &&
chartLogic.includes("fetchCityDetailWithTimeout") &&
chartLogic.includes("fetchCityDetailBatchWithTimeout") &&
chartLogic.includes("signal: controller.signal") &&
chartLogic.includes("controller.abort()"),
"city detail chart fetches must have a frontend timeout so panels cannot stay on 加载图表 forever",
frontend/components/dashboard/scan-terminal/temperature-chart-logic.ts
+6 -46
View File
@@ -1033,16 +1033,6 @@ function primeCityDetailCache(
return data;
}
async function fetchSingleHourlyForecastForCity(
city: string,
resolution: string,
): Promise<HourlyForecast> {
const res = await fetchCityDetailWithTimeout(city, resolution);
if (!res || !res.ok) return null;
const json = await res.json() as CityDetail;
return primeCityDetailCache(city, resolution, json);
}
function queueCityDetailBatch(city: string, resolution: string): Promise<HourlyForecast> {
return new Promise<HourlyForecast>((resolve, reject) => {
const queue = _cityDetailBatchQueues.get(resolution) || {
@@ -1073,13 +1063,6 @@ function resolveBatchWaiters(
(waiters || []).forEach((waiter) => waiter.resolve(value));
}
function rejectBatchWaiters(
waiters: CityDetailBatchWaiter[] | undefined,
reason: unknown,
) {
(waiters || []).forEach((waiter) => waiter.reject(reason));
}
function resolveCityDetailFromBatch(
details: Record<string, CityDetail | null | undefined> | undefined,
city: string,
@@ -1119,7 +1102,7 @@ async function flushCityDetailBatch(resolution: string) {
try {
const payload = await fetchCityDetailBatchWithTimeout(cities, resolution);
if (!payload) {
await resolveCityDetailBatchWithSingleFallback(cities, queue, resolution);
resolveAllBatchWaitersAsNull(cities, queue);
return;
}
@@ -1145,29 +1128,17 @@ async function flushCityDetailBatch(resolution: string) {
}),
);
} catch (error) {
await resolveCityDetailBatchWithSingleFallback(cities, queue, resolution, error);
resolveAllBatchWaitersAsNull(cities, queue);
}
}
async function resolveCityDetailBatchWithSingleFallback(
function resolveAllBatchWaitersAsNull(
cities: string[],
queue: CityDetailBatchQueue,
resolution: string,
reason?: unknown,
) {
await Promise.all(
cities.map(async (city) => {
const waiters = queue.waiters.get(city);
try {
resolveBatchWaiters(
waiters,
await runQueuedHourlyDetailRequest(() => fetchSingleHourlyForecastForCity(city, resolution)),
);
} catch (fallbackError) {
rejectBatchWaiters(waiters, fallbackError || reason);
}
}),
);
cities.forEach((city) => {
resolveBatchWaiters(queue.waiters.get(city), null);
});
}
function fetchCityDetailBatchWithTimeout(cities: string[], resolution: string) {
@@ -1227,17 +1198,6 @@ async function fetchHourlyForecastForCity(
return request;
}
function fetchCityDetailWithTimeout(city: string, resolution: string) {
const controller = new AbortController();
const timeoutId = globalThis.setTimeout(() => controller.abort(), HOURLY_DETAIL_REQUEST_TIMEOUT_MS);
return fetch(`/api/city/${encodeURIComponent(city)}/detail?depth=full&force_refresh=false&resolution=${resolution}`, {
headers: { Accept: "application/json" },
signal: controller.signal,
})
.catch(() => null)
.finally(() => globalThis.clearTimeout(timeoutId));
}
function shouldPollLiveChart({
city,
compact,
tests/test_web_observability.py
+56 -31
View File
@@ -31,6 +31,27 @@ def test_healthz_returns_ok_shape():
assert 'cities_count' in payload
def test_healthz_keeps_liveness_200_when_db_health_is_degraded(monkeypatch):
from web.services import system_api
monkeypatch.setattr(
system_api,
"build_health_payload",
lambda: {
"status": "degraded",
"time_utc": "2026-05-30T00:00:00+00:00",
"db": {"ok": False, "error": "database is locked"},
"state_storage_mode": "sqlite",
"cities_count": 50,
},
)
response = client.get("/healthz")
assert response.status_code == 200
assert response.json()["status"] == "degraded"
def test_system_status_returns_summary_shape():
response = client.get('/api/system/status')
assert response.status_code == 200
@@ -697,41 +718,45 @@ def test_city_detail_batch_chart_scope_returns_only_chart_fields(monkeypatch):
assert kind == "full"
return {
"payload": {
"city": city,
"name": city,
"display_name": city.title(),
"local_date": "2026-05-30",
"local_time": "15:20",
"temp_symbol": "°C",
"current": {
"temp": 20.0,
"settlement_source": "metar",
"settlement_source_label": "METAR",
},
"hourly": {"times": ["2026-05-30T00:00:00Z"], "temps": [20.0]},
"forecast": {
"today_high": 22.0,
"daily": [{"date": "2026-05-30", "max_temp": 22.0}],
},
"multi_model": {
"hourly_times": ["15:00"],
"hourly_forecasts": {"ECMWF": [21.0]},
},
"deb": {"prediction": 21.5, "hourly_path": {"times": ["15:00"], "temps": [21.5]}},
"probabilities": {"mu": 21.4, "distribution": [{"value": 21, "probability": 0.4}]},
"runway_plate_history": {"01/19": [{"time": "2026-05-30T15:20:00Z", "temp": 20.1}]},
"airport_current": {"temp": 20.0},
"airport_primary": {"temp": 20.0},
"airport_primary_today_obs": [["15:20", 20.0]],
"wunderground_current": {"max_so_far": 20.5},
"settlement_station": {"settlement_station_label": "Station"},
"amos": {"runway_obs": {"point_temperatures": []}},
"metar_today_obs": [{"time": "15:20", "temp": 20.0}],
"settlement_today_obs": [],
"dynamic_commentary": {"summary": "large text"},
"official_nearby": [{"name": "unused"}],
"taf": {"raw": "unused"},
"ai_analysis": "unused",
}
}
def build_detail(data, market_slug, target_date, resolution):
return {
"city": data["city"],
"overview": {
"local_date": "2026-05-30",
"local_time": "15:20",
"deb_prediction": 21.5,
"airport_primary_today_obs": [["15:20", 20.0]],
},
"timeseries": {
"hourly": data["hourly"],
"metar_today_obs": [{"time": "15:20", "temp": 20.0}],
"settlement_today_obs": [],
"forecast_daily": [{"date": "2026-05-30", "max_temp": 22.0}],
},
"models_hourly": {"times": ["15:00"], "curves": {"ECMWF": [21.0]}},
"deb": {"prediction": 21.5, "hourly_path": {"times": ["15:00"], "temps": [21.5]}},
"probabilities": {"mu": 21.4, "distribution": [{"value": 21, "probability": 0.4}]},
"runway_plate_history": {"01/19": [{"timestamp": "15:20", "temp_c": 20.1}]},
"runway_band_history": [{"time": "2026-05-30T15:20:00Z", "high_temp": 20.1}],
"airport_current": {"temp": 20.0},
"airport_primary": {"temp": 20.0},
"wunderground_current": {"max_so_far": 20.5},
"settlement_station": {"settlement_station_label": "Station"},
"amos": {"runway_obs": {"point_temperatures": []}},
"dynamic_commentary": {"summary": "large text"},
"official_nearby": [{"name": "unused"}],
"taf": {"raw": "unused"},
"ai_analysis": "unused",
}
def build_detail(_data, _market_slug, _target_date, _resolution):
raise AssertionError("chart scope must not build the full city detail payload")
monkeypatch.setattr(city_api.legacy_routes, "_CACHE_DB", FakeCache())
monkeypatch.setattr(city_api.legacy_routes, "_build_city_detail_payload", build_detail)
web/analysis_service.py
+8
View File
@@ -37,6 +37,7 @@ from src.data_collection.city_registry import ALIASES, CITY_REGISTRY
from src.data_collection.city_time import get_city_utc_offset_seconds
from src.database.runtime_state import IntradayPathSnapshotRepository
from web.services.city_payloads import (
build_city_chart_detail_payload as _city_chart_payload_detail,
build_city_detail_payload as _city_payload_detail,
build_city_summary_payload as _city_payload_summary
)
@@ -2312,6 +2313,13 @@ def _build_city_detail_payload(
)
def _build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str] = "10m",
) -> Dict[str, Any]:
return _city_chart_payload_detail(data, resolution=resolution)
# ──────────────────────────────────────────────────────────
# Routes
web/core.py
+1 -1
View File
@@ -511,7 +511,7 @@ def _is_excluded_model_name(model_name: str) -> bool:
def _sqlite_health() -> Dict[str, Any]:
try:
with sqlite3.connect(_account_db.db_path) as conn:
with sqlite3.connect(_account_db.db_path, timeout=0.05) as conn:
conn.execute("SELECT 1").fetchone()
return {"ok": True, "db_path": _account_db.db_path}
except Exception as exc:
web/services/city_api.py
+44
View File
@@ -146,6 +146,24 @@ async def _get_city_full_data(city: str, *, force_refresh: bool) -> Dict[str, An
return await _refresh_city_full_data(city, False)
async def _get_city_chart_data(city: str, *, force_refresh: bool) -> Dict[str, Any]:
if force_refresh:
return await _get_city_full_data(city, force_refresh=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 {}
if payload:
if not legacy_routes._city_cache_is_fresh(cached_entry, legacy_routes.CITY_FULL_CACHE_TTL_SEC):
_start_city_full_stale_refresh(city)
return await _overlay_cached_wunderground(city, payload)
return {
"name": city,
"display_name": str((legacy_routes.CITY_REGISTRY.get(city, {}) or {}).get("display_name") or city.title()),
}
def _city_detail_payload_cache_key(
data: Dict[str, Any],
market_slug: Optional[str],
@@ -228,6 +246,17 @@ async def _build_city_detail_payload_cached(
return payload
async def _build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str],
) -> Dict[str, Any]:
return await run_in_threadpool(
legacy_routes._build_city_chart_detail_payload,
data,
resolution,
)
def _default_deb_recent() -> Dict[str, object]:
return {
"tier": "other",
@@ -577,6 +606,21 @@ async def _build_city_detail_batch_item_async(
detail_scope: str = "full",
timing_recorder: Optional[ServerTimingRecorder] = None,
) -> Tuple[str, Dict[str, Any]]:
if detail_scope == "chart":
if timing_recorder is not None:
data = await timing_recorder.measure_async(
f"chart_data_{city}",
lambda: _get_city_chart_data(city, force_refresh=force_refresh),
)
detail = await timing_recorder.measure_async(
f"chart_payload_{city}",
lambda: _build_city_chart_detail_payload(data, resolution),
)
else:
data = await _get_city_chart_data(city, force_refresh=force_refresh)
detail = await _build_city_chart_detail_payload(data, resolution)
return city, detail
if timing_recorder is not None:
data = await timing_recorder.measure_async(
f"full_data_{city}",
web/services/city_payloads.py
+145
View File
@@ -161,6 +161,151 @@ def build_runway_band_history(raw_history: Dict[str, List[Dict[str, Any]]], reso
return band_history
def build_runway_chart_histories(
raw_history: Dict[str, List[Dict[str, Any]]],
resolution: str,
) -> tuple[Dict[str, List[Dict[str, Any]]], List[Dict[str, Any]]]:
if not raw_history:
return {}, []
try:
if resolution.endswith("m"):
minutes = int(resolution[:-1])
elif resolution.endswith("h"):
minutes = int(resolution[:-1]) * 60
else:
minutes = 10
except Exception:
minutes = 10
seconds = max(60, minutes * 60)
per_runway_buckets: Dict[str, Dict[int, List[float]]] = {}
all_buckets: Dict[int, List[float]] = {}
for rwy, points in raw_history.items():
if not points:
continue
rwy_key = str(rwy or "").strip()
if not rwy_key:
continue
runway_buckets = per_runway_buckets.setdefault(rwy_key, {})
for pt in points:
t_str = pt.get("time") or pt.get("timestamp")
temp = pt.get("temp") or pt.get("temp_c") or pt.get("value")
if temp is None or not isinstance(t_str, str):
continue
dt = _parse_time_val(t_str)
if not dt:
continue
try:
temp_value = float(temp)
except Exception:
continue
bucket_ts = (int(dt.timestamp()) // seconds) * seconds
runway_buckets.setdefault(bucket_ts, []).append(temp_value)
all_buckets.setdefault(bucket_ts, []).append(temp_value)
aggregated_history: Dict[str, List[Dict[str, Any]]] = {}
for rwy, buckets in per_runway_buckets.items():
bucket_points = []
for bucket_ts in sorted(buckets.keys()):
temps = buckets[bucket_ts]
if not temps:
continue
bucket_dt = datetime.fromtimestamp(bucket_ts, tz=timezone.utc)
bucket_points.append(
{
"time": bucket_dt.isoformat(),
"temp": round(temps[-1], 1),
}
)
if bucket_points:
aggregated_history[rwy] = bucket_points
band_history = []
for bucket_ts in sorted(all_buckets.keys()):
temps = all_buckets[bucket_ts]
if not temps:
continue
bucket_dt = datetime.fromtimestamp(bucket_ts, tz=timezone.utc)
band_history.append(
{
"time": bucket_dt.isoformat(),
"high_temp": round(max(temps), 1),
"low_temp": round(min(temps), 1),
"avg_temp": round(sum(temps) / len(temps), 1),
}
)
return aggregated_history, band_history
def build_city_chart_detail_payload(
data: Dict[str, Any],
resolution: Optional[str] = "10m",
) -> Dict[str, Any]:
forecast = data.get("forecast") if isinstance(data.get("forecast"), dict) else {}
current = data.get("current") if isinstance(data.get("current"), dict) else {}
multi_model = data.get("multi_model") if isinstance(data.get("multi_model"), dict) else {}
runway_history, runway_band_history = build_runway_chart_histories(
data.get("runway_plate_history") or {},
resolution or "10m",
)
airport_primary_today_obs = data.get("airport_primary_today_obs") or []
local_date = data.get("local_date")
local_time = data.get("local_time")
return {
"city": data.get("name") or data.get("city"),
"fetched_at": data.get("updated_at"),
"local_date": local_date,
"local_time": local_time,
"overview": {
"name": data.get("name") or data.get("city"),
"display_name": data.get("display_name"),
"local_date": local_date,
"local_time": local_time,
"temp_symbol": data.get("temp_symbol"),
"current_temp": current.get("temp"),
"deb_prediction": (data.get("deb") or {}).get("prediction"),
"settlement_source": current.get("settlement_source"),
"settlement_source_label": current.get("settlement_source_label"),
},
"timeseries": {
"hourly": data.get("hourly") or {},
"metar_today_obs": data.get("metar_today_obs") or [],
"settlement_today_obs": data.get("settlement_today_obs") or [],
"forecast_daily": forecast.get("daily") or [],
},
"hourly": data.get("hourly") or {},
"models_hourly": {
"times": multi_model.get("hourly_times") or [],
"curves": {
model: values
for model, values in (multi_model.get("hourly_forecasts") or {}).items()
if not _is_excluded_model_name(model)
},
},
"deb": data.get("deb") or {},
"forecast": {
"today_high": forecast.get("today_high"),
"daily": forecast.get("daily") or [],
},
"multi_model_daily": data.get("multi_model_daily") or {},
"probabilities": data.get("probabilities") or {"mu": None, "distribution": []},
"runway_plate_history": runway_history,
"runway_band_history": runway_band_history,
"amos": data.get("amos") or {},
"airport_current": data.get("airport_current") or {},
"airport_primary": data.get("airport_primary") or {},
"airport_primary_today_obs": airport_primary_today_obs,
"official": {"airport_primary_today_obs": airport_primary_today_obs},
"wunderground_current": data.get("wunderground_current") or {},
"settlement_station": data.get("settlement_station") or {},
}
def build_city_detail_payload(
data: Dict[str, Any],
market_slug: Optional[str] = None,
web/services/city_runtime.py
+1
View File
@@ -25,6 +25,7 @@ from src.utils.refresh_policy import OBSERVATION_REFRESH_SEC, SCAN_ROWS_REFRESH_
from web.analysis_service import (
_analyze,
_analyze_summary,
_build_city_chart_detail_payload, # noqa: F401 - compatibility export for chart detail batches
_build_city_detail_payload, # noqa: F401 - compatibility export for tests and transitional routers
_build_city_market_scan_payload,
_build_city_summary_payload,
web/services/system_api.py
+2 -5
View File
@@ -5,7 +5,7 @@ from __future__ import annotations
import time
from typing import Any, Dict, Optional
from fastapi import BackgroundTasks, HTTPException, Request
from fastapi import BackgroundTasks, Request
from fastapi.concurrency import run_in_threadpool
from fastapi.responses import PlainTextResponse
from loguru import logger
@@ -16,10 +16,7 @@ import web.routes as legacy_routes
def get_health_payload() -> Dict[str, Any]:
payload = build_health_payload()
if payload.get("status") != "ok":
raise HTTPException(status_code=503, detail=payload)
return payload
return build_health_payload()
async def get_system_status_payload() -> Dict[str, Any]: