From 8f7adba65a6cf60418648c4444dc90be73665e69 Mon Sep 17 00:00:00 2001 From: l5y <220195275+l5yth@users.noreply.github.com> Date: Tue, 18 Nov 2025 18:04:40 +0100 Subject: [PATCH] rate limit host device telemetry (#467) * rate limit host device telemetry * Spec: add more unit tests --- data/mesh_ingestor/daemon.py | 3 + data/mesh_ingestor/handlers.py | 69 ++++++++++++++++++++ data/mesh_ingestor/interfaces.py | 42 +++++++++++++ tests/test_mesh.py | 105 +++++++++++++++++++++++++++++++ 4 files changed, 219 insertions(+) diff --git a/data/mesh_ingestor/daemon.py b/data/mesh_ingestor/daemon.py index 7c88247..1be0c03 100644 --- a/data/mesh_ingestor/daemon.py +++ b/data/mesh_ingestor/daemon.py @@ -284,6 +284,9 @@ def main(existing_interface=None) -> None: active_candidate = resolved_target interfaces._ensure_radio_metadata(iface) interfaces._ensure_channel_metadata(iface) + handlers.register_host_node_id( + interfaces._extract_host_node_id(iface) + ) retry_delay = max(0.0, config._RECONNECT_INITIAL_DELAY_SECS) initial_snapshot_sent = False if not announced_target and resolved_target: diff --git a/data/mesh_ingestor/handlers.py b/data/mesh_ingestor/handlers.py index 8405eb1..c2f8ecf 100644 --- a/data/mesh_ingestor/handlers.py +++ b/data/mesh_ingestor/handlers.py @@ -20,6 +20,7 @@ import base64 import contextlib import importlib import json +import math import sys import threading import time @@ -35,6 +36,15 @@ _IGNORED_PACKET_LOG_PATH = Path(__file__).resolve().parents[2] / "ignored.txt" _IGNORED_PACKET_LOCK = threading.Lock() """Lock guarding writes to :data:`_IGNORED_PACKET_LOG_PATH`.""" +_HOST_TELEMETRY_INTERVAL_SECS = 60 * 60 +"""Minimum interval between accepted host telemetry packets.""" + +_host_node_id: str | None = None +"""Canonical ``!xxxxxxxx`` identifier for the connected host device.""" + +_host_telemetry_last_rx: int | None = None +"""Receive timestamp of the last accepted host telemetry packet.""" + def _ignored_packet_default(value: object) -> object: """Return a JSON-serialisable representation for ignored packet data.""" @@ -90,6 +100,50 @@ from .serialization import ( ) +def register_host_node_id(node_id: str | None) -> None: + """Record the canonical identifier for the connected host device. + + Parameters: + node_id: Identifier reported by the connected device. ``None`` clears + the current host assignment. + """ + + global _host_node_id, _host_telemetry_last_rx + canonical = _canonical_node_id(node_id) + _host_node_id = canonical + _host_telemetry_last_rx = None + if canonical: + config._debug_log( + "Registered host device node id", + context="handlers.host_device", + host_node_id=canonical, + ) + + +def host_node_id() -> str | None: + """Return the canonical identifier for the connected host device.""" + + return _host_node_id + + +def _mark_host_telemetry_seen(rx_time: int) -> None: + """Update the last receive time for the host telemetry window.""" + + global _host_telemetry_last_rx + _host_telemetry_last_rx = rx_time + + +def _host_telemetry_suppressed(rx_time: int) -> tuple[bool, int]: + """Return suppression state and minutes remaining for host telemetry.""" + + if _host_telemetry_last_rx is None: + return False, 0 + remaining_secs = (_host_telemetry_last_rx + _HOST_TELEMETRY_INTERVAL_SECS) - rx_time + if remaining_secs <= 0: + return False, 0 + return True, int(math.ceil(remaining_secs / 60.0)) + + def _radio_metadata_fields() -> dict[str, object]: """Return the shared radio metadata fields for payload enrichment.""" @@ -534,6 +588,19 @@ def store_telemetry_packet(packet: Mapping, decoded: Mapping) -> None: rx_time = int(time.time()) rx_iso = _iso(rx_time) + host_id = host_node_id() + if host_id is not None and node_id == host_id: + suppressed, minutes_remaining = _host_telemetry_suppressed(rx_time) + if suppressed: + config._debug_log( + "Suppressed host telemetry update", + context="handlers.store_telemetry", + host_node_id=host_id, + minutes_remaining=minutes_remaining, + ) + return + _mark_host_telemetry_seen(rx_time) + telemetry_time = _coerce_int(_first(telemetry_section, "time", default=None)) channel = _coerce_int(_first(decoded, "channel", default=None)) @@ -1492,8 +1559,10 @@ def on_receive(packet, interface) -> None: __all__ = [ "_queue_post_json", + "host_node_id", "last_packet_monotonic", "on_receive", + "register_host_node_id", "store_neighborinfo_packet", "store_nodeinfo_packet", "store_packet_dict", diff --git a/data/mesh_ingestor/interfaces.py b/data/mesh_ingestor/interfaces.py index df815b0..f35f4c0 100644 --- a/data/mesh_ingestor/interfaces.py +++ b/data/mesh_ingestor/interfaces.py @@ -113,6 +113,47 @@ def _candidate_node_id(mapping: Mapping | None) -> str | None: return None +def _extract_host_node_id(iface) -> str | None: + """Return the canonical node identifier for the connected host device.""" + + if iface is None: + return None + + def _as_mapping(candidate) -> Mapping | None: + mapping = _ensure_mapping(candidate) + if mapping is not None: + return mapping + if callable(candidate): + with contextlib.suppress(Exception): + return _ensure_mapping(candidate()) + return None + + candidates: list[Mapping] = [] + for attr in ("myInfo", "my_node_info", "myNodeInfo", "my_node", "localNode"): + mapping = _as_mapping(getattr(iface, attr, None)) + if mapping is None: + continue + candidates.append(mapping) + nested_info = _ensure_mapping(mapping.get("info")) + if nested_info: + candidates.append(nested_info) + + for mapping in candidates: + node_id = _candidate_node_id(mapping) + if node_id: + return node_id + for key in ("myNodeNum", "my_node_num", "myNodeId", "my_node_id"): + node_id = serialization._canonical_node_id(mapping.get(key)) + if node_id: + return node_id + + node_id = serialization._canonical_node_id(getattr(iface, "myNodeNum", None)) + if node_id: + return node_id + + return None + + def _normalise_nodeinfo_packet(packet) -> dict | None: """Return a dictionary view of ``packet`` with a guaranteed ``id`` when known.""" @@ -771,6 +812,7 @@ __all__ = [ "NoAvailableMeshInterface", "_ensure_channel_metadata", "_ensure_radio_metadata", + "_extract_host_node_id", "_DummySerialInterface", "_DEFAULT_TCP_PORT", "_DEFAULT_TCP_TARGET", diff --git a/tests/test_mesh.py b/tests/test_mesh.py index 5b7839d..6a7c22d 100644 --- a/tests/test_mesh.py +++ b/tests/test_mesh.py @@ -228,6 +228,60 @@ def test_snapshot_interval_defaults_to_60_seconds(mesh_module): assert mesh.SNAPSHOT_SECS == 60 +def test_extract_host_node_id_prefers_my_info_fields(mesh_module): + mesh = mesh_module + + class DummyInterface: + def __init__(self): + self.myInfo = {"my_node_num": 0x9E95CF60} + + iface = DummyInterface() + + assert mesh._extract_host_node_id(iface) == "!9e95cf60" + + +def test_extract_host_node_id_from_nested_info(mesh_module): + mesh = mesh_module + + class DummyInterface: + def __init__(self): + self.myInfo = {"info": {"id": "!cafebabe"}} + + iface = DummyInterface() + + assert mesh._extract_host_node_id(iface) == "!cafebabe" + + +def test_extract_host_node_id_from_callable(mesh_module): + mesh = mesh_module + + class CallableNoDict: + __slots__ = () + + def __call__(self): + return {"id": "!f00ba4"} + + class DummyInterface: + def __init__(self): + self.localNode = CallableNoDict() + + iface = DummyInterface() + + assert mesh._extract_host_node_id(iface) == "!00f00ba4" + + +def test_extract_host_node_id_from_my_node_num_attribute(mesh_module): + mesh = mesh_module + + class DummyInterface: + def __init__(self): + self.myNodeNum = 0xDEADBEEF + + iface = DummyInterface() + + assert mesh._extract_host_node_id(iface) == "!deadbeef" + + @pytest.mark.parametrize("value", ["mock", "Mock", " disabled "]) def test_create_serial_interface_allows_mock(mesh_module, value): mesh = mesh_module @@ -1978,6 +2032,57 @@ def test_store_packet_dict_handles_environment_telemetry(mesh_module, monkeypatc assert payload["modem_preset"] == "MediumFast" +def test_store_packet_dict_throttles_host_telemetry(mesh_module, monkeypatch): + mesh = mesh_module + captured = [] + logs = [] + monkeypatch.setattr( + mesh, + "_queue_post_json", + lambda path, payload, *, priority: captured.append((path, payload, priority)), + ) + monkeypatch.setattr( + mesh.config, + "_debug_log", + lambda message, **metadata: logs.append((message, metadata)), + ) + + mesh.register_host_node_id("!9e95cf60") + + base_packet = { + "id": 1_234, + "fromId": "!9e95cf60", + "decoded": { + "portnum": "TELEMETRY_APP", + "telemetry": { + "time": 1_000, + "deviceMetrics": { + "batteryLevel": 50, + }, + }, + }, + } + + mesh.store_packet_dict({**base_packet, "rxTime": 1_000}) + mesh.store_packet_dict({**base_packet, "id": 1_235, "rxTime": 1_300}) + mesh.store_packet_dict({**base_packet, "id": 1_236, "rxTime": 4_700}) + + assert len(captured) == 2 + first_path, first_payload, _ = captured[0] + second_path, second_payload, _ = captured[1] + assert first_path == "/api/telemetry" + assert second_path == "/api/telemetry" + assert first_payload["id"] == 1_234 + assert second_payload["id"] == 1_236 + + suppression_logs = [ + entry for entry in logs if entry[0] == "Suppressed host telemetry update" + ] + assert suppression_logs + assert suppression_logs[0][1]["host_node_id"] == "!9e95cf60" + assert suppression_logs[0][1]["minutes_remaining"] == 55 + + def test_store_packet_dict_handles_traceroute_packet(mesh_module, monkeypatch): mesh = mesh_module captured = []