perf: inline cache fast path; cache /health checks
CI/CD Pipeline - Northern Thailand Ping River Monitor / Test Suite (3.11) (push) Failing after 23s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Build Docker Image (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Integration Test with Services (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Staging (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Deploy to Production (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Performance Test (push) Skipped
CI/CD Pipeline - Northern Thailand Ping River Monitor / Code Quality (push) Successful in 12s
Documentation / Validate Documentation (push) Failing after 9s
Documentation / Build Sphinx Documentation (push) Successful in 14s
Documentation / Generate API Documentation (push) Successful in 8s
CI/CD Pipeline - Northern Thailand Ping River Monitor / Cleanup (push) Successful in 0s
Documentation / Documentation Summary (push) Successful in 3s

The on-box load test exposed the real stall: cache HITS were dispatched
through asyncio.to_thread, so under load a microsecond lookup queued
~11s behind slow work in the ~8-thread default executor (endpoints
answered inline — /forecast 4ms, /api/stats 2ms — while every to_thread
endpoint sat at p50 8-17s). Handlers now check TTL caches inline in the
async path via _cache_fresh() and only pay for a thread on a miss.

/health results are cached for HEALTH_CACHE_TTL_SECONDS (10): its
external RID-API probe plus DB query were occupying executor threads on
every hit, which is what jammed the pool in the first place.
This commit is contained in:
2026-08-12 11:48:05 +07:00
parent 1ec5cfb4df
commit d9c65bcf0c
2 changed files with 64 additions and 13 deletions
+3
View File
@@ -91,6 +91,9 @@ class Config:
# TTL for the /measurements/latest response cache (hottest endpoint) # TTL for the /measurements/latest response cache (hottest endpoint)
LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45")) 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 # Web server worker processes. Above 1, uvicorn forks workers and a
# localhost lock port elects a single background-collection leader. # localhost lock port elects a single background-collection leader.
WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2")) WEB_WORKERS = int(os.getenv("WEB_WORKERS", "2"))
+55 -7
View File
@@ -408,11 +408,27 @@ async def get_health():
if not health_manager: if not health_manager:
raise HTTPException(status_code=503, detail="Health manager not initialized") raise HTTPException(status_code=503, detail="Health manager not initialized")
# Run health checks (populates state read by get_health_summary). # Health checks include a DB query and an external RID-API probe (seconds
# In a thread: DatabaseHealthCheck and APIHealthCheck do blocking I/O and # of blocking I/O). Cached briefly so hammering /health cannot flood the
# would otherwise stall the event loop for every other request. # shared thread-pool executor; fresh hits answer inline.
await asyncio.to_thread(health_manager.run_all_checks) summary = _cache_fresh(
summary = health_manager.get_health_summary() 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) return HealthResponse(**summary)
@@ -702,6 +718,21 @@ _HII_COMPUTE_LOCKS = {"rain": Lock(), "waterlevel": Lock()}
LATEST_CACHE: Dict[str, Any] = {} LATEST_CACHE: Dict[str, Any] = {}
LATEST_CACHE_LOCK = Lock() LATEST_CACHE_LOCK = Lock()
_LATEST_COMPUTE_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): 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 GROUP BY station_id) latest
ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp 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 asyncio.to_thread(_hii_cached, "rain", hours, sql)
if stale: if stale:
response.headers["X-Data-Stale"] = "true" response.headers["X-Data-Stale"] = "true"
@@ -789,6 +825,11 @@ async def get_hii_waterlevel_latest(
GROUP BY station_id) latest GROUP BY station_id) latest
ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp 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 asyncio.to_thread(_hii_cached, "waterlevel", hours, sql)
if stale: if stale:
response.headers["X-Data-Stale"] = "true" response.headers["X-Data-Stale"] = "true"
@@ -904,8 +945,15 @@ async def get_latest_measurements(response: Response, limit: int = 100):
return scraper.get_latest_data(limit=limit) return scraper.get_latest_data(limit=limit)
try: try:
# Cached + single-flight in a thread: this is the most frequently hit # Fresh cache hits are answered inline (no executor round-trip); only
# endpoint (every dashboard poll); the source data changes hourly. # misses pay for a thread. This is the most frequently hit endpoint.
measurements = _cache_fresh(
LATEST_CACHE,
LATEST_CACHE_LOCK,
f"latest:{limit}",
Config.LATEST_CACHE_TTL_SECONDS,
)
if measurements is None:
measurements, stale = await asyncio.to_thread( measurements, stale = await asyncio.to_thread(
_ttl_cached_stale, _ttl_cached_stale,
LATEST_CACHE, LATEST_CACHE,