mirror of
https://github.com/agessaman/meshcore-bot.git
synced 2026-08-15 23:20:28 +00:00
agessaman
96c4a01788
Remaining findings from the same triage pass. None are security
relevant; each produces a wrong user-visible result.
- greeter: treat rollout_started_at as UTC. It is written by SQLite
CURRENT_TIMESTAMP, but datetime.timestamp() read the naive value as
local time, shifting the backfill cutoff by the host's offset (7h on
PT). Users who posted inside that window were never marked as already
greeted and could be sent a welcome they should not have received.
- greeter: write greeted_at in SQLite's own format. The rollout backfill
used isoformat() while every other path used CURRENT_TIMESTAMP. "T"
sorts above a space, so ORDER BY greeted_at interleaved the two
formats wrongly, corrupting duplicate cleanup and the web viewer's
recently-greeted list.
- greeter: let an empty `channels =` disable the command. BaseCommand
reads an empty value as disabled-on-channels, but the greeter
collapsed "key absent" and "key present but empty" into one fallback
and kept greeting via monitor_channels.
- greeter: stop comma-splitting greeting text. channel_greetings split
entries on ",", so "Public:Welcome to the mesh, {sender}!" was stored
as "Welcome to the mesh" with the placeholder silently dropped. A
fragment now starts a new entry only when the text before its first
colon looks like a channel name, which keeps commas, URLs and clock
times attached to the greeting they belong to.
- wxsim_parser: match condition abbreviations longest-first. Substring
matching in dict order let RAIN shadow CHNC. RAIN, so five conditions
lost their "chance" qualifier (rain, snow, drizzle, t-storm, and
FAIR-P.C.).
- wxsim_parser: re-anchor "now" on each parse. current_date and
current_year were fixed at construction and wx_command builds one
parser at startup, so after a few days forecast dates rolled back a
year and staleness checks read permanently true.
- thesportsdb_client: hold the rate-limit read/sleep/write under a lock.
Concurrent callers read the same last_request_time, slept the same
interval and fired together, bursting past the 2.1s throttle. This
exposes an unrelated request fan-out problem in fetch_league_scores;
filed in TODO.md rather than fixed here.
- alert_command: stop duplicating the first incident. The tail treated
any single-line buffer as "header only" and appended incidents[0],
but messages after the first carry no header, so a final chunk holding
one incident had incident 0 pasted onto it. The same confusion inside
the loop also let a second incident be appended past the 130-character
limit.
- feed: reject non-positive poll intervals. -1 is truthy and was stored,
and the poller's `now - last_check >= interval` then treats the feed
as permanently due and re-fetches the URL every cycle; 0 was ignored
while still reporting success. The web viewer, which is the primary
editor, had no validation at all, and a JSON null there raised a
TypeError that aborted the poll cycle for every feed rather than one.
feed_manager falls back to the default for rows written before this.
- transmission_tracker: age out confirmed transmissions that have
repeats. Cleanup removed only repeat_count == 0, so repeated records
accumulated for the lifetime of the process, which matters on a Pi
Zero. Repeat counts are already persisted to packet_stream, so the
longer 30-minute retention loses nothing.
- mesh_graph: weighted-merge avg_hop_position when promoting an edge.
Promotion overwrote the average with the single new observation, so an
edge averaging 2.0 over 4 observations became 9.0 instead of 3.4 and
skewed subsequent path scoring.
- multitest_command: show the path that ends exactly at the display LCP.
The suffix helpers are correct in isolation; the loss happens in the
cluster formatters, where _shrink_display_lcp refuses to shrink a
single-token LCP and the trunk then rendered as "96 ┐", meaning
"everything continues past here" and dropping the bare 96 route while
the header still counted it. Now uses the file's existing "├ common"
marker. Empty suffixes are also no longer handed to the nested
renderer, which mapped them onto the same [] as a route ending at the
inner LCP and drew a row for a route that did not exist.
- sports_mappings: stop shadowing seven unique team nicknames. In one
flat dict a repeated key silently drops the earlier team, so hawks
resolved to the NBA Hawks rather than the Seahawks alias it was added
as, blazers to Kamloops rather than Portland, and rockets to Kelowna
rather than Houston. First definition now wins for hawks, giants,
jets, rangers, kings, blazers and rockets; every shadowed team keeps
its unambiguous full-name alias. The 55 city and abbreviation
collisions (chicago, sf, la, ...) are genuinely ambiguous and stay
last-wins, now pinned by a test so a new collision fails loudly
instead of passing unnoticed.
1476 lines
67 KiB
Python
1476 lines
67 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Mesh Graph Module
|
|
Tracks observed connections between repeaters to improve path guessing accuracy.
|
|
Persists graph state across bot restarts for development scenarios.
|
|
|
|
Multi-resolution storage and node identity:
|
|
Edges are stored at the resolution observed (2, 4, or 6 hex chars per node). Nodes
|
|
with the same logical identity (e.g. 01, 0101, 0101C1) are treated as one: on read,
|
|
get_edge uses prefix matching and returns the best-matching edge (longest prefix,
|
|
then observation_count, then last_seen). On write:
|
|
- 1-byte observations (2-char prefix): merge into an existing 2/3-byte edge only
|
|
when the edge is unique (exactly one prefix-matching edge). If zero or multiple
|
|
matches exist, create or update only the 1-byte edge to avoid false coalescing.
|
|
- 2/3-byte observations: merge into the best-matching edge when present; when the
|
|
new observation is more specific than the existing edge, promote (remove old edge,
|
|
add new at higher resolution with merged observation count; DB: delete old row,
|
|
insert new). Distinct links (e.g. 7e42→8611 and 7e99→86ff) stay separate.
|
|
"""
|
|
|
|
import sqlite3
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections import defaultdict
|
|
from datetime import datetime, timedelta
|
|
from typing import Any, Optional
|
|
|
|
|
|
def _merge_avg_hop_position(
|
|
current_avg: Optional[float], prior_count: int, hop_position: Optional[float]
|
|
) -> Optional[float]:
|
|
"""Fold one new hop-position observation into a running average.
|
|
|
|
``prior_count`` is the observation count *before* this observation. Returns
|
|
the existing average unchanged when there is nothing new to fold in.
|
|
"""
|
|
if hop_position is None:
|
|
return current_avg
|
|
if current_avg is None or prior_count <= 0:
|
|
return hop_position
|
|
return ((current_avg * prior_count) + hop_position) / (prior_count + 1)
|
|
|
|
|
|
class MeshGraph:
|
|
"""Graph structure tracking observed connections between mesh nodes."""
|
|
|
|
def __init__(self, bot):
|
|
"""Initialize the mesh graph.
|
|
|
|
Args:
|
|
bot: Bot instance with db_manager and config access.
|
|
"""
|
|
self.bot = bot
|
|
self.logger = bot.logger
|
|
self.db_manager = bot.db_manager
|
|
|
|
# Capture/validation feature flags
|
|
# graph_capture_enabled: controls whether new edge data is collected from packets
|
|
# When False, no new edges are added and the batch writer thread is not started.
|
|
self.capture_enabled = bot.config.getboolean('Path_Command', 'graph_capture_enabled', fallback=True)
|
|
|
|
# In-memory graph storage: {(from_prefix, to_prefix): edge_data}
|
|
self.edges: dict[tuple[str, str], dict] = {}
|
|
|
|
# Adjacency indexes for O(1) neighbour lookups (derived from self.edges)
|
|
self._outgoing_index: dict[str, set[str]] = defaultdict(set) # from_prefix -> set of to_prefixes
|
|
self._incoming_index: dict[str, set[str]] = defaultdict(set) # to_prefix -> set of from_prefixes
|
|
|
|
# First-byte (2 hex chars) buckets over the adjacency indexes, so prefix-match lookups scan
|
|
# only nodes sharing the query's first byte instead of every node. A prefix can only
|
|
# _prefix_match a query if they share their first byte (all prefixes are >= 1 byte and one is
|
|
# a prefix of the other), so bucketing by [:2] is exact. Kept in sync via _index_add/_remove.
|
|
self._outgoing_byte0: dict[str, set[str]] = defaultdict(set) # from_prefix[:2] -> set of from_prefixes
|
|
self._incoming_byte0: dict[str, set[str]] = defaultdict(set) # to_prefix[:2] -> set of to_prefixes
|
|
|
|
# Per-edge last-notification timestamps for web viewer throttling (unix float)
|
|
self._notification_timestamps: dict[tuple[str, str], float] = {}
|
|
|
|
# Track pending updates for batched writes
|
|
self.pending_updates: set[tuple[str, str]] = set()
|
|
self.pending_lock = threading.Lock()
|
|
|
|
# Write strategy configuration
|
|
self.write_strategy = bot.config.get('Path_Command', 'graph_write_strategy', fallback='hybrid')
|
|
self.batch_interval = bot.config.getint('Path_Command', 'graph_batch_interval_seconds', fallback=30)
|
|
self.batch_max_pending = bot.config.getint('Path_Command', 'graph_batch_max_pending', fallback=100)
|
|
# Default 14 days: edges older than this carry near-zero recency confidence anyway.
|
|
# Set to 0 in config.ini to load all historical edges (e.g. on servers with ample RAM).
|
|
self.startup_load_days = bot.config.getint('Path_Command', 'graph_startup_load_days', fallback=14)
|
|
self.edge_expiration_days = bot.config.getint('Path_Command', 'graph_edge_expiration_days', fallback=7)
|
|
|
|
# Background task for batched writes
|
|
self._batch_task = None
|
|
self._shutdown_event = threading.Event()
|
|
|
|
# Load graph from database on startup
|
|
self._load_from_database()
|
|
|
|
# Start background batch writer only when capture is active
|
|
if self.capture_enabled and self.write_strategy in ('batched', 'hybrid'):
|
|
self._start_batch_writer()
|
|
|
|
def _prefix_len(self) -> int:
|
|
"""Return configured prefix length in hex chars (always an int for slicing)."""
|
|
n = getattr(self.bot, 'prefix_hex_chars', 2)
|
|
try:
|
|
return max(2, int(n))
|
|
except (TypeError, ValueError):
|
|
return 2
|
|
|
|
def _valid_prefix_length(self, prefix: str) -> bool:
|
|
"""Return True if prefix has 2, 4, or 6 hex chars (after stripping)."""
|
|
if not prefix or not isinstance(prefix, str):
|
|
return False
|
|
s = prefix.strip().lower()
|
|
if len(s) not in (2, 4, 6):
|
|
return False
|
|
return all(c in '0123456789abcdef' for c in s)
|
|
|
|
def _prefix_match(self, a: str, b: str) -> bool:
|
|
"""Return True if a and b match: exact or one is a prefix of the other (after lowercasing)."""
|
|
if not a or not b:
|
|
return False
|
|
a, b = a.lower().strip(), b.lower().strip()
|
|
return a == b or a.startswith(b) or b.startswith(a)
|
|
|
|
def _get_edge_by_prefix_match(self, from_q: str, to_q: str) -> Optional[dict]:
|
|
"""Return the best matching edge for a prefix query. Single edge only; never merge counts.
|
|
Tie-break: exact key if present, else longest combined prefix length, then max
|
|
observation_count, then most recent last_seen.
|
|
"""
|
|
matches = self._find_all_matching_edges(from_q, to_q)
|
|
if not matches:
|
|
return None
|
|
# Return the edge data of the best match (first in list)
|
|
return matches[0][1]
|
|
|
|
def _find_all_matching_edges(
|
|
self, from_prefix: str, to_prefix: str
|
|
) -> list[tuple[tuple[str, str], dict]]:
|
|
"""Return all edges that prefix-match (from_prefix, to_prefix), ordered by best match first.
|
|
|
|
Best = longest combined prefix length, then observation_count desc, then last_seen desc.
|
|
Used for 1-byte uniqueness check (merge only when len==1) and for 2/3-byte merge/promote.
|
|
"""
|
|
from_q = from_prefix.lower().strip() if from_prefix else ""
|
|
to_q = to_prefix.lower().strip() if to_prefix else ""
|
|
if not from_q or not to_q:
|
|
return []
|
|
|
|
# Only from-nodes sharing from_q's first byte can prefix-match it, so scan just that bucket
|
|
# rather than every from-node (and never the full edge set). Same candidate set as a full
|
|
# scan (indexes mirror self.edges), but O(bucket + matched). This is the hot path for
|
|
# get_edge()'s prefix fallback during graph path inference.
|
|
candidates: list[tuple[tuple[str, str], dict]] = []
|
|
for from_p in self._outgoing_byte0.get(from_q[:2], ()):
|
|
if not self._prefix_match(from_p, from_q):
|
|
continue
|
|
for to_p in self._outgoing_index[from_p]:
|
|
if to_p[:2] == to_q[:2] and self._prefix_match(to_p, to_q):
|
|
edge = self.edges.get((from_p, to_p))
|
|
if edge is not None:
|
|
candidates.append(((from_p, to_p), edge))
|
|
|
|
if not candidates:
|
|
return []
|
|
|
|
def sort_key(item):
|
|
edge_key, edge = item
|
|
from_p, to_p = edge_key
|
|
spec = len(from_p) + len(to_p)
|
|
obs = edge.get("observation_count", 0)
|
|
last = edge.get("last_seen")
|
|
if isinstance(last, str):
|
|
try:
|
|
last = datetime.fromisoformat(last.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
last = datetime.min
|
|
last_ts = last if isinstance(last, datetime) else datetime.min
|
|
return (-spec, -obs, last_ts) # desc spec, desc obs, asc time -> most recent last
|
|
|
|
candidates.sort(key=sort_key)
|
|
return candidates
|
|
|
|
def _index_add(self, from_p: str, to_p: str) -> None:
|
|
"""Register an edge in all adjacency indexes (both directions + first-byte buckets).
|
|
|
|
Single point of truth for index maintenance — call whenever self.edges gains a key.
|
|
"""
|
|
self._outgoing_index[from_p].add(to_p)
|
|
self._incoming_index[to_p].add(from_p)
|
|
self._outgoing_byte0[from_p[:2]].add(from_p)
|
|
self._incoming_byte0[to_p[:2]].add(to_p)
|
|
|
|
def _index_remove(self, from_p: str, to_p: str) -> None:
|
|
"""Unregister an edge from all adjacency indexes, pruning now-empty buckets.
|
|
|
|
Call whenever self.edges loses a key. The byte0 buckets are only pruned of a node once it
|
|
has no remaining edges in the corresponding direction.
|
|
"""
|
|
if from_p in self._outgoing_index:
|
|
self._outgoing_index[from_p].discard(to_p)
|
|
if not self._outgoing_index[from_p]:
|
|
del self._outgoing_index[from_p]
|
|
self._outgoing_byte0[from_p[:2]].discard(from_p)
|
|
if not self._outgoing_byte0[from_p[:2]]:
|
|
del self._outgoing_byte0[from_p[:2]]
|
|
if to_p in self._incoming_index:
|
|
self._incoming_index[to_p].discard(from_p)
|
|
if not self._incoming_index[to_p]:
|
|
del self._incoming_index[to_p]
|
|
self._incoming_byte0[to_p[:2]].discard(to_p)
|
|
if not self._incoming_byte0[to_p[:2]]:
|
|
del self._incoming_byte0[to_p[:2]]
|
|
|
|
def _remove_edge_from_memory(self, edge_key: tuple[str, str]) -> None:
|
|
"""Remove an edge from in-memory graph and adjacency indexes.
|
|
Does not touch the database. Used when promoting to a higher-resolution key.
|
|
"""
|
|
if edge_key not in self.edges:
|
|
return
|
|
from_p, to_p = edge_key
|
|
del self.edges[edge_key]
|
|
self._index_remove(from_p, to_p)
|
|
self._notification_timestamps.pop(edge_key, None)
|
|
|
|
def _delete_edge_from_db(
|
|
self, edge_key: tuple[str, str], conn: Optional[sqlite3.Connection] = None
|
|
) -> int:
|
|
"""Delete a single edge row from mesh_connections. Returns rows affected."""
|
|
from_p, to_p = edge_key
|
|
query = "DELETE FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?"
|
|
params = (from_p, to_p)
|
|
if conn is not None:
|
|
return self.db_manager.execute_update_on_connection(conn, query, params)
|
|
return self.db_manager.execute_update(query, params)
|
|
|
|
def _update_edge_data(
|
|
self,
|
|
edge: dict,
|
|
now: datetime,
|
|
hop_position: Optional[int] = None,
|
|
from_public_key: Optional[str] = None,
|
|
to_public_key: Optional[str] = None,
|
|
geographic_distance: Optional[float] = None,
|
|
prefix_bytes: int = 1,
|
|
) -> None:
|
|
"""Apply one observation to an edge dict (increment count, last_seen, optional fields)."""
|
|
edge["observation_count"] = edge.get("observation_count", 0) + 1
|
|
edge["last_seen"] = now
|
|
if hop_position is not None:
|
|
current_avg = edge.get("avg_hop_position")
|
|
count = edge["observation_count"]
|
|
if current_avg is not None:
|
|
edge["avg_hop_position"] = ((current_avg * (count - 1)) + hop_position) / count
|
|
else:
|
|
edge["avg_hop_position"] = hop_position
|
|
if from_public_key:
|
|
edge["from_public_key"] = from_public_key
|
|
if to_public_key:
|
|
edge["to_public_key"] = to_public_key
|
|
if geographic_distance is not None:
|
|
edge["geographic_distance"] = geographic_distance
|
|
if prefix_bytes == 2:
|
|
edge["confirmed_2byte"] = True
|
|
|
|
def _load_from_database(self):
|
|
"""Load graph edges from database on startup."""
|
|
try:
|
|
query = '''
|
|
SELECT from_prefix, to_prefix, from_public_key, to_public_key,
|
|
observation_count, first_seen, last_seen, avg_hop_position,
|
|
geographic_distance
|
|
FROM mesh_connections
|
|
'''
|
|
|
|
# Build WHERE clause combining startup_load_days and edge_expiration_days.
|
|
# startup_load_days: explicit cap on how far back to load (0 = no cap).
|
|
# edge_expiration_days: always applied — never load edges we would immediately
|
|
# evict as expired (this bounds memory even when startup_load_days=0).
|
|
where_parts = []
|
|
if self.startup_load_days > 0:
|
|
cutoff_date = datetime.now() - timedelta(days=self.startup_load_days)
|
|
where_parts.append(f"last_seen >= '{cutoff_date.isoformat()}'")
|
|
if self.edge_expiration_days > 0:
|
|
expiry_date = datetime.now() - timedelta(days=self.edge_expiration_days)
|
|
where_parts.append(f"last_seen >= '{expiry_date.isoformat()}'")
|
|
if where_parts:
|
|
# Use the most restrictive (most recent) cutoff
|
|
query += " WHERE " + " AND ".join(where_parts)
|
|
|
|
query += " ORDER BY last_seen DESC"
|
|
|
|
results = self.db_manager.execute_query(query)
|
|
|
|
edge_count = 0
|
|
for row in results:
|
|
from_prefix = row['from_prefix']
|
|
to_prefix = row['to_prefix']
|
|
edge_key = (from_prefix, to_prefix)
|
|
|
|
# Intern public key strings so identical keys across many edges share
|
|
# a single string object in memory rather than duplicating bytes.
|
|
from_pk = row.get('from_public_key')
|
|
to_pk = row.get('to_public_key')
|
|
if from_pk:
|
|
from_pk = sys.intern(from_pk)
|
|
if to_pk:
|
|
to_pk = sys.intern(to_pk)
|
|
|
|
self.edges[edge_key] = {
|
|
'from_prefix': from_prefix,
|
|
'to_prefix': to_prefix,
|
|
'from_public_key': from_pk,
|
|
'to_public_key': to_pk,
|
|
'observation_count': row.get('observation_count', 1),
|
|
'first_seen': row.get('first_seen'),
|
|
'last_seen': row.get('last_seen'),
|
|
'avg_hop_position': row.get('avg_hop_position'),
|
|
'geographic_distance': row.get('geographic_distance')
|
|
}
|
|
|
|
# Maintain adjacency indexes
|
|
self._index_add(from_prefix, to_prefix)
|
|
|
|
edge_count += 1
|
|
|
|
self.logger.info(f"Loaded {edge_count} graph edges from database")
|
|
|
|
# Log statistics
|
|
if edge_count > 0:
|
|
total_observations = sum(e['observation_count'] for e in self.edges.values())
|
|
self.logger.info(f"Graph statistics: {edge_count} edges, {total_observations} total observations")
|
|
|
|
# Belt-and-suspenders: prune any edges that slipped through the SQL filter
|
|
# (e.g. timezone edge cases or edges loaded before expiration_days was set)
|
|
self.prune_expired_edges()
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error loading graph from database: {e}")
|
|
# Continue with empty graph
|
|
|
|
def add_edge(self, from_prefix: str, to_prefix: str,
|
|
from_public_key: Optional[str] = None,
|
|
to_public_key: Optional[str] = None,
|
|
hop_position: Optional[int] = None,
|
|
geographic_distance: Optional[float] = None,
|
|
prefix_bytes: int = 1):
|
|
"""Add or update an edge in the graph.
|
|
|
|
Prefixes are stored at the resolution provided (2, 4, or 6 hex chars).
|
|
No truncation; the same physical link can be recorded at different
|
|
resolutions and will create/update the matching edge.
|
|
|
|
Args:
|
|
from_prefix: Source node prefix (2, 4, or 6 hex chars depending on path encoding).
|
|
to_prefix: Destination node prefix (2, 4, or 6 hex chars depending on path encoding).
|
|
from_public_key: Full public key of source node (optional).
|
|
to_public_key: Full public key of destination node (optional).
|
|
hop_position: Position in path where this edge was observed (optional).
|
|
geographic_distance: Distance in km between nodes (optional).
|
|
prefix_bytes: 1 = 1-byte (2-char) prefix (default); 2 = 2-byte confirmed (stored as edge flag for weighting).
|
|
"""
|
|
if not from_prefix or not to_prefix:
|
|
return
|
|
|
|
# Respect the capture kill-switch — allow reads but block writes
|
|
if not self.capture_enabled:
|
|
return
|
|
|
|
# Normalize to lowercase; validate length (2, 4, or 6 hex chars). No truncation.
|
|
from_prefix = from_prefix.lower().strip()
|
|
to_prefix = to_prefix.lower().strip()
|
|
if not self._valid_prefix_length(from_prefix) or not self._valid_prefix_length(to_prefix):
|
|
return
|
|
|
|
# Intern public key strings so repeated identical keys share one object in RAM
|
|
if from_public_key:
|
|
from_public_key = sys.intern(from_public_key)
|
|
if to_public_key:
|
|
to_public_key = sys.intern(to_public_key)
|
|
|
|
edge_key = (from_prefix, to_prefix)
|
|
now = datetime.now()
|
|
|
|
matches = self._find_all_matching_edges(from_prefix, to_prefix)
|
|
best = matches[0] if matches else None
|
|
incoming_1byte = len(from_prefix) == 2 or len(to_prefix) == 2
|
|
|
|
# 1-byte: merge only when exactly one matching edge (unique link)
|
|
if incoming_1byte and len(matches) == 1:
|
|
target_key, target_edge = matches[0]
|
|
self._update_edge_data(
|
|
target_edge, now, hop_position,
|
|
from_public_key, to_public_key, geographic_distance, prefix_bytes,
|
|
)
|
|
self._persist_and_notify_edge(target_key, is_new_edge=False)
|
|
return
|
|
|
|
# 2/3-byte: merge into best match or promote
|
|
if not incoming_1byte and best is not None:
|
|
best_key, best_edge = best
|
|
best_spec = len(best_key[0]) + len(best_key[1])
|
|
new_spec = len(from_prefix) + len(to_prefix)
|
|
|
|
if best_key == edge_key:
|
|
# Update in place
|
|
self._update_edge_data(
|
|
best_edge, now, hop_position,
|
|
from_public_key, to_public_key, geographic_distance, prefix_bytes,
|
|
)
|
|
self._persist_and_notify_edge(edge_key, is_new_edge=False)
|
|
return
|
|
|
|
if best_spec > new_spec:
|
|
# Best match is more specific — update that edge and return
|
|
self._update_edge_data(
|
|
best_edge, now, hop_position,
|
|
from_public_key, to_public_key, geographic_distance, prefix_bytes,
|
|
)
|
|
self._persist_and_notify_edge(best_key, is_new_edge=False)
|
|
return
|
|
|
|
if best_spec < new_spec:
|
|
# Don't promote 1-byte to 3-byte when the existing 1-byte edge has no public_key:
|
|
# keep the 1-byte edge so we don't attribute its observations to one specific 3-byte node
|
|
# (other e0 repeaters would otherwise have no edges and appear removed).
|
|
best_is_1byte = len(best_key[0]) == 2 and len(best_key[1]) == 2
|
|
if best_is_1byte and not best_edge.get("from_public_key") and not best_edge.get("to_public_key"):
|
|
self._update_edge_data(
|
|
best_edge, now, hop_position,
|
|
from_public_key, to_public_key, geographic_distance, prefix_bytes,
|
|
)
|
|
self._persist_and_notify_edge(best_key, is_new_edge=False)
|
|
return
|
|
# Promote: remove old edge, add new at higher resolution with merged count
|
|
merged_count = best_edge["observation_count"] + 1
|
|
first_seen = best_edge.get("first_seen") or now
|
|
self._remove_edge_from_memory(best_key)
|
|
self._delete_edge_from_db(best_key)
|
|
with self.pending_lock:
|
|
self.pending_updates.discard(best_key)
|
|
|
|
self.edges[edge_key] = {
|
|
"from_prefix": from_prefix,
|
|
"to_prefix": to_prefix,
|
|
"from_public_key": from_public_key or best_edge.get("from_public_key"),
|
|
"to_public_key": to_public_key or best_edge.get("to_public_key"),
|
|
"observation_count": merged_count,
|
|
"first_seen": first_seen,
|
|
"last_seen": now,
|
|
# Weighted merge, matching the in-place update path below:
|
|
# the promoted edge carries merged_count observations, so
|
|
# overwriting the average with this single observation would
|
|
# throw away every earlier one and skew path scoring.
|
|
"avg_hop_position": _merge_avg_hop_position(
|
|
best_edge.get("avg_hop_position"),
|
|
best_edge["observation_count"],
|
|
hop_position,
|
|
),
|
|
"geographic_distance": geographic_distance if geographic_distance is not None else best_edge.get("geographic_distance"),
|
|
"confirmed_2byte": True if prefix_bytes == 2 else best_edge.get("confirmed_2byte", False),
|
|
}
|
|
self._index_add(from_prefix, to_prefix)
|
|
|
|
self._persist_and_notify_edge(edge_key, is_new_edge=True)
|
|
return
|
|
|
|
# Exact-key create or update (1-byte with 0 or 2+ matches, or 2/3-byte with no match)
|
|
if edge_key in self.edges:
|
|
edge = self.edges[edge_key]
|
|
edge['observation_count'] += 1
|
|
edge['last_seen'] = now
|
|
|
|
# Update average hop position
|
|
if hop_position is not None:
|
|
current_avg = edge.get('avg_hop_position')
|
|
count = edge['observation_count']
|
|
if current_avg is not None:
|
|
# Weighted average: (old_avg * (count-1) + new_pos) / count
|
|
edge['avg_hop_position'] = ((current_avg * (count - 1)) + hop_position) / count
|
|
else:
|
|
# First time setting hop position
|
|
edge['avg_hop_position'] = hop_position
|
|
|
|
# Update public keys if provided (always update if we have a better key)
|
|
# This allows us to fill in missing keys on existing edges
|
|
if from_public_key:
|
|
edge['from_public_key'] = from_public_key
|
|
if to_public_key:
|
|
edge['to_public_key'] = to_public_key
|
|
|
|
# Update geographic distance if provided
|
|
if geographic_distance is not None:
|
|
edge['geographic_distance'] = geographic_distance
|
|
|
|
# 2-byte trace confirmation (for path weighting when supported)
|
|
if prefix_bytes == 2:
|
|
edge['confirmed_2byte'] = True
|
|
|
|
is_new_edge = False
|
|
else:
|
|
# New edge — also update adjacency indexes
|
|
self.edges[edge_key] = {
|
|
'from_prefix': from_prefix,
|
|
'to_prefix': to_prefix,
|
|
'from_public_key': from_public_key,
|
|
'to_public_key': to_public_key,
|
|
'observation_count': 1,
|
|
'first_seen': now,
|
|
'last_seen': now,
|
|
'avg_hop_position': hop_position if hop_position is not None else None,
|
|
'geographic_distance': geographic_distance,
|
|
'confirmed_2byte': prefix_bytes == 2,
|
|
}
|
|
self._index_add(from_prefix, to_prefix)
|
|
is_new_edge = True
|
|
|
|
self._persist_and_notify_edge(edge_key, is_new_edge)
|
|
|
|
def _persist_and_notify_edge(self, edge_key: tuple[str, str], is_new_edge: bool) -> None:
|
|
"""Persist edge to DB (according to write strategy) and notify web viewer."""
|
|
self.logger.debug(f"Mesh graph: Edge {edge_key} - new={is_new_edge}, strategy={self.write_strategy}")
|
|
if self.write_strategy == 'immediate':
|
|
self._write_edge_to_db(edge_key, is_new_edge)
|
|
elif self.write_strategy == 'batched':
|
|
flush_needed = False
|
|
with self.pending_lock:
|
|
self.pending_updates.add(edge_key)
|
|
if len(self.pending_updates) >= self.batch_max_pending:
|
|
flush_needed = True
|
|
if flush_needed:
|
|
self._flush_pending_updates_sync()
|
|
elif self.write_strategy == 'hybrid':
|
|
if is_new_edge:
|
|
self._write_edge_to_db(edge_key, True)
|
|
else:
|
|
flush_needed = False
|
|
with self.pending_lock:
|
|
self.pending_updates.add(edge_key)
|
|
if len(self.pending_updates) >= self.batch_max_pending:
|
|
flush_needed = True
|
|
if flush_needed:
|
|
self._flush_pending_updates_sync()
|
|
self._notify_web_viewer_edge(edge_key, is_new_edge)
|
|
|
|
def _notify_web_viewer_edge(self, edge_key: tuple[str, str], is_new: bool):
|
|
"""Notify web viewer of edge update via bot integration.
|
|
|
|
New edges always trigger an immediate notification. Updates to existing
|
|
edges are throttled to at most once every 10 seconds to reduce HTTP
|
|
traffic on busy meshes.
|
|
"""
|
|
try:
|
|
if not hasattr(self.bot, 'web_viewer_integration') or not self.bot.web_viewer_integration:
|
|
return
|
|
|
|
if not hasattr(self.bot.web_viewer_integration, 'bot_integration'):
|
|
return
|
|
|
|
edge = self.edges.get(edge_key)
|
|
if not edge:
|
|
return
|
|
|
|
# Throttle repeated updates for the same edge (new edges always notify)
|
|
now_ts = time.time()
|
|
if not is_new:
|
|
last_notified = self._notification_timestamps.get(edge_key, 0.0)
|
|
if (now_ts - last_notified) < 10.0:
|
|
return # Skip — notified recently enough
|
|
self._notification_timestamps[edge_key] = now_ts
|
|
|
|
# Prepare edge data for web viewer
|
|
edge_data = {
|
|
'from_prefix': edge['from_prefix'],
|
|
'to_prefix': edge['to_prefix'],
|
|
'from_public_key': edge.get('from_public_key'),
|
|
'to_public_key': edge.get('to_public_key'),
|
|
'observation_count': edge['observation_count'],
|
|
'first_seen': edge['first_seen'].isoformat() if isinstance(edge['first_seen'], datetime) else str(edge['first_seen']),
|
|
'last_seen': edge['last_seen'].isoformat() if isinstance(edge['last_seen'], datetime) else str(edge['last_seen']),
|
|
'avg_hop_position': edge.get('avg_hop_position'),
|
|
'geographic_distance': edge.get('geographic_distance'),
|
|
'is_new': is_new
|
|
}
|
|
|
|
# Send update asynchronously
|
|
self.bot.web_viewer_integration.bot_integration.send_mesh_edge_update(edge_data)
|
|
except Exception as e:
|
|
self.logger.debug(f"Error notifying web viewer of edge update: {e}")
|
|
|
|
def _recalculate_distance_if_needed(
|
|
self,
|
|
edge: dict,
|
|
conn: Optional[sqlite3.Connection] = None,
|
|
location_cache: Optional[dict[str, tuple[float, float]]] = None,
|
|
) -> Optional[float]:
|
|
"""Recalculate geographic distance using full public keys if available.
|
|
|
|
This ensures we get the correct location when there are prefix collisions.
|
|
|
|
Args:
|
|
edge: Edge dictionary with prefix and optional public keys.
|
|
conn: Optional existing DB connection for batch operations.
|
|
location_cache: Optional cache for location lookups within a flush (keyed by pk: or prefix:).
|
|
|
|
Returns:
|
|
Optional[float]: Recalculated distance in km, or None if can't calculate.
|
|
"""
|
|
from .utils import calculate_distance
|
|
|
|
# Get location for 'from' node (conn optional for single-connection batch flush)
|
|
if edge.get('from_public_key'):
|
|
from_location = self._get_location_by_public_key(
|
|
edge['from_public_key'], conn=conn, location_cache=location_cache
|
|
)
|
|
else:
|
|
from_location = None
|
|
if not from_location:
|
|
to_location_temp = None
|
|
if edge.get('to_public_key'):
|
|
to_location_temp = self._get_location_by_public_key(
|
|
edge['to_public_key'], conn=conn, location_cache=location_cache
|
|
)
|
|
if not to_location_temp:
|
|
to_location_temp = self._get_location_by_prefix(
|
|
edge['to_prefix'], conn=conn, location_cache=location_cache
|
|
)
|
|
from_location = self._get_location_by_prefix(
|
|
edge['from_prefix'], to_location_temp, conn=conn, location_cache=location_cache
|
|
)
|
|
|
|
# Get location for 'to' node
|
|
if edge.get('to_public_key'):
|
|
to_location = self._get_location_by_public_key(
|
|
edge['to_public_key'], conn=conn, location_cache=location_cache
|
|
)
|
|
else:
|
|
to_location = None
|
|
if not to_location:
|
|
to_location = self._get_location_by_prefix(
|
|
edge['to_prefix'], from_location, conn=conn, location_cache=location_cache
|
|
)
|
|
|
|
# Calculate distance if we have both locations
|
|
if from_location and to_location:
|
|
return calculate_distance(
|
|
from_location[0], from_location[1],
|
|
to_location[0], to_location[1]
|
|
)
|
|
|
|
return None
|
|
|
|
def _get_location_by_public_key(
|
|
self,
|
|
public_key: str,
|
|
conn: Optional[sqlite3.Connection] = None,
|
|
location_cache: Optional[dict[str, tuple[float, float]]] = None,
|
|
) -> Optional[tuple[float, float]]:
|
|
"""Get location for a full public key (more accurate than prefix lookup).
|
|
|
|
Prefers starred repeaters if there are somehow multiple entries (shouldn't happen with full key).
|
|
"""
|
|
cache_key = f"pk:{public_key}" if location_cache is not None else None
|
|
if cache_key is not None and location_cache is not None and cache_key in location_cache:
|
|
return location_cache[cache_key]
|
|
try:
|
|
query = '''
|
|
SELECT latitude, longitude
|
|
FROM complete_contact_tracking
|
|
WHERE public_key = ?
|
|
AND latitude IS NOT NULL AND longitude IS NOT NULL
|
|
AND latitude != 0 AND longitude != 0
|
|
AND role IN ('repeater', 'roomserver')
|
|
ORDER BY is_starred DESC, COALESCE(last_advert_timestamp, last_heard) DESC
|
|
LIMIT 1
|
|
'''
|
|
if conn is not None:
|
|
results = self.db_manager.execute_query_on_connection(conn, query, (public_key,))
|
|
else:
|
|
results = self.db_manager.execute_query(query, (public_key,))
|
|
if results:
|
|
row = results[0]
|
|
lat = row.get('latitude')
|
|
lon = row.get('longitude')
|
|
if lat is not None and lon is not None:
|
|
result = (float(lat), float(lon))
|
|
if cache_key is not None and location_cache is not None:
|
|
location_cache[cache_key] = result
|
|
return result
|
|
except Exception as e:
|
|
self.logger.debug(f"Error getting location by public key {public_key[:16]}...: {e}")
|
|
return None
|
|
|
|
def _get_location_by_prefix(
|
|
self,
|
|
prefix: str,
|
|
reference_location: Optional[tuple[float, float]] = None,
|
|
conn: Optional[sqlite3.Connection] = None,
|
|
location_cache: Optional[dict[str, tuple[float, float]]] = None,
|
|
) -> Optional[tuple[float, float]]:
|
|
"""Get location for a prefix (fallback when full public key not available).
|
|
|
|
For LoRa networks, prefers shorter distances when there are prefix collisions,
|
|
as LoRa range is limited by the curve of the earth.
|
|
|
|
Args:
|
|
prefix: 2-character hex prefix.
|
|
reference_location: Optional (lat, lon) to calculate distance from for LoRa preference.
|
|
conn: Optional existing DB connection for batch operations.
|
|
location_cache: Optional cache for location lookups within a flush.
|
|
"""
|
|
if location_cache is not None:
|
|
if reference_location is not None:
|
|
cache_key = f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}"
|
|
else:
|
|
cache_key = f"prefix:{prefix}"
|
|
if cache_key in location_cache:
|
|
return location_cache[cache_key]
|
|
try:
|
|
prefix_pattern = f"{prefix}%"
|
|
|
|
# Get all candidates with locations
|
|
query = '''
|
|
SELECT latitude, longitude, is_starred,
|
|
COALESCE(last_advert_timestamp, last_heard) as last_seen
|
|
FROM complete_contact_tracking
|
|
WHERE public_key LIKE ?
|
|
AND latitude IS NOT NULL AND longitude IS NOT NULL
|
|
AND latitude != 0 AND longitude != 0
|
|
AND role IN ('repeater', 'roomserver')
|
|
'''
|
|
if conn is not None:
|
|
results = self.db_manager.execute_query_on_connection(conn, query, (prefix_pattern,))
|
|
else:
|
|
results = self.db_manager.execute_query(query, (prefix_pattern,))
|
|
|
|
if not results:
|
|
return None
|
|
|
|
# If we have a reference location, prefer shorter distances (LoRa range limitation)
|
|
if reference_location and len(results) > 1:
|
|
from .utils import calculate_distance
|
|
ref_lat, ref_lon = reference_location
|
|
|
|
# Calculate distances and sort by distance (shorter first)
|
|
candidates_with_distance = []
|
|
for row in results:
|
|
lat = row.get('latitude')
|
|
lon = row.get('longitude')
|
|
if lat is not None and lon is not None:
|
|
distance = calculate_distance(ref_lat, ref_lon, float(lat), float(lon))
|
|
is_starred = row.get('is_starred', False)
|
|
last_seen = row.get('last_seen', '')
|
|
candidates_with_distance.append((distance, is_starred, last_seen, row))
|
|
|
|
if candidates_with_distance:
|
|
# Sort by: starred first (False < True), then distance (shorter = better for LoRa), then recency
|
|
candidates_with_distance.sort(key=lambda x: (
|
|
not x[1], # Starred first (False < True, so starred=True comes before starred=False)
|
|
x[0], # Distance (shorter first)
|
|
x[2] if x[2] else '' # More recent first (newer timestamps sort later in string comparison)
|
|
))
|
|
|
|
# Get the best candidate
|
|
best_row = candidates_with_distance[0][3]
|
|
lat = best_row.get('latitude')
|
|
lon = best_row.get('longitude')
|
|
if lat is not None and lon is not None:
|
|
result = (float(lat), float(lon))
|
|
if location_cache is not None and reference_location is not None:
|
|
cache_key = f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}"
|
|
location_cache[cache_key] = result
|
|
return result
|
|
|
|
# No reference location or single result - use standard ordering
|
|
# Prefer starred, then most recent
|
|
results.sort(key=lambda x: (
|
|
not x.get('is_starred', False), # Starred first (False < True)
|
|
x.get('last_seen', '') if x.get('last_seen') else '' # More recent first
|
|
))
|
|
|
|
row = results[0]
|
|
lat = row.get('latitude')
|
|
lon = row.get('longitude')
|
|
if lat is not None and lon is not None:
|
|
result = (float(lat), float(lon))
|
|
if location_cache is not None:
|
|
cache_key = f"prefix:{prefix}" if reference_location is None else f"prefix:{prefix}:{reference_location[0]}:{reference_location[1]}"
|
|
location_cache[cache_key] = result
|
|
return result
|
|
except Exception as e:
|
|
self.logger.debug(f"Error getting location by prefix {prefix}: {e}")
|
|
return None
|
|
|
|
def _write_edge_to_db(
|
|
self,
|
|
edge_key: tuple[str, str],
|
|
is_new: bool,
|
|
conn: Optional[sqlite3.Connection] = None,
|
|
location_cache: Optional[dict[str, tuple[float, float]]] = None,
|
|
skip_distance_recalc: bool = False,
|
|
):
|
|
"""Write a single edge to the database.
|
|
|
|
Args:
|
|
edge_key: (from_prefix, to_prefix) tuple.
|
|
is_new: True if this is a new edge, False if updating existing.
|
|
conn: Optional existing DB connection for batch operations (caller commits).
|
|
location_cache: Optional cache for location lookups within a flush.
|
|
skip_distance_recalc: If True, skip distance recalculation (used by
|
|
_flush_pending_updates_sync which already recalculates before calling here).
|
|
"""
|
|
if edge_key not in self.edges:
|
|
return
|
|
|
|
edge = self.edges[edge_key]
|
|
|
|
# Recalculate distance using full public keys if available (more accurate).
|
|
# Skipped when called from the batch flush loop, which already recalculated.
|
|
if not skip_distance_recalc and (edge.get('from_public_key') or edge.get('to_public_key')):
|
|
recalculated_distance = self._recalculate_distance_if_needed(
|
|
edge, conn=conn, location_cache=location_cache
|
|
)
|
|
if recalculated_distance is not None:
|
|
edge['geographic_distance'] = recalculated_distance
|
|
self.logger.debug(f"Mesh graph: Recalculated distance for {edge_key} using public keys: {recalculated_distance:.1f} km")
|
|
|
|
try:
|
|
query: str
|
|
params: tuple[Any, ...]
|
|
if is_new:
|
|
# Upsert new edge.
|
|
# Use INSERT ... ON CONFLICT DO UPDATE so that if the row already exists
|
|
# in the database (e.g. it was filtered out of the in-memory graph at
|
|
# startup by startup_load_days / edge_expiration_days, or written by a
|
|
# concurrent process), we merge rather than fail with UNIQUE constraint.
|
|
query = '''
|
|
INSERT INTO mesh_connections
|
|
(from_prefix, to_prefix, from_public_key, to_public_key,
|
|
observation_count, first_seen, last_seen, avg_hop_position,
|
|
geographic_distance)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
|
|
ON CONFLICT(from_prefix, to_prefix) DO UPDATE SET
|
|
observation_count = MAX(observation_count, excluded.observation_count),
|
|
last_seen = MAX(last_seen, excluded.last_seen),
|
|
avg_hop_position = excluded.avg_hop_position,
|
|
geographic_distance = COALESCE(excluded.geographic_distance, geographic_distance),
|
|
from_public_key = COALESCE(excluded.from_public_key, from_public_key),
|
|
to_public_key = COALESCE(excluded.to_public_key, to_public_key)
|
|
'''
|
|
params = (
|
|
edge['from_prefix'],
|
|
edge['to_prefix'],
|
|
edge.get('from_public_key'),
|
|
edge.get('to_public_key'),
|
|
edge['observation_count'],
|
|
edge['first_seen'].isoformat() if isinstance(edge['first_seen'], datetime) else edge['first_seen'],
|
|
edge['last_seen'].isoformat() if isinstance(edge['last_seen'], datetime) else edge['last_seen'],
|
|
edge.get('avg_hop_position'),
|
|
edge.get('geographic_distance')
|
|
)
|
|
else:
|
|
# Update existing edge — recalculate distance if we now have public keys,
|
|
# but only when not already done by the caller (skip_distance_recalc=False).
|
|
current_distance = edge.get('geographic_distance')
|
|
if not skip_distance_recalc and (edge.get('from_public_key') or edge.get('to_public_key')) and current_distance:
|
|
recalculated = self._recalculate_distance_if_needed(
|
|
edge, conn=conn, location_cache=location_cache
|
|
)
|
|
if recalculated is not None:
|
|
# Update if recalculated distance is significantly different (more than 20% difference)
|
|
if abs(recalculated - current_distance) / max(current_distance, 1.0) > 0.2:
|
|
edge['geographic_distance'] = recalculated
|
|
self.logger.info(f"Mesh graph: Corrected distance for {edge_key}: {current_distance:.1f} -> {recalculated:.1f} km")
|
|
|
|
# Update existing edge
|
|
# Always update public keys if provided (allows filling in missing keys on existing edges)
|
|
from_key = edge.get('from_public_key')
|
|
to_key = edge.get('to_public_key')
|
|
query = self._MESH_EDGE_UPDATE_QUERY
|
|
params = (
|
|
edge['observation_count'],
|
|
edge['last_seen'].isoformat() if isinstance(edge['last_seen'], datetime) else edge['last_seen'],
|
|
edge.get('avg_hop_position'),
|
|
edge.get('geographic_distance'),
|
|
from_key, # First occurrence for CASE WHEN check
|
|
from_key, # Second occurrence for value assignment
|
|
to_key, # First occurrence for CASE WHEN check
|
|
to_key, # Second occurrence for value assignment
|
|
edge['from_prefix'],
|
|
edge['to_prefix']
|
|
)
|
|
|
|
if conn is not None:
|
|
rows_affected = self.db_manager.execute_update_on_connection(conn, query, params)
|
|
else:
|
|
rows_affected = self.db_manager.execute_update(query, params)
|
|
if rows_affected > 0:
|
|
self.logger.debug(f"Mesh graph: Successfully wrote edge {edge_key} to database ({'INSERT' if is_new else 'UPDATE'}, {rows_affected} rows)")
|
|
else:
|
|
self.logger.warning(f"Mesh graph: Edge write returned 0 rows affected for {edge_key}")
|
|
|
|
except Exception as e:
|
|
self.logger.warning(f"Error writing edge to database: {e}")
|
|
import traceback
|
|
self.logger.debug(traceback.format_exc())
|
|
|
|
# UPDATE statement used for both single-edge writes and batch executemany
|
|
_MESH_EDGE_UPDATE_QUERY = '''
|
|
UPDATE mesh_connections
|
|
SET observation_count = ?, last_seen = ?,
|
|
avg_hop_position = ?, geographic_distance = ?,
|
|
from_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE from_public_key END,
|
|
to_public_key = CASE WHEN ? IS NOT NULL THEN ? ELSE to_public_key END
|
|
WHERE from_prefix = ? AND to_prefix = ?
|
|
'''
|
|
|
|
def _build_update_params_for_edge(
|
|
self,
|
|
edge_key: tuple[str, str],
|
|
conn: Optional[sqlite3.Connection],
|
|
location_cache: Optional[dict[str, tuple[float, float]]],
|
|
) -> Optional[tuple]:
|
|
"""Build UPDATE params for an edge (for batch executemany). Returns None to skip."""
|
|
if edge_key not in self.edges:
|
|
return None
|
|
try:
|
|
edge = self.edges[edge_key]
|
|
# Recalculate distance if we have public keys (same logic as _write_edge_to_db)
|
|
if edge.get('from_public_key') or edge.get('to_public_key'):
|
|
recalculated_distance = self._recalculate_distance_if_needed(
|
|
edge, conn=conn, location_cache=location_cache
|
|
)
|
|
if recalculated_distance is not None:
|
|
edge['geographic_distance'] = recalculated_distance
|
|
current_distance = edge.get('geographic_distance')
|
|
if (edge.get('from_public_key') or edge.get('to_public_key')) and current_distance:
|
|
recalculated = self._recalculate_distance_if_needed(
|
|
edge, conn=conn, location_cache=location_cache
|
|
)
|
|
if recalculated is not None and abs(recalculated - current_distance) / max(current_distance, 1.0) > 0.2:
|
|
edge['geographic_distance'] = recalculated
|
|
from_key = edge.get('from_public_key')
|
|
to_key = edge.get('to_public_key')
|
|
last_seen = edge['last_seen']
|
|
if isinstance(last_seen, datetime):
|
|
last_seen = last_seen.isoformat()
|
|
return (
|
|
edge['observation_count'],
|
|
last_seen,
|
|
edge.get('avg_hop_position'),
|
|
edge.get('geographic_distance'),
|
|
from_key,
|
|
from_key,
|
|
to_key,
|
|
to_key,
|
|
edge['from_prefix'],
|
|
edge['to_prefix'],
|
|
)
|
|
except Exception as e:
|
|
self.logger.debug(f"Error building update params for {edge_key}: {e}")
|
|
return None
|
|
|
|
def prune_expired_edges(self) -> int:
|
|
"""Remove edges from the in-memory graph that have exceeded graph_edge_expiration_days.
|
|
|
|
Only evicts from RAM — the database rows are kept so that historical data is
|
|
preserved and can be reloaded if the expiration window is later widened.
|
|
|
|
Returns:
|
|
int: Number of edges evicted.
|
|
"""
|
|
if self.edge_expiration_days <= 0:
|
|
return 0
|
|
|
|
cutoff = datetime.now() - timedelta(days=self.edge_expiration_days)
|
|
expired_keys = []
|
|
for edge_key, edge in self.edges.items():
|
|
last_seen = edge.get('last_seen')
|
|
if last_seen is None:
|
|
continue
|
|
if isinstance(last_seen, str):
|
|
try:
|
|
last_seen = datetime.fromisoformat(last_seen.replace('Z', '+00:00'))
|
|
except ValueError:
|
|
continue
|
|
if last_seen < cutoff:
|
|
expired_keys.append(edge_key)
|
|
|
|
for edge_key in expired_keys:
|
|
from_prefix, to_prefix = edge_key
|
|
del self.edges[edge_key]
|
|
# Clean up adjacency indexes
|
|
self._index_remove(from_prefix, to_prefix)
|
|
# Drop stale notification timestamp if present
|
|
self._notification_timestamps.pop(edge_key, None)
|
|
|
|
if expired_keys:
|
|
self.logger.debug(f"Pruned {len(expired_keys)} expired graph edges (older than {self.edge_expiration_days} days)")
|
|
return len(expired_keys)
|
|
|
|
def delete_expired_edges_from_db(self, days: int) -> int:
|
|
"""Delete mesh_connections rows older than the given days.
|
|
Keeps the on-disk table aligned with in-memory pruning and prevents unbounded growth.
|
|
Called from the scheduler (e.g. daily). Use Data_Retention mesh_connections_retention_days
|
|
or Path_Command graph_edge_expiration_days.
|
|
Returns:
|
|
int: Number of rows deleted.
|
|
"""
|
|
if days <= 0:
|
|
return 0
|
|
try:
|
|
deleted = self.db_manager.execute_update(
|
|
"DELETE FROM mesh_connections WHERE last_seen < datetime('now', ?)",
|
|
(f'-{days} days',)
|
|
)
|
|
if deleted > 0:
|
|
self.logger.info(f"Cleaned up {deleted} old mesh_connections entries (older than {days} days)")
|
|
return deleted
|
|
except Exception as e:
|
|
self.logger.error(f"Error cleaning up mesh_connections: {e}")
|
|
return 0
|
|
|
|
def _start_batch_writer(self):
|
|
"""Start background task for batched writes."""
|
|
def batch_writer_loop():
|
|
while not self._shutdown_event.is_set():
|
|
self._shutdown_event.wait(self.batch_interval)
|
|
if not self._shutdown_event.is_set():
|
|
# Flush synchronously (database operations are synchronous)
|
|
self._flush_pending_updates_sync()
|
|
# Periodically evict expired edges from RAM
|
|
self.prune_expired_edges()
|
|
|
|
import threading
|
|
batch_thread = threading.Thread(target=batch_writer_loop, daemon=True)
|
|
batch_thread.start()
|
|
self._batch_thread = batch_thread
|
|
|
|
def _flush_pending_updates_sync(self):
|
|
"""Flush all pending edge updates to database (synchronous version).
|
|
Uses a single connection for the entire batch to avoid 'unable to open database file'
|
|
when many edges are written in quick succession.
|
|
Handles both new edges (INSERT) and existing edges (UPDATE).
|
|
"""
|
|
with self.pending_lock:
|
|
if not self.pending_updates:
|
|
return
|
|
|
|
updates = list(self.pending_updates)
|
|
self.pending_updates.clear()
|
|
|
|
location_cache: dict[str, tuple[float, float]] = {}
|
|
try:
|
|
with self.db_manager.connection() as conn:
|
|
cursor = conn.cursor()
|
|
for edge_key in updates:
|
|
if edge_key not in self.edges:
|
|
continue
|
|
edge = self.edges[edge_key]
|
|
# Recalculate distance if we have public keys
|
|
if edge.get('from_public_key') or edge.get('to_public_key'):
|
|
recalculated = self._recalculate_distance_if_needed(
|
|
edge, conn=conn, location_cache=location_cache
|
|
)
|
|
if recalculated is not None:
|
|
edge['geographic_distance'] = recalculated
|
|
# Check if edge exists in DB
|
|
cursor.execute(
|
|
'SELECT 1 FROM mesh_connections WHERE from_prefix = ? AND to_prefix = ?',
|
|
(edge_key[0], edge_key[1]),
|
|
)
|
|
is_new = cursor.fetchone() is None
|
|
# Distance was already recalculated above — tell _write_edge_to_db to skip it
|
|
self._write_edge_to_db(edge_key, is_new, conn=conn, location_cache=location_cache,
|
|
skip_distance_recalc=True)
|
|
conn.commit()
|
|
except Exception as e:
|
|
self.logger.warning(f"Error flushing graph updates: {e}")
|
|
# Connection already closed by context manager; rollback happened on exit if needed
|
|
|
|
if updates:
|
|
self.logger.debug(f"Flushed {len(updates)} pending graph edge updates")
|
|
|
|
async def _flush_pending_updates(self):
|
|
"""Flush all pending edge updates to database (async wrapper)."""
|
|
self._flush_pending_updates_sync()
|
|
|
|
def has_edge(self, from_prefix: str, to_prefix: str) -> bool:
|
|
"""Check if an edge exists in the graph (exact or prefix match).
|
|
|
|
Args:
|
|
from_prefix: Source node prefix.
|
|
to_prefix: Destination node prefix.
|
|
|
|
Returns:
|
|
bool: True if edge exists.
|
|
"""
|
|
return self.get_edge(from_prefix, to_prefix) is not None
|
|
|
|
def get_edge(self, from_prefix: str, to_prefix: str) -> Optional[dict]:
|
|
"""Get edge data if it exists (exact key first, then prefix match).
|
|
|
|
Args:
|
|
from_prefix: Source node prefix.
|
|
to_prefix: Destination node prefix.
|
|
|
|
Returns:
|
|
Dict with edge data or None if not found.
|
|
"""
|
|
from_norm = from_prefix.lower().strip() if from_prefix else ""
|
|
to_norm = to_prefix.lower().strip() if to_prefix else ""
|
|
if not from_norm or not to_norm:
|
|
return None
|
|
# Exact key first
|
|
exact = self.edges.get((from_norm, to_norm))
|
|
if exact is not None:
|
|
return exact
|
|
return self._get_edge_by_prefix_match(from_norm, to_norm)
|
|
|
|
def get_outgoing_edges(self, prefix: str) -> list[dict]:
|
|
"""Get all edges originating from a node (prefix match: returns edges where from_prefix matches prefix).
|
|
|
|
Args:
|
|
prefix: Node prefix (2, 4, or 6 hex chars).
|
|
|
|
Returns:
|
|
List of edge dictionaries.
|
|
"""
|
|
prefix = prefix.lower().strip() if prefix else ""
|
|
if not prefix:
|
|
return []
|
|
result = []
|
|
# Only from-nodes sharing prefix's first byte can prefix-match it, so scan just that bucket
|
|
# of the adjacency index rather than every from-node. O(bucket + matched); identical edge
|
|
# set (indexes mirror self.edges).
|
|
for from_p in self._outgoing_byte0.get(prefix[:2], ()):
|
|
if self._prefix_match(from_p, prefix):
|
|
for to_p in self._outgoing_index[from_p]:
|
|
edge = self.edges.get((from_p, to_p))
|
|
if edge is not None:
|
|
result.append(edge)
|
|
return result
|
|
|
|
def get_incoming_edges(self, prefix: str) -> list[dict]:
|
|
"""Get all edges ending at a node (prefix match: returns edges where to_prefix matches prefix).
|
|
|
|
Args:
|
|
prefix: Node prefix (2, 4, or 6 hex chars).
|
|
|
|
Returns:
|
|
List of edge dictionaries.
|
|
"""
|
|
prefix = prefix.lower().strip() if prefix else ""
|
|
if not prefix:
|
|
return []
|
|
result = []
|
|
# Only to-nodes sharing prefix's first byte can prefix-match it, so scan just that bucket
|
|
# of the adjacency index rather than every to-node. O(bucket + matched).
|
|
for to_p in self._incoming_byte0.get(prefix[:2], ()):
|
|
if self._prefix_match(to_p, prefix):
|
|
for from_p in self._incoming_index[to_p]:
|
|
edge = self.edges.get((from_p, to_p))
|
|
if edge is not None:
|
|
result.append(edge)
|
|
return result
|
|
|
|
def validate_path_segment(self, from_prefix: str, to_prefix: str,
|
|
min_observations: int = 1,
|
|
check_bidirectional: bool = False) -> tuple[bool, float]:
|
|
"""Validate a path segment using graph data.
|
|
|
|
Args:
|
|
from_prefix: Source node prefix.
|
|
to_prefix: Destination node prefix.
|
|
min_observations: Minimum observations required for confidence.
|
|
check_bidirectional: If True, check if reverse edge exists and boost confidence.
|
|
|
|
Returns:
|
|
Tuple of (is_valid, confidence_score) where confidence is 0.0-1.0.
|
|
"""
|
|
edge = self.get_edge(from_prefix, to_prefix)
|
|
|
|
if not edge:
|
|
return (False, 0.0)
|
|
|
|
if edge['observation_count'] < min_observations:
|
|
return (False, 0.0)
|
|
|
|
# Confidence based on observation count and recency
|
|
obs_count = edge['observation_count']
|
|
last_seen = edge['last_seen']
|
|
|
|
if isinstance(last_seen, str):
|
|
last_seen = datetime.fromisoformat(last_seen.replace('Z', '+00:00'))
|
|
|
|
hours_ago = (datetime.now() - last_seen).total_seconds() / 3600.0
|
|
|
|
# Observation count confidence (logarithmic scale)
|
|
obs_confidence = min(1.0, 0.3 + (0.7 * (1.0 - 1.0 / (1.0 + obs_count / 10.0))))
|
|
|
|
# Recency confidence (exponential decay, 48 hour half-life for longer advert intervals)
|
|
recency_confidence = 1.0 if hours_ago < 1 else max(0.0, 2.0 ** (-hours_ago / 48.0))
|
|
|
|
# Combined confidence
|
|
confidence = (obs_confidence * 0.6) + (recency_confidence * 0.4)
|
|
|
|
# Bidirectional edge bonus
|
|
if check_bidirectional:
|
|
reverse_edge = self.get_edge(to_prefix, from_prefix)
|
|
if reverse_edge and reverse_edge['observation_count'] >= min_observations:
|
|
# Bidirectional connection is more reliable
|
|
confidence = min(1.0, confidence + 0.15)
|
|
|
|
return (True, confidence)
|
|
|
|
def validate_path(self, path_nodes: list[str], min_observations: int = 1) -> tuple[bool, float]:
|
|
"""Validate an entire path using graph data.
|
|
|
|
Args:
|
|
path_nodes: List of node prefixes in path order.
|
|
min_observations: Minimum observations required per edge.
|
|
|
|
Returns:
|
|
Tuple of (is_valid, average_confidence).
|
|
"""
|
|
if len(path_nodes) < 2:
|
|
return (True, 1.0) # Single node or empty path is always valid
|
|
|
|
validations = []
|
|
for i in range(len(path_nodes) - 1):
|
|
from_node = path_nodes[i]
|
|
to_node = path_nodes[i + 1]
|
|
is_valid, confidence = self.validate_path_segment(from_node, to_node, min_observations)
|
|
|
|
if not is_valid:
|
|
return (False, 0.0)
|
|
|
|
validations.append(confidence)
|
|
|
|
# Return average confidence
|
|
avg_confidence = sum(validations) / len(validations) if validations else 0.0
|
|
return (True, avg_confidence)
|
|
|
|
def get_candidate_score(self, candidate_prefix: str, prev_prefix: Optional[str],
|
|
next_prefix: Optional[str], min_observations: int = 1,
|
|
hop_position: Optional[int] = None,
|
|
use_bidirectional: bool = True,
|
|
use_hop_position: bool = True) -> float:
|
|
"""Get graph-based score for a candidate node in a path.
|
|
|
|
Args:
|
|
candidate_prefix: The candidate node prefix.
|
|
prev_prefix: Previous node in path (if available).
|
|
next_prefix: Next node in path (if available).
|
|
min_observations: Minimum observations required.
|
|
hop_position: Current position in path (0-based index) for hop position validation.
|
|
use_bidirectional: If True, check bidirectional edges for higher confidence.
|
|
use_hop_position: If True, validate against avg_hop_position if available.
|
|
|
|
Returns:
|
|
Score from 0.0 to 1.0 based on graph evidence.
|
|
"""
|
|
score = 0.0
|
|
evidence_count = 0
|
|
scores = []
|
|
|
|
# Check edge from previous node
|
|
if prev_prefix:
|
|
is_valid, confidence = self.validate_path_segment(
|
|
prev_prefix, candidate_prefix, min_observations,
|
|
check_bidirectional=use_bidirectional
|
|
)
|
|
if is_valid:
|
|
scores.append(confidence)
|
|
score += confidence
|
|
evidence_count += 1
|
|
|
|
# Check edge to next node
|
|
if next_prefix:
|
|
is_valid, confidence = self.validate_path_segment(
|
|
candidate_prefix, next_prefix, min_observations,
|
|
check_bidirectional=use_bidirectional
|
|
)
|
|
if is_valid:
|
|
scores.append(confidence)
|
|
score += confidence
|
|
evidence_count += 1
|
|
|
|
if evidence_count == 0:
|
|
return 0.0
|
|
|
|
# Calculate base score as average
|
|
base_score = score / evidence_count
|
|
|
|
# Hop position validation bonus
|
|
if use_hop_position and hop_position is not None:
|
|
# Check if candidate appears in expected position based on avg_hop_position
|
|
# Check both incoming and outgoing edges for hop position data
|
|
hop_position_match = False
|
|
|
|
if prev_prefix:
|
|
edge = self.get_edge(prev_prefix, candidate_prefix)
|
|
if edge and edge.get('avg_hop_position') is not None:
|
|
# Allow some tolerance (within 0.5 of expected position)
|
|
expected_pos = edge['avg_hop_position']
|
|
if abs(hop_position - expected_pos) <= 0.5:
|
|
hop_position_match = True
|
|
|
|
if not hop_position_match and next_prefix:
|
|
edge = self.get_edge(candidate_prefix, next_prefix)
|
|
if edge and edge.get('avg_hop_position') is not None:
|
|
# For outgoing edge, expected position is one less (since it's the from node)
|
|
expected_pos = edge['avg_hop_position'] - 1
|
|
if abs(hop_position - expected_pos) <= 0.5:
|
|
hop_position_match = True
|
|
|
|
if hop_position_match:
|
|
base_score = min(1.0, base_score + 0.1)
|
|
|
|
# Geographic distance validation (if available)
|
|
# Use stored geographic_distance from edges when available (more accurate)
|
|
if prev_prefix or next_prefix:
|
|
# Check if we have geographic distance data that suggests reasonable routing
|
|
# This is informational - we don't heavily penalize based on distance alone
|
|
# but can use it as a tie-breaker
|
|
geographic_available = False
|
|
if prev_prefix:
|
|
edge = self.get_edge(prev_prefix, candidate_prefix)
|
|
if edge and edge.get('geographic_distance') is not None:
|
|
geographic_available = True
|
|
if not geographic_available and next_prefix:
|
|
edge = self.get_edge(candidate_prefix, next_prefix)
|
|
if edge and edge.get('geographic_distance') is not None:
|
|
geographic_available = True
|
|
|
|
# Having geographic data increases confidence slightly (indicates well-tracked edge)
|
|
if geographic_available:
|
|
base_score = min(1.0, base_score + 0.05)
|
|
|
|
return base_score
|
|
|
|
def find_intermediate_nodes(self, from_prefix: str, to_prefix: str,
|
|
min_observations: int = 1,
|
|
max_hops: int = 2) -> list[tuple[str, float]]:
|
|
"""Find intermediate nodes that connect from_prefix to to_prefix.
|
|
|
|
Uses multi-hop path inference to find nodes that connect two prefixes
|
|
when a direct edge may not exist or have low confidence.
|
|
|
|
Args:
|
|
from_prefix: Source node prefix.
|
|
to_prefix: Destination node prefix.
|
|
min_observations: Minimum observations required per edge.
|
|
max_hops: Maximum number of hops to search (default: 2, fallback to 3).
|
|
|
|
Returns:
|
|
List of (candidate_prefix, score) tuples sorted by score (highest first).
|
|
Score is 0.0-1.0 based on path strength.
|
|
"""
|
|
from_prefix = from_prefix.lower().strip() if from_prefix else ""
|
|
to_prefix = to_prefix.lower().strip() if to_prefix else ""
|
|
if not from_prefix or not to_prefix:
|
|
return []
|
|
|
|
candidates: dict[str, float] = {}
|
|
|
|
# Try 2-hop paths first: from_prefix -> intermediate -> to_prefix
|
|
outgoing_edges = self.get_outgoing_edges(from_prefix)
|
|
|
|
for edge in outgoing_edges:
|
|
intermediate_prefix = edge['to_prefix']
|
|
|
|
# Skip if this is the destination (direct edge case)
|
|
if intermediate_prefix == to_prefix:
|
|
continue
|
|
|
|
# Check if intermediate connects to destination
|
|
to_edge = self.get_edge(intermediate_prefix, to_prefix)
|
|
if not to_edge or to_edge['observation_count'] < min_observations:
|
|
continue
|
|
|
|
# Validate both edges
|
|
from_valid, from_confidence = self.validate_path_segment(
|
|
from_prefix, intermediate_prefix, min_observations,
|
|
check_bidirectional=True
|
|
)
|
|
to_valid, to_confidence = self.validate_path_segment(
|
|
intermediate_prefix, to_prefix, min_observations,
|
|
check_bidirectional=True
|
|
)
|
|
|
|
if from_valid and to_valid:
|
|
# Score is minimum of both edges (weakest link)
|
|
path_score = min(from_confidence, to_confidence)
|
|
|
|
# Bidirectional path bonus
|
|
reverse_from = self.get_edge(intermediate_prefix, from_prefix)
|
|
reverse_to = self.get_edge(to_prefix, intermediate_prefix)
|
|
bidirectional_bonus = 1.0
|
|
if reverse_from and reverse_from['observation_count'] >= min_observations:
|
|
if reverse_to and reverse_to['observation_count'] >= min_observations:
|
|
# Both edges are bidirectional - strong evidence
|
|
bidirectional_bonus = 1.2
|
|
else:
|
|
bidirectional_bonus = 1.1
|
|
elif reverse_to and reverse_to['observation_count'] >= min_observations:
|
|
bidirectional_bonus = 1.1
|
|
|
|
path_score = min(1.0, path_score * bidirectional_bonus)
|
|
|
|
# Use best score if we've seen this candidate before
|
|
if intermediate_prefix not in candidates or path_score > candidates[intermediate_prefix]:
|
|
candidates[intermediate_prefix] = path_score
|
|
|
|
# If no 2-hop paths found and max_hops >= 3, try 3-hop paths
|
|
if not candidates and max_hops >= 3:
|
|
# Find 3-hop paths: from_prefix -> intermediate1 -> intermediate2 -> to_prefix
|
|
for edge1 in outgoing_edges:
|
|
intermediate1 = edge1['to_prefix']
|
|
if intermediate1 == to_prefix:
|
|
continue
|
|
|
|
# Get edges from intermediate1
|
|
intermediate1_edges = self.get_outgoing_edges(intermediate1)
|
|
|
|
for edge2 in intermediate1_edges:
|
|
intermediate2 = edge2['to_prefix']
|
|
if intermediate2 in (from_prefix, intermediate1):
|
|
continue
|
|
|
|
# Check if intermediate2 connects to destination
|
|
to_edge = self.get_edge(intermediate2, to_prefix)
|
|
if not to_edge or to_edge['observation_count'] < min_observations:
|
|
continue
|
|
|
|
# Validate all three edges
|
|
valid1, conf1 = self.validate_path_segment(
|
|
from_prefix, intermediate1, min_observations
|
|
)
|
|
valid2, conf2 = self.validate_path_segment(
|
|
intermediate1, intermediate2, min_observations
|
|
)
|
|
valid3, conf3 = self.validate_path_segment(
|
|
intermediate2, to_prefix, min_observations
|
|
)
|
|
|
|
if valid1 and valid2 and valid3:
|
|
# Score is minimum of all three edges
|
|
path_score = min(conf1, conf2, conf3)
|
|
|
|
# 3-hop paths are less reliable, so reduce score
|
|
path_score *= 0.8
|
|
|
|
# Use intermediate2 as candidate (the one before destination)
|
|
if intermediate2 not in candidates or path_score > candidates[intermediate2]:
|
|
candidates[intermediate2] = path_score
|
|
|
|
# Sort by score (highest first) and return
|
|
sorted_candidates = sorted(candidates.items(), key=lambda x: x[1], reverse=True)
|
|
return sorted_candidates
|
|
|
|
def shutdown(self):
|
|
"""Shutdown graph, flushing all pending writes."""
|
|
# Do not log here: atexit may run after the logger's stream is closed.
|
|
# Signal shutdown
|
|
self._shutdown_event.set()
|
|
|
|
# Flush pending updates
|
|
try:
|
|
self._flush_pending_updates_sync()
|
|
except Exception:
|
|
pass # Avoid logging; stream may be closed during atexit
|