diff --git a/src/config.py b/src/config.py index a6dbe00..65b6855 100644 --- a/src/config.py +++ b/src/config.py @@ -92,7 +92,7 @@ class Config: LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45")) # TTL for /health check results (includes an external RID-API probe) - HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "10")) + HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "30")) # Web server worker processes. Above 1, uvicorn forks workers and a # localhost lock port elects a single background-collection leader. diff --git a/src/health_check.py b/src/health_check.py index 7a67035..bcf307f 100644 --- a/src/health_check.py +++ b/src/health_check.py @@ -140,6 +140,10 @@ class APIHealthCheck(HealthCheck): def __init__(self, api_url: str, session, name: str = "api"): super().__init__(name) + # A liveness probe should fail fast: the default 30s timeout meant a + # slow upstream pinned executor threads for longer than the /health + # cache TTL, so the pool never drained under load. + self.timeout_seconds = 5 self.api_url = api_url self.session = session diff --git a/src/web_api.py b/src/web_api.py index 03ac09a..114faaf 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -64,6 +64,7 @@ FORECAST_TTL = 900 # 15 minutes DB_STATS_CACHE: Dict[str, tuple] = {} DB_STATS_CACHE_LOCK = Lock() +_DB_STATS_COMPUTE_LOCK = Lock() DB_STATS_TTL = 300 # 5 minutes # Admin API protection. Read/dashboard endpoints stay public; anything that @@ -1042,20 +1043,9 @@ async def get_database_stats(): if not scraper or not scraper.db_adapter: raise HTTPException(status_code=503, detail="Database not available") - now = time.monotonic() - with DB_STATS_CACHE_LOCK: - cached = DB_STATS_CACHE.get("all") - if cached and now - cached[0] < DB_STATS_TTL: - return cached[1] - - try: - stats = await asyncio.to_thread(scraper.db_adapter.get_database_stats) - except Exception as e: - logger.error(f"Error fetching database stats: {e}") - raise HTTPException(status_code=500, detail=str(e)) - - if stats is None: - raise HTTPException(status_code=503, detail="Database statistics unavailable") + cached = _cache_fresh(DB_STATS_CACHE, DB_STATS_CACHE_LOCK, "all", DB_STATS_TTL) + if cached is not None: + return cached def hii_totals(): engine = _hii_engine() @@ -1079,47 +1069,69 @@ async def get_database_stats(): ) ).one() - hii = None + def compute(): + # Heavy: full-table counts and coverage over ~1.7M rows. Runs at most + # once per TTL thanks to the single-flight; concurrent misses wait for + # this one result instead of piling identical queries onto Postgres. + stats = scraper.db_adapter.get_database_stats() + if stats is None: + raise RuntimeError("Database statistics unavailable") + + hii = None + try: + hii = hii_totals() + except Exception as e: + logger.warning(f"HII stats unavailable: {e}") + + def as_dt(value): + if isinstance(value, str): + return datetime.fromisoformat(value) + return value + + first_ts = stats["first_timestamp"] + last_ts = stats["last_timestamp"] + rain_n = wl_n = hii_stations = 0 + if hii is not None: + rain_n, wl_n = hii.rain_n or 0, hii.wl_n or 0 + hii_stations = (hii.rain_s or 0) + (hii.wl_s or 0) + for lo in (as_dt(hii.rain_lo), as_dt(hii.wl_lo)): + if lo is not None and lo < first_ts: + first_ts = lo + for hi in (as_dt(hii.rain_hi), as_dt(hii.wl_hi)): + if hi is not None and hi > last_ts: + last_ts = hi + + return { + # Whole-DB totals (RID + HII feeds); breakdown fields alongside + "total_measurements": stats["total_measurements"] + rain_n + wl_n, + "rid_measurements": stats["total_measurements"], + "hii_rainfall_measurements": rain_n, + "hii_waterlevel_measurements": wl_n, + "station_count": stats["station_count"] + hii_stations, + "rid_station_count": stats["station_count"], + "hii_station_count": hii_stations, + "first_timestamp": first_ts.isoformat(), + "last_timestamp": last_ts.isoformat(), + "days_spanned": (last_ts.date() - first_ts.date()).days + 1, + "coverage_percent": stats["coverage_percent"], + } + try: - hii = await asyncio.to_thread(hii_totals) + data, _ = await asyncio.to_thread( + _ttl_cached_stale, + DB_STATS_CACHE, + DB_STATS_CACHE_LOCK, + _DB_STATS_COMPUTE_LOCK, + "all", + DB_STATS_TTL, + compute, + ) + return data + except RuntimeError as e: + raise HTTPException(status_code=503, detail=str(e)) except Exception as e: - logger.warning(f"HII stats unavailable: {e}") - - def as_dt(value): - if isinstance(value, str): - return datetime.fromisoformat(value) - return value - - first_ts = stats["first_timestamp"] - last_ts = stats["last_timestamp"] - rain_n = wl_n = hii_stations = 0 - if hii is not None: - rain_n, wl_n = hii.rain_n or 0, hii.wl_n or 0 - hii_stations = (hii.rain_s or 0) + (hii.wl_s or 0) - for lo in (as_dt(hii.rain_lo), as_dt(hii.wl_lo)): - if lo is not None and lo < first_ts: - first_ts = lo - for hi in (as_dt(hii.rain_hi), as_dt(hii.wl_hi)): - if hi is not None and hi > last_ts: - last_ts = hi - - data = { - # Whole-DB totals (RID + HII feeds); breakdown fields alongside - "total_measurements": stats["total_measurements"] + rain_n + wl_n, - "rid_measurements": stats["total_measurements"], - "hii_rainfall_measurements": rain_n, - "hii_waterlevel_measurements": wl_n, - "station_count": stats["station_count"] + hii_stations, - "rid_station_count": stats["station_count"], - "hii_station_count": hii_stations, - "first_timestamp": first_ts.isoformat(), - "last_timestamp": last_ts.isoformat(), - "days_spanned": (last_ts.date() - first_ts.date()).days + 1, - "coverage_percent": stats["coverage_percent"], - } - with DB_STATS_CACHE_LOCK: - DB_STATS_CACHE["all"] = (now, data) - return data + logger.error(f"Error fetching database stats: {e}") + raise HTTPException(status_code=500, detail=str(e)) @app.post("/scrape/trigger", dependencies=[Depends(require_admin_key)])