adam/pyMC_Repeater
1
0
Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-08-07 17:33:16 +02:00
Code Issues Packages Projects Releases Wiki Activity

feat(neighbors): persist neighbour scopes and expose them over the API

Browse Source 
Scope answers were built into the MQTT payload and then discarded, so the web UI
had nothing to show between cycles and no way to ask a single repeater. Adds a
store for them, a read endpoint, and a single-target query.

Migration 15 adds neighbor_scopes: one row per queried neighbour holding the last
answer (`scopes`, `responded_at`) alongside the last query's outcome
(`status`, `queried_at`). The two are kept apart deliberately -- a failed query
updates the outcome but leaves the answer in place, because the responder
rate-limits anonymous replies to 4 every 3 minutes and one timeout is weak
evidence that a neighbour's scopes changed. An empty answer is stored as a real
answer: it means the neighbour serves unscoped traffic only.

A row is written for any query the node attempted. `timeout` alone cannot carry
that: the sweep reports it both for a neighbour that was asked and stayed silent
and for one it never reached, which is what ScopeResult.transmitted separates.
`send_failed` is recorded too even though nothing reached the air -- it was
attempted and the duty cycle refused, and skipping it left a repeater that keeps
refusing reading as "never queried" however often it was asked.

Scope rows follow their neighbour out of the database, on all four paths that
delete adverts: the two explicit deletes, the 6-hourly retention cleanup, and a
purge of the adverts table. Without the last two the table grew without bound and
a purge left scope counts on screen for repeaters no longer listed.

GET /api/neighbor_scopes serves the table; it stays separate from the paginated
advert queries, which are read per contact type on every page load.

POST /api/query_neighbor_scopes asks one neighbour now. Unlike publish_neighbors
it holds the request open for the reply, since the response window is normally its
5 s floor; past 45 s it returns and leaves the query running so a late answer
still lands rather than throwing away spent airtime. It returns the stored view,
not the raw result, so a failed query cannot tell the client to forget scopes the
database still holds. Nothing is published -- the periodic cycle owns the topic.

A query and a cycle must not collide over the scope helper. A cycle holds it for
its whole run including the discovery window, so a query refuses while one is
active; a cycle defers while a query is in flight; and if the two still race, the
cycle abandons the pass on the short retry delay instead of publishing a table
with every scope missing or dying and re-spending its discovery broadcast. Queries
are tracked so shutdown cancels them rather than transmitting through teardown.

