perf: cache /measurements/latest; stale-on-error fallback for cached endpoints

The response caches (HII feeds, and now /measurements/latest at 45s TTL
— the endpoint every dashboard poll hits) share one helper,
_ttl_cached_stale: single-flight per key, empty results never cached,
and expired entries kept as a fallback. If a recompute fails (DB
unreachable), the last good response is served with an X-Data-Stale:
true header instead of a 5xx — during an outage the dashboard keeps
showing the last real readings with their honest timestamps. TTLs:
LATEST_CACHE_TTL_SECONDS (45), HII_CACHE_TTL_SECONDS (120).

src/config.py

@@ -88,6 +88,8 @@ class Config:
     HII_BASIN_CODE = int(os.getenv("HII_BASIN_CODE", "6"))  # 6 = Ping Basin
     # TTL for the /api/hii/*/latest response cache; source data changes hourly
     HII_CACHE_TTL_SECONDS = int(os.getenv("HII_CACHE_TTL_SECONDS", "120"))
+    # TTL for the /measurements/latest response cache (hottest endpoint)
+    LATEST_CACHE_TTL_SECONDS = int(os.getenv("LATEST_CACHE_TTL_SECONDS", "45"))
 
     # Umami analytics (self-hosted). The website id is public (it ships in the
     # dashboard <script> tag); server-side API tracking posts to /api/send.

src/web_api.py

@@ -14,7 +14,15 @@ from threading import Lock
 from typing import Any, Dict, List, Optional
 
 import requests
-from fastapi import BackgroundTasks, Depends, FastAPI, Header, HTTPException, Query
+from fastapi import (
+    BackgroundTasks,
+    Depends,
+    FastAPI,
+    Header,
+    HTTPException,
+    Query,
+    Response,
+)
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import FileResponse, HTMLResponse
 from fastapi.staticfiles import StaticFiles
@@ -641,39 +649,68 @@ def _hii_rows(sql: str, params: Dict[str, Any]) -> List[Dict[str, Any]]:
     return rows
 
 
-# Short-TTL cache for the HII feeds: the collector writes hourly, but these
-# endpoints carried ~30% of the load-test mix at 0.6-0.9 s per uncached query.
-# Process-local by design (single-worker deployment); the helper is the seam
-# where a shared backend (e.g. Redis) would slot in if we ever run multiple
-# workers. Thread-based single-flight keeps it event-loop-agnostic.
+# Short-TTL response caches with stale-on-error. Process-local by design
+# (single-worker deployment); _ttl_cached_stale is the seam where a shared
+# backend (e.g. Redis) would slot in if we ever run multiple workers.
+# Thread-based single-flight keeps it event-loop-agnostic. Expired entries are
+# kept as a fallback: if the recompute fails (typically the DB briefly
+# unreachable), the last good response is served instead of a 5xx — during a
+# flood, slightly stale readings with a visible timestamp beat an error page.
 HII_CACHE: Dict[str, Any] = {}
 HII_CACHE_LOCK = Lock()
 _HII_COMPUTE_LOCKS = {"rain": Lock(), "waterlevel": Lock()}
+LATEST_CACHE: Dict[str, Any] = {}
+LATEST_CACHE_LOCK = Lock()
+_LATEST_COMPUTE_LOCK = Lock()
 
 
-def _hii_cached(feed: str, hours: int, sql: str) -> List[Dict[str, Any]]:
-    key = f"{feed}:{hours}"
-    ttl = Config.HII_CACHE_TTL_SECONDS
-    now = time.monotonic()
-    with HII_CACHE_LOCK:
-        cached = HII_CACHE.get(key)
-        if cached and now - cached[0] < ttl:
-            return cached[1]
-    with _HII_COMPUTE_LOCKS[feed]:  # single-flight per feed
-        with HII_CACHE_LOCK:
-            cached = HII_CACHE.get(key)
-            if cached and time.monotonic() - cached[0] < ttl:
-                return cached[1]
+def _ttl_cached_stale(cache, cache_lock, compute_lock, key, ttl, compute):
+    """Return (value, is_stale). Single-flight; keeps expired entries as an
+    error fallback; never caches empty results so recovery is immediate."""
+    with cache_lock:
+        entry = cache.get(key)
+        if entry and time.monotonic() - entry[0] < ttl:
+            return entry[1], False
+    with compute_lock:
+        with cache_lock:
+            entry = cache.get(key)
+            if entry and time.monotonic() - entry[0] < ttl:
+                return entry[1], False
+        try:
+            value = compute()
+        except Exception as error:
+            if entry is not None:
+                age = int(time.monotonic() - entry[0])
+                logger.warning(
+                    f"{key}: recompute failed ({error}); serving {age}s-stale copy"
+                )
+                return entry[1], True
+            raise
+        if value:
+            with cache_lock:
+                cache[key] = (time.monotonic(), value)
+        return value, False
+
+
+def _hii_cached(feed: str, hours: int, sql: str):
+    def compute():
         cutoff = datetime.now() - timedelta(hours=hours)
