From 731f10910e2366cbfc0174a03db8610c4355a9d3 Mon Sep 17 00:00:00 2001 From: grabowski Date: Wed, 12 Aug 2026 12:32:38 +0700 Subject: [PATCH] perf: stale-while-revalidate caching, bigger executor, cheap DB health probe MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Third on-box run: p50s healthy everywhere but tails at 60s — when a TTL expired under load, every concurrent miss parked an executor thread on the single-flight lock, exhausting the ~12-thread pool and timing out unrelated endpoints. All cached endpoints (latest, HII, stats, health) now use _cached_swr: fresh -> inline; expired-but-present -> the stale value is returned immediately and ONE background task refreshes; only a cold key (first request since startup) waits. Plus: dedicated ThreadPoolExecutor (EXECUTOR_THREADS, default 48) replaces the cpu+4 default, and DatabaseHealthCheck no longer re-runs the CREATE TABLE DDL suite on every probe (connect only when no live engine). --- src/config.py | 5 ++ src/health_check.py | 7 +- src/web_api.py | 131 +++++++++++++++++++++++------------- tests/test_hii_collector.py | 4 ++ 4 files changed, 97 insertions(+), 50 deletions(-) diff --git a/src/config.py b/src/config.py index 65b6855..05b8f30 100644 --- a/src/config.py +++ b/src/config.py @@ -94,6 +94,11 @@ class Config: # TTL for /health check results (includes an external RID-API probe) HEALTH_CACHE_TTL_SECONDS = int(os.getenv("HEALTH_CACHE_TTL_SECONDS", "30")) + # Thread-pool size for blocking work in the web process (DB queries, + # inference, health probes). Waiting threads are cheap; starving the pool + # stalls every endpoint that needs a thread. + EXECUTOR_THREADS = int(os.getenv("EXECUTOR_THREADS", "48")) + # 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/health_check.py b/src/health_check.py index bcf307f..d331356 100644 --- a/src/health_check.py +++ b/src/health_check.py @@ -88,8 +88,11 @@ class DatabaseHealthCheck(HealthCheck): } try: - # Try to connect - if hasattr(self.db_adapter, "connect"): + # Connect only when there is no live engine yet: connect() re-runs + # the CREATE TABLE DDL suite, which is far too heavy per probe. + if getattr(self.db_adapter, "engine", None) is None and hasattr( + self.db_adapter, "connect" + ): connected = self.db_adapter.connect() if not connected: return { diff --git a/src/web_api.py b/src/web_api.py index 114faaf..5975fab 100644 --- a/src/web_api.py +++ b/src/web_api.py @@ -138,6 +138,15 @@ async def lifespan(app: FastAPI): # Startup logger.info("Starting Water Monitor API...") + # Larger dedicated executor: the default (cpu+4 threads) was exhausted + # under load by concurrent blocking work; waiting threads are cheap. + from concurrent.futures import ThreadPoolExecutor + + app_state["executor"] = ThreadPoolExecutor( + max_workers=Config.EXECUTOR_THREADS, thread_name_prefix="api" + ) + asyncio.get_running_loop().set_default_executor(app_state["executor"]) + # Initialize configuration try: Config.validate_config() @@ -198,6 +207,8 @@ async def lifespan(app: FastAPI): pass if app_state.get("leader_lock"): app_state["leader_lock"].close() + if app_state.get("executor"): + app_state["executor"].shutdown(wait=False) logger.info("Water Monitor API shutdown complete") @@ -441,24 +452,18 @@ async def get_health(): # 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 + def compute(): + health_manager.run_all_checks() + return health_manager.get_health_summary() + + summary, _ = await _cached_swr( + HEALTH_CACHE, + HEALTH_CACHE_LOCK, + _HEALTH_COMPUTE_LOCK, + "health", + Config.HEALTH_CACHE_TTL_SECONDS, + compute, ) - 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) @@ -753,6 +758,56 @@ HEALTH_CACHE_LOCK = Lock() _HEALTH_COMPUTE_LOCK = Lock() +_REFRESH_IN_FLIGHT: set = set() +_REFRESH_FLAG_LOCK = Lock() + + +async def _cached_swr(cache, cache_lock, compute_lock, key, ttl, compute): + """Stale-while-revalidate. Returns (value, served_stale). + + fresh -> return inline; expired-but-present -> return the stale value + immediately and kick ONE background refresh (nobody waits, and no executor + thread is parked on the single-flight lock — under stampede that parking + exhausted the pool and timed out unrelated endpoints); absent -> cold + single-flight compute (first request per key since startup). + """ + fresh = _cache_fresh(cache, cache_lock, key, ttl) + if fresh is not None: + return fresh, False + with cache_lock: + entry = cache.get(key) + if entry is not None: + with _REFRESH_FLAG_LOCK: + should_start = key not in _REFRESH_IN_FLIGHT + if should_start: + _REFRESH_IN_FLIGHT.add(key) + if should_start: + + async def _refresh(): + try: + await asyncio.to_thread( + _ttl_cached_stale, + cache, + cache_lock, + compute_lock, + key, + ttl, + compute, + ) + except Exception as error: + logger.warning(f"{key}: background refresh failed: {error}") + finally: + with _REFRESH_FLAG_LOCK: + _REFRESH_IN_FLIGHT.discard(key) + + asyncio.create_task(_refresh()) + return entry[1], True + value, stale = await asyncio.to_thread( + _ttl_cached_stale, cache, cache_lock, compute_lock, key, ttl, compute + ) + return value, stale + + 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 @@ -793,12 +848,12 @@ def _ttl_cached_stale(cache, cache_lock, compute_lock, key, ttl, compute): return value, False -def _hii_cached(feed: str, hours: int, sql: str): +async def _hii_swr(feed: str, hours: int, sql: str): def compute(): cutoff = datetime.now() - timedelta(hours=hours) return _hii_rows(sql, {"cutoff": cutoff}) - return _ttl_cached_stale( + return await _cached_swr( HII_CACHE, HII_CACHE_LOCK, _HII_COMPUTE_LOCKS[feed], @@ -825,12 +880,7 @@ 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) + rows, stale = await _hii_swr("rain", hours, sql) if stale: response.headers["X-Data-Stale"] = "true" return rows @@ -855,12 +905,7 @@ 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) + rows, stale = await _hii_swr("waterlevel", hours, sql) if stale: response.headers["X-Data-Stale"] = "true" return rows @@ -975,26 +1020,17 @@ async def get_latest_measurements(response: Response, limit: int = 100): return scraper.get_latest_data(limit=limit) try: - # 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( + # SWR-cached: the most frequently hit endpoint (every dashboard poll) + measurements, stale = await _cached_swr( LATEST_CACHE, LATEST_CACHE_LOCK, + _LATEST_COMPUTE_LOCK, f"latest:{limit}", Config.LATEST_CACHE_TTL_SECONDS, + compute, ) - 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" + if stale: + response.headers["X-Data-Stale"] = "true" return [_to_measurement_response(m) for m in measurements] except Exception as e: @@ -1117,8 +1153,7 @@ async def get_database_stats(): } try: - data, _ = await asyncio.to_thread( - _ttl_cached_stale, + data, _ = await _cached_swr( DB_STATS_CACHE, DB_STATS_CACHE_LOCK, _DB_STATS_COMPUTE_LOCK, diff --git a/tests/test_hii_collector.py b/tests/test_hii_collector.py index 388e293..214dd21 100644 --- a/tests/test_hii_collector.py +++ b/tests/test_hii_collector.py @@ -296,6 +296,7 @@ class TestHiiApiEndpoints: assert collector.store.save_waterlevel(wl) == 2 monkeypatch.setitem(web_api.app_state, "hii_collector", collector) web_api.HII_CACHE.clear() # response cache would leak across tests + web_api._REFRESH_IN_FLIGHT.clear() return web_api @staticmethod @@ -331,6 +332,7 @@ class TestHiiApiEndpoints: monkeypatch.setitem(web_api.app_state, "hii_collector", None) web_api.HII_CACHE.clear() + web_api._REFRESH_IN_FLIGHT.clear() assert self._get(web_api, "get_hii_rainfall_latest", hours=26)[0] == [] assert self._get(web_api, "get_hii_waterlevel_latest", hours=26)[0] == [] @@ -378,6 +380,7 @@ class TestHiiApiEndpoints: ) monkeypatch.setitem(web_api.app_state, "hii_collector", collector) web_api.HII_CACHE.clear() + web_api._REFRESH_IN_FLIGHT.clear() assert self._get(web_api, "get_hii_rainfall_latest", hours=26)[0] == [] assert web_api.HII_CACHE == {} # empty response left uncached @@ -406,6 +409,7 @@ class TestHiiApiEndpoints: scraper = SimpleNamespace(db_adapter=True, get_latest_data=get_latest_data) monkeypatch.setitem(web_api.app_state, "scraper", scraper) web_api.LATEST_CACHE.clear() + web_api._REFRESH_IN_FLIGHT.clear() rows, response = self._get(web_api, "get_latest_measurements", limit=500) rows2, _ = self._get(web_api, "get_latest_measurements", limit=500)