perf: 120s TTL cache with single-flight on /api/hii/*/latest

These endpoints ran an uncached latest-per-station aggregation on every
request (p50 0.6-0.9s under load, ~30% of the traffic mix) for data the
collector refreshes hourly. Responses are now cached per (feed, hours)
for HII_CACHE_TTL_SECONDS (default 120) with a per-feed compute lock so
a cache miss runs one query regardless of concurrency; empty results are
never cached so recovery is immediate. Cached hits measure ~8ms. Cache
is process-local behind a single helper — the seam where a shared
backend (Redis) would slot in if the deployment ever moves to multiple
workers; not warranted at one.

src/config.py

@@ -86,6 +86,8 @@ class Config:
         "yes",
     )
     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"))
 
     # 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

@@ -641,6 +641,37 @@ 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.
+HII_CACHE: Dict[str, Any] = {}
+HII_CACHE_LOCK = Lock()
+_HII_COMPUTE_LOCKS = {"rain": Lock(), "waterlevel": 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]
+        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
+
+
 @app.get("/api/hii/rainfall/latest")
 async def get_hii_rainfall_latest(hours: int = Query(26, ge=1, le=168)):
     """Latest rainfall reading per HII station (Ping basin, collected hourly)."""
@@ -656,8 +687,7 @@ 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
     """
-    cutoff = datetime.now() - timedelta(hours=hours)
+    return await asyncio.to_thread(_hii_cached, "rain", hours, sql)
-    return await asyncio.to_thread(_hii_rows, sql, {"cutoff": cutoff})
 
 
 @app.get("/api/hii/waterlevel/latest")
@@ -677,8 +707,7 @@ 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
     """
-    cutoff = datetime.now() - timedelta(hours=hours)
+    return await asyncio.to_thread(_hii_cached, "waterlevel", hours, sql)
-    return await asyncio.to_thread(_hii_rows, sql, {"cutoff": cutoff})
 
 
 @app.get("/measurements/history/{station_code}")

tests/test_hii_collector.py

@@ -295,6 +295,7 @@ class TestHiiApiEndpoints:
         assert collector.store.save_rain(rain) == 1
         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
         return web_api
 
     def test_rainfall_latest(self, web_api):
@@ -323,9 +324,43 @@ class TestHiiApiEndpoints:
         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)) == []
 
+    def test_latest_responses_are_cached(self, web_api, monkeypatch):
+        import asyncio
+
+        calls = {"n": 0}
+        real = web_api._hii_rows
+
+        def counting(sql, params):
+            calls["n"] += 1
+            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))
+        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))
+        assert calls["n"] == 2
+
+    def test_empty_results_are_not_cached(self, tmp_path, monkeypatch):
+        import asyncio
+
+        from src import web_api
+        from src.hii_collector import HiiCollector
+
+        collector = HiiCollector(
+            {"type": "sqlite", "connection_string": f"sqlite:///{tmp_path}/empty.db"}
+        )
+        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 web_api.HII_CACHE == {}  # empty response left uncached
+
 
 class TestBackfillHelpers:
     def test_chunk_date_range(self):