feat: enhance SQLiteHandler and StorageCollector for improved caching and async performance

- Added caching for packet type stats and cumulative counts in SQLiteHandler to reduce database load.
- Implemented a dedicated writer thread in StorageCollector to handle blocking storage operations, preventing asyncio event loop stalls.
- Updated record_packet method to utilize the new writer thread for efficient packet processing.

repeater/data_acquisition/sqlite_handler.py

@@ -19,7 +19,16 @@ class SQLiteHandler:
         self._api_token_last_used_interval_sec = 300
         self._hot_cache_ttl_sec = 60
         self._packet_stats_cache = {}
+        self._packet_type_stats_cache = {}
         self._neighbors_cache = {"timestamp": 0.0, "value": None}
+        # Short time-based cache for the per-packet cumulative-counts aggregate
+        # (two full-table scans). The storage writer thread calls this once per
+        # recorded packet/duplicate; a few seconds of staleness is fine for the
+        # RRD/UI counters and stops a full scan running on every packet.
+        # Intentionally NOT cleared by _invalidate_hot_caches() — that runs on
+        # every write, which would defeat the cache under load.
+        self._cumulative_counts_cache = {"timestamp": 0.0, "value": None}
+        self._cumulative_counts_ttl_sec = 3.0
         # Thread-local storage for persistent SQLite connections.
         # Opening a new connection on every DB call is expensive on SD-card
         # storage: each sqlite3.connect() call triggers file-system operations
@@ -1186,7 +1195,11 @@ class SQLiteHandler:
 
     def get_packet_type_stats(self, hours: int = 24) -> dict:
         try:
-            cutoff = time.time() - (hours * 3600)
+            now = time.time()
+            cached = self._packet_type_stats_cache.get(hours)
+            if cached and (now - cached["timestamp"]) < self._hot_cache_ttl_sec:
+                return cached["value"]
+            cutoff = now - (hours * 3600)
 
             # Align with pyMC_core feat/newRadios PAYLOAD_TYPES (0x0B = CONTROL)
             try:
@@ -1262,13 +1275,15 @@ class SQLiteHandler:
                 if other_count > 0:
                     type_counts["Other Types (>15)"] = other_count
 
