perf: batch health-check writes, PRAGMA synchronous=NORMAL, concurrent exchange fetch (CashPilot-perf) (#123)

From the research performance panel — three low-risk hot-path fixes:

1. Health-check cycle committed one health event PER service (up to ~49 fsync'd
   commits/cycle, 288 cycles/day). Collect the cycle's events and write them in a
   single transaction via new database.record_health_events (one executemany + one
   commit) — 27-98x fewer commits on the write-heaviest path.
2. PRAGMA synchronous=NORMAL in _get_db: durable in WAL across app crashes, skips an
   fsync on every commit — a fleet-wide win on top of #1.
3. exchange_rates.refresh() fetched CoinGecko then Frankfurter sequentially, awaited
   on the startup critical path (up to 30s worst case with two 15s timeouts). Fetch
   both concurrently with asyncio.gather(return_exceptions) — halves the worst case
   and makes one source's failure not discard the other's success.

Behavior-preserving. Tests: record_health_events batched write + empty no-op; the
health-check tests updated to the batched contract. Full suite 1173 green; ruff clean.

Author: GeiserX

app/database.py

@@ -280,6 +280,10 @@ async def _get_db() -> _BorrowedConnection:
         await conn.execute("PRAGMA journal_mode=WAL")
         await conn.execute("PRAGMA foreign_keys=ON")
         await conn.execute("PRAGMA busy_timeout=5000")
+        # In WAL mode NORMAL is durable across app crashes (only a power loss can lose
+        # the last transactions) and skips an fsync on every commit — a large win on the
+        # write-heavy health-check path that commits per service each cycle.
+        await conn.execute("PRAGMA synchronous=NORMAL")
         _shared_conns[key] = conn
 
     return _BorrowedConnection(conn)
@@ -1178,6 +1182,26 @@ async def record_health_event(slug: str, event: str, detail: str = "") -> None:
         await db.close()
 
 
+async def record_health_events(events: list[tuple[str, str, str]]) -> None:
+    """Record many health events in ONE transaction/commit.
+
+    The health-check cycle writes one event per deployed service; committing each
+    separately fsync'd the WAL up to ~49 times per cycle. One executemany + one commit
+    collapses that to a single write — the dominant fix for that path's I/O.
+    """
+    if not events:
+        return
+    db = await _get_db()
+    try:
+        await db.executemany(
+            "INSERT INTO health_events (slug, event, detail) VALUES (?, ?, ?)",
+            events,
+        )
+        await db.commit()
+    finally:
+        await db.close()
+
+
 async def get_health_scores(days: int = 7) -> list[dict[str, Any]]:
     """Compute health score per service over the last N days.
 

app/exchange_rates.py

@@ -9,6 +9,7 @@
 
 from __future__ import annotations
 
+import asyncio
 import logging
 import time
 from typing import Any
@@ -52,53 +53,20 @@ async def refresh() -> None:
     time -- a non-200 (or an unreachable API) leaves it untouched so staleness is
     tracked per-source and a partial failure can't mislabel the other source.
     """
-    global _fiat_rates, _crypto_usd, _last_fetch, _crypto_last_fetch, _fiat_last_fetch
+    global _last_fetch
 
     try:
         async with httpx.AsyncClient(timeout=15, follow_redirects=True) as client:
-            # --- Crypto rates from CoinGecko (free, no key) ---
-            if CRYPTO_IDS:
-                ids = ",".join(CRYPTO_IDS.values())
-                resp = await client.get(
-                    "https://api.coingecko.com/api/v3/simple/price",
-                    params={"ids": ids, "vs_currencies": "usd"},
-                )
-                if resp.status_code == 200:
-                    data = resp.json()
-                    for token, cg_id in CRYPTO_IDS.items():
-                        price = (data.get(cg_id) or {}).get("usd")
-                        if price is not None:
-                            _crypto_usd[token] = float(price)
-                    _crypto_last_fetch = time.time()
-                else:
-                    logger.warning(
-                        "exchange rate fetch got HTTP %s from %s",
-                        resp.status_code,
-                        "CoinGecko",
-                    )
-            else:
-                # Nothing to fetch -- don't let an empty crypto map hold the
-                # aggregate staleness clock back forever.
-                _crypto_last_fetch = time.time()
-
-            # --- Fiat rates from Frankfurter (free, no key) ---
-            resp = await client.get(
-                "https://api.frankfurter.app/latest",
-                params={"from": "USD"},
-            )
-            if resp.status_code == 200:
-                data = resp.json()
-                new_rates: dict[str, float] = {"USD": 1.0}
-                for code, rate in data.get("rates", {}).items():
-                    new_rates[code] = float(rate)
-                _fiat_rates = new_rates
-                _fiat_last_fetch = time.time()
-            else:
-                logger.warning(
-                    "exchange rate fetch got HTTP %s from %s",
-                    resp.status_code,
-                    "Frankfurter",
-                )
+            # Fetch both sources concurrently — they update independent state, so a slow
+            # provider no longer serializes behind the other. This roughly halves the
+            # worst case (two 15s-timeout calls: 30s -> 15s), which matters most on the
+            # startup path where refresh() is awaited before the app serves requests.
+            # return_exceptions so one source's network failure can't discard the other's
+            # success (each source still tracks its own last-fetch / non-200 internally).
+            results = await asyncio.gather(_fetch_crypto(client), _fetch_fiat(client), return_exceptions=True)
+        for r in results:
+            if isinstance(r, Exception):
+                logger.error("Exchange rate fetch failed: %s", r)
 
         _last_fetch = min(_crypto_last_fetch, _fiat_last_fetch)
         logger.info(
@@ -110,6 +78,48 @@ async def refresh() -> None:
         logger.error("Exchange rate fetch failed: %s", exc)
 
 
+async def _fetch_crypto(client: httpx.AsyncClient) -> None:
+    """Fetch crypto→USD from CoinGecko (free, no key). Updates crypto state in place."""
+    global _crypto_usd, _crypto_last_fetch
+    if not CRYPTO_IDS:
+        # Nothing to fetch -- don't let an empty crypto map hold the aggregate
+        # staleness clock back forever.
+        _crypto_last_fetch = time.time()
+        return
+    ids = ",".join(CRYPTO_IDS.values())
+    resp = await client.get(
+        "https://api.coingecko.com/api/v3/simple/price",
+        params={"ids": ids, "vs_currencies": "usd"},
+    )
+    if resp.status_code == 200:
+        data = resp.json()
+        for token, cg_id in CRYPTO_IDS.items():
+            price = (data.get(cg_id) or {}).get("usd")
+            if price is not None:
+                _crypto_usd[token] = float(price)
+        _crypto_last_fetch = time.time()
+    else:
+        logger.warning("exchange rate fetch got HTTP %s from %s", resp.status_code, "CoinGecko")
+
+
+async def _fetch_fiat(client: httpx.AsyncClient) -> None:
+    """Fetch USD→fiat from Frankfurter (free, no key). Updates fiat state in place."""
+    global _fiat_rates, _fiat_last_fetch
+    resp = await client.get(
+        "https://api.frankfurter.app/latest",
+        params={"from": "USD"},
+    )
+    if resp.status_code == 200:
+        data = resp.json()
+        new_rates: dict[str, float] = {"USD": 1.0}
+        for code, rate in data.get("rates", {}).items():
+            new_rates[code] = float(rate)
+        _fiat_rates = new_rates
+        _fiat_last_fetch = time.time()
+    else:
+        logger.warning("exchange rate fetch got HTTP %s from %s", resp.status_code, "Frankfurter")
+
+
 def rates_stale() -> bool:
     """Return True if cached rates are stale (refreshes appear to be failing).
 

app/main.py

@@ -208,11 +208,14 @@ async def _run_health_check() -> None:
             status = s.get("status", "unknown")
             if slug_best.get(slug) != "running":
                 slug_best[slug] = status
+        # Collect every event for this cycle and write them in a single transaction
+        # rather than one fsync'd commit per service (see database.record_health_events).
+        events: list[tuple[str, str, str]] = []
         for slug, status in slug_best.items():
             if status == "running":
-                await database.record_health_event(slug, "check_ok")
+                events.append((slug, "check_ok", ""))
             else:
-                await database.record_health_event(slug, "check_down", status)
+                events.append((slug, "check_down", status))
 
         workers = await database.list_workers()
         if any(w.get("status") == "online" for w in workers):
@@ -221,7 +224,9 @@ async def _run_health_check() -> None:
                 slug = d["slug"]
                 if d.get("status") == "external" or slug in slug_best:
                     continue
-                await database.record_health_event(slug, "check_down", "missing from heartbeat")
+                events.append((slug, "check_down", "missing from heartbeat"))
+
+        await database.record_health_events(events)
     except Exception as exc:
         logger.warning("Health check skipped: %s", exc)
 

tests/test_coverage_gaps.py

@@ -339,15 +339,18 @@ def test_health_check_records_events(self):
         mock_record = AsyncMock()
         with (
             patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
-            patch("app.main.database.record_health_event", mock_record),
+            patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=[]),
+            patch("app.main.database.record_health_events", mock_record),
         ):
             asyncio.run(_run_health_check())
 
-        # Should record check_ok for honeygain (running) and check_down for earnapp (stopped)
-        calls = mock_record.call_args_list
-        slugs_events = [(c.args[0], c.args[1]) for c in calls]
-        assert ("honeygain", "check_ok") in slugs_events
-        assert ("earnapp", "check_down") in slugs_events
+        # One batched write carrying check_ok for honeygain (running) and check_down for
+        # earnapp (stopped).
+        mock_record.assert_awaited_once()
+        events = mock_record.call_args.args[0]
+        pairs = [(e[0], e[1]) for e in events]
+        assert ("honeygain", "check_ok") in pairs
+        assert ("earnapp", "check_down") in pairs
 
 
 class TestMainStaleWorkers:

tests/test_database.py

@@ -427,6 +427,26 @@ async def run():
 
         asyncio.run(run())
 
+    def test_record_health_events_batched(self, db):
+        async def run():
+            # One batched write of several events across two services (the health-check
+            # path uses this instead of a commit per service).
+            await database.record_health_events(
+                [
+                    ("honeygain", "check_ok", ""),
+                    ("honeygain", "restart", ""),
+                    ("earnapp", "check_down", "stopped"),
+                ]
+            )
+            by_slug = {s["slug"]: s for s in await database.get_health_scores(7)}
+            assert set(by_slug) == {"honeygain", "earnapp"}
+            assert by_slug["honeygain"]["restarts"] == 1
+            # An empty batch is a no-op — no crash, nothing written.
+            await database.record_health_events([])
+            assert len(await database.get_health_scores(7)) == 2
+
+        asyncio.run(run())
+
 
 class TestWorkerKeys:
     def test_workers_table_has_api_key_enc_column(self, db):

tests/test_workers.py

@@ -390,6 +390,11 @@ def test_bad_worker_entry_does_not_abort_others(self):
 # ---------------------------------------------------------------------------
 
 
+def _batched_events(mock):
+    """The single batched event list passed to record_health_events ([] if uncalled)."""
+    return list(mock.call_args.args[0]) if mock.call_args else []
+
+
 class TestHealthCheckVanishedService:
     def test_known_deployment_missing_from_heartbeat_gets_check_down(self):
         """A Docker-backed deployment absent from every online worker's
@@ -402,10 +407,10 @@ def test_known_deployment_missing_from_heartbeat_gets_check_down(self):
         with (
             patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
             patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=deployments),
-            patch("app.main.database.record_health_event", mock_record),
+            patch("app.main.database.record_health_events", mock_record),
         ):
             _run(_run_health_check())
-        mock_record.assert_any_call("honeygain", "check_down", "missing from heartbeat")
+        assert ("honeygain", "check_down", "missing from heartbeat") in _batched_events(mock_record)
 
     def test_external_deployment_never_flagged_missing(self):
         """External (no-container) deployments like Grass/Bytelixir are never
@@ -417,10 +422,10 @@ def test_external_deployment_never_flagged_missing(self):
         with (
             patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
             patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=deployments),
-            patch("app.main.database.record_health_event", mock_record),
+            patch("app.main.database.record_health_events", mock_record),
         ):
             _run(_run_health_check())
-        mock_record.assert_not_called()
+        assert all(e[0] != "grass" for e in _batched_events(mock_record))
 
     def test_fully_offline_fleet_does_not_flag_missing(self):
         """With no worker online there is no heartbeat data to trust either
@@ -432,10 +437,10 @@ def test_fully_offline_fleet_does_not_flag_missing(self):
         with (
             patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
             patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=deployments),
-            patch("app.main.database.record_health_event", mock_record),
+            patch("app.main.database.record_health_events", mock_record),
         ):
             _run(_run_health_check())
-        mock_record.assert_not_called()
+        assert _batched_events(mock_record) == []
 
     def test_service_present_in_heartbeat_not_double_flagged(self):
         """A service still reported by a worker must get exactly one health
@@ -453,10 +458,11 @@ def test_service_present_in_heartbeat_not_double_flagged(self):
         with (
             patch("app.main.database.list_workers", new_callable=AsyncMock, return_value=workers),
             patch("app.main.database.get_deployments", new_callable=AsyncMock, return_value=deployments),
-            patch("app.main.database.record_health_event", mock_record),
+            patch("app.main.database.record_health_events", mock_record),
         ):
             _run(_run_health_check())
-        mock_record.assert_called_once_with("honeygain", "check_ok")
+        # Exactly one event (the normal check_ok), never a second "missing" on top.
+        assert _batched_events(mock_record) == [("honeygain", "check_ok", "")]
 
 
 # ---------------------------------------------------------------------------