Keep the spool's file list in memory instead of rescanning it

Every oldest(), stats() and put() globbed and sorted the whole spool
directory, so draining a backlog cost O(N^2 log N). After an outage on
a Pi 3B+ the writer thread sat at 100% CPU for minutes working through
3,300 batches. The directory is now scanned once at startup and the
names, sizes and total are tracked as batches are added and removed;
draining 3,300 batches drops from 59s to 0.24s on a laptop.

Bump to 0.1.6.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

openhop-plugin.json

@@ -2,7 +2,7 @@
   "schema": 1,
   "id": "community.influxdb",
   "name": "InfluxDB Exporter",
-  "version": "0.1.5",
+  "version": "0.1.6",
   "description": "Pushes Repeater metrics, per-packet events, logs and node/GPS snapshots to a remote InfluxDB (v1, v2 or v3) for Grafana dashboards.",
   "runtime": {
     "type": "python",

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "openhop-influxdb-plugin"
-version = "0.1.5"
+version = "0.1.6"
 description = "openHop Repeater plugin that pushes metrics, packets, logs and snapshots to InfluxDB"
 readme = "README.md"
 requires-python = ">=3.10"

src/openhop_influxdb_plugin/__init__.py

@@ -1 +1 @@
-__version__ = "0.1.5"
+__version__ = "0.1.6"

src/openhop_influxdb_plugin/spool.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import bisect
 import gzip
 import logging
 import os
@@ -15,6 +16,9 @@ class Spool:
 
     Batches that cannot be delivered are written here and replayed oldest first.
     When the size cap is exceeded the oldest batches are discarded.
+
+    The directory is scanned once, at startup; after that the file names and sizes
+    are tracked in memory, so a large backlog costs nothing to drain or measure.
     """
 
     def __init__(self, directory: str | Path, max_bytes: int):
@@ -23,17 +27,39 @@ class Spool:
         self._lock = threading.Lock()
         self._seq = 0
         self.dropped_batches = 0
+        self._names: list[str] = []  # sorted, so oldest first
+        self._sizes: dict[str, int] = {}
+        self._total = 0
         if self.enabled:
             self.directory.mkdir(parents=True, exist_ok=True)
+            self._load()
 
     @property
     def enabled(self) -> bool:
         return self.max_bytes > 0
 
-    def _files(self) -> list[Path]:
-        if not self.directory.exists():
-            return []
-        return sorted(self.directory.glob("*.lp.gz"))
+    def _load(self) -> None:
+        for path in self.directory.glob("*.lp.gz"):
+            try:
+                size = path.stat().st_size
+            except OSError:
+                continue
+            self._sizes[path.name] = size
+            self._total += size
+        self._names = sorted(self._sizes)
+
+    def _forget(self, name: str) -> None:
+        size = self._sizes.pop(name, None)
+        if size is None:
+            return
+        self._total -= size
+        i = bisect.bisect_left(self._names, name)
+        if i < len(self._names) and self._names[i] == name:
+            del self._names[i]
+
+    def _discard(self, name: str) -> None:
+        self._forget(name)
+        (self.directory / name).unlink(missing_ok=True)
 
     def put(self, body: bytes) -> bool:
         if not self.enabled:
@@ -42,37 +68,39 @@ class Spool:
         with self._lock:
             self._seq += 1
             name = f"{time.time_ns():020d}-{self._seq:06d}.lp.gz"
+            data = gzip.compress(body)
             tmp = self.directory / (name + ".tmp")
-            tmp.write_bytes(gzip.compress(body))
+            tmp.write_bytes(data)
             os.replace(tmp, self.directory / name)
+            self._sizes[name] = len(data)
+            self._total += len(data)
+            bisect.insort(self._names, name)
             self._enforce_cap()
         return True
 
     def _enforce_cap(self) -> None:
-        files = self._files()
-        total = sum(f.stat().st_size for f in files)
-        while files and total > self.max_bytes:
-            oldest = files.pop(0)
-            total -= oldest.stat().st_size
-            oldest.unlink(missing_ok=True)
+        while self._names and self._total > self.max_bytes:
+            oldest = self._names[0]
+            self._discard(oldest)
             self.dropped_batches += 1
-            logger.warning(
-                "Spool over %d bytes; discarded oldest batch %s", self.max_bytes, oldest.name
-            )
+            logger.warning("Spool over %d bytes; discarded oldest batch %s", self.max_bytes, oldest)
 
     def oldest(self) -> tuple[Path, bytes] | None:
         with self._lock:
-            for path in self._files():
+            while self._names:
+                name = self._names[0]
+                path = self.directory / name
                 try:
                     return path, gzip.decompress(path.read_bytes())
                 except (OSError, EOFError) as exc:
-                    logger.warning("Discarding unreadable spool file %s: %s", path.name, exc)
-                    path.unlink(missing_ok=True)
+                    logger.warning("Discarding unreadable spool file %s: %s", name, exc)
+                    self._discard(name)
             return None
 
     def remove(self, path: Path) -> None:
-        path.unlink(missing_ok=True)
+        with self._lock:
+            self._discard(path.name)
 
     def stats(self) -> tuple[int, int]:
-        files = self._files()
-        return len(files), sum(f.stat().st_size for f in files)
+        with self._lock:
+            return len(self._names), self._total

tests/test_spool.py

@@ -0,0 +1,73 @@
+from pathlib import Path
+
+from openhop_influxdb_plugin.spool import Spool
+
+
+def _drain(spool):
+    bodies = []
+    while (item := spool.oldest()) is not None:
+        path, body = item
+        bodies.append(body)
+        spool.remove(path)
+    return bodies
+
+
+def test_drains_oldest_first(tmp_path):
+    spool = Spool(tmp_path, max_bytes=1024 * 1024)
+    for i in range(5):
+        spool.put(f"m v={i}".encode())
+    assert _drain(spool) == [f"m v={i}".encode() for i in range(5)]
+    assert spool.stats() == (0, 0)
+    assert list(tmp_path.glob("*.lp.gz")) == []
+
+
+def test_reloads_backlog_left_by_previous_run(tmp_path):
+    first = Spool(tmp_path, max_bytes=1024 * 1024)
+    for i in range(3):
+        first.put(f"m v={i}".encode())
+    files, size = first.stats()
+
+    second = Spool(tmp_path, max_bytes=1024 * 1024)
+    assert second.stats() == (files, size)
+    second.put(b"m v=3")
+    assert _drain(second) == [f"m v={i}".encode() for i in range(4)]
+
+
+def test_stats_track_puts_and_removes(tmp_path):
+    spool = Spool(tmp_path, max_bytes=1024 * 1024)
+    for i in range(4):
+        spool.put(f"m v={i}".encode())
+    on_disk = sum(p.stat().st_size for p in tmp_path.glob("*.lp.gz"))
+    assert spool.stats() == (4, on_disk)
+    path, _ = spool.oldest()
+    spool.remove(path)
+    on_disk = sum(p.stat().st_size for p in tmp_path.glob("*.lp.gz"))
+    assert spool.stats() == (3, on_disk)
+
+
+def test_skips_unreadable_and_vanished_files(tmp_path):
+    spool = Spool(tmp_path, max_bytes=1024 * 1024)
+    for i in range(3):
+        spool.put(f"m v={i}".encode())
+    names = sorted(p.name for p in tmp_path.glob("*.lp.gz"))
+    (tmp_path / names[0]).write_bytes(b"not gzip")
+    (tmp_path / names[1]).unlink()
+    path, body = spool.oldest()
+    assert body == b"m v=2"
+    assert spool.stats()[0] == 1
+    assert not (tmp_path / names[0]).exists()
+
+
+def test_large_backlog_is_not_rescanned(tmp_path, monkeypatch):
+    spool = Spool(tmp_path, max_bytes=100 * 1024 * 1024)
+    for i in range(500):
+        spool.put(f"m v={i}".encode())
+
+    def no_scan(*args, **kwargs):
+        raise AssertionError("spool directory rescanned")
+
+    monkeypatch.setattr(Path, "glob", no_scan)
+    monkeypatch.setattr(Path, "iterdir", no_scan)
+    spool.put(b"m v=500")
+    assert spool.stats()[0] == 501
+    assert len(_drain(spool)) == 501