-                return {
+                result = {
                     "hours": hours,
                     "packet_type_totals": type_counts,
                     "total_packets": sum(type_counts.values()),
                     "period": f"{hours} hours",
                     "data_source": "sqlite",
                 }
+                self._packet_type_stats_cache[hours] = {"timestamp": now, "value": result}
+                return result
 
         except Exception as e:
             logger.error(f"Failed to get packet type stats from SQLite: {e}")
@@ -1602,6 +1617,11 @@ class SQLiteHandler:
             logger.error(f"Failed to cleanup old data: {e}")
 
     def get_cumulative_counts(self) -> dict:
+        now = time.time()
+        cached = self._cumulative_counts_cache.get("value")
+        cached_ts = float(self._cumulative_counts_cache.get("timestamp", 0.0))
+        if cached is not None and (now - cached_ts) < self._cumulative_counts_ttl_sec:
+            return cached
         try:
             with self._connect() as conn:
                 conn.row_factory = sqlite3.Row
@@ -1630,12 +1650,14 @@ class SQLiteHandler:
                 """
                 ).fetchone()
 
-                return {
+                result = {
                     "rx_total": int(totals["rx_total"] or 0),
                     "tx_total": int(totals["tx_total"] or 0),
                     "drop_total": int(totals["drop_total"] or 0),
                     "type_counts": type_counts,
                 }
+                self._cumulative_counts_cache = {"timestamp": now, "value": result}
+                return result
 
         except Exception as e:
             logger.error(f"Failed to get cumulative counts: {e}")

repeater/data_acquisition/storage_collector.py

@@ -1,4 +1,5 @@
 import asyncio
+import concurrent.futures
 import logging
 import time
 from typing import Optional
@@ -20,6 +21,17 @@ class StorageCollector:
         self.glass_publish_callback = None
         self._pending_tasks = set()
 
+        # Dedicated single writer thread for all blocking storage work (the SQLite
+        # write, the cumulative-counts aggregate, RRD updates, and network
+        # publishing). This keeps that work off the asyncio event loop, which it
+        # was previously stalling for seconds per packet on a busy mesh — starving
+        # every other coroutine (e.g. send_advert would time out). One worker
+        # preserves packet write ordering and reuses a single thread-local SQLite
+        # connection (no WAL writer contention, no connection fan-out).
+        self._db_executor = concurrent.futures.ThreadPoolExecutor(
+            max_workers=1, thread_name_prefix="storage-writer"
+        )
+
         self.storage_dir = resolve_storage_dir(config)
         self.storage_dir.mkdir(parents=True, exist_ok=True)
 
@@ -145,7 +157,12 @@ class StorageCollector:
         return stats
 
     def record_packet(self, packet_record: dict, skip_mqtt_if_invalid: bool = True):
-        """Record packet to storage and publish to MQTT
+        """Record a packet to storage and publish it.
+
+        All blocking work — the SQLite write, the cumulative-counts aggregate, the
+        RRD update, and network publishing — runs on the dedicated writer thread so
+        it never blocks the asyncio event loop. Callers treat this as
+        fire-and-forget (the previous synchronous version blocked the loop).
 
         Args:
             packet_record: Dictionary containing packet information
@@ -155,27 +172,32 @@ class StorageCollector:
             f"Recording packet: type={packet_record.get('type')}, "
             f"transmitted={packet_record.get('transmitted')}"
         )
+        self._submit_db(self._record_packet_blocking, packet_record, skip_mqtt_if_invalid)
 
-        # HOT PATH: Store to local databases only (fast, non-blocking)
+    def _submit_db(self, fn, *args):
+        """Run a blocking storage operation on the dedicated writer thread.
+
+        Falls back to running inline only if the executor has already been shut
+        down (process teardown), so late records are not silently dropped.
+        """
+        try:
+            self._db_executor.submit(self._run_db_task, fn, *args)
+        except RuntimeError:
+            self._run_db_task(fn, *args)
+
+    def _run_db_task(self, fn, *args):
+        """Execute a writer-thread task, logging (not raising) on failure."""
+        try:
+            fn(*args)
+        except Exception as e:
+            logger.error(f"Storage writer task failed: {e}", exc_info=True)
+
+    def _record_packet_blocking(self, packet_record: dict, skip_mqtt: bool):
+        """Store, aggregate, update metrics, and publish one packet (writer thread)."""
         self.sqlite_handler.store_packet(packet_record)
         cumulative_counts = self.sqlite_handler.get_cumulative_counts()
         self.rrd_handler.update_packet_metrics(packet_record, cumulative_counts)
-
-        # DEFERRED: Publish to network sinks and WebSocket in background tasks
-        # This prevents network latency from blocking packet processing
-        self._schedule_background(
-            self._deferred_publish,
-            packet_record,
-            skip_mqtt_if_invalid,
-            sync_fallback=self._publish_packet_sync,
-        )
-
-    async def _deferred_publish(self, packet_record: dict, skip_mqtt: bool):
-        """Deferred background task for all network publishing operations."""
-        try:
-            self._publish_packet_sync(packet_record, skip_mqtt)
-        except Exception as e:
-            logger.error(f"Deferred publish failed: {e}", exc_info=True)
+        self._publish_packet_sync(packet_record, skip_mqtt)
 
     def _publish_packet_sync(self, packet_record: dict, skip_mqtt: bool):
         """Publish packet updates synchronously (used when no asyncio loop is active)."""
@@ -421,6 +443,10 @@ class StorageCollector:
         return self.sqlite_handler.get_noise_floor_stats(hours)
 
     def close(self):
+        # Drain and stop the storage writer thread first so pending writes and
+        # publishes complete before MQTT and the DB connections are torn down.
+        self._db_executor.shutdown(wait=True)
+
         # Cancel all pending background tasks
         for task in self._pending_tasks:
             if not task.done():