Merge pull request #29 from yellowcooln/dev

v1.6.6 fix peer history retention

AGENTS.md

@@ -1,6 +1,6 @@
 # Repository Guidelines
 
-Current version: `1.6.5` (see `VERSIONS.md`).
+Current version: `1.6.6` (see `VERSIONS.md`).
 
 ## Project Structure & Module Organization
 - `backend/app.py` wires FastAPI routes, MQTT lifecycle, and websocket broadcast flow.
@@ -17,7 +17,7 @@ Current version: `1.6.5` (see `VERSIONS.md`).
 - `docker-compose.yaml` runs the service as `meshmap-live`.
 - `data/` stores persisted state (`state.json`), route history (`route_history.jsonl`), role overrides (`device_roles.json`), and optional neighbor overrides (`neighbor_overrides.json`).
 - `.env` holds dev runtime settings; `.env.example` mirrors template defaults.
-- `VERSION.txt` tracks the current version (now `1.6.5`); append changes in `VERSIONS.md`.
+- `VERSION.txt` tracks the current version (now `1.6.6`); append changes in `VERSIONS.md`.
 
 ## Build, Test, and Development Commands
 - `docker compose up -d --build` rebuilds and restarts the backend (preferred workflow).
@@ -110,7 +110,7 @@ Current version: `1.6.5` (see `VERSIONS.md`).
 - Propagation render stays visible until a new render; origin changes only mark it dirty.
 - Propagation now has an adjustable TX antenna gain (dBi) control, and Rx AGL defaults to 1m.
 - Preview image endpoint renders in-bounds device dots for shared links.
-- Peers tool opens a right-side panel showing incoming/outgoing neighbors (counts + %) based on recent route history; selecting a node draws peer lines on the map.
+- Peers tool opens a right-side panel showing incoming/outgoing neighbors (counts + %) based on rolling peer-history buckets; selecting a node draws peer lines on the map.
 - Peers tool ignores nodes listed in `MQTT_ONLINE_FORCE_NAMES` (used for observer listeners).
 - Units toggle (km/mi) is stored in localStorage and defaults to `DISTANCE_UNITS`.
 - PWA support is enabled via `/manifest.webmanifest` + `/sw.js` so mobile browsers can install the app.

ARCHITECTURE.md

@@ -1,7 +1,7 @@
 # Architecture Guide
 
 This document explains how the Mesh Live Map codebase is organized and how the components interact.
-Current version: `1.6.5` (see `VERSIONS.md`).
+Current version: `1.6.6` (see `VERSIONS.md`).
 
 ## High-Level Overview
 
@@ -134,6 +134,7 @@ routes: Dict[str, Dict]             # Active route visualizations
 heat_events: List[Dict]             # Recent activity points
 route_history_segments: List[Dict]  # 24h route history
 route_history_edges: Dict[str, Dict]# Aggregated edge counts
+peer_history_pairs: Dict[str, Dict] # Rolling peer buckets for /peers stats
 neighbor_edges: Dict[str, Dict]     # Neighbor adjacency cache
 ```
 
@@ -392,4 +393,4 @@ npx eslint backend/static/app.js
 ```
 
 Versioning:
-- See `VERSIONS.md` for the changelog; `VERSION.txt` mirrors the latest entry (`1.6.5`).
+- See `VERSIONS.md` for the changelog; `VERSION.txt` mirrors the latest entry (`1.6.6`).

CONTRIBUTING.md

@@ -8,7 +8,7 @@ Thanks for helping improve the MeshCore Live Map. This repo is intentionally lig
 3) Verify: `curl -s http://localhost:8080/snapshot`
 
 ## Versioning
-- Current version: `1.6.5` (see `VERSIONS.md`).
+- Current version: `1.6.6` (see `VERSIONS.md`).
 - Update `VERSION.txt` when adding features.
 - Append a new section to `VERSIONS.md` describing the change set.
 

README.md

@@ -1,6 +1,6 @@
 # Mesh Live Map
 
