feat(neighbors): zero-hop neighbour discovery in packet capture

Port the observer firmware's neighbours feature into the bot's packet capture
service, by way of meshcore-packet-capture (upstream PRs #42/#43). On a long
interval the bot asks which repeaters it hears directly and records each
confirmed link with its measured SNR.

This is the strongest link evidence the bot collects: a first-party RF
measurement between two full 32-byte public keys. Path inference works from
1-3 byte prefixes with no keys, and complete_contact_tracking.hop_count
over-claims zero-hop (800 claimed vs 68 corroborated on the live database).

modules/neighbors_discovery.py keeps upstream's public names so its fixes and
tests stay portable. Two deliberate divergences:

- No command_lock plumbing. _SerializedCommands in modules/core.py already
  serialises and paces every radio command, strictly more than upstream's
  reentrant lock did.
- neighbors_collect_scopes defaults off. Upstream's zero-hop scope probe
  relies on a neighbour not being a known contact; this bot tracks contacts,
  and for a repeater with no stored path the library reaches zero-hop by
  calling change_contact_path() then reset_path() -- mutating the device's
  contact table per neighbour. Scope requests also hold the radio lock for
  their whole round trip (~25s), stalling bot replies. The default cycle is
  one command plus a passive listen window, during which the bot stays
  responsive.

Evidence lands in neighbor_links and neighbor_observations (migration 22)
rather than mesh_connections, which cannot persist provenance. The viewer
exposes it as evidence=neighbors on /api/mesh/edges and a Neighbours Only
mode on the mesh page, with populated public keys and real SNR; confirmed
neighbours also relabel edges in the combined view and count as
provenance-trusted when framing the initial map.

neighbors_enabled is the single switch. Every enabled broker publishes once
it is on (mqttN_neighbors defaults true; set false to hold one back). The
topic derives from each broker's packets topic with the last segment swapped,
so a templated broker gets meshcore/{IATA}/{PUBLIC_KEY}/neighbors -- the
topic the firmware uses -- instead of an unrelated flat one. A derived
location-routed topic is skipped with a warning when no iata is set, rather
than publishing into meshcore/XYZ/... on a shared namespace. Snapshots are
non-retained: heard_secs_ago is relative to publish time, so a retained copy
would read as current days later.

Also adds a DM-gated `neighbors` command (the 12h interval floor makes
waiting for the scheduler impractical), which acks immediately and reports in
a second message once the window closes.

Requires meshcore >= 2.3.8 for send_node_discover_req / req_regions_sync.
This commit is contained in:
agessaman
2026-08-04 19:24:18 -07:00
parent 7882148f90
commit ebc66992dd
20 changed files with 3755 additions and 16 deletions
+42
View File
@@ -8,6 +8,48 @@ semantic versioning.
### Added
- Zero-hop neighbour discovery in the packet capture service, ported from
`meshcore-packet-capture` (itself a port of the observer firmware's neighbours
feature). On a long interval (12336 h, default 24) the bot asks which
repeaters it hears **directly** and records each confirmed link with its
measured SNR. `[PacketCapture] neighbors_enabled` is the single switch and is
off by default; every enabled broker publishes the snapshot once it is on
(`mqttN_neighbors` defaults true, so set it false to hold a broker back). The
neighbours topic is derived from each broker's packets topic by swapping the
last segment, so a templated broker gets
`meshcore/{IATA}/{PUBLIC_KEY}/neighbors` — the topic the firmware uses. A
derived location-routed topic is skipped with a warning when no `iata` is set,
rather than publishing into `meshcore/XYZ/...`. Snapshots are non-retained,
because `heard_secs_ago` is relative to publish time.
- Confirmed direct links are now the strongest evidence class in the database:
two full 32-byte public keys plus a first-party RF measurement, where path
inference has only 13 byte prefixes and no keys. Stored in `neighbor_links`
(adjacency, migration 22) and `neighbor_observations` (per-cycle history,
pruned by `neighbor_observations_retention_days`, default 365).
- Mesh graph integration: a **Neighbours Only** evidence mode on the mesh page
and `GET /api/mesh/edges?evidence=neighbors`, deriving edges purely from
`neighbor_links`. Unlike the multi-byte mode these edges carry populated public
keys and real SNR, and they render as heavier lines. Confirmed neighbours are
also labelled as such in the combined view, and count as provenance-trusted
when framing the initial map. `neighbors_feed_mesh_graph` (default on) also
writes them to `mesh_connections`.
- `neighbors` DM command to run one cycle on demand — the scheduled interval has
a 12 h floor, which makes testing impractical otherwise. Acknowledges
immediately and reports in a second DM once the listen window closes. Worth
adding to `[Admin_ACL] admin_commands`, since a cycle spends airtime.
- Optional region-scope collection (`neighbors_collect_scopes`), **off by
default**: each request holds the bot's single radio command lock for up to
~25 s, and for a repeater with no stored path the meshcore library reaches
zero-hop by temporarily rewriting that contact's path on the device. The
default cycle costs one radio command plus a passive listen window, during
which the bot stays fully responsive.
### Changed
- `meshcore` minimum raised from 2.3.6 to 2.3.8. Required for
`send_node_discover_req` / `req_regions_sync`, and for the bounded, serialised
BLE write.
- Rebuilt web-viewer dashboard, served from a background snapshot instead of
recomputing statistics on every request. A refresher thread in the viewer
process writes `daily_rollup` (one row per local date) and
+88 -2
View File
@@ -350,7 +350,8 @@ admin_pubkeys =
# These commands will only work for users in the admin_pubkeys list
# reload: Reload configuration without restarting (radio settings cannot be changed)
# channelpause: DM-only; channelpause / channelresume — pause or resume bot responses on channels (not persisted)
admin_commands = repeater,webviewer,reload,channelpause
# neighbors: DM-only; runs a zero-hop neighbour discovery cycle (spends airtime — admin-gate this)
admin_commands = repeater,webviewer,reload,channelpause,neighbors
[Plugin_Overrides]
# Plugin Overrides - Use alternative plugin implementations
# Format: command_name = alternative_file_name
@@ -954,6 +955,12 @@ purging_log_retention_days = 90
#
# Mesh connections (path graph edges). Should be >= Path_Command graph_edge_expiration_days.
mesh_connections_retention_days = 7
#
# Zero-hop neighbour observation history (see [PacketCapture] neighbors_enabled).
# At most one row per neighbour per cycle, and cycles are >= 12h apart, so this
# grows very slowly and a long window is cheap. The neighbor_links aggregate that
# the mesh graph reads is never pruned.
neighbor_observations_retention_days = 365
[Path_Command]
# Enable or disable the path command
enabled = true
@@ -1466,6 +1473,13 @@ enabled = true
[Advert_Command]
enabled = true
# channels =
[Neighbors_Command]
# Runs one zero-hop neighbour discovery cycle on demand (DM only, 15min cooldown).
# The scheduled interval has a 12h floor, so this is how you test a change or
# refresh after moving the node. Requires [PacketCapture] neighbors_enabled = true.
# Worth adding to Admin_ACL admin_commands, since a cycle spends airtime.
enabled = true
# channels =
[Test_Command]
enabled = true
# Require minimum path byte length before responding to test/t
@@ -1753,9 +1767,70 @@ log_max_bytes = 50MB
log_rotation_when = midnight
log_backup_count = 5
# --- Neighbour discovery (zero-hop) ---
# Periodically asks which repeaters this node can hear DIRECTLY, then records each
# confirmed link with its measured SNR. This is the strongest link evidence the bot
# has: two full 32-byte public keys and a first-party RF measurement, as opposed to
# path inference over 1-3 byte prefixes.
#
# Results go to the neighbor_links / neighbor_observations tables, feed the mesh
# graph, and are published to every enabled broker (mqttN_neighbors defaults true;
# set it false on a broker you want to hold back).
# Trigger one on demand with the `neighbors` DM command (see [Neighbors_Command]).
#
# This is the single switch for the whole feature. Off by default because a cycle
# spends real airtime; turning it on needs no other change.
neighbors_enabled = false
# How often a cycle runs, in hours. Clamped to 12-336 to match the firmware; an
# out-of-range value is clamped with a warning rather than silently honoured.
neighbors_interval_hours = 24
# Seconds spent listening for responses. Repeaters answer after a randomised delay,
# so a short window simply finds fewer neighbours (floor: 5s). The bot stays fully
# responsive during this window — it is a passive listen, not a busy wait.
neighbors_discover_window = 60
# Cap on neighbours recorded per cycle, most useful first (most recently heard,
# then strongest SNR).
neighbors_max = 32
# Also add each confirmed direct link to the mesh graph as an edge.
neighbors_feed_mesh_graph = true
# Also ask each neighbour for its region scopes. OFF BY DEFAULT, and worth
# understanding before enabling:
# - Every bot radio command is serialised, and a scope request waits for its
# reply while holding that lock — up to ~25s per neighbour. With many
# neighbours the bot's own replies stall in bursts for minutes.
# - The zero-hop probe assumes the neighbour is not a known contact. This bot
# does track contacts, and for a repeater with no stored path the meshcore
# library reaches zero-hop by temporarily rewriting that contact's path on the
# device, then restoring it.
# With this off, the snapshot reports every neighbour it heard with empty scopes.
neighbors_collect_scopes = false
# Cap on the discover request itself; a stalled BLE/serial write can otherwise
# block far longer than the library's own reply timeout.
neighbors_command_timeout = 20
# Scope-request tuning (only used when neighbors_collect_scopes = true)
# neighbors_scope_timeout: per-request wait; 0 = use the device's own airtime estimate
# neighbors_scope_min_timeout: floor under the device-suggested timeout
# neighbors_scope_gap: settle delay between requests (they must not overlap on air)
# neighbors_cycle_timeout: overall budget for the pass; unreached neighbours report timeout
neighbors_scope_timeout = 0
neighbors_scope_min_timeout = 8
neighbors_scope_gap = 2.0
neighbors_cycle_timeout = 600
# Override for this node's own "self.scopes" value in the published payload.
# Empty asks the device for its default flood scope.
neighbors_self_scopes =
# Owner information (for packet analyzer registration)
# Owner public key (64-character hex string)
owner_public_key =
owner_public_key =
# Owner email address
owner_email =
@@ -1803,6 +1878,17 @@ iata = XYZ
# mqttN_client_id = # MQTT client ID (optional, auto-generated from bot name)
# mqttN_upload_packet_types = # Comma-separated packet types to upload (e.g. 2,4); empty = all
# mqttN_include_decoded = true/false # Publish the decoded object to this broker (default: include_decoded)
# mqttN_neighbors = true/false # Publish the zero-hop neighbours snapshot to this broker.
# # Default TRUE — neighbors_enabled below is the single
# # switch, so nothing is sent until you turn that on.
# # Set false to hold just this broker back.
# mqttN_topic_neighbors = # Neighbours topic template. Defaults to this broker's
# # packets topic with its last segment swapped for
# # "neighbors" (so a templated packets topic yields
# # meshcore/{IATA}/{PUBLIC_KEY}/neighbors), else
# # <mqttN_topic_prefix>/neighbors. A derived
# # location-routed topic is skipped with a warning when
# # no iata is set, rather than publishing to XYZ.
#
# Topic template placeholders:
# {IATA} - Uppercase IATA code (e.g., SEA)
+121
View File
@@ -326,6 +326,127 @@ Tokens are valid for 24 hours and auto-renewed. The service tries on-device sign
---
## Neighbour Discovery (zero-hop)
Periodically asks which repeaters this node can hear **directly**, and records each
confirmed link with its measured SNR. Ported from the observer firmware's neighbours
feature by way of `meshcore-packet-capture`. **Off by default.**
This is the strongest link evidence the bot collects. Everything else is weaker:
path inference works from 13 byte prefixes with no public keys, and
`complete_contact_tracking.hop_count` over-claims zero-hop (it asserts ~800 zero-hop
contacts where only ~68 have corroborating path evidence). A discover response is a
first-party RF measurement between two full 32-byte public keys.
```ini
[PacketCapture]
enabled = true
neighbors_enabled = true # the only switch you need
neighbors_interval_hours = 24 # clamped to 12-336
```
That one setting turns the whole feature on. Every enabled broker publishes the
snapshot by default (`mqttN_neighbors` defaults to true) — set it false on any broker
you want to hold back:
```ini
mqtt2_neighbors = false
# mqtt1_topic_neighbors = meshcore/{IATA}/{PUBLIC_KEY}/neighbors # optional override
```
The neighbours topic is derived from the broker's packets topic by swapping its last
segment, so a broker configured with `meshcore/{IATA}/{PUBLIC_KEY}/packets` publishes to
`meshcore/{IATA}/{PUBLIC_KEY}/neighbors` — the same topic the firmware uses. Brokers
with only a `topic_prefix` get `<prefix>/neighbors`. If a derived topic is
location-routed but no `iata` is set, that broker is skipped with a warning rather than
publishing into `meshcore/XYZ/...` on a shared namespace.
Each cycle sends one zero-hop node-discover request, then listens for
`neighbors_discover_window` seconds (60 by default). **The bot stays fully responsive
during the window** — it is a passive listen, and only the single discover command
touches the radio.
Results go three places, independently of each other:
- **`neighbor_links`** — current adjacency (full public keys, observation count,
best/last/mean SNR). This is the source of truth.
- **`neighbor_observations`** — one row per neighbour per cycle, for signal history.
Pruned by `[Data_Retention] neighbor_observations_retention_days` (default 365).
- **The mesh graph** — as edges, when `neighbors_feed_mesh_graph = true` (default),
plus a dedicated **Neighbours Only** evidence mode on the mesh page and
`GET /api/mesh/edges?evidence=neighbors`. Confirmed neighbours render as heavier
lines and show their SNR.
Snapshots are published **non-retained**. `heard_secs_ago` is relative to publish time,
so a retained copy replayed days later would still claim the neighbour was heard seconds
ago. Consumers that want the current picture should subscribe and wait for the next
cycle, or read `timestamp` and correct for the age. No broker is required at all — the
database is a perfectly good consumer on its own.
### Triggering a cycle manually
The minimum interval is 12 hours, so use the DM command to test:
```
neighbors
```
It acknowledges immediately and reports the result in a second DM once the window
closes. Enabled via `[Neighbors_Command]`; add `neighbors` to
`[Admin_ACL] admin_commands` to restrict it, since a cycle spends airtime.
### Region scopes are opt-in, and why
`neighbors_collect_scopes` additionally asks each neighbour for its region scopes.
It defaults to **false** for two reasons specific to running inside the bot:
1. **It stalls bot replies.** Every bot radio command is serialised through one lock
(`modules/core.py` `_SerializedCommands`), and `req_regions_sync` waits for its
reply *inside* the call — so one request holds the radio for up to ~25 s. With 32
neighbours the bot's own messages stall in bursts for minutes.
2. **It mutates device contact state.** The zero-hop probe relies on the neighbour
*not* being a known contact. The bot does track contacts, and for a repeater with
no stored path the meshcore library reaches zero-hop by calling
`change_contact_path()` and then `reset_path()` — temporarily rewriting that
contact's path on the device.
With it off, the snapshot reports every neighbour it heard with empty `scopes` and
`status: responded`. Enable it on a bench radio first.
### Published payload
```json
{
"timestamp": "2026-08-04T12:00:00.000000+00:00",
"origin": "MeshCore-HOWL",
"origin_id": "A1B2C3D4E5F67890...",
"total_neighbors": 6,
"queried_neighbors": 6,
"truncated": false,
"self": { "scopes": "" },
"neighbors": [
{
"pubkey": "0011223344556677...",
"snr": 9.75,
"heard_secs_ago": 42,
"scopes": "",
"status": "responded"
}
]
}
```
`total_neighbors` is how many were discovered, `queried_neighbors` how many were kept
after the `neighbors_max` cap, and `truncated` is true when either that cap or the
10 KB payload budget dropped entries. Entries are ordered most- to least-useful (most
recently heard, then stronger SNR). `status` is `responded`, `timeout`, or
`send_failed`.
Requires `meshcore >= 2.3.8` and a firmware build exposing `CMD_SEND_CONTROL_DATA`;
the service logs once and skips the feature if either is missing.
---
## FAQ
**Q: Do I need to provide a private key?**
+169
View File
@@ -0,0 +1,169 @@
#!/usr/bin/env python3
"""
Neighbors command for the MeshCore Bot
Triggers one zero-hop neighbor discovery cycle on demand
"""
import asyncio
from typing import Any, Optional
from ..models import MeshMessage
from .base_command import BaseCommand
class NeighborsCommand(BaseCommand):
"""Runs one neighbor discovery cycle immediately.
The scheduled interval has a 12 hour floor (the firmware's band), which makes
waiting for the scheduler impractical when testing or after moving the node.
This is the bot's equivalent of the firmware's ``discover.neighbors``.
A cycle takes at least ``neighbors_discover_window`` seconds 60 by default
so this acknowledges immediately and reports the result in a second message
rather than holding the reply open.
"""
# Plugin metadata
name = "neighbors"
keywords = ['neighbors', 'neighbours']
description = "Runs a zero-hop neighbor discovery cycle (DM only)"
requires_dm = True
cooldown_seconds = 900 # 15 minutes; a cycle costs airtime
category = "special"
def __init__(self, bot: Any):
"""Initialize the neighbors command.
Args:
bot: The bot instance.
"""
super().__init__(bot)
self.command_enabled = self.get_config_value(
'Neighbors_Command', 'enabled', fallback=True, value_type='bool'
)
# Tracked so a second invocation cannot start an overlapping cycle: two
# concurrent discover rounds would collect into each other's window.
self._cycle_task: Optional[asyncio.Task] = None
def get_help_text(self) -> str:
"""Get help text for the neighbors command.
Returns:
str: The help text for this command.
"""
return self.translate('commands.neighbors.description')
def can_execute(self, message: MeshMessage, skip_channel_check: bool = False) -> bool:
"""Check if the neighbors command can be executed.
Args:
message: The message triggering the command.
skip_channel_check: Passed through to the base implementation.
Returns:
bool: True if the command can be executed, False otherwise.
"""
if not self.command_enabled:
return False
return super().can_execute(message, skip_channel_check=skip_channel_check)
def _get_capture_service(self) -> Any:
"""The packet capture service instance, or None when unavailable."""
service = getattr(self.bot, 'packet_capture_service', None)
if service is not None:
return service
# The alias is set up at init; fall back to the service registry in case
# the service was loaded but not aliased.
services = getattr(self.bot, 'services', None) or {}
try:
return services.get('packetcapture')
except AttributeError:
return None
async def execute(self, message: MeshMessage) -> bool:
"""Execute the neighbors command.
Args:
message: The message triggering the command.
Returns:
bool: True if handled (including the error and busy notices).
"""
service = self._get_capture_service()
if service is None or not getattr(service, 'neighbors_enabled', False):
await self.send_response(message, self.translate('commands.neighbors.disabled'))
return True
if self._cycle_task is not None and not self._cycle_task.done():
await self.send_response(message, self.translate('commands.neighbors.busy'))
return True
cfg = service.neighbors_config
self.logger.info(f"User {message.sender_id} requested a neighbors discovery cycle")
await self.send_response(
message,
self.translate('commands.neighbors.started', seconds=int(cfg.discover_window)),
)
# Run detached so the discover window does not hold the command open, and
# bound it so a stalled radio link cannot leave the task alive forever.
self._cycle_task = asyncio.create_task(
self._run_and_report(message, service, cfg.cycle_budget)
)
return True
async def _run_and_report(self, message: MeshMessage, service: Any, budget: float) -> None:
"""Run one cycle and DM the outcome."""
try:
summary = await asyncio.wait_for(service.run_neighbors_cycle(), timeout=budget)
except asyncio.TimeoutError:
self.logger.error(f"Neighbors: manual cycle exceeded {budget:.0f}s and was abandoned")
await self.send_response(
message,
self.translate('commands.neighbors.error',
error=f"timed out after {budget:.0f}s"),
skip_user_rate_limit=True,
)
return
except asyncio.CancelledError:
raise
except Exception as e:
self.logger.error(f"Neighbors: manual cycle failed: {e}", exc_info=True)
await self.send_response(
message,
self.translate('commands.neighbors.error', error=str(e)),
skip_user_rate_limit=True,
)
return
await self.send_response(
message, self._format_summary(summary), skip_user_rate_limit=True
)
def _format_summary(self, summary: dict[str, Any]) -> str:
"""Render a cycle summary short enough for a mesh DM."""
if not summary.get('ok'):
reason = summary.get('reason') or 'unknown error'
return self.translate('commands.neighbors.failed', reason=reason)
found = summary.get('queried', 0)
if not found:
return self.translate('commands.neighbors.none')
best = summary.get('best_snr')
best_text = f"{best:.1f}dB" if isinstance(best, (int, float)) else "n/a"
text = self.translate(
'commands.neighbors.success',
count=found,
best_snr=best_text,
recorded=summary.get('recorded', 0),
)
# Only mention brokers when at least one was actually tried, so an
# operator with no MQTT does not see a confusing "0/0".
if summary.get('attempted'):
text += " " + self.translate(
'commands.neighbors.published',
succeeded=summary.get('succeeded', 0),
attempted=summary.get('attempted', 0),
)
return text
+2
View File
@@ -59,6 +59,8 @@ class DBManager:
'purging_log', # Repeater manager
'mesh_connections', # Mesh graph for path validation
'observed_paths', # Repeater manager - observed paths from adverts and messages
'neighbor_links', # Zero-hop neighbor discovery - current adjacency
'neighbor_observations', # Zero-hop neighbor discovery - per-cycle history
}
def __init__(self, bot: Any, db_path: str = "meshcore_bot.db"):
+61
View File
@@ -728,6 +728,66 @@ def _m0021_daily_rollup_packet_type_encoding(cursor: sqlite3.Cursor) -> None:
_add_column(cursor, "daily_rollup", "packet_type_encoding", "TEXT")
def _m0022_neighbor_tables(cursor: sqlite3.Cursor) -> None:
"""Tables for zero-hop neighbor discovery (see modules/neighbors_discovery.py).
A discover response is the strongest link evidence this codebase has: a
confirmed direct RF reception between two *full* 32-byte public keys, with a
measured SNR. Every other edge source is weaker ``observed_paths`` carries
13 byte prefixes and no keys, and ``complete_contact_tracking.hop_count``
over-claims zero-hop. So these keep the full keys rather than prefixes, and
stay separate from ``mesh_connections``, which cannot represent provenance
(its ``confirmed_2byte`` flag is memory-only and never persisted).
Two tables because they answer different questions: ``neighbor_links`` is the
current adjacency the mesh graph reads, ``neighbor_observations`` is the
per-cycle history a signal trend needs. SNR is stored as sum+count rather
than a mean, matching ``daily_rollup``, so any window re-aggregates exactly.
Index names are database-global in SQLite (see the note on migration 20), so
every name here is table-qualified.
"""
cursor.executescript(
"""
CREATE TABLE IF NOT EXISTS neighbor_links (
id INTEGER PRIMARY KEY AUTOINCREMENT,
self_public_key TEXT NOT NULL,
neighbor_public_key TEXT NOT NULL,
first_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
last_seen TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
observation_count INTEGER DEFAULT 1,
snr_sum REAL DEFAULT 0,
snr_count INTEGER DEFAULT 0,
best_snr REAL,
last_snr REAL,
last_status TEXT,
scopes TEXT,
UNIQUE(self_public_key, neighbor_public_key)
);
CREATE TABLE IF NOT EXISTS neighbor_observations (
id INTEGER PRIMARY KEY AUTOINCREMENT,
observed_at TIMESTAMP NOT NULL,
self_public_key TEXT NOT NULL,
neighbor_public_key TEXT NOT NULL,
snr REAL,
heard_secs_ago INTEGER,
scopes TEXT,
status TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_neighbor_links_last_seen
ON neighbor_links(last_seen);
CREATE INDEX IF NOT EXISTS idx_neighbor_links_neighbor
ON neighbor_links(neighbor_public_key);
CREATE INDEX IF NOT EXISTS idx_neighbor_observations_observed_at
ON neighbor_observations(observed_at);
CREATE INDEX IF NOT EXISTS idx_neighbor_observations_neighbor
ON neighbor_observations(neighbor_public_key, observed_at);
"""
)
# ---------------------------------------------------------------------------
# Migration registry — append new entries here, never remove or reorder.
# ---------------------------------------------------------------------------
@@ -756,6 +816,7 @@ MIGRATIONS: list[MigrationEntry] = [
(19, "packet_stream: denormalized packet dimensions", _m0019_packet_stream_denorm_dims),
(20, "mesh_connections: table-specific last_seen index", _m0020_mesh_connections_last_seen_index),
(21, "daily_rollup: per-payload-type multibyte split", _m0021_daily_rollup_packet_type_encoding),
(22, "neighbor discovery tables", _m0022_neighbor_tables),
]
+38
View File
@@ -155,6 +155,9 @@ class MaintenanceRunner:
daily_stats_days = get_retention_days('Data_Retention', 'daily_stats_retention_days', 90)
observed_paths_days = get_retention_days('Data_Retention', 'observed_paths_retention_days', 90)
mesh_connections_days = get_retention_days('Data_Retention', 'mesh_connections_retention_days', 7)
neighbor_observations_days = get_retention_days(
'Data_Retention', 'neighbor_observations_retention_days', 365
)
stats_days = get_retention_days('Stats_Command', 'data_retention_days', 7)
try:
@@ -197,6 +200,11 @@ class MaintenanceRunner:
if hasattr(self.bot, 'mesh_graph') and self.bot.mesh_graph and hasattr(self.bot.mesh_graph, 'delete_expired_edges_from_db'):
self.bot.mesh_graph.delete_expired_edges_from_db(mesh_connections_days)
# neighbor_observations is per-cycle history; neighbor_links is the
# aggregate adjacency the mesh graph reads and is deliberately not
# pruned here (losing it would silently drop confirmed direct links).
self._cleanup_neighbor_observations(neighbor_observations_days)
ran_at = _utc_now().isoformat()
self._last_retention_stats['ran_at'] = ran_at
try:
@@ -215,6 +223,36 @@ class MaintenanceRunner:
except Exception:
pass
def _cleanup_neighbor_observations(self, retention_days: int) -> None:
"""Prune zero-hop neighbor observation history past the retention window.
Volume is tiny (at most one row per neighbor per cycle, and cycles are
12h apart at minimum), so the default window is generous the point of
this table is the long-run signal history.
"""
if retention_days <= 0:
return
db_manager = getattr(self.bot, 'db_manager', None)
if not db_manager or not hasattr(db_manager, 'delete_timestamp_rows_in_chunks'):
return
try:
cutoff = (_utc_now() - datetime.timedelta(days=retention_days)).isoformat()
deleted = db_manager.delete_timestamp_rows_in_chunks(
'neighbor_observations',
'observed_at',
cutoff,
progress_label='neighbor observations',
)
if deleted > 0:
self.logger.info(
f"Cleaned up {deleted} old neighbor_observations entries "
f"(older than {retention_days} days)"
)
except Exception as e:
# A missing table (pre-migration-22 database) must not abort the rest
# of the retention run.
self.logger.debug(f"Neighbor observation retention skipped: {e}")
def collect_email_stats(self) -> dict[str, Any]:
"""Gather summary stats for the nightly digest."""
stats: dict[str, Any] = {}
+577
View File
@@ -0,0 +1,577 @@
#!/usr/bin/env python3
"""Zero-hop neighbor discovery and scope collection.
Ported from the ``meshcore-packet-capture`` project's ``neighbors.py``, which is
itself a port of the observer firmware's ``WITH_MQTT_NEIGHBORS`` feature (see
``examples/simple_repeater/MyMesh.cpp``). Two stages per cycle:
1. A zero-hop node-discover request; repeaters answer with their pubkey and the
SNR we heard them at. Responses are collected for a fixed window.
2. One anon-regions request per discovered neighbor, yielding that neighbor's
scope names.
Both requests are issued through meshcore_py (``send_node_discover_req`` and
``req_regions_sync``); nothing here re-encodes packets.
Stage 1 is the valuable part for this bot: it produces a *confirmed direct RF
link* between two full 32-byte public keys, with a measured SNR. That is
stronger evidence than anything path inference can offer, and it costs one radio
command plus a passive listen window.
Stage 2 (scopes) is deliberately optional here and defaults off, for two reasons
that do not apply to the upstream capture tool:
* ``req_regions_sync`` awaits its reply inside the call, and every bot command
is serialized through ``modules.core._SerializedCommands``, so one scope
request holds the radio for its whole round trip -- stalling message sends.
* Upstream relies on a freshly discovered neighbor *not* being a known contact,
which is what makes ``send_anon_req`` ask for a zero-hop reply path. This bot
populates the library's contact cache. For a contact with no path
(``out_path_len == -1``, the common case for a flood repeater) the library
reaches zero-hop by calling ``change_contact_path()`` and then
``reset_path()`` -- i.e. it *mutates the device's contact table* per neighbor.
No device-command lock is passed around: unlike upstream, every coroutine on
``meshcore.commands`` is already serialized and paced by
``modules.core._SerializedCommands``.
"""
from __future__ import annotations
import asyncio
import json
import logging
import random
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Any, Callable, Optional
from meshcore import EventType
from .enums import AdvertFlags
# The firmware discovers repeaters only. The request filter is a bitmask over
# advert *types*, so the bit index is the type value (2 -> 0x04).
DISCOVER_FILTER_REPEATER = 1 << AdvertFlags.ADV_TYPE_REPEATER.value
# Matches MQTTBridge::NEIGHBORS_JSON_BUFFER_SIZE. Entries past this budget are
# dropped from the tail so the payload stays within the firmware's contract.
NEIGHBORS_JSON_BUDGET = 10240
STATUS_RESPONDED = "responded"
STATUS_TIMEOUT = "timeout"
STATUS_SEND_FAILED = "send_failed"
# Firmware interval band (MQTTPrefsStorage.h).
MIN_INTERVAL_HOURS = 12
MAX_INTERVAL_HOURS = 336
DEFAULT_INTERVAL_HOURS = 24
# Floors that keep a misconfiguration from producing an empty snapshot forever.
# Repeaters answer a discover request after a randomised delay, so a very short
# window collects nothing at all.
MIN_DISCOVER_WINDOW = 5.0
MIN_CYCLE_TIMEOUT = 10.0
MIN_COMMAND_TIMEOUT = 1.0
# meshcore_py's CommandHandlerBase.DEFAULT_TIMEOUT: how long req_regions_sync
# waits for MSG_SENT before it even begins waiting for the response.
LIBRARY_MSG_SENT_TIMEOUT = 15.0
def clamp_interval_hours(hours: int) -> int:
"""Clamp to the firmware's 12-336h band, falling back to the 24h default."""
if hours <= 0:
return DEFAULT_INTERVAL_HOURS
return max(MIN_INTERVAL_HOURS, min(MAX_INTERVAL_HOURS, hours))
@dataclass
class NeighborsConfig:
"""Tuning for one neighbors cycle. Defaults mirror the firmware."""
interval_hours: int = DEFAULT_INTERVAL_HOURS
discover_window: float = 60.0 # stage 1 collection window
command_timeout: float = 20.0 # cap on the discover request itself
collect_scopes: bool = False # stage 2 opt-in (bot-only; see module docstring)
scope_timeout: float = 0.0 # 0 = let the device suggest it
scope_min_timeout: float = 8.0 # floor under the suggested timeout
scope_gap: float = 2.0 # settle delay between scope requests
cycle_timeout: float = 600.0 # overall budget for the scope pass
max_neighbors: int = 32
self_scopes: str = "" # explicit override; "" = ask the device
def __post_init__(self) -> None:
# Values that would silently produce a permanently empty snapshot get a
# floor rather than being honoured: a 0s discover window collects nothing,
# and max_neighbors = 0 queries nobody.
self.interval_hours = clamp_interval_hours(self.interval_hours)
if self.discover_window < MIN_DISCOVER_WINDOW:
self.discover_window = MIN_DISCOVER_WINDOW
if self.max_neighbors < 1:
self.max_neighbors = 1
if self.cycle_timeout < MIN_CYCLE_TIMEOUT:
self.cycle_timeout = MIN_CYCLE_TIMEOUT
if self.scope_gap < 0:
self.scope_gap = 0.0
# wait_for(timeout=0) raises immediately, so 0 here would break every
# cycle forever -- and it reads as "no cap" by analogy with scope_timeout.
if self.command_timeout < MIN_COMMAND_TIMEOUT:
self.command_timeout = MIN_COMMAND_TIMEOUT
@property
def scope_request_budget(self) -> float:
"""Hard ceiling on one scope request.
req_regions_sync waits for MSG_SENT (the library default, 15s) and then
for the response, so the budget has to exceed both or healthy requests
would be cut off.
"""
wait = self.scope_timeout if self.scope_timeout > 0 else self.scope_min_timeout
return LIBRARY_MSG_SENT_TIMEOUT + max(wait, self.scope_min_timeout) + self.command_timeout
@property
def interval_seconds(self) -> float:
return self.interval_hours * 3600.0
@property
def cycle_budget(self) -> float:
"""Worst-case wall time for one full cycle, for callers that bound it.
Covers stage 1, the scope pass, and the self-scope query. Used by the
manual trigger so a stalled link cannot hang the caller indefinitely.
"""
budget = self.discover_window + self.command_timeout + self.command_timeout
if self.collect_scopes:
budget += self.cycle_timeout + self.scope_request_budget + self.scope_gap
return budget
@dataclass
class NeighborEntry:
"""One discovered neighbor. Snapshotted so later table changes can't alter it."""
pubkey: str
snr: float
heard_at: float # wall clock of the response
scopes: str = ""
status: str = STATUS_TIMEOUT
def heard_secs_ago(self, now: Optional[float] = None) -> int:
now = time.time() if now is None else now
return max(0, int(now - self.heard_at))
def sort_key(entry: NeighborEntry, now: Optional[float] = None) -> tuple[int, float, str]:
"""Firmware ordering: most recently heard, then stronger SNR, then pubkey.
Mirrors neighborPublishEntryComesBefore() and the pre-query sort added in
firmware commit aba571ed, so query order and publish order agree -- a
truncated cycle still covers the most useful neighbors.
Recency is compared as the published heard_secs_ago, ascending -- the exact
value the firmware's comparator uses. Sorting on the raw float clock instead
would quantise differently from the field we publish, so the output could be
non-monotonic in heard_secs_ago, and SNR would never break a tie: the most
recent response would always win outright. That matters because this order
decides which entries survive the payload budget.
"""
return (entry.heard_secs_ago(now), -entry.snr, entry.pubkey)
def sort_entries(entries: list[NeighborEntry],
now: Optional[float] = None) -> list[NeighborEntry]:
# One `now` for the whole sort, and the same one the payload uses, so the
# published heard_secs_ago values are ordered exactly as sorted.
now = time.time() if now is None else now
return sorted(entries, key=lambda e: sort_key(e, now))
async def discover_neighbors(
meshcore: Any,
cfg: NeighborsConfig,
self_pubkey: Optional[str],
logger: logging.Logger,
*,
debug: bool = False,
still_valid: Optional[Callable[[], bool]] = None,
) -> Optional[list[NeighborEntry]]:
"""Stage 1: zero-hop node-discover, collecting responses for the window.
Returns the discovered entries (possibly empty), or None if the request
could not be sent or the session was invalidated mid-window.
``still_valid`` is an optional predicate that must keep returning True for
the collected data to mean anything. A reconnect part-way through the window
tears down every event subscription (the service clears them wholesale), so
our response handler silently stops firing -- without this check the cycle
would happily record "0 neighbours" as though the mesh were empty.
"""
collected: dict[str, NeighborEntry] = {}
self_key = (self_pubkey or "").lower()
# EventDispatcher spawns async callbacks as background tasks, so an event
# already dequeued can still reach the handler after unsubscribe() returns.
# This latch keeps a straggler from mutating entries we have already returned.
closed = False
# Generate the tag ourselves rather than letting the library pick one, so it is
# known *before* the subscription goes live. Otherwise a response arriving
# between subscribe and send-completion would be accepted with no tag check --
# which is how a stale round, or another client's round, could leak in.
# DISCOVER_RESPONSE reports the tag as little-endian hex.
tag = random.randint(1, 0xFFFFFFFF)
expected_tag = tag.to_bytes(4, "little").hex()
async def on_discover_response(event: Any) -> None:
if closed:
return
payload = getattr(event, "payload", None) or {}
pubkey = str(payload.get("pubkey", "")).lower()
# Full 32-byte pubkeys only; the firmware rejects short prefixes too.
if len(pubkey) != 64:
return
if self_key and pubkey == self_key:
return
if payload.get("node_type") != AdvertFlags.ADV_TYPE_REPEATER.value:
return
if str(payload.get("tag", "")).lower() != expected_tag:
return
snr = float(payload.get("SNR", 0) or 0)
existing = collected.get(pubkey)
if existing is None:
collected[pubkey] = NeighborEntry(pubkey=pubkey, snr=snr, heard_at=time.time())
else:
# Same neighbor heard again (repeaters delay responses randomly).
# putNeighbour() refreshes the timestamp on every response, so do
# that unconditionally; keep the strongest SNR seen.
existing.heard_at = time.time()
existing.snr = max(existing.snr, snr)
# Subscribe per-run rather than at startup: the service drops every
# subscription on reconnect, and this handler is only wanted for the
# duration of the window.
subscription = meshcore.subscribe(EventType.DISCOVER_RESPONSE, on_discover_response)
try:
# Bounded: on a stalled link the underlying BLE/serial write can block far
# longer than the library's own response timeout, and an unbounded wait
# here stalls the whole cycle.
try:
result = await asyncio.wait_for(
meshcore.commands.send_node_discover_req(
DISCOVER_FILTER_REPEATER,
prefix_only=False, # we need the full 32-byte pubkey
tag=tag,
),
timeout=cfg.command_timeout,
)
except asyncio.TimeoutError:
logger.warning(
f"Neighbors: node-discover request did not complete within "
f"{cfg.command_timeout:.0f}s, abandoning this cycle"
)
return None
if result is None or result.type == EventType.ERROR:
reason = ""
if result is not None:
reason = (getattr(result, "payload", None) or {}).get("reason", "")
# Logged at debug: the caller owns the user-facing message, because on a
# build that lacks the command this fails every cycle forever.
logger.debug(
f"Neighbors: node-discover request failed{f' ({reason})' if reason else ''}"
)
return None
if debug:
logger.debug(
f"Neighbors: node-discover sent (tag={expected_tag}), "
f"collecting for {cfg.discover_window:.0f}s"
)
await asyncio.sleep(cfg.discover_window)
if still_valid is not None and not still_valid():
logger.warning(
"Neighbors: device session was reset during the discovery window "
"(event subscriptions are torn down on reconnect), abandoning this cycle"
)
return None
finally:
closed = True
try:
meshcore.unsubscribe(subscription)
except Exception as exc:
logger.debug(f"Neighbors: error unsubscribing discover handler: {exc}")
return sort_entries(list(collected.values()))
async def collect_scopes(
meshcore: Any,
entries: list[NeighborEntry],
cfg: NeighborsConfig,
logger: logging.Logger,
*,
debug: bool = False,
) -> None:
"""Stage 2: one anon-regions request per neighbor, paced, updating in place.
Entries must already be sorted (see sort_entries) so that if the cycle
budget runs out the most useful neighbors have been covered. Anything not
reached keeps its initial ``timeout`` status, matching the firmware's
fallback.
Opt-in only -- see the module docstring for why. In particular, for a
neighbor that *is* a known contact with no stored path, the library reaches
zero-hop by temporarily rewriting that contact's path on the device.
This is the single entry point for stage 2, including when it is disabled,
so that the meaning of ``status`` is decided in exactly one place.
"""
if not entries:
return
if not cfg.collect_scopes:
# Stage 2 disabled. Every entry here answered the discover request, and
# that reception is the entire claim the snapshot makes, so `responded`
# is accurate -- whereas leaving the `timeout` default would report a
# live neighbor as unreachable. Scopes stay empty.
for entry in entries:
entry.status = STATUS_RESPONDED
if debug:
logger.debug(
f"Neighbors: scope collection disabled, reporting "
f"{len(entries)} neighbor(s) without scopes"
)
return
deadline = time.time() + cfg.cycle_timeout
# 0 means "let the device decide": req_regions_sync derives the wait from the
# suggested_timeout the radio returns, which is its own airtime estimate.
timeout = cfg.scope_timeout if cfg.scope_timeout > 0 else 0
for index, entry in enumerate(entries):
if time.time() >= deadline:
dropped = len(entries) - index
logger.warning(
f"Neighbors: cycle budget ({cfg.cycle_timeout:.0f}s) reached, "
f"{dropped} of {len(entries)} neighbor(s) left unqueried (reported as timeout)"
)
break
# Settle gap between requests, standing in for the firmware's
# wait-for-TX-completion gating.
if index > 0 and cfg.scope_gap > 0:
await asyncio.sleep(cfg.scope_gap)
try:
# Bounded: this holds the shared radio command lock for its whole
# round trip, and an unbounded stall here would block every other
# bot command for as long as the write hangs.
scopes = await asyncio.wait_for(
meshcore.commands.req_regions_sync(
entry.pubkey,
timeout=timeout,
min_timeout=cfg.scope_min_timeout,
),
timeout=cfg.scope_request_budget,
)
except asyncio.CancelledError:
raise
except asyncio.TimeoutError:
entry.status = STATUS_SEND_FAILED
logger.warning(
f"Neighbors: scope request to {entry.pubkey[:12]} exceeded "
f"{cfg.scope_request_budget:.0f}s; the device link may be stalled"
)
continue
except Exception as exc:
entry.status = STATUS_SEND_FAILED
logger.debug(f"Neighbors: scope request to {entry.pubkey[:12]} failed: {exc}")
continue
if scopes is None:
# req_regions_sync collapses every failure to None: a real timeout,
# but also a device-level send rejection. We report timeout, which is
# the firmware's own fallback for anything that isn't a clean send
# failure or response.
entry.status = STATUS_TIMEOUT
if debug:
logger.debug(f"Neighbors: no scope response from {entry.pubkey[:12]}")
continue
entry.scopes = str(scopes).strip()
entry.status = STATUS_RESPONDED
if debug:
logger.debug(
f"Neighbors: {entry.pubkey[:12]} scopes="
f"{entry.scopes if entry.scopes else '(none)'}"
)
async def fetch_self_scopes(meshcore: Any, cfg: NeighborsConfig,
logger: logging.Logger) -> str:
"""This node's own scope names for the message's ``self`` object.
A companion radio has no region_map, so the closest analogue to the
firmware's exportNamesTo(REGION_DENY_FLOOD) is the default flood scope name.
``neighbors_self_scopes`` overrides it outright and is honoured even when
stage 2 is disabled.
"""
if cfg.self_scopes:
return cfg.self_scopes
# With stage 2 off the snapshot reports no scopes for anyone, so spending a
# radio command to learn our own would be inconsistent as well as wasteful:
# the default cycle is meant to cost one command plus a listen window.
if not cfg.collect_scopes:
return ""
# A reconnect can null out the device handle mid-cycle (a cycle spans minutes).
commands = getattr(meshcore, "commands", None)
getter = getattr(commands, "get_default_flood_scope", None)
if not callable(getter):
return ""
try:
result = await asyncio.wait_for(getter(), timeout=cfg.command_timeout)
except Exception as exc:
logger.debug(f"Neighbors: could not read default flood scope: {exc}")
return ""
if result is None or result.type == EventType.ERROR:
return ""
return str((getattr(result, "payload", None) or {}).get("scope_name", "") or "").strip()
def record_neighbors(
db_manager: Any,
self_pubkey: str,
entries: list[NeighborEntry],
logger: logging.Logger,
*,
observed_at: Optional[str] = None,
) -> int:
"""Persist one cycle to ``neighbor_observations`` and ``neighbor_links``.
Returns the number of links written. Only entries we actually heard are
recorded: an entry left at ``timeout`` because the scope pass ran out of
budget was still a real zero-hop reception, but one that never answered
discovery never reaches this function at all.
SNR is accumulated as sum+count rather than a running mean so that any later
window can re-aggregate exactly (the ``daily_rollup`` convention).
"""
if not entries or not self_pubkey:
return 0
self_key = self_pubkey.lower()
stamp = observed_at or datetime.now(timezone.utc).isoformat()
written = 0
try:
with db_manager.connection() as conn:
cursor = conn.cursor()
for entry in entries:
cursor.execute(
"""
INSERT INTO neighbor_observations
(observed_at, self_public_key, neighbor_public_key,
snr, heard_secs_ago, scopes, status)
VALUES (?, ?, ?, ?, ?, ?, ?)
""",
(stamp, self_key, entry.pubkey.lower(), entry.snr,
entry.heard_secs_ago(), entry.scopes or "", entry.status),
)
# ON CONFLICT rather than a read-modify-write: the aggregate must
# stay correct even if two cycles were ever to overlap.
cursor.execute(
"""
INSERT INTO neighbor_links
(self_public_key, neighbor_public_key, first_seen, last_seen,
observation_count, snr_sum, snr_count, best_snr, last_snr,
last_status, scopes)
VALUES (?, ?, ?, ?, 1, ?, 1, ?, ?, ?, ?)
ON CONFLICT(self_public_key, neighbor_public_key) DO UPDATE SET
last_seen = excluded.last_seen,
observation_count = observation_count + 1,
snr_sum = snr_sum + excluded.snr_sum,
snr_count = snr_count + 1,
best_snr = MAX(COALESCE(best_snr, excluded.best_snr), excluded.best_snr),
last_snr = excluded.last_snr,
last_status = excluded.last_status,
-- Keep the last non-empty scopes: a cycle with stage 2
-- disabled must not erase scopes an earlier cycle learned.
scopes = CASE
WHEN excluded.scopes != '' THEN excluded.scopes ELSE scopes
END
""",
(self_key, entry.pubkey.lower(), stamp, stamp,
entry.snr, entry.snr, entry.snr, entry.status, entry.scopes or ""),
)
written += 1
conn.commit()
except Exception as exc:
logger.error(f"Neighbors: could not persist snapshot: {exc}")
return 0
return written
def build_neighbors_message(
origin: str,
origin_id: str,
self_scopes: str,
entries: list[NeighborEntry],
*,
timestamp: Optional[str] = None,
now: Optional[float] = None,
budget: int = NEIGHBORS_JSON_BUDGET,
total_neighbors: Optional[int] = None,
) -> tuple[dict[str, Any], int]:
"""Build the neighbors payload, dropping the tail past ``budget`` bytes.
Matches MQTTPayloadBuilder::buildNeighborsMessage. Returns the message and
the number of entries dropped. Key order is part of the contract.
"""
now = time.time() if now is None else now
# total_neighbors is how many were discovered before any max_neighbors cap;
# queried_neighbors is how many we actually asked. Firmware key order
# (MQTTPayloadBuilder.cpp): these sit between origin_id and self.
queried = len(entries)
total = queried if total_neighbors is None else total_neighbors
message: dict[str, Any] = {
"timestamp": timestamp or datetime.now(timezone.utc).isoformat(),
"origin": origin,
"origin_id": origin_id,
"total_neighbors": total,
"queried_neighbors": queried,
# Set below once we know whether the payload budget dropped a tail.
"truncated": total > queried,
"self": {"scopes": self_scopes or ""},
"neighbors": [],
}
neighbors = message["neighbors"]
dropped = 0
for position, entry in enumerate(sort_entries(entries, now)):
neighbors.append(
{
"pubkey": entry.pubkey.upper(),
"snr": entry.snr,
"heard_secs_ago": entry.heard_secs_ago(now),
"scopes": entry.scopes or "",
"status": entry.status,
}
)
# Entries are ordered most- to least-useful, so once one overflows, drop
# it and everything after it.
if len(json.dumps(message)) >= budget:
neighbors.pop()
dropped = len(entries) - position
break
# Truncated covers both causes: the max_neighbors cap and the payload budget.
message["truncated"] = bool(dropped) or total > queried
return message, dropped
@@ -24,6 +24,17 @@ from ..meshcore_payload_decode import (
ChannelKeyStore,
decode_payload,
)
from ..neighbors_discovery import (
MAX_INTERVAL_HOURS,
MIN_INTERVAL_HOURS,
STATUS_RESPONDED,
NeighborsConfig,
build_neighbors_message,
collect_scopes,
discover_neighbors,
fetch_self_scopes,
record_neighbors,
)
# Import bot's utilities for packet hash
from ..utils import (
@@ -47,6 +58,13 @@ import contextlib
from .base_service import BaseServicePlugin
from .packet_capture_utils import create_auth_token_async, read_private_key_file
# bot_metadata key holding the last neighbors cycle timestamp. Namespaced because
# bot_metadata is shared across the whole bot.
NEIGHBORS_STATE_KEY = "packet_capture.last_neighbors_publish"
# Sentinel meaning "no IATA configured" (documented as invalid in config.ini.example).
DEFAULT_IATA = "XYZ"
def _decode_key_str(key_str: str) -> Optional[bytes]:
"""Decode a 16-byte channel key from a hex (32 chars) or base64 string."""
@@ -176,6 +194,45 @@ class PacketCaptureService(BaseServicePlugin):
"min": 1, "default": 43200, "unit": "s", "help": "Default JWT renewal interval (per-broker overrides apply)."},
{"key": "jwt_ttl_seconds", "label": "JWT TTL", "type": "int", "group": "Advanced",
"min": 1, "default": 86400, "unit": "s", "help": "Default JWT lifetime (per-broker overrides apply)."},
# --- Neighbors (zero-hop discovery; see modules/neighbors_discovery.py) ---
{"key": "neighbors_enabled", "label": "Neighbors discovery", "type": "bool", "group": "Neighbors",
"default": False,
"help": "Periodically ask which repeaters this node hears directly. Records confirmed "
"direct links with SNR, and publishes to brokers that opted in."},
{"key": "neighbors_interval_hours", "label": "Interval", "type": "int", "group": "Neighbors",
"min": 12, "max": 336, "default": 24, "unit": "h",
"help": "How often a discovery cycle runs. Clamped to 12-336h, matching the firmware."},
{"key": "neighbors_discover_window", "label": "Discover window", "type": "float", "group": "Neighbors",
"min": 5, "default": 60.0, "unit": "s",
"help": "How long responses are collected. Repeaters reply after a random delay, so a "
"short window finds fewer neighbours. The bot stays responsive throughout."},
{"key": "neighbors_max", "label": "Max neighbours", "type": "int", "group": "Neighbors",
"min": 1, "default": 32, "help": "Cap per cycle, most useful first (recent, then stronger SNR)."},
{"key": "neighbors_feed_mesh_graph", "label": "Feed mesh graph", "type": "bool", "group": "Neighbors",
"default": True, "help": "Also add confirmed direct links as mesh graph edges."},
{"key": "neighbors_collect_scopes", "label": "Collect region scopes", "type": "bool",
"group": "Neighbors", "default": False,
"help": "SLOW — also ask each neighbour for its region scopes. Each request holds the radio "
"for up to ~25s, delaying bot replies, and for a repeater with no stored path the "
"library temporarily rewrites that contact's path on the device. Leave off unless "
"you specifically need scopes."},
{"key": "neighbors_command_timeout", "label": "Command timeout", "type": "float",
"group": "Neighbors", "min": 1, "default": 20.0, "unit": "s",
"help": "Cap on the discover request itself; a stalled link can otherwise block."},
{"key": "neighbors_scope_timeout", "label": "Scope timeout", "type": "float", "group": "Neighbors",
"min": 0, "default": 0.0, "unit": "s",
"help": "Per-scope-request wait. 0 uses the device's own airtime estimate."},
{"key": "neighbors_scope_min_timeout", "label": "Scope min timeout", "type": "float",
"group": "Neighbors", "min": 0, "default": 8.0, "unit": "s",
"help": "Floor under the device-suggested scope timeout."},
{"key": "neighbors_scope_gap", "label": "Scope gap", "type": "float", "group": "Neighbors",
"min": 0, "default": 2.0, "unit": "s", "help": "Settle delay between scope requests."},
{"key": "neighbors_cycle_timeout", "label": "Scope pass budget", "type": "float",
"group": "Neighbors", "min": 10, "default": 600.0, "unit": "s",
"help": "Overall budget for the scope pass; unreached neighbours report as timeout."},
{"key": "neighbors_self_scopes", "label": "Own scopes override", "type": "str",
"group": "Neighbors", "default": "", "width": "lg",
"help": "Override for this node's own \"self.scopes\" value. Empty asks the device."},
]
# Repeating structured blocks (see modules/settings_schema.py). Each MQTT
@@ -206,6 +263,13 @@ class PacketCaptureService(BaseServicePlugin):
"help": "JWT audience claim (when using auth token)."},
{"key": "topic_status", "label": "Status topic", "type": "str", "default": ""},
{"key": "topic_packets", "label": "Packets topic", "type": "str", "default": ""},
{"key": "neighbors", "label": "Publish neighbours", "type": "bool", "default": True,
"help": "Send the zero-hop neighbours snapshot to this broker. On by default, but "
"nothing is sent until Neighbours discovery is enabled above. Turn off to "
"hold just this broker back."},
{"key": "topic_neighbors", "label": "Neighbours topic", "type": "str", "default": "",
"help": "Defaults to the packets topic with its last segment swapped for "
"'neighbors', else <topic prefix>/neighbors."},
{"key": "websocket_path", "label": "WebSocket path", "type": "str", "default": "/mqtt",
"help": "Path when transport is websockets."},
{"key": "client_id", "label": "Client ID", "type": "str", "default": ""},
@@ -305,6 +369,15 @@ class PacketCaptureService(BaseServicePlugin):
self.cached_firmware_info = None
self.radio_info = None
# Neighbors discovery runtime state. Kept out of _load_config so a future
# config reload cannot orphan a running scheduler or replay a cycle
# (map_uploader_service re-invokes its own _load_config on reload).
self.neighbors_task = None
self.neighbors_capability_state = None
self.neighbors_discover_failures = 0
self.neighbors_topic_warned: set[int] = set()
self.last_neighbors_publish = self._load_neighbors_state()
# Background tasks
self.background_tasks: list[asyncio.Task] = []
self.should_exit = False
@@ -366,7 +439,7 @@ class PacketCaptureService(BaseServicePlugin):
self.mqtt_brokers = self._parse_mqtt_brokers(config)
# Global IATA
self.global_iata = config.get("PacketCapture", "iata", fallback="XYZ").lower()
self.global_iata = config.get("PacketCapture", "iata", fallback=DEFAULT_IATA).lower()
# Owner information
self.owner_public_key = config.get("PacketCapture", "owner_public_key", fallback=None)
@@ -403,6 +476,76 @@ class PacketCaptureService(BaseServicePlugin):
# The create_auth_token_async function will automatically try to export the key
# from the device if private_key_hex is None and meshcore_instance is available
self._load_neighbors_config(config)
def _load_neighbors_config(self, config) -> None:
"""Build the neighbors cycle configuration (see modules/neighbors_discovery.py).
Off by default: a cycle spends real airtime and, with scope collection
enabled, holds the shared radio command lock for seconds at a time.
"""
self.neighbors_enabled = config.getboolean("PacketCapture", "neighbors_enabled", fallback=False)
self.neighbors_feed_mesh_graph = config.getboolean(
"PacketCapture", "neighbors_feed_mesh_graph", fallback=True
)
requested_interval = config.getint("PacketCapture", "neighbors_interval_hours", fallback=24)
self.neighbors_config = NeighborsConfig(
interval_hours=requested_interval,
discover_window=config.getfloat("PacketCapture", "neighbors_discover_window", fallback=60.0),
command_timeout=config.getfloat("PacketCapture", "neighbors_command_timeout", fallback=20.0),
collect_scopes=config.getboolean("PacketCapture", "neighbors_collect_scopes", fallback=False),
scope_timeout=config.getfloat("PacketCapture", "neighbors_scope_timeout", fallback=0.0),
scope_min_timeout=config.getfloat("PacketCapture", "neighbors_scope_min_timeout", fallback=8.0),
scope_gap=config.getfloat("PacketCapture", "neighbors_scope_gap", fallback=2.0),
cycle_timeout=config.getfloat("PacketCapture", "neighbors_cycle_timeout", fallback=600.0),
max_neighbors=config.getint("PacketCapture", "neighbors_max", fallback=32),
self_scopes=config.get("PacketCapture", "neighbors_self_scopes", fallback="").strip(),
)
# NeighborsConfig clamps out-of-range values; say so rather than silently
# honouring something different from what was configured.
if self.neighbors_enabled and self.neighbors_config.interval_hours != requested_interval:
self.logger.warning(
f"neighbors_interval_hours {requested_interval} is outside the supported "
f"{MIN_INTERVAL_HOURS}-{MAX_INTERVAL_HOURS}h range, "
f"using {self.neighbors_config.interval_hours}h"
)
def _load_neighbors_state(self) -> float:
"""Last cycle timestamp, from bot_metadata.
Neighbors intervals are long (12-336h), so surviving a restart is what
keeps a bot that reboots often from re-running discovery every start.
"""
try:
raw = self.bot.db_manager.get_metadata(NEIGHBORS_STATE_KEY)
except Exception as e:
self.logger.debug(f"Could not read neighbors state: {e}")
return 0.0
if not raw:
return 0.0
try:
value = float(raw)
except (TypeError, ValueError):
self.logger.warning(f"Ignoring malformed neighbors state value: {raw!r}")
return 0.0
# A clock jump either way would otherwise pin the scheduler: a future
# timestamp suppresses cycles indefinitely, a nonsensical old one is noise.
now = time.time()
if value > now + 300 or value < now - (400 * 86400):
self.logger.warning(
f"Ignoring out-of-range neighbors state timestamp ({value}); will run a cycle"
)
return 0.0
return value
def _save_neighbors_state(self) -> None:
"""Persist the last cycle timestamp to bot_metadata."""
try:
self.bot.db_manager.set_metadata(NEIGHBORS_STATE_KEY, str(self.last_neighbors_publish))
except Exception as e:
self.logger.debug(f"Could not save neighbors state: {e}")
def _build_channel_key_store(self, config) -> ChannelKeyStore:
"""Build a comprehensive channel key store for GRP_TXT decryption.
@@ -528,6 +671,15 @@ class PacketCaptureService(BaseServicePlugin):
"topic_prefix": config.get("PacketCapture", f"mqtt{broker_num}_topic_prefix", fallback=None),
"topic_status": config.get("PacketCapture", f"mqtt{broker_num}_topic_status", fallback=None),
"topic_packets": config.get("PacketCapture", f"mqtt{broker_num}_topic_packets", fallback=None),
"topic_neighbors": config.get(
"PacketCapture", f"mqtt{broker_num}_topic_neighbors", fallback=None
),
# On by default, so `neighbors_enabled` is the single switch that
# turns the feature on everywhere. Set false per broker to hold one
# back. The whole feature is still off until neighbors_enabled.
"neighbors": config.getboolean(
"PacketCapture", f"mqtt{broker_num}_neighbors", fallback=True
),
"use_auth_token": config.getboolean(
"PacketCapture", f"mqtt{broker_num}_use_auth_token", fallback=False
),
@@ -1864,6 +2016,380 @@ class PacketCaptureService(BaseServicePlugin):
task = asyncio.create_task(self.mqtt_reconnection_monitor())
self.background_tasks.append(task)
# Neighbors discovery. Unlike the upstream capture tool this does not
# require a broker: the database is a legitimate consumer on its own, so
# the cycle runs whenever the feature is enabled and publishes to
# whichever brokers opted in (possibly none).
if self.neighbors_enabled:
self.neighbors_task = asyncio.create_task(self.neighbors_scheduler())
self.background_tasks.append(self.neighbors_task)
def neighbors_brokers(self) -> list[dict[str, Any]]:
"""Connected MQTT clients whose broker opted into the neighbors topic."""
return [
info for info in self.mqtt_clients
if info.get("connected", False) and info["config"].get("neighbors", False)
]
def neighbors_commands_available(self) -> bool:
"""Detect whether the connected build exposes the neighbors commands.
``send_node_discover_req`` needs companion CMD_SEND_CONTROL_DATA (v8+);
``req_regions_sync`` needs CMD_SEND_ANON_REQ and is only required when
scope collection is enabled. Logged once per state change, like stats.
"""
if not self.meshcore or not hasattr(self.meshcore, "commands"):
return False
commands = self.meshcore.commands
required = ["send_node_discover_req"]
if self.neighbors_config.collect_scopes:
required.append("req_regions_sync")
available = all(callable(getattr(commands, attr, None)) for attr in required)
state = "available" if available else "missing"
if state != self.neighbors_capability_state:
if available:
self.logger.info("MeshCore neighbors commands detected - neighbors discovery enabled")
else:
self.logger.warning(
"MeshCore neighbors commands not available - neighbors discovery disabled "
"(needs a newer firmware/meshcore build)"
)
self.neighbors_capability_state = state
return available
def _neighbors_topic_template(self, broker_config: dict[str, Any]) -> Optional[str]:
"""Unresolved neighbors topic template for one broker.
Order matters because brokers publish neighbors by default, so the derived
value has to be *correct*, not merely non-empty:
1. Explicit ``topic_neighbors`` always wins.
2. Otherwise swap the last segment of ``topic_packets``. A broker
configured with ``meshcore/{IATA}/{PUBLIC_KEY}/packets`` then gets
``meshcore/{IATA}/{PUBLIC_KEY}/neighbors`` the topic the firmware
actually publishes instead of an unrelated flat topic.
3. Otherwise ``<topic_prefix>/neighbors``, mirroring how the packet path
falls back to ``<prefix>/packet``.
"""
explicit = broker_config.get("topic_neighbors")
if explicit:
return explicit
packets = broker_config.get("topic_packets")
if packets and "/" in packets:
return packets.rsplit("/", 1)[0] + "/neighbors"
prefix = broker_config.get("topic_prefix")
if prefix:
return f"{prefix}/neighbors"
return None
def _resolve_neighbors_topic(self, broker_config: dict[str, Any]) -> Optional[str]:
"""Resolved topic for one broker's neighbors snapshot, or None if unroutable."""
template = self._neighbors_topic_template(broker_config)
if not template:
return None
# An unset IATA is the documented sentinel "XYZ", and this topic is
# location-routed on the community brokers. Publishing a snapshot to
# meshcore/XYZ/... would pollute a shared namespace, so refuse instead —
# matching the upstream rule that neighbors needs a real IATA to route.
# Only applies to a template we derived; an explicit topic is the
# operator's call.
# .upper() catches both the {IATA} and {iata} placeholder spellings.
if (
not broker_config.get("topic_neighbors")
and "{IATA}" in template.upper()
and self.global_iata == DEFAULT_IATA.lower()
):
return None
return self._resolve_topic_template(template, "neighbors")
def publish_neighbors_mqtt(self, message: dict[str, Any]) -> dict[str, int]:
"""Publish a neighbors snapshot to every opted-in, connected broker.
Non-retained: a snapshot taken every 12-336h is not a useful last-will
value, and a stale retained copy would read as current.
"""
metrics = {"attempted": 0, "succeeded": 0}
if not self.mqtt_enabled or mqtt is None:
return metrics
payload = json.dumps(message, default=str)
for mqtt_client_info in self.neighbors_brokers():
config = mqtt_client_info["config"]
broker_num = config.get("broker_num", 0)
host = config.get("host", "unknown")
topic = self._resolve_neighbors_topic(config)
if not topic:
# Warn once per broker: the cycle spent real airtime, so silently
# dropping the result would be worse than noisy.
if broker_num not in self.neighbors_topic_warned:
self.neighbors_topic_warned.add(broker_num)
template = self._neighbors_topic_template(config)
if template and "{IATA}" in template.upper():
reason = (
f"its topic is location-routed ({template}) but no IATA is set; "
f"set [PacketCapture] iata, or mqtt{broker_num}_topic_neighbors "
f"to a topic that does not need one"
)
else:
reason = (
f"no neighbors topic could be resolved (set "
f"mqtt{broker_num}_topic_neighbors or a topic prefix)"
)
self.logger.warning(
f"Not publishing neighbors to {host}: {reason}"
)
continue
try:
metrics["attempted"] += 1
result = mqtt_client_info["client"].publish(topic, payload, qos=0, retain=False)
if result.rc == mqtt.MQTT_ERR_SUCCESS:
metrics["succeeded"] += 1
self.logger.debug(f"Published neighbors to '{topic}' on {host}")
else:
self.logger.warning(
f"Failed to publish neighbors to '{topic}' on {host}: "
f"{result.rc} ({mqtt.error_string(result.rc)})"
)
except Exception as e:
self.logger.error(f"Error publishing neighbors to MQTT on {host}: {e}")
return metrics
def _feed_mesh_graph(self, self_pubkey: str, entries: list[Any]) -> None:
"""Add each confirmed direct link to the mesh graph, both directions.
Both directions are correct: a discover response proves we transmitted,
they received, they transmitted, and we received.
Note that ``mesh_connections`` cannot represent *why* an edge exists
its multi-byte confirmation flag is memory-only and never persisted so
``neighbor_links`` remains the source of truth for this evidence, and the
viewer derives its neighbors evidence mode from that table instead.
"""
mesh_graph = getattr(self.bot, "mesh_graph", None)
if not mesh_graph or not self_pubkey:
return
self_prefix = self_pubkey.lower()[:6]
added = 0
for entry in entries:
neighbor_prefix = entry.pubkey.lower()[:6]
if not neighbor_prefix or neighbor_prefix == self_prefix:
continue
try:
# hop_position 1: a direct link is the first hop of any path
# through it. add_edge honours the graph capture kill-switch.
mesh_graph.add_edge(
self_prefix, neighbor_prefix,
from_public_key=self_pubkey.lower(),
to_public_key=entry.pubkey.lower(),
hop_position=1, prefix_bytes=2,
)
mesh_graph.add_edge(
neighbor_prefix, self_prefix,
from_public_key=entry.pubkey.lower(),
to_public_key=self_pubkey.lower(),
hop_position=1, prefix_bytes=2,
)
added += 2
except Exception as e:
self.logger.debug(f"Neighbors: could not add graph edge for {neighbor_prefix}: {e}")
if added and self.debug:
self.logger.debug(f"Neighbors: fed {added} mesh graph edge(s)")
def _device_public_key(self) -> str:
"""This node's public key, lowercase hex, or '' when unavailable."""
if not self.meshcore or not hasattr(self.meshcore, "self_info"):
return ""
try:
self_info = self.meshcore.self_info
if isinstance(self_info, dict):
key = self_info.get("public_key", "")
else:
key = getattr(self_info, "public_key", "")
if isinstance(key, (bytes, bytearray)):
key = bytes(key).hex()
key = str(key or "").replace("0x", "").replace(" ", "").strip().lower()
return "" if key in ("", "unknown") else key
except Exception as e:
self.logger.debug(f"Could not read device public key: {e}")
return ""
async def run_neighbors_cycle(self) -> dict[str, Any]:
"""Run one discovery cycle: record it, feed the graph, publish it.
Returns a summary dict (also used by the ``neighbors`` command):
``{'ok', 'reason', 'discovered', 'queried', 'best_snr', 'attempted',
'succeeded', 'recorded'}``.
"""
summary: dict[str, Any] = {
"ok": False, "reason": "", "discovered": 0, "queried": 0,
"best_snr": None, "attempted": 0, "succeeded": 0, "recorded": 0,
}
if not self.neighbors_enabled:
summary["reason"] = "neighbors discovery is disabled"
return summary
if not self.meshcore or not self.bot.connected:
summary["reason"] = "radio not connected"
return summary
if not self.neighbors_commands_available():
summary["reason"] = "radio build does not support neighbor discovery"
return summary
cfg = self.neighbors_config
self_pubkey = self._device_public_key()
self.logger.info("Neighbors: starting discovery cycle")
# A reconnect swaps in a fresh MeshCore object and clears every event
# subscription, so a cycle spanning one is collecting into a dead handler.
session = self.meshcore
def session_intact() -> bool:
return self.meshcore is session and self.bot.connected
entries = await discover_neighbors(
self.meshcore, cfg, self_pubkey, self.logger,
debug=self.debug, still_valid=session_intact,
)
if entries is None:
# A build that rejects the discover command fails every cycle; warn
# once rather than on every wakeup forever.
self.neighbors_discover_failures += 1
if self.neighbors_discover_failures == 1:
self.logger.warning(
"Neighbors: discovery request failed; will keep retrying quietly "
"on the configured interval"
)
else:
self.logger.debug(
f"Neighbors: discovery request failed "
f"({self.neighbors_discover_failures} consecutive)"
)
summary["reason"] = "discovery request failed"
return summary
self.neighbors_discover_failures = 0
total_discovered = len(entries)
summary["discovered"] = total_discovered
if total_discovered > cfg.max_neighbors:
self.logger.info(
f"Neighbors: {total_discovered} discovered, keeping the "
f"{cfg.max_neighbors} most useful"
)
entries = entries[:cfg.max_neighbors]
self.logger.info(f"Neighbors: {len(entries)} neighbor(s) discovered")
await collect_scopes(self.meshcore, entries, cfg, self.logger, debug=self.debug)
if not session_intact():
self.logger.warning(
"Neighbors: device session was reset mid-cycle, discarding this cycle "
"rather than recording partial data"
)
summary["reason"] = "device session reset mid-cycle"
return summary
summary["queried"] = len(entries)
if entries:
summary["best_snr"] = max(e.snr for e in entries)
# Record before publishing: the data is valuable on its own, and a broker
# problem must not cost us the observation.
if entries and self_pubkey:
summary["recorded"] = record_neighbors(
self.bot.db_manager, self_pubkey, entries, self.logger
)
elif entries and not self_pubkey:
self.logger.warning(
"Neighbors: device public key unavailable, cannot record links "
"(an unattributed link is not usable evidence)"
)
if entries and self.neighbors_feed_mesh_graph and self_pubkey:
self._feed_mesh_graph(self_pubkey, entries)
self_scopes = await fetch_self_scopes(self.meshcore, cfg, self.logger)
origin_id = self_pubkey.upper() if self_pubkey else "DEVICE"
message, dropped = build_neighbors_message(
self._get_bot_name() or "MeshCore Device",
origin_id, self_scopes, entries,
total_neighbors=total_discovered,
)
if dropped:
self.logger.warning(
f"Neighbors: payload budget reached, dropped {dropped} least-useful entry(ies)"
)
metrics = self.publish_neighbors_mqtt(message)
summary["attempted"] = metrics["attempted"]
summary["succeeded"] = metrics["succeeded"]
responded = sum(1 for e in entries if e.status == STATUS_RESPONDED)
if metrics["attempted"]:
self.logger.info(
f"Neighbors: published {len(message['neighbors'])} entry(ies) "
f"({responded} responded) to {metrics['succeeded']}/{metrics['attempted']} broker(s)"
)
else:
self.logger.info(
f"Neighbors: recorded {summary['recorded']} link(s); "
f"no broker has neighbors enabled, nothing published"
)
# Stamp the attempt even when nothing was published, so a persistently
# failing broker cannot turn every wakeup into a fresh discovery burst.
self.last_neighbors_publish = time.time()
self._save_neighbors_state()
summary["ok"] = True
return summary
async def neighbors_scheduler(self) -> None:
"""Run a discovery cycle on the configured interval."""
interval_seconds = self.neighbors_config.interval_seconds
if self.debug:
self.logger.debug(
f"Starting neighbors scheduler "
f"({self.neighbors_config.interval_hours}h interval)"
)
while not self.should_exit:
try:
time_since_last = time.time() - self.last_neighbors_publish
if time_since_last < interval_seconds:
sleep_time = interval_seconds - time_since_last
if self.debug:
self.logger.debug(f"Next neighbors cycle in {sleep_time / 3600:.1f} hours")
if await self._wait_with_shutdown(sleep_time):
break
continue
await self.run_neighbors_cycle()
# run_neighbors_cycle stamps last_neighbors_publish when it got as
# far as a result. If it bailed early (not connected, unsupported
# build) back off before retrying so we re-check periodically
# rather than spinning.
if (time.time() - self.last_neighbors_publish) >= interval_seconds:
if await self._wait_with_shutdown(300):
break
except asyncio.CancelledError:
if self.debug:
self.logger.debug("Neighbors scheduler cancelled")
break
except Exception as e:
self.logger.error(f"Error in neighbors scheduler: {e}")
if await self._wait_with_shutdown(300):
break
async def stats_refresh_scheduler(self) -> None:
"""Periodically refresh stats and publish them via MQTT (matches original script).
+146 -3
View File
@@ -974,6 +974,122 @@ class BotDataViewer:
result.sort(key=lambda e: e['last_seen'] or '', reverse=True)
return result
# Nodes in the neighbor tables are stored as full 32-byte public keys, so the
# graph's highest resolution (3 bytes) is always available for edge identity.
NEIGHBOR_PREFIX_HEX_CHARS = 6
def _neighbor_evidence_edge_keys(self) -> set[tuple[str, str]]:
"""Directed prefix pairs that confirmed zero-hop discovery has proven.
Used to upgrade the evidence label in the combined view, where the edge
itself comes from ``mesh_connections`` and so has lost its provenance.
"""
chars = self.NEIGHBOR_PREFIX_HEX_CHARS
try:
with self._with_db_connection() as conn:
rows = conn.execute(
'SELECT self_public_key, neighbor_public_key FROM neighbor_links'
).fetchall()
except Exception as exc:
# A pre-migration-22 database simply has no neighbor evidence.
self.logger.debug(f"Neighbor evidence keys unavailable: {exc}")
return set()
keys: set[tuple[str, str]] = set()
for row in rows:
a = (row['self_public_key'] or '').lower()[:chars]
b = (row['neighbor_public_key'] or '').lower()[:chars]
if a and b:
keys.add((a, b))
keys.add((b, a))
return keys
def _compute_neighbor_evidence_edges(self) -> list[dict[str, Any]]:
"""Derive mesh edges from confirmed zero-hop neighbor discovery.
This is the strongest evidence class in the database: each row is a
direct RF reception between two *full* public keys with a measured SNR,
recorded by modules/neighbors_discovery.py. Two differences from the
multi-byte path derivation are worth noting:
* ``from_public_key``/``to_public_key`` are populated. Path-derived edges
cannot fill these in, because a path carries prefixes only.
* ``snr``/``best_snr`` are real measurements. Unlike the dashboard's
one-hop panel, which withholds SNR unless two sources agree because
``complete_contact_tracking.hop_count`` over-claims zero-hop, a
discover response *is* the authoritative first-party measurement.
Both directions are emitted per link: a discover response proves we
transmitted, they received, they transmitted, and we received.
"""
chars = self.NEIGHBOR_PREFIX_HEX_CHARS
query = '''
SELECT
self_public_key,
neighbor_public_key,
observation_count,
snr_sum,
snr_count,
best_snr,
last_snr,
first_seen,
last_seen
FROM neighbor_links
'''
try:
with self._with_db_connection() as conn:
rows = conn.execute(query).fetchall()
except Exception as exc:
self.logger.debug(f"Neighbor evidence edges unavailable: {exc}")
return []
edges: list[dict[str, Any]] = []
for row in rows:
self_key = (row['self_public_key'] or '').lower()
neighbor_key = (row['neighbor_public_key'] or '').lower()
if not self_key or not neighbor_key:
continue
snr_count = row['snr_count'] or 0
mean_snr = (row['snr_sum'] / snr_count) if snr_count else None
for from_key, to_key in ((self_key, neighbor_key), (neighbor_key, self_key)):
edges.append({
'from_prefix': from_key[:chars],
'to_prefix': to_key[:chars],
'from_public_key': from_key,
'to_public_key': to_key,
'observation_count': row['observation_count'] or 1,
'first_seen': row['first_seen'],
'last_seen': row['last_seen'],
# A direct link is by definition the first hop of any path
# that crosses it.
'avg_hop_position': 1.0,
'geographic_distance': None,
'snr': mean_snr,
'best_snr': row['best_snr'],
'last_snr': row['last_snr'],
'evidence': 'neighbors',
})
edges.sort(key=lambda e: e['last_seen'] or '', reverse=True)
return edges
def _derive_neighbor_evidence_graph(
self,
days: int | None = None,
min_observations: int | None = None,
) -> tuple[list[dict[str, Any]], int]:
"""Filtered neighbor-evidence edges plus their prefix resolution.
Reuses the multi-byte view filter: it only touches ``last_seen`` and
``observation_count`` (handling both naive and aware timestamps), which
is exactly the filtering these edges need.
"""
all_edges = self._compute_neighbor_evidence_edges()
filtered = self._filter_multibyte_evidence_edges(
all_edges, days=days, min_observations=min_observations
)
return filtered, self.NEIGHBOR_PREFIX_HEX_CHARS
def _resolve_path(self, path_input: str) -> dict[str, Any]:
"""Resolve a hex path to repeater names/locations for the mesh map.
@@ -2739,6 +2855,10 @@ class BotDataViewer:
evidence=multibyte derives edges purely from unique multi-byte path
observations (observed_paths, bytes_per_hop >= 2), bypassing the
mesh_connections merge heuristics that single-byte evidence feeds into.
evidence=neighbors derives edges purely from confirmed zero-hop
discovery (neighbor_links) full public keys on both ends plus a
measured SNR, the strongest evidence class available.
"""
conn = None
try:
@@ -2762,6 +2882,21 @@ class BotDataViewer:
'evidence': 'multibyte',
})
if evidence == 'neighbors':
edges, prefix_hex_chars = self._derive_neighbor_evidence_graph(
days=days,
min_observations=min_observations,
)
return jsonify({
'edges': edges,
'prefix_hex_chars': prefix_hex_chars,
'evidence': 'neighbors',
})
# Combined view: mesh_connections cannot record *why* an edge
# exists, so re-derive the strongest label from neighbor_links.
neighbor_keys = self._neighbor_evidence_edge_keys()
conn = self._get_db_connection()
cursor = conn.cursor()
@@ -2831,9 +2966,17 @@ class BotDataViewer:
# by a multi-byte path observation; 2-char keys carry only ambiguous
# single-byte evidence.
is_multibyte = bool(fp) and bool(tp) and len(fp) >= 4 and len(tp) >= 4
from_lower = fp.lower() if fp else ''
to_lower = tp.lower() if tp else ''
if (from_lower, to_lower) in neighbor_keys:
edge_evidence = 'neighbors'
elif is_multibyte:
edge_evidence = 'multibyte'
else:
edge_evidence = 'singlebyte'
edges.append({
'from_prefix': fp.lower() if fp else '',
'to_prefix': tp.lower() if tp else '',
'from_prefix': from_lower,
'to_prefix': to_lower,
'from_public_key': row['from_public_key'],
'to_public_key': row['to_public_key'],
'observation_count': row['observation_count'],
@@ -2841,7 +2984,7 @@ class BotDataViewer:
'last_seen': row['last_seen'],
'avg_hop_position': row['avg_hop_position'],
'geographic_distance': row['geographic_distance'],
'evidence': 'multibyte' if is_multibyte else 'singlebyte'
'evidence': edge_evidence
})
return jsonify({'edges': edges, 'prefix_hex_chars': prefix_hex_chars or 2})
+40 -8
View File
@@ -243,9 +243,10 @@
</div>
<div class="col-md-2">
<label for="filter-evidence" class="form-label">Evidence</label>
<select class="form-select" id="filter-evidence" title="Multi-byte only: build the graph solely from unique multi-byte path observations, ignoring ambiguous single-byte evidence">
<select class="form-select" id="filter-evidence" title="Multi-byte only: build the graph solely from unique multi-byte path observations, ignoring ambiguous single-byte evidence. Neighbours only: only links this node confirmed directly by zero-hop discovery, with measured SNR.">
<option value="all">All Evidence</option>
<option value="multibyte" selected>Multi-byte Only</option>
<option value="neighbors">Neighbours Only</option>
</select>
</div>
<div class="col-md-2">
@@ -294,6 +295,10 @@
<span>Bidirectional</span>
</div>
<hr style="margin: 4px 0;">
<div class="connection-legend-item">
<span style="display: inline-block; width: 18px; border-top: 4px solid currentColor; vertical-align: middle;"></span>
<span>Confirmed neighbour (zero-hop)</span>
</div>
<div class="connection-legend-item">
<span style="display: inline-block; width: 18px; border-top: 2px solid currentColor; vertical-align: middle;"></span>
<span>Multi-byte evidence</span>
@@ -741,7 +746,21 @@
return edge.evidence === 'singlebyte';
}
// A zero-hop discovery response is a direct RF measurement between two full
// public keys — the strongest evidence class, so it gets a heavier line.
function isNeighborEdge(edge) {
return edge.evidence === 'neighbors';
}
function formatSnr(value) {
return (typeof value === 'number' && isFinite(value)) ? `${value.toFixed(1)} dB` : null;
}
function getEvidenceLabel(edge) {
if (edge.evidence === 'neighbors') {
const snr = formatSnr(edge.snr);
return snr ? `Confirmed neighbour (avg SNR ${snr})` : 'Confirmed neighbour (zero-hop)';
}
if (edge.evidence === 'multibyte') {
return edge.path_count
? `Multi-byte (${edge.path_count} unique path${edge.path_count === 1 ? '' : 's'})`
@@ -931,14 +950,18 @@
// --- Initial map framing: home mesh component ---
// Nodes occasionally carry wrong GPS, and fitting the view to every node can
// stretch the initial frame across continents. Instead, frame the component
// reachable from the bot's location over multi-byte (provenance-trusted)
// evidence edges only; wrong-GPS phantoms are rarely connected to it.
// reachable from the bot's location over provenance-trusted evidence edges
// only; wrong-GPS phantoms are rarely connected to it.
let botLocation = null; // {latitude, longitude} from /api/mesh/stats
function getHomeComponentBounds() {
if (!botLocation) return null;
// Both evidence modes tag edges, so this works even in "All Evidence"
const mbEdges = filteredEdges.filter(e => e.evidence === 'multibyte');
// Every evidence mode tags its edges, so this works in "All Evidence" too.
// Confirmed neighbours count as trusted (they are measured, not inferred),
// and are the only tagged edges present in "Neighbours Only" mode.
const mbEdges = filteredEdges.filter(
e => e.evidence === 'multibyte' || e.evidence === 'neighbors'
);
if (!mbEdges.length) return null;
const adj = new Map(); // nodeId -> Set of neighbor nodeIds
@@ -1636,7 +1659,7 @@
const edgeDays = document.getElementById('filter-edge-days').value;
const nodeDays = document.getElementById('filter-node-days').value;
const edgeParams = new URLSearchParams();
if (evidence === 'multibyte') edgeParams.set('evidence', 'multibyte');
if (evidence === 'multibyte' || evidence === 'neighbors') edgeParams.set('evidence', evidence);
if (edgeDays) edgeParams.set('days', edgeDays);
if (forceRefresh) edgeParams.set('refresh', '1');
const nodeParams = new URLSearchParams();
@@ -1763,7 +1786,7 @@
if (filters.starredOnly !== undefined) {
document.getElementById('filter-starred').checked = filters.starredOnly;
}
if (filters.evidence === 'all' || filters.evidence === 'multibyte') {
if (['all', 'multibyte', 'neighbors'].includes(filters.evidence)) {
document.getElementById('filter-evidence').value = filters.evidence;
}
if (filters.colorMode === 'age' || filters.colorMode === 'heat') {
@@ -2135,6 +2158,10 @@
<td style="padding: 2px 8px 2px 0; color: #888;"><strong>Evidence:</strong></td>
<td style="padding: 2px 0;">${evidenceLabel}</td>
</tr>` : ''}
${isNeighborEdge(edge) && formatSnr(edge.best_snr) ? `<tr>
<td style="padding: 2px 8px 2px 0; color: #888;"><strong>Best SNR:</strong></td>
<td style="padding: 2px 0;">${formatSnr(edge.best_snr)}${formatSnr(edge.last_snr) ? ` (last ${formatSnr(edge.last_snr)})` : ''}</td>
</tr>` : ''}
<tr>
<td style="padding: 2px 8px 2px 0; color: #888;"><strong>Distance:</strong></td>
<td style="padding: 2px 0;">${edgeDistance ? edgeDistance.toFixed(2) + ' km' : 'N/A'}</td>
@@ -3151,7 +3178,12 @@
// Shared edge width: log-scaled, capped so heavy edges don't blanket dense areas
function getEdgeWeight(edge) {
return Math.max(1, Math.min(5, Math.log10(Math.max(1, edge.observation_count || 1)) * 1.5));
const weight = Math.max(1, Math.min(5, Math.log10(Math.max(1, edge.observation_count || 1)) * 1.5));
// Confirmed zero-hop links are the strongest evidence class, but they
// accumulate observations slowly (one per cycle, cycles are >=12h apart).
// Without a floor an inferred edge seen thousands of times would always
// outweigh a directly measured one, which inverts the confidence order.
return isNeighborEdge(edge) ? Math.max(weight, 3) : weight;
}
function getMapEdgeWeight(edge) {
+2 -1
View File
@@ -33,7 +33,8 @@ dependencies = [
"maidenhead>=1.4.0",
"pytz>=2023.3",
"aiohttp>=3.8.0",
"meshcore>=2.3.6",
# 2.3.8+ required for zero-hop neighbour discovery (see requirements.txt)
"meshcore>=2.3.8",
"openmeteo-requests>=1.7.2",
"requests-cache>=1.1.1",
"retry-requests>=1.0.0",
+3 -1
View File
@@ -13,7 +13,9 @@ geopy>=2.3.0
maidenhead>=1.4.0
pytz>=2023.3
aiohttp>=3.8.0
meshcore>=2.3.6
# 2.3.8+ is required, not merely preferred: send_node_discover_req / req_regions_sync
# (zero-hop neighbour discovery) and the bounded, serialised BLE write both landed there.
meshcore>=2.3.8
openmeteo-requests>=1.7.2
requests-cache>=1.1.1
retry-requests>=1.0.0
+201
View File
@@ -0,0 +1,201 @@
"""Tests for the neighbors command (modules/commands/neighbors_command.py)."""
from __future__ import annotations
import asyncio
import types
import pytest
from modules.commands.neighbors_command import NeighborsCommand
from tests.conftest import mock_message
def make_service(*, neighbors_enabled=True, summary=None, hang=False,
raises=None, cycle_budget=5.0):
"""A stand-in for PacketCaptureService's neighbors surface."""
service = types.SimpleNamespace()
service.neighbors_enabled = neighbors_enabled
service.neighbors_config = types.SimpleNamespace(
discover_window=60.0, cycle_budget=cycle_budget
)
service.calls = 0
async def run_cycle():
service.calls += 1
if hang:
await asyncio.sleep(30)
if raises is not None:
raise raises
return summary or {"ok": True, "queried": 0, "recorded": 0, "attempted": 0}
service.run_neighbors_cycle = run_cycle
return service
def make_command(command_mock_bot, service, *, enabled=True):
"""Build the command without __init__ (which reads config and the ACL)."""
command_mock_bot.packet_capture_service = service
command = object.__new__(NeighborsCommand)
command.bot = command_mock_bot
command.logger = command_mock_bot.logger
command.command_enabled = enabled
command._cycle_task = None
sent: list[str] = []
async def send_response(message, content, **kwargs):
sent.append(content)
return True
command.send_response = send_response
command.translate = lambda key, **kwargs: (
key.split(".")[-1] + (" " + " ".join(f"{k}={v}" for k, v in sorted(kwargs.items()))
if kwargs else "")
)
return command, sent
@pytest.fixture
def message():
return mock_message(content="neighbors", is_dm=True, sender_id="TestUser")
def test_command_metadata():
# A cycle spends airtime and takes a minute, so it is DM-only with a cooldown.
assert NeighborsCommand.requires_dm is True
assert NeighborsCommand.cooldown_seconds == 900
assert "neighbours" in NeighborsCommand.keywords
async def test_reports_when_discovery_is_disabled(command_mock_bot, message):
command, sent = make_command(command_mock_bot, make_service(neighbors_enabled=False))
assert await command.execute(message) is True
assert sent == ["disabled"]
async def test_reports_when_the_service_is_absent(command_mock_bot, message):
command, sent = make_command(command_mock_bot, None)
command_mock_bot.services = {}
assert await command.execute(message) is True
assert sent == ["disabled"]
async def test_acknowledges_before_running_the_cycle(command_mock_bot, message):
"""The listen window is ~60s, far too long to hold the reply open."""
service = make_service(summary={"ok": True, "queried": 2, "best_snr": 8.0,
"recorded": 2, "attempted": 0})
command, sent = make_command(command_mock_bot, service)
assert await command.execute(message) is True
assert sent == ["started seconds=60"]
assert service.calls == 0 # not awaited inline
await command._cycle_task
assert len(sent) == 2
assert sent[1].startswith("success")
async def test_summary_reports_count_snr_and_records(command_mock_bot, message):
service = make_service(summary={"ok": True, "queried": 3, "best_snr": 8.25,
"recorded": 3, "attempted": 0})
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert "count=3" in sent[1]
assert "recorded=3" in sent[1]
assert "best_snr=8.2dB" in sent[1]
async def test_summary_mentions_brokers_only_when_one_was_tried(command_mock_bot, message):
"""A operator with no MQTT should not be shown a confusing 0/0."""
without = make_service(summary={"ok": True, "queried": 1, "best_snr": 1.0,
"recorded": 1, "attempted": 0})
command, sent = make_command(command_mock_bot, without)
await command.execute(message)
await command._cycle_task
assert "published" not in sent[1]
with_broker = make_service(summary={"ok": True, "queried": 1, "best_snr": 1.0,
"recorded": 1, "attempted": 2, "succeeded": 1})
command2, sent2 = make_command(command_mock_bot, with_broker)
await command2.execute(message)
await command2._cycle_task
assert "published" in sent2[1]
assert "succeeded=1" in sent2[1]
assert "attempted=2" in sent2[1]
async def test_reports_when_nothing_answered(command_mock_bot, message):
service = make_service(summary={"ok": True, "queried": 0, "recorded": 0, "attempted": 0})
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert sent[1] == "none"
async def test_reports_the_reason_a_cycle_did_not_run(command_mock_bot, message):
service = make_service(summary={"ok": False, "reason": "radio not connected"})
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert sent[1] == "failed reason=radio not connected"
async def test_missing_snr_does_not_break_the_summary(command_mock_bot, message):
service = make_service(summary={"ok": True, "queried": 1, "best_snr": None,
"recorded": 1, "attempted": 0})
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert "best_snr=n/a" in sent[1]
async def test_cycle_errors_are_reported(command_mock_bot, message):
service = make_service(raises=RuntimeError("radio exploded"))
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert sent[1].startswith("error")
assert "radio exploded" in sent[1]
async def test_a_stalled_cycle_is_abandoned(command_mock_bot, message):
service = make_service(hang=True, cycle_budget=0.05)
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
assert sent[1].startswith("error")
assert "timed out" in sent[1]
async def test_a_second_request_will_not_overlap_the_first(command_mock_bot, message):
"""Two concurrent discover rounds would collect into each other's window."""
service = make_service(hang=True, cycle_budget=5.0)
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command.execute(message)
assert sent == ["started seconds=60", "busy"]
command._cycle_task.cancel()
with pytest.raises(asyncio.CancelledError):
await command._cycle_task
async def test_a_finished_cycle_does_not_block_the_next_request(command_mock_bot, message):
service = make_service(summary={"ok": True, "queried": 0, "recorded": 0, "attempted": 0})
command, sent = make_command(command_mock_bot, service)
await command.execute(message)
await command._cycle_task
await command.execute(message)
await command._cycle_task
assert service.calls == 2
assert "busy" not in sent
def test_disabled_command_cannot_execute(command_mock_bot, message):
command, _ = make_command(command_mock_bot, make_service(), enabled=False)
assert command.can_execute(message) is False
+76
View File
@@ -392,3 +392,79 @@ class TestSchema:
cursor = conn.cursor()
assert _column_exists(cursor, "purging_log", "details") is True
# ---------------------------------------------------------------------------
# TestNeighborTables (migration 22)
# ---------------------------------------------------------------------------
class TestNeighborTables:
"""Zero-hop neighbor discovery tables (see modules/neighbors_discovery.py)."""
def test_neighbor_tables_created(self, runner, conn):
runner.run()
for table in ["neighbor_links", "neighbor_observations"]:
cur = conn.execute(
"SELECT name FROM sqlite_master WHERE type='table' AND name=?",
(table,),
)
assert cur.fetchone() is not None
def test_neighbor_links_keeps_full_public_keys(self, runner, conn):
"""Full 32-byte keys, not prefixes — that is the point of this evidence."""
runner.run()
cursor = conn.cursor()
for column in ["self_public_key", "neighbor_public_key", "best_snr",
"last_snr", "snr_sum", "snr_count", "observation_count",
"first_seen", "last_seen", "last_status", "scopes"]:
assert _column_exists(cursor, "neighbor_links", column) is True
def test_neighbor_observations_columns(self, runner, conn):
runner.run()
cursor = conn.cursor()
for column in ["observed_at", "self_public_key", "neighbor_public_key",
"snr", "heard_secs_ago", "scopes", "status"]:
assert _column_exists(cursor, "neighbor_observations", column) is True
def test_neighbor_indexes_are_table_qualified(self, runner, conn):
"""SQLite index names are database-global (see migration 20)."""
runner.run()
for idx, table in [
("idx_neighbor_links_last_seen", "neighbor_links"),
("idx_neighbor_links_neighbor", "neighbor_links"),
("idx_neighbor_observations_observed_at", "neighbor_observations"),
("idx_neighbor_observations_neighbor", "neighbor_observations"),
]:
row = conn.execute(
"SELECT tbl_name FROM sqlite_master WHERE type='index' AND name=?",
(idx,),
).fetchone()
assert row is not None, f"missing index {idx}"
assert row[0] == table
def test_neighbor_links_is_unique_per_directed_pair(self, runner, conn):
runner.run()
conn.execute(
"INSERT INTO neighbor_links (self_public_key, neighbor_public_key) VALUES ('a', 'b')"
)
with pytest.raises(sqlite3.IntegrityError):
conn.execute(
"INSERT INTO neighbor_links (self_public_key, neighbor_public_key) VALUES ('a', 'b')"
)
def test_migration_is_idempotent(self, conn, logger):
MigrationRunner(conn, logger).run()
MigrationRunner(conn, logger).run()
applied = conn.execute(
"SELECT COUNT(*) FROM schema_version WHERE version = 22"
).fetchone()[0]
assert applied == 1
def test_neighbor_tables_are_writable_by_the_db_manager(self, runner, conn):
"""DBManager.ALLOWED_TABLES gates create/drop/retention helpers."""
from modules.db_manager import DBManager
runner.run()
assert "neighbor_links" in DBManager.ALLOWED_TABLES
assert "neighbor_observations" in DBManager.ALLOWED_TABLES
+138
View File
@@ -135,3 +135,141 @@ def test_retention_settings_are_configurable_and_bounded():
config.set("Data_Retention", "retention_delete_pause_seconds", "9")
assert retention_delete_settings(config) == (10_000, 5.0)
# ---------------------------------------------------------------------------
# Neighbor observation retention (modules/maintenance.py)
# ---------------------------------------------------------------------------
def _utc_now():
"""Clock callable MaintenanceRunner takes for injection."""
import datetime
return datetime.datetime.now(datetime.timezone.utc)
def _maintenance_with_db(tmp_path, config, db_path=None):
"""A MaintenanceRunner bound to a real migrated database."""
import logging
import sqlite3
from contextlib import contextmanager
from modules.db_migrations import MigrationRunner
from modules.maintenance import MaintenanceRunner
logger = logging.getLogger("test-neighbor-retention")
path = db_path or (tmp_path / "retention_neighbors.db")
with closing(sqlite3.connect(path)) as conn:
MigrationRunner(conn, logger).run()
class DBManager:
@contextmanager
def connection(self):
with closing(sqlite3.connect(path)) as conn:
conn.row_factory = sqlite3.Row
yield conn
def delete_timestamp_rows_in_chunks(self, table, column, cutoff, **kwargs):
return delete_timestamp_rows_in_chunks(
self.connection, table, column, cutoff, **kwargs
)
bot = Mock()
bot.config = config
bot.logger = logger
bot.db_manager = DBManager()
return MaintenanceRunner(bot, _utc_now), bot.db_manager, path
def _seed_observations(path, stamps):
import sqlite3
with closing(sqlite3.connect(path)) as conn:
conn.executemany(
"""
INSERT INTO neighbor_observations
(observed_at, self_public_key, neighbor_public_key, snr,
heard_secs_ago, scopes, status)
VALUES (?, 'ff', 'aa', 1.0, 0, '', 'responded')
""",
[(s,) for s in stamps],
)
conn.commit()
def _observation_count(path):
import sqlite3
with closing(sqlite3.connect(path)) as conn:
return conn.execute("SELECT COUNT(*) FROM neighbor_observations").fetchone()[0]
def test_neighbor_observation_retention_deletes_only_old_rows(tmp_path):
import datetime
maintenance, _, path = _maintenance_with_db(tmp_path, ConfigParser())
now = datetime.datetime.now(datetime.timezone.utc)
recent = (now - datetime.timedelta(days=5)).isoformat()
ancient = (now - datetime.timedelta(days=500)).isoformat()
_seed_observations(path, [recent, ancient, ancient])
maintenance._cleanup_neighbor_observations(365)
assert _observation_count(path) == 1
def test_neighbor_observation_retention_disabled_when_not_positive(tmp_path):
import datetime
maintenance, _, path = _maintenance_with_db(tmp_path, ConfigParser())
ancient = (
datetime.datetime.now(datetime.timezone.utc) - datetime.timedelta(days=5000)
).isoformat()
_seed_observations(path, [ancient])
maintenance._cleanup_neighbor_observations(0)
assert _observation_count(path) == 1
def test_neighbor_observation_retention_survives_a_pre_migration_database(tmp_path):
"""A database without migration 22 must not abort the whole retention run."""
import logging
import sqlite3
from contextlib import contextmanager
from modules.maintenance import MaintenanceRunner
path = tmp_path / "old.db"
sqlite3.connect(path).close()
class DBManager:
@contextmanager
def connection(self):
with closing(sqlite3.connect(path)) as conn:
yield conn
def delete_timestamp_rows_in_chunks(self, table, column, cutoff, **kwargs):
return delete_timestamp_rows_in_chunks(
self.connection, table, column, cutoff, **kwargs
)
bot = Mock()
bot.config = ConfigParser()
bot.logger = logging.getLogger("test-neighbor-retention")
bot.db_manager = DBManager()
# Must not raise.
MaintenanceRunner(bot, _utc_now)._cleanup_neighbor_observations(365)
def test_neighbor_observation_retention_tolerates_a_missing_db_manager(tmp_path):
import logging
from modules.maintenance import MaintenanceRunner
bot = Mock()
bot.config = ConfigParser()
bot.logger = logging.getLogger("test-neighbor-retention")
bot.db_manager = None
MaintenanceRunner(bot, _utc_now)._cleanup_neighbor_observations(365)
+201
View File
@@ -0,0 +1,201 @@
"""Mesh-graph edges derived from confirmed zero-hop neighbor discovery.
Covers BotDataViewer._compute_neighbor_evidence_edges and friends, which back
GET /api/mesh/edges?evidence=neighbors.
"""
from __future__ import annotations
import contextlib
import logging
import sqlite3
import pytest
from modules.db_migrations import MigrationRunner
from modules.web_viewer.app import BotDataViewer
LOGGER = logging.getLogger("test-neighbor-edges")
SELF_KEY = "ff" * 32
KEY_A = "aa" * 32
KEY_B = "bb" * 32
RECENT = "2026-08-01T00:00:00+00:00"
ANCIENT = "2020-01-01T00:00:00+00:00"
class ViewerStub:
"""Just the db plumbing the derivation methods need.
Binding the real functions keeps this a test of production code rather than a
reimplementation; BotDataViewer.__init__ would want a Flask app and a bot.
"""
logger = LOGGER
NEIGHBOR_PREFIX_HEX_CHARS = BotDataViewer.NEIGHBOR_PREFIX_HEX_CHARS
_compute_neighbor_evidence_edges = BotDataViewer._compute_neighbor_evidence_edges
_derive_neighbor_evidence_graph = BotDataViewer._derive_neighbor_evidence_graph
_neighbor_evidence_edge_keys = BotDataViewer._neighbor_evidence_edge_keys
_filter_multibyte_evidence_edges = staticmethod(
BotDataViewer._filter_multibyte_evidence_edges
)
def __init__(self, path):
self._path = path
@contextlib.contextmanager
def _with_db_connection(self):
conn = sqlite3.connect(self._path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def insert_link(conn, self_key, neighbor_key, *, observations=1, snr_sum=5.0,
snr_count=1, best_snr=5.0, last_snr=5.0, first_seen=RECENT,
last_seen=RECENT):
conn.execute(
"""
INSERT INTO neighbor_links
(self_public_key, neighbor_public_key, first_seen, last_seen,
observation_count, snr_sum, snr_count, best_snr, last_snr,
last_status, scopes)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'responded', '')
""",
(self_key, neighbor_key, first_seen, last_seen, observations,
snr_sum, snr_count, best_snr, last_snr),
)
@pytest.fixture
def viewer(tmp_path):
path = tmp_path / "viewer.db"
conn = sqlite3.connect(path)
MigrationRunner(conn, LOGGER).run()
conn.close()
return ViewerStub(path)
@pytest.fixture
def seeded(viewer):
with viewer._with_db_connection() as conn:
insert_link(conn, SELF_KEY, KEY_A, observations=4, snr_sum=30.0,
snr_count=4, best_snr=9.5, last_snr=7.0,
first_seen="2026-01-01T00:00:00+00:00", last_seen=RECENT)
insert_link(conn, SELF_KEY, KEY_B, observations=1, snr_sum=-3.0,
snr_count=1, best_snr=-3.0, last_snr=-3.0,
last_seen=ANCIENT)
conn.commit()
return viewer
def find_edge(edges, from_key, to_key):
chars = ViewerStub.NEIGHBOR_PREFIX_HEX_CHARS
matches = [
e for e in edges
if e["from_prefix"] == from_key[:chars] and e["to_prefix"] == to_key[:chars]
]
assert len(matches) == 1
return matches[0]
def test_each_link_yields_both_directions(seeded):
"""A discover response proves both directions of the link."""
edges = seeded._compute_neighbor_evidence_edges()
assert len(edges) == 4
find_edge(edges, SELF_KEY, KEY_A)
find_edge(edges, KEY_A, SELF_KEY)
def test_edges_carry_full_public_keys(seeded):
"""The multi-byte path derivation cannot supply these — paths have no keys."""
edge = find_edge(seeded._compute_neighbor_evidence_edges(), SELF_KEY, KEY_A)
assert edge["from_public_key"] == SELF_KEY
assert edge["to_public_key"] == KEY_A
def test_edges_report_measured_snr(seeded):
edge = find_edge(seeded._compute_neighbor_evidence_edges(), SELF_KEY, KEY_A)
assert edge["snr"] == pytest.approx(30.0 / 4)
assert edge["best_snr"] == 9.5
assert edge["last_snr"] == 7.0
def test_edges_are_tagged_and_treated_as_first_hop(seeded):
edge = find_edge(seeded._compute_neighbor_evidence_edges(), SELF_KEY, KEY_A)
assert edge["evidence"] == "neighbors"
# A direct link is the first hop of any path crossing it.
assert edge["avg_hop_position"] == 1.0
def test_edges_preserve_lifetime_counts_and_timestamps(seeded):
edge = find_edge(seeded._compute_neighbor_evidence_edges(), SELF_KEY, KEY_A)
assert edge["observation_count"] == 4
assert edge["first_seen"] == "2026-01-01T00:00:00+00:00"
assert edge["last_seen"] == RECENT
def test_missing_snr_samples_yield_null_mean(viewer):
with viewer._with_db_connection() as conn:
insert_link(conn, SELF_KEY, KEY_A, snr_count=0, snr_sum=0.0, best_snr=None)
conn.commit()
edge = find_edge(viewer._compute_neighbor_evidence_edges(), SELF_KEY, KEY_A)
assert edge["snr"] is None
def test_days_filter_drops_stale_links(seeded):
edges, prefix_hex_chars = seeded._derive_neighbor_evidence_graph(days=30)
assert prefix_hex_chars == ViewerStub.NEIGHBOR_PREFIX_HEX_CHARS
# Only the recent link survives, in both directions.
assert len(edges) == 2
assert {e["to_prefix"] for e in edges} | {e["from_prefix"] for e in edges} == {
SELF_KEY[:6], KEY_A[:6]
}
def test_min_observations_filter(seeded):
edges, _ = seeded._derive_neighbor_evidence_graph(min_observations=3)
assert len(edges) == 2
assert all(e["observation_count"] >= 3 for e in edges)
def test_unfiltered_returns_everything(seeded):
edges, _ = seeded._derive_neighbor_evidence_graph()
assert len(edges) == 4
def test_edge_keys_cover_both_directions(seeded):
"""Used to relabel mesh_connections edges, which lost their provenance."""
keys = seeded._neighbor_evidence_edge_keys()
assert (SELF_KEY[:6], KEY_A[:6]) in keys
assert (KEY_A[:6], SELF_KEY[:6]) in keys
assert (SELF_KEY[:6], KEY_B[:6]) in keys
assert ("11", "22") not in keys
def test_blank_keys_are_skipped(viewer):
with viewer._with_db_connection() as conn:
insert_link(conn, "", KEY_A)
conn.commit()
assert viewer._compute_neighbor_evidence_edges() == []
assert viewer._neighbor_evidence_edge_keys() == set()
def test_empty_database_yields_nothing(viewer):
assert viewer._compute_neighbor_evidence_edges() == []
assert viewer._neighbor_evidence_edge_keys() == set()
def test_pre_migration_database_degrades_quietly(tmp_path):
"""A database without migration 22 simply has no neighbor evidence."""
path = tmp_path / "old.db"
sqlite3.connect(path).close()
stub = ViewerStub(path)
assert stub._compute_neighbor_evidence_edges() == []
assert stub._neighbor_evidence_edge_keys() == set()
edges, chars = stub._derive_neighbor_evidence_graph(days=7)
assert edges == []
assert chars == ViewerStub.NEIGHBOR_PREFIX_HEX_CHARS
+643
View File
@@ -0,0 +1,643 @@
"""Tests for zero-hop neighbor discovery (modules/neighbors_discovery.py).
Ported from the meshcore-packet-capture project's test_neighbors.py, minus the
device-lock tests (the bot serialises radio commands globally in
modules/core.py _SerializedCommands, so the module takes no lock of its own).
"""
from __future__ import annotations
import asyncio
import contextlib
import json
import logging
import sqlite3
import types
import pytest
from meshcore import EventType
from modules import neighbors_discovery as nb
from modules.db_migrations import MigrationRunner
from modules.enums import AdvertFlags
LOGGER = logging.getLogger("test-neighbors")
SELF_KEY = "ff" * 32
KEY_A = "aa" * 32
KEY_B = "bb" * 32
# ---------------------------------------------------------------------------
# Fakes
# ---------------------------------------------------------------------------
class FakeEvent:
def __init__(self, event_type, payload=None):
self.type = event_type
self.payload = payload or {}
class FakeRadio:
"""Delivers DISCOVER_RESPONSE events synchronously from the send call."""
def __init__(self, responses=None, *, send_result="ok", scopes="DEN,APRS",
flood_scope="SEA"):
self.responses = responses or []
self.send_result = send_result
self.scopes = scopes
self.handler = None
self.subscribed = 0
self.unsubscribed = 0
self.sent_tag = None
self.regions_calls = []
self.self_info = {"public_key": SELF_KEY}
# A real contact cache; stage 2 must never populate it.
self.contacts = {}
self.commands = types.SimpleNamespace(
send_node_discover_req=self._send,
req_regions_sync=self._regions,
get_default_flood_scope=self._flood,
)
self._flood_scope = flood_scope
def subscribe(self, event_type, callback):
assert event_type == EventType.DISCOVER_RESPONSE
self.subscribed += 1
self.handler = callback
return "subscription"
def unsubscribe(self, subscription):
self.unsubscribed += 1
async def _send(self, filter_bits, prefix_only=True, tag=None):
self.sent_filter = filter_bits
self.sent_prefix_only = prefix_only
self.sent_tag = tag
if self.send_result == "error":
return FakeEvent(EventType.ERROR, {"reason": "unsupported"})
if self.send_result == "none":
return None
if self.send_result == "hang":
await asyncio.sleep(30)
little_endian = tag.to_bytes(4, "little").hex()
for response in self.responses:
payload = dict(response)
payload.setdefault("tag", little_endian)
await self.handler(FakeEvent(EventType.DISCOVER_RESPONSE, payload))
return FakeEvent(EventType.MSG_SENT, {})
async def _regions(self, pubkey, timeout=0, min_timeout=0):
self.regions_calls.append(pubkey)
if self.scopes == "raise":
raise RuntimeError("device rejected request")
if self.scopes == "hang":
await asyncio.sleep(120)
return self.scopes
async def _flood(self):
return FakeEvent(EventType.SELF_INFO, {"scope_name": self._flood_scope})
def response(pubkey, snr=5.0, node_type=None, **extra):
payload = {
"pubkey": pubkey,
"SNR": snr,
"node_type": AdvertFlags.ADV_TYPE_REPEATER.value if node_type is None else node_type,
}
payload.update(extra)
return payload
def entry(pubkey_char, snr, heard_at, **kwargs):
return nb.NeighborEntry(pubkey=pubkey_char * 64, snr=snr, heard_at=heard_at, **kwargs)
@pytest.fixture
def fast_cfg():
"""Config with the smallest window the module permits (floored at 5s).
Tests that must not wait use monkeypatched sleep instead.
"""
return nb.NeighborsConfig(discover_window=nb.MIN_DISCOVER_WINDOW)
@pytest.fixture
def no_sleep(monkeypatch):
"""Make asyncio.sleep inside the module a no-op so windows are instant."""
async def instant(_seconds):
return None
monkeypatch.setattr(nb.asyncio, "sleep", instant)
# ---------------------------------------------------------------------------
# Interval clamping (firmware band: 12-336h, default 24h)
# ---------------------------------------------------------------------------
@pytest.mark.parametrize(
"given,expected",
[(11, 12), (12, 12), (24, 24), (48, 48), (336, 336), (400, 336), (0, 24), (-5, 24)],
)
def test_clamp_interval_hours(given, expected):
assert nb.clamp_interval_hours(given) == expected
def test_interval_seconds_uses_clamped_value():
assert nb.NeighborsConfig(interval_hours=1).interval_seconds == 12 * 3600
assert nb.NeighborsConfig(interval_hours=9999).interval_seconds == 336 * 3600
def test_config_floors_values_that_would_never_produce_data():
cfg = nb.NeighborsConfig(
discover_window=0, max_neighbors=0, cycle_timeout=1, scope_gap=-3, command_timeout=0
)
assert cfg.discover_window == nb.MIN_DISCOVER_WINDOW
assert cfg.max_neighbors == 1
assert cfg.cycle_timeout == nb.MIN_CYCLE_TIMEOUT
assert cfg.scope_gap == 0.0
# wait_for(timeout=0) raises immediately, which would break every cycle.
assert cfg.command_timeout == nb.MIN_COMMAND_TIMEOUT
def test_scope_request_budget_exceeds_library_msg_sent_wait():
cfg = nb.NeighborsConfig(scope_min_timeout=8.0, command_timeout=20.0)
assert cfg.scope_request_budget > nb.LIBRARY_MSG_SENT_TIMEOUT
def test_cycle_budget_is_smaller_when_scopes_are_disabled():
off = nb.NeighborsConfig(collect_scopes=False)
on = nb.NeighborsConfig(collect_scopes=True)
assert off.cycle_budget < on.cycle_budget
# Must still cover the listen window plus the discover request.
assert off.cycle_budget >= off.discover_window + off.command_timeout
def test_discover_filter_targets_repeaters():
# A bitmask over advert types, so the bit index is the type value.
assert 1 << AdvertFlags.ADV_TYPE_REPEATER.value == nb.DISCOVER_FILTER_REPEATER
assert nb.DISCOVER_FILTER_REPEATER == 0x04
# ---------------------------------------------------------------------------
# Ordering: most recently heard, then stronger SNR, then pubkey
# ---------------------------------------------------------------------------
def test_sort_prefers_most_recently_heard():
older = entry("a", 20.0, 100.0)
newer = entry("b", -5.0, 200.0)
assert nb.sort_entries([older, newer], now=200.0) == [newer, older]
def test_sort_breaks_recency_tie_by_snr():
weak = entry("a", 1.0, 100.0)
strong = entry("b", 9.0, 100.0)
assert nb.sort_entries([weak, strong], now=100.0) == [strong, weak]
def test_sort_breaks_full_tie_by_pubkey_ascending():
first = entry("1", 5.0, 100.0)
second = entry("2", 5.0, 100.0)
assert nb.sort_entries([second, first], now=100.0) == [first, second]
def test_sort_compares_whole_seconds_so_snr_can_break_ties():
"""Recency is compared as the *published* heard_secs_ago.
Sorting on the raw float clock would quantise differently from the field we
publish, so SNR could never break a tie and the output could be non-monotonic
in heard_secs_ago. That order decides which entries survive the size budget.
"""
slightly_newer_weak = entry("a", 1.0, 100.4)
slightly_older_strong = entry("b", 9.0, 100.0)
ordered = nb.sort_entries([slightly_newer_weak, slightly_older_strong], now=100.9)
assert [e.heard_secs_ago(100.9) for e in ordered] == [0, 0]
assert ordered[0] is slightly_older_strong
# ---------------------------------------------------------------------------
# Payload shape and tail-drop
# ---------------------------------------------------------------------------
def test_message_matches_firmware_contract():
item = nb.NeighborEntry(pubkey=KEY_A, snr=9.75, heard_at=100.0,
scopes="DEN,APRS", status=nb.STATUS_RESPONDED)
message, dropped = nb.build_neighbors_message(
"MeshCore-HOWL", "A1B2", "DEN", [item], timestamp="TS", now=142.0
)
assert dropped == 0
# Key order is part of the contract (MQTTPayloadBuilder.cpp).
assert list(message.keys()) == [
"timestamp", "origin", "origin_id", "total_neighbors",
"queried_neighbors", "truncated", "self", "neighbors",
]
assert message["self"] == {"scopes": "DEN"}
assert message["neighbors"] == [{
"pubkey": KEY_A.upper(),
"snr": 9.75,
"heard_secs_ago": 42,
"scopes": "DEN,APRS",
"status": "responded",
}]
def test_message_reports_truncation_from_the_max_cap():
items = [entry("a", 1.0, 100.0)]
message, dropped = nb.build_neighbors_message(
"o", "i", "", items, timestamp="TS", now=100.0, total_neighbors=9
)
assert dropped == 0
assert message["total_neighbors"] == 9
assert message["queried_neighbors"] == 1
assert message["truncated"] is True
def test_message_drops_the_tail_past_the_size_budget():
items = [nb.NeighborEntry(pubkey=f"{i:064x}", snr=float(i), heard_at=100.0)
for i in range(300)]
message, dropped = nb.build_neighbors_message(
"o", "i", "", items, timestamp="TS", now=100.0
)
assert dropped > 0
assert len(message["neighbors"]) == len(items) - dropped
assert len(json.dumps(message)) < nb.NEIGHBORS_JSON_BUDGET
assert message["truncated"] is True
def test_message_publishes_pubkeys_uppercase():
message, _ = nb.build_neighbors_message(
"o", "i", "", [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)],
timestamp="TS", now=0.0,
)
assert message["neighbors"][0]["pubkey"] == KEY_A.upper()
# ---------------------------------------------------------------------------
# Stage 1: discovery
# ---------------------------------------------------------------------------
async def test_discover_collects_and_sorts_responses(fast_cfg, no_sleep):
radio = FakeRadio([response(KEY_A, 3.0), response(KEY_B, 9.0)])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert [e.pubkey for e in entries] == [KEY_B, KEY_A]
assert radio.sent_filter == nb.DISCOVER_FILTER_REPEATER
# Full 32-byte pubkeys are required, so prefix_only must be off.
assert radio.sent_prefix_only is False
async def test_discover_dedupes_keeping_strongest_snr(fast_cfg, no_sleep):
radio = FakeRadio([response(KEY_A, 3.0), response(KEY_A, 9.0), response(KEY_A, 5.0)])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert len(entries) == 1
assert entries[0].snr == 9.0
async def test_discover_accepts_uppercase_pubkeys_as_lowercase(fast_cfg, no_sleep):
radio = FakeRadio([response(KEY_A.upper(), 1.0)])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert entries[0].pubkey == KEY_A
@pytest.mark.parametrize("bad", [
pytest.param({"pubkey": "aa" * 10}, id="short-pubkey"),
pytest.param({"node_type": AdvertFlags.ADV_TYPE_CHAT.value}, id="not-a-repeater"),
pytest.param({"tag": "deadbeef"}, id="wrong-tag"),
])
async def test_discover_rejects_unwanted_responses(fast_cfg, no_sleep, bad):
payload = response(KEY_A, 5.0)
payload.update(bad)
radio = FakeRadio([payload])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert entries == []
async def test_discover_filters_out_self(fast_cfg, no_sleep):
radio = FakeRadio([response(SELF_KEY, 9.0), response(KEY_A, 1.0)])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert [e.pubkey for e in entries] == [KEY_A]
async def test_discover_tag_is_known_before_subscribing(fast_cfg, no_sleep):
"""The tag must exist before the handler goes live.
Otherwise a response arriving between subscribe and send-completion would be
accepted with no tag check, letting a stale or foreign round leak in.
"""
radio = FakeRadio([response(KEY_A, 1.0)])
await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert radio.sent_tag is not None
assert 1 <= radio.sent_tag <= 0xFFFFFFFF
@pytest.mark.parametrize("result", ["error", "none"])
async def test_discover_returns_none_when_send_fails(fast_cfg, no_sleep, result):
radio = FakeRadio([], send_result=result)
assert await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER) is None
async def test_discover_bounds_a_stalled_send():
"""A stalled BLE/serial write must not hang the cycle."""
cfg = nb.NeighborsConfig(command_timeout=nb.MIN_COMMAND_TIMEOUT)
radio = FakeRadio([], send_result="hang")
assert await nb.discover_neighbors(radio, cfg, SELF_KEY, LOGGER) is None
async def test_discover_abandons_cycle_when_session_was_reset(fast_cfg, no_sleep):
"""A reconnect tears down subscriptions, so the collector goes deaf.
Without this check the cycle would record "0 neighbours" as though the mesh
were empty.
"""
radio = FakeRadio([response(KEY_A, 1.0)])
result = await nb.discover_neighbors(
radio, fast_cfg, SELF_KEY, LOGGER, still_valid=lambda: False
)
assert result is None
async def test_discover_always_unsubscribes(fast_cfg, no_sleep):
radio = FakeRadio([response(KEY_A, 1.0)])
await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert radio.subscribed == 1
assert radio.unsubscribed == 1
failing = FakeRadio([], send_result="error")
await nb.discover_neighbors(failing, fast_cfg, SELF_KEY, LOGGER)
assert failing.unsubscribed == 1
async def test_discover_ignores_stragglers_after_the_window(fast_cfg, no_sleep):
"""Callbacks are spawned as tasks, so one can land after unsubscribe returns."""
radio = FakeRadio([response(KEY_A, 1.0)])
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
await radio.handler(FakeEvent(EventType.DISCOVER_RESPONSE, response(KEY_B, 9.0)))
assert [e.pubkey for e in entries] == [KEY_A]
async def test_discover_survives_an_unsubscribe_error(fast_cfg, no_sleep):
radio = FakeRadio([response(KEY_A, 1.0)])
radio.unsubscribe = lambda s: (_ for _ in ()).throw(RuntimeError("gone"))
entries = await nb.discover_neighbors(radio, fast_cfg, SELF_KEY, LOGGER)
assert [e.pubkey for e in entries] == [KEY_A]
# ---------------------------------------------------------------------------
# Stage 2: scopes
# ---------------------------------------------------------------------------
async def test_collect_scopes_disabled_marks_entries_responded():
"""Stage 2 off must not report a live neighbour as unreachable."""
cfg = nb.NeighborsConfig(collect_scopes=False)
radio = FakeRadio()
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].status == nb.STATUS_RESPONDED
assert entries[0].scopes == ""
# And it must not spend any airtime.
assert radio.regions_calls == []
async def test_collect_scopes_populates_scopes_when_enabled():
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0)
radio = FakeRadio(scopes="DEN,APRS")
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].scopes == "DEN,APRS"
assert entries[0].status == nb.STATUS_RESPONDED
assert radio.regions_calls == [KEY_A]
async def test_collect_scopes_reports_timeout_when_response_is_none():
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0)
radio = FakeRadio(scopes=None)
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].status == nb.STATUS_TIMEOUT
async def test_collect_scopes_reports_send_failed_on_exception():
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0)
radio = FakeRadio(scopes="raise")
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].status == nb.STATUS_SEND_FAILED
async def test_collect_scopes_bounds_a_stalled_request(monkeypatch):
"""A hung request must be cut off, not left holding the radio."""
# scope_request_budget is floored by the library's 15s MSG_SENT wait, so patch
# that constant instead of waiting it out.
monkeypatch.setattr(nb, "LIBRARY_MSG_SENT_TIMEOUT", 0.01)
cfg = nb.NeighborsConfig(
collect_scopes=True, scope_gap=0, scope_min_timeout=0.01,
command_timeout=nb.MIN_COMMAND_TIMEOUT,
)
cfg.scope_timeout = 0.01
radio = FakeRadio(scopes="hang")
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].status == nb.STATUS_SEND_FAILED
async def test_collect_scopes_stops_at_the_cycle_budget(monkeypatch):
"""Unreached neighbours keep the timeout status, matching the firmware."""
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0, cycle_timeout=nb.MIN_CYCLE_TIMEOUT)
radio = FakeRadio(scopes="DEN")
entries = [nb.NeighborEntry(pubkey=k, snr=1.0, heard_at=0.0) for k in (KEY_A, KEY_B)]
clock = {"t": 0.0}
monkeypatch.setattr(nb.time, "time", lambda: clock["t"])
real_regions = radio._regions
async def advance_then_answer(pubkey, timeout=0, min_timeout=0):
clock["t"] += 1000.0 # blow the budget after the first request
return await real_regions(pubkey, timeout, min_timeout)
radio.commands.req_regions_sync = advance_then_answer
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert entries[0].status == nb.STATUS_RESPONDED
assert entries[1].status == nb.STATUS_TIMEOUT
assert len(radio.regions_calls) == 1
async def test_collect_scopes_reraises_cancellation():
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0)
radio = FakeRadio()
async def cancelled(pubkey, timeout=0, min_timeout=0):
raise asyncio.CancelledError
radio.commands.req_regions_sync = cancelled
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
with pytest.raises(asyncio.CancelledError):
await nb.collect_scopes(radio, entries, cfg, LOGGER)
async def test_collect_scopes_never_populates_the_contact_cache():
"""Zero-hop probing depends on the neighbour not being a known contact.
Nothing in this module may add one; the bot's own contact management is a
separate concern and is why scope collection defaults off.
"""
cfg = nb.NeighborsConfig(collect_scopes=True, scope_gap=0)
radio = FakeRadio(scopes="DEN")
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)]
await nb.collect_scopes(radio, entries, cfg, LOGGER)
assert radio.contacts == {}
async def test_collect_scopes_handles_no_entries():
await nb.collect_scopes(FakeRadio(), [], nb.NeighborsConfig(collect_scopes=True), LOGGER)
# ---------------------------------------------------------------------------
# Self scopes
# ---------------------------------------------------------------------------
async def test_self_scopes_override_wins_even_when_stage_two_is_off():
cfg = nb.NeighborsConfig(collect_scopes=False, self_scopes="OVERRIDE")
assert await nb.fetch_self_scopes(FakeRadio(), cfg, LOGGER) == "OVERRIDE"
async def test_self_scopes_skipped_when_stage_two_is_off():
"""No scope is published for anyone, so spending a command on ours is wrong."""
radio = FakeRadio(flood_scope="SEA")
cfg = nb.NeighborsConfig(collect_scopes=False)
assert await nb.fetch_self_scopes(radio, cfg, LOGGER) == ""
async def test_self_scopes_read_from_device_when_stage_two_is_on():
radio = FakeRadio(flood_scope="SEA")
cfg = nb.NeighborsConfig(collect_scopes=True)
assert await nb.fetch_self_scopes(radio, cfg, LOGGER) == "SEA"
async def test_self_scopes_tolerates_a_build_without_the_command():
radio = FakeRadio()
radio.commands = types.SimpleNamespace()
cfg = nb.NeighborsConfig(collect_scopes=True)
assert await nb.fetch_self_scopes(radio, cfg, LOGGER) == ""
async def test_self_scopes_tolerates_a_missing_device_handle():
cfg = nb.NeighborsConfig(collect_scopes=True)
assert await nb.fetch_self_scopes(types.SimpleNamespace(commands=None), cfg, LOGGER) == ""
# ---------------------------------------------------------------------------
# Persistence
# ---------------------------------------------------------------------------
@pytest.fixture
def db(tmp_path):
"""A real file-based database at the current schema version.
File-based, not :memory:, because each in-memory connection would be a
separate database (matching the project's test_db fixture).
"""
path = tmp_path / "neighbors.db"
conn = sqlite3.connect(path)
MigrationRunner(conn, LOGGER).run()
conn.close()
class Manager:
@contextlib.contextmanager
def connection(self):
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
return Manager()
def _link_rows(db):
with db.connection() as conn:
return [dict(r) for r in conn.execute("SELECT * FROM neighbor_links")]
def test_record_neighbors_writes_history_and_aggregate(db):
entries = [nb.NeighborEntry(pubkey=KEY_A, snr=7.5, heard_at=0.0,
status=nb.STATUS_RESPONDED)]
assert nb.record_neighbors(db, SELF_KEY, entries, LOGGER, observed_at="T1") == 1
rows = _link_rows(db)
assert len(rows) == 1
assert rows[0]["self_public_key"] == SELF_KEY
assert rows[0]["neighbor_public_key"] == KEY_A
assert rows[0]["observation_count"] == 1
assert rows[0]["best_snr"] == 7.5
with db.connection() as conn:
history = [dict(r) for r in conn.execute("SELECT * FROM neighbor_observations")]
assert len(history) == 1
assert history[0]["observed_at"] == "T1"
assert history[0]["status"] == nb.STATUS_RESPONDED
def test_record_neighbors_accumulates_snr_as_sum_and_count(db):
"""Sums, not means, so any later window re-aggregates exactly."""
for snr, stamp in ((5.0, "T1"), (9.0, "T2"), (3.0, "T3")):
nb.record_neighbors(
db, SELF_KEY,
[nb.NeighborEntry(pubkey=KEY_A, snr=snr, heard_at=0.0)],
LOGGER, observed_at=stamp,
)
row = _link_rows(db)[0]
assert row["observation_count"] == 3
assert row["snr_sum"] == pytest.approx(17.0)
assert row["snr_count"] == 3
assert row["snr_sum"] / row["snr_count"] == pytest.approx(17.0 / 3)
assert row["best_snr"] == 9.0
assert row["last_snr"] == 3.0
assert row["first_seen"] == "T1"
assert row["last_seen"] == "T3"
def test_record_neighbors_keeps_previously_learned_scopes(db):
"""A stage-2-disabled cycle must not erase scopes an earlier cycle learned."""
nb.record_neighbors(
db, SELF_KEY,
[nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0, scopes="DEN,APRS")],
LOGGER, observed_at="T1",
)
nb.record_neighbors(
db, SELF_KEY,
[nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0, scopes="")],
LOGGER, observed_at="T2",
)
assert _link_rows(db)[0]["scopes"] == "DEN,APRS"
def test_record_neighbors_stores_keys_lowercase(db):
nb.record_neighbors(
db, SELF_KEY.upper(),
[nb.NeighborEntry(pubkey=KEY_A.upper(), snr=1.0, heard_at=0.0)],
LOGGER,
)
row = _link_rows(db)[0]
assert row["self_public_key"] == SELF_KEY
assert row["neighbor_public_key"] == KEY_A
def test_record_neighbors_noops_without_entries_or_self_key(db):
assert nb.record_neighbors(db, SELF_KEY, [], LOGGER) == 0
assert nb.record_neighbors(
db, "", [nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)], LOGGER
) == 0
assert _link_rows(db) == []
def test_record_neighbors_reports_zero_on_database_error(tmp_path):
class Broken:
@contextlib.contextmanager
def connection(self):
raise sqlite3.OperationalError("database is locked")
yield # pragma: no cover
written = nb.record_neighbors(
Broken(), SELF_KEY,
[nb.NeighborEntry(pubkey=KEY_A, snr=1.0, heard_at=0.0)], LOGGER,
)
assert written == 0
+665
View File
@@ -0,0 +1,665 @@
"""Neighbors wiring inside PacketCaptureService: config, gating, publish, cycle."""
from __future__ import annotations
import asyncio
import configparser
import contextlib
import json
import logging
import sqlite3
import types
from unittest.mock import MagicMock
import pytest
from meshcore import EventType
from modules import neighbors_discovery as nb
from modules.db_migrations import MigrationRunner
from modules.service_plugins.packet_capture_service import (
NEIGHBORS_STATE_KEY,
PacketCaptureService,
)
LOGGER = logging.getLogger("test-pc-neighbors")
SELF_KEY = "ff" * 32
KEY_A = "aa" * 32
KEY_B = "bb" * 32
class FakeEvent:
def __init__(self, event_type, payload=None):
self.type = event_type
self.payload = payload or {}
class FakeRadio:
"""Minimal radio that answers a discover request with canned responses."""
def __init__(self, responses=None, *, commands=("send_node_discover_req",)):
self.responses = responses or []
self.handler = None
self.self_info = {"public_key": SELF_KEY}
self.contacts = {}
available = {}
if "send_node_discover_req" in commands:
available["send_node_discover_req"] = self._send
if "req_regions_sync" in commands:
available["req_regions_sync"] = self._regions
self.commands = types.SimpleNamespace(**available)
def subscribe(self, event_type, callback):
self.handler = callback
return "sub"
def unsubscribe(self, subscription):
return None
async def _send(self, filter_bits, prefix_only=True, tag=None):
little_endian = tag.to_bytes(4, "little").hex()
for item in self.responses:
payload = dict(item)
payload.setdefault("tag", little_endian)
await self.handler(FakeEvent(EventType.DISCOVER_RESPONSE, payload))
return FakeEvent(EventType.MSG_SENT, {})
async def _regions(self, pubkey, timeout=0, min_timeout=0):
return "DEN"
class FakeMqttClient:
def __init__(self, rc=0):
self.published = []
self._rc = rc
def publish(self, topic, payload, qos=0, retain=False):
self.published.append({"topic": topic, "payload": payload,
"qos": qos, "retain": retain})
return types.SimpleNamespace(rc=self._rc)
def response(pubkey, snr=5.0, node_type=2):
return {"pubkey": pubkey, "SNR": snr, "node_type": node_type}
@pytest.fixture
def db_manager(tmp_path):
"""Real migrated database plus in-memory bot_metadata."""
path = tmp_path / "bot.db"
conn = sqlite3.connect(path)
MigrationRunner(conn, LOGGER).run()
conn.close()
class Manager:
def __init__(self):
self.metadata = {}
@contextlib.contextmanager
def connection(self):
conn = sqlite3.connect(path)
conn.row_factory = sqlite3.Row
try:
yield conn
finally:
conn.close()
def get_metadata(self, key):
return self.metadata.get(key)
def set_metadata(self, key, value):
self.metadata[key] = value
return Manager()
def build_service(ini: str, *, db_manager=None, radio=None, connected=True):
"""Construct the service without running __init__.
__init__ installs logging handlers and opens files; the project's other
packet-capture unit tests use this same idiom.
"""
config = configparser.ConfigParser()
config.read_string(ini.strip())
bot = MagicMock()
bot.config = config
bot.logger = LOGGER
bot.connected = connected
bot.meshcore = radio
if db_manager is not None:
bot.db_manager = db_manager
else:
bot.db_manager.get_metadata.return_value = None
service = object.__new__(PacketCaptureService)
service.bot = bot
service.logger = LOGGER
service._load_config()
# Runtime state __init__ sets up alongside _load_config().
service.neighbors_task = None
service.neighbors_capability_state = None
service.neighbors_discover_failures = 0
service.neighbors_topic_warned = set()
service.last_neighbors_publish = service._load_neighbors_state()
service.debug = False
# global_iata deliberately not overridden: it comes from the ini, so tests can
# exercise both a real IATA and the unset sentinel.
service.mqtt_clients = []
service.background_tasks = []
service.should_exit = False
service._get_bot_name = lambda: "TestBot"
return service
BASE_INI = """
[PacketCapture]
enabled = true
neighbors_enabled = true
"""
# ---------------------------------------------------------------------------
# Configuration
# ---------------------------------------------------------------------------
def test_neighbors_defaults_to_disabled():
service = build_service("[PacketCapture]\nenabled = true\n")
assert service.neighbors_enabled is False
# Scope collection is separately off, for radio-lock and contact-path reasons.
assert service.neighbors_config.collect_scopes is False
# Graph feeding is on: the data is only useful if something consumes it.
assert service.neighbors_feed_mesh_graph is True
def test_neighbors_config_reads_every_key():
service = build_service("""
[PacketCapture]
enabled = true
neighbors_enabled = true
neighbors_interval_hours = 48
neighbors_discover_window = 30
neighbors_command_timeout = 11
neighbors_collect_scopes = true
neighbors_scope_timeout = 4
neighbors_scope_min_timeout = 6
neighbors_scope_gap = 1.5
neighbors_cycle_timeout = 300
neighbors_max = 4
neighbors_self_scopes = DEN,APRS
neighbors_feed_mesh_graph = false
""")
cfg = service.neighbors_config
assert (cfg.interval_hours, cfg.discover_window, cfg.command_timeout) == (48, 30.0, 11.0)
assert cfg.collect_scopes is True
assert (cfg.scope_timeout, cfg.scope_min_timeout, cfg.scope_gap) == (4.0, 6.0, 1.5)
assert (cfg.cycle_timeout, cfg.max_neighbors) == (300.0, 4)
assert cfg.self_scopes == "DEN,APRS"
assert service.neighbors_feed_mesh_graph is False
def test_out_of_range_interval_is_clamped_with_a_warning(caplog):
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
service = build_service(BASE_INI + "neighbors_interval_hours = 3\n")
assert service.neighbors_config.interval_hours == nb.MIN_INTERVAL_HOURS
assert "outside the supported" in caplog.text
def test_in_range_interval_does_not_warn(caplog):
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
service = build_service(BASE_INI + "neighbors_interval_hours = 24\n")
assert service.neighbors_config.interval_hours == 24
assert "outside the supported" not in caplog.text
def test_disabled_feature_does_not_warn_about_the_interval(caplog):
"""A stale value in a disabled section is not worth a startup warning."""
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
build_service("[PacketCapture]\nenabled = true\nneighbors_interval_hours = 1\n")
assert "outside the supported" not in caplog.text
# ---------------------------------------------------------------------------
# Per-broker opt-in and topic resolution
# ---------------------------------------------------------------------------
BROKER_INI = BASE_INI + """
iata = SEA
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt2_enabled = true
mqtt2_server = two.example.com
mqtt2_neighbors = false
mqtt3_enabled = true
mqtt3_server = three.example.com
mqtt3_topic_neighbors = custom/{IATA}/{PUBLIC_KEY}/nbrs
"""
def test_broker_neighbors_flag_defaults_on():
"""neighbors_enabled is the single switch; brokers opt out, not in."""
service = build_service(BROKER_INI)
flags = {b["broker_num"]: b["neighbors"] for b in service.mqtt_brokers}
assert flags == {1: True, 2: False, 3: True}
def test_neighbors_topic_derives_from_the_broker_prefix():
service = build_service(BROKER_INI)
broker = service.mqtt_brokers[0]
# Mirrors how topic_packets falls back to <prefix>/packet.
assert service._resolve_neighbors_topic(broker) == "meshcore/packets/neighbors"
def test_neighbors_topic_follows_a_templated_packets_topic():
"""Brokers publish by default, so the derived topic has to be *right*.
A broker configured with meshcore/{IATA}/{PUBLIC_KEY}/packets should get the
matching neighbors topic, not an unrelated flat one.
"""
service = build_service(BASE_INI + """
iata = SEA
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets
""", radio=FakeRadio())
broker = service.mqtt_brokers[0]
assert service._neighbors_topic_template(broker) == "meshcore/{IATA}/{PUBLIC_KEY}/neighbors"
assert service._resolve_neighbors_topic(broker) == (
f"meshcore/SEA/{SELF_KEY.upper()}/neighbors"
)
def test_explicit_topic_wins_over_the_packets_topic():
service = build_service(BASE_INI + """
iata = SEA
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets
mqtt1_topic_neighbors = my/own/topic
""", radio=FakeRadio())
assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) == "my/own/topic"
def test_location_routed_topic_is_refused_without_an_iata():
"""meshcore/XYZ/... would pollute a shared namespace on a community broker."""
service = build_service(BASE_INI + """
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets
""", radio=FakeRadio())
assert service.global_iata == "xyz"
assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) is None
def test_a_flat_derived_topic_still_works_without_an_iata():
"""The guard is about location routing, not about having an IATA at all."""
service = build_service(BASE_INI + """
mqtt1_enabled = true
mqtt1_server = one.example.com
""")
assert service.global_iata == "xyz"
assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) == "meshcore/packets/neighbors"
def test_an_explicit_topic_is_honoured_even_without_an_iata():
"""An operator naming the topic themselves has made the call."""
service = build_service(BASE_INI + """
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt1_topic_neighbors = my/{IATA}/nbrs
""", radio=FakeRadio())
assert service._resolve_neighbors_topic(service.mqtt_brokers[0]) == "my/XYZ/nbrs"
def test_explicit_neighbors_topic_resolves_placeholders():
service = build_service(BROKER_INI, radio=FakeRadio())
topic = service._resolve_neighbors_topic(service.mqtt_brokers[2])
assert topic == f"custom/SEA/{SELF_KEY.upper()}/nbrs"
def test_unroutable_broker_yields_no_topic():
service = build_service(BROKER_INI)
assert service._resolve_neighbors_topic({"topic_prefix": None, "topic_neighbors": None}) is None
def test_neighbors_brokers_requires_connected_and_opted_in():
service = build_service(BROKER_INI)
service.mqtt_clients = [
{"client": FakeMqttClient(), "config": service.mqtt_brokers[0], "connected": True},
{"client": FakeMqttClient(), "config": service.mqtt_brokers[1], "connected": True},
{"client": FakeMqttClient(), "config": service.mqtt_brokers[2], "connected": False},
]
assert [i["config"]["broker_num"] for i in service.neighbors_brokers()] == [1]
# ---------------------------------------------------------------------------
# Publishing
# ---------------------------------------------------------------------------
def test_publish_is_non_retained_and_only_to_opted_in_brokers():
service = build_service(BROKER_INI)
service.mqtt_enabled = True
opted_in, not_opted_in = FakeMqttClient(), FakeMqttClient()
service.mqtt_clients = [
{"client": opted_in, "config": service.mqtt_brokers[0], "connected": True},
{"client": not_opted_in, "config": service.mqtt_brokers[1], "connected": True},
]
metrics = service.publish_neighbors_mqtt({"neighbors": []})
assert metrics == {"attempted": 1, "succeeded": 1}
assert not_opted_in.published == []
assert len(opted_in.published) == 1
# A snapshot taken every 12-336h is not a useful last-will value.
assert opted_in.published[0]["retain"] is False
assert opted_in.published[0]["qos"] == 0
def test_publish_warns_once_for_an_unroutable_broker(caplog):
service = build_service(BROKER_INI)
service.mqtt_enabled = True
broker = dict(service.mqtt_brokers[0])
broker["topic_prefix"] = None
broker["topic_neighbors"] = None
service.mqtt_clients = [{"client": FakeMqttClient(), "config": broker, "connected": True}]
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
first = service.publish_neighbors_mqtt({"neighbors": []})
second = service.publish_neighbors_mqtt({"neighbors": []})
assert first == {"attempted": 0, "succeeded": 0}
assert second == {"attempted": 0, "succeeded": 0}
# Warned once, not on every cycle forever.
assert caplog.text.count("no neighbors topic could be resolved") == 1
def test_publish_warning_names_a_missing_iata_as_the_cause(caplog):
"""The generic "set a topic prefix" advice would be misleading here."""
service = build_service(BASE_INI + """
mqtt1_enabled = true
mqtt1_server = one.example.com
mqtt1_topic_packets = meshcore/{IATA}/{PUBLIC_KEY}/packets
""", radio=FakeRadio())
service.mqtt_enabled = True
service.mqtt_clients = [
{"client": FakeMqttClient(), "config": service.mqtt_brokers[0], "connected": True}
]
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
assert service.publish_neighbors_mqtt({"neighbors": []}) == {
"attempted": 0, "succeeded": 0
}
assert "no IATA is set" in caplog.text
def test_publish_counts_a_broker_failure():
service = build_service(BROKER_INI)
service.mqtt_enabled = True
service.mqtt_clients = [
{"client": FakeMqttClient(rc=4), "config": service.mqtt_brokers[0], "connected": True}
]
assert service.publish_neighbors_mqtt({"neighbors": []}) == {"attempted": 1, "succeeded": 0}
def test_publish_noops_when_mqtt_is_disabled():
service = build_service(BROKER_INI)
service.mqtt_enabled = False
service.mqtt_clients = [
{"client": FakeMqttClient(), "config": service.mqtt_brokers[0], "connected": True}
]
assert service.publish_neighbors_mqtt({"neighbors": []}) == {"attempted": 0, "succeeded": 0}
# ---------------------------------------------------------------------------
# Capability detection
# ---------------------------------------------------------------------------
def test_capability_needs_only_discover_when_scopes_are_off():
service = build_service(BASE_INI, radio=FakeRadio(commands=("send_node_discover_req",)))
assert service.neighbors_commands_available() is True
def test_capability_needs_regions_when_scopes_are_on():
service = build_service(
BASE_INI + "neighbors_collect_scopes = true\n",
radio=FakeRadio(commands=("send_node_discover_req",)),
)
assert service.neighbors_commands_available() is False
both = build_service(
BASE_INI + "neighbors_collect_scopes = true\n",
radio=FakeRadio(commands=("send_node_discover_req", "req_regions_sync")),
)
assert both.neighbors_commands_available() is True
def test_capability_false_without_a_radio():
service = build_service(BASE_INI, radio=None)
assert service.neighbors_commands_available() is False
# ---------------------------------------------------------------------------
# State persistence
# ---------------------------------------------------------------------------
def test_state_round_trips_through_bot_metadata(db_manager):
service = build_service(BASE_INI, db_manager=db_manager)
assert service.last_neighbors_publish == 0.0
service.last_neighbors_publish = 1774482900.0
service._save_neighbors_state()
assert db_manager.metadata[NEIGHBORS_STATE_KEY] == "1774482900.0"
reloaded = build_service(BASE_INI, db_manager=db_manager)
assert reloaded.last_neighbors_publish == 1774482900.0
def test_malformed_state_is_ignored(db_manager):
db_manager.metadata[NEIGHBORS_STATE_KEY] = "not-a-number"
assert build_service(BASE_INI, db_manager=db_manager).last_neighbors_publish == 0.0
def test_far_future_state_is_ignored(db_manager):
"""A clock jump forward would otherwise suppress cycles indefinitely."""
db_manager.metadata[NEIGHBORS_STATE_KEY] = str(2**31)
assert build_service(BASE_INI, db_manager=db_manager).last_neighbors_publish == 0.0
# ---------------------------------------------------------------------------
# Full cycle
# ---------------------------------------------------------------------------
@pytest.fixture
def no_sleep(monkeypatch):
async def instant(_seconds):
return None
monkeypatch.setattr(nb.asyncio, "sleep", instant)
async def test_cycle_records_feeds_graph_and_publishes(db_manager, no_sleep):
radio = FakeRadio([response(KEY_A, 7.5), response(KEY_B, -2.0)])
service = build_service(BROKER_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = True
client = FakeMqttClient()
service.mqtt_clients = [
{"client": client, "config": service.mqtt_brokers[0], "connected": True}
]
graph_calls = []
service.bot.mesh_graph.add_edge.side_effect = lambda *a, **k: graph_calls.append((a, k))
summary = await service.run_neighbors_cycle()
assert summary["ok"] is True
assert summary["discovered"] == 2
assert summary["queried"] == 2
assert summary["best_snr"] == 7.5
assert summary["recorded"] == 2
assert summary["attempted"] == 1
assert summary["succeeded"] == 1
# Persisted, both tables.
with db_manager.connection() as conn:
links = [dict(r) for r in conn.execute(
"SELECT neighbor_public_key, best_snr FROM neighbor_links ORDER BY neighbor_public_key"
)]
history = conn.execute("SELECT COUNT(*) FROM neighbor_observations").fetchone()[0]
assert [row["neighbor_public_key"] for row in links] == [KEY_A, KEY_B]
assert history == 2
# Graph: both directions per link.
assert len(graph_calls) == 4
assert {call[0] for call in graph_calls} == {
(SELF_KEY[:6], KEY_A[:6]), (KEY_A[:6], SELF_KEY[:6]),
(SELF_KEY[:6], KEY_B[:6]), (KEY_B[:6], SELF_KEY[:6]),
}
# Full public keys, which path-derived edges cannot supply.
assert all(len(call[1]["from_public_key"]) == 64 for call in graph_calls)
# Published payload.
message = json.loads(client.published[0]["payload"])
assert message["origin"] == "TestBot"
assert message["origin_id"] == SELF_KEY.upper()
assert [n["pubkey"] for n in message["neighbors"]] == [KEY_A.upper(), KEY_B.upper()]
# State stamped.
assert float(db_manager.metadata[NEIGHBORS_STATE_KEY]) > 0
async def test_cycle_never_touches_the_contact_cache(db_manager, no_sleep):
"""Stage 1 is a broadcast; nothing here may add a contact."""
radio = FakeRadio([response(KEY_A, 1.0)])
service = build_service(BASE_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
await service.run_neighbors_cycle()
assert radio.contacts == {}
async def test_cycle_records_without_any_broker(db_manager, no_sleep):
"""The database is a legitimate consumer on its own."""
radio = FakeRadio([response(KEY_A, 1.0)])
service = build_service(BASE_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
summary = await service.run_neighbors_cycle()
assert summary["ok"] is True
assert summary["recorded"] == 1
assert summary["attempted"] == 0
with db_manager.connection() as conn:
assert conn.execute("SELECT COUNT(*) FROM neighbor_links").fetchone()[0] == 1
async def test_cycle_respects_the_max_cap(db_manager, no_sleep):
radio = FakeRadio([response(KEY_A, 9.0), response(KEY_B, 1.0)])
service = build_service(BASE_INI + "neighbors_max = 1\n",
db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
summary = await service.run_neighbors_cycle()
assert summary["discovered"] == 2
assert summary["queried"] == 1
# Kept the strongest, and said so in the payload contract.
with db_manager.connection() as conn:
kept = conn.execute("SELECT neighbor_public_key FROM neighbor_links").fetchone()[0]
assert kept == KEY_A
async def test_cycle_skips_graph_when_disabled(db_manager, no_sleep):
radio = FakeRadio([response(KEY_A, 1.0)])
service = build_service(BASE_INI + "neighbors_feed_mesh_graph = false\n",
db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
await service.run_neighbors_cycle()
service.bot.mesh_graph.add_edge.assert_not_called()
async def test_cycle_marks_entries_responded_with_scopes_off(db_manager, no_sleep):
"""Reporting a live neighbour as `timeout` would be wrong."""
radio = FakeRadio([response(KEY_A, 1.0)])
service = build_service(BASE_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
await service.run_neighbors_cycle()
with db_manager.connection() as conn:
row = conn.execute("SELECT last_status, scopes FROM neighbor_links").fetchone()
assert row["last_status"] == nb.STATUS_RESPONDED
assert row["scopes"] == ""
@pytest.mark.parametrize("ini,radio,connected,expected", [
("[PacketCapture]\nenabled = true\n", FakeRadio(), True, "disabled"),
(BASE_INI, None, True, "not connected"),
(BASE_INI, FakeRadio(), False, "not connected"),
(BASE_INI, FakeRadio(commands=()), True, "does not support"),
])
async def test_cycle_bails_with_a_reason(db_manager, ini, radio, connected, expected):
service = build_service(ini, db_manager=db_manager, radio=radio, connected=connected)
service.mqtt_enabled = False
summary = await service.run_neighbors_cycle()
assert summary["ok"] is False
assert expected in summary["reason"]
async def test_cycle_discards_when_the_session_resets_mid_cycle(db_manager, monkeypatch):
"""Recording partial data from a torn-down session would be misleading."""
radio = FakeRadio([response(KEY_A, 1.0)])
service = build_service(BASE_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
async def swap_radio(_seconds):
# Simulate a reconnect replacing the meshcore object mid-window.
service.bot.meshcore = FakeRadio([])
monkeypatch.setattr(nb.asyncio, "sleep", swap_radio)
summary = await service.run_neighbors_cycle()
assert summary["ok"] is False
with db_manager.connection() as conn:
assert conn.execute("SELECT COUNT(*) FROM neighbor_links").fetchone()[0] == 0
async def test_repeated_discover_failure_warns_only_once(db_manager, caplog, no_sleep):
radio = FakeRadio([])
async def failing(filter_bits, prefix_only=True, tag=None):
return FakeEvent(EventType.ERROR, {"reason": "unsupported"})
radio.commands.send_node_discover_req = failing
service = build_service(BASE_INI, db_manager=db_manager, radio=radio)
service.mqtt_enabled = False
with caplog.at_level(logging.WARNING, logger=LOGGER.name):
for _ in range(3):
assert (await service.run_neighbors_cycle())["ok"] is False
assert caplog.text.count("discovery request failed; will keep retrying") == 1
assert service.neighbors_discover_failures == 3
# ---------------------------------------------------------------------------
# Scheduler registration
# ---------------------------------------------------------------------------
async def _drain(service):
service.should_exit = True
for task in service.background_tasks:
task.cancel()
with contextlib.suppress(asyncio.CancelledError):
await task
def _quiet_other_tasks(service):
service.stats_status_enabled = False
service.stats_refresh_interval = 0
service.health_check_interval = 0
service.mqtt_enabled = False
async def test_scheduler_starts_only_when_enabled(db_manager):
disabled = build_service("[PacketCapture]\nenabled = true\n", db_manager=db_manager)
_quiet_other_tasks(disabled)
await disabled.start_background_tasks()
assert disabled.neighbors_task is None
assert disabled.background_tasks == []
enabled = build_service(BASE_INI, db_manager=db_manager)
_quiet_other_tasks(enabled)
await enabled.start_background_tasks()
assert enabled.neighbors_task is not None
assert enabled.neighbors_task in enabled.background_tasks
await _drain(enabled)
+15
View File
@@ -165,6 +165,10 @@
"advert": [
"advert"
],
"neighbors": [
"neighbors",
"neighbours"
],
"multitest": [
"multitest",
"mt"
@@ -407,6 +411,17 @@
"success": "Flood advert sent successfully!",
"error": "Error sending flood advert: {error}"
},
"neighbors": {
"description": "Runs a zero-hop neighbour discovery cycle (DM only)",
"disabled": "Neighbour discovery is not enabled. Set neighbors_enabled = true in the PacketCapture config.",
"busy": "A neighbour discovery cycle is already running. Try again once it finishes.",
"started": "Discovering neighbours - listening for about {seconds}s. I'll report back when done.",
"success": "Heard {count} neighbour(s), best SNR {best_snr}. Recorded {recorded} link(s).",
"published": "Published to {succeeded}/{attempted} broker(s).",
"none": "Discovery finished but no repeaters answered. Nothing is in direct range right now.",
"failed": "Neighbour discovery did not run: {reason}.",
"error": "Error running neighbour discovery: {error}"
},
"path": {
"description": "Decode hex path data to show which repeaters were involved in message routing",
"help": "Path: path [hex] - Decode path to show repeaters. Use path alone for current message path, or path [7e,01] for specific path.",