-        rows = _hii_rows(sql, {"cutoff": cutoff})
-        if rows:  # never cache empty: recover immediately once data appears
-            with HII_CACHE_LOCK:
-                HII_CACHE[key] = (time.monotonic(), rows)
-        return rows
+        return _hii_rows(sql, {"cutoff": cutoff})
+
+    return _ttl_cached_stale(
+        HII_CACHE,
+        HII_CACHE_LOCK,
+        _HII_COMPUTE_LOCKS[feed],
+        f"{feed}:{hours}",
+        Config.HII_CACHE_TTL_SECONDS,
+        compute,
+    )
 
 
 @app.get("/api/hii/rainfall/latest")
-async def get_hii_rainfall_latest(hours: int = Query(26, ge=1, le=168)):
+async def get_hii_rainfall_latest(
+    response: Response, hours: int = Query(26, ge=1, le=168)
+):
     """Latest rainfall reading per HII station (Ping basin, collected hourly)."""
     increment_counter("api_requests", labels={"endpoint": "hii_rainfall"})
     sql = """
@@ -687,11 +724,16 @@ async def get_hii_rainfall_latest(hours: int = Query(26, ge=1, le=168)):
               GROUP BY station_id) latest
           ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp
     """
-    return await asyncio.to_thread(_hii_cached, "rain", hours, sql)
+    rows, stale = await asyncio.to_thread(_hii_cached, "rain", hours, sql)
+    if stale:
+        response.headers["X-Data-Stale"] = "true"
+    return rows
 
 
 @app.get("/api/hii/waterlevel/latest")
-async def get_hii_waterlevel_latest(hours: int = Query(26, ge=1, le=168)):
+async def get_hii_waterlevel_latest(
+    response: Response, hours: int = Query(26, ge=1, le=168)
+):
     """Latest water-level reading per HII station (Ping basin, m MSL)."""
     increment_counter("api_requests", labels={"endpoint": "hii_waterlevel"})
     sql = """
@@ -707,7 +749,10 @@ async def get_hii_waterlevel_latest(hours: int = Query(26, ge=1, le=168)):
               GROUP BY station_id) latest
           ON latest.station_id = m.station_id AND latest.latest_ts = m.timestamp
     """
-    return await asyncio.to_thread(_hii_cached, "waterlevel", hours, sql)
+    rows, stale = await asyncio.to_thread(_hii_cached, "waterlevel", hours, sql)
+    if stale:
+        response.headers["X-Data-Stale"] = "true"
+    return rows
 
 
 @app.get("/measurements/history/{station_code}")
@@ -807,7 +852,7 @@ async def _compute_forecasts(get_latest_forecasts):
 
 
 @app.get("/measurements/latest", response_model=List[MeasurementResponse])
-async def get_latest_measurements(limit: int = 100):
+async def get_latest_measurements(response: Response, limit: int = 100):
     """Get latest measurements from all stations"""
     increment_counter("api_requests", labels={"endpoint": "measurements_latest"})
 
@@ -815,11 +860,23 @@ async def get_latest_measurements(limit: int = 100):
     if not scraper or not scraper.db_adapter:
         raise HTTPException(status_code=503, detail="Database not available")
 
-    try:
-        # In a thread: this is a synchronous DB query, and this is the most
-        # frequently hit endpoint — inline it would block the event loop.
-        measurements = await asyncio.to_thread(scraper.get_latest_data, limit)
+    def compute():
+        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,
+            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:

tests/test_hii_collector.py

@@ -298,19 +298,27 @@ class TestHiiApiEndpoints:
         web_api.HII_CACHE.clear()  # response cache would leak across tests
         return web_api
 
-    def test_rainfall_latest(self, web_api):
+    @staticmethod
+    def _get(web_api_module, endpoint, **kwargs):
         import asyncio
 
-        rows = asyncio.run(web_api.get_hii_rainfall_latest(hours=26))
+        from fastapi import Response
+
+        response = Response()
+        rows = asyncio.run(
+            getattr(web_api_module, endpoint)(response=response, **kwargs)
+        )
+        return rows, response
+
+    def test_rainfall_latest(self, web_api):
+        rows, _ = self._get(web_api, "get_hii_rainfall_latest", hours=26)
         assert len(rows) == 1
         assert rows[0]["oldcode"] == "CHM005"
         assert rows[0]["rain_24h"] == 49.6
         assert rows[0]["latitude"] == pytest.approx(19.12207)
 
     def test_waterlevel_latest(self, web_api):
-        import asyncio
-
-        rows = asyncio.run(web_api.get_hii_waterlevel_latest(hours=26))
+        rows, _ = self._get(web_api, "get_hii_waterlevel_latest", hours=26)
         assert len(rows) == 2
         p1 = next(r for r in rows if r["oldcode"] == "P.1")
         assert p1["rid_code"] == "P.1"