-Version: `1.6.5` (see [VERSIONS.md](VERSIONS.md))
+Version: `1.6.6` (see [VERSIONS.md](VERSIONS.md))
 
 Live MeshCore traffic map that renders nodes, routes, and activity in real time on a Leaflet map. The backend subscribes to MQTT over WebSockets+TLS or TCP, decodes MeshCore packets with [`meshcore-decoder-multibyte-patch`](https://www.npmjs.com/package/meshcore-decoder-multibyte-patch), and streams updates to the browser via WebSockets.
 
@@ -251,7 +251,7 @@ Use it:
 - On mobile, long‑press a node to select it for LOS.
 - LOS elevations are fetched via `/los/elevations` and LOS/relay math runs client-side (with `/los` fallback).
 - History tool always loads off (use the button or `history=on` in the URL).
-- Peers tool uses route history segments; forced MQTT listeners are excluded from peer lists.
+- Peers tool uses dedicated rolling peer-history buckets so 24h counts stay accurate even on high-volume meshes; forced MQTT listeners are excluded from peer lists.
 - URL params override stored settings: `lat`, `lon`/`lng`/`long`, `zoom`, `layer`, `history`, `heat`, `coverage`, `weather`, `weather_radar`, `weather_wind`, `labels`, `nodes`, `legend`, `menu`, `units`, `history_filter`.
 - Dark map also darkens node popups for readability.
 - Route styling uses payload type: 2/5 = Message (blue), 8/9 = Trace (orange), 4 = Advert (green).

VERSION.txt

@@ -1 +1 @@
-1.6.5
+1.6.6

VERSIONS.md

@@ -1,5 +1,11 @@
 # Versions
 
+## v1.6.6 (03-19-2026)
+- Fixed the Peers tool 24h counts on large meshes by moving peer statistics off raw `route_history_segments` and onto dedicated rolling peer-history buckets.
+- Peer counts are now time-windowed independently of `ROUTE_HISTORY_MAX_SEGMENTS`, so high traffic no longer causes the effective 24h window to shrink to a few hours.
+- Added peer-history bucket pruning/state persistence so `/peers/{device_id}` survives restarts and keeps the expected rolling window behavior.
+- Added regression tests covering peer counts after segment-cap pruning and peer-history state round trips.
+
 ## v1.6.5 (03-14-2026)
 - Made MQTT role detection conservative for accuracy: the map now only assigns roles from explicit role fields and numeric MeshCore role codes.
 - Stopped inferring device roles from weak MQTT name/model/client/origin/description hints that could mislabel normal nodes as room servers.

backend/app.py

@@ -47,7 +47,11 @@ from decoder import (
 )
 from history import (
   _load_route_history,
+  PEER_HISTORY_BUCKET_SECONDS,
+  _peer_history_cutoff,
+  _prune_peer_history,
   _prune_route_history,
+  _rebuild_peer_history_from_segments,
   _record_route_history,
   _route_history_saver,
 )
@@ -181,6 +185,7 @@ from state import (
   heat_events,
   route_history_segments,
   route_history_edges,
+  peer_history_pairs,
   node_hash_to_device,
   node_hash_collisions,
   node_hash_candidates,
@@ -527,6 +532,7 @@ def _serialize_state() -> Dict[str, Any]:
     "last_seen_in_path": state.last_seen_in_path,
     "first_seen_devices": first_seen_devices,
     "last_seen_in_advert": last_seen_in_advert,
+    "peer_history_pairs": peer_history_pairs,
   }
 
 
@@ -1137,6 +1143,46 @@ def _load_state() -> None:
         last_seen_in_advert[str(key)] = float(value)
       except (TypeError, ValueError):
         continue
+  raw_peer_history = data.get("peer_history_pairs") or {}
+  peer_history_pairs.clear()
+  if isinstance(raw_peer_history, dict):
+    for pair_key, value in raw_peer_history.items():
+      if not isinstance(pair_key, str) or not isinstance(value, dict):
+        continue
+      a_id = value.get("a_id")
+      b_id = value.get("b_id")
+      buckets = value.get("buckets")
+      if not isinstance(a_id, str) or not a_id.strip():
+        continue
+      if not isinstance(b_id, str) or not b_id.strip():
+        continue
+      if not isinstance(buckets, dict):
+        continue
+      clean_buckets: Dict[str, int] = {}
+      for bucket_key, count in buckets.items():
+        try:
+          bucket_ts = int(float(bucket_key))
+          bucket_count = int(count)
+        except (TypeError, ValueError):
+          continue
+        if bucket_count <= 0:
+          continue
+        clean_buckets[str(bucket_ts)] = bucket_count
+      if not clean_buckets:
+        continue
+      last_ts = value.get("last_ts")
+      try:
+        last_ts_val = float(last_ts) if last_ts is not None else 0.0
+      except (TypeError, ValueError):
+        last_ts_val = 0.0
+      peer_history_pairs[pair_key] = {
+        "a_id": a_id.strip(),
+        "b_id": b_id.strip(),
+        "buckets": clean_buckets,
+        "last_ts": last_ts_val,
+      }
+  if _prune_peer_history():
+    state.state_dirty = True
   # If first_seen data is missing, fall back to loaded last_seen values.
   if not raw_first_seen:
     for device_id, seen_ts in seen_devices.items():
@@ -2000,6 +2046,8 @@ async def reaper():
         first_seen_devices.pop(dev_id, None)
         last_seen_in_advert.pop(dev_id, None)
         state.last_seen_in_path.pop(dev_id, None)
+    if _prune_peer_history(now):
+      state.state_dirty = True
 
     await asyncio.sleep(5)
 
@@ -2901,24 +2949,46 @@ def _peer_stats_for_device(device_id: str, limit: int) -> Dict[str, Any]:
   outbound: Dict[str, int] = {}
   inbound_last: Dict[str, float] = {}
   outbound_last: Dict[str, float] = {}
-  for entry in route_history_segments:
+  cutoff = _peer_history_cutoff()
+  if not peer_history_pairs and route_history_segments:
+    _rebuild_peer_history_from_segments()
+  for entry in peer_history_pairs.values():
     if not isinstance(entry, dict):
       continue
     a_id = entry.get("a_id")
     b_id = entry.get("b_id")
     if not a_id or not b_id:
       continue
-    ts = entry.get("ts") or 0
+    buckets = entry.get("buckets")
+    if not isinstance(buckets, dict):
+      continue
+    count = 0
+    last_ts = 0.0
+    for bucket_key, bucket_count in buckets.items():
+      try:
+        bucket_start = float(bucket_key)
+        bucket_value = int(bucket_count)
+      except (TypeError, ValueError):
+        continue
+      if bucket_value <= 0:
+        continue
+      bucket_end = bucket_start + PEER_HISTORY_BUCKET_SECONDS
+      if bucket_end < cutoff:
+        continue
+      count += bucket_value
+      last_ts = max(last_ts, bucket_end)
+    if count <= 0:
+      continue
     if a_id == device_id and b_id != device_id:
       if _peer_is_excluded(b_id):
         continue
-      outbound[b_id] = outbound.get(b_id, 0) + 1
-      outbound_last[b_id] = max(outbound_last.get(b_id, 0), float(ts))
+      outbound[b_id] = outbound.get(b_id, 0) + count
+      outbound_last[b_id] = max(outbound_last.get(b_id, 0), float(last_ts))
     if b_id == device_id and a_id != device_id:
       if _peer_is_excluded(a_id):
         continue
-      inbound[a_id] = inbound.get(a_id, 0) + 1
-      inbound_last[a_id] = max(inbound_last.get(a_id, 0), float(ts))
+      inbound[a_id] = inbound.get(a_id, 0) + count
+      inbound_last[a_id] = max(inbound_last.get(a_id, 0), float(last_ts))
 
   inbound_total = sum(inbound.values())
   outbound_total = sum(outbound.values())

backend/history.py

@@ -29,6 +29,115 @@ for _part in ROUTE_HISTORY_PAYLOAD_TYPES.split(","):
   except ValueError:
     pass
 
+PEER_HISTORY_BUCKET_SECONDS = 300
+
+
+def _peer_history_cutoff(now: Optional[float] = None) -> float:
+  current = time.time() if now is None else float(now)
+  return current - (ROUTE_HISTORY_HOURS * 3600)
+
+
+def _peer_history_bucket_start(ts: float) -> int:
+  return int(float(ts) // PEER_HISTORY_BUCKET_SECONDS) * PEER_HISTORY_BUCKET_SECONDS
+
+
+def _peer_history_pair_key(a_id: str, b_id: str) -> str:
+  return f"{a_id}|{b_id}"
+
+
+def _record_peer_history_segment(a_id: Any, b_id: Any, ts: float) -> bool:
+  if not ROUTE_HISTORY_ENABLED or ROUTE_HISTORY_HOURS <= 0:
+    return False
+  if not isinstance(a_id, str) or not a_id.strip():
+    return False
+  if not isinstance(b_id, str) or not b_id.strip():
+    return False
+  a_id = a_id.strip()
+  b_id = b_id.strip()
+  if a_id == b_id:
+    return False
+  bucket_start = _peer_history_bucket_start(ts)
+  pair_key = _peer_history_pair_key(a_id, b_id)
+  entry = state.peer_history_pairs.get(pair_key)
+  if not entry:
+    entry = {
+      "a_id": a_id,
+      "b_id": b_id,
+      "buckets": {},
+      "last_ts": float(ts),
+    }
+    state.peer_history_pairs[pair_key] = entry
+  buckets = entry.get("buckets")
+  if not isinstance(buckets, dict):
+    buckets = {}
+    entry["buckets"] = buckets
+  bucket_key = str(bucket_start)
+  buckets[bucket_key] = int(buckets.get(bucket_key, 0)) + 1
+  entry["last_ts"] = max(float(entry.get("last_ts", 0.0)), float(ts))
+  return True
+
+
+def _prune_peer_history(now: Optional[float] = None) -> bool:
+  if not state.peer_history_pairs:
+    return False
+  cutoff = _peer_history_cutoff(now)
+  changed = False
+  for pair_key in list(state.peer_history_pairs.keys()):
+    entry = state.peer_history_pairs.get(pair_key)
+    if not isinstance(entry, dict):
+      state.peer_history_pairs.pop(pair_key, None)
+      changed = True
+      continue
+    buckets = entry.get("buckets")
+    if not isinstance(buckets, dict):
+      state.peer_history_pairs.pop(pair_key, None)
+      changed = True
+      continue
+    new_buckets: Dict[str, int] = {}
+    last_ts = 0.0
+    for bucket_key, count in buckets.items():
+      try:
+        bucket_start = float(bucket_key)
+        bucket_count = int(count)
+      except (TypeError, ValueError):
+        changed = True
+        continue
+      if bucket_count <= 0:
+        changed = True
+        continue
+      bucket_end = bucket_start + PEER_HISTORY_BUCKET_SECONDS
+      if bucket_end < cutoff:
+        changed = True
+        continue
+      new_buckets[str(int(bucket_start))] = bucket_count
+      last_ts = max(last_ts, bucket_end)
+    if not new_buckets:
+      state.peer_history_pairs.pop(pair_key, None)
+      changed = True
+      continue
+    if len(new_buckets) != len(buckets):
+      changed = True
+    entry["buckets"] = new_buckets
+    entry["last_ts"] = max(last_ts, float(entry.get("last_ts", 0.0)))
+  return changed
+
+
+def _rebuild_peer_history_from_segments() -> bool:
+  state.peer_history_pairs.clear()
+  changed = False
+  for entry in state.route_history_segments:
+    if not isinstance(entry, dict):
+      continue
+    a_id = entry.get("a_id")
+    b_id = entry.get("b_id")
+    ts = entry.get("ts")
+    if not isinstance(ts, (int, float)):
+      continue
+    if _record_peer_history_segment(a_id, b_id, float(ts)):
+      changed = True
+  _prune_peer_history()
+  return changed
+
 
 def _history_payload_allowed(payload_type: Optional[int]) -> bool:
   if not ROUTE_HISTORY_ENABLED or ROUTE_HISTORY_HOURS <= 0:
@@ -162,6 +271,8 @@ def _record_route_history(
     edge["count"] = int(edge.get("count", 0)) + 1
     edge["last_ts"] = max(edge.get("last_ts", float(ts)), float(ts))
     _update_history_edge_recent(edge, sample)
+    if a_id and b_id:
+      _record_peer_history_segment(a_id, b_id, float(ts))
     updated_keys.add(key)
 
   if not new_entries:
@@ -169,6 +280,9 @@ def _record_route_history(
 
   state.route_history_segments.extend(new_entries)
   _append_route_history_file(new_entries)
+  if new_entries:
+    _prune_peer_history(ts)
+    state.state_dirty = True
 
   updates = [
     state.route_history_edges[key]
@@ -332,6 +446,9 @@ def _load_route_history() -> None:
   if not loaded_any:
     return
 
+  if not state.peer_history_pairs:
+    _rebuild_peer_history_from_segments()
+
   if ROUTE_HISTORY_MAX_SEGMENTS > 0 and len(
     state.route_history_segments
   ) > ROUTE_HISTORY_MAX_SEGMENTS:

backend/state.py

@@ -49,6 +49,7 @@ routes: Dict[str, Dict[str, Any]] = {}
 heat_events: List[Dict[str, float]] = []
 route_history_segments: Deque[Dict[str, Any]] = deque()
 route_history_edges: Dict[str, Dict[str, Any]] = {}
+peer_history_pairs: Dict[str, Dict[str, Any]] = {}
 route_history_compact = False
 route_history_last_compact = 0.0
 node_hash_to_device: Dict[str, str] = {}

docs.md

@@ -1,13 +1,13 @@
 # Mesh Map Live: Implementation Notes
 
 This document captures the state of the project and the key changes made so far, so a new Codex session can pick up without losing context.
-Current version: `1.6.5` (see `VERSIONS.md`).
+Current version: `1.6.6` (see `VERSIONS.md`).
 
 ## Overview
 This project renders live MeshCore traffic on a Leaflet + OpenStreetMap map. A FastAPI backend subscribes to MQTT (WSS/TLS or TCP), decodes MeshCore packets using [`meshcore-decoder-multibyte-patch`](https://www.npmjs.com/package/meshcore-decoder-multibyte-patch), and broadcasts device updates and routes over WebSockets to the frontend. Core logic is split into config/state/decoder/LOS/history modules so changes are localized. The UI includes heatmap, LOS tools, map mode toggles, and a 24‑hour route history layer.
 
 ## Versioning
-- `VERSION.txt` holds the current version string (`1.6.5`).
+- `VERSION.txt` holds the current version string (`1.6.6`).
 - `VERSIONS.md` is an append-only changelog by version.
 
 ## Key Paths
@@ -95,7 +95,7 @@ This project renders live MeshCore traffic on a Leaflet + OpenStreetMap map. A F
 - History panel can be dismissed with the X button while keeping history lines visible (toggle History tool to show it again).
 - History slider modes: 0 = All, 1 = Blue only, 2 = Yellow only, 3 = Yellow + Red, 4 = Red only.
 - History legend swatch is hidden unless the History tool is active.
-- Peers tool shows incoming/outgoing neighbors for a selected node, with counts and percentages pulled from route history.
+- Peers tool shows incoming/outgoing neighbors for a selected node, with counts and percentages pulled from dedicated rolling peer-history buckets instead of raw route-history segments.
 - Peers tool skips nodes listed in `MQTT_ONLINE_FORCE_NAMES` (observer listeners).
 - Peers panel legend clarifies line colors (incoming = blue, outgoing = purple).
 - Coverage tool only appears when `COVERAGE_API_URL` is set; it fetches tiles on demand.

howto.md

@@ -1,7 +1,7 @@
 # How-To: MQTT Broker + Live Map
 
 This guide covers two parts: stand up a MeshCore MQTT broker and point the live map at it.
-Current version: `1.6.5` (see `VERSIONS.md`).
+Current version: `1.6.6` (see `VERSIONS.md`).
 
 ## 1) MQTT broker (meshcore-mqtt-broker)
 

tests/test_peer_history_stats.py

@@ -0,0 +1,80 @@
+import json
+import time
+
+import app
+import history
+import state
+
+
+def _clear_peer_state():
+  state.route_history_segments.clear()
+  state.route_history_edges.clear()
+  state.peer_history_pairs.clear()
+
+
+def _route(ts):
+  return {
+    "points": [[42.3601, -71.0589], [42.3611, -71.0579]],
+    "point_ids": ["AA001111", "BB001111"],
+    "payload_type": 5,
+    "message_hash": f"msg-{int(ts)}",
+    "origin_id": "AA001111",
+    "receiver_id": "BB001111",
+    "route_mode": "path",
+    "topic": "meshcore/test",
+    "ts": ts,
+  }
+
+
+def test_peer_stats_use_bucket_history_even_when_segment_limit_prunes(monkeypatch):
+  _clear_peer_state()
+  now = time.time()
+
+  monkeypatch.setattr(history, "ROUTE_HISTORY_ENABLED", True)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_HOURS", 24)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_MAX_SEGMENTS", 2)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_ALLOWED_MODES_SET", {"path"})
+  monkeypatch.setattr(app, "ROUTE_HISTORY_HOURS", 24)
+
+  history._record_route_history(_route(now - 3 * 3600))
+  history._record_route_history(_route(now - 2 * 3600))
+  history._record_route_history(_route(now - 1 * 3600))
+
+  assert len(state.route_history_segments) == 2
+
+  outbound = app._peer_stats_for_device("AA001111", limit=8)
+  inbound = app._peer_stats_for_device("BB001111", limit=8)
+
+  assert outbound["outgoing_total"] == 3
+  assert inbound["incoming_total"] == 3
+  assert outbound["outgoing"][0]["peer_id"] == "BB001111"
+  assert inbound["incoming"][0]["peer_id"] == "AA001111"
+
+
+def test_state_round_trip_preserves_peer_history_buckets(tmp_path, monkeypatch):
+  _clear_peer_state()
+  now = time.time()
+  state_file = tmp_path / "state.json"
+
+  monkeypatch.setattr(history, "ROUTE_HISTORY_ENABLED", True)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_HOURS", 24)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_MAX_SEGMENTS", 10)
+  monkeypatch.setattr(history, "ROUTE_HISTORY_ALLOWED_MODES_SET", {"path"})
+  monkeypatch.setattr(app, "ROUTE_HISTORY_HOURS", 24)
+  monkeypatch.setattr(app, "STATE_FILE", str(state_file))
+  monkeypatch.setattr(app, "DEVICE_ROLES_FILE", "")
+  monkeypatch.setattr(app, "DEVICE_COORDS_FILE", "")
+
+  history._record_route_history(_route(now - 600))
+  history._record_route_history(_route(now - 300))
+
+  state_file.write_text(json.dumps(app._serialize_state()), encoding="utf-8")
+
+  _clear_peer_state()
+  app._load_state()
+
+  assert len(state.peer_history_pairs) == 1
+
+  outbound = app._peer_stats_for_device("AA001111", limit=8)
+  assert outbound["outgoing_total"] == 2
+  assert outbound["outgoing"][0]["peer_id"] == "BB001111"

tests/test_state_persistence.py

@@ -78,6 +78,7 @@ def test_load_state_drops_zero_devices_and_keeps_valid_entries(
   state.device_roles.clear()
   state.device_role_sources.clear()
   state.last_seen_in_path.clear()
+  state.peer_history_pairs.clear()
 
   monkeypatch.setattr(app, "STATE_FILE", str(state_file))
   monkeypatch.setattr(app, "DEVICE_ROLES_FILE", "")
@@ -117,14 +118,17 @@ def test_route_history_round_trip_file_load(tmp_path, monkeypatch):
 
   state.route_history_segments.clear()
   state.route_history_edges.clear()
+  state.peer_history_pairs.clear()
   history._append_route_history_file([entry])
 
   state.route_history_segments.clear()
   state.route_history_edges.clear()
+  state.peer_history_pairs.clear()
   history._load_route_history()
 
   assert len(state.route_history_segments) == 1
   assert len(state.route_history_edges) == 1
+  assert len(state.peer_history_pairs) == 1
   loaded = state.route_history_segments[0]
   assert loaded["a_id"] == "AA001111"
   assert loaded["b_id"] == "BB001111"
@@ -192,9 +196,11 @@ def test_route_history_load_skips_bad_lines_and_marks_compact(
 
   state.route_history_segments.clear()
   state.route_history_edges.clear()
+  state.peer_history_pairs.clear()
   state.route_history_compact = False
   history._load_route_history()
 
   assert len(state.route_history_segments) == 1
   assert len(state.route_history_edges) == 1
+  assert len(state.peer_history_pairs) == 1
   assert state.route_history_compact is True