From ddae287ee6722c9b214841dab631ebe3d5db74f1 Mon Sep 17 00:00:00 2001 From: Afri <220195275+l5yth@users.noreply.github.com> Date: Thu, 8 Oct 2026 15:55:12 +0200 Subject: [PATCH] data: keep snapshot node ages when the radio clock is off (#1007) * data: keep snapshot node ages when the radio clock is off * data: add the snapshot clock unit tests left out of the first commit --- CHANGELOG.md | 1 + data/mesh_ingestor/CONTRACTS.md | 4 +- data/mesh_ingestor/daemon.py | 20 +- data/mesh_ingestor/handlers/receive_time.py | 32 +- data/mesh_ingestor/protocols/meshtastic.py | 348 +++++++- tests/conftest.py | 19 + tests/test_channel_scope_unit.py | 26 +- tests/test_receive_time_unit.py | 12 +- tests/test_snapshot_clock_unit.py | 886 ++++++++++++++++++++ 9 files changed, 1327 insertions(+), 21 deletions(-) create mode 100644 tests/test_snapshot_clock_unit.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f7bc00a2..40ded3a6 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -81,6 +81,7 @@ * Data: `PROTOCOL=reticulum` needs `INGESTOR_NODE_ID` in Docker, when nothing on the ingestor's RNS stack announces, or when two local identities tie; otherwise the node id is derived once a local app announces (SPEC RE5/RE8) * Data: a set `CONNECTION` is no longer silently ignored under `PROTOCOL=reticulum` - the ingestor now logs that it is dropping the value (SPEC RN10) * Data/Web: Meshtastic records whose radio clock is off by more than an hour get the ingestor's receive time, so they show in charts and tables again; the node page shows a battery above 100% as the nodes table does (SPEC RK1-RK3, UX10) +* Data: when a Meshtastic radio's clock is off by more than an hour, the node list it sends at connect is posted with each `lastHeard` shifted to the ingestor's clock; entries stamped more than an hour before the radio's current boot are posted as stamped when the radio reports its uptime, entries stamped ahead of its clock are skipped, and the ingestor logs one warning. Without a clock reading, an entry more than an hour ahead of the ingestor's clock is skipped (SPEC RK4) * Data: `RETICULUM_INTERFACES` now resolves interface names through the shared RNS instance, fixing an allowlist that matched nothing on a shared stack (SPEC RE3) * Data: a Reticulum announce with no display name no longer overwrites a stored name * Docs: document the Reticulum config directory, its Docker volume, and how to add interfaces to it diff --git a/data/mesh_ingestor/CONTRACTS.md b/data/mesh_ingestor/CONTRACTS.md index 6769769e..879d3725 100644 --- a/data/mesh_ingestor/CONTRACTS.md +++ b/data/mesh_ingestor/CONTRACTS.md @@ -113,13 +113,13 @@ Deployment ordering. The web whitelist must accept a protocol before any ingesto - An index whose name was never captured matches no `ALLOWED_CHANNELS` or `HIDDEN_CHANNELS` entry, except channel 0, which is matched by `PRIMARY_CHANNEL_NAME` (the passive UDP transport captures no channel names). That name is used for matching only; message payloads keep their `channel_name` rules. - `DROP_VIA_MQTT` drops a packet carrying `viaMqtt: true` or `via_mqtt: true` (the latter from a proto-field-name conversion) and a nodeDB entry carrying `viaMqtt: true`; presence means true. The UDP transport maps the same `viaMqtt` key. - `POST /api/nodes` bodies never carry the meshtastic library's `lastReceived` copy of a node's last packet. -- Known limit: a snapshot entry is published when its NodeInfo channel passes, and it carries the nodeDB's latest position, metrics and signal details (`lastHeard`, `snr`, `hopLimit`), which the radio stores whatever channel they were heard on. +- Known limit: a snapshot entry is published when its NodeInfo channel passes, and it carries the nodeDB's latest position, metrics and signal details (`lastHeard`, `snr`, `hopLimit`), which the radio stores whatever channel they were heard on. Its `lastHeard` follows the snapshot rule under "Receive time" below, which can also skip the entry (SPEC RK4). ### Ingest HTTP routes and payload shapes Future providers should emit payloads that match these shapes (keys + types), which are validated by existing tests (notably `tests/test_mesh.py`). -Receive time (SPEC RK1-RK3). `rx_time` is the ingestor's receive time, in Unix seconds. A Meshtastic packet's `rxTime` comes from a radio's clock, not the ingestor's. `handlers.on_receive`, which both Meshtastic transports feed, keeps it only within 1 hour of the ingestor's clock, in either direction (`RX_TIME_TOLERANCE_SECS` in `handlers/receive_time.py`); otherwise it posts the ingestor's clock and logs a warning that names the sender and the offset, at most once per 10 minutes. MeshCore and Reticulum records keep the times their providers assign: a MeshCore roster position carries the contact's `last_advert` (SPEC RS1). The node-list snapshot's `lastHeard` comes from the radio's node database and is not checked (SPEC RK3). The web app stores `rx_time` as posted and clamps only future values, and the GET time windows below filter messages, positions and telemetry on it, so a record stamped before the window is stored but never served. +Receive time (SPEC RK1-RK4). `rx_time` is the ingestor's receive time, in Unix seconds. A Meshtastic packet's `rxTime` comes from a radio's clock, not the ingestor's. `handlers.on_receive`, which both Meshtastic transports feed, keeps it only within 1 hour of the ingestor's clock, in either direction (`RX_TIME_TOLERANCE_SECS` in `handlers/receive_time.py`); otherwise it posts the ingestor's clock and logs a warning that names the sender and the offset, at most once per 10 minutes. MeshCore and Reticulum records keep the times their providers assign: a MeshCore roster position carries the contact's `last_advert` (SPEC RS1). The node-list snapshot's `lastHeard` comes from the radio's node database, stamped by the radio's clock. The serial, TCP and BLE transports read that clock once per snapshot from the `lastHeard` of the radio's own entry (SPEC RK4). When the reading is more than 1 hour off the ingestor's clock, in either direction, each entry is posted with its `lastHeard` shifted by that offset, so it keeps its age by the radio's clock; an entry stamped more than 1 hour after the reading, or one the shift would post at 0 or below, came from an earlier boot and is skipped (`clock-epoch`). When the same entry gives the radio's uptime (`deviceMetrics.uptimeSeconds`), an entry stamped more than 1 hour before this boot started came from an earlier boot too: it is posted as stamped, or skipped when it is more than 1 hour ahead of the ingestor's clock. One warning per snapshot gives the offset and the counts. Within 1 hour, entries are posted as stamped. Without a reading, entries are posted as stamped, except an entry more than 1 hour ahead of the ingestor's clock, which is skipped. An entry without a positive `lastHeard` is always posted as it is. The web app stores `rx_time` as posted and clamps only future values, and the GET time windows below filter messages, positions and telemetry on it, so a record stamped before the window is stored but never served. #### `POST /api/nodes` diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 54f1ac2d..4411df74 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -413,7 +413,12 @@ def _try_send_snapshot(state: _DaemonState) -> bool: ``snapshot_filter_reason(node_id, node)`` hook (SPEC CF3): an entry it names a reason for is skipped, not upserted. Skipped entries still count as processed, so a nodeDB whose every entry is filtered still latches - ``initial_snapshot_sent``. + ``initial_snapshot_sent``. Each entry it lets through passes the + optional ``snapshot_entry(node_id, node)`` hook, inside the same per-node + error handling, and what that returns is upserted: the Meshtastic + provider shifts ``lastHeard`` off a wrong radio clock there (SPEC RK4). + A provider whose ``snapshot_clock_warning`` then names fields gets one + ``warn`` line per snapshot with them. Returns: ``True`` when the snapshot succeeded (or no nodes exist yet); ``False`` @@ -423,6 +428,7 @@ def _try_send_snapshot(state: _DaemonState) -> bool: try: node_items = state.provider.node_snapshot_items(state.iface) filter_reason = getattr(state.provider, "snapshot_filter_reason", None) + snapshot_entry = getattr(state.provider, "snapshot_entry", None) processed_any = False for node_id, node in node_items: processed_any = True @@ -438,6 +444,8 @@ def _try_send_snapshot(state: _DaemonState) -> bool: reason=reason, ) continue + if callable(snapshot_entry): + node = snapshot_entry(node_id, node) handlers.upsert_node(node_id, node) except Exception as exc: config._debug_log( @@ -456,6 +464,16 @@ def _try_send_snapshot(state: _DaemonState) -> bool: ) if processed_any: state.initial_snapshot_sent = True + # Read once every entry is handled, so the counts are this snapshot's. + clock_warning = getattr(state.provider, "snapshot_clock_warning", None) + if isinstance(clock_warning, dict): + config._debug_log( + "Radio clock is off; posting snapshot lastHeard shifted to the " + "ingestor clock", + context="daemon.snapshot", + severity="warn", + **clock_warning, + ) return True except Exception as exc: config._debug_log( diff --git a/data/mesh_ingestor/handlers/receive_time.py b/data/mesh_ingestor/handlers/receive_time.py index b5d0df67..d6df55b8 100644 --- a/data/mesh_ingestor/handlers/receive_time.py +++ b/data/mesh_ingestor/handlers/receive_time.py @@ -12,7 +12,7 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Receive-time check for Meshtastic packets (SPEC RK1-RK3). +"""Receive-time checks for Meshtastic packets and snapshots (SPEC RK1-RK4). A Meshtastic packet's ``rxTime`` comes from the clock of a radio, not from the ingestor host. A radio whose clock is wrong stamps every packet it hears @@ -154,8 +154,38 @@ def _warn_replaced(packet: Mapping, offset_secs: int) -> None: _unlogged_replacements = 0 +def radio_clock_offset(radio_now: object, host_now: int) -> int | None: + """Return how far a radio's clock reading lies off the host clock (SPEC RK4). + + The Meshtastic node-list snapshot posts each nodeDB entry's ``lastHeard``, + which the radio stamped with its own clock. The radio stamps its own + entry with that clock's reading while it sends the nodeDB at connect, so + the reading tells how far every entry's time is off. A reading within + :data:`RX_TIME_TOLERANCE_SECS` of the host clock, the bound included, + counts as the host clock, as a packet's receive time does (SPEC RK2). + + Parameters: + radio_now: The radio's clock reading in Unix seconds, as its nodeDB + holds it. Anything that does not parse as a positive integer is + no reading. + host_now: The host clock in whole Unix seconds. + + Returns: + ``None`` when ``radio_now`` is no clock reading; ``0`` when it lies + within the tolerance; otherwise ``radio_now - host_now``, negative for + a radio clock that runs behind. + """ + + radio_time = _coerce_int(radio_now) + if radio_time is None or radio_time <= 0: + return None + offset = radio_time - host_now + return 0 if abs(offset) <= RX_TIME_TOLERANCE_SECS else offset + + __all__ = [ "RX_TIME_TOLERANCE_SECS", "RX_TIME_WARNING_INTERVAL_SECS", + "radio_clock_offset", "replace_skewed_rx_time", ] diff --git a/data/mesh_ingestor/protocols/meshtastic.py b/data/mesh_ingestor/protocols/meshtastic.py index d1605809..dfcdf41f 100644 --- a/data/mesh_ingestor/protocols/meshtastic.py +++ b/data/mesh_ingestor/protocols/meshtastic.py @@ -16,6 +16,7 @@ from __future__ import annotations +import time from collections.abc import Mapping from pubsub import pub @@ -29,6 +30,7 @@ from .. import ( interfaces, tx_policy, ) +from ..handlers.receive_time import RX_TIME_TOLERANCE_SECS, radio_clock_offset from ..node_identity import canonical_node_id from ..serialization import _coerce_int from ..utils import _retry_dict_snapshot @@ -41,6 +43,7 @@ class MeshtasticProvider: def __init__(self): self._subscribed: list[str] = [] + self._keep_snapshot_clock(0, None) def subscribe(self) -> list[str]: """Subscribe Meshtastic pubsub receive topics.""" @@ -88,7 +91,10 @@ class MeshtasticProvider: Uses :func:`~data.mesh_ingestor.utils._retry_dict_snapshot` to tolerate concurrent modifications from the Meshtastic background - thread. + thread. The entries are the library's own dicts, not copies (SPEC + CF3). It also reads the radio's clock and uptime for this snapshot + from the radio's own entry, which :meth:`snapshot_filter_reason` and + :meth:`snapshot_entry` apply to each entry (SPEC RK4). Parameters: iface: Live Meshtastic interface whose ``nodes`` dict to snapshot. @@ -101,13 +107,112 @@ class MeshtasticProvider: nodes = getattr(iface, "nodes", {}) or {} result = _retry_dict_snapshot(lambda: list(nodes.items())) if result is None: + # No entries to apply a clock to; keep none from a past snapshot. + self._keep_snapshot_clock(0, None) config._debug_log( "Skipping node snapshot due to concurrent modification", context="meshtastic.snapshot", ) return [] + self._read_snapshot_clock(iface, nodes) return result + def _read_snapshot_clock(self, iface: object, nodes: object) -> None: + """Keep this snapshot's clock correction, read from the own entry (SPEC RK4). + + The radio's own entry (:func:`_own_entry`) carries its clock's + reading at connect (see + :func:`~data.mesh_ingestor.handlers.receive_time.radio_clock_offset`) + and its uptime (:func:`_uptime_secs`). Three cases: + + - A reading more than :data:`RX_TIME_TOLERANCE_SECS` off the host + clock: every entry is posted with ``lastHeard`` minus that offset. + Another clock, an earlier boot's, stamped an entry more than the + tolerance after the reading, or one the shift would post at ``0`` + or below; it is skipped. With an uptime, an entry stamped more + than the tolerance before this boot started, by the radio's clock, + is an earlier boot's too: it is posted as stamped, unless it lies + more than the tolerance ahead of the host clock, which skips it. + - A reading within the tolerance: every entry is posted as stamped. + - No reading: entries are posted as stamped, except one stamped more + than the tolerance ahead of the host clock: a fast clock stamped + it, and the web app would store it as heard now. + + In every case an entry without a positive ``lastHeard`` is posted as + it is. Only the first case uses the uptime. + + Parameters: + iface: Live Meshtastic interface. + nodes: Its ``nodes`` object, already read for the snapshot. + """ + + host_now = int(time.time()) + own = _own_entry(iface, nodes) + offset = radio_clock_offset( + None if own is None else own.get("lastHeard"), host_now + ) + if offset is None: + self._keep_snapshot_clock(0, host_now + RX_TIME_TOLERANCE_SECS) + elif offset == 0: + self._keep_snapshot_clock(0, None) + else: + # host_now + offset is the radio's reading, which came from + # ``own``, so ``own`` is a mapping here. + radio_now = host_now + offset + uptime = _uptime_secs(own) + # This boot started ``uptime`` seconds before the reading, by the + # radio's clock. A previous boot's uptime, left in the own entry + # until this boot's first device telemetry, is mostly larger: it + # moves the boot start earlier, so fewer entries count as an + # earlier boot's, down to the rule without an uptime. An uptime + # this boot refreshed lags by up to one telemetry interval and + # moves the boot start later by as much; the tolerance absorbs a + # lag of up to 1 hour. + self._keep_snapshot_clock( + offset, + radio_now + RX_TIME_TOLERANCE_SECS, + boot_floor=( + None + if uptime is None + else radio_now - uptime - RX_TIME_TOLERANCE_SECS + ), + host_newest=host_now + RX_TIME_TOLERANCE_SECS, + ) + + def _keep_snapshot_clock( + self, + offset: int, + newest: int | None, + *, + boot_floor: int | None = None, + host_newest: int | None = None, + ) -> None: + """Keep the clock correction of the snapshot that starts (SPEC RK4). + + The counts of shifted, skipped and unshifted entries restart at zero. + + Parameters: + offset: Radio clock minus host clock, subtracted from each posted + ``lastHeard``; ``0`` shifts nothing. + newest: Newest ``lastHeard`` the snapshot posts; ``None`` posts + any. + boot_floor: Oldest ``lastHeard`` this boot's clock can have + stamped, by the radio's uptime; an older one is an earlier + boot's and is posted as stamped. ``None``, without an + uptime, counts every entry as this boot's. + host_newest: Newest earlier-boot ``lastHeard`` the snapshot + posts, the host clock plus the tolerance; read only with a + ``boot_floor``. + """ + + self._clock_offset = offset + self._clock_newest = newest + self._clock_boot_floor = boot_floor + self._clock_host_newest = host_newest + self._clock_shifted = 0 + self._clock_skipped = 0 + self._clock_unshifted = 0 + def snapshot_filter_reason(self, node_id: str, node: object) -> str | None: """Return why the node-list snapshot must not publish ``node`` (SPEC CF3, NI1). @@ -129,6 +234,16 @@ class MeshtasticProvider: metrics, which the radio stores whatever channel they were heard on (a documented limit, SPEC CF3). + An entry whose ``lastHeard`` is newer than this snapshot's clock + allows (see :meth:`node_snapshot_items`), or one the shift of + :meth:`snapshot_entry` would post at ``0`` or below, was stamped by + another clock, an earlier boot's, and no shift of this snapshot fits + it (SPEC RK4). So is one stamped before this boot started, which is + posted as stamped, and skipped only when it lies more than the + tolerance ahead of the host clock. The channel reasons come first, + so the snapshot's radio-clock warning counts only the entries + skipped for their time. + Parameters: node_id: Id the entry is filed under in ``iface.nodes``. node: The nodeDB entry. @@ -136,18 +251,142 @@ class MeshtasticProvider: Returns: ``"num-mismatch"`` for an entry filed under another node's id, else the :func:`~data.mesh_ingestor.channels.ingest_filter_reason` - verdict; ``None`` for an entry that is not a mapping. + verdict, else ``"clock-epoch"`` for an entry stamped by another + clock than the snapshot's; ``None`` for an entry that is not a + mapping. """ if not isinstance(node, Mapping): return None - num = _coerce_int(node.get("num")) - if num is not None and canonical_node_id(num) != canonical_node_id(node_id): + if _filed_under_another_id(node_id, node): return "num-mismatch" - return channels.ingest_filter_reason( + reason = channels.ingest_filter_reason( _coerce_int(node.get("channel")) or 0, via_mqtt=bool(node.get("viaMqtt")), ) + if reason is None and self._clock_newest is not None: + last_heard = _stamped_last_heard(node) + if last_heard is not None and self._clock_epoch(last_heard): + self._clock_skipped += 1 + return "clock-epoch" + return reason + + def _clock_epoch(self, last_heard: int) -> bool: + """Return whether this snapshot skips an entry for its time (SPEC RK4). + + Read only while the snapshot keeps a newest time (see + :meth:`_keep_snapshot_clock`). + + Parameters: + last_heard: The entry's positive ``lastHeard``. + + Returns: + ``True`` for a stamp later than the newest the snapshot posts, an + earlier boot's stamp more than the tolerance ahead of the host + clock, or a stamp the shift would post at ``0`` or below, 1970 or + before by the host clock. + """ + + if last_heard > self._clock_newest: + return True + if self._earlier_boot(last_heard): + # Posted as stamped, so the guard without a reading applies. + return last_heard > self._clock_host_newest + # With no offset kept, a positive time is never shifted to 0. + return last_heard - self._clock_offset <= 0 + + def _earlier_boot(self, last_heard: int) -> bool: + """Return whether ``last_heard`` was stamped before this boot (SPEC RK4). + + Parameters: + last_heard: The entry's positive ``lastHeard``. + + Returns: + ``True`` for a stamp older than the snapshot's boot floor, more + than the tolerance before the boot started, which only a snapshot + that read the radio's uptime keeps. + """ + + return ( + self._clock_boot_floor is not None and last_heard < self._clock_boot_floor + ) + + def snapshot_entry(self, node_id: str, node: object) -> object: + """Return ``node`` as the node-list snapshot posts it (SPEC RK4). + + An **optional**, duck-typed provider hook like + :meth:`snapshot_filter_reason`: + :func:`~data.mesh_ingestor.daemon._try_send_snapshot` passes it each + entry the filter lets through, inside its per-node error handling, + and upserts what it returns. + + A radio whose clock reading at connect lay more than + :data:`~data.mesh_ingestor.handlers.receive_time.RX_TIME_TOLERANCE_SECS` + off the host clock stamped every ``lastHeard`` in its nodeDB by that + clock. The entry is posted with ``lastHeard`` minus the offset, so it + keeps the age the radio's clock gives it, counted from the host + clock: a node the radio heard 10 minutes ago is posted 10 minutes + old. The shift is made on a shallow copy, so the library's dict + keeps the radio's value, and the copy is made inside the per-node + handling, so a dict the library mutates meanwhile fails that entry + alone (SPEC CF3). An entry the shift would post at ``0`` or below + never gets here: :meth:`snapshot_filter_reason` skips it. An entry + stamped before this boot started, by the radio's uptime, was stamped + by an earlier boot's clock and is posted as stamped. + + Parameters: + node_id: Id the entry is filed under in ``iface.nodes``. + node: The nodeDB entry. + + Returns: + A shallow copy of ``node`` with the shifted ``lastHeard``; + ``node`` itself when nothing is shifted: no offset kept, an entry + that is not a mapping, one without a positive ``lastHeard``, or + an earlier boot's, which counts as ``unshifted``. + """ + + if not self._clock_offset or not isinstance(node, Mapping): + return node + last_heard = _stamped_last_heard(node) + if last_heard is None: + return node + if self._earlier_boot(last_heard): + self._clock_unshifted += 1 + return node + entry = dict(node) + entry["lastHeard"] = last_heard - self._clock_offset + self._clock_shifted += 1 + return entry + + @property + def snapshot_clock_warning(self) -> dict[str, int] | None: + """Fields of this snapshot's radio-clock warning (SPEC RK4). + + Read by :func:`~data.mesh_ingestor.daemon._try_send_snapshot` once it + has handled every entry, so a snapshot logs one warning however many + entries it shifts. + + Returns: + ``None`` unless this snapshot shifts its entries for a radio + clock past the tolerance; then ``offset_secs`` (radio clock minus + host clock), ``shifted`` (entries :meth:`snapshot_entry` shifted) + and ``skipped`` (entries skipped as ``clock-epoch``), plus + ``unshifted`` (earlier-boot entries posted as stamped) when the + snapshot read the radio's uptime. + """ + + if not self._clock_offset: + return None + fields = { + "offset_secs": self._clock_offset, + "shifted": self._clock_shifted, + "skipped": self._clock_skipped, + } + # Without an uptime no entry can count as unshifted, so the field + # also tells whether the snapshot read one. + if self._clock_boot_floor is not None: + fields["unshifted"] = self._clock_unshifted + return fields def send_channel_announcement(self, iface: object, text: str) -> None: """Broadcast an activity announcement on the default channel (SPEC MA6/MA9). @@ -184,4 +423,103 @@ class MeshtasticProvider: ) +def _filed_under_another_id(node_id: str, node: Mapping) -> bool: + """Return whether ``node`` is another node's entry filed under ``node_id`` (SPEC NI1). + + The meshtastic library files a node under the ``user.id`` of its last + NodeInfo, so one naming another node files the sender's entry under the + other node's id. An entry without a ``num`` is not checked. + + Parameters: + node_id: Id the entry is filed under in ``iface.nodes``. + node: The nodeDB entry. + + Returns: + ``True`` when the entry's ``num`` maps to an id other than + ``node_id``. + """ + + num = _coerce_int(node.get("num")) + return num is not None and canonical_node_id(num) != canonical_node_id(node_id) + + +def _stamped_last_heard(node: Mapping) -> int | None: + """Return the time a radio's clock stamped on ``node`` (SPEC RK4). + + Parameters: + node: The nodeDB entry. + + Returns: + ``lastHeard`` as an integer, or ``None`` when it does not parse as a + positive one: the entry then counts as one without ``lastHeard``, + as the own entry does for the clock reading + (:func:`~data.mesh_ingestor.handlers.receive_time.radio_clock_offset`). + """ + + last_heard = _coerce_int(node.get("lastHeard")) + return last_heard if last_heard is not None and last_heard > 0 else None + + +def _own_entry(iface: object, nodes: object) -> Mapping | None: + """Return the radio's own nodeDB entry (SPEC RK4). + + The entry is looked up once, under the host id ``daemon._try_connect`` + registers when its ``num`` maps to that id (the ``num-mismatch`` check of + :meth:`MeshtasticProvider.snapshot_filter_reason`), else through the + meshtastic library's ``getMyNodeInfo()``, which finds it by the radio's + node number. Another node's entry filed under the host id (SPEC NI1) + carries that node's time, which read as the radio's clock would shift a + correct clock's entries by the node's age; it is never returned. + Neither lookup copies the nodeDB. + + Parameters: + iface: Live Meshtastic interface. + nodes: Its ``nodes`` object, already read for the snapshot. + + Returns: + The own entry, whose ``lastHeard`` is the radio's clock reading and + whose ``deviceMetrics`` carries its uptime, or ``None`` when none is + found. A ``getMyNodeInfo`` that raises counts as no entry: the + library formats the whole nodeDB into a debug line first, which a + concurrent update can break, and the snapshot must not fail for it. + """ + + host_id = handlers.host_node_id() + if host_id is not None and isinstance(nodes, Mapping): + own = nodes.get(host_id) + if isinstance(own, Mapping) and not _filed_under_another_id(host_id, own): + return own + get_my_node_info = getattr(iface, "getMyNodeInfo", None) + if not callable(get_my_node_info): + return None + try: + own = get_my_node_info() + except Exception: + return None + return own if isinstance(own, Mapping) else None + + +def _uptime_secs(own: Mapping) -> int | None: + """Return the radio's uptime from its own nodeDB entry (SPEC RK4). + + The firmware refreshes ``deviceMetrics.uptimeSeconds`` of its own entry + whenever it sends device telemetry (``DeviceTelemetry`` ``sendTelemetry`` + calls ``nodeDB->updateTelemetry`` for its own node number). + + Parameters: + own: The radio's own entry (:func:`_own_entry`). + + Returns: + The uptime in seconds, or ``None`` unless it is a positive ``int``: + a ``bool`` (an ``int`` subclass), a float, a string or a value of + ``0`` or below is no uptime. + """ + + metrics = own.get("deviceMetrics") + uptime = metrics.get("uptimeSeconds") if isinstance(metrics, Mapping) else None + if isinstance(uptime, bool) or not isinstance(uptime, int) or uptime <= 0: + return None + return uptime + + __all__ = ["MeshtasticProvider"] diff --git a/tests/conftest.py b/tests/conftest.py index 15b697f6..f08c23ef 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -17,7 +17,9 @@ from __future__ import annotations import importlib import sys +import time from pathlib import Path +from types import SimpleNamespace import pytest @@ -66,6 +68,23 @@ def permit_tx(monkeypatch): return config +@pytest.fixture +def host_clock(monkeypatch, host_now): + """Drive the ingestor host clock by hand, starting at ``host_now``. + + ``host_now`` is a fixture the requesting test module defines, returning + its start time in whole Unix seconds (the receive-time checks, SPEC + RK1-RK4). The clock stays there until a test moves it. + + Returns: + A namespace whose ``now`` attribute :func:`time.time` returns. + """ + + clock = SimpleNamespace(now=host_now) + monkeypatch.setattr(time, "time", lambda: float(clock.now)) + return clock + + @pytest.fixture def load_config(monkeypatch): """Reload :mod:`~data.mesh_ingestor.config` under the given environment. diff --git a/tests/test_channel_scope_unit.py b/tests/test_channel_scope_unit.py index d6b8324a..5ee3cf5b 100644 --- a/tests/test_channel_scope_unit.py +++ b/tests/test_channel_scope_unit.py @@ -153,15 +153,35 @@ def _mesh_packet(port: str, *, channel: int, via_mqtt: bool = False): return packet -def _node_info_frame(num: int, *, channel: int = 0, via_mqtt: bool = False) -> bytes: - """Serialized ``FromRadio.node_info`` as the radio sends its nodeDB.""" +def _node_info_frame( + num: int, + *, + channel: int = 0, + via_mqtt: bool = False, + last_heard: int = 0, + user_id: str | None = None, + uptime_seconds: int = 0, +) -> bytes: + """Serialized ``FromRadio.node_info`` as the radio sends its nodeDB. + + ``last_heard`` is the radio clock's reading when it last heard the node + (SPEC RK4); proto3 omits it at ``0``, as it omits ``channel``. + ``user_id`` is the ``user.id`` the entry names, the node's own id by + default; the library files the entry under it (SPEC NI1). + ``uptime_seconds`` goes into the entry's device metrics, as the radio's + own entry carries its uptime (SPEC RK4); at ``0`` the entry has none. + """ info = mesh_pb2.NodeInfo( num=num, - user=mesh_pb2.User(id=f"!{num:08x}", long_name=f"Node {num:x}"), + user=mesh_pb2.User(id=user_id or f"!{num:08x}", long_name=f"Node {num:x}"), channel=channel, via_mqtt=via_mqtt, + last_heard=last_heard, ) + if uptime_seconds: + # Only a set submessage is sent, so other entries keep no metrics. + info.device_metrics.uptime_seconds = uptime_seconds return mesh_pb2.FromRadio(node_info=info).SerializeToString() diff --git a/tests/test_receive_time_unit.py b/tests/test_receive_time_unit.py index 348eb78b..3b8523f9 100644 --- a/tests/test_receive_time_unit.py +++ b/tests/test_receive_time_unit.py @@ -137,16 +137,10 @@ def reset_warning_state(monkeypatch): @pytest.fixture -def host_clock(monkeypatch): - """Drive the ingestor host clock by hand, starting at :data:`NOW`. +def host_now(): + """Start the shared ``host_clock`` fixture (``tests/conftest.py``) at :data:`NOW`.""" - Returns: - A namespace whose ``now`` attribute :func:`time.time` returns. - """ - - clock = SimpleNamespace(now=NOW) - monkeypatch.setattr(time, "time", lambda: float(clock.now)) - return clock + return NOW @pytest.fixture diff --git a/tests/test_snapshot_clock_unit.py b/tests/test_snapshot_clock_unit.py new file mode 100644 index 00000000..c7dafe4f --- /dev/null +++ b/tests/test_snapshot_clock_unit.py @@ -0,0 +1,886 @@ +# Copyright © 2025-26 l5yth & contributors +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. +"""Regression guard: the Meshtastic node-list snapshot posts ``lastHeard`` +inside the web app's read windows when the radio's clock is wrong (SPEC RK4). + +The snapshot posts each nodeDB entry's ``lastHeard``, stamped by the radio's +clock. The field host's radio (meshtasticd, no RTC) ran 97 days slow, so a +node it heard 10 minutes ago was posted 97 days old: outside the 7-day +``GET /api/nodes`` window and the 28-day per-id window, and stored as the +node's ``first_heard`` for good. The radio stamps its own entry with its +clock's reading while it sends the nodeDB at connect, so the snapshot reads +the offset there and posts every entry with the age the radio's clock gives +it, counted from the host clock. + +The nodeDB is loaded through the pinned meshtastic library's +``_handleFromRadio`` the way a radio sends it, and +``daemon._try_send_snapshot`` runs with the real :class:`MeshtasticProvider`. +The host clock is frozen; only the HTTP queue is replaced. The cases after +the regression guard pin the rule: the 1-hour tolerance in both directions, +the skip of entries from an earlier boot, the snapshot without a clock +reading, another node's entry under the host id, a ``lastHeard`` at or +shifted to 0 or below, the radio's uptime telling an earlier boot's entries +apart, one warning per snapshot, and the library's dicts left as they were. +""" + +from __future__ import annotations + +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +REPO_ROOT = Path(__file__).resolve().parents[1] +if str(REPO_ROOT) not in sys.path: # pragma: no cover - conftest adds it + sys.path.insert(0, str(REPO_ROOT)) + +from meshtastic.protobuf import mesh_pb2 # noqa: E402 + +import data.mesh_ingestor.config as config # noqa: E402 +import data.mesh_ingestor.daemon as daemon # noqa: E402 +import data.mesh_ingestor.handlers as handlers # noqa: E402 +import data.mesh_ingestor.handlers.receive_time as receive_time # noqa: E402 +from daemon_fakes import make_state # noqa: E402 - shared daemon doubles +from data.mesh_ingestor.protocols.meshtastic import MeshtasticProvider # noqa: E402 +from data.mesh_ingestor.serialization import _iso # noqa: E402 +from test_channel_scope_unit import ( # noqa: E402,F401 - shared fixtures + SECONDARY, + SENDER, + SENDER_ID, + _node_info_frame, + _snapshot, + radio, + scope, +) + +DAY = 86_400 +"""Seconds in a day.""" + +HOST_NOW = 1_791_460_800 +"""Ingestor host clock for every case: 2026-10-08T12:00:00Z.""" + +RADIO_OFFSET = -97 * DAY +"""The field radio's clock minus the host clock: 97 days slow (-8380800 s).""" + +RADIO_NOW = HOST_NOW + RADIO_OFFSET +"""The field radio's clock reading at connect: 2026-07-03T12:00:00Z.""" + +TOLERANCE = 60 * 60 +"""SPEC RK2's tolerance in seconds, which the snapshot reuses (RK4).""" + +HOST_NUM = 0x0D15EA5E +"""Node number of the ingestor's own radio.""" + +HOST_ID = "!0d15ea5e" +"""Canonical id of :data:`HOST_NUM`, as ``daemon._try_connect`` registers it.""" + +PREV_BOOT_NUM = 0x0B007B00 +"""Node last heard during the radio's previous boot.""" + +PREV_BOOT_ID = "!0b007b00" +"""Canonical id of :data:`PREV_BOOT_NUM`.""" + +IMPOSTOR_NUM = 0x0BADBEEF +"""Node whose NodeInfo names :data:`HOST_ID`, so the library files its entry +under the host id (SPEC NI1).""" + +AHEAD_NUM = 0x0FA57000 +"""Node a fast clock stamped 61 minutes ahead of the host clock.""" + +AHEAD_ID = "!0fa57000" +"""Canonical id of :data:`AHEAD_NUM`.""" + +FAST_OFFSET = HOST_NOW +"""A radio clock ahead of the host clock by as long as the host clock is past +1970: shifted by it, an entry stamped at the host clock would post at 0.""" + +FIELD_HOST_NOW = 1_791_455_465 +"""The field host's clock at its snapshot with the uptime read: +2026-10-08T10:31:05Z.""" + +FIELD_READING = 1_783_067_010 +"""The field radio's own ``lastHeard`` at that snapshot, its clock's reading.""" + +FIELD_OFFSET = FIELD_READING - FIELD_HOST_NOW +"""The field radio's clock minus the host clock: -8388455 s, 97.1 days slow.""" + +FIELD_UPTIME = 343_965 +"""The field radio's own ``deviceMetrics.uptimeSeconds``: up 3.98 days.""" + +FIELD_BOOT_START = FIELD_READING - FIELD_UPTIME +"""When the field radio's boot started, by its own clock: 1782723045.""" + +OLD_BOOT_NUM = 0x0B00701D +"""Node last heard during an earlier boot, before the field radio's boot.""" + +OLD_BOOT_ID = "!0b00701d" +"""Canonical id of :data:`OLD_BOOT_NUM`.""" + +OLD_BOOT_LAST_HEARD = FIELD_HOST_NOW - 120 * DAY +"""Heard 120 days ago by an earlier boot's right clock: 1781087465, more than +an hour before :data:`FIELD_BOOT_START`.""" + +REMOTE_LAST_HEARD = RADIO_NOW - 600 +""":data:`SENDER` heard 10 minutes ago, stamped by the field radio's clock.""" + +PREV_BOOT_LAST_HEARD = HOST_NOW - 5 * DAY - 4_294_967 +"""Heard 5 days ago during the previous boot, whose clock ran 2^32 ms behind. + +That is 42.3 days after the field radio's reading: no single shift fits both +boots, and shifted by this boot's offset it would lie 42 days in the future. +""" + +BULK_WINDOW = 7 * DAY +"""Floor of ``GET /api/nodes``: a node last heard earlier is not listed.""" + +FIELD_WARNING = { + "message": "Radio clock is off; posting snapshot lastHeard shifted to the " + "ingestor clock", + "severity": "warn", + "offset_secs": -8_380_800, + "shifted": 2, + "skipped": 1, +} +"""The one warning the field snapshot logs: own entry and :data:`SENDER` +shifted, the previous-boot entry skipped.""" + +FIELD_UPTIME_WARNING = { + **FIELD_WARNING, + "offset_secs": -8_388_455, + "unshifted": 1, +} +"""The one warning of the field host's snapshot with the uptime read: own +entry and :data:`SENDER` shifted, the previous-boot entry skipped, and the +entry from before the boot posted as stamped.""" + + +@pytest.fixture +def host_now(): + """Start the shared ``host_clock`` fixture (``tests/conftest.py``) at + :data:`HOST_NOW`, where every case keeps it.""" + + return HOST_NOW + + +@pytest.fixture +def snapshot_log(monkeypatch): + """Capture the daemon's snapshot log lines instead of printing them. + + Returns: + A list that collects each ``daemon.snapshot`` line's message, + severity and metadata. + """ + + lines: list[dict] = [] + + def record(message, *, context=None, severity="debug", **metadata): + """Keep the snapshot lines and drop every other log line.""" + + if context == "daemon.snapshot": + lines.append({"message": message, "severity": severity, **metadata}) + + monkeypatch.setattr(config, "_debug_log", record) + return lines + + +def _load_node_db( + radio, + own_last_heard: int | None, + entries: dict[int, int], + *, + register_host: bool = True, + own_uptime: int = 0, +) -> None: + """Send ``radio`` its nodeDB the way a radio sends it at connect. + + Parameters: + radio: The ``radio`` fixture's namespace. + own_last_heard: ``lastHeard`` of the radio's own entry, its clock's + reading. ``0`` sends the entry without one; ``None`` sends + neither ``my_info`` nor the entry. + entries: ``lastHeard`` of each other entry, by node number. + register_host: Register the host id afterwards, as + ``daemon._try_connect`` does right before the snapshot. + own_uptime: ``deviceMetrics.uptimeSeconds`` of the own entry, the + radio's uptime; ``0`` sends no device metrics. + """ + + iface = radio.iface + if own_last_heard is not None: + my_info = mesh_pb2.MyNodeInfo(my_node_num=HOST_NUM) + iface._handleFromRadio(mesh_pb2.FromRadio(my_info=my_info).SerializeToString()) + iface._handleFromRadio( + _node_info_frame( + HOST_NUM, last_heard=own_last_heard, uptime_seconds=own_uptime + ) + ) + for num, last_heard in entries.items(): + iface._handleFromRadio(_node_info_frame(num, last_heard=last_heard)) + if register_host: + handlers.register_host_node_id(HOST_ID) + + +def _load_field_host(radio, host_clock, entries: dict[int, int]) -> None: + """Set the field host's clock and send its radio's nodeDB (SPEC RK4). + + The own entry reads :data:`FIELD_READING` and carries + :data:`FIELD_UPTIME`. + + Parameters: + radio: The ``radio`` fixture's namespace. + host_clock: The shared ``host_clock`` fixture, moved to + :data:`FIELD_HOST_NOW`. + entries: ``lastHeard`` of each other entry, by node number. + """ + + host_clock.now = FIELD_HOST_NOW + _load_node_db(radio, FIELD_READING, entries, own_uptime=FIELD_UPTIME) + + +@pytest.fixture +def field(radio, host_clock): + """The field host's radio at connect, its clock 97 days slow. + + Its nodeDB holds its own entry, stamped with the clock's reading, a node it + heard 10 minutes ago and one heard during the previous boot. The own + entry's stamp is read in meshtastic/firmware master: ``PhoneAPI.cpp`` + ``STATE_SEND_MY_INFO`` calls ``refreshLocalMeshNode``, which sets its + ``last_heard`` to the radio's clock. On the field host it read + :data:`FIELD_READING` against the host clock :data:`FIELD_HOST_NOW`. + """ + + _load_node_db( + radio, + RADIO_NOW, + {SENDER: REMOTE_LAST_HEARD, PREV_BOOT_NUM: PREV_BOOT_LAST_HEARD}, + ) + return radio + + +def _warnings(lines: list[dict]) -> list[dict]: + """Return the ``warn`` lines of a :func:`snapshot_log` capture.""" + + return [line for line in lines if line["severity"] == "warn"] + + +def _skipped(lines: list[dict]) -> dict[str, str]: + """Return the reason of each entry a :func:`snapshot_log` capture skipped.""" + + return { + line["node_id"]: line["reason"] for line in lines if line["severity"] == "debug" + } + + +def _file_impostor_under_host_id(radio, last_heard: int) -> None: + """Send :data:`IMPOSTOR_NUM`'s entry, which names the host id (SPEC NI1). + + The library files it under :data:`HOST_ID`, replacing the radio's own + entry there; ``getMyNodeInfo`` still finds the own entry by number. + + Parameters: + radio: The ``radio`` fixture's namespace. + last_heard: The impostor's ``lastHeard``. + """ + + radio.iface._handleFromRadio( + _node_info_frame(IMPOSTOR_NUM, last_heard=last_heard, user_id=HOST_ID) + ) + + +# --------------------------------------------------------------------------- +# Regression guard: the field case +# --------------------------------------------------------------------------- + + +def test_snapshot_last_heard_lands_inside_bulk_window(field): + """A node the radio heard 10 minutes ago is posted inside the 7-day + ``/api/nodes`` window, 10 minutes old by the host clock.""" + + posted = _snapshot(field)[SENDER_ID]["lastHeard"] + + assert HOST_NOW - BULK_WINDOW <= posted <= HOST_NOW, ( + f"snapshot lastHeard {posted} ({_iso(posted)}) is the radio clock, " + f"{(HOST_NOW - posted) / DAY:.0f} days behind the ingestor clock " + f"({_iso(HOST_NOW)}); GET /api/nodes does not list it (7-day bulk and " + "28-day per-id windows)" + ) + assert posted == HOST_NOW - 600 + + +def test_own_entry_is_posted_at_the_host_clock(field): + """The radio's own entry, its clock's reading, is posted as the host clock.""" + + assert _snapshot(field)[HOST_ID]["lastHeard"] == HOST_NOW + + +def test_entry_from_an_earlier_boot_is_skipped_as_clock_epoch(field, snapshot_log): + """An entry stamped more than an hour after the radio's reading was stamped + during an earlier boot: it is skipped with reason ``clock-epoch``.""" + + upserts = _snapshot(field) + + assert set(upserts) == {HOST_ID, SENDER_ID} + skipped = [line for line in snapshot_log if line["severity"] == "debug"] + assert skipped == [ + { + "message": "Skipped snapshot node", + "severity": "debug", + "node_id": PREV_BOOT_ID, + "reason": "clock-epoch", + } + ] + + +def test_one_warning_per_snapshot(field, snapshot_log): + """Each snapshot logs one warning with the offset and both counts; the + counts start again with the next connection's snapshot, which reuses the + provider.""" + + state = make_state(provider=MeshtasticProvider(), iface=field.iface) + + assert daemon._try_send_snapshot(state) is True + assert daemon._try_send_snapshot(state) is True + + assert _warnings(snapshot_log) == [FIELD_WARNING, FIELD_WARNING] + + +def test_library_dicts_keep_the_radio_clock(field): + """The shift is made on copies: the library's nodeDB keeps the radio's times.""" + + nodes = field.iface.nodes + library_entry = nodes[SENDER_ID] + + upserts = _snapshot(field) + + assert upserts[SENDER_ID]["lastHeard"] == HOST_NOW - 600 + assert nodes[SENDER_ID] is library_entry + assert library_entry["lastHeard"] == REMOTE_LAST_HEARD + assert nodes[HOST_ID]["lastHeard"] == RADIO_NOW + assert nodes[PREV_BOOT_ID]["lastHeard"] == PREV_BOOT_LAST_HEARD + assert field.iface.getMyNodeInfo() is nodes[HOST_ID] + + +# --------------------------------------------------------------------------- +# The rule: tolerance, earlier boots, no reading +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("minutes", "shifted"), + [(-61, True), (-60, False), (-59, False), (59, False), (60, False), (61, True)], +) +def test_radio_clock_offset_boundaries( + radio, host_clock, snapshot_log, minutes, shifted +): + """A radio clock 59 or exactly 60 minutes off the host clock, behind or + ahead, leaves the entries as stamped; 61 minutes off shifts them by the + offset, with one warning.""" + + reading = HOST_NOW + minutes * 60 + _load_node_db(radio, reading, {SENDER: reading - 600}) + + posted = _snapshot(radio)[SENDER_ID]["lastHeard"] + + assert posted == (HOST_NOW - 600 if shifted else reading - 600) + expected = [{**FIELD_WARNING, "offset_secs": minutes * 60, "skipped": 0}] + assert _warnings(snapshot_log) == (expected if shifted else []) + + +@pytest.mark.parametrize(("minutes", "posted"), [(59, True), (60, True), (61, False)]) +def test_entry_ahead_of_the_radio_reading_boundaries( + radio, host_clock, minutes, posted +): + """An entry stamped up to 60 minutes after the radio's reading is posted, + shifted; one 61 minutes after is skipped.""" + + stamp = RADIO_NOW + minutes * 60 + _load_node_db(radio, RADIO_NOW, {SENDER: stamp}) + + upserts = _snapshot(radio) + + if posted: + assert upserts[SENDER_ID]["lastHeard"] == stamp - RADIO_OFFSET + else: + assert SENDER_ID not in upserts + + +@pytest.mark.parametrize(("shifted", "posted"), [(1, True), (0, False), (-1, False)]) +def test_entry_shifted_to_zero_or_below_is_skipped( + radio, host_clock, snapshot_log, shifted, posted +): + """An entry the shift would post at 0 or below, 1970 or before by the host + clock, was stamped by another clock than this boot's: it is skipped as + ``clock-epoch`` and counted in the warning. One the shift posts at 1 is + posted.""" + + _load_node_db(radio, HOST_NOW + FAST_OFFSET, {SENDER: FAST_OFFSET + shifted}) + + upserts = _snapshot(radio) + + if posted: + assert upserts[SENDER_ID]["lastHeard"] == shifted + else: + assert _skipped(snapshot_log) == {SENDER_ID: "clock-epoch"} + expected = { + **FIELD_WARNING, + "offset_secs": FAST_OFFSET, + "shifted": 2 if posted else 1, + "skipped": 0 if posted else 1, + } + assert _warnings(snapshot_log) == [expected] + + +@pytest.mark.parametrize( + "own_last_heard", [0, None], ids=["own entry without lastHeard", "no own entry"] +) +def test_without_a_reading_entries_post_as_stamped( + radio, host_clock, snapshot_log, own_last_heard +): + """Without a clock reading from the own entry, entries are posted as + stamped, a radio clock 97 days slow included, and nothing is logged.""" + + _load_node_db(radio, own_last_heard, {SENDER: REMOTE_LAST_HEARD}) + + upserts = _snapshot(radio) + + assert upserts[SENDER_ID]["lastHeard"] == REMOTE_LAST_HEARD + assert snapshot_log == [] + + +@pytest.mark.parametrize(("minutes", "posted"), [(59, True), (60, True), (61, False)]) +def test_without_a_reading_an_entry_ahead_of_the_host_clock( + radio, host_clock, snapshot_log, minutes, posted +): + """Without a clock reading, an entry stamped up to 60 minutes ahead of the + host clock is posted as stamped; one 61 minutes ahead is skipped as + ``clock-epoch``, without a warning.""" + + stamp = HOST_NOW + minutes * 60 + _load_node_db(radio, 0, {SENDER: stamp}) + + upserts = _snapshot(radio) + + if posted: + assert upserts[SENDER_ID]["lastHeard"] == stamp + else: + assert SENDER_ID not in upserts + assert snapshot_log[0]["reason"] == "clock-epoch" + assert _warnings(snapshot_log) == [] + + +def test_within_the_tolerance_every_entry_posts_as_stamped(radio, host_clock): + """A radio clock within the tolerance changes nothing: an entry two hours + ahead of its reading is posted as stamped, as before RK4.""" + + reading = HOST_NOW + 30 * 60 + _load_node_db(radio, reading, {SENDER: reading + 2 * TOLERANCE}) + + assert _snapshot(radio)[SENDER_ID]["lastHeard"] == reading + 2 * TOLERANCE + + +def test_reading_falls_back_to_get_my_node_info(radio, host_clock): + """Before the host id is registered, the library's ``getMyNodeInfo`` finds + the own entry by the radio's node number.""" + + _load_node_db(radio, RADIO_NOW, {SENDER: REMOTE_LAST_HEARD}, register_host=False) + + assert _snapshot(radio)[SENDER_ID]["lastHeard"] == HOST_NOW - 600 + + +def test_mismatched_own_entry_falls_back_to_get_my_node_info( + radio, host_clock, snapshot_log +): + """Another node's entry filed under the host id (SPEC NI1) carries that + node's time, not the radio's clock: the reading comes from + ``getMyNodeInfo``. With the radio's clock right, a node heard 10 minutes + ago and one 61 minutes ahead of the host clock post as stamped, and + nothing is shifted.""" + + _load_node_db( + radio, HOST_NOW, {SENDER: HOST_NOW - 600, AHEAD_NUM: HOST_NOW + 61 * 60} + ) + _file_impostor_under_host_id(radio, HOST_NOW - 3 * DAY) + + upserts = _snapshot(radio) + + assert _skipped(snapshot_log) == {HOST_ID: "num-mismatch"} + assert upserts[SENDER_ID]["lastHeard"] == HOST_NOW - 600 + assert upserts[AHEAD_ID]["lastHeard"] == HOST_NOW + 61 * 60 + assert _warnings(snapshot_log) == [] + + +def test_mismatched_own_entry_without_a_fallback_gives_no_reading( + radio, host_clock, snapshot_log +): + """Another node's entry under the host id, and no own entry for + ``getMyNodeInfo`` to find, give no clock reading: a node heard 10 minutes + ago posts as stamped, one 61 minutes ahead of the host clock is skipped + as ``clock-epoch``, and no warning is logged.""" + + _load_node_db(radio, None, {SENDER: HOST_NOW - 600, AHEAD_NUM: HOST_NOW + 61 * 60}) + _file_impostor_under_host_id(radio, HOST_NOW - 3 * DAY) + + upserts = _snapshot(radio) + + assert _skipped(snapshot_log) == { + HOST_ID: "num-mismatch", + AHEAD_ID: "clock-epoch", + } + assert upserts[SENDER_ID]["lastHeard"] == HOST_NOW - 600 + assert _warnings(snapshot_log) == [] + + +def test_mismatched_own_entry_without_get_my_node_info_gives_no_reading( + scope, host_clock +): + """An interface without ``getMyNodeInfo`` leaves another node's entry + under the host id unread: there is no reading, so a node heard 10 minutes + ago is neither skipped by the impostor's time nor shifted.""" + + handlers.register_host_node_id(HOST_ID) + impostor = {"num": IMPOSTOR_NUM, "lastHeard": HOST_NOW - 3 * DAY} + recent = {"num": SENDER, "lastHeard": HOST_NOW - 600} + provider = MeshtasticProvider() + + provider.node_snapshot_items( + SimpleNamespace(nodes={HOST_ID: impostor, SENDER_ID: recent}) + ) + + assert provider.snapshot_filter_reason(HOST_ID, impostor) == "num-mismatch" + assert provider.snapshot_filter_reason(SENDER_ID, recent) is None + assert provider.snapshot_entry(SENDER_ID, recent) is recent + assert provider.snapshot_clock_warning is None + + +def _no_node_info(): + """A ``getMyNodeInfo`` that fails the way a nodeDB mutated mid-read does.""" + + raise RuntimeError("dictionary changed size during iteration") + + +class _ItemsOnly: + """A nodes object that is no mapping: it offers ``items()`` only.""" + + def items(self): + """Return the one entry, stamped 61 minutes ahead of the host clock.""" + + return [(SENDER_ID, {"num": SENDER, "lastHeard": HOST_NOW + 61 * 60})] + + +@pytest.mark.parametrize( + "make_iface", + [ + lambda: SimpleNamespace( + nodes=dict(_ItemsOnly().items()), getMyNodeInfo=_no_node_info + ), + lambda: SimpleNamespace(nodes=_ItemsOnly()), + ], + ids=["getMyNodeInfo raises", "nodes without get"], +) +def test_unreadable_own_entry_counts_as_no_reading(scope, host_clock, make_iface): + """A failing ``getMyNodeInfo``, or a nodes object without ``get`` and an + interface without ``getMyNodeInfo``, gives no clock reading: the snapshot + still runs, with the host-clock guard.""" + + handlers.register_host_node_id(HOST_ID) + provider = MeshtasticProvider() + + ((node_id, node),) = provider.node_snapshot_items(make_iface()) + + assert provider.snapshot_filter_reason(node_id, node) == "clock-epoch" + assert provider.snapshot_entry(node_id, node) is node + assert provider.snapshot_clock_warning is None + + +def test_channel_filter_reason_wins_over_clock_epoch(field, snapshot_log): + """An entry from an earlier boot heard on a hidden channel is skipped as + ``hidden-channel``; the warning counts only the ``clock-epoch`` skips.""" + + field.apply("hidden") + field.iface._handleFromRadio( + _node_info_frame(0x0E0E0E0E, channel=SECONDARY, last_heard=PREV_BOOT_LAST_HEARD) + ) + + _snapshot(field) + + assert _skipped(snapshot_log) == { + "!0e0e0e0e": "hidden-channel", + PREV_BOOT_ID: "clock-epoch", + } + assert _warnings(snapshot_log) == [FIELD_WARNING] + + +# --------------------------------------------------------------------------- +# The radio's uptime: entries from before this boot +# --------------------------------------------------------------------------- + + +def test_field_host_posts_an_entry_older_than_the_boot_as_stamped( + radio, host_clock, snapshot_log +): + """The field host: its radio 8388455 s slow and up 343965 s. A node + heard 10 minutes ago this boot is shifted, one stamped after the reading + is skipped, and one heard 120 days ago, before the boot started, is + posted as stamped; one warning counts it as ``unshifted``, and the + library's dicts keep their values.""" + + _load_field_host( + radio, + host_clock, + { + SENDER: FIELD_READING - 600, + PREV_BOOT_NUM: PREV_BOOT_LAST_HEARD, + OLD_BOOT_NUM: OLD_BOOT_LAST_HEARD, + }, + ) + nodes = radio.iface.nodes + old_entry = nodes[OLD_BOOT_ID] + + upserts = _snapshot(radio) + + assert upserts[HOST_ID]["lastHeard"] == FIELD_HOST_NOW + assert upserts[SENDER_ID]["lastHeard"] == FIELD_HOST_NOW - 600 + assert upserts[OLD_BOOT_ID]["lastHeard"] == OLD_BOOT_LAST_HEARD + assert _skipped(snapshot_log) == {PREV_BOOT_ID: "clock-epoch"} + assert _warnings(snapshot_log) == [FIELD_UPTIME_WARNING] + assert nodes[OLD_BOOT_ID] is old_entry + assert old_entry["lastHeard"] == OLD_BOOT_LAST_HEARD + assert nodes[SENDER_ID]["lastHeard"] == FIELD_READING - 600 + assert nodes[HOST_ID]["lastHeard"] == FIELD_READING + assert nodes[HOST_ID]["deviceMetrics"] == {"uptimeSeconds": FIELD_UPTIME} + + +@pytest.mark.parametrize( + ("before_boot", "shifted"), [(TOLERANCE, True), (TOLERANCE + 1, False)] +) +def test_entry_before_the_boot_start_boundaries( + radio, host_clock, snapshot_log, before_boot, shifted +): + """An entry stamped exactly 1 hour before the boot started is shifted; one + a second earlier is an earlier boot's, posted as stamped and counted as + ``unshifted``.""" + + stamp = FIELD_BOOT_START - before_boot + _load_field_host(radio, host_clock, {SENDER: stamp}) + + posted = _snapshot(radio)[SENDER_ID]["lastHeard"] + + assert posted == (stamp - FIELD_OFFSET if shifted else stamp) + expected = { + **FIELD_UPTIME_WARNING, + "shifted": 2 if shifted else 1, + "skipped": 0, + "unshifted": 0 if shifted else 1, + } + assert _warnings(snapshot_log) == [expected] + + +@pytest.mark.parametrize(("minutes", "posted"), [(60, True), (61, False)]) +def test_earlier_boot_entry_ahead_of_the_host_clock( + radio, host_clock, snapshot_log, minutes, posted +): + """With a radio clock 30 days fast and up 1 day, an entry from before its + boot up to 60 minutes ahead of the host clock is posted as stamped; one + 61 minutes ahead is skipped as ``clock-epoch``, as without a reading.""" + + stamp = HOST_NOW + minutes * 60 + _load_node_db(radio, HOST_NOW + 30 * DAY, {SENDER: stamp}, own_uptime=DAY) + + upserts = _snapshot(radio) + + if posted: + assert upserts[SENDER_ID]["lastHeard"] == stamp + else: + assert _skipped(snapshot_log) == {SENDER_ID: "clock-epoch"} + expected = { + **FIELD_WARNING, + "offset_secs": 30 * DAY, + "shifted": 1, + "skipped": 0 if posted else 1, + "unshifted": 1 if posted else 0, + } + assert _warnings(snapshot_log) == [expected] + + +@pytest.mark.parametrize( + "uptime", + [None, 0, -5, "junk", True, float(FIELD_UPTIME)], + ids=["absent", "zero", "negative", "junk", "bool", "float"], +) +def test_without_an_uptime_every_entry_is_shifted( + radio, host_clock, snapshot_log, uptime +): + """An uptime that is absent or no positive int counts as none: the entry + from before the boot is shifted, as without the uptime, and the warning + has no ``unshifted``.""" + + _load_field_host(radio, host_clock, {OLD_BOOT_NUM: OLD_BOOT_LAST_HEARD}) + own = radio.iface.nodes[HOST_ID] + if uptime is None: + del own["deviceMetrics"] + else: + own["deviceMetrics"] = {"uptimeSeconds": uptime} + + posted = _snapshot(radio)[OLD_BOOT_ID]["lastHeard"] + + assert posted == OLD_BOOT_LAST_HEARD - FIELD_OFFSET + expected = {**FIELD_WARNING, "offset_secs": FIELD_OFFSET, "skipped": 0} + assert _warnings(snapshot_log) == [expected] + + +# --------------------------------------------------------------------------- +# Mechanics: the hooks, the daemon's per-node handling +# --------------------------------------------------------------------------- + + +class _MutatingEntry(dict): + """A nodeDB entry the library's thread mutates while it is copied.""" + + def keys(self): + """Fail the way iterating a dict that changes size does.""" + + raise RuntimeError("dictionary changed size during iteration") + + __iter__ = keys + + +def test_an_entry_failing_its_copy_costs_only_that_entry( + scope, host_clock, snapshot_log +): + """``snapshot_entry`` runs inside the per-node error handling: an entry + that fails while it is copied is logged and skipped, the others are + posted, the latch is set and the interface stays open.""" + + iface = SimpleNamespace( + nodes={ + HOST_ID: {"num": HOST_NUM, "lastHeard": RADIO_NOW}, + "!0000000a": _MutatingEntry(num=10, lastHeard=RADIO_NOW - 60), + "!0000000b": {"num": 11, "lastHeard": RADIO_NOW - 120}, + } + ) + handlers.register_host_node_id(HOST_ID) + state = make_state(provider=MeshtasticProvider(), iface=iface) + + assert daemon._try_send_snapshot(state) is True + + assert state.iface is iface and state.initial_snapshot_sent is True + posted = { + node_id: entry["lastHeard"] + for _path, body in scope.posts + for node_id, entry in body.items() + if node_id.startswith("!") + } + assert posted == {HOST_ID: HOST_NOW, "!0000000b": HOST_NOW - 120} + failed = [line for line in _warnings(snapshot_log) if "node_id" in line] + assert [line["node_id"] for line in failed] == ["!0000000a"] + assert _warnings(snapshot_log)[-1] == {**FIELD_WARNING, "skipped": 0} + + +def test_hooks_leave_entries_alone_before_a_snapshot(scope): + """A provider that has not read a snapshot's clock changes no entry.""" + + provider = MeshtasticProvider() + entry = {"num": SENDER, "lastHeard": HOST_NOW + 10 * DAY} + + assert provider.snapshot_entry(SENDER_ID, entry) is entry + assert provider.snapshot_filter_reason(SENDER_ID, entry) is None + assert provider.snapshot_clock_warning is None + + +def test_snapshot_entry_passes_entries_without_a_time(field): + """With an offset kept, an entry without ``lastHeard`` and one that is not + a mapping are returned as they are.""" + + provider = MeshtasticProvider() + provider.node_snapshot_items(field.iface) + untimed = {"num": 5} + opaque = SimpleNamespace(lastHeard=RADIO_NOW) + + assert provider.snapshot_filter_reason("!00000005", untimed) is None + assert provider.snapshot_entry("!00000005", untimed) is untimed + assert provider.snapshot_entry("!00000006", opaque) is opaque + assert provider.snapshot_clock_warning == { + "offset_secs": RADIO_OFFSET, + "shifted": 0, + "skipped": 0, + } + + +@pytest.mark.parametrize("last_heard", [0, -5]) +def test_last_heard_of_zero_or_below_posts_as_it_is(field, last_heard): + """With an offset kept, a ``lastHeard`` of 0 or below counts as none: the + entry is not skipped, and it is posted as it is, not shifted to + ``lastHeard - offset`` (8380800 for 0).""" + + provider = MeshtasticProvider() + provider.node_snapshot_items(field.iface) + entry = {"num": 5, "lastHeard": last_heard} + + assert provider.snapshot_filter_reason("!00000005", entry) is None + posted = provider.snapshot_entry("!00000005", entry) + assert posted["lastHeard"] == last_heard + assert posted is entry + assert provider.snapshot_clock_warning == { + "offset_secs": RADIO_OFFSET, + "shifted": 0, + "skipped": 0, + } + + +def test_failed_node_list_read_reports_no_warning(field): + """A snapshot whose node list cannot be read keeps no offset from the + snapshot before it.""" + + class _AlwaysMutating: + """A nodes object whose iteration never settles.""" + + def items(self): + """Fail every attempt.""" + + raise RuntimeError("dictionary changed size during iteration") + + provider = MeshtasticProvider() + provider.node_snapshot_items(field.iface) + assert provider.snapshot_clock_warning is not None + + assert provider.node_snapshot_items(SimpleNamespace(nodes=_AlwaysMutating())) == [] + assert provider.snapshot_clock_warning is None + + +# --------------------------------------------------------------------------- +# The pure offset helper +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize( + ("reading", "expected"), + [ + (None, None), + ("", None), + ("junk", None), + (0, None), + (-5, None), + (HOST_NOW - TOLERANCE, 0), + (HOST_NOW + TOLERANCE, 0), + (HOST_NOW - TOLERANCE - 1, -TOLERANCE - 1), + (HOST_NOW + TOLERANCE + 1, TOLERANCE + 1), + (str(RADIO_NOW), RADIO_OFFSET), + ], +) +def test_radio_clock_offset(reading, expected): + """``None`` for no clock reading, ``0`` within the tolerance (the bound + included), else the reading minus the host clock.""" + + assert receive_time.radio_clock_offset(reading, HOST_NOW) == expected