diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index ca53243..5db5f36 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -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: diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 245c227..0a609e0 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -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) diff --git a/repeater/neighbors_publisher.py b/repeater/neighbors_publisher.py index a4083ad..893c450 100644 --- a/repeater/neighbors_publisher.py +++ b/repeater/neighbors_publisher.py @@ -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. diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index b3f2610..5b1b672 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -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. diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index 234c3b5..c6e6d09 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -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: diff --git a/tests/test_mqtt_neighbors.py b/tests/test_mqtt_neighbors.py index 765455a..fc182a6 100644 --- a/tests/test_mqtt_neighbors.py +++ b/tests/test_mqtt_neighbors.py @@ -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() == {}