diff --git a/config.yaml.example b/config.yaml.example index 3d3816c..910dcb6 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -562,6 +562,12 @@ mqtt_brokers: # max_neighbors: 32 # cap on neighbours queried per cycle # max_neighbor_age_seconds: 86400 # ignore zero-hop rows older than this # max_sweep_seconds: 900 # give up on a cycle after this long + # duty_cycle_abort_seconds: 30 # abandon the sweep when the airtime + # # backlog exceeds this + # + # This must be a block, not a boolean - `neighbors: true` here is ignored (with + # a startup warning) because the on/off switch that matters is the per-broker + # `neighbors: true` flag documented in the broker schema below. # # `discover.scopes` over the mesh CLI runs one cycle immediately. diff --git a/repeater/data_acquisition/sqlite_handler.py b/repeater/data_acquisition/sqlite_handler.py index d4dca65..ca53243 100644 --- a/repeater/data_acquisition/sqlite_handler.py +++ b/repeater/data_acquisition/sqlite_handler.py @@ -735,11 +735,78 @@ class SQLiteHandler: ) logger.info(f"Migration '{migration_name}' applied successfully") + # Migration 14: Small key/value store for daemon state that must + # outlive a restart. Added for the neighbours publisher, whose + # schedule otherwise resets on every boot and re-runs a discovery + # sweep; kept generic so the next such need does not add another + # table. Not for anything hot -- one row per writer, rewritten at + # whatever cadence that writer already has. + migration_name = "add_daemon_state" + existing = conn.execute( + "SELECT migration_name FROM migrations WHERE migration_name = ?", + (migration_name,), + ).fetchone() + if not existing: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS daemon_state ( + key TEXT PRIMARY KEY, + value_json TEXT NOT NULL, + updated_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}") + # Daemon state methods + def get_daemon_state(self, key: str) -> Optional[dict]: + """Read a persisted daemon-state blob, or None when absent/unreadable. + + Never raises: every caller treats missing state as "no history", so a + corrupt row must degrade to that rather than break startup. + """ + try: + with self._connect() as conn: + row = conn.execute( + "SELECT value_json FROM daemon_state WHERE key = ?", (key,) + ).fetchone() + if not row: + return None + value = json.loads(row[0]) + return value if isinstance(value, dict) else None + except Exception as e: + logger.debug(f"Could not read daemon state '{key}': {e}") + return None + + def set_daemon_state(self, key: str, value: dict) -> bool: + """Upsert a daemon-state blob. Returns whether it was written.""" + try: + with self._connect() as conn: + conn.execute( + """ + INSERT INTO daemon_state (key, value_json, updated_at) + VALUES (?, ?, ?) + ON CONFLICT(key) DO UPDATE SET + value_json = excluded.value_json, + updated_at = excluded.updated_at + """, + (key, json.dumps(value), time.time()), + ) + conn.commit() + return True + except Exception as e: + logger.warning(f"Could not persist daemon state '{key}': {e}") + return False + # API Token methods def create_api_token(self, name: str, token_hash: str) -> int: """Create a new API token entry""" diff --git a/repeater/data_acquisition/storage_collector.py b/repeater/data_acquisition/storage_collector.py index 1410a5d..245c227 100644 --- a/repeater/data_acquisition/storage_collector.py +++ b/repeater/data_acquisition/storage_collector.py @@ -539,6 +539,12 @@ class StorageCollector: def get_neighbors(self) -> dict: return self.sqlite_handler.get_neighbors() + def get_daemon_state(self, key: str) -> Optional[dict]: + return self.sqlite_handler.get_daemon_state(key) + + def set_daemon_state(self, key: str, value: dict) -> bool: + return self.sqlite_handler.set_daemon_state(key, value) + def get_node_name_by_pubkey(self, pubkey: str) -> Optional[str]: """ Lookup node name from adverts table by public key. diff --git a/repeater/handler_helpers/neighbor_scopes.py b/repeater/handler_helpers/neighbor_scopes.py index 181658d..cb9e516 100644 --- a/repeater/handler_helpers/neighbor_scopes.py +++ b/repeater/handler_helpers/neighbor_scopes.py @@ -68,6 +68,32 @@ DEFAULT_MAX_SWEEP_SECONDS = 900.0 DEFAULT_DUTY_CYCLE_ABORT_SECONDS = 30.0 +def neighbors_config_block(config: Optional[dict]) -> dict: + """Return ``mqtt_brokers.neighbors`` as a mapping, or ``{}`` when it is not one. + + ``neighbors`` names a settings block under ``mqtt_brokers`` but a plain + boolean on each broker entry, and ``config.yaml.example`` documents both, so + ``mqtt_brokers.neighbors: true`` is an easy hand-edit to make. A truthy + non-mapping used to reach ``.get()`` directly, which raised AttributeError + inside :meth:`NeighborScopeHelper.refresh_config` — and because that helper is + built during daemon init, it took the whole daemon down on startup. Ignore the + value instead; saving from the API rewrites it as a proper block. + """ + if not isinstance(config, dict): + return {} + brokers_cfg = config.get("mqtt_brokers", {}) + if not isinstance(brokers_cfg, dict): + return {} + block = brokers_cfg.get("neighbors", {}) + if not isinstance(block, dict): + logger.debug( + "Ignoring mqtt_brokers.neighbors: expected a settings block, got %s", + type(block).__name__, + ) + return {} + return block + + @dataclass(frozen=True) class NeighborSnapshot: """One neighbour, frozen at sweep start. @@ -91,6 +117,12 @@ class NeighborSnapshot: class ScopeResult: status: str scopes: str = "" + # Whether the request actually reached the air. Feeds the payload's + # ``queried_neighbors``, which firmware increments in ``logTx`` when a QUEUED + # entry becomes PENDING. It is not derivable from ``status``: a target the + # sweep never reached is reported as ``timeout`` exactly like one that was + # asked and stayed silent. + transmitted: bool = False @dataclass @@ -149,7 +181,7 @@ class NeighborScopeHelper: config = self.config else: self.config = config - neighbors_cfg = (config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {} + neighbors_cfg = neighbors_config_block(config) self._response_timeout_override = _as_float( neighbors_cfg.get("scope_response_timeout_seconds", 0), 0.0 @@ -163,8 +195,10 @@ class NeighborScopeHelper: neighbors_cfg.get("duty_cycle_abort_seconds"), DEFAULT_DUTY_CYCLE_ABORT_SECONDS ), ) + delays_cfg = config.get("delays", {}) if isinstance(config, dict) else {} self._direct_tx_delay_factor = _as_float( - (config.get("delays", {}) or {}).get("direct_tx_delay_factor"), 0.5 + delays_cfg.get("direct_tx_delay_factor") if isinstance(delays_cfg, dict) else None, + 0.5, ) def response_timeout(self) -> float: @@ -308,13 +342,15 @@ class NeighborScopeHelper: logger.debug(f"Scope request not transmitted for {target.pubkey[:8]}") return ScopeResult(STATUS_SEND_FAILED) + # Past this point the request is on air, so every outcome counts as + # queried regardless of whether the neighbour answers. try: scopes = await asyncio.wait_for(pending.future, timeout) except asyncio.TimeoutError: logger.debug(f"Scope query timed out for {target.pubkey[:8]}") - return ScopeResult(STATUS_TIMEOUT) + return ScopeResult(STATUS_TIMEOUT, transmitted=True) logger.debug(f"Scope response from {target.pubkey[:8]}: '{scopes}'") - return ScopeResult(STATUS_RESPONDED, scopes) + return ScopeResult(STATUS_RESPONDED, scopes, transmitted=True) finally: self._pending = None diff --git a/repeater/neighbors_publisher.py b/repeater/neighbors_publisher.py index 77c1014..a4083ad 100644 --- a/repeater/neighbors_publisher.py +++ b/repeater/neighbors_publisher.py @@ -31,6 +31,7 @@ from repeater.handler_helpers.neighbor_scopes import ( STATUS_TIMEOUT, NeighborSnapshot, ScopeResult, + neighbors_config_block, ) logger = logging.getLogger("NeighborsPublisher") @@ -58,6 +59,19 @@ _TICK_SECONDS = 30.0 # waiting out the full interval. RETRY_DELAY_SECONDS = 900.0 +# How an unset default region is spelled in the payload. Matches the wildcard +# LoginHelper._format_region_names already emits for unscoped flood. +DEFAULT_SCOPE_WILDCARD = "*" + +# daemon_state key holding the persisted schedule. +STATE_KEY = "neighbors_publisher" + +# A restored schedule never fires inside this window after boot, even when it is +# already overdue. Keeps a restart quiet and lets the radio and brokers settle +# before a cycle claims the airtime. Firmware has no equivalent -- it treats +# every boot as immediately due. +STARTUP_GRACE_SECONDS = 300.0 + # Phases reported by status(), mirroring the firmware's NeighborsPhase. PHASE_DISABLED = "disabled" PHASE_SCHEDULED = "scheduled" @@ -72,6 +86,9 @@ def build_neighbors_payload( self_scopes: str, entries: List[dict], timestamp: Optional[str] = None, + total_neighbors: Optional[int] = None, + queried_neighbors: Optional[int] = None, + self_default_scope: str = DEFAULT_SCOPE_WILDCARD, ) -> dict: """Assemble the ``neighbors`` topic payload. @@ -80,18 +97,47 @@ def build_neighbors_payload( needs that order so it can drop the tail when its fixed buffer fills; we keep it because it is the documented shape of the topic and it puts the useful rows first for consumers. + + ``self`` carries this node's own advertised scopes plus ``default_scope``, the + region it stamps on outgoing floods (``*`` when it floods unscoped). The + firmware tracks a ``default_scope`` internally but does not publish it; it is + included here because a consumer reading the table cannot otherwise tell which + of several scopes this node actually transmits under. + + Two progress counters mirror firmware ``buildNeighborsMessage``: + + * ``total_neighbors`` — neighbours in this cycle's table, which is also how + many rows ``neighbors`` carries. + * ``queried_neighbors`` — how many scope requests reached the air. + + The firmware's third field, ``truncated``, is deliberately not emitted: it + reports that a fixed PSRAM JSON buffer filled and the tail was dropped, and + openhop has no such buffer, so it could only ever be false. That also keeps + ``total_neighbors`` equal to the published row count here, where firmware + allows it to run ahead. Both are emitted only when the caller supplies the + counts, matching the firmware's ``total_neighbors >= 0`` guard. """ ordered = sorted( entries, key=lambda e: (e.get("heard_secs_ago", 0), -float(e.get("snr", 0.0)), e.get("pubkey", "")), ) - return { + payload = { "timestamp": timestamp or datetime.now(timezone.utc).isoformat(), "origin": origin, "origin_id": origin_id, - "self": {"scopes": self_scopes or ""}, - "neighbors": ordered, } + # Key order matches the firmware writer so the two payloads diff cleanly. + if total_neighbors is not None: + payload["total_neighbors"] = int(total_neighbors) + payload["queried_neighbors"] = int( + queried_neighbors if queried_neighbors is not None else total_neighbors + ) + payload["self"] = { + "scopes": self_scopes or "", + "default_scope": self_default_scope or DEFAULT_SCOPE_WILDCARD, + } + payload["neighbors"] = ordered + return payload class NeighborsPublisher: @@ -124,14 +170,26 @@ class NeighborsPublisher: self._discovery_seen: Dict[str, dict] = {} self._next_publish_at: Optional[float] = None self._last_result: Optional[str] = None + # When the last cycle finished, whatever its outcome -- this is what + # status() reports alongside last_result. self._last_publish_at: Optional[float] = None + # When a cycle last actually reached a broker. The schedule is measured + # from this, not from the above: a cycle that failed to publish reschedules + # on the short retry delay, and restoring from a failed attempt would + # silently turn that retry into a full interval. + self._last_success_at: Optional[float] = None + # Whether the feature has been enabled at any point in this process. The + # disabled branch of _tick clears the schedule so re-enabling publishes + # promptly, which must not fire before we have ever been enabled -- that + # would throw away a schedule just restored from disk. + self._was_enabled = False # ------------------------------------------------------------------ # Config accessors (re-read every cycle so live edits take effect) # ------------------------------------------------------------------ @property def _neighbors_config(self) -> dict: - return (self.config.get("mqtt_brokers", {}) or {}).get("neighbors", {}) or {} + return neighbors_config_block(self.config) @property def master_enabled(self) -> bool: @@ -181,10 +239,87 @@ class NeighborsPublisher: def start(self) -> None: if self._task is not None and not self._task.done(): return + # Say so once, loudly, rather than from the config accessors: those run + # several times per tick and would bury the log. + raw_block = (self.config.get("mqtt_brokers") or {}) if isinstance(self.config, dict) else {} + if isinstance(raw_block, dict) and not isinstance(raw_block.get("neighbors", {}), dict): + logger.warning( + "mqtt_brokers.neighbors is %s, not a settings block - using defaults. " + "The per-broker 'neighbors: true' flag is what opts a broker in.", + type(raw_block.get("neighbors")).__name__, + ) + self._restore_schedule() self._running = True self._task = asyncio.create_task(self._run_loop(), name="neighbors-publisher") logger.info("Neighbors publisher started (interval %.1fh)", self.interval_seconds / 3600.0) + # ------------------------------------------------------------------ + # Schedule persistence + # ------------------------------------------------------------------ + def _restore_schedule(self) -> None: + """Resume the schedule from the last publish instead of restarting it. + + Without this a restart leaves ``_next_publish_at`` unset, which reads as + "due" and spends a discovery broadcast plus a serialized scope query per + neighbour on every boot. ``_next_publish_at`` is monotonic and so cannot + be stored directly; the persisted value is the wall-clock publish time, + converted back to a monotonic deadline here. + """ + storage = self._storage() + reader = getattr(storage, "get_daemon_state", None) if storage else None + if not callable(reader): + # Older storage backend, or none wired up: behave as before. + self._next_publish_at = time.monotonic() + STARTUP_GRACE_SECONDS + return + + state = reader(STATE_KEY) or {} + last = _as_epoch(state.get("last_success_at")) + + # Restore the display fields too, so a restart does not report the node as + # having never run. + self._last_result = state.get("last_result") or None + self._last_publish_at = _as_epoch(state.get("last_publish_at")) or None + interval = self.interval_seconds + now = time.time() + + if last <= 0 or last > now: + # No successful publish on record, or a timestamp from the future -- + # the clock moved backwards, or the row is junk. Either way it cannot + # place the next cycle, so fall back to the grace delay. + if last > now: + logger.warning( + "Persisted neighbours publish time is %.0fs in the future; ignoring it", + last - now, + ) + self._next_publish_at = time.monotonic() + STARTUP_GRACE_SECONDS + return + + self._last_success_at = last + # Never sooner than the grace window, never later than a full interval + # from now -- the latter bounds the damage from an interval that shrank + # since the last publish. + delay = min(max((last + interval) - now, STARTUP_GRACE_SECONDS), interval) + self._next_publish_at = time.monotonic() + delay + logger.info( + "Neighbours schedule resumed: last published %.1fh ago, next in %.1fh", + (now - last) / 3600.0, + delay / 3600.0, + ) + + def _persist_schedule(self) -> None: + storage = self._storage() + writer = getattr(storage, "set_daemon_state", None) if storage else None + if not callable(writer): + return + writer( + STATE_KEY, + { + "last_success_at": self._last_success_at, + "last_publish_at": self._last_publish_at, + "last_result": self._last_result, + }, + ) + def trigger_cycle(self) -> bool: """Start a manual cycle, tracked so shutdown can cancel it. @@ -303,10 +438,16 @@ class NeighborsPublisher: async def _tick(self) -> None: if not self.enabled(): # Drop the schedule so re-enabling runs a pass promptly, matching the - # firmware's next_neighbors_publish = 0 reset. - self._next_publish_at = None + # firmware's next_neighbors_publish = 0 reset. Only once the feature + # has actually been on in this process, though: at boot the MQTT + # handler may not have its connections up yet, and clearing here would + # discard the schedule just restored from disk and re-run the sweep + # anyway -- the exact thing persistence exists to prevent. + if self._was_enabled: + self._next_publish_at = None return + self._was_enabled = True handler = self._mqtt_handler() if not handler or not handler.has_connected_neighbors_brokers(): logger.debug("Neighbors publish deferred: no connected opted-in broker") @@ -361,6 +502,9 @@ class NeighborsPublisher: else "publish failed (broker unreachable or rejected the payload)" ) self._last_publish_at = time.time() + if published: + self._last_success_at = self._last_publish_at + self._persist_schedule() logger.info( "Neighbors %s cycle finished in %.1fs: %d neighbour(s), %d with scopes, " "published=%s", @@ -486,6 +630,8 @@ class NeighborsPublisher: # the same ordering the firmware applies before it starts querying. snapshots.sort(key=lambda s: (-s.last_seen, -s.snr, s.pubkey)) if len(snapshots) > self.max_neighbors: + # Only logged, not published: the payload reports the capped table, so + # the dropped rows would otherwise be invisible here. logger.info( "Neighbour table has %d entries; querying the freshest %d", len(snapshots), @@ -495,12 +641,17 @@ class NeighborsPublisher: return snapshots def _build_payload( - self, targets: List[NeighborSnapshot], scope_results: Dict[str, ScopeResult] + self, + targets: List[NeighborSnapshot], + scope_results: Dict[str, ScopeResult], ) -> dict: now = time.time() entries = [] + queried = 0 for target in targets: result = scope_results.get(target.pubkey) or ScopeResult(STATUS_TIMEOUT) + if result.transmitted: + queried += 1 heard_secs_ago = int(max(0.0, now - target.last_seen)) if target.last_seen else 0 entries.append( { @@ -527,9 +678,33 @@ class NeighborsPublisher: origin=origin, origin_id=origin_id, self_scopes=self._self_scopes(), + self_default_scope=self._self_default_scope(), entries=entries, + total_neighbors=len(entries), + queried_neighbors=queried, ) + def _self_default_scope(self) -> str: + """The region this node stamps on outgoing floods, or ``*`` when unset. + + Read from live config on every cycle because ``region default `` over + the mesh CLI writes straight into ``config["mesh"]`` (the same dict this + holds) and expects to take effect without a restart. + + Normalised like the ``scopes`` field beside it: the transport-key table + stores region names with a leading ``#``, which is not part of the name a + consumer matches on, so it is stripped here as + ``LoginHelper._format_region_names`` strips it there. + """ + mesh_cfg = self.config.get("mesh", {}) if isinstance(self.config, dict) else {} + if not isinstance(mesh_cfg, dict): + return DEFAULT_SCOPE_WILDCARD + raw = mesh_cfg.get("default_region") + name = str(raw).strip() if raw not in (None, "") else "" + if name.startswith("#"): + name = name[1:].strip() + return name or DEFAULT_SCOPE_WILDCARD + def _self_scopes(self) -> str: if not self._self_scopes_fn: return "" @@ -551,6 +726,16 @@ class NeighborsPublisher: return bool(results) +def _as_epoch(value) -> float: + """Coerce a persisted timestamp to a float, or 0.0 when it is unusable.""" + try: + if value is None: + return 0.0 + return float(value) + except (TypeError, ValueError): + return 0.0 + + def normalize_interval_hours(value) -> float: """Validate an interval in hours, falling back to the default when invalid. diff --git a/repeater/web/api_endpoints.py b/repeater/web/api_endpoints.py index 8b89225..b3f2610 100644 --- a/repeater/web/api_endpoints.py +++ b/repeater/web/api_endpoints.py @@ -2337,6 +2337,60 @@ class APIEndpoints: logger.error(f"Error listing broker presets: {e}") return self._error(str(e)) + @cherrypy.expose + @cherrypy.tools.json_out() + def publish_neighbors(self): + """Run one neighbours discovery + publish cycle now. + + POST /api/publish_neighbors + + The HTTP equivalent of the ``discover.scopes`` mesh CLI command. A cycle + runs for minutes -- a discovery window plus one serialized scope query per + neighbour -- so this schedules it on the event loop and returns + immediately rather than holding the request open. Poll ``mqtt_status`` + for the outcome. + """ + self._set_cors_headers() + + if cherrypy.request.method == "OPTIONS": + return "" + + future = None + try: + self._require_post() + publisher = getattr(self.daemon_instance, "neighbors_publisher", None) + if not publisher: + return self._error("Neighbors publisher not available") + if not publisher.enabled(): + return self._error( + "Neighbours publishing is disabled - enable it and opt a broker in first" + ) + if self.event_loop is None: + return self._error("Event loop not available") + + import asyncio + + # trigger_cycle owns the already-running check and tracks the task so + # shutdown can cancel it. Doing either here would race: this runs on a + # cherrypy thread while the cycle lives on the event loop. + async def _start(): + return publisher.trigger_cycle() + + future = asyncio.run_coroutine_threadsafe(_start(), self.event_loop) + if future.result(timeout=10): + return self._success("Neighbours discovery cycle started") + return self._error("A neighbours cycle is already running") + except FutureTimeoutError: + logger.error("Timed out starting the neighbours cycle", exc_info=True) + if future is not None: + future.cancel() + return self._error("Timed out starting the neighbours cycle") + except cherrypy.HTTPError: + raise + except Exception as e: + logger.error(f"Error starting neighbours cycle: {e}", exc_info=True) + return self._error(str(e)) + @staticmethod def _validate_neighbors_settings(raw): """Validate the ``mqtt_brokers.neighbors`` block. @@ -2489,16 +2543,22 @@ class APIEndpoints: if "email" in data: mqtt_updates["email"] = str(data["email"]).strip() if "neighbors" in data: + from repeater.handler_helpers.neighbor_scopes import neighbors_config_block + neighbors_settings, error = self._validate_neighbors_settings(data["neighbors"]) if error: return self._error(error) # update_and_save replaces a section's key outright, so merge onto # the stored block instead of letting a partial POST drop the - # settings it did not mention. - existing_neighbors = (self.config.get("mqtt_brokers", {}) or {}).get( - "neighbors", {} - ) or {} - mqtt_updates["neighbors"] = {**existing_neighbors, **neighbors_settings} + # settings it did not mention. neighbors_config_block returns {} for + # a hand-edited scalar (`neighbors: true` is an easy mistake, since + # the per-broker key of the same name is a boolean), which would + # otherwise fail the merge with a TypeError; the save then rewrites + # it as a proper block. + mqtt_updates["neighbors"] = { + **neighbors_config_block(self.config), + **neighbors_settings, + } # if "disallowed_packet_types" in data: # mqtt_updates["disallowed_packet_types"] = list(data["disallowed_packet_types"]) if "brokers" in data: diff --git a/repeater/web/openapi.yaml b/repeater/web/openapi.yaml index 62d4f74..234c3b5 100644 --- a/repeater/web/openapi.yaml +++ b/repeater/web/openapi.yaml @@ -3417,6 +3417,28 @@ paths: schema: type: object + /publish_neighbors: + post: + tags: [System] + summary: Publish the neighbours table now + description: > + Runs one neighbours cycle immediately: a zero-hop discovery broadcast, a + serialized scope query per neighbour, then a publish to every opted-in + broker. Returns as soon as the cycle is scheduled; the cycle itself takes + minutes. Poll /mqtt_status for the outcome. + security: + - BearerAuth: [] + - ApiKeyAuth: [] + responses: + '200': + description: Cycle started, or an error when one is already running + content: + application/json: + schema: + $ref: '#/components/schemas/SuccessResponse' + '405': + description: Method not allowed + /update_web_config: post: tags: [System] diff --git a/tests/test_mqtt_neighbors.py b/tests/test_mqtt_neighbors.py index f3158cf..765455a 100644 --- a/tests/test_mqtt_neighbors.py +++ b/tests/test_mqtt_neighbors.py @@ -21,6 +21,7 @@ from openhop_core.protocol import CryptoUtils, Identity, LocalIdentity from openhop_core.protocol.constants import PAYLOAD_TYPE_RESPONSE from repeater.data_acquisition.mqtt_handler import MeshCoreToMqttPusher from repeater.handler_helpers.neighbor_scopes import ( + DEFAULT_MAX_SWEEP_SECONDS, STATUS_RESPONDED, STATUS_SEND_FAILED, STATUS_TIMEOUT, @@ -28,8 +29,11 @@ from repeater.handler_helpers.neighbor_scopes import ( NeighborSnapshot, ) from repeater.neighbors_publisher import ( + DEFAULT_INTERVAL_HOURS, MAX_INTERVAL_HOURS, MIN_INTERVAL_HOURS, + STARTUP_GRACE_SECONDS, + STATE_KEY, NeighborsPublisher, build_neighbors_payload, normalize_interval_hours, @@ -100,7 +104,7 @@ def test_payload_orders_most_useful_first(): ) assert [e["pubkey"] for e in payload["neighbors"]] == ["bb", "aa", "cc"] - assert payload["self"] == {"scopes": "DEN,APRS"} + assert payload["self"] == {"scopes": "DEN,APRS", "default_scope": "*"} assert payload["origin_id"] == "AA" * 32 assert payload["timestamp"] @@ -823,7 +827,7 @@ async def test_cycle_publishes_table_with_unanswered_neighbors(): assert result["responded"] == 1 payload = published[0] - assert payload["self"] == {"scopes": "DEN,APRS"} + assert payload["self"] == {"scopes": "DEN,APRS", "default_scope": "*"} statuses = {e["pubkey"]: e["status"] for e in payload["neighbors"]} assert statuses == {"aa" * 32: STATUS_RESPONDED, "bb" * 32: STATUS_TIMEOUT} # A cycle always arms the next one, so a failure cannot wedge the schedule. @@ -1015,3 +1019,576 @@ def test_invalid_neighbors_interval_is_rejected_by_the_endpoint(monkeypatch): assert out["success"] is False assert "between 12 and 336" in out["error"] api.config_manager.update_and_save.assert_not_called() + + +# ==================================================================== +# Malformed config block +# ==================================================================== +# `neighbors` is a settings block under mqtt_brokers but a boolean on each broker +# entry, and config.yaml.example documents both, so `mqtt_brokers.neighbors: true` +# is an easy hand-edit to make. Every reader used to call .get() on it directly. +@pytest.mark.parametrize("bogus", [True, 24, "on", ["DEN"]]) +def test_scalar_neighbors_block_is_ignored_not_fatal(bogus): + """A truthy scalar took the daemon down: the scope helper reads it in __init__.""" + config = {"mqtt_brokers": {"neighbors": bogus}} + + helper = NeighborScopeHelper( + local_identity=LocalIdentity(), packet_injector=None, config=config + ) + assert helper._max_sweep_seconds == DEFAULT_MAX_SWEEP_SECONDS + + publisher = NeighborsPublisher(config=config) + assert publisher._neighbors_config == {} + assert publisher.master_enabled is True + assert publisher.interval_seconds == DEFAULT_INTERVAL_HOURS * 3600.0 + assert publisher.enabled() is False # no handler, so nothing is published + + +@pytest.mark.parametrize("bogus", [True, 24, "on"]) +def test_scalar_neighbors_block_is_repaired_by_a_save(monkeypatch, bogus): + """The merge onto the stored block must not fail on a non-mapping.""" + api, request = _api_with_stored_brokers(monkeypatch, [], neighbors_block=bogus) + + request.json = {"neighbors": {"enabled": True, "interval_hours": 36}} + assert api.update_mqtt_config()["success"] is True + + saved = api.config_manager.update_and_save.call_args.kwargs["updates"]["mqtt_brokers"] + assert saved["neighbors"] == {"enabled": True, "interval_hours": 36} + + +def test_scalar_delays_block_does_not_break_the_response_window(): + helper = NeighborScopeHelper( + local_identity=LocalIdentity(), + packet_injector=None, + config={"delays": True}, + ) + assert helper._direct_tx_delay_factor == 0.5 + + +# ==================================================================== +# Progress metadata (firmware total_neighbors / queried_neighbors) +# ==================================================================== +def test_payload_reports_progress_metadata_in_firmware_key_order(): + payload = build_neighbors_payload( + origin="node", + origin_id="AA" * 32, + self_scopes="DEN", + entries=[ + { + "pubkey": "aa", + "snr": 1.0, + "heard_secs_ago": 5, + "scopes": "DEN", + "status": "responded", + } + ], + total_neighbors=4, + queried_neighbors=2, + ) + + assert payload["total_neighbors"] == 4 + assert payload["queried_neighbors"] == 2 + # Firmware's buildNeighborsMessageBase writes the counters between origin_id + # and self; keeping the order lets the two payloads diff cleanly. + assert list(payload) == [ + "timestamp", + "origin", + "origin_id", + "total_neighbors", + "queried_neighbors", + "self", + "neighbors", + ] + + +def test_payload_never_emits_the_firmware_truncated_field(): + """It reports a fixed PSRAM buffer overflowing, which openhop cannot have.""" + payload = build_neighbors_payload( + origin="node", + origin_id="AA" * 32, + self_scopes="", + entries=[], + total_neighbors=3, + queried_neighbors=1, + ) + + assert "truncated" not in payload + + +def test_payload_omits_progress_metadata_when_counts_are_absent(): + """Mirrors the firmware's `total_neighbors >= 0` guard.""" + payload = build_neighbors_payload( + origin="node", origin_id="AA" * 32, self_scopes="", entries=[] + ) + + assert "total_neighbors" not in payload + assert "queried_neighbors" not in payload + + +def test_queried_count_excludes_neighbors_never_put_on_air(): + """`status` cannot carry this: an unreached target also reports `timeout`.""" + from repeater.handler_helpers.neighbor_scopes import ScopeResult + + targets = [ + NeighborSnapshot(pubkey=f"{i:064x}", last_seen=time.time(), snr=1.0) for i in range(4) + ] + publisher = _publisher({"mqtt_brokers": {}}) + + payload = publisher._build_payload( + targets, + { + targets[0].pubkey: ScopeResult(STATUS_RESPONDED, "DEN", transmitted=True), + targets[1].pubkey: ScopeResult(STATUS_TIMEOUT, transmitted=True), + targets[2].pubkey: ScopeResult(STATUS_SEND_FAILED), # never transmitted + targets[3].pubkey: ScopeResult(STATUS_TIMEOUT), # sweep never reached it + }, + ) + + assert payload["queried_neighbors"] == 2 + + +def test_total_neighbors_always_matches_the_published_row_count(): + """Without `truncated` to flag a gap, the two must not diverge.""" + publisher = _publisher({"mqtt_brokers": {}}) + targets = [ + NeighborSnapshot(pubkey=f"{i:064x}", last_seen=time.time(), snr=1.0) for i in range(3) + ] + + payload = publisher._build_payload(targets, {}) + + assert payload["total_neighbors"] == len(payload["neighbors"]) == 3 + + +@pytest.mark.asyncio +async def test_transmitted_flag_tracks_whether_the_request_reached_the_air(): + local = LocalIdentity() + peer = LocalIdentity() + target = NeighborSnapshot(pubkey=peer.get_public_key().hex(), last_seen=time.time()) + + async def failed(packet, wait_for_ack=False): + return False + + results = await _helper_with_injector(local, failed).sweep([target]) + assert results[target.pubkey].transmitted is False + + async def sent(packet, wait_for_ack=False): + return True + + helper = _helper_with_injector( + local, + sent, + config={"mqtt_brokers": {"neighbors": {"scope_response_timeout_seconds": 0.05}}}, + ) + results = await helper.sweep([target]) + assert results[target.pubkey].status == STATUS_TIMEOUT + assert results[target.pubkey].transmitted is True + + +# ==================================================================== +# Manual trigger endpoint +# ==================================================================== +def _api_with_publisher(monkeypatch, publisher, method="POST"): + import cherrypy + + from repeater.web.api_endpoints import APIEndpoints + + request = SimpleNamespace(method=method, params={}, json={}) + 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.event_loop = asyncio.new_event_loop() + api.send_advert_func = None + api.stats_getter = None + api._config_path = "/tmp/test-config.yaml" + api.config_manager = MagicMock() + return api + + +class _FakePublisher: + def __init__(self, *, is_enabled=True, starts=True): + self._enabled = is_enabled + self._starts = starts + self.triggered = 0 + + def enabled(self): + return self._enabled + + def trigger_cycle(self): + self.triggered += 1 + return self._starts + + +def _run_endpoint(api): + """Drive the endpoint's run_coroutine_threadsafe against a real loop.""" + import threading + + loop = api.event_loop + thread = threading.Thread(target=loop.run_forever, daemon=True) + thread.start() + try: + return api.publish_neighbors() + finally: + loop.call_soon_threadsafe(loop.stop) + thread.join(timeout=5) + # Closing a loop whose thread has not finished unwinding can raise on its + # self-pipe descriptors; leak it rather than risk a flaky teardown. + if not thread.is_alive(): + loop.close() + + +def test_publish_neighbors_endpoint_starts_a_cycle(monkeypatch): + publisher = _FakePublisher() + api = _api_with_publisher(monkeypatch, publisher) + + out = _run_endpoint(api) + + assert out["success"] is True + assert publisher.triggered == 1 + + +def test_publish_neighbors_endpoint_reports_an_already_running_cycle(monkeypatch): + publisher = _FakePublisher(starts=False) + api = _api_with_publisher(monkeypatch, publisher) + + out = _run_endpoint(api) + + assert out["success"] is False + assert "already running" in out["error"] + + +def test_publish_neighbors_endpoint_refuses_when_disabled(monkeypatch): + """No broker opted in: refuse rather than burn airtime on an unpublishable cycle.""" + publisher = _FakePublisher(is_enabled=False) + api = _api_with_publisher(monkeypatch, publisher) + + out = api.publish_neighbors() + + assert out["success"] is False + assert publisher.triggered == 0 + + +def test_publish_neighbors_endpoint_without_a_publisher(monkeypatch): + api = _api_with_publisher(monkeypatch, None) + + out = api.publish_neighbors() + + assert out["success"] is False + assert "not available" in out["error"] + + +# ==================================================================== +# self.default_scope +# ==================================================================== +@pytest.mark.parametrize( + "mesh_cfg,expected", + [ + ({"default_region": "DEN"}, "DEN"), + ({"default_region": "#DEN"}, "DEN"), # transport-key tables prefix with '#' + ({"default_region": " DEN "}, "DEN"), + ({"default_region": None}, "*"), # unset -> floods unscoped + ({"default_region": ""}, "*"), + ({"default_region": " "}, "*"), + ({}, "*"), # key absent entirely + ({"default_region": "*"}, "*"), + (True, "*"), # hand-edited scalar must not raise + ], +) +def test_self_default_scope_normalisation(mesh_cfg, expected): + publisher = _publisher({"mqtt_brokers": {}, "mesh": mesh_cfg}) + assert publisher._self_default_scope() == expected + + +def test_default_scope_is_published_inside_self(): + publisher = _publisher({"mqtt_brokers": {}, "mesh": {"default_region": "#PDX"}}) + + payload = publisher._build_payload([], {}) + + assert payload["self"]["default_scope"] == "PDX" + assert "scopes" in payload["self"] + + +def test_default_scope_tracks_a_live_config_edit(): + """`region default ` writes into the same dict and must not need a restart.""" + config = {"mqtt_brokers": {}, "mesh": {"default_region": None}} + publisher = _publisher(config) + + assert publisher._build_payload([], {})["self"]["default_scope"] == "*" + + config["mesh"]["default_region"] = "DEN" + assert publisher._build_payload([], {})["self"]["default_scope"] == "DEN" + + +def test_payload_defaults_default_scope_to_the_wildcard(): + """Callers that omit it still emit a valid self block.""" + payload = build_neighbors_payload( + origin="node", origin_id="AA" * 32, self_scopes="DEN", entries=[] + ) + + assert payload["self"] == {"scopes": "DEN", "default_scope": "*"} + + +# ==================================================================== +# Schedule persistence across restarts +# ==================================================================== +class _FakeStateStore: + """Stands in for the daemon_state table.""" + + def __init__(self, initial=None): + self.rows = dict(initial or {}) + self.writes = 0 + + def get_daemon_state(self, key): + return self.rows.get(key) + + def set_daemon_state(self, key, value): + self.rows[key] = dict(value) + self.writes += 1 + return True + + +def _enabled_handler(): + return SimpleNamespace( + has_neighbors_brokers=lambda: True, has_connected_neighbors_brokers=lambda: True + ) + + +def test_restore_resumes_the_interval_from_the_last_successful_publish(): + now = time.time() + store = _FakeStateStore({STATE_KEY: {"last_success_at": now - 3600, "last_result": "ok"}}) + publisher = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=_enabled_handler(), + storage=store, + ) + + publisher._restore_schedule() + + # 1h since the last publish on a 24h interval -> ~23h to go, not "due now". + secs = publisher.status()["secs_until_next"] + assert 22.9 * 3600 < secs < 23.1 * 3600 + assert publisher.status()["phase"] == "scheduled" + assert publisher._last_result == "ok" + + +def test_restore_applies_the_grace_delay_when_already_overdue(): + """A node off for a week must not transmit the instant it boots.""" + store = _FakeStateStore( + {STATE_KEY: {"last_success_at": time.time() - 7 * 86400, "last_result": "ok"}} + ) + publisher = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=_enabled_handler(), + storage=store, + ) + + publisher._restore_schedule() + + assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2) + + +@pytest.mark.parametrize( + "state", + [ + None, # fresh install + {}, + {"last_success_at": None}, + {"last_success_at": "nonsense"}, + {"last_success_at": 0}, + # Clock moved backwards, or a junk row: must not park the schedule in the + # far future where nothing would ever publish again. + {"last_success_at": time.time() + 5 * 86400}, + ], +) +def test_restore_falls_back_to_the_grace_delay_on_unusable_state(state): + store = _FakeStateStore({STATE_KEY: state} if state is not None else {}) + publisher = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=_enabled_handler(), + storage=store, + ) + + publisher._restore_schedule() + + assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2) + + +def test_restore_clamps_a_delay_longer_than_the_interval(): + """An interval shortened since the last publish must take effect now.""" + store = _FakeStateStore({STATE_KEY: {"last_success_at": time.time()}}) + publisher = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 12}}}, + handler=_enabled_handler(), + storage=store, + ) + publisher._restore_schedule() + assert publisher.status()["secs_until_next"] <= 12 * 3600 + + # Same stored publish time, but the config now says 24h -> still bounded. + publisher.config["mqtt_brokers"]["neighbors"]["interval_hours"] = 24 + publisher._restore_schedule() + assert publisher.status()["secs_until_next"] <= 24 * 3600 + + +def test_restore_without_a_state_capable_storage_backend(): + """An older storage backend must still start, just without resuming.""" + publisher = _publisher({"mqtt_brokers": {}}, handler=_enabled_handler(), storage=object()) + + publisher._restore_schedule() + + assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2) + + +@pytest.mark.asyncio +async def test_failed_publish_is_not_recorded_as_a_successful_one(): + """Otherwise a restart turns the 15-minute retry into a full interval.""" + store = _FakeStateStore() + handler = SimpleNamespace( + has_neighbors_brokers=lambda: True, + has_connected_neighbors_brokers=lambda: True, + publish_neighbors=lambda payload: [], # reached no broker + node_name="n", + public_key="AB" * 32, + ) + store.get_neighbors = lambda: {} + publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store) + + await publisher.run_cycle(trigger="test") + + saved = store.rows[STATE_KEY] + assert saved["last_success_at"] is None + assert saved["last_publish_at"] is not None # the attempt is still recorded + assert "publish failed" in saved["last_result"] + + # A restart therefore retries promptly rather than waiting out the interval. + restarted = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=_enabled_handler(), + storage=store, + ) + restarted._restore_schedule() + assert restarted.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2) + + +@pytest.mark.asyncio +async def test_successful_publish_persists_a_resumable_schedule(): + store = _FakeStateStore() + 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, + ) + store.get_neighbors = lambda: {} + publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store) + + await publisher.run_cycle(trigger="test") + + assert store.rows[STATE_KEY]["last_success_at"] == pytest.approx(time.time(), abs=5) + + restarted = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=_enabled_handler(), + storage=store, + ) + restarted._restore_schedule() + assert restarted.status()["secs_until_next"] > 23 * 3600 + + +@pytest.mark.asyncio +async def test_boot_tick_does_not_discard_the_restored_schedule(): + """The disabled branch of _tick must not fire before we were ever enabled. + + At boot the MQTT connections may not be up, so enabled() can briefly be + False. Clearing the schedule there would mark the node due and re-run the + sweep on every restart -- the exact thing persistence prevents. + """ + store = _FakeStateStore({STATE_KEY: {"last_success_at": time.time() - 3600}}) + not_ready = SimpleNamespace(has_neighbors_brokers=lambda: False) + publisher = _publisher( + {"mqtt_brokers": {"neighbors": {"interval_hours": 24}}}, + handler=not_ready, + storage=store, + ) + publisher._restore_schedule() + restored = publisher._next_publish_at + + await publisher._tick() + + assert publisher._next_publish_at == restored + + +@pytest.mark.asyncio +async def test_disabling_after_being_enabled_still_clears_the_schedule(): + """Re-enabling should publish promptly; that behaviour is preserved.""" + store = _FakeStateStore() + enabled = True + handler = SimpleNamespace( + has_neighbors_brokers=lambda: enabled, + has_connected_neighbors_brokers=lambda: False, + ) + publisher = _publisher({"mqtt_brokers": {}}, handler=handler, storage=store) + publisher._next_publish_at = time.monotonic() + 3600 + + await publisher._tick() # enabled, but no connected broker -> schedule kept + assert publisher._next_publish_at is not None + + enabled = False + await publisher._tick() + assert publisher._next_publish_at is None + + +def test_daemon_state_round_trips_through_real_sqlite(tmp_path): + """The fake store above cannot prove the migration or the accessors work.""" + from repeater.data_acquisition.sqlite_handler import SQLiteHandler + + handler = SQLiteHandler(tmp_path) + + assert handler.get_daemon_state(STATE_KEY) is None # absent, not an error + + assert handler.set_daemon_state(STATE_KEY, {"last_success_at": 1785372000.0}) is True + assert handler.get_daemon_state(STATE_KEY) == {"last_success_at": 1785372000.0} + + # Upsert, not a second row. + assert handler.set_daemon_state(STATE_KEY, {"last_success_at": 1785458400.0}) is True + assert handler.get_daemon_state(STATE_KEY)["last_success_at"] == 1785458400.0 + + # A fresh handler on the same file sees it -- this is the restart path. + assert SQLiteHandler(tmp_path).get_daemon_state(STATE_KEY)["last_success_at"] == 1785458400.0 + + +def test_daemon_state_survives_a_corrupt_row(tmp_path): + """Unparseable state must read as "no history", never break startup.""" + import sqlite3 + + from repeater.data_acquisition.sqlite_handler import SQLiteHandler + + handler = SQLiteHandler(tmp_path) + handler.set_daemon_state(STATE_KEY, {"last_success_at": 1.0}) + with sqlite3.connect(handler.sqlite_path) as conn: + conn.execute("UPDATE daemon_state SET value_json = ?", ("{not json",)) + conn.commit() + + assert handler.get_daemon_state(STATE_KEY) is None + + publisher = _publisher({"mqtt_brokers": {}}, handler=_enabled_handler(), storage=handler) + publisher._restore_schedule() + assert publisher.status()["secs_until_next"] == pytest.approx(STARTUP_GRACE_SECONDS, abs=2) + + +def test_migration_is_idempotent_on_an_existing_database(tmp_path): + """Migration 14 runs against nibbler's populated DB, not a fresh file.""" + from repeater.data_acquisition.sqlite_handler import SQLiteHandler + + first = SQLiteHandler(tmp_path) + first.set_daemon_state(STATE_KEY, {"last_success_at": 42.0}) + + # Re-running migrations must not drop the table or its contents. + for _ in range(3): + SQLiteHandler(tmp_path)._run_migrations() + + assert SQLiteHandler(tmp_path).get_daemon_state(STATE_KEY) == {"last_success_at": 42.0}