routing tests, launch.configs, redirect, stats and route statistics

packages/tracker/meshnet_tracker/routing_stats.py

@@ -23,6 +23,7 @@ with a different shard produces a new arm automatically.
 from __future__ import annotations
 
 import random
+import sqlite3
 import threading
 import time
 from dataclasses import dataclass, field
@@ -45,6 +46,7 @@ class RoutingConfig:
 @dataclass
 class RouteStat:
     ewma_tps: float = 0.0
+    ewma_latency_ms: float = 0.0
     weight: float = 0.0  # decayed effective sample mass
     last_sample_ts: float = 0.0
     epoch: int = 0
@@ -75,11 +77,15 @@ def route_signature(model_key: str, nodes: Iterable[Any]) -> str:
 class RouteStatsStore:
     """Thread-safe per-route decayed throughput statistics."""
 
-    def __init__(self, config: RoutingConfig | None = None) -> None:
+    def __init__(self, config: RoutingConfig | None = None, db_path: str | None = None) -> None:
         self.config = config or RoutingConfig()
         self._lock = threading.Lock()
         self._stats: dict[str, RouteStat] = {}
         self._epochs: dict[str, int] = {}
+        self._db_path = db_path
+        if db_path:
+            self._init_db()
+            self._load_from_db()
 
     def epoch(self, model_key: str) -> int:
         with self._lock:
@@ -119,6 +125,7 @@ class RouteStatsStore:
             carried = stat.decayed_weight(ts, cfg.stats_half_life_seconds)
             total = carried + 1.0
             stat.ewma_tps = (stat.ewma_tps * carried + tps) / total
+            stat.ewma_latency_ms = (stat.ewma_latency_ms * carried + elapsed_seconds * 1000.0) / total
             stat.weight = total
             stat.last_sample_ts = ts
             stat.epoch = self._epochs.get(model_key, 0)
@@ -143,11 +150,64 @@ class RouteStatsStore:
             status = "proven"
         return {
             "tps": round(stat.ewma_tps, 4) if stat.samples else None,
+            "latency_ms": round(stat.ewma_latency_ms, 3) if stat.samples else None,
             "weight": round(weight, 4),
             "samples": stat.samples,
             "status": status,
         }
 
+    def model_rows(self, model_key: str, now: float | None = None) -> list[dict]:
+        """All measured route samples, including pinned experiment routes."""
+        prefix = f"{model_key}|"
+        with self._lock:
+            signatures = [signature for signature in self._stats if signature.startswith(prefix)]
+        rows = [
+            {
+                "signature": signature,
+                "hop_count": signature.count("->") + 1,
+                **self.snapshot(signature, model_key, now=now),
+            }
+            for signature in signatures
+        ]
+        return sorted(rows, key=lambda row: (row["hop_count"], row["signature"]))
+
+    def _init_db(self) -> None:
+        con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
+        con.execute(
+            "CREATE TABLE IF NOT EXISTS route_stats "
+            "(signature TEXT PRIMARY KEY, ewma_tps REAL NOT NULL, ewma_latency_ms REAL NOT NULL, "
+            "weight REAL NOT NULL, last_sample_ts REAL NOT NULL, epoch INTEGER NOT NULL, samples INTEGER NOT NULL)"
+        )
+        con.execute("CREATE TABLE IF NOT EXISTS route_stat_epochs (model_key TEXT PRIMARY KEY, epoch INTEGER NOT NULL)")
+        con.commit()
+        con.close()
+
+    def save_to_db(self) -> None:
+        if not self._db_path:
+            return
+        with self._lock:
+            rows = [
+                (signature, stat.ewma_tps, stat.ewma_latency_ms, stat.weight, stat.last_sample_ts, stat.epoch, stat.samples)
+                for signature, stat in self._stats.items()
+            ]
+            epochs = list(self._epochs.items())
+        con = sqlite3.connect(self._db_path)
+        con.executemany("INSERT OR REPLACE INTO route_stats VALUES (?,?,?,?,?,?,?)", rows)
+        con.executemany("INSERT OR REPLACE INTO route_stat_epochs VALUES (?,?)", epochs)
+        con.commit()
+        con.close()
+
+    def _load_from_db(self) -> None:
+        con = sqlite3.connect(self._db_path)  # type: ignore[arg-type]
+        rows = con.execute("SELECT signature, ewma_tps, ewma_latency_ms, weight, last_sample_ts, epoch, samples FROM route_stats").fetchall()
+        epochs = con.execute("SELECT model_key, epoch FROM route_stat_epochs").fetchall()
+        con.close()
+        self._stats = {
+            signature: RouteStat(float(tps), float(latency), float(weight), float(last_sample_ts), int(epoch), int(samples))
+            for signature, tps, latency, weight, last_sample_ts, epoch, samples in rows
+        }
+        self._epochs = {str(model_key): int(epoch) for model_key, epoch in epochs}
+
     def prune(self, now: float | None = None) -> int:
         """Drop routes with no samples for `prune_after_seconds`."""
         ts = time.time() if now is None else now
@@ -246,6 +306,8 @@ def route_table(
                 for n in cand.nodes
             ],
             "tps": r["tps"],
+            "latency_ms": r["latency_ms"],
+            "hop_count": len(cand.nodes),
             "coefficient": coefficient,
             "expected_share": round(share, 4),
             "samples": r["samples"],