The endpoint needs json_in: it is not on globally for /api and cherrypy's Request
has no `json` attribute without it, so reading the body would have failed on every
real request. A test asserts the decorator rather than trusting a fabricated
cherrypy.request, which is how it went unnoticed.
This commit is contained in:
agessaman
2026-07-29 09:58:41 -07:00
parent 7f79b6797c
commit 9c417478f3
6 changed files with 1114 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
repeater/data_acquisition/sqlite_handler.py
+140
View File
@@ -762,11 +762,123 @@ class SQLiteHandler:
)
logger.info(f"Migration '{migration_name}' applied successfully")
# Migration 15: Last-known region scopes per neighbour, from the
# anon-regions query the neighbours publisher issues. The MQTT
# payload was the only consumer, so the answers were discarded as
# soon as they were published and the web UI had nothing to show.
# One row per queried neighbour, rewritten at most once per query.
migration_name = "add_neighbor_scopes"
existing = conn.execute(
"SELECT migration_name FROM migrations WHERE migration_name = ?",
(migration_name,),
).fetchone()
if not existing:
conn.execute(
"""
CREATE TABLE IF NOT EXISTS neighbor_scopes (
pubkey TEXT PRIMARY KEY,
scopes TEXT NOT NULL DEFAULT '',
responded_at REAL,
status TEXT NOT NULL,
queried_at REAL NOT NULL
)
"""
)
conn.execute(
"INSERT INTO migrations (migration_name, applied_at) VALUES (?, ?)",
(migration_name, time.time()),
)
logger.info(f"Migration '{migration_name}' applied successfully")
conn.commit()
except Exception as e:
logger.error(f"Failed to run migrations: {e}")
# Neighbour scope methods
def get_neighbor_scopes(self) -> dict:
"""Return every stored scope record, keyed by lowercase pubkey hex.
Never raises: the caller renders a column from this, and a missing or
corrupt table must degrade to "nothing known" rather than fail the
request that also carries the neighbour table.
"""
try:
with self._connect() as conn:
conn.row_factory = sqlite3.Row
rows = conn.execute(
"SELECT pubkey, scopes, responded_at, status, queried_at FROM neighbor_scopes"
).fetchall()
return {
row["pubkey"]: {
"scopes": row["scopes"] or "",
"responded_at": row["responded_at"],
"status": row["status"],
"queried_at": row["queried_at"],
}
for row in rows
}
except Exception as e:
logger.debug(f"Could not read neighbour scopes: {e}")
return {}
def record_neighbor_scope(
self,
pubkey: str,
status: str,
scopes: Optional[str] = None,
queried_at: Optional[float] = None,
) -> bool:
"""Record the outcome of one scope query.
``scopes`` is the answer the neighbour gave, and is only supplied when it
actually answered. A failed query updates ``status``/``queried_at`` but
leaves the last known ``scopes`` and ``responded_at`` in place: the
responder rate-limits anon replies (4 per 3 minutes, shared across
identities), so a single timeout is weak evidence that a neighbour's
scopes changed and is not worth discarding a good answer over.
An empty ``scopes`` string is a real answer -- it means the neighbour
serves unscoped traffic only -- so it is stored, not treated as absent.
"""
pubkey = str(pubkey or "").strip().lower()
if not pubkey:
return False
when = time.time() if queried_at is None else float(queried_at)
try:
with self._connect() as conn:
if scopes is None:
conn.execute(
"""
INSERT INTO neighbor_scopes (pubkey, scopes, responded_at,
status, queried_at)
VALUES (?, '', NULL, ?, ?)
ON CONFLICT(pubkey) DO UPDATE SET
status = excluded.status,
queried_at = excluded.queried_at
""",
(pubkey, str(status), when),
)
else:
conn.execute(
"""
INSERT INTO neighbor_scopes (pubkey, scopes, responded_at,
status, queried_at)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT(pubkey) DO UPDATE SET
scopes = excluded.scopes,
responded_at = excluded.responded_at,
status = excluded.status,
queried_at = excluded.queried_at
""",
(pubkey, str(scopes), when, str(status), when),
)
conn.commit()
return True
except Exception as e:
logger.warning(f"Could not persist neighbour scopes for {pubkey[:8]}: {e}")
return False
# Daemon state methods
def get_daemon_state(self, key: str) -> Optional[dict]:
"""Read a persisted daemon-state blob, or None when absent/unreadable.
@@ -2631,6 +2743,11 @@ class SQLiteHandler:
try:
with self._connect() as conn:
result = conn.execute(purge_queries[table_name])
if table_name == "adverts":
# Purging the neighbour table has to take the scopes with it,
# or the UI shows scope counts for repeaters it no longer lists
# and presents them as current.
conn.execute("DELETE FROM neighbor_scopes")
conn.commit()
logger.info(f"Purged {result.rowcount} rows from {table_name}")
return result.rowcount
@@ -2667,6 +2784,14 @@ class SQLiteHandler:
result = conn.execute("DELETE FROM adverts WHERE timestamp < ?", (cutoff,))
adverts_deleted = result.rowcount
# A scope row describes a neighbour, so it has nothing left to be
# displayed against once that neighbour's advert is pruned. Without
# this it would survive every retention pass and grow without bound.
conn.execute(
"DELETE FROM neighbor_scopes WHERE pubkey NOT IN "
"(SELECT lower(pubkey) FROM adverts)"
)
result = conn.execute("DELETE FROM noise_floor WHERE timestamp < ?", (cutoff,))
noise_deleted = result.rowcount
@@ -3152,7 +3277,17 @@ class SQLiteHandler:
def delete_advert(self, advert_id: int) -> bool:
try:
with self._connect() as conn:
# Adverts are one row per pubkey (migration `adverts_unique_pubkey`),
# so deleting one drops the neighbour entirely; its scope row would
# otherwise outlive it with nothing left to display it against.
row = conn.execute(
"SELECT pubkey FROM adverts WHERE id = ?", (advert_id,)
).fetchone()
cursor = conn.execute("DELETE FROM adverts WHERE id = ?", (advert_id,))
if row and row[0]:
conn.execute(
"DELETE FROM neighbor_scopes WHERE pubkey = ?", (str(row[0]).lower(),)
)
self._neighbors_cache = {"timestamp": 0.0, "value": None}
return cursor.rowcount > 0
except Exception as e:
@@ -3165,11 +3300,16 @@ class SQLiteHandler:
with self._connect() as conn:
if pubkey_prefix is None:
cursor = conn.execute("DELETE FROM adverts")
conn.execute("DELETE FROM neighbor_scopes")
else:
cursor = conn.execute(
"DELETE FROM adverts WHERE lower(pubkey) LIKE ?",
(f"{pubkey_prefix.lower()}%",),
)
conn.execute(
"DELETE FROM neighbor_scopes WHERE pubkey LIKE ?",
(f"{pubkey_prefix.lower()}%",),
)
self._neighbors_cache = {"timestamp": 0.0, "value": None}
return int(cursor.rowcount)
except Exception as e:
repeater/data_acquisition/storage_collector.py
+12
View File
@@ -539,6 +539,18 @@ class StorageCollector:
def get_neighbors(self) -> dict:
return self.sqlite_handler.get_neighbors()
def get_neighbor_scopes(self) -> dict:
return self.sqlite_handler.get_neighbor_scopes()
def record_neighbor_scope(
self,
pubkey: str,
status: str,
scopes: Optional[str] = None,
queried_at: Optional[float] = None,
) -> bool:
return self.sqlite_handler.record_neighbor_scope(pubkey, status, scopes, queried_at)
def get_daemon_state(self, key: str) -> Optional[dict]:
return self.sqlite_handler.get_daemon_state(key)
repeater/neighbors_publisher.py
+196 -2
View File
@@ -28,6 +28,7 @@ from typing import Any, Callable, Dict, List, Optional
from repeater.handler_helpers.discovery import persist_discovery_result
from repeater.handler_helpers.neighbor_scopes import (
STATUS_RESPONDED,
STATUS_SEND_FAILED,
STATUS_TIMEOUT,
NeighborSnapshot,
ScopeResult,
@@ -166,6 +167,12 @@ class NeighborsPublisher:
self._manual_task: Optional[asyncio.Task] = None
self._running = False
self._active = False
# Single-neighbour queries running outside a cycle. A cycle must not start
# on top of one: it would reach the sweep and find the helper's lock held,
# which raises and costs the whole cycle (including its discovery
# broadcast). Tracked as a count, and as tasks so shutdown can cancel them.
self._queries_in_flight = 0
self._query_tasks: set = set()
# Discovery responses collected during the current cycle, keyed by pubkey.
self._discovery_seen: Dict[str, dict] = {}
self._next_publish_at: Optional[float] = None
@@ -329,6 +336,8 @@ class NeighborsPublisher:
"""
if self._active or (self._manual_task is not None and not self._manual_task.done()):
return False
if self._queries_in_flight:
return False
self._manual_task = asyncio.create_task(
self.run_cycle(trigger="manual"), name="neighbors-manual-cycle"
)
@@ -347,9 +356,14 @@ class NeighborsPublisher:
async def stop(self) -> None:
self._running = False
tasks = [t for t in (self._task, self._manual_task) if t and not t.done()]
tasks = [
t
for t in (self._task, self._manual_task, *self._query_tasks)
if t and not t.done() and t is not asyncio.current_task()
]
self._task = None
self._manual_task = None
self._query_tasks.clear()
for task in tasks:
task.cancel()
for task in tasks:
@@ -457,6 +471,12 @@ class NeighborsPublisher:
if self._next_publish_at is not None and now < self._next_publish_at:
return
if self._queries_in_flight:
# Deferred, not rescheduled: the next tick is 30 s away and a single
# query is far shorter than that, so the cycle simply runs then.
logger.debug("Neighbors cycle deferred: a manual scope query is in flight")
return
await self.run_cycle(trigger="periodic")
def _reschedule(self, *, retry: bool = False) -> None:
@@ -488,7 +508,18 @@ class NeighborsPublisher:
scope_results: Dict[str, ScopeResult] = {}
if targets and self.scope_helper:
scope_results = await self.scope_helper.sweep(targets)
try:
scope_results = await self.scope_helper.sweep(targets)
except RuntimeError as e:
# A manual query took the helper between the checks in _tick /
# trigger_cycle and here -- the discovery window above leaves a
# wide gap for that. Give up on this pass rather than publish a
# table with every scope missing; the finally below reschedules
# on the short retry delay.
logger.warning("Neighbors cycle abandoned: %s", e)
self._last_result = f"deferred: {e}"
return {"success": False, "error": str(e)}
self._persist_scope_results(scope_results)
elif targets:
logger.warning("No scope helper available; publishing without scopes")
@@ -524,6 +555,169 @@ class NeighborsPublisher:
self._active = False
self._reschedule(retry=not published)
@staticmethod
def _was_asked(result: ScopeResult) -> bool:
"""Whether this outcome represents a query the node actually attempted.
``timeout`` is ambiguous on its own: the sweep reports it both for a
neighbour that was asked and stayed silent and for one it never reached
(budget exhausted), so ``transmitted`` is what separates those. A
``send_failed`` was attempted and refused by the transmit path -- the
duty-cycle pre-flight -- which is worth recording even though nothing went
on air, otherwise a repeater that keeps refusing reads as "never queried"
forever.
"""
return result.transmitted or result.status == STATUS_SEND_FAILED
def _persist_scope_results(
self, results: Dict[str, ScopeResult], now: Optional[float] = None
) -> None:
"""Store what a sweep learned, so it outlives the MQTT publish.
The payload used to be the only consumer, which left the web UI with
nothing to show between cycles. One row per neighbour the node actually
asked (see :meth:`_was_asked`); targets the sweep never reached are left
untouched rather than credited with a query that never happened.
"""
storage = self._storage()
writer = getattr(storage, "record_neighbor_scope", None) if storage else None
if not callable(writer):
return
stamp = time.time() if now is None else now
for pubkey, result in (results or {}).items():
if not self._was_asked(result):
continue
try:
# scopes is passed only for an answer, so a failed query keeps the
# last known value instead of blanking it.
writer(
pubkey,
result.status,
result.scopes if result.status == STATUS_RESPONDED else None,
stamp,
)
except Exception as e:
logger.debug(f"Could not persist scopes for {pubkey[:8]}: {e}")
async def query_one(self, pubkey: str) -> dict:
"""Query a single neighbour's scopes now, outside the periodic cycle.
Deliberately independent of ``enabled()``: the answer is stored for the
web UI, so this is useful on a repeater that publishes to no broker at
all. Nothing is published as a result -- the periodic cycle owns that.
Raises ``ValueError`` for a key that cannot be queried and ``RuntimeError``
when a cycle or another query already holds the scope helper; the caller
turns both into a message. The cycle check is on ``_active`` rather than on
the helper's lock because a cycle spends its first minute in the discovery
window without holding that lock, and colliding later -- once the sweep has
taken it -- would cost the whole cycle.
"""
key = str(pubkey or "").strip().lower()
if len(key) != 64:
# ECDH against the responder needs the full 32-byte key; a prefix
# (which is all an advert-only sighting may carry) cannot be used.
raise ValueError("A full 64-character public key is required")
try:
bytes.fromhex(key)
except ValueError:
raise ValueError("Public key is not valid hex") from None
if key == self._local_pubkey_hex():
raise ValueError("Cannot query this repeater's own scopes")
if not self.scope_helper:
raise RuntimeError("Scope helper not available")
if self._active:
# A cycle owns the helper for its whole run, including the discovery
# window before the sweep takes the lock. Refusing here is the mirror
# of the sweep-side guard below and keeps the two from colliding.
raise RuntimeError("A neighbours cycle is running - try again once it finishes")
snapshot = self._snapshot_for(key)
self._queries_in_flight += 1
task = asyncio.current_task()
if task is not None:
# Tracked for the same reason trigger_cycle tracks its task: this holds
# the radio for up to a response window, and shutdown has to be able to
# cut it short rather than transmit through the teardown.
self._query_tasks.add(task)
try:
try:
results = await self.scope_helper.sweep([snapshot])
except RuntimeError:
# The helper's own wording names its internals; say what the
# operator can act on instead.
raise RuntimeError(
"A neighbour scope sweep is already running - try again once it finishes"
) from None
now = time.time()
self._persist_scope_results(results, now=now)
result = results.get(key) or ScopeResult(STATUS_TIMEOUT)
return self._scope_record(key, result, now)
finally:
self._queries_in_flight = max(0, self._queries_in_flight - 1)
if task is not None:
self._query_tasks.discard(task)
def _scope_record(self, pubkey: str, result: ScopeResult, now: float) -> dict:
"""The stored view of one query's outcome, as the API returns it.
Read back through the stored row rather than reported straight from
``result``: a failed query deliberately keeps the neighbour's last known
scopes, so returning this query's empty string would tell the client to
forget an answer the database still holds.
"""
responded = result.status == STATUS_RESPONDED
record = {
"pubkey": pubkey,
"status": result.status,
"scopes": result.scopes if responded else "",
"transmitted": result.transmitted,
"queried_at": now if self._was_asked(result) else None,
"responded_at": now if responded else None,
}
if responded:
return record
storage = self._storage()
reader = getattr(storage, "get_neighbor_scopes", None) if storage else None
if not callable(reader):
return record
try:
stored = (reader() or {}).get(pubkey) or {}
except Exception as e:
logger.debug(f"Could not re-read stored scopes for {pubkey[:8]}: {e}")
return record
if stored.get("responded_at") is not None:
record["scopes"] = stored.get("scopes") or ""
record["responded_at"] = stored.get("responded_at")
return record
def _snapshot_for(self, pubkey: str) -> NeighborSnapshot:
"""Build a one-target snapshot, borrowing last_seen/snr when we have them.
Neither field affects the query -- they are carried so a single query goes
through exactly the same path as a sweep entry.
"""
storage = self._storage()
if storage:
try:
# The adverts table does not normalise key case, so match on the
# lowercased form rather than indexing directly.
for candidate, info in (storage.get_neighbors() or {}).items():
if str(candidate or "").lower() != pubkey:
continue
return NeighborSnapshot(
pubkey=pubkey,
last_seen=float(info.get("last_seen") or 0.0),
snr=float(info.get("snr") or 0.0),
)
except Exception as e:
logger.debug(f"Could not read neighbour row for {pubkey[:8]}: {e}")
return NeighborSnapshot(pubkey=pubkey)
async def _refresh_neighbor_table(self) -> None:
"""Stage 1: zero-hop node discovery, awaited to completion.
repeater/web/api_endpoints.py
+105
View File
@@ -191,6 +191,14 @@ POLICY_GROUP_KINDS = {
class APIEndpoints:
# How long /api/query_neighbor_scopes holds a request open. The scope helper's
# response window is normally its 5 s floor; a slow radio config (SF12) or a
# duty-cycle deferral can push a single query past this, in which case the
# query is left running so its answer still reaches the stored table and the
# client is told to re-read it. Chosen to stay inside a default reverse-proxy
# read timeout rather than to cover the helper's 120 s ceiling.
SCOPE_QUERY_HTTP_TIMEOUT = 45.0
def __init__(
self,
stats_getter: Optional[Callable] = None,
@@ -2391,6 +2399,103 @@ class APIEndpoints:
logger.error(f"Error starting neighbours cycle: {e}", exc_info=True)
return self._error(str(e))
@cherrypy.expose
@cherrypy.tools.json_out()
def neighbor_scopes(self):
"""Last known region scopes per neighbour.
GET /api/neighbor_scopes
Served as its own endpoint rather than folded into the advert queries: the
advert list is paginated per contact type and read on every neighbours-page
load, and this table is small enough (one row per queried neighbour) that
the client can hold the whole thing and join on pubkey.
``scopes`` is the last answer a neighbour gave; an empty string is a real
answer meaning it serves unscoped traffic only. ``status``/``queried_at``
describe the most recent query, which may have failed after a good answer,
so ``responded_at`` is how fresh the scopes themselves are.
"""
self._set_cors_headers()
if cherrypy.request.method == "OPTIONS":
return ""
try:
storage = self._get_storage()
scopes = storage.get_neighbor_scopes() or {}
return self._success(scopes, count=len(scopes))
except Exception as e:
logger.error(f"Error reading neighbour scopes: {e}")
return self._error(str(e))
@cherrypy.expose
@cherrypy.tools.json_out()
@cherrypy.tools.json_in()
def query_neighbor_scopes(self):
"""Ask one neighbour for its region scopes now.
POST /api/query_neighbor_scopes {"pubkey": "<64 hex chars>"}
Unlike ``publish_neighbors`` this holds the request open for the answer:
one query is a single route-direct request whose response window is
normally the 5 s floor (only a slow radio config approaches the 120 s
ceiling), so a spinner is a better fit than a poll. Nothing is published
as a result -- the answer is stored and returned, and the periodic cycle
remains the only thing that writes to the MQTT topic.
The request is route-direct with an empty path, so it only reaches a
zero-hop neighbour, and the responder rate-limits anonymous replies (4 per
3 minutes, shared across its identities) -- both show up here as a
``timeout``.
"""
self._set_cors_headers()
if cherrypy.request.method == "OPTIONS":
return ""
try:
self._require_post()
data = cherrypy.request.json or {}
pubkey = str(data.get("pubkey") or "").strip()
if not pubkey:
return self._error("Missing pubkey parameter")
publisher = getattr(self.daemon_instance, "neighbors_publisher", None)
if not publisher:
return self._error("Neighbors publisher not available")
if self.event_loop is None:
return self._error("Event loop not available")
import asyncio
future = asyncio.run_coroutine_threadsafe(publisher.query_one(pubkey), self.event_loop)
result = future.result(timeout=self.SCOPE_QUERY_HTTP_TIMEOUT)
return self._success(result)
except FutureTimeoutError:
# Deliberately not cancelled: the query has already spent the airtime,
# and it persists its own outcome, so letting it finish means the answer
# still shows up on a re-read instead of being thrown away here.
logger.warning(
"Neighbour scope query still in flight after %.0fs",
self.SCOPE_QUERY_HTTP_TIMEOUT,
)
return self._error("Still waiting for the neighbour to reply - check back in a moment")
except cherrypy.HTTPError:
raise
except ValueError as e:
return self._error(str(e))
except RuntimeError as e:
# query_one raises this when a cycle or another query already holds the
# scope helper -- one request in flight at a time, by design. Still
# logged, because an unrelated RuntimeError from the event loop lands
# here too and would otherwise leave no trace.
logger.warning(f"Neighbour scope query refused: {e}")
return self._error(str(e))
except Exception as e:
logger.error(f"Error querying neighbour scopes: {e}", exc_info=True)
return self._error(str(e))
@staticmethod
def _validate_neighbors_settings(raw):
"""Validate the ``mqtt_brokers.neighbors`` block.
repeater/web/openapi.yaml
+120
View File
@@ -3439,6 +3439,105 @@ paths:
'405':
description: Method not allowed
/neighbor_scopes:
get:
tags: [Network Policy]
summary: Last known region scopes per neighbour
description: >
Region scopes learned from the anon-regions query the neighbours publisher
issues, keyed by lowercase pubkey hex. One row per neighbour that has been
queried; neighbours never queried are simply absent. `scopes` is the last
answer given and an empty string is a real answer meaning the neighbour
serves unscoped traffic only. `status`/`queried_at` describe the most
recent query, which may have failed after a good answer, so `responded_at`
is what says how fresh `scopes` is.
security:
- BearerAuth: []
- ApiKeyAuth: []
responses:
'200':
description: >
Stored scope records, or an error when storage is unavailable (as it
briefly is during daemon startup) — `data` is absent in that case.
content:
application/json:
schema:
type: object
required: [success]
properties:
success:
type: boolean
error:
type: string
count:
type: integer
data:
type: object
additionalProperties:
$ref: '#/components/schemas/NeighborScopeRecord'
/query_neighbor_scopes:
post:
tags: [Network Policy]
summary: Query one neighbour's region scopes now
description: >
Sends a single route-direct anon-regions request and waits for the reply,
then stores and returns the outcome. Nothing is published to MQTT; the
periodic cycle owns the neighbors topic. The request only reaches a
zero-hop neighbour, and the responder rate-limits anonymous replies (4 per
3 minutes), so both a multi-hop target and a repeated query show up as
`timeout`. Errors when a neighbours cycle already holds the scope helper.
security:
- BearerAuth: []
- ApiKeyAuth: []
requestBody:
required: true
content:
application/json:
schema:
type: object
required: [pubkey]
properties:
pubkey:
type: string
description: Full 64-character public key hex of the neighbour
responses:
'200':
description: Query outcome, or an error when it could not be run
content:
application/json:
schema:
type: object
required: [success]
properties:
success:
type: boolean
error:
type: string
data:
type: object
required: [pubkey, status, scopes, transmitted]
properties:
pubkey:
type: string
status:
type: string
enum: [responded, timeout, send_failed]
scopes:
type: string
description: Comma-separated scope names; empty when unscoped
transmitted:
type: boolean
description: Whether the request actually reached the air
queried_at:
type: number
nullable: true
responded_at:
type: number
nullable: true
'405':
description: Method not allowed
/update_web_config:
post:
tags: [System]
@@ -4407,6 +4506,27 @@ components:
type: string
description: Error message
NeighborScopeRecord:
type: object
required: [scopes, status, queried_at]
properties:
scopes:
type: string
description: >
Comma-separated region scope names from the neighbour's last answer.
Empty means it answered that it serves unscoped traffic only.
responded_at:
type: number
nullable: true
description: Epoch seconds of the answer `scopes` came from
status:
type: string
enum: [responded, timeout, send_failed]
description: Outcome of the most recent query
queried_at:
type: number
description: Epoch seconds of the most recent query
NeighborLinkSnapshot:
type: object
properties:
tests/test_mqtt_neighbors.py
+541 -5
View File
@@ -27,6 +27,7 @@ from repeater.handler_helpers.neighbor_scopes import (
STATUS_TIMEOUT,
NeighborScopeHelper,
NeighborSnapshot,
ScopeResult,
)
from repeater.neighbors_publisher import (
DEFAULT_INTERVAL_HOURS,
@@ -1187,19 +1188,22 @@ async def test_transmitted_flag_tracks_whether_the_request_reached_the_air():
# ====================================================================
# Manual trigger endpoint
# ====================================================================
def _api_with_publisher(monkeypatch, publisher, method="POST"):
def _api_with_publisher(monkeypatch, publisher, method="POST", json=None, storage=None):
import cherrypy
from repeater.web.api_endpoints import APIEndpoints
request = SimpleNamespace(method=method, params={}, json={})
request = SimpleNamespace(method=method, params={}, json=json if json is not None else {})
response = SimpleNamespace(headers={}, status=200)
monkeypatch.setattr(cherrypy, "request", request, raising=False)
monkeypatch.setattr(cherrypy, "response", response, raising=False)
api = APIEndpoints.__new__(APIEndpoints)
api.config = {}
api.daemon_instance = SimpleNamespace(neighbors_publisher=publisher)
api.daemon_instance = SimpleNamespace(
neighbors_publisher=publisher,
repeater_handler=SimpleNamespace(storage=storage),
)
api.event_loop = asyncio.new_event_loop()
api.send_advert_func = None
api.stats_getter = None
@@ -1222,7 +1226,7 @@ class _FakePublisher:
return self._starts
def _run_endpoint(api):
def _run_endpoint(api, call=None):
"""Drive the endpoint's run_coroutine_threadsafe against a real loop."""
import threading
@@ -1230,7 +1234,7 @@ def _run_endpoint(api):
thread = threading.Thread(target=loop.run_forever, daemon=True)
thread.start()
try:
return api.publish_neighbors()
return (call or api.publish_neighbors)()
finally:
loop.call_soon_threadsafe(loop.stop)
thread.join(timeout=5)
@@ -1592,3 +1596,535 @@ def test_migration_is_idempotent_on_an_existing_database(tmp_path):
SQLiteHandler(tmp_path)._run_migrations()
assert SQLiteHandler(tmp_path).get_daemon_state(STATE_KEY) == {"last_success_at": 42.0}
# ====================================================================
# Scope persistence
# ====================================================================
class _FakeScopeStore:
"""Neighbour table plus the scope rows, without touching sqlite."""
def __init__(self, neighbors=None, scopes=None):
self._neighbors = dict(neighbors or {})
self.scopes = dict(scopes or {})
self.writes = []
def get_neighbors(self):
return self._neighbors
def get_neighbor_scopes(self):
return self.scopes
def record_neighbor_scope(self, pubkey, status, scopes=None, queried_at=None):
self.writes.append((pubkey, status, scopes))
row = self.scopes.setdefault(pubkey, {"scopes": "", "responded_at": None})
row["status"] = status
row["queried_at"] = queried_at
if scopes is not None:
row["scopes"] = scopes
row["responded_at"] = queried_at
return True
class _StubSweep:
"""Scope helper stand-in returning canned results for one sweep."""
def __init__(self, results):
self.results = results
self.targets = None
async def sweep(self, targets):
self.targets = list(targets)
return dict(self.results)
def _repeater_row(last_seen=None, snr=6.0):
return {
"is_repeater": True,
"zero_hop": True,
"last_seen": time.time() if last_seen is None else last_seen,
"snr": snr,
}
@pytest.mark.asyncio
async def test_cycle_persists_what_the_sweep_learned():
"""The MQTT payload used to be the only consumer, leaving the UI nothing."""
answered, silent = "aa" * 32, "bb" * 32
store = _FakeScopeStore({answered: _repeater_row(), silent: _repeater_row()})
helper = _StubSweep(
{
answered: ScopeResult(STATUS_RESPONDED, "DEN,BOU", transmitted=True),
silent: ScopeResult(STATUS_TIMEOUT, transmitted=True),
}
)
publisher = _publisher(
{"mqtt_brokers": {}},
handler=SimpleNamespace(
has_neighbors_brokers=lambda: True,
has_connected_neighbors_brokers=lambda: True,
publish_neighbors=lambda payload: [("broker", None)],
node_name="n",
public_key="AB" * 32,
),
storage=store,
scope_helper=helper,
)
await publisher.run_cycle(trigger="test")
assert store.scopes[answered]["scopes"] == "DEN,BOU"
assert store.scopes[answered]["responded_at"] is not None
# A silent neighbour is recorded as asked, but claims no scopes.
assert store.scopes[silent]["status"] == STATUS_TIMEOUT
assert store.scopes[silent]["responded_at"] is None
@pytest.mark.asyncio
async def test_neighbors_never_put_on_air_are_not_recorded_as_queried():
"""`timeout` covers both "asked, silent" and "never reached"; only the first counts."""
unreached = "cc" * 32
store = _FakeScopeStore({unreached: _repeater_row()})
publisher = _publisher(
{"mqtt_brokers": {}},
handler=SimpleNamespace(
has_neighbors_brokers=lambda: True,
has_connected_neighbors_brokers=lambda: True,
publish_neighbors=lambda payload: [("broker", None)],
node_name="n",
public_key="AB" * 32,
),
storage=store,
scope_helper=_StubSweep({unreached: ScopeResult(STATUS_TIMEOUT, transmitted=False)}),
)
await publisher.run_cycle(trigger="test")
assert store.writes == []
assert store.scopes == {}
@pytest.mark.asyncio
async def test_query_one_returns_and_stores_the_answer():
target = "dd" * 32
store = _FakeScopeStore({target: _repeater_row(snr=3.5)})
helper = _StubSweep({target: ScopeResult(STATUS_RESPONDED, "DEN", transmitted=True)})
publisher = _publisher({"mqtt_brokers": {}}, storage=store, scope_helper=helper)
out = await publisher.query_one(target.upper())
assert out["status"] == STATUS_RESPONDED
assert out["scopes"] == "DEN"
assert out["responded_at"] is not None
assert store.scopes[target]["scopes"] == "DEN"
# The snapshot borrows the stored row so a single query walks the sweep path.
assert helper.targets[0].snr == 3.5
@pytest.mark.asyncio
async def test_query_one_does_not_publish():
"""Only the periodic cycle writes to the neighbors topic."""
target = "ee" * 32
published = []
handler = SimpleNamespace(
has_neighbors_brokers=lambda: True,
has_connected_neighbors_brokers=lambda: True,
publish_neighbors=lambda payload: published.append(payload) or [("broker", None)],
node_name="n",
public_key="AB" * 32,
)
publisher = _publisher(
{"mqtt_brokers": {}},
handler=handler,
storage=_FakeScopeStore({target: _repeater_row()}),
scope_helper=_StubSweep({target: ScopeResult(STATUS_RESPONDED, "DEN", transmitted=True)}),
)
await publisher.query_one(target)
assert published == []
@pytest.mark.asyncio
@pytest.mark.parametrize("bad", ["", "abcd", "zz" * 32, "aa" * 31])
async def test_query_one_rejects_keys_it_cannot_query(bad):
"""ECDH needs the full 32-byte key; a prefix or non-hex cannot be used."""
publisher = _publisher({"mqtt_brokers": {}}, scope_helper=_StubSweep({}))
with pytest.raises(ValueError):
await publisher.query_one(bad)
@pytest.mark.asyncio
async def test_query_one_refuses_our_own_key():
local = LocalIdentity()
publisher = _publisher({"mqtt_brokers": {}}, scope_helper=_StubSweep({}), local_identity=local)
with pytest.raises(ValueError):
await publisher.query_one(local.get_public_key().hex())
@pytest.mark.asyncio
async def test_query_one_reports_a_sweep_already_in_progress():
"""One request in flight at a time; the wording has to be actionable."""
local = LocalIdentity()
helper = _helper_with_injector(local, lambda packet, wait_for_ack=False: True)
publisher = _publisher({"mqtt_brokers": {}}, scope_helper=helper)
async with helper._sweep_lock:
with pytest.raises(RuntimeError, match="already running"):
await publisher.query_one("ff" * 32)
@pytest.mark.asyncio
async def test_query_one_without_storage_still_answers():
"""Scope queries are useful on a repeater with no storage wired up."""
target = "ab" * 32
helper = _StubSweep({target: ScopeResult(STATUS_RESPONDED, "", transmitted=True)})
publisher = _publisher({"mqtt_brokers": {}}, storage=None, scope_helper=helper)
out = await publisher.query_one(target)
assert out["status"] == STATUS_RESPONDED
assert out["scopes"] == "" # a real answer: unscoped traffic only
def test_neighbor_scopes_round_trip_through_real_sqlite(tmp_path):
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
handler = SQLiteHandler(tmp_path)
pubkey = "1f" * 32
assert handler.get_neighbor_scopes() == {}
assert handler.record_neighbor_scope(pubkey, STATUS_RESPONDED, "DEN,BOU", 1785372000.0) is True
row = handler.get_neighbor_scopes()[pubkey]
assert row == {
"scopes": "DEN,BOU",
"responded_at": 1785372000.0,
"status": STATUS_RESPONDED,
"queried_at": 1785372000.0,
}
# An empty answer is an answer -- the neighbour serves unscoped traffic only.
handler.record_neighbor_scope(pubkey, STATUS_RESPONDED, "", 1785372600.0)
assert handler.get_neighbor_scopes()[pubkey]["scopes"] == ""
# Keys are normalised, so an upper-case caller does not create a second row.
handler.record_neighbor_scope(pubkey.upper(), STATUS_RESPONDED, "DEN", 1785373000.0)
assert list(handler.get_neighbor_scopes()) == [pubkey]
# Survives a restart.
assert SQLiteHandler(tmp_path).get_neighbor_scopes()[pubkey]["scopes"] == "DEN"
def test_failed_query_keeps_the_last_known_scopes(tmp_path):
"""The responder rate-limits anon replies, so one timeout is weak evidence."""
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
handler = SQLiteHandler(tmp_path)
pubkey = "2f" * 32
handler.record_neighbor_scope(pubkey, STATUS_RESPONDED, "DEN", 1785372000.0)
handler.record_neighbor_scope(pubkey, STATUS_TIMEOUT, None, 1785375600.0)
row = handler.get_neighbor_scopes()[pubkey]
assert row["scopes"] == "DEN"
assert row["responded_at"] == 1785372000.0 # still says how fresh the scopes are
assert row["status"] == STATUS_TIMEOUT
assert row["queried_at"] == 1785375600.0
def test_deleting_a_neighbor_drops_its_scope_row(tmp_path):
"""Otherwise the row outlives the neighbour with nothing to display it against."""
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
handler = SQLiteHandler(tmp_path)
pubkey = "3f" * 32
handler.store_advert(
{
"pubkey": pubkey,
"node_name": "gone",
"is_repeater": True,
"contact_type": "Repeater",
"timestamp": time.time(),
}
)
handler.record_neighbor_scope(pubkey, STATUS_RESPONDED, "DEN", time.time())
advert_id = handler.get_adverts_by_contact_type(contact_type="Repeater")[0]["id"]
assert handler.delete_advert(advert_id) is True
assert handler.get_neighbor_scopes() == {}
def test_scope_migration_is_idempotent_on_an_existing_database(tmp_path):
"""Migration 15 runs against nibbler's populated DB, not a fresh file."""
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
pubkey = "4f" * 32
SQLiteHandler(tmp_path).record_neighbor_scope(pubkey, STATUS_RESPONDED, "DEN", 42.0)
for _ in range(3):
SQLiteHandler(tmp_path)._run_migrations()
assert SQLiteHandler(tmp_path).get_neighbor_scopes()[pubkey]["scopes"] == "DEN"
# ====================================================================
# Scope endpoints
# ====================================================================
class _QueryPublisher:
def __init__(self, result=None, error=None):
self.result = result
self.error = error
self.calls = []
async def query_one(self, pubkey):
self.calls.append(pubkey)
if self.error:
raise self.error
return self.result
def test_neighbor_scopes_endpoint_serves_the_stored_table(monkeypatch):
store = _FakeScopeStore(
scopes={"5f" * 32: {"scopes": "DEN", "status": STATUS_RESPONDED, "queried_at": 1.0}}
)
api = _api_with_publisher(monkeypatch, None, method="GET", storage=store)
out = api.neighbor_scopes()
assert out["success"] is True
assert out["count"] == 1
assert out["data"]["5f" * 32]["scopes"] == "DEN"
def test_query_neighbor_scopes_endpoint_returns_the_answer(monkeypatch):
target = "6f" * 32
publisher = _QueryPublisher(
{"pubkey": target, "status": STATUS_RESPONDED, "scopes": "DEN", "transmitted": True}
)
api = _api_with_publisher(monkeypatch, publisher, json={"pubkey": target})
out = _run_endpoint(api, api.query_neighbor_scopes)
assert out["success"] is True
assert out["data"]["scopes"] == "DEN"
assert publisher.calls == [target]
def test_query_neighbor_scopes_endpoint_requires_a_pubkey(monkeypatch):
publisher = _QueryPublisher()
api = _api_with_publisher(monkeypatch, publisher, json={})
out = api.query_neighbor_scopes()
assert out["success"] is False
assert publisher.calls == []
@pytest.mark.parametrize(
"error",
[ValueError("A full 64-character public key is required"), RuntimeError("already running")],
)
def test_query_neighbor_scopes_endpoint_surfaces_refusals(monkeypatch, error):
"""Both a bad key and a busy sweep must read as a message, not a 500."""
publisher = _QueryPublisher(error=error)
api = _api_with_publisher(monkeypatch, publisher, json={"pubkey": "7f" * 32})
out = _run_endpoint(api, api.query_neighbor_scopes)
assert out["success"] is False
assert str(error) in out["error"]
def test_query_neighbor_scopes_endpoint_without_a_publisher(monkeypatch):
api = _api_with_publisher(monkeypatch, None, json={"pubkey": "8f" * 32})
out = api.query_neighbor_scopes()
assert out["success"] is False
assert "not available" in out["error"]
@pytest.mark.parametrize("name", ["query_neighbor_scopes"])
def test_json_body_endpoints_enable_the_json_in_tool(name):
"""cherrypy.request.json only exists when json_in is on for the handler.
It is not enabled globally for /api, and Request has no __getattr__, so a
handler that reads .json without the decorator raises AttributeError on every
real request -- while a test that fabricates cherrypy.request still passes.
That is exactly how this was missed, hence a check on the decorator itself.
"""
from repeater.web.api_endpoints import APIEndpoints
handler = getattr(APIEndpoints, name)
config = getattr(handler, "_cp_config", {})
assert config.get("tools.json_in.on") is True
# ====================================================================
# Scope persistence: the rules the review found broken
# ====================================================================
@pytest.mark.asyncio
async def test_duty_cycle_refusal_is_recorded_as_a_query():
"""Nothing reached the air, but the node did try -- and kept trying.
Recording only `transmitted` outcomes left a repeater that refuses on duty
cycle reading as "never queried" forever, however often it was asked.
"""
target = "1a" * 32
store = _FakeScopeStore({target: _repeater_row()})
publisher = _publisher(
{"mqtt_brokers": {}},
storage=store,
scope_helper=_StubSweep({target: ScopeResult(STATUS_SEND_FAILED, transmitted=False)}),
)
out = await publisher.query_one(target)
assert out["status"] == STATUS_SEND_FAILED
assert out["queried_at"] is not None
assert store.scopes[target]["status"] == STATUS_SEND_FAILED
@pytest.mark.asyncio
async def test_failed_query_returns_the_scopes_still_on_record():
"""The row keeps the last answer, so the response must not tell the UI to drop it."""
target = "2a" * 32
store = _FakeScopeStore({target: _repeater_row()})
publisher = _publisher(
{"mqtt_brokers": {}},
storage=store,
scope_helper=_StubSweep(
{target: ScopeResult(STATUS_RESPONDED, "DEN,BOU", transmitted=True)}
),
)
await publisher.query_one(target)
publisher.scope_helper = _StubSweep({target: ScopeResult(STATUS_TIMEOUT, transmitted=True)})
out = await publisher.query_one(target)
assert out["status"] == STATUS_TIMEOUT
assert out["scopes"] == "DEN,BOU" # not blanked
assert out["responded_at"] is not None
assert out["responded_at"] < out["queried_at"]
@pytest.mark.asyncio
async def test_query_refuses_while_a_cycle_is_running():
"""The cycle owns the helper for its whole run, discovery window included."""
publisher = _publisher({"mqtt_brokers": {}}, scope_helper=_StubSweep({}))
publisher._active = True
with pytest.raises(RuntimeError, match="cycle is running"):
await publisher.query_one("3a" * 32)
@pytest.mark.asyncio
async def test_a_cycle_defers_rather_than_dies_when_a_query_holds_the_helper():
"""Colliding with a manual query must not cost a whole cycle's airtime."""
target = "4a" * 32
store = _FakeScopeStore({target: _repeater_row()})
published = []
handler = SimpleNamespace(
has_neighbors_brokers=lambda: True,
has_connected_neighbors_brokers=lambda: True,
publish_neighbors=lambda payload: published.append(payload) or [("broker", None)],
node_name="n",
public_key="AB" * 32,
)
class _BusyHelper:
async def sweep(self, targets):
raise RuntimeError("neighbor scope sweep already active")
publisher = _publisher(
{"mqtt_brokers": {}}, handler=handler, storage=store, scope_helper=_BusyHelper()
)
out = await publisher.run_cycle(trigger="test")
assert out["success"] is False
# No scope-less table published, and the short retry delay applies.
assert published == []
assert "deferred" in publisher._last_result
assert publisher.status()["secs_until_next"] <= 900
@pytest.mark.asyncio
async def test_a_query_in_flight_blocks_a_cycle_from_starting():
publisher = _publisher(
{"mqtt_brokers": {}}, handler=_enabled_handler(), scope_helper=_StubSweep({})
)
publisher._queries_in_flight = 1
assert publisher.trigger_cycle() is False
publisher._next_publish_at = None # due
await publisher._tick()
assert publisher._last_publish_at is None # no cycle ran
@pytest.mark.asyncio
async def test_shutdown_cancels_an_in_flight_query():
"""A query holds the radio for a response window; teardown must cut it short."""
local = LocalIdentity()
peer = LocalIdentity() # a real key, so the request actually builds
on_air = asyncio.Event()
async def _never_returns(packet, wait_for_ack=False):
on_air.set()
await asyncio.sleep(3600)
helper = _helper_with_injector(local, _never_returns)
publisher = _publisher({"mqtt_brokers": {}}, scope_helper=helper, local_identity=local)
task = asyncio.create_task(publisher.query_one(peer.get_public_key().hex()))
# Waited on an event rather than polling the task set: if the query fails
# before it gets on air this fails fast instead of spinning.
await asyncio.wait_for(on_air.wait(), timeout=5)
assert publisher._query_tasks
await publisher.stop()
assert task.done()
assert publisher._queries_in_flight == 0
def test_retention_cleanup_drops_scope_rows_for_pruned_neighbours(tmp_path):
"""cleanup_old_data is the path that actually runs unattended on the Pi."""
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
handler = SQLiteHandler(tmp_path)
gone, kept = "6a" * 32, "7a" * 32
old = time.time() - (60 * 24 * 3600)
for pubkey, ts in ((gone, old), (kept, time.time())):
handler.store_advert(
{
"pubkey": pubkey,
"node_name": pubkey[:4],
"is_repeater": True,
"contact_type": "Repeater",
"timestamp": ts,
}
)
handler.record_neighbor_scope(pubkey, STATUS_RESPONDED, "DEN", time.time())
handler.cleanup_old_data(days=31)
remaining = handler.get_neighbor_scopes()
assert kept in remaining
assert gone not in remaining
def test_purging_the_advert_table_takes_the_scopes_with_it(tmp_path):
"""Otherwise the UI shows scope counts for repeaters it no longer lists."""
from repeater.data_acquisition.sqlite_handler import SQLiteHandler
handler = SQLiteHandler(tmp_path)
handler.record_neighbor_scope("8a" * 32, STATUS_RESPONDED, "DEN", time.time())
handler.purge_table("adverts")
assert handler.get_neighbor_scopes() == {}