diff --git a/src/config.py b/src/config.py index 35f82d9..a6dbe00 100644 --- a/src/config.py +++ b/src/config.py @@ -91,6 +91,9 @@ class Config: # TTL for the /measurements/latest response cache (hottest endpoint) 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")) + # Web server worker processes. Above 1, uvicorn forks workers and a # localhost lock port elects a single background-collection leader. WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2")) diff --git a/src/web_api.py b/src/web_api.py index 73b987d..4b06ec5 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -408,11 +408,27 @@ async def get_health(): if not health_manager: raise HTTPException(status_code=503, detail="Health manager not initialized") - # Run health checks (populates state read by get_health_summary). - # In a thread: DatabaseHealthCheck and APIHealthCheck do blocking I/O and - # would otherwise stall the event loop for every other request. - await asyncio.to_thread(health_manager.run_all_checks) - summary = health_manager.get_health_summary() + # Health checks include a DB query and an external RID-API probe (seconds + # of blocking I/O). Cached briefly so hammering /health cannot flood the + # shared thread-pool executor; fresh hits answer inline. + summary = _cache_fresh( + HEALTH_CACHE, HEALTH_CACHE_LOCK, "health", Config.HEALTH_CACHE_TTL_SECONDS + ) + if summary is None: + + def compute(): + health_manager.run_all_checks() + return health_manager.get_health_summary() + + summary, _ = await asyncio.to_thread( + _ttl_cached_stale, + HEALTH_CACHE, + HEALTH_CACHE_LOCK, + _HEALTH_COMPUTE_LOCK, + "health", + Config.HEALTH_CACHE_TTL_SECONDS, + compute, + ) return HealthResponse(**summary) @@ -702,6 +718,21 @@ _HII_COMPUTE_LOCKS = {"rain": Lock(), "waterlevel": Lock()} LATEST_CACHE: Dict[str, Any] = {} LATEST_CACHE_LOCK = Lock() _LATEST_COMPUTE_LOCK = Lock() +HEALTH_CACHE: Dict[str, Any] = {} +HEALTH_CACHE_LOCK = Lock() +_HEALTH_COMPUTE_LOCK = Lock() + + +def _cache_fresh(cache, cache_lock, key, ttl): + """Non-blocking fresh-cache read for the async fast path. On-box load + testing showed cache HITS queuing ~11s behind slow work in the shared + thread-pool executor — so handlers must check the cache inline and only + dispatch to a thread on a miss.""" + with cache_lock: + entry = cache.get(key) + if entry and time.monotonic() - entry[0] < ttl: + return entry[1] + return None def _ttl_cached_stale(cache, cache_lock, compute_lock, key, ttl, compute): @@ -764,6 +795,11 @@ async def get_hii_rainfall_latest( GROUP BY station_id) latest ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp """ + rows = _cache_fresh( + HII_CACHE, HII_CACHE_LOCK, f"rain:{hours}", Config.HII_CACHE_TTL_SECONDS + ) + if rows is not None: + return rows rows, stale = await asyncio.to_thread(_hii_cached, "rain", hours, sql) if stale: response.headers["X-Data-Stale"] = "true" @@ -789,6 +825,11 @@ async def get_hii_waterlevel_latest( GROUP BY station_id) latest ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp """ + rows = _cache_fresh( + HII_CACHE, HII_CACHE_LOCK, f"waterlevel:{hours}", Config.HII_CACHE_TTL_SECONDS + ) + if rows is not None: + return rows rows, stale = await asyncio.to_thread(_hii_cached, "waterlevel", hours, sql) if stale: response.headers["X-Data-Stale"] = "true" @@ -904,19 +945,26 @@ async def get_latest_measurements(response: Response, limit: int = 100): return scraper.get_latest_data(limit=limit) try: - # Cached + single-flight in a thread: this is the most frequently hit - # endpoint (every dashboard poll); the source data changes hourly. - measurements, stale = await asyncio.to_thread( - _ttl_cached_stale, + # Fresh cache hits are answered inline (no executor round-trip); only + # misses pay for a thread. This is the most frequently hit endpoint. + measurements = _cache_fresh( LATEST_CACHE, LATEST_CACHE_LOCK, - _LATEST_COMPUTE_LOCK, f"latest:{limit}", Config.LATEST_CACHE_TTL_SECONDS, - compute, ) - if stale: - response.headers["X-Data-Stale"] = "true" + if measurements is None: + measurements, stale = await asyncio.to_thread( + _ttl_cached_stale, + LATEST_CACHE, + LATEST_CACHE_LOCK, + _LATEST_COMPUTE_LOCK, + f"latest:{limit}", + Config.LATEST_CACHE_TTL_SECONDS, + compute, + ) + if stale: + response.headers["X-Data-Stale"] = "true" return [_to_measurement_response(m) for m in measurements] except Exception as e: