Fix recurring WAL corruption with RESTART checkpointing

Root cause: SQLite's default PASSIVE checkpoint mode cannot complete
under sustained write load (4 nodes, hundreds of events/min) because
active readers always block it. The WAL grows unbounded and eventually
corrupts the main DB file.

- Add dedicated checkpoint_wal() using PRAGMA wal_checkpoint(RESTART),
  which waits for readers to finish then checkpoints fully and resets
  the WAL back to the start.
- Add background checkpointer task running every 30 minutes, separate
  from the 6-hour prune cycle.
- Remove PASSIVE checkpoint from prune_old_events(), replace with call
  to checkpoint_wal() after pruning completes.
- Set wal_autocheckpoint=500 on every connection (~2MB threshold) so
  SQLite's own auto-checkpoint also triggers more frequently.
- Reduce HOT_DAYS from 30 to 7: at current data rate (~500k beacons/day)
  30-day retention would grow to several GB before first prune runs.
  7 days keeps the DB under ~200MB and queries fast permanently.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
bot
2026-04-08 08:34:52 +03:00
parent 7f702133cc
commit 9fd97e6925
+77 -4
View File
@@ -19,7 +19,7 @@ JS_PATH = os.path.join(os.path.dirname(__file__), "static", "dashboard.js")
OUI_PATH = os.path.join(os.path.dirname(__file__), "oui.txt") OUI_PATH = os.path.join(os.path.dirname(__file__), "oui.txt")
ONLINE_SECS = 30 # node considered online if heartbeat within this many seconds ONLINE_SECS = 30 # node considered online if heartbeat within this many seconds
EVENT_CAP = 100 # max events returned / shown in dashboard EVENT_CAP = 100 # max events returned / shown in dashboard
HOT_DAYS = 30 # dashboard queries cover this many days; Search is always all-time HOT_DAYS = 7 # dashboard queries cover this many days; Search is always all-time
# ─── Deauth reason codes ───────────────────────────────────────────────────── # ─── Deauth reason codes ─────────────────────────────────────────────────────
@@ -105,6 +105,7 @@ def get_conn() -> sqlite3.Connection:
conn = sqlite3.connect(DB_PATH) conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row conn.row_factory = sqlite3.Row
conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA wal_autocheckpoint=500") # auto-checkpoint every 500 pages (~2MB)
return conn return conn
@@ -481,6 +482,7 @@ def _finalize_session(active: dict) -> dict:
else: else:
severity = "normal" severity = "normal"
desc, _ = deauth_reason_info(dominant)
src_vendor, src_rand = oui_lookup(active["src"]) src_vendor, src_rand = oui_lookup(active["src"])
return { return {
"src": active["src"], "src": active["src"],
@@ -497,6 +499,8 @@ def _finalize_session(active: dict) -> dict:
"peak_rssi": max(rssi_vals) if rssi_vals else None, "peak_rssi": max(rssi_vals) if rssi_vals else None,
"avg_rssi": round(sum(rssi_vals) / len(rssi_vals), 1) if rssi_vals else None, "avg_rssi": round(sum(rssi_vals) / len(rssi_vals), 1) if rssi_vals else None,
"dominant_reason": dominant, "dominant_reason": dominant,
"dominant_reason_desc": desc,
"dominant_reason_class": classification,
"severity": severity, "severity": severity,
"events": active["frames"][:50], "events": active["frames"][:50],
} }
@@ -654,8 +658,30 @@ def build_alerts() -> dict:
GROUP BY bssid GROUP BY bssid
""", tuple(bssid_list)) """, tuple(bssid_list))
ssid_map = {r["bssid"]: r["ssid"] for r in ssid_rows} ssid_map = {r["bssid"]: r["ssid"] for r in ssid_rows}
# Dominant reason per impersonated BSSID
reason_rows_t = query(f"""
SELECT bssid, reason, COUNT(*) AS cnt
FROM deauth_events
WHERE received_at > ? AND bssid IN ({placeholders})
GROUP BY bssid, reason
ORDER BY bssid, cnt DESC
""", (cutoff_7d,) + tuple(bssid_list))
dom_reason_by_bssid: dict = {}
for r in reason_rows_t:
if r["bssid"] not in dom_reason_by_bssid:
dom_reason_by_bssid[r["bssid"]] = r["reason"]
for t in top_targets: for t in top_targets:
t["ssid"] = ssid_map.get(t["bssid"]) t["ssid"] = ssid_map.get(t["bssid"])
code = dom_reason_by_bssid.get(t["bssid"])
if code is not None:
desc, cls = deauth_reason_info(code)
t["dominant_reason"] = code
t["dominant_reason_desc"] = desc
t["dominant_reason_class"] = cls
else:
t["dominant_reason"] = None
t["dominant_reason_desc"] = None
t["dominant_reason_class"] = None
# Most targeted devices — last 7 days (grouped by dst, the actual victim) # Most targeted devices — last 7 days (grouped by dst, the actual victim)
top_targeted_devices = query(""" top_targeted_devices = query("""
@@ -672,10 +698,38 @@ def build_alerts() -> dict:
ORDER BY total_frames DESC ORDER BY total_frames DESC
LIMIT 10 LIMIT 10
""", (cutoff_7d,)) """, (cutoff_7d,))
if top_targeted_devices:
dst_list = [d["dst"] for d in top_targeted_devices]
placeholders = ",".join("?" * len(dst_list))
# Dominant reason per targeted dst MAC
reason_rows_d = query(f"""
SELECT dst, reason, COUNT(*) AS cnt
FROM deauth_events
WHERE received_at > ? AND dst IN ({placeholders})
GROUP BY dst, reason
ORDER BY dst, cnt DESC
""", (cutoff_7d,) + tuple(dst_list))
dom_reason_by_dst: dict = {}
for r in reason_rows_d:
if r["dst"] not in dom_reason_by_dst:
dom_reason_by_dst[r["dst"]] = r["reason"]
else:
dom_reason_by_dst = {}
for d in top_targeted_devices: for d in top_targeted_devices:
vendor, randomized = oui_lookup(d["dst"]) vendor, randomized = oui_lookup(d["dst"])
d["vendor"] = vendor d["vendor"] = vendor
d["randomized"] = randomized d["randomized"] = randomized
code = dom_reason_by_dst.get(d["dst"])
if code is not None:
desc, cls = deauth_reason_info(code)
d["dominant_reason"] = code
d["dominant_reason_desc"] = desc
d["dominant_reason_class"] = cls
else:
d["dominant_reason"] = None
d["dominant_reason_desc"] = None
d["dominant_reason_class"] = None
# Reason code breakdown — last 7 days # Reason code breakdown — last 7 days
reason_rows = query(""" reason_rows = query("""
@@ -989,17 +1043,29 @@ def ensure_schema():
PRUNE_INTERVAL_SECS = 6 * 3600 # run pruning every 6 hours PRUNE_INTERVAL_SECS = 6 * 3600 # run pruning every 6 hours
CHECKPOINT_INTERVAL_SECS = 30 * 60 # force WAL checkpoint every 30 minutes
def prune_old_events() -> None: def prune_old_events() -> None:
"""Delete events older than HOT_DAYS from all event tables.""" """Delete events older than HOT_DAYS from all event tables, then checkpoint."""
cutoff = _hot_cutoff() cutoff = _hot_cutoff()
tables = ["beacon_events", "probe_events", "deauth_events", "assoc_events"] tables = ["beacon_events", "probe_events", "deauth_events", "assoc_events"]
with get_conn() as conn: with get_conn() as conn:
for table in tables: for table in tables:
conn.execute(f"DELETE FROM {table} WHERE received_at < ?", (cutoff,)) conn.execute(f"DELETE FROM {table} WHERE received_at < ?", (cutoff,))
conn.execute("PRAGMA wal_checkpoint(PASSIVE)")
conn.commit() conn.commit()
checkpoint_wal()
def checkpoint_wal() -> None:
"""
Force a RESTART checkpoint — writes all WAL frames to the main DB file
and resets the WAL back to the beginning. Prevents the WAL from growing
unbounded under sustained write load, which causes DB corruption over time.
RESTART blocks until all current readers finish, then checkpoints fully.
"""
with get_conn() as conn:
conn.execute("PRAGMA wal_checkpoint(RESTART)")
app = FastAPI(title="ESP32 Recon Dashboard") app = FastAPI(title="ESP32 Recon Dashboard")
@@ -1007,12 +1073,19 @@ ensure_schema()
@app.on_event("startup") @app.on_event("startup")
async def start_pruning_task(): async def start_background_tasks():
async def _pruner(): async def _pruner():
while True: while True:
await asyncio.sleep(PRUNE_INTERVAL_SECS) await asyncio.sleep(PRUNE_INTERVAL_SECS)
await asyncio.to_thread(prune_old_events) await asyncio.to_thread(prune_old_events)
async def _checkpointer():
while True:
await asyncio.sleep(CHECKPOINT_INTERVAL_SECS)
await asyncio.to_thread(checkpoint_wal)
asyncio.create_task(_pruner()) asyncio.create_task(_pruner())
asyncio.create_task(_checkpointer())
@app.get("/", response_class=HTMLResponse) @app.get("/", response_class=HTMLResponse)