From eb0cbb331ca67f708882736d7207a005ce0c2f7d Mon Sep 17 00:00:00 2001 From: roses Date: Thu, 10 Sep 2026 06:35:43 +0200 Subject: [PATCH 1/2] Traceroutes are now saved to the database the moment they arrive instead of being re-read from raw packets every time a page needs them. Two new tables keep track of each route that has ever been seen and the hops between the two endpoints. This prepares for faster queries in PR2 --- pyproject.toml | 1 + src/malla/backfill_traceroutes.py | 146 ++++++ src/malla/database/schema.py | 5 + src/malla/database/traceroute_schema.py | 109 ++++ src/malla/database/traceroutes.py | 198 ++++++++ src/malla/mqtt_capture.py | 23 +- src/malla/utils/traceroute_utils.py | 25 +- tests/unit/test_traceroute_materialization.py | 471 ++++++++++++++++++ 8 files changed, 962 insertions(+), 16 deletions(-) create mode 100644 src/malla/backfill_traceroutes.py create mode 100644 src/malla/database/traceroute_schema.py create mode 100644 src/malla/database/traceroutes.py create mode 100644 tests/unit/test_traceroute_materialization.py diff --git a/pyproject.toml b/pyproject.toml index dc5b8b78..63ab355d 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -63,6 +63,7 @@ Issues = "https://github.com/zenitraM/malla/issues" malla-web = "malla.web_ui:main" malla-web-gunicorn = "malla.wsgi:main" malla-capture = "malla.mqtt_capture:main" +malla-backfill-traceroutes = "malla.backfill_traceroutes:main" [project.optional-dependencies] dev = [ diff --git a/src/malla/backfill_traceroutes.py b/src/malla/backfill_traceroutes.py new file mode 100644 index 00000000..975611da --- /dev/null +++ b/src/malla/backfill_traceroutes.py @@ -0,0 +1,146 @@ +"""Explicit, resumable preparation of stored traceroute packets.""" + +import argparse +import json +import sqlite3 +import sys +from collections.abc import Callable +from contextlib import closing +from pathlib import Path + +from .database.traceroute_schema import TRACEROUTE_PREDICATE, ensure_traceroute_schema +from .database.traceroutes import ( + PARSER_VERSION, + inspect_traceroutes, + write_traceroute, +) + + +def prepare_traceroutes( + conn: sqlite3.Connection, + batch_size: int = 1000, + progress: Callable[[int], None] | None = None, +) -> dict: + """Fill missing/pending/older-version records, then validate under a write lock. + + Each batch commits independently. An ID boundary keeps continuous capture + from extending the run indefinitely; captures beyond it use the same writer. + Final validation includes any raw-only imports that arrived during the run. + """ + if batch_size < 1: + raise ValueError("batch_size must be positive") + conn.row_factory = sqlite3.Row + with conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.cursor() + ensure_traceroute_schema(cursor) + # Building this raw-history index is explicit maintenance work, never + # part of web/capture startup. It also speeds subsequent resumptions. + cursor.execute( + "CREATE INDEX IF NOT EXISTS idx_packet_history_traceroute_id " + f"ON packet_history(id) WHERE {TRACEROUTE_PREDICATE}" + ) + if cursor.execute( + "SELECT 1 FROM traceroute_routes WHERE parser_version > ? LIMIT 1", + (PARSER_VERSION,), + ).fetchone(): + raise ValueError("Database uses a newer traceroute decoder; upgrade Malla") + upper_id = cursor.execute( + "SELECT COALESCE(MAX(id), 0) FROM packet_history" + ).fetchone()[0] + + last_id = None + processed = 0 + while True: + with conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.cursor() + lower_bound = "" if last_id is None else "AND id > ?" + params = [upper_id] + if last_id is not None: + params.append(last_id) + rows = cursor.execute( + f""" + SELECT id, timestamp, mesh_packet_id, from_node_id, to_node_id, + hop_start, hop_limit, raw_payload + FROM packet_history + WHERE {TRACEROUTE_PREDICATE} AND id <= ? {lower_bound} + AND NOT EXISTS ( + SELECT 1 FROM traceroute_routes + WHERE packet_id = packet_history.id AND parser_version = ? + ) + ORDER BY id LIMIT ? + """, + [*params, PARSER_VERSION, batch_size], + ).fetchall() + if not rows: + break + for row in rows: + write_traceroute(cursor, dict(row)) + last_id = rows[-1]["id"] + processed += len(rows) + if progress is not None: + progress(processed) + + with conn: + conn.execute("BEGIN IMMEDIATE") + cursor = conn.cursor() + result = inspect_traceroutes(cursor) + return {**result, "processed_this_run": processed} + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "--database", required=True, type=Path, help="Existing database path" + ) + parser.add_argument( + "--batch-size", type=int, default=1000, help="Packets per transaction" + ) + parser.add_argument( + "--check", + "--dry-run", + action="store_true", + help="Report counts without modifying the database", + ) + args = parser.parse_args(argv) + if args.batch_size < 1: + parser.error("--batch-size must be positive") + path = args.database.expanduser().resolve() + print(f"Database: {path}", flush=True) + try: + # mode=rw refuses to create a database for a misspelled path. Neither + # configuration defaults nor capture startup participate in this command. + mode = "ro" if args.check else "rw" + with closing( + sqlite3.connect(f"{path.as_uri()}?mode={mode}", uri=True, timeout=30.0) + ) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys=ON") + if args.check: + with conn: + conn.execute("BEGIN") + result = inspect_traceroutes(conn.cursor()) + else: + result = prepare_traceroutes( + conn, + batch_size=args.batch_size, + progress=lambda count: print( + f"Prepared {count} traceroutes", flush=True + ), + ) + print(json.dumps(result, indent=2), flush=True) + return 0 if result["complete"] else 1 + except KeyboardInterrupt: + print( + "Interrupted. Committed batches are safe; run the same command to resume.", + file=sys.stderr, + ) + return 130 + except (OSError, sqlite3.Error, ValueError) as exc: + print(f"Traceroute preparation failed: {exc}", file=sys.stderr) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/src/malla/database/schema.py b/src/malla/database/schema.py index c84fbfcd..618276c5 100644 --- a/src/malla/database/schema.py +++ b/src/malla/database/schema.py @@ -3,6 +3,8 @@ import logging import sqlite3 +from .traceroute_schema import ensure_traceroute_schema + logger = logging.getLogger(__name__) @@ -208,6 +210,9 @@ def ensure_startup_schema( cursor.execute(ACTIVITY_ROLLUP_TABLE_SQL) + if "packet_history" in existing_tables: + ensure_traceroute_schema(cursor) + if "node_info" in existing_tables: cursor.execute("PRAGMA table_info(node_info)") node_info_columns = {row[1] for row in cursor.fetchall()} diff --git a/src/malla/database/traceroute_schema.py b/src/malla/database/traceroute_schema.py new file mode 100644 index 00000000..c26dea50 --- /dev/null +++ b/src/malla/database/traceroute_schema.py @@ -0,0 +1,109 @@ +"""Rebuildable traceroute tables and pending records for raw-only writers.""" + +import sqlite3 + +TRACEROUTE_PORT = 70 +TRACEROUTE_PREDICATE = "(portnum = 70 OR portnum_name = 'TRACEROUTE_APP')" + + +def ensure_traceroute_schema(cursor: sqlite3.Cursor) -> None: + """Create empty derived tables; historical decoding is always explicit. + + Triggers leave raw-only inserts/updates pending, including imports and older + capture versions. The shared writer replaces the pending record in the same + transaction. These triggers never decode payloads or examine packet history. + """ + cursor.execute(""" + CREATE TABLE IF NOT EXISTS traceroute_routes ( + packet_id INTEGER PRIMARY KEY REFERENCES packet_history(id) ON DELETE CASCADE, + timestamp REAL NOT NULL, + mesh_packet_id INTEGER, + from_node_id INTEGER, + to_node_id INTEGER, + route_nodes_json TEXT NOT NULL DEFAULT '[]', + snr_towards_json TEXT NOT NULL DEFAULT '[]', + route_back_json TEXT NOT NULL DEFAULT '[]', + snr_back_json TEXT NOT NULL DEFAULT '[]', + forward_complete INTEGER NOT NULL DEFAULT 0 CHECK (forward_complete IN (0, 1)), + return_complete INTEGER NOT NULL DEFAULT 0 CHECK (return_complete IN (0, 1)), + parse_status TEXT NOT NULL DEFAULT 'pending' + CHECK (parse_status IN ('pending', 'parsed', 'valid_empty', 'invalid_payload')), + parse_error TEXT, + parser_version INTEGER NOT NULL DEFAULT 0, + materialized_at REAL, + CHECK ((parse_status = 'pending' AND parser_version = 0) + OR (parse_status != 'pending' AND parser_version > 0)), + CHECK ((parse_status = 'invalid_payload' AND parse_error IS NOT NULL) + OR (parse_status != 'invalid_payload' AND parse_error IS NULL)) + ) + """) + cursor.execute(""" + CREATE TABLE IF NOT EXISTS traceroute_hops ( + packet_id INTEGER NOT NULL REFERENCES traceroute_routes(packet_id) ON DELETE CASCADE, + direction TEXT NOT NULL CHECK (direction IN ('forward', 'return')), + hop_index INTEGER NOT NULL CHECK (hop_index >= 0), + timestamp REAL NOT NULL, + from_node_id INTEGER NOT NULL, + to_node_id INTEGER NOT NULL, + snr REAL, + PRIMARY KEY (packet_id, direction, hop_index) + ) + """) + + for name, table, columns in ( + ("routes_time", "traceroute_routes", "timestamp"), + ("routes_source_time", "traceroute_routes", "from_node_id, timestamp"), + ("routes_target_time", "traceroute_routes", "to_node_id, timestamp"), + ( + "routes_group_time", + "traceroute_routes", + "mesh_packet_id, from_node_id, to_node_id, timestamp", + ), + ("routes_version", "traceroute_routes", "parser_version"), + ("hops_link_time", "traceroute_hops", "from_node_id, to_node_id, timestamp"), + ("hops_time_link", "traceroute_hops", "timestamp, from_node_id, to_node_id"), + ("hops_source_time", "traceroute_hops", "from_node_id, timestamp"), + ("hops_target_time", "traceroute_hops", "to_node_id, timestamp"), + ): + cursor.execute( + f"CREATE INDEX IF NOT EXISTS idx_traceroute_{name} ON {table} ({columns})" + ) + + # Explicit child deletion also handles raw import connections with foreign + # keys disabled, and INSERT OR REPLACE reusing a packet row ID. + for event, operation in ( + ("insert", "INSERT"), + ( + "update", + "UPDATE OF id, timestamp, mesh_packet_id, from_node_id, to_node_id, " + "portnum, portnum_name, raw_payload, hop_start, hop_limit", + ), + ): + old_cleanup = ( + "DELETE FROM traceroute_hops WHERE packet_id = OLD.id; " + "DELETE FROM traceroute_routes WHERE packet_id = OLD.id;" + if event == "update" + else "" + ) + cursor.execute(f""" + CREATE TRIGGER IF NOT EXISTS traceroute_packet_{event} + AFTER {operation} ON packet_history + BEGIN + {old_cleanup} + DELETE FROM traceroute_hops WHERE packet_id = NEW.id; + DELETE FROM traceroute_routes WHERE packet_id = NEW.id; + INSERT INTO traceroute_routes + (packet_id, timestamp, mesh_packet_id, from_node_id, to_node_id) + SELECT NEW.id, NEW.timestamp, NEW.mesh_packet_id, + NEW.from_node_id, NEW.to_node_id + WHERE NEW.portnum = {TRACEROUTE_PORT} OR NEW.portnum_name = 'TRACEROUTE_APP'; + END + """) + cursor.execute(""" + CREATE TRIGGER IF NOT EXISTS traceroute_packet_delete + AFTER DELETE ON packet_history + BEGIN + DELETE FROM traceroute_hops WHERE packet_id = OLD.id; + DELETE FROM traceroute_routes WHERE packet_id = OLD.id; + END + """) diff --git a/src/malla/database/traceroutes.py b/src/malla/database/traceroutes.py new file mode 100644 index 00000000..696114f8 --- /dev/null +++ b/src/malla/database/traceroutes.py @@ -0,0 +1,198 @@ +"""Shared capture/backfill decoder, writer, and traceroute readiness checks.""" + +import json +import sqlite3 +import time +from dataclasses import dataclass +from typing import Any + +from google.protobuf.message import DecodeError + +from ..models.traceroute import TracerouteHop, TraceroutePacket +from ..utils.traceroute_utils import RouteData, decode_traceroute_payload +from .traceroute_schema import TRACEROUTE_PREDICATE + +PARSER_VERSION = 1 + + +@dataclass(frozen=True) +class DecodedTraceroute: + route: RouteData + hops: tuple[TracerouteHop, ...] = () + forward_complete: bool = False + return_complete: bool = False + parse_error: str | None = None + + @property + def parse_status(self) -> str: + if self.parse_error is not None: + return "invalid_payload" + return "parsed" if any(self.route.values()) else "valid_empty" + + +def decode_traceroute(packet: dict[str, Any]) -> DecodedTraceroute: + """Decode once, without database/name/location lookups or discarded hops.""" + try: + if packet.get("raw_payload") is None: + raise TypeError("Traceroute payload is missing") + route = decode_traceroute_payload(packet["raw_payload"]) + except (DecodeError, TypeError) as exc: + return DecodedTraceroute( + route=RouteData(route_nodes=[], snr_towards=[], route_back=[], snr_back=[]), + parse_error=str(exc), + ) + + traceroute = TraceroutePacket( + packet, resolve_names=False, pre_parsed_route_data=route + ) + return DecodedTraceroute( + route=route, + hops=tuple(traceroute.get_rf_hops()), + forward_complete=traceroute.is_complete(), + return_complete=traceroute.is_return_complete(), + ) + + +def write_traceroute(cursor: sqlite3.Cursor, packet: dict[str, Any]) -> None: + """Replace one reception's decoded data inside the caller's transaction. + + The caller must already have inserted the raw packet. This function neither + commits nor swallows storage errors, so raw and derived data stay atomic. + """ + if not cursor.connection.in_transaction: + raise ValueError("Traceroute writes require an active transaction") + decoded = decode_traceroute(packet) + cursor.execute( + """ + INSERT INTO traceroute_routes ( + packet_id, timestamp, mesh_packet_id, from_node_id, to_node_id, + route_nodes_json, snr_towards_json, route_back_json, snr_back_json, + forward_complete, return_complete, parse_status, parse_error, + parser_version, materialized_at + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ON CONFLICT(packet_id) DO UPDATE SET + timestamp = excluded.timestamp, + mesh_packet_id = excluded.mesh_packet_id, + from_node_id = excluded.from_node_id, + to_node_id = excluded.to_node_id, + route_nodes_json = excluded.route_nodes_json, + snr_towards_json = excluded.snr_towards_json, + route_back_json = excluded.route_back_json, + snr_back_json = excluded.snr_back_json, + forward_complete = excluded.forward_complete, + return_complete = excluded.return_complete, + parse_status = excluded.parse_status, + parse_error = excluded.parse_error, + parser_version = excluded.parser_version, + materialized_at = excluded.materialized_at + """, + ( + packet["id"], + packet["timestamp"], + packet.get("mesh_packet_id"), + packet.get("from_node_id"), + packet.get("to_node_id"), + json.dumps(decoded.route["route_nodes"]), + json.dumps(decoded.route["snr_towards"]), + json.dumps(decoded.route["route_back"]), + json.dumps(decoded.route["snr_back"]), + decoded.forward_complete, + decoded.return_complete, + decoded.parse_status, + decoded.parse_error, + PARSER_VERSION, + time.time(), + ), + ) + cursor.execute("DELETE FROM traceroute_hops WHERE packet_id = ?", (packet["id"],)) + indices = {"forward_rf": 0, "return_rf": 0} + for hop in decoded.hops: + cursor.execute( + """ + INSERT INTO traceroute_hops + (packet_id, direction, hop_index, timestamp, from_node_id, to_node_id, snr) + VALUES (?, ?, ?, ?, ?, ?, ?) + """, + ( + packet["id"], + "forward" if hop.direction == "forward_rf" else "return", + indices[hop.direction], + packet["timestamp"], + hop.from_node_id, + hop.to_node_id, + hop.snr, + ), + ) + indices[hop.direction] += 1 + + +def inspect_traceroutes(cursor: sqlite3.Cursor) -> dict[str, Any]: + """Audit stored history explicitly, including databases not yet prepared. + + This is an offline validation query, not a web request helper. The caller + holds a transaction for a consistent snapshot. + """ + tables = { + row[0] + for row in cursor.execute("SELECT name FROM sqlite_master WHERE type = 'table'") + } + raw = cursor.execute( + f"SELECT COUNT(*) FROM packet_history WHERE {TRACEROUTE_PREDICATE}" + ).fetchone()[0] + counts: dict[str, Any] = { + "raw_traceroutes": raw, + "routes": 0, + "parsed": 0, + "valid_empty": 0, + "invalid_payload": 0, + "pending": 0, + "outdated": 0, + "missing": raw, + "hops": 0, + "orphan_routes": 0, + "orphan_hops": 0, + "parser_version": PARSER_VERSION, + "complete": False, + } + if not {"traceroute_routes", "traceroute_hops"} <= tables: + return counts + + for status, version, count in cursor.execute( + "SELECT parse_status, parser_version, COUNT(*) FROM traceroute_routes " + "GROUP BY parse_status, parser_version" + ): + counts["routes"] += count + counts[status] += count + if version not in (0, PARSER_VERSION): + counts["outdated"] += count + counts["missing"] = cursor.execute( + f""" + SELECT COUNT(*) FROM packet_history + WHERE {TRACEROUTE_PREDICATE} AND NOT EXISTS ( + SELECT 1 FROM traceroute_routes WHERE packet_id = packet_history.id + ) + """ + ).fetchone()[0] + counts["hops"] = cursor.execute("SELECT COUNT(*) FROM traceroute_hops").fetchone()[ + 0 + ] + counts["orphan_routes"] = cursor.execute( + f""" + SELECT COUNT(*) FROM traceroute_routes WHERE NOT EXISTS ( + SELECT 1 FROM packet_history + WHERE id = traceroute_routes.packet_id AND {TRACEROUTE_PREDICATE} + ) + """ + ).fetchone()[0] + counts["orphan_hops"] = cursor.execute( + """ + SELECT COUNT(*) FROM traceroute_hops WHERE NOT EXISTS ( + SELECT 1 FROM traceroute_routes WHERE packet_id = traceroute_hops.packet_id + ) + """ + ).fetchone()[0] + counts["complete"] = not any( + counts[key] + for key in ("missing", "pending", "outdated", "orphan_routes", "orphan_hops") + ) + return counts diff --git a/src/malla/mqtt_capture.py b/src/malla/mqtt_capture.py index a9831877..e8003020 100644 --- a/src/malla/mqtt_capture.py +++ b/src/malla/mqtt_capture.py @@ -38,6 +38,7 @@ import sqlite3 import threading import time +from contextlib import closing from typing import Any import paho.mqtt.client as mqtt @@ -59,6 +60,7 @@ from .database.connection import seed_query_planner_stats_async from .database.schema import ensure_startup_schema +from .database.traceroutes import write_traceroute # Load the singleton configuration once at module import time. This ensures the # capture tool honours the same YAML + optional environment override mechanism @@ -846,10 +848,10 @@ def log_packet_to_database( relay_node = getattr(mesh_packet, "relay_node", None) if mesh_packet else None tx_after = getattr(mesh_packet, "tx_after", None) if mesh_packet else None - with db_lock: - conn = sqlite3.connect(DATABASE_FILE, timeout=30.0) + with db_lock, closing(sqlite3.connect(DATABASE_FILE, timeout=30.0)) as conn, conn: conn.row_factory = sqlite3.Row cursor = conn.cursor() + cursor.execute("PRAGMA foreign_keys=ON") cursor.execute( """ @@ -894,8 +896,20 @@ def log_packet_to_database( ), ) - conn.commit() - conn.close() + if portnum == portnums_pb2.PortNum.TRACEROUTE_APP: + write_traceroute( + cursor, + { + "id": cursor.lastrowid, + "timestamp": current_time, + "mesh_packet_id": mesh_packet_id, + "from_node_id": from_node_id, + "to_node_id": to_node_id, + "hop_start": hop_start, + "hop_limit": hop_limit, + "raw_payload": raw_payload, + }, + ) def get_packet_history( @@ -945,6 +959,7 @@ def cleanup_old_data() -> None: conn = sqlite3.connect(DATABASE_FILE, timeout=30.0) conn.row_factory = sqlite3.Row cursor = conn.cursor() + cursor.execute("PRAGMA foreign_keys=ON") try: # Delete old packet history records diff --git a/src/malla/utils/traceroute_utils.py b/src/malla/utils/traceroute_utils.py index adf54d62..7b94b435 100644 --- a/src/malla/utils/traceroute_utils.py +++ b/src/malla/utils/traceroute_utils.py @@ -19,6 +19,18 @@ class RouteData(TypedDict): snr_back: list[float] +def decode_traceroute_payload(raw_payload: bytes) -> RouteData: + """Decode RouteDiscovery, raising on malformed payloads instead of hiding them.""" + route_discovery = mesh_pb2.RouteDiscovery() + route_discovery.ParseFromString(raw_payload) + return RouteData( + route_nodes=list(route_discovery.route), + snr_towards=[snr / 4.0 for snr in route_discovery.snr_towards], + route_back=list(route_discovery.route_back), + snr_back=[snr / 4.0 for snr in route_discovery.snr_back], + ) + + def parse_traceroute_payload(raw_payload: bytes) -> RouteData: """ Parse traceroute payload from raw bytes using protobuf parsing. @@ -44,18 +56,7 @@ def parse_traceroute_payload(raw_payload: bytes) -> RouteData: return RouteData(route_nodes=[], snr_towards=[], route_back=[], snr_back=[]) try: - # Try protobuf parsing - route_discovery = mesh_pb2.RouteDiscovery() - route_discovery.ParseFromString(raw_payload) - - result = RouteData( - route_nodes=[int(node_id) for node_id in route_discovery.route], - # Convert SNR from scaled integer to actual dB (divide by 4) - snr_towards=[float(snr) / 4.0 for snr in route_discovery.snr_towards], - route_back=[int(node_id) for node_id in route_discovery.route_back], - # Convert SNR from scaled integer to actual dB (divide by 4) - snr_back=[float(snr) / 4.0 for snr in route_discovery.snr_back], - ) + result = decode_traceroute_payload(raw_payload) logger.debug( f"Protobuf parsing successful: {len(result['route_nodes'])} nodes, " diff --git a/tests/unit/test_traceroute_materialization.py b/tests/unit/test_traceroute_materialization.py new file mode 100644 index 00000000..21ff64f9 --- /dev/null +++ b/tests/unit/test_traceroute_materialization.py @@ -0,0 +1,471 @@ +"""Behavioral checks for capture, raw-only imports, and resumable preparation.""" + +import json +import sqlite3 +from contextlib import closing +from unittest.mock import patch + +import pytest +from meshtastic import mesh_pb2, mqtt_pb2, portnums_pb2 + +from malla import mqtt_capture +from malla.backfill_traceroutes import main, prepare_traceroutes +from malla.database.schema import ensure_startup_schema +from malla.database.traceroute_schema import ensure_traceroute_schema +from malla.database.traceroutes import ( + PARSER_VERSION, + decode_traceroute, + inspect_traceroutes, + write_traceroute, +) +from malla.models.traceroute import TraceroutePacket + +pytestmark = pytest.mark.unit + + +def payload(route=(), snr=(), back=(), snr_back=()): + return mesh_pb2.RouteDiscovery( + route=route, snr_towards=snr, route_back=back, snr_back=snr_back + ).SerializeToString() + + +@pytest.fixture +def database(tmp_path, monkeypatch): + path = tmp_path / "history.db" + monkeypatch.setattr(mqtt_capture, "DATABASE_FILE", str(path)) + monkeypatch.setattr( + mqtt_capture, "seed_query_planner_stats_async", lambda *_: False + ) + mqtt_capture.init_database() + with closing(sqlite3.connect(path)) as conn: + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA foreign_keys=ON") + yield path, conn + + +def insert_raw(conn, raw, **overrides): + fields = { + "timestamp": 1000.0, + "topic": "msh/test/e/LongFast/!12345678", + "from_node_id": 100, + "to_node_id": 200, + "portnum": 70, + "portnum_name": "TRACEROUTE_APP", + "mesh_packet_id": 42, + "gateway_id": "!12345678", + "hop_start": 5, + "hop_limit": 3, + "raw_payload": raw, + **overrides, + } + cursor = conn.execute( + f"INSERT INTO packet_history ({', '.join(fields)}) " + f"VALUES ({', '.join('?' for _ in fields)})", + list(fields.values()), + ) + return dict( + conn.execute( + "SELECT * FROM packet_history WHERE id = ?", (cursor.lastrowid,) + ).fetchone() + ) + + +def capture(raw, gateway="!12345678"): + packet = mesh_pb2.MeshPacket(id=42, to=200, hop_start=5, hop_limit=3) + setattr(packet, "from", 100) + packet.decoded.portnum = portnums_pb2.PortNum.TRACEROUTE_APP + packet.decoded.payload = raw + envelope = mqtt_pb2.ServiceEnvelope(gateway_id=gateway, channel_id="LongFast") + mqtt_capture.log_packet_to_database("msh/test/e/LongFast", envelope, packet) + + +@pytest.mark.parametrize( + "raw", + [ + payload(snr=[-16]), + payload(route=[110], snr=[-20]), + payload(route=[110], snr=[-20, -32], back=[120], snr_back=[4, -12]), + payload(route=[110, 100, 110], snr=[4, 8, 12, 16]), + payload(route=[110, 120]), + payload(snr=[-128]), + b"", + ], +) +def test_decoder_preserves_existing_path_rules_without_lookups(raw): + packet = { + "from_node_id": 100, + "to_node_id": 200, + "hop_start": 5, + "hop_limit": 3, + "raw_payload": raw, + } + expected = TraceroutePacket(packet, resolve_names=False) + with patch( + "malla.models.traceroute.TraceroutePacket._resolve_node_names", + side_effect=AssertionError("no lookup"), + ): + decoded = decode_traceroute(packet) + assert decoded.route == expected.route_data + assert decoded.hops == tuple(expected.get_rf_hops()) + assert decoded.forward_complete == expected.is_complete() + assert decoded.return_complete == expected.is_return_complete() + + +def test_capture_preserves_repeated_hops_and_gateway_receptions(database): + _, conn = database + raw = payload(route=[110, 100, 110], snr=[4, 8, 12, 16]) + capture(raw) + capture(raw, gateway="!87654321") + routes = conn.execute( + "SELECT * FROM traceroute_routes ORDER BY packet_id" + ).fetchall() + assert len(routes) == 2 + assert routes[0]["mesh_packet_id"] == routes[1]["mesh_packet_id"] == 42 + for route in routes: + assert route["parse_status"] == "parsed" + assert json.loads(route["route_nodes_json"]) == [110, 100, 110] + hops = conn.execute( + "SELECT hop_index, from_node_id, to_node_id, snr FROM traceroute_hops " + "WHERE packet_id = ? ORDER BY hop_index", + (route["packet_id"],), + ).fetchall() + assert [tuple(hop) for hop in hops] == [ + (0, 200, 110, 1.0), + (1, 110, 100, 2.0), + (2, 100, 110, 3.0), + (3, 110, 100, 4.0), + ] + + +def test_capture_and_backfill_are_identical(database): + _, conn = database + raw = payload(route=[110], snr=[-20, -32], back=[120], snr_back=[4, -128]) + capture(raw) + before_route = dict(conn.execute("SELECT * FROM traceroute_routes").fetchone()) + before_hops = [ + tuple(row) + for row in conn.execute( + "SELECT * FROM traceroute_hops ORDER BY direction, hop_index" + ) + ] + with conn: + conn.execute("DELETE FROM traceroute_routes") + result = prepare_traceroutes(conn) + after_route = dict(conn.execute("SELECT * FROM traceroute_routes").fetchone()) + before_route.pop("materialized_at") + after_route.pop("materialized_at") + assert before_route == after_route + assert before_hops == [ + tuple(row) + for row in conn.execute( + "SELECT * FROM traceroute_hops ORDER BY direction, hop_index" + ) + ] + assert result["complete"] + assert result["processed_this_run"] == 1 + + +@pytest.mark.parametrize( + "raw,status", [(b"\xff", "invalid_payload"), (b"", "valid_empty")] +) +def test_capture_keeps_malformed_and_empty_packets_distinct(database, raw, status): + _, conn = database + capture(raw) + assert conn.execute("SELECT raw_payload FROM packet_history").fetchone()[0] == raw + route = conn.execute("SELECT * FROM traceroute_routes").fetchone() + assert route["parse_status"] == status + assert bool(route["parse_error"]) == (status == "invalid_payload") + assert conn.execute("SELECT COUNT(*) FROM traceroute_hops").fetchone()[0] == 0 + assert prepare_traceroutes(conn)["processed_this_run"] == 0 + + +def test_capture_storage_failure_rolls_back_raw_and_all_derived_rows(database): + _, conn = database + with conn: + conn.execute(""" + CREATE TRIGGER reject_second_hop BEFORE INSERT ON traceroute_hops + WHEN NEW.hop_index = 1 BEGIN SELECT RAISE(ABORT, 'test storage failure'); END + """) + with pytest.raises(sqlite3.IntegrityError, match="test storage failure"): + capture(payload(route=[110], snr=[4, 8])) + for table in ("packet_history", "traceroute_routes", "traceroute_hops"): + assert conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0 + # The failed capture connection was closed and its lock released. + capture(payload(snr=[4])) + assert conn.execute("SELECT COUNT(*) FROM traceroute_routes").fetchone()[0] == 1 + + +def test_pending_imports_and_decoder_versions_affect_completeness(database): + _, conn = database + assert inspect_traceroutes(conn.cursor())["complete"] + with conn: + packet = insert_raw(conn, payload(snr=[4])) + assert not inspect_traceroutes(conn.cursor())["complete"] + assert prepare_traceroutes(conn)["complete"] + assert inspect_traceroutes(conn.cursor())["complete"] + with conn: + packet = insert_raw(conn, payload(snr=[4])) + assert not inspect_traceroutes(conn.cursor())["complete"] + with conn: + conn.execute("BEGIN") + write_traceroute(conn.cursor(), packet) + assert inspect_traceroutes(conn.cursor())["complete"] + with patch("malla.database.traceroutes.PARSER_VERSION", PARSER_VERSION + 1): + assert not inspect_traceroutes(conn.cursor())["complete"] + capture(payload(snr=[8])) + assert inspect_traceroutes(conn.cursor())["complete"] + + +def test_raw_only_updates_and_replacements_remove_stale_hops(database): + _, conn = database + capture(payload(route=[110], snr=[4, 8])) + assert prepare_traceroutes(conn)["complete"] + with conn: + conn.execute( + "UPDATE packet_history SET raw_payload = ?", (payload(route=[120]),) + ) + assert not inspect_traceroutes(conn.cursor())["complete"] + assert conn.execute("SELECT COUNT(*) FROM traceroute_hops").fetchone()[0] == 0 + assert prepare_traceroutes(conn)["complete"] + assert json.loads( + conn.execute("SELECT route_nodes_json FROM traceroute_routes").fetchone()[0] + ) == [120] + with conn: + conn.execute( + "INSERT OR REPLACE INTO packet_history (id, timestamp, topic, portnum) VALUES (1, 1000, 'test', 1)" + ) + assert conn.execute("SELECT COUNT(*) FROM traceroute_routes").fetchone()[0] == 0 + + +@pytest.mark.parametrize("foreign_keys", [True, False]) +def test_raw_deletion_cleans_derived_rows_even_for_raw_only_importers( + database, foreign_keys +): + _, conn = database + capture(payload(snr=[4])) + conn.execute(f"PRAGMA foreign_keys={'ON' if foreign_keys else 'OFF'}") + with conn: + conn.execute("DELETE FROM packet_history") + assert conn.execute("SELECT COUNT(*) FROM traceroute_routes").fetchone()[0] == 0 + assert conn.execute("SELECT COUNT(*) FROM traceroute_hops").fetchone()[0] == 0 + + +def test_retention_deletes_route_and_hops(database, monkeypatch): + _, conn = database + with patch("malla.mqtt_capture.time.time", return_value=1000.0): + capture(payload(snr=[4])) + monkeypatch.setattr(mqtt_capture, "DATA_RETENTION_HOURS", 1) + mqtt_capture.cleanup_old_data() + for table in ("packet_history", "traceroute_routes", "traceroute_hops"): + assert conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0] == 0 + + +def test_backfill_batches_resume_without_retrying_invalid_packets(database): + _, conn = database + with conn: + for raw in ( + payload(snr=[4]), + b"\xff", + payload(route=[110]), + b"", + payload(snr=[8]), + ): + insert_raw(conn, raw) + insert_raw(conn, b"ordinary text", portnum=1, portnum_name="TEXT_MESSAGE_APP") + + def interrupt(_): + raise KeyboardInterrupt + + with pytest.raises(KeyboardInterrupt): + prepare_traceroutes(conn, batch_size=2, progress=interrupt) + assert ( + conn.execute( + "SELECT COUNT(*) FROM traceroute_routes WHERE parser_version = ?", + (PARSER_VERSION,), + ).fetchone()[0] + == 2 + ) + assert not inspect_traceroutes(conn.cursor())["complete"] + result = prepare_traceroutes(conn, batch_size=2) + assert result["raw_traceroutes"] == result["routes"] == 5 + assert result["invalid_payload"] == 1 + assert result["processed_this_run"] == 3 + assert result["complete"] + rows = [ + tuple(row) + for row in conn.execute("SELECT * FROM traceroute_routes ORDER BY packet_id") + ] + assert prepare_traceroutes(conn)["processed_this_run"] == 0 + assert rows == [ + tuple(row) + for row in conn.execute("SELECT * FROM traceroute_routes ORDER BY packet_id") + ] + + +def test_backfill_failed_batch_is_atomic_and_can_resume(database): + _, conn = database + with conn: + for _ in range(3): + insert_raw(conn, payload(snr=[4])) + conn.execute(""" + CREATE TRIGGER reject_packet BEFORE INSERT ON traceroute_hops + WHEN NEW.packet_id = 2 BEGIN SELECT RAISE(ABORT, 'batch failure'); END + """) + with pytest.raises(sqlite3.IntegrityError, match="batch failure"): + prepare_traceroutes(conn, batch_size=3) + assert ( + conn.execute( + "SELECT COUNT(*) FROM traceroute_routes WHERE parse_status = 'pending'" + ).fetchone()[0] + == 3 + ) + assert conn.execute("SELECT COUNT(*) FROM traceroute_hops").fetchone()[0] == 0 + with conn: + conn.execute("DROP TRIGGER reject_packet") + assert prepare_traceroutes(conn)["complete"] + + +def test_capture_can_commit_between_backfill_batches(database): + _, conn = database + with conn: + for _ in range(4): + insert_raw(conn, payload(snr=[4])) + result = prepare_traceroutes( + conn, batch_size=2, progress=lambda _: capture(payload(snr=[8])) + ) + assert result["processed_this_run"] == 4 + assert result["routes"] == result["raw_traceroutes"] == 6 + assert result["complete"] + + +def test_raw_only_import_during_backfill_cannot_look_complete(database): + path, conn = database + with conn: + insert_raw(conn, payload(snr=[4])) + + def raw_import(_): + with closing(sqlite3.connect(path)) as other, other: + other.row_factory = sqlite3.Row + insert_raw(other, payload(snr=[8])) + + result = prepare_traceroutes(conn, progress=raw_import) + assert not result["complete"] + assert result["pending"] == 1 + assert prepare_traceroutes(conn)["complete"] + + +def test_missing_endpoints_are_not_coerced_and_missing_payload_is_reported(database): + _, conn = database + with conn: + insert_raw(conn, None, mesh_packet_id=None, from_node_id=None, to_node_id=None) + insert_raw( + conn, + payload(route=[110]), + mesh_packet_id=None, + from_node_id=None, + to_node_id=None, + ) + result = prepare_traceroutes(conn) + assert result["invalid_payload"] == 1 + assert result["parsed"] == 1 + assert result["hops"] == 0 + for row in conn.execute("SELECT * FROM traceroute_routes"): + assert row["from_node_id"] is row["to_node_id"] is row["mesh_packet_id"] is None + + +def test_version_downgrade_is_rejected(database): + _, conn = database + capture(payload(snr=[4])) + with conn: + conn.execute( + "UPDATE traceroute_routes SET parser_version = ?", (PARSER_VERSION + 1,) + ) + with pytest.raises(ValueError, match="newer traceroute decoder"): + prepare_traceroutes(conn) + assert not inspect_traceroutes(conn.cursor())["complete"] + + +def test_cli_requires_explicit_database_and_rejects_missing_path(tmp_path): + with pytest.raises(SystemExit) as exc: + main([]) + assert exc.value.code == 2 + path = tmp_path / "missing.db" + assert main(["--database", str(path)]) == 1 + assert not path.exists() + + +def test_cli_check_does_not_create_schema_or_change_history(database, capsys): + path, conn = database + with conn: + for event in ("insert", "update", "delete"): + conn.execute(f"DROP TRIGGER traceroute_packet_{event}") + conn.execute("DROP TABLE traceroute_hops") + conn.execute("DROP TABLE traceroute_routes") + insert_raw(conn, payload(snr=[4])) + before = list(conn.iterdump()) + assert main(["--database", str(path), "--check"]) == 1 + assert list(conn.iterdump()) == before + output = capsys.readouterr().out + assert str(path.resolve()) in output + assert '"missing": 1' in output + # Shared startup creates tables and pending triggers, but does not backfill. + with conn: + ensure_startup_schema(conn.cursor()) + assert conn.execute("SELECT COUNT(*) FROM traceroute_routes").fetchone()[0] == 0 + assert not inspect_traceroutes(conn.cursor())["complete"] + assert main(["--database", str(path), "--batch-size", "1"]) == 0 + assert inspect_traceroutes(conn.cursor())["complete"] + + +def test_validation_reports_missing_and_orphaned_rows(database): + _, conn = database + capture(payload(snr=[4])) + conn.execute("PRAGMA foreign_keys=OFF") + with conn: + conn.execute("DELETE FROM traceroute_routes") + ensure_traceroute_schema(conn.cursor()) + with conn: + conn.execute("BEGIN") + result = inspect_traceroutes(conn.cursor()) + assert result["missing"] == 1 + assert result["orphan_hops"] == 1 + assert not result["complete"] + + +def test_clearing_derived_history_invalidates_completeness(database): + _, conn = database + capture(payload(snr=[4])) + assert prepare_traceroutes(conn)["complete"] + with conn: + conn.execute("DELETE FROM traceroute_routes") + assert not inspect_traceroutes(conn.cursor())["complete"] + assert prepare_traceroutes(conn)["complete"] + with conn: + conn.execute("DELETE FROM packet_history") + assert inspect_traceroutes(conn.cursor())["complete"] + + +def test_backfill_includes_explicit_nonpositive_ids_and_either_port_field(database): + _, conn = database + with conn: + insert_raw(conn, payload(snr=[4]), id=-1, portnum=None) + insert_raw(conn, payload(snr=[4]), id=0, portnum_name=None) + result = prepare_traceroutes(conn, batch_size=1) + assert result["processed_this_run"] == 2 + assert result["complete"] + + +def test_new_decoder_reprepares_old_records(database, monkeypatch): + _, conn = database + capture(payload(snr=[4])) + assert prepare_traceroutes(conn)["complete"] + monkeypatch.setattr("malla.database.traceroutes.PARSER_VERSION", PARSER_VERSION + 1) + monkeypatch.setattr("malla.backfill_traceroutes.PARSER_VERSION", PARSER_VERSION + 1) + assert not inspect_traceroutes(conn.cursor())["complete"] + result = prepare_traceroutes(conn) + assert result["complete"] + assert result["processed_this_run"] == 1 + assert ( + conn.execute("SELECT parser_version FROM traceroute_routes").fetchone()[0] + == PARSER_VERSION + 1 + ) From 95c62156d72fc3bc283fc8aab520ae361004ea14 Mon Sep 17 00:00:00 2001 From: roses Date: Thu, 10 Sep 2026 11:28:02 +0200 Subject: [PATCH 2/2] fix(ui): prevent URLFilterManager from overwriting user input during async restoration Track user-edited fields in a Set so async applyURLParameters() does not overwrite user input in form pickers if typing occurs while parameter resolution is in flight (e.g. following a timezone toggle or page reload). --- src/malla/static/js/url-filter-manager.js | 34 +++++++++++++++++++++++ 1 file changed, 34 insertions(+) diff --git a/src/malla/static/js/url-filter-manager.js b/src/malla/static/js/url-filter-manager.js index 9f96b0c9..9fad8e17 100644 --- a/src/malla/static/js/url-filter-manager.js +++ b/src/malla/static/js/url-filter-manager.js @@ -12,6 +12,29 @@ class URLFilterManager { this.form = document.querySelector(this.options.formSelector); this.groupingCheckbox = document.querySelector(this.options.groupingSelector); + this._userEditedFields = new Set(); + this._trackUserEdits(); + } + + /** + * Track fields the user has interacted with, so that the async + * applyURLParameters() restoration never overwrites user input + * (it can still be in flight when a user starts typing after a reload). + */ + _trackUserEdits() { + const markEdited = (event) => { + const target = event.target; + if (!target) return; + // Prefer the field name; picker search inputs only carry an id + const field = target.name || target.id; + if (field) { + this._userEditedFields.add(field); + } + }; + // Listen at document level: some controls (e.g. the grouping + // checkbox) may live outside the form element + document.addEventListener('input', markEdited, true); + document.addEventListener('change', markEdited, true); } /** @@ -58,6 +81,7 @@ class URLFilterManager { // Apply simple form field values Object.entries(urlParams).forEach(([key, value]) => { + if (this._userEditedFields.has(key)) return; // never clobber user input if (key === 'group_packets') { if (this.groupingCheckbox) { this.groupingCheckbox.checked = value === 'true'; @@ -102,6 +126,8 @@ class URLFilterManager { * Set node picker value by node ID */ async setNodePickerValue(fieldName, nodeId) { + // Skip if the user edited this field while we were initialising + if (this._userEditedFields.has(fieldName)) return; try { let displayName = `Node ${nodeId}`; // Default fallback @@ -124,6 +150,9 @@ class URLFilterManager { document.querySelector(`input[data-field="${fieldName}"]`) || document.querySelector(`.node-picker-input[data-field="${fieldName}"]`); + // Re-check: the user may have edited this field while we fetched + if (this._userEditedFields.has(fieldName)) return; + if (hiddenField && visibleField) { hiddenField.value = nodeId; visibleField.value = displayName; @@ -141,6 +170,8 @@ class URLFilterManager { * Set gateway picker value by gateway ID */ async setGatewayPickerValue(fieldName, gatewayId) { + // Skip if the user edited this field while we were initialising + if (this._userEditedFields.has(fieldName)) return; try { // Normalise: convert !hex to decimal string if applicable let nodeId = gatewayId; @@ -197,6 +228,9 @@ class URLFilterManager { document.querySelector(`input[data-field="${fieldName}"]`) || document.querySelector(`.gateway-picker-input[data-field="${fieldName}"]`); + // Re-check: the user may have edited this field while we fetched + if (this._userEditedFields.has(fieldName)) return; + if (hiddenField && visibleField) { hiddenField.value = nodeId; visibleField.value = displayName;