Optional retention: trim old raw JSON, keep every column forever

README.md

@@ -51,6 +51,7 @@ systemd service on an always-on box, see [Running wallmonitor](https://github.co
   serial is pinned on first contact so a swapped or second unit can never
   blend into your history
 - Full-fidelity recording — every response stored with its complete raw JSON
+  (optional retention trims old raw blobs so the database stays near-flat)
 - Live dashboard (SSE) with rolling charts, an active-alert banner, and a
   live derate-forecast chart: measured handle temperature against the
   model's projected plateau and the trip threshold, projection drawn

docs/recording.md

@@ -7,6 +7,9 @@ around gaps, glitches, and unverified labels. The numbered points below are
 the recording pipeline in the order data flows through it.
 
 1. **Records everything the charger reports, at the highest safe fidelity.**
+   (With the optional `--retain-raw-days` policy, samples older than the
+   window keep every extracted column but drop the raw JSON blob — see
+   [Retention](running.md#retention).)
    Vitals every 2 s while a vehicle is attached (5 s idle), Wi-Fi status every
    30 s, lifetime counters every 60 s, firmware info every 6 h. Requests are
    strictly sequential and back off exponentially on failures, so the charger's

docs/running.md

@@ -45,6 +45,8 @@ is the charger's microcontroller.
 | `--discover [RANGE]` | — | — | Sweep the LAN for Wall Connectors and exit (own subnet, or a private CIDR) |
 | `--label` | `WM_LABEL` | — | Name for this charger, shown in the header, tab title and notifications |
 | `--peer LABEL=URL` | `WM_PEERS` | — | Link another instance from the header switcher; repeatable (env: comma-separated) |
+| `--retain-raw-days` | `WM_RETAIN_RAW_DAYS` | `0` (off) | Trim raw JSON from samples older than N days (min 7); columns stay forever |
+| `--compact` | — | — | One-shot VACUUM to reclaim trimmed space; run with the service stopped |
 | `--port` | `WM_PORT` | `8480` | Web UI port |
 | `--bind` | `WM_BIND` | `127.0.0.1` | Web UI bind address |
 | `--db` | `WM_DB` | `wallmonitor.db` | SQLite path |
@@ -139,6 +141,30 @@ sudo ./deploy/install-service.sh --name right --host 192.168.1.51 --port 8481 --
 
 A single-process, cross-device view is tracked as future work in issue #10.
 
+## Retention
+
+By default nothing is ever deleted: every sample keeps its complete raw
+JSON forever, and the database grows by roughly a gigabyte a month on a
+busy install (about 85 % of that is the raw blobs on vitals samples).
+
+`--retain-raw-days N` caps that. Samples older than N days keep **every
+extracted column** — charts, session pages, the thermal model, and the
+degradation watch see identical history — but their raw JSON blob is
+blanked by a daily background pass (chunked, so live polling never waits
+more than a moment; vitals wait until the diagnostics backfill has
+finished). What is given up is *re-interpretability*: extracting a field
+that was never a column, the way the diagnostics columns themselves were
+backfilled from raw, becomes impossible for trimmed rows. Forecast
+snapshots and version info are never trimmed.
+
+Freed pages are reused by new inserts, so the file stops growing rather
+than shrinking. To hand the space back to the filesystem once, stop the
+service and run:
+
+```bash
+uv run python -m wallmonitor --compact --db /path/to/wallmonitor.db
+```
+
 ## Run as a service (Ubuntu / systemd)
 
 For an always-on box, `deploy/install-service.sh` installs wallmonitor as a

tests/test_retention.py

@@ -0,0 +1,86 @@
+"""Retention: raw-JSON trim, its gates and cursor, readers on trimmed rows."""
+
+import time
+
+import pytest
+
+from wallmonitor.config import parse_args
+from wallmonitor.db import Database
+
+
+@pytest.fixture
+def db(tmp_path):
+    database = Database(str(tmp_path / "test.db"))
+    yield database
+    database.close()
+
+
+def _fill(db, days_ago, n=20):
+    base = time.time() - days_ago * 86400
+    for i in range(n):
+        ts = base + i * 10
+        db.insert_vitals(ts, {"handle_temp_c": 30.0, "vehicle_current_a": 16.0,
+                              "prox_v": 1.5, "pilot_high_v": 8.9}, 1, 3680.0)
+        db.insert_wifi(ts, {"wifi_rssi": -60, "wifi_connected": True})
+        db.insert_lifetime(ts, {"energy_wh": 1000})
+        db.insert_ambient(ts, 25.0, raw={"t": 25.0}, source="test")
+
+
+def _raw_counts(db):
+    return {t: db._rows(f"SELECT COUNT(*) AS n FROM {t} WHERE raw != ''")[0]["n"]
+            for t in Database.RAW_TRIM_TABLES}
+
+
+def test_trim_respects_cutoff_and_keeps_columns(db):
+    _fill(db, days_ago=30)
+    _fill(db, days_ago=1)
+    db.set_setting("diag_backfill_done", "1")
+    counts = db.trim_raw(time.time() - 7 * 86400)
+    assert counts["vitals_samples"] == 20 and counts["wifi_samples"] == 20
+    remaining = _raw_counts(db)
+    assert all(v == 20 for v in remaining.values()), remaining  # recent rows untouched
+    # Columns intact and served for the trimmed era.
+    old = db.vitals_range(time.time() - 31 * 86400, time.time() - 29 * 86400)
+    assert len(old) == 20 and all(s["handle_temp_c"] == 30.0 and s["prox_v"] == 1.5 for s in old)
+    # Bucketed path too (must not touch the blanked raw).
+    bucketed = db.vitals_range(time.time() - 31 * 86400, time.time() - 29 * 86400, max_points=4)
+    assert bucketed and all(b["prox_v"] == pytest.approx(1.5) for b in bucketed)
+
+
+def test_vitals_trim_waits_for_diag_backfill(db):
+    _fill(db, days_ago=30)
+    counts = db.trim_raw(time.time())
+    assert "vitals_samples" not in counts  # gated: backfill not done
+    assert counts["wifi_samples"] == 20    # other tables trim regardless
+    db.set_setting("diag_backfill_done", "1")
+    assert db.trim_raw(time.time())["vitals_samples"] == 20
+
+
+def test_trim_cursor_makes_second_run_cheap_and_correct(db):
+    _fill(db, days_ago=30)
+    db.set_setting("diag_backfill_done", "1")
+    db.trim_raw(time.time() - 7 * 86400)
+    # New rows age past the cutoff later; the cursor resumes, not rescans.
+    _fill(db, days_ago=8)
+    counts = db.trim_raw(time.time() - 7 * 86400)
+    assert counts["vitals_samples"] == 20
+    assert float(db.get_setting("raw_trim_ts:vitals_samples")) > time.time() - 9 * 86400
+
+
+def test_vacuum_reclaims_trimmed_space(db):
+    for i in range(300):
+        db.insert_vitals(time.time() - 40 * 86400 + i, {"handle_temp_c": 30.0, "pad": "x" * 2000}, 1, 0.0)
+    db.set_setting("diag_backfill_done", "1")
+    db.trim_raw(time.time())
+    before, after = db.vacuum()
+    # 300 rows x ~2KB of trimmed blob must come back from the file.
+    assert before - after > 300_000
+
+
+def test_retain_flag_validation():
+    with pytest.raises(SystemExit):
+        parse_args(["--demo", "--retain-raw-days", "3"])
+    cfg = parse_args(["--demo", "--retain-raw-days", "30"])
+    assert cfg.retain_raw_days == 30.0
+    cfg = parse_args(["--demo"])
+    assert cfg.retain_raw_days == 0.0

wallmonitor/__main__.py

@@ -30,17 +30,30 @@
 log = logging.getLogger("wallmonitor")
 
 
-async def _backfill(db: Database) -> None:
-    """One-time, in the background: fill diagnostics columns for rows
-    recorded before those columns existed. Chunked, so live polling only
-    ever waits a moment; a settings flag makes later startups a no-op."""
+async def _maintenance(db: Database, cfg) -> None:
+    """Background housekeeping: the one-time diagnostics backfill, then —
+    when retention is enabled — a daily raw-JSON trim of samples older
+    than the retention window. Chunked throughout, so live polling only
+    ever waits a moment."""
+    import time as _time
+
     def report(done: int, total: int) -> None:
         if done % 200_000 < 10_000 or done >= total:
             log.info("diagnostics backfill: %d / %d rows", done, total)
 
     touched = await asyncio.to_thread(db.backfill_diag_columns, 10_000, report)
     if touched:
         log.info("diagnostics backfill complete: %d rows", touched)
+    if not cfg.retain_raw_days:
+        return
+    while True:
+        cutoff = _time.time() - cfg.retain_raw_days * 86400.0
+        counts = await asyncio.to_thread(db.trim_raw, cutoff)
+        total = sum(counts.values())
+        if total:
+            log.info("retention: trimmed raw JSON from %d rows (%s)", total,
+                     ", ".join(f"{k}={v}" for k, v in counts.items() if v))
+        await asyncio.sleep(24 * 3600.0)
 
 
 async def run(argv: list[str] | None = None) -> None:
@@ -52,6 +65,15 @@ async def run(argv: list[str] | None = None) -> None:
     cfg = parse_args(argv)
     if cfg.discover is not None:
         raise SystemExit(await run_discovery(cfg.discover, split_phase_hint=cfg.split_phase))
+    if cfg.compact:
+        db = Database(cfg.db_path)
+        try:
+            before, after = db.vacuum()
+        finally:
+            db.close()
+        print(f"{cfg.db_path}: {before / 1e6:.1f} MB -> {after / 1e6:.1f} MB "
+              f"({(before - after) / 1e6:.1f} MB reclaimed)")
+        raise SystemExit(0)
     logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s %(message)s")
 
     sim_runner = None
@@ -65,7 +87,7 @@ async def run(argv: list[str] | None = None) -> None:
     client = aiohttp.ClientSession()
     poller = Poller(cfg, db, bus, client)
     await poller.start()
-    backfill = asyncio.create_task(_backfill(db), name="wallmonitor-backfill")
+    backfill = asyncio.create_task(_maintenance(db, cfg), name="wallmonitor-maintenance")
 
     app = make_app(db, bus, poller)
     runner = web.AppRunner(app)

wallmonitor/config.py

@@ -33,6 +33,12 @@ class Config:
     # Human name for this charger, shown in the UI header, tab title and
     # notifications so several instances side by side stay attributable.
     label: str = ""
+    # Optional retention: rows older than this many days keep every extracted
+    # column forever but have their raw JSON blob trimmed to "". 0 = never
+    # trim (the default — full re-interpretability retained indefinitely).
+    retain_raw_days: float = 0.0
+    # One-shot: VACUUM the database to reclaim trimmed space, print sizes, exit.
+    compact: bool = False
     # Sibling instances watching other chargers, as (label, url) pairs;
     # rendered as a switcher in the header so one browser hops between them.
     peers: tuple[tuple[str, str], ...] = ()
@@ -149,6 +155,20 @@ def parse_args(argv: list[str] | None = None) -> Config:
         help="Name for this charger, shown in the UI and notifications — useful when "
         "running one instance per Wall Connector (env: WM_LABEL)",
     )
+    parser.add_argument(
+        "--retain-raw-days",
+        type=float,
+        default=_env("WM_RETAIN_RAW_DAYS", 0.0),
+        help="Trim the raw JSON blob from samples older than this many days (columns are "
+        "kept forever; 0 disables — the default). Minimum 7. (env: WM_RETAIN_RAW_DAYS)",
+    )
+    parser.add_argument(
+        "--compact",
+        action="store_true",
+        default=False,
+        help="VACUUM the database to reclaim space freed by retention, print sizes, and exit. "
+        "Run it while the monitor service is stopped.",
+    )
     parser.add_argument(
         "--peer",
         action="append",
@@ -195,7 +215,10 @@ def parse_args(argv: list[str] | None = None) -> Config:
     except ValueError as ex:
         parser.error(str(ex))
 
-    if not args.demo and not args.host and args.discover is None:
+    if args.retain_raw_days and args.retain_raw_days < 7:
+        parser.error("--retain-raw-days must be at least 7 (or 0 to disable)")
+
+    if not args.demo and not args.host and args.discover is None and not args.compact:
         parser.error(
             "--host (or WM_WC_HOST) is required — run `wallmonitor --discover` to find "
             "your Wall Connector's address, or --demo for the built-in simulator"
@@ -209,6 +232,8 @@ def parse_args(argv: list[str] | None = None) -> Config:
         demo=bool(args.demo),
         split_phase=bool(args.split_phase),
         label=args.label,
+        retain_raw_days=float(args.retain_raw_days),
+        compact=bool(args.compact),
         peers=peers,
         discover=args.discover,
         notify_url=args.notify_url,

wallmonitor/db.py

@@ -221,6 +221,7 @@ def __init__(self, path: str):
         telemetry). The schema is pure CREATE IF NOT EXISTS, re-run on
         every startup: "migrations" are additive statements appended here.
         """
+        self.path = path
         self._conn = sqlite3.connect(path, check_same_thread=False)
         self._conn.row_factory = sqlite3.Row
         self._lock = threading.Lock()
@@ -245,6 +246,73 @@ def _migrate(self) -> None:
                 self._conn.execute(f"ALTER TABLE vitals_samples ADD COLUMN {col} REAL")
         self._conn.commit()
 
+    # Tables whose raw JSON blob is trimmed by the retention policy. The
+    # extracted columns stay forever; forecast_samples and version_info are
+    # exempt (tiny, and their raw payloads carry fields with no column).
+    RAW_TRIM_TABLES = ("vitals_samples", "wifi_samples", "lifetime_samples", "ambient_samples")
+
+    def trim_raw(self, cutoff_ts: float, chunk: int = 10_000) -> dict[str, int]:
+        """Retention: blank the raw JSON on samples older than cutoff_ts.
+
+        Columns are untouched, so charts, fits and the degradation watch see
+        exactly the same history — what is given up is re-interpreting old
+        rows for fields that were never extracted. Vitals are only trimmed
+        once the diagnostics backfill has finished (its json_extract source
+        is the raw blob). A per-table timestamp cursor in settings makes the
+        daily run scan only rows that newly aged past the cutoff, and the
+        chunked updates hold the write lock briefly each. Freed pages are
+        reused by new inserts, so the file stops growing; a one-shot VACUUM
+        (--compact) reclaims the space for the filesystem.
+        """
+        counts: dict[str, int] = {}
+        for table in self.RAW_TRIM_TABLES:
+            if table == "vitals_samples" and self.get_setting(DIAG_BACKFILL_SETTING) != "1":
+                continue
+            cursor_key = f"raw_trim_ts:{table}"
+            start_ts = float(self.get_setting(cursor_key) or 0.0)
+            trimmed = 0
+            while True:
+                with self._lock:
+                    rows = self._conn.execute(
+                        f"SELECT id, ts FROM {table} WHERE ts >= ? AND ts < ? "
+                        "ORDER BY ts LIMIT ?",
+                        (start_ts, cutoff_ts, chunk),
+                    ).fetchall()
+                    if not rows:
+                        break
+                    ids = [row[0] for row in rows]
+                    placeholders = ",".join("?" for _ in ids)
+                    cur = self._execute(
+                        f"UPDATE {table} SET raw = '' WHERE id IN ({placeholders}) AND raw != ''",
+                        tuple(ids),
+                    )
+                    trimmed += cur.rowcount
+                    start_ts = rows[-1][1]
+                    self._execute(
+                        "INSERT INTO settings(key, value) VALUES (?, ?) "
+                        "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
+                        (cursor_key, repr(start_ts)),
+                    )
+                if len(rows) < chunk:
+                    break
+            counts[table] = trimmed
+        return counts
+
+    def vacuum(self) -> tuple[int, int]:
+        """VACUUM, returning (bytes_before, bytes_after). Exclusive — run
+        while nothing else is writing (the --compact command exists so this
+        never happens implicitly under a live poller)."""
+        import os
+
+        with self._lock:
+            self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+        before = os.path.getsize(self.path)
+        with self._lock:
+            self._conn.execute("VACUUM")
+            self._conn.execute("PRAGMA wal_checkpoint(TRUNCATE)")
+        after = os.path.getsize(self.path)
+        return before, after
+
     def backfill_diag_columns(self, chunk: int = 10_000, progress=None) -> int:
         """Fill the diagnostics columns for rows recorded before they
         existed, from each row's raw JSON, in id-ordered chunks so the
@@ -274,7 +342,7 @@ def backfill_diag_columns(self, chunk: int = 10_000, progress=None) -> int:
                            relay_k1_v   = COALESCE(json_extract(raw, '$.relay_k1_v'),
                                                    json_extract(raw, '$.relay_coil_v')),
                            relay_k2_v   = json_extract(raw, '$.relay_k2_v')
-                       WHERE id > ? AND id <= ? AND prox_v IS NULL""",
+                       WHERE id > ? AND id <= ? AND prox_v IS NULL AND raw != ''""",
                     (last, upto),
                 )
                 touched += cur.rowcount
@@ -617,13 +685,18 @@ def vitals_range(self, t_from: float, t_to: float, max_points: int = 1500) -> li
         if self._diag_backfilled:
             d = {col: col for col in VITALS_LATER_COLUMNS}
         else:
+            # CASE-guarded: a retention-trimmed row has raw = '' and
+            # json_extract('') is a hard error, not NULL — and SQLite does
+            # not promise to short-circuit COALESCE past it even when the
+            # column is populated.
+            j = "CASE WHEN raw = '' THEN NULL ELSE json_extract(raw, '{key}') END"
             d = {
-                "pilot_high_v": "COALESCE(pilot_high_v, json_extract(raw, '$.pilot_high_v'))",
-                "pilot_low_v": "COALESCE(pilot_low_v, json_extract(raw, '$.pilot_low_v'))",
-                "prox_v": "COALESCE(prox_v, json_extract(raw, '$.prox_v'))",
-                "relay_k1_v": "COALESCE(relay_k1_v, json_extract(raw, '$.relay_k1_v'), "
-                              "json_extract(raw, '$.relay_coil_v'))",
-                "relay_k2_v": "COALESCE(relay_k2_v, json_extract(raw, '$.relay_k2_v'))",
+                "pilot_high_v": f"COALESCE(pilot_high_v, {j.format(key='$.pilot_high_v')})",
+                "pilot_low_v": f"COALESCE(pilot_low_v, {j.format(key='$.pilot_low_v')})",
+                "prox_v": f"COALESCE(prox_v, {j.format(key='$.prox_v')})",
+                "relay_k1_v": f"COALESCE(relay_k1_v, {j.format(key='$.relay_k1_v')}, "
+                              f"{j.format(key='$.relay_coil_v')})",
+                "relay_k2_v": f"COALESCE(relay_k2_v, {j.format(key='$.relay_k2_v')})",
             }
         diag = ", ".join(f"{expr} AS {col}" for col, expr in d.items())
         if sample_count <= max_points:

wallmonitor/poller.py

@@ -689,6 +689,7 @@ def status(self) -> dict[str, Any]:
             "host": self.cfg.host,
             "label": self.cfg.label,
             "peers": [{"label": label, "url": url} for label, url in self.cfg.peers],
+            "retain_raw_days": self.cfg.retain_raw_days,
             "device_serial": self.device_serial,
             "serial_mismatch": self.serial_mismatch,
             "offline": self._offline,