@@ -319,18 +327,14 @@ class TestHiiApiEndpoints:
         assert p1["situation_level"] == 4
 
     def test_empty_when_collector_disabled(self, monkeypatch):
-        import asyncio
-
         from src import web_api
 
         monkeypatch.setitem(web_api.app_state, "hii_collector", None)
         web_api.HII_CACHE.clear()
-        assert asyncio.run(web_api.get_hii_rainfall_latest(hours=26)) == []
-        assert asyncio.run(web_api.get_hii_waterlevel_latest(hours=26)) == []
+        assert self._get(web_api, "get_hii_rainfall_latest", hours=26)[0] == []
+        assert self._get(web_api, "get_hii_waterlevel_latest", hours=26)[0] == []
 
     def test_latest_responses_are_cached(self, web_api, monkeypatch):
-        import asyncio
-
         calls = {"n": 0}
         real = web_api._hii_rows
 
@@ -339,17 +343,33 @@ class TestHiiApiEndpoints:
             return real(sql, params)
 
         monkeypatch.setattr(web_api, "_hii_rows", counting)
-        first = asyncio.run(web_api.get_hii_rainfall_latest(hours=26))
-        second = asyncio.run(web_api.get_hii_rainfall_latest(hours=26))
+        first, _ = self._get(web_api, "get_hii_rainfall_latest", hours=26)
+        second, _ = self._get(web_api, "get_hii_rainfall_latest", hours=26)
         assert first == second and len(first) == 1
         assert calls["n"] == 1  # second call served from the TTL cache
         # different hours -> different cache key -> fresh query
-        asyncio.run(web_api.get_hii_rainfall_latest(hours=48))
+        self._get(web_api, "get_hii_rainfall_latest", hours=48)
         assert calls["n"] == 2
 
-    def test_empty_results_are_not_cached(self, tmp_path, monkeypatch):
-        import asyncio
+    def test_stale_served_on_recompute_failure(self, web_api, monkeypatch):
+        # Prime the cache, expire it, break the DB: the stale copy is served
+        # and flagged via the X-Data-Stale header.
+        good, response = self._get(web_api, "get_hii_rainfall_latest", hours=26)
+        assert good and "x-data-stale" not in response.headers
 
+        from src.config import Config
+
+        monkeypatch.setattr(Config, "HII_CACHE_TTL_SECONDS", 0)
+
+        def broken(sql, params):
+            raise RuntimeError("db unreachable")
+
+        monkeypatch.setattr(web_api, "_hii_rows", broken)
+        rows, response = self._get(web_api, "get_hii_rainfall_latest", hours=26)
+        assert rows == good
+        assert response.headers["X-Data-Stale"] == "true"
+
+    def test_empty_results_are_not_cached(self, tmp_path, monkeypatch):
         from src import web_api
         from src.hii_collector import HiiCollector
 
@@ -358,9 +378,51 @@ class TestHiiApiEndpoints:
         )
         monkeypatch.setitem(web_api.app_state, "hii_collector", collector)
         web_api.HII_CACHE.clear()
-        assert asyncio.run(web_api.get_hii_rainfall_latest(hours=26)) == []
+        assert self._get(web_api, "get_hii_rainfall_latest", hours=26)[0] == []
         assert web_api.HII_CACHE == {}  # empty response left uncached
 
+    def test_measurements_latest_cached_and_stale(self, monkeypatch):
+        from types import SimpleNamespace
+
+        from src import web_api
+        from src.config import Config
+
+        calls = {"n": 0}
+
+        def get_latest_data(limit=100):
+            calls["n"] += 1
+            return [
+                {
+                    "timestamp": "2026-08-12 10:00:00",
+                    "station_code": "P.1",
+                    "station_name_en": "Nawarat Bridge",
+                    "station_name_th": "สะพานนวรัฐ",
+                    "water_level": 2.76,
+                    "discharge": 331.0,
+                    "discharge_percent": 77.9,
+                }
+            ]
+
+        scraper = SimpleNamespace(db_adapter=True, get_latest_data=get_latest_data)
+        monkeypatch.setitem(web_api.app_state, "scraper", scraper)
+        web_api.LATEST_CACHE.clear()
+
+        rows, response = self._get(web_api, "get_latest_measurements", limit=500)
+        rows2, _ = self._get(web_api, "get_latest_measurements", limit=500)
+        assert calls["n"] == 1  # second call cached
+        assert rows2[0].station_code == "P.1"
+
+        # DB failure after expiry -> stale copy + header
+        monkeypatch.setattr(Config, "LATEST_CACHE_TTL_SECONDS", 0)
+
+        def broken(limit=100):
+            raise RuntimeError("db unreachable")
+
+        scraper.get_latest_data = broken
+        rows3, response = self._get(web_api, "get_latest_measurements", limit=500)
+        assert rows3[0].water_level == 2.76
+        assert response.headers["X-Data-Stale"] == "true"
+
 
 class TestBackfillHelpers:
     def test_chunk_date_range(self):