import asyncio import copy import logging import math import secrets import time from collections import OrderedDict, deque from collections.abc import Mapping from dataclasses import dataclass, field from enum import Enum from typing import Dict, Optional, Tuple from openhop_core.node.handlers.base import BaseHandler from openhop_core.protocol import Packet from openhop_core.protocol.constants import ( MAX_PATH_SIZE, PAYLOAD_TYPE_ACK, PAYLOAD_TYPE_ADVERT, PAYLOAD_TYPE_ANON_REQ, PAYLOAD_TYPE_MULTIPART, PAYLOAD_TYPE_TRACE, PH_ROUTE_MASK, PH_TYPE_SHIFT, ROUTE_TYPE_DIRECT, ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_DIRECT, ROUTE_TYPE_TRANSPORT_FLOOD, ) from openhop_core.protocol.packet_utils import ( PacketHashingUtils, PacketHeaderUtils, PathUtils, flood_rx_metrics, packet_score, ) from repeater.airtime import AirtimeBudgets from repeater.config import fabric_origin_tx, fabric_selects_by_ingress_radio from repeater.data_acquisition import StorageCollector from repeater.modem_config import normalize_modem_config, redact_modem_tokens_in_place from repeater.neighbour_links import NeighbourLinkTracker from repeater.policy_engine import PolicyDecision, PolicyEngine logger = logging.getLogger("RepeaterHandler") NOISE_FLOOR_INTERVAL = 30.0 # seconds LOOP_DETECT_OFF = "off" LOOP_DETECT_MINIMAL = "minimal" LOOP_DETECT_MODERATE = "moderate" LOOP_DETECT_STRICT = "strict" # Thresholds for flood loop detection, keyed by loop mode and path-hash width. # MeshCore treats 4-byte hashes as reserved, so only 1-3 byte widths are valid here. LOOP_DETECT_MAX_COUNTERS = { LOOP_DETECT_MINIMAL: {1: 4, 2: 2, 3: 1}, LOOP_DETECT_MODERATE: {1: 2, 2: 1, 3: 1}, LOOP_DETECT_STRICT: {1: 1, 2: 1, 3: 1}, } class DropReason(str, Enum): """Canonical, non-alarming reasons a packet was intentionally not retransmitted. Single source of truth shared with the packet router, replacing the string prefixes that used to be duplicated there. Members subclass ``str`` so they compare equal to their text, JSON-serialize, and persist to SQLite as plain strings unchanged. Where the engine still needs a detail suffix (hop counts, loop mode, multipart sub-case) it formats the member into a larger string; the router accepts both the bare member and any string that begins with one. """ DUPLICATE = "Duplicate" MAX_FLOOD_HOPS = "Max flood hops limit reached" PATH_HOP_COUNT_MAX = "Path hop count at maximum" PATH_EXCEEDS_MAX_SIZE = "Path would exceed MAX_PATH_SIZE" DIRECT_NO_PATH = "Direct: no path" DIRECT_NOT_FOR_US = "Direct: not for us" UNSCOPED_FLOOD_DISABLED = "Unscoped flood policy disabled" TRANSPORT_CODE_NOT_ALLOWED = "Transport code not allowed to flood" FLOOD_LOOP_DETECTED = "FLOOD loop detected" MARKED_DO_NOT_RETRANSMIT = "Marked do not retransmit" REPEAT_DISABLED = "Repeat disabled" NO_TX_MODE = "No TX mode" DUTY_CYCLE_LIMIT = "Duty cycle limit" EMPTY_PAYLOAD = "Empty payload" PATH_TOO_LONG = "Path too long" INVALID_ADVERT = "Invalid advert packet" MULTIPART = "Multipart" # Python 3.11+ formats a (str, Enum) member as "DropReason.X" under str()/%s; # return the value so log lines and stored records keep the plain reason text. def __str__(self) -> str: return self.value class ForwardResult(tuple): """A forwarding decision that unpacks as a plain ``(packet, delay_seconds)`` pair. ``extras`` carries additional ``(packet, delay_seconds)`` transmissions that accompany the primary packet: MeshCore's routeDirectRecvAcks queues its optional multi-ack redundancy copies ahead of the plain ACK at the same scheduled time, so the caller must create the extra TX tasks before the primary one. """ def __new__(cls, packet: Packet, delay_s: float, extras=()): result = super().__new__(cls, (packet, delay_s)) result.extras = tuple(extras) return result @dataclass class RadioTxResult: """Outcome of one physical transmission through one radio endpoint.""" radio_id: Optional[str] success: bool metadata: Optional[dict] = None error: Optional[BaseException] = None @dataclass class FanoutTxResult: """Per-egress outcomes of one logical transmission, in attempt order. One logical packet can go out through several Fabric radios; each egress succeeds or fails on its own, and the logical send succeeds when any does. """ results: list = field(default_factory=list) @property def any_success(self) -> bool: return any(result.success for result in self.results) @property def all_success(self) -> bool: return bool(self.results) and all(result.success for result in self.results) @property def successful_radio_ids(self) -> list: return [r.radio_id for r in self.results if r.success and r.radio_id] @property def failed_radio_ids(self) -> list: return [r.radio_id for r in self.results if not r.success and r.radio_id] @property def primary(self) -> Optional[RadioTxResult]: """The first successful egress, else the first one attempted.""" for result in self.results: if result.success: return result return self.results[0] if self.results else None class RepeaterHandler(BaseHandler): @staticmethod def payload_type() -> int: return 0xFF # Special marker (not a real payload type) def __init__( self, config: dict, dispatcher, local_hash: int, *, local_hash_bytes=None, send_advert_func=None, periodic_advert_tick_func=None, ): self.config = config self.dispatcher = dispatcher self.local_hash = local_hash self.local_hash_bytes = local_hash_bytes or bytes([local_hash]) self.send_advert_func = send_advert_func self.periodic_advert_tick_func = periodic_advert_tick_func # One budget per channel; on a single-radio node this is the one manager # it always had. airtime_mgr stays the default radio's, because # everything that reports a single utilisation figure reads it. self.airtime_budgets = AirtimeBudgets(config) self.airtime_mgr = self.airtime_budgets.default self.policy_engine = PolicyEngine.from_runtime_config(config) self.seen_packets = OrderedDict() self.cache_ttl = max( 300, config.get("repeater", {}).get("cache_ttl", 3600) ) # Min 5 min, default 1 hour self.max_cache_size = 1000 self.max_duplicates_per_packet = 20 self.tx_delay_factor = config.get("delays", {}).get("tx_delay_factor", 1.0) self.direct_tx_delay_factor = config.get("delays", {}).get("direct_tx_delay_factor", 0.5) # These are airtime multipliers, not seconds: TX delay is a random value in # [0, 5 * airtime * factor] (matching MeshCore getRetransmitDelay). Flood and # direct share the formula and differ only in the factor. logger.info( "TX delay factors (airtime multipliers, not seconds): " f"flood random 0-5*airtime*F (F={self.tx_delay_factor}), " f"direct random 0-5*airtime*F (F={self.direct_tx_delay_factor})" ) self.use_score_for_tx = config.get("repeater", {}).get("use_score_for_tx", False) self.score_threshold = config.get("repeater", {}).get("score_threshold", 0.3) self.multi_acks = self._normalize_multi_acks(config) self.max_flood_hops = config.get("repeater", {}).get("max_flood_hops", 64) self.send_advert_interval_hours = config.get("repeater", {}).get( "send_advert_interval_hours", 10 ) self.last_advert_time = time.time() self.loop_detect_mode = self._normalize_loop_detect_mode( config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF) ) self.neighbour_link_tracker = NeighbourLinkTracker(config) # (packet hash, path) -> recent physical sends, each with its radio and # timing. Two radios sharing one frequency hear each other send; that echo # ends in our own path hash and must not make this node its own neighbour. self._recent_own_tx: OrderedDict = OrderedDict() radio = dispatcher.radio if dispatcher else None if radio: self.radio_config = { "spreading_factor": getattr(radio, "spreading_factor", 8), "bandwidth": getattr(radio, "bandwidth", 125000), "coding_rate": getattr(radio, "coding_rate", 8), "preamble_length": getattr(radio, "preamble_length", 32), "frequency": getattr(radio, "frequency", 915000000), "tx_power": getattr(radio, "tx_power", 14), } logger.info( f"radio settings: SF={self.radio_config['spreading_factor']}, " f"BW={self.radio_config['bandwidth']}Hz, CR={self.radio_config['coding_rate']}" ) else: raise RuntimeError("Radio object not available - cannot initialize repeater") # Statistics tracking for dashboard self.rx_count = 0 self.forwarded_count = 0 self.dropped_count = 0 self.max_recent_packets = 50 self.recent_packets = deque(maxlen=self.max_recent_packets) self._recent_hash_index = {} self.start_time = time.time() # Flood/direct and duplicate counters (for GET_STATUS / firmware RepeaterStats) self.recv_flood_count = 0 self.recv_direct_count = 0 self.sent_flood_count = 0 self.sent_direct_count = 0 self.flood_dup_count = 0 self.direct_dup_count = 0 # Storage collector for persistent packet logging try: local_identity = dispatcher.local_identity if dispatcher else None self.storage = StorageCollector(config, local_identity, repeater_handler=self) logger.info("StorageCollector initialized successfully") except Exception as e: logger.error(f"Failed to initialize StorageCollector: {e}") self.storage = None # Initialize background timer tracking self.last_noise_measurement = time.time() self.last_cache_cleanup = time.time() self.last_db_cleanup = time.time() self.noise_floor_interval = NOISE_FLOOR_INTERVAL # 30 seconds self._background_task = None self._cached_noise_floor = None # Radio hardware CRC counter per radio id (None on a single-radio node), # so each radio's delta is measured against its own last reading. self._crc_error_baselines: Dict[Optional[str], int] = {} # Cache transport keys for efficient lookup self._transport_keys_cache = None self._transport_keys_cache_time = 0 self._transport_keys_cache_ttl = 60 # Cache for 60 seconds # Serialise all radio TX calls. # # Background: since the queue loop dispatches each packet as an # asyncio.create_task, multiple _route_packet coroutines can have their # TX delay timers running concurrently — which is the intended behaviour # (firmware nodes do the same with a hardware timer). However, the # LoRa radio is half-duplex: it can only transmit one packet at a time. # Without serialisation, two tasks whose delay timers expire near- # simultaneously both call dispatcher.send_packet, interleaving SPI/serial # commands to the radio and both passing the LBT check before either has # actually transmitted. # # _tx_lock is acquired after each delay sleep and held for the entire # send_packet call. Delays still run concurrently; only the radio # access is serialised. This also eliminates the TOCTOU gap in duty-cycle # enforcement — see schedule_retransmit / delayed_send for details. self._tx_lock = asyncio.Lock() self._start_background_tasks() async def __call__( self, packet: Packet, metadata: Optional[dict] = None, local_transmission: bool = False ) -> bool: if metadata is None: metadata = {} # Only count as receive when packet came from the radio (not locally injected) if not local_transmission: self.rx_count += 1 route_type = packet.header & PH_ROUTE_MASK if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.recv_flood_count += 1 elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT): self.recv_direct_count += 1 try: rx_mgr = self.airtime_budgets.for_radio(self._rx_radio_id_of(packet, metadata)) rx_airtime_ms = rx_mgr.calculate_airtime(packet.get_raw_length()) rx_mgr.record_rx(rx_airtime_ms) except Exception as exc: logger.debug(f"Failed to record RX airtime: {exc}") route_type = packet.header & PH_ROUTE_MASK pkt_hash_full = packet.calculate_packet_hash().hex().upper() snr = metadata.get("snr", 0.0) rssi = metadata.get("rssi", 0) # TX mode: forward (repeat on), monitor (no repeat, tenants can TX), no_tx (all TX off) mode = self.config.get("repeater", {}).get("mode", "forward") if mode not in ("forward", "monitor", "no_tx"): mode = "forward" allow_forward = mode == "forward" allow_local_tx = mode != "no_tx" policy_context = { "route_type": route_type, "payload_type": packet.get_payload_type() if hasattr(packet, "get_payload_type") else None, "payload_length": len(packet.payload or b""), "path_hash_size": packet.get_path_hash_size() if hasattr(packet, "get_path_hash_size") else None, "hop_count": packet.get_path_hash_count() if hasattr(packet, "get_path_hash_count") else None, "rssi": metadata.get("rssi", 0), "snr": metadata.get("snr", 0.0), "rx_radio_id": metadata.get("rx_radio_id") or getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None), "local_transmission": local_transmission, "mode": mode, } prechecked_decision = metadata.get("_policy_precheck_decision") if isinstance(prechecked_decision, PolicyDecision): policy_decision = prechecked_decision else: policy_decision = self.policy_engine.evaluate(packet, policy_context) policy_reason = None if policy_decision.matched: logger.info(policy_decision.reason) if policy_decision.action == "drop": allow_forward = False allow_local_tx = False policy_reason = self._policy_drop_reason(policy_decision) if logger.isEnabledFor(logging.DEBUG): logger.debug( f"RX packet: header=0x{packet.header:02x}, payload_len={len(packet.payload or b'')}, " f"path_len={len(packet.path) if packet.path else 0}, " f"rssi={metadata.get('rssi', 'N/A')}, snr={metadata.get('snr', 'N/A')}, mode={mode}" ) # clone the packet to avoid modifying the original processed_packet = copy.deepcopy(packet) transmitted = False tx_delay_ms = 0.0 drop_reason = None tx_radio_id = None lbt_attempts = 0 lbt_backoff_delays_ms = None lbt_channel_busy = False rx_radio_id = ( metadata.get("rx_radio_id") or getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) ) tx_radio_ids_sent = None tx_egress = None if local_transmission and not rx_radio_id: # Originated here: companion, repeater/room identity, protocol reply. tx_radio_ids = self._resolve_origin_tx_radio_ids() else: # Heard on RF and being repeated. A helper re-emitting a received # packet through the local path (TRACE forwarding) still carries its # ingress radio, so it resolves as the relay it is. tx_radio_ids = self._resolve_relay_tx_radio_ids(rx_radio_id) original_path_hashes = packet.get_path_hashes_hex() path_hash_size = packet.get_path_hash_size() if not local_transmission: payload_type = ( packet.get_payload_type() if hasattr(packet, "get_payload_type") else None ) frame_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0 score = flood_rx_metrics( frame_len, snr, self.radio_config["spreading_factor"], self.radio_config["bandwidth"], self.radio_config["coding_rate"], self.radio_config["preamble_length"], ).score if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id): self.neighbour_link_tracker.observe( packet, route_type=route_type, payload_type=payload_type, rssi=rssi, snr=snr, score=score, is_duplicate=(pkt_hash_full in self.seen_packets), rx_radio_id=rx_radio_id, ) # Process for forwarding (skip if repeat disabled or if this is a local transmission). # Pass pkt_hash_full so flood_forward / direct_forward don't recompute SHA-256. result = ( None if (not allow_forward or local_transmission) else self.process_packet(processed_packet, snr, packet_hash=pkt_hash_full) ) forwarded_path_hashes = None # For local transmissions, create a direct transmission result (if local TX allowed) if local_transmission and allow_local_tx: # Mark local packet as seen to prevent duplicate processing when received back self.mark_seen(packet, packet_hash=pkt_hash_full) # Calculate transmission delay for local packets delay = self._calculate_tx_delay(packet, snr) result = (packet, delay) forwarded_path_hashes = packet.get_path_hashes_hex() if logger.isEnabledFor(logging.DEBUG): logger.debug(f"Local transmission: calculated delay {delay:.3f}s") if result: fwd_pkt, delay = result tx_delay_ms = delay * 1000.0 # Capture the forwarded path (after modification) forwarded_path_hashes = fwd_pkt.get_path_hashes_hex() # MeshCore queues multi-ack redundancy copies ahead of the primary # ACK at the same scheduled time, so create their TX tasks first. # Each task re-checks the duty-cycle gate before transmitting, and # every packet from one forwarding decision shares its egress set. # A fan-out decides send-time normalisation (flood scope, hash mode) # up front; do it before metering so every duty-cycle gate sees the # bytes that actually go on air. if tx_radio_ids is None: # Nothing above named an egress, so the fabric picks one inside # send(). Ask it now, with this packet's own ingress radio, and # keep the answer: on a core that passes the ingress radio to # the selector the answer is a function of the packet, so it is # the same answer send() would reach and nothing moves on the # air. What it buys is that the duty cycle is charged to the # channel that carries the packet instead of to whichever radio # the node lists first. An older core cannot answer, and # tx_radio_ids stays None: the fabric goes on choosing at send # time and the gate holds every radio it could pick to its own # budget. planned = self._planned_egress_radio_id(fwd_pkt, rx_radio_id) if planned is not None: tx_radio_ids = (planned,) fanout = bool(tx_radio_ids) and len(tx_radio_ids) > 1 extra_tx_tasks = [] for extra_pkt, extra_delay in getattr(result, "extras", ()): if fanout: self._freeze_tx_normalisation(extra_pkt) extra_airtime_ms = self.airtime_mgr.calculate_airtime(extra_pkt.get_raw_length()) extra_tx_tasks.append( ( extra_pkt, await self._schedule_egress_tx( extra_pkt, extra_delay, extra_airtime_ms, tx_radio_ids ), ) ) # Check duty-cycle before scheduling TX if fanout: self._freeze_tx_normalisation(fwd_pkt) airtime_ms = self.airtime_mgr.calculate_airtime(fwd_pkt.get_raw_length()) # Advisory, and asked of the radios this packet is headed for: one # radio being over budget is no reason to drop a forward the other # can carry. The authoritative gate is per egress, under the TX lock. can_tx, wait_time = self._egress_can_transmit(fwd_pkt, tx_radio_ids) # LBT metadata (set after any TX path that awaits send) tx_metadata = None lbt_attempts = 0 lbt_backoff_delays_ms = None lbt_channel_busy = False if not can_tx and not local_transmission: logger.warning( f"Duty-cycle limit exceeded. Airtime={airtime_ms:.1f}ms, " f"wait={wait_time:.1f}s before retry" ) self.dropped_count += 1 drop_reason = DropReason.DUTY_CYCLE_LIMIT else: if can_tx: failure_reason = "TX failed" failure_label = "Local TX failed" else: # Defer local TX until duty cycle allows instead of dropping logger.info( f"Duty-cycle limit: deferring local TX by {wait_time:.1f}s " f"(airtime={airtime_ms:.1f}ms)" ) delay += wait_time failure_reason = "TX failed (deferred)" failure_label = "Deferred local TX failed" tx_task = await self._schedule_egress_tx( fwd_pkt, delay, airtime_ms, tx_radio_ids, local_transmission=local_transmission, ) try: tx_result = await self._await_egress_tx(tx_task, fwd_pkt) except Exception as e: transmitted = False drop_reason = failure_reason logger.warning(f"{failure_label}: {e}") raise # One logical forward, however many radios carried it: the # per-radio sent_flood/sent_direct counters track physical sends. if tx_result.any_success: self.forwarded_count += 1 transmitted = True else: transmitted = False drop_reason = failure_reason self.dropped_count += 1 if len(tx_result.results) > 1 and not tx_result.all_success: logger.warning( "Fan-out %s TX failure: %s successful=%s failed=%s", "partial" if tx_result.any_success else "total", f"rx_radio={rx_radio_id}" if rx_radio_id else "origin=local", tx_result.successful_radio_ids, tx_result.failed_radio_ids, ) tx_radio_ids_sent = tx_result.successful_radio_ids or None tx_egress = self._egress_records(tx_result) # Scalar TX fields describe the primary egress. primary = tx_result.primary if primary is not None and primary.success: tx_radio_id = primary.radio_id tx_metadata = primary.metadata if primary is not None else None if tx_metadata: lbt_attempts = tx_metadata.get("lbt_attempts", 0) lbt_backoff_delays_ms = tx_metadata.get("lbt_backoff_delays_ms", []) lbt_channel_busy = tx_metadata.get("lbt_channel_busy", False) if lbt_attempts > 0: total_lbt_delay = sum(lbt_backoff_delays_ms) logger.info( f"LBT: {lbt_attempts} attempts, {total_lbt_delay:.0f}ms delay, " f"backoffs={lbt_backoff_delays_ms}" ) # Redundancy copies ride alongside the primary result: collect them # without letting a failed extra change the primary outcome. Their # sends are physical transmissions like any other, so they belong in # the egress rows -- leaving them out would undercount exactly the # traffic a multi-ack burst adds to a busy channel. for extra_pkt, extra_task in extra_tx_tasks: try: extra_result = await self._await_egress_tx(extra_task, extra_pkt) if extra_result.any_success: self.forwarded_count += 1 extra_egress = self._egress_records(extra_result) if extra_egress: tx_egress = (tx_egress or []) + extra_egress except Exception as e: logger.warning(f"Multi-ack redundancy TX failed: {e}") else: self.dropped_count += 1 # Determine drop reason if local_transmission and not allow_local_tx: drop_reason = policy_reason or DropReason.NO_TX_MODE elif not allow_forward: drop_reason = policy_reason or DropReason.REPEAT_DISABLED else: # Check if packet has a specific drop reason set by handlers drop_reason = processed_packet.drop_reason or self._get_drop_reason( processed_packet, packet_hash=pkt_hash_full ) if logger.isEnabledFor(logging.DEBUG): logger.debug(f"Packet not forwarded: {drop_reason}") # Extract packet type and route from header if not hasattr(packet, "header") or packet.header is None: logger.error(f"Packet missing header attribute! Packet: {packet}") payload_type = 0 route_type = 0 else: header_info = PacketHeaderUtils.parse_header(packet.header) payload_type = header_info["payload_type"] route_type = header_info["route_type"] if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Packet header=0x{packet.header:02x}, type={payload_type}, route={route_type}" ) # Check if this is a duplicate is_dupe = pkt_hash_full in self.seen_packets and not transmitted # Set drop reason for duplicates and count flood vs direct dups if is_dupe and drop_reason is None: drop_reason = DropReason.DUPLICATE if is_dupe: if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.flood_dup_count += 1 elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT): self.direct_dup_count += 1 # Expose effective drop reason to PacketRouter. # Some Packet implementations are slot-based and cannot accept dynamic attrs. metadata["_repeater_drop_reason"] = drop_reason try: setattr(packet, "_repeater_drop_reason", drop_reason) except (AttributeError, TypeError): pass display_hashes = ( original_path_hashes if original_path_hashes else packet.get_path_hashes_hex() ) path_hash = self._path_hash_display(display_hashes) src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type) # Record packet for charts packet_record = self._build_packet_record( packet, payload_type, route_type, rssi, snr, original_path_hashes, path_hash_size, path_hash, src_hash, dst_hash, transmitted=transmitted, drop_reason=drop_reason, is_duplicate=is_dupe, forwarded_path=forwarded_path_hashes, tx_delay_ms=tx_delay_ms, rx_radio_id=rx_radio_id, tx_radio_id=tx_radio_id, tx_radio_ids=tx_radio_ids_sent, lbt_attempts=lbt_attempts, lbt_backoff_delays_ms=lbt_backoff_delays_ms, lbt_channel_busy=lbt_channel_busy, packet_hash=pkt_hash_full, ) # Store packet record to persistent storage # Skip mqtt only for invalid packets (not duplicates or operational drops) if self.storage: try: # Only skip mqtt for actual invalid/bad packets invalid_reasons = ( DropReason.INVALID_ADVERT, DropReason.EMPTY_PAYLOAD, DropReason.PATH_TOO_LONG, ) skip_mqtt = drop_reason in invalid_reasons if drop_reason else False self.storage.record_packet( packet_record, skip_mqtt_if_invalid=skip_mqtt, tx_egress=tx_egress ) except Exception as e: logger.error(f"Failed to store packet record: {e}") # If this is a duplicate, try to attach it to the original packet if is_dupe and len(self.recent_packets) > 0: prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"]) if prev_pkt is not None: # Add duplicate to original packet's duplicate list if "duplicates" not in prev_pkt: prev_pkt["duplicates"] = [] if len(prev_pkt["duplicates"]) < self.max_duplicates_per_packet: prev_pkt["duplicates"].append(packet_record) # Don't add duplicate to main list, just track in original else: # Original not found, add as regular packet self._append_recent_packet(packet_record) else: # Not a duplicate or first occurrence self._append_recent_packet(packet_record) return transmitted @staticmethod def _policy_drop_reason(decision: PolicyDecision) -> str: if decision.rule_id is None: return "Policy blocked packet" return f"Policy blocked packet (rule {decision.rule_id})" def log_trace_record(self, packet_record: dict) -> None: """Manually log a packet trace record (used by external callers)""" self._append_recent_packet(packet_record) self.rx_count += 1 if packet_record.get("transmitted", False): self.forwarded_count += 1 else: self.dropped_count += 1 # Store to persistent storage (same as __call__ does) if self.storage: try: self.storage.record_packet(packet_record) except Exception as e: logger.error(f"Failed to store packet record: {e}") def record_packet_only(self, packet: Packet, metadata: dict) -> None: """Record a packet for UI/storage without running forwarding or duplicate logic. Used by the packet router for injection-only types (ANON_REQ, ACK, PATH, etc.) so they still appear in the web UI. TRACE packets are excluded: TraceHelper.log_trace_record stores the real trace path; packet.path on TRACE holds SNR bytes, not routing hashes. """ if not self.storage: return rssi = metadata.get("rssi", 0) snr = metadata.get("snr", 0.0) if not hasattr(packet, "header") or packet.header is None: logger.debug("record_packet_only: packet missing header, skipping") return header_info = PacketHeaderUtils.parse_header(packet.header) payload_type = header_info["payload_type"] route_type = header_info["route_type"] if payload_type == PAYLOAD_TYPE_TRACE: return original_path_hashes = packet.get_path_hashes_hex() path_hash_size = packet.get_path_hash_size() path_hash = self._path_hash_display(original_path_hashes) src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type) packet_record = self._build_packet_record( packet, payload_type, route_type, rssi, snr, original_path_hashes, path_hash_size, path_hash, src_hash, dst_hash, rx_radio_id=( metadata.get("rx_radio_id") or getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) ), packet_hash=packet.calculate_packet_hash().hex().upper(), ) try: self.storage.record_packet(packet_record, skip_mqtt_if_invalid=False) except Exception as e: logger.error(f"Failed to store packet record (record_packet_only): {e}") return self._append_recent_packet(packet_record) # Another radio of this node hears a send only while it is on the air or # shortly after, allowing for a remote modem's delivery delay. OWN_ECHO_MARGIN_SECONDS = 2.0 # A send that never reports back stops counting after this long. OWN_ECHO_INFLIGHT_LIMIT_SECONDS = 60.0 OWN_ECHO_MAX_ENTRIES = 256 def _note_own_send_started(self, packet: Packet, radio_id: Optional[str]): """Record a physical send about to go on air; return a handle for its outcome. Only on a node with two or more radios: a lone radio cannot hear itself. """ _, radio_ids = self._fabric_endpoints() if not radio_ids or len(radio_ids) < 2: return None path_hashes = packet.get_path_hashes_hex() if not path_hashes: return None key = (packet.calculate_packet_hash().hex().upper(), tuple(path_hashes)) record = {"radio_id": radio_id, "started": time.monotonic(), "finished": None} sends = self._recent_own_tx.pop(key, []) sends.append(record) self._recent_own_tx[key] = sends while len(self._recent_own_tx) > self.OWN_ECHO_MAX_ENTRIES: self._recent_own_tx.popitem(last=False) return key, record def _note_own_send_finished(self, handle, packet: Packet, *, sent: bool) -> None: """Close a send: keep it for its echo window, or forget it if nothing went on air.""" if handle is None: return key, record = handle if record["finished"] is not None: return record["finished"] = time.monotonic() if not sent: sends = [r for r in self._recent_own_tx.get(key, []) if r is not record] if sends: self._recent_own_tx[key] = sends else: self._recent_own_tx.pop(key, None) return if record["radio_id"] is None: metadata = getattr(packet, "_tx_metadata", None) if isinstance(metadata, dict): record["radio_id"] = metadata.get("radio_id") or metadata.get("tx_radio_id") def _own_send_is_live(self, record: dict, now: float) -> bool: if record["finished"] is None: return now - record["started"] <= self.OWN_ECHO_INFLIGHT_LIMIT_SECONDS return now - record["finished"] <= self.OWN_ECHO_MARGIN_SECONDS def _is_own_echo(self, packet_hash: str, path_hashes, rx_radio_id) -> bool: """True when another radio of this node is hearing our own transmission. The reception must carry exactly the path we sent, arrive on a radio other than the one that sent it, and arrive while that send is on the air or within OWN_ECHO_MARGIN_SECONDS of it finishing. A neighbour repeating our copy appends its own hash; one whose 1-byte hash collides with ours is only mistaken for an echo if it repeats the same packet on the same path inside that window. """ if rx_radio_id is None or not path_hashes or not self._recent_own_tx: return False now = time.monotonic() while self._recent_own_tx: sends = next(iter(self._recent_own_tx.values())) if any(self._own_send_is_live(record, now) for record in sends): break self._recent_own_tx.popitem(last=False) for record in self._recent_own_tx.get((packet_hash, tuple(path_hashes)), ()): if not self._own_send_is_live(record, now): continue sender = record["radio_id"] if sender is None or str(sender) != str(rx_radio_id): return True return False def record_duplicate(self, packet: Packet, rssi: int = 0, snr: float = 0.0) -> None: """Record a known-duplicate packet for UI/storage visibility without forwarding. Called by the raw_packet_subscriber path so that path variants blocked by the Dispatcher's payload-based dedup still appear in the UI. """ self.rx_count += 1 route_type = packet.header & PH_ROUTE_MASK if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.recv_flood_count += 1 self.flood_dup_count += 1 elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT): self.recv_direct_count += 1 self.direct_dup_count += 1 header_info = PacketHeaderUtils.parse_header(packet.header) payload_type = header_info["payload_type"] route_type_parsed = header_info["route_type"] original_path_hashes = packet.get_path_hashes_hex() path_hash_size = packet.get_path_hash_size() path_hash = self._path_hash_display(original_path_hashes) src_hash, dst_hash = self._packet_record_src_dst(packet, payload_type) pkt_hash_full = packet.calculate_packet_hash().hex().upper() frame_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0 score = flood_rx_metrics( frame_len, snr, self.radio_config["spreading_factor"], self.radio_config["bandwidth"], self.radio_config["coding_rate"], self.radio_config["preamble_length"], ).score rx_radio_id = getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) if not self._is_own_echo(pkt_hash_full, original_path_hashes, rx_radio_id): self.neighbour_link_tracker.observe( packet, route_type=route_type, payload_type=payload_type, rssi=float(rssi), snr=float(snr), score=score, is_duplicate=True, rx_radio_id=rx_radio_id, ) packet_record = self._build_packet_record( packet, payload_type, route_type_parsed, rssi, snr, original_path_hashes, path_hash_size, path_hash, src_hash, dst_hash, transmitted=False, drop_reason=DropReason.DUPLICATE, is_duplicate=True, rx_radio_id=( getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) ), packet_hash=pkt_hash_full, ) if self.storage: try: self.storage.record_packet(packet_record, skip_mqtt_if_invalid=False) except Exception as e: logger.error(f"Failed to store duplicate record: {e}") # Group under original in recent_packets if len(self.recent_packets) > 0: prev_pkt = self._recent_hash_index.get(packet_record["packet_hash"]) if prev_pkt is not None: if "duplicates" not in prev_pkt: prev_pkt["duplicates"] = [] prev_pkt["duplicates"].append(packet_record) else: self._append_recent_packet(packet_record) else: self._append_recent_packet(packet_record) def cleanup_cache(self): now = time.time() expired = [k for k, ts in self.seen_packets.items() if now - ts > self.cache_ttl] for k in expired: del self.seen_packets[k] def _path_hash_display(self, display_hashes) -> Optional[str]: """Build path hash string for packet record from path hashes list.""" if not display_hashes: return None display = display_hashes[:8] if len(display_hashes) > 8: display = list(display) + ["..."] return "[" + ", ".join(display) + "]" def _packet_record_src_dst( self, packet: Packet, payload_type: int ) -> Tuple[Optional[str], Optional[str]]: """Return (src_hash, dst_hash) for packet_record from packet and payload_type.""" src_hash = None dst_hash = None payload = getattr(packet, "payload", None) if payload_type in [0x00, 0x01, 0x02, 0x08]: if payload and len(payload) >= 2: dst_hash = f"{payload[0]:02X}" src_hash = f"{payload[1]:02X}" elif payload_type == PAYLOAD_TYPE_ADVERT: if payload and len(payload) >= 1: src_hash = f"{payload[0]:02X}" elif payload_type == PAYLOAD_TYPE_ANON_REQ: if payload and len(payload) >= 1: dst_hash = f"{payload[0]:02X}" return (src_hash, dst_hash) def _build_packet_record( self, packet: Packet, payload_type: int, route_type: int, rssi: int, snr: float, original_path_hashes, path_hash_size: int, path_hash: Optional[str], src_hash: Optional[str], dst_hash: Optional[str], *, transmitted: bool = False, drop_reason: Optional[str] = None, is_duplicate: bool = False, forwarded_path=None, tx_delay_ms: float = 0.0, rx_radio_id: Optional[str] = None, tx_radio_id: Optional[str] = None, tx_radio_ids=None, lbt_attempts: int = 0, lbt_backoff_delays_ms=None, lbt_channel_busy: bool = False, packet_hash: Optional[str] = None, ) -> dict: """Build a single packet_record dict for storage and recent_packets. ``tx_radio_id`` is the primary successful egress; ``tx_radio_ids`` lists every radio that successfully transmitted this logical packet. """ pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() payload = getattr(packet, "payload", None) payload_len = len(payload or b"") upstream_hash, upstream_hash_size = self.neighbour_link_tracker.get_upstream_peer_identity( packet, route_type, payload_type, path_hashes=original_path_hashes, path_hash_size=path_hash_size, ) # LoRa time-on-air for this packet (Semtech reference formula). # Computed once here so every downstream consumer (MQTT, SQLite, Glass, # websocket) reads the same value instead of recomputing or shipping # zeros. The same calculator is used by the RX accumulator and the # duty-cycle gate, keeping reporting and metering aligned. airtime_ms = 0.0 try: raw_len = packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0 if raw_len > 0: # The radio that carried it: as heard, else as first sent. On a # bridge the same bytes are a different time on air per radio, so # a figure computed on another radio's modulation describes a # transmission that never happened. carrier = rx_radio_id or getattr(packet, "_rx_radio_id", None) or tx_radio_id airtime_ms = float( self.airtime_budgets.for_radio(carrier).calculate_airtime(raw_len) ) except Exception as e: logger.debug(f"Could not compute airtime for packet record: {e}") return { "timestamp": time.time(), "header": ( f"0x{packet.header:02X}" if hasattr(packet, "header") and packet.header is not None else None ), "payload": payload.hex() if payload else None, "payload_length": len(payload) if payload else 0, "type": payload_type, "route": route_type, "length": payload_len, "rssi": rssi, "snr": snr, "score": flood_rx_metrics( packet.get_raw_length() if hasattr(packet, "get_raw_length") else 0, snr, self.radio_config["spreading_factor"], self.radio_config["bandwidth"], self.radio_config["coding_rate"], self.radio_config["preamble_length"], ).score, "tx_delay_ms": tx_delay_ms, "rx_radio_id": rx_radio_id or getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None), "tx_radio_id": tx_radio_id, "tx_radio_ids": list(tx_radio_ids) if tx_radio_ids else None, "airtime_ms": airtime_ms, "transmitted": transmitted, "is_duplicate": is_duplicate, "packet_hash": pkt_hash[:16], "drop_reason": drop_reason, "path_hash": path_hash, "src_hash": src_hash, "dst_hash": dst_hash, "original_path": original_path_hashes or None, "forwarded_path": forwarded_path, "path_hash_size": path_hash_size, "upstream_hash": upstream_hash, "upstream_hash_size": upstream_hash_size, "raw_packet": packet.write_to().hex() if hasattr(packet, "write_to") else None, "lbt_attempts": lbt_attempts, "lbt_backoff_delays_ms": lbt_backoff_delays_ms, "lbt_channel_busy": lbt_channel_busy, } def _append_recent_packet(self, packet_record: dict) -> None: """Append packet to bounded recent list and keep hash index aligned.""" if len(self.recent_packets) >= self.max_recent_packets: oldest = self.recent_packets.popleft() oldest_hash = oldest.get("packet_hash") if isinstance(oldest, dict) else None if oldest_hash and self._recent_hash_index.get(oldest_hash) is oldest: del self._recent_hash_index[oldest_hash] self.recent_packets.append(packet_record) pkt_hash = packet_record.get("packet_hash") if isinstance(packet_record, dict) else None if pkt_hash: self._recent_hash_index[pkt_hash] = packet_record def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str: if self.is_duplicate(packet, packet_hash=packet_hash): return DropReason.DUPLICATE if not packet or not packet.payload: return DropReason.EMPTY_PAYLOAD if len(packet.path or []) > MAX_PATH_SIZE: return DropReason.PATH_TOO_LONG route_type = packet.header & PH_ROUTE_MASK if route_type == ROUTE_TYPE_FLOOD: # Check if unscoped flood policy blocked it unscoped_flood_allow = self.config.get("mesh", {}).get( "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True) ) if not unscoped_flood_allow: return DropReason.UNSCOPED_FLOOD_DISABLED if route_type == ROUTE_TYPE_DIRECT: hash_size = packet.get_path_hash_size() if not packet.path or len(packet.path) < hash_size: return DropReason.DIRECT_NO_PATH next_hop = bytes(packet.path[:hash_size]) if next_hop != self.local_hash_bytes[:hash_size]: return DropReason.DIRECT_NOT_FOR_US # Default reason return "Unknown" def is_duplicate(self, packet: Packet, packet_hash: Optional[str] = None) -> bool: """Return True if this packet has already been seen. Accepts an optional pre-computed packet_hash to avoid a redundant SHA-256 when the caller (e.g. __call__ → process_packet → flood/direct_forward) has already calculated the hash. Falls back to computing it if not provided. INVARIANT: this method is synchronous with no await points. The caller (process_packet / __call__) relies on is_duplicate + mark_seen being effectively atomic within the asyncio event loop. Do NOT add any await here without revisiting that invariant. """ pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() return pkt_hash in self.seen_packets def mark_seen(self, packet: Packet, packet_hash: Optional[str] = None): pkt_hash = packet_hash or packet.calculate_packet_hash().hex().upper() self.seen_packets[pkt_hash] = time.time() if len(self.seen_packets) > self.max_cache_size: self.seen_packets.popitem(last=False) def validate_packet(self, packet: Packet) -> Tuple[bool, str]: if not packet or not packet.payload: return False, DropReason.EMPTY_PAYLOAD if packet.get_path_hash_size() > 3: return False, "Reserved path hash size is invalid" if len(packet.path or []) > MAX_PATH_SIZE: return ( False, f"Path length {len(packet.path or [])} exceeds MAX_PATH_SIZE ({MAX_PATH_SIZE})", ) return True, "" @staticmethod def _normalize_multi_acks(config: dict) -> int: """Read ``repeater.multi_acks``, constrained to 0..1 like MeshCore's CommonCLI prefs load.""" try: value = int(config.get("repeater", {}).get("multi_acks", 0)) except (TypeError, ValueError): return 0 return max(0, min(1, value)) def _normalize_loop_detect_mode(self, mode) -> str: if isinstance(mode, str): normalized = mode.strip().lower() if normalized in { LOOP_DETECT_OFF, LOOP_DETECT_MINIMAL, LOOP_DETECT_MODERATE, LOOP_DETECT_STRICT, }: return normalized return LOOP_DETECT_OFF def _get_loop_detect_mode(self) -> str: return self.loop_detect_mode def _is_flood_looped(self, packet: Packet, mode: Optional[str] = None) -> bool: mode = mode or self._get_loop_detect_mode() if mode == LOOP_DETECT_OFF: return False max_counters = LOOP_DETECT_MAX_COUNTERS.get(mode) if max_counters is None: return False hash_size = packet.get_path_hash_size() if hash_size not in max_counters: return False max_counter = max_counters[hash_size] hop_count = packet.get_path_hash_count() path = packet.path or bytearray() local_hash = self.local_hash_bytes[:hash_size] local_count = sum( 1 for i in range(hop_count) if bytes(path[i * hash_size : (i + 1) * hash_size]) == local_hash ) return local_count >= max_counter def _check_transport_codes(self, packet: Packet) -> Tuple[bool, str]: if not self.storage: logger.warning("Transport code check failed: no storage available") return False, "No storage available for transport key validation" try: from openhop_core.protocol.transport_keys import calc_transport_code # Check cache validity current_time = time.time() if ( self._transport_keys_cache is None or current_time - self._transport_keys_cache_time > self._transport_keys_cache_ttl ): # Refresh cache self._transport_keys_cache = self.storage.get_transport_keys() self._transport_keys_cache_time = current_time transport_keys = self._transport_keys_cache if not transport_keys: return False, "No transport keys configured" # Check if packet has transport codes if not packet.has_transport_codes(): return False, "No transport codes present" transport_code_0 = packet.transport_codes[0] # First transport code # Check packet against each transport key for key_record in transport_keys: transport_key_encoded = key_record.get("transport_key") key_name = key_record.get("name", "unknown") flood_policy = key_record.get("flood_policy", "deny") if not transport_key_encoded: continue try: import base64 transport_key = base64.b64decode(transport_key_encoded) expected_code = calc_transport_code(transport_key, packet) if transport_code_0 == expected_code: if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Transport code validated for key '{key_name}' with policy '{flood_policy}'" ) # Update last_used timestamp for this key try: key_id = key_record.get("id") if key_id: self.storage.update_transport_key( key_id=key_id, last_used=time.time() ) if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Updated last_used timestamp for transport key '{key_name}'" ) except Exception as e: logger.warning( f"Failed to update last_used for transport key '{key_name}': {e}" ) # Check flood policy for this key if flood_policy == "allow": return True, "" else: return False, f"Transport key '{key_name}' flood policy denied" except Exception as e: logger.warning(f"Error checking transport key '{key_name}': {e}") continue # No matching transport code found if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Transport code 0x{transport_code_0:04X} denied (checked {len(transport_keys)} keys)" ) return False, "No matching transport code" except Exception as e: logger.error(f"Transport code validation error: {e}") return False, f"Transport code validation error: {e}" def flood_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]: """Forward a FLOOD packet, appending our hash to the path. INVARIANT: purely synchronous — no await points. The is_duplicate + mark_seen pair is atomic within the asyncio event loop. Do NOT add any await here without revisiting that invariant in __call__ / process_packet. """ # Validate valid, reason = self.validate_packet(packet) if not valid: packet.drop_reason = reason return None # Check if packet is marked do-not-retransmit if packet.is_marked_do_not_retransmit(): # Check if packet has custom drop reason if not packet.drop_reason: packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None # Check unscoped flood policy unscoped_flood_allow = self.config.get("mesh", {}).get( "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True) ) route_type = packet.header & PH_ROUTE_MASK if route_type == ROUTE_TYPE_FLOOD: if not unscoped_flood_allow: packet.drop_reason = DropReason.UNSCOPED_FLOOD_DISABLED return None # Check transport scopes flood policy if route_type == ROUTE_TYPE_TRANSPORT_FLOOD: allowed, check_reason = self._check_transport_codes(packet) if not allowed: packet.drop_reason = DropReason.TRANSPORT_CODE_NOT_ALLOWED return None mode = self._get_loop_detect_mode() if self._is_flood_looped(packet, mode): packet.drop_reason = f"{DropReason.FLOOD_LOOP_DETECTED} ({mode})" return None # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256. if self.is_duplicate(packet, packet_hash=packet_hash): packet.drop_reason = DropReason.DUPLICATE return None if packet.path is None: packet.path = bytearray() elif not isinstance(packet.path, bytearray): packet.path = bytearray(packet.path) hash_size = packet.get_path_hash_size() hop_count = packet.get_path_hash_count() if self.max_flood_hops > 0 and hop_count >= self.max_flood_hops: packet.drop_reason = f"{DropReason.MAX_FLOOD_HOPS} ({hop_count}/{self.max_flood_hops})" return None # path_len encodes hop count in 6 bits (0-63); adding ourselves must not exceed 63 if hop_count >= 63: packet.drop_reason = f"{DropReason.PATH_HOP_COUNT_MAX} (63), cannot append" return None # Check path won't exceed MAX_PATH_SIZE after append if (hop_count + 1) * hash_size > MAX_PATH_SIZE: packet.drop_reason = DropReason.PATH_EXCEEDS_MAX_SIZE return None self.mark_seen(packet, packet_hash=packet_hash) # Append hash_size bytes from our public key prefix packet.path.extend(self.local_hash_bytes[:hash_size]) packet.path_len = PathUtils.encode_path_len(hash_size, hop_count + 1) # This is a transit packet received over RF. Its existing scope decision # is authoritative: a plain flood must remain plain, and a transport # flood must retain the originator's codes when the shared dispatcher # performs its send-time scope resolution. packet._flood_scope_applied = True return packet def direct_forward(self, packet: Packet, packet_hash: Optional[str] = None) -> Optional[Packet]: """Forward a DIRECT packet, removing the first hop from the path. INVARIANT: purely synchronous — no await points. The is_duplicate + mark_seen pair is atomic within the asyncio event loop. Do NOT add any await here without revisiting that invariant in __call__ / process_packet. """ # Validate packet (empty payload, oversized path, etc.) valid, reason = self.validate_packet(packet) if not valid: packet.drop_reason = reason return None # Check if packet is marked do-not-retransmit if packet.is_marked_do_not_retransmit(): if not packet.drop_reason: packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None hash_size = packet.get_path_hash_size() hop_count = packet.get_path_hash_count() # Check if we're the next hop if not packet.path or len(packet.path) < hash_size: packet.drop_reason = DropReason.DIRECT_NO_PATH return None next_hop = bytes(packet.path[:hash_size]) if next_hop != self.local_hash_bytes[:hash_size]: packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256. if self.is_duplicate(packet, packet_hash=packet_hash): packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet, packet_hash=packet_hash) # Remove first hash entry (hash_size bytes) packet.path = bytearray(packet.path[hash_size:]) packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) return packet # Multipart ACKs are spaced ~300ms apart per remaining fragment, matching # MeshCore's forwardMultipartDirect (`(remaining + 1) * 300`). MULTIPART_ACK_SPACING_MS = 300 def forward_multipart_direct( self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None ) -> Optional["ForwardResult"]: """Forward a MULTIPART packet the way MeshCore's forwardMultipartDirect does. MeshCore never relays the multipart wrapper as an ordinary directed packet. Only a multipart-ACK at an intermediate direct hop is relayed, and it is relayed by regenerating the embedded ACK as a plain DIRECT ACK (the wrapper byte stripped) rather than repeating the wrapper. Every other multipart case — flood routing, a final hop, or a non-ACK embedded type — is dropped. Returns a ``ForwardResult`` — ``(ack_packet, delay_seconds)`` plus any multi-ack redundancy ``extras`` — when a regenerated ACK should be sent, otherwise ``None``. INVARIANT: purely synchronous, mirroring flood_forward / direct_forward — the is_duplicate + mark_seen pair must stay atomic within the event loop. """ # MeshCore only forwards multipart on the direct-route branch; the flood # switch case updates ACK state locally but never re-routes it. if not packet.is_route_direct(): packet.drop_reason = f"{DropReason.MULTIPART}: not direct-routed" return None valid, reason = self.validate_packet(packet) if not valid: packet.drop_reason = reason return None if packet.is_marked_do_not_retransmit(): packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT return None payload = packet.payload or b"" remaining = payload[0] >> 4 embedded_type = payload[0] & 0x0F # Only multipart-ACKs are relayed; MeshCore leaves other types for future # use and does not forward them. if embedded_type != PAYLOAD_TYPE_ACK or len(payload) < 5: packet.drop_reason = f"{DropReason.MULTIPART}: unsupported embedded type" return None hash_size = packet.get_path_hash_size() hop_count = packet.get_path_hash_count() # We must be the next hop (a final-hop multipart has no remaining path and # is handled locally, not forwarded). if not packet.path or len(packet.path) < hash_size: packet.drop_reason = DropReason.DIRECT_NO_PATH return None if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None # Dedupe before regenerating, on MeshCore's exact seen key: hasSeen() # runs on a copy that keeps the MULTIPART payload type but carries the # unwrapped ACK payload (the `remaining` count is stripped). Keeping # that key distinct from the plain ACK's is what lets a multi-ack # redundancy pair — MULTIPART copy plus plain ACK — survive every hop # instead of the second half being swallowed as a duplicate. The # pre-computed hash belongs to the wrapper as received, so recompute. seen_key = ( PacketHashingUtils.calculate_packet_hash( PAYLOAD_TYPE_MULTIPART, packet.path_len, bytes(payload[1:]) ) .hex() .upper() ) if self.is_duplicate(packet, packet_hash=seen_key): packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet, packet_hash=seen_key) # Rebuild the received wrapper as the plain DIRECT ACK MeshCore would emit: # drop the multipart header byte and force the ACK payload/route type, then # remove this node from the path (hasSeen()/removeSelfFromPath() order). packet.payload = bytearray(payload[1:]) packet.payload_len = len(packet.payload) packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT packet.transport_codes = [0, 0] packet.path = bytearray(packet.path[hash_size:]) packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) extras, delay_ms = self._multi_ack_extras( packet, float((remaining + 1) * self.MULTIPART_ACK_SPACING_MS), snr ) return ForwardResult(packet, delay_ms / 1000.0, extras) def _multi_ack_extras( self, ack_packet: Packet, base_delay_ms: float, snr: float ) -> Tuple[list, float]: """Build MeshCore's optional multi-ack redundancy copies for a relayed ACK. With ``multi_acks`` enabled, routeDirectRecvAcks precedes the plain ACK with a MULTIPART-wrapped copy (``remaining<<4 | ACK`` prefix byte) on the already self-removed path, pushing the accumulated delay out by a direct retransmit delay + 300 ms per copy; the plain ACK itself slides to the final accumulated delay. Returns ``(extras, plain_ack_delay_ms)``. """ extra = self.multi_acks extras = [] delay_ms = base_delay_ms while extra > 0: delay_ms += ( self._calculate_tx_delay(ack_packet, snr) * 1000.0 + self.MULTIPART_ACK_SPACING_MS ) wrapped = Packet() wrapped.header = (PAYLOAD_TYPE_MULTIPART << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT wrapped.payload = bytearray( bytes([(extra << 4) | PAYLOAD_TYPE_ACK]) + bytes(ack_packet.payload) ) wrapped.payload_len = len(wrapped.payload) wrapped.path = bytearray(ack_packet.path) wrapped.path_len = ack_packet.path_len extras.append((wrapped, delay_ms / 1000.0)) extra -= 1 return extras, delay_ms def forward_routed_ack( self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None ) -> Optional["ForwardResult"]: """Relay a routed ACK the way MeshCore's routeDirectRecvAcks does. An ACK at an intermediate direct hop is not repeated verbatim: MeshCore regenerates it (createAck) as a plain DIRECT ACK — transport codes dropped, every other header bit cleared — removes this node from the path, and transmits it with zero retransmit delay rather than the generic direct forwarding delay. With ``multi_acks`` enabled, the plain ACK is preceded by a MULTIPART-wrapped redundancy copy and both slide out by a direct retransmit delay + 300 ms. INVARIANT: purely synchronous, mirroring direct_forward — the is_duplicate + mark_seen pair must stay atomic within the event loop. """ valid, reason = self.validate_packet(packet) if not valid: packet.drop_reason = reason return None if packet.is_marked_do_not_retransmit(): packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT return None hash_size = packet.get_path_hash_size() hop_count = packet.get_path_hash_count() # A final-hop ACK (no remaining path) is consumed locally, never relayed. if not packet.path or len(packet.path) < hash_size: packet.drop_reason = DropReason.DIRECT_NO_PATH return None if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None # The packet hash covers only the payload type and payload, so the # received form and the regenerated plain ACK share one key — matching # MeshCore's hasSeen() on the incoming ACK before removeSelfFromPath(). if self.is_duplicate(packet, packet_hash=packet_hash): packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet, packet_hash=packet_hash) packet.header = (PAYLOAD_TYPE_ACK << PH_TYPE_SHIFT) | ROUTE_TYPE_DIRECT packet.transport_codes = [0, 0] packet.path = bytearray(packet.path[hash_size:]) packet.path_len = PathUtils.encode_path_len(hash_size, hop_count - 1) extras, delay_ms = self._multi_ack_extras(packet, 0.0, snr) return ForwardResult(packet, delay_ms / 1000.0, extras) @staticmethod def calculate_packet_score(snr: float, packet_len: int, spreading_factor: int = 8) -> float: """Reception-quality score in [0, 1] via the shared core scorer (MeshCore RadioLibWrappers packetScoreInt).""" return packet_score(snr, spreading_factor, packet_len) def _calculate_tx_delay(self, packet: Packet, snr: float = 0.0) -> float: packet_len = packet.get_raw_length() airtime_ms = self.airtime_mgr.calculate_airtime(packet_len) if packet.is_route_flood(): delay_factor = self.tx_delay_factor elif packet.is_route_direct(): delay_factor = self.direct_tx_delay_factor else: delay_factor = self.direct_tx_delay_factor # MeshCore uses the same random window for flood and direct variants: # choose a value from 0 through 5 * airtime * factor. base_delay_ms = airtime_ms * delay_factor random_mult = secrets.randbelow(5001) / 1000.0 delay_s = (base_delay_ms * random_mult) / 1000.0 # OpenHop's optional score gate is applied only after the 50 ms threshold. if delay_s >= 0.05 and self.use_score_for_tx: score = self.calculate_packet_score(snr, packet_len) # Higher score = shorter delay: max(0.2, 1.0 - score) # score 1.0 → multiplier 0.2 (20% of original) # score 0.0 → multiplier 1.0 (100% of original) score_multiplier = max(0.2, 1.0 - score) delay_s = delay_s * score_multiplier if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Congestion detected (delay >= 50ms), score={score:.2f}, " f"delay multiplier={score_multiplier:.2f}" ) # Cap at 5 seconds maximum delay_s = min(delay_s, 5.0) if logger.isEnabledFor(logging.DEBUG): logger.debug( f"Route={'FLOOD' if packet.is_route_flood() else 'DIRECT'}, " f"len={packet_len}B, airtime={airtime_ms:.1f}ms, delay={delay_s:.3f}s" ) return delay_s def process_packet( self, packet: Packet, snr: float = 0.0, packet_hash: Optional[str] = None, ) -> Optional[Tuple[Packet, float]]: """Route a received packet to flood_forward or direct_forward. packet_hash is the pre-computed SHA-256 hex string from __call__. Passing it here avoids recomputing the hash in flood_forward / direct_forward / is_duplicate / mark_seen — reducing SHA-256 calls from 3 per forwarded packet to 1. """ route_type = packet.header & PH_ROUTE_MASK # MeshCore routes multipart traffic through its own branch, never the # generic flood/direct path — it regenerates embedded ACKs and drops # everything else. Its delay is fixed by the fragment count, so return # the (packet, delay) pair directly instead of _calculate_tx_delay. payload_type = packet.get_payload_type() if payload_type == PAYLOAD_TYPE_MULTIPART: return self.forward_multipart_direct(packet, snr, packet_hash=packet_hash) # Routed ACKs also get their own MeshCore branch: an intermediate direct # hop re-emits the ACK immediately (routeDirectRecvAcks) instead of the # delayed generic direct forward. Flood ACKs stay on the generic path. if payload_type == PAYLOAD_TYPE_ACK and route_type in ( ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT, ): return self.forward_routed_ack(packet, snr, packet_hash=packet_hash) if route_type == ROUTE_TYPE_FLOOD or route_type == ROUTE_TYPE_TRANSPORT_FLOOD: fwd_pkt = self.flood_forward(packet, packet_hash=packet_hash) if fwd_pkt is None: return None delay = self._calculate_tx_delay(fwd_pkt, snr) return fwd_pkt, delay elif route_type == ROUTE_TYPE_DIRECT or route_type == ROUTE_TYPE_TRANSPORT_DIRECT: fwd_pkt = self.direct_forward(packet, packet_hash=packet_hash) if fwd_pkt is None: return None delay = self._calculate_tx_delay(fwd_pkt, snr) return fwd_pkt, delay else: packet.drop_reason = f"Unknown route type: {route_type}" return None # A local transmit whose radio link has dropped waits this long for it to come # back before giving up, polling at LOCAL_TX_LINK_POLL_S. Kept well inside the # companion client's own command timeout (meshcore_py: 15s), which is already # partly spent on the TX delay by the time we get here. DEFAULT_LOCAL_TX_LINK_WAIT_S = 5.0 # Hard ceiling regardless of configuration: the companion client is already # waiting on this send, and an unbounded (or infinite) value would hold its # command open forever rather than reporting a failure it can act on. MAX_LOCAL_TX_LINK_WAIT_S = 30.0 LOCAL_TX_LINK_POLL_S = 0.1 def _local_tx_link_wait_s(self) -> float: """Configured ceiling for waiting out a radio-link outage on a local TX.""" delays = self.config.get("delays") or {} try: wait = float( delays.get("local_tx_link_wait_seconds", self.DEFAULT_LOCAL_TX_LINK_WAIT_S) ) except (TypeError, ValueError): return self.DEFAULT_LOCAL_TX_LINK_WAIT_S if not math.isfinite(wait): logger.warning( "local_tx_link_wait_seconds=%r is not a finite number; using %.1fs", wait, self.DEFAULT_LOCAL_TX_LINK_WAIT_S, ) return self.DEFAULT_LOCAL_TX_LINK_WAIT_S return min(max(0.0, wait), self.MAX_LOCAL_TX_LINK_WAIT_S) def _resolve_tx_radio(self, radio_id: Optional[str] = None): """The radio a send will actually use, or None when that cannot be told. Attribute access on a FabricRadio passes through to its *default* endpoint (rf_fabric/fabric.py), so reading link state off it while the send targets another radio_id answers about the wrong radio -- suppressing a healthy one, or waving through a dead one. Resolve the named endpoint instead, and return None rather than guess: no answer leaves the send to proceed as before. """ radio = getattr(self.dispatcher, "radio", None) if radio is None: return None fabric = getattr(radio, "fabric", None) if fabric is None and hasattr(radio, "get_radio"): fabric = radio # an RFFabric handed to the dispatcher directly radios = getattr(fabric, "radios", None) if fabric is not None else None # isinstance, not truthiness: RFFabric.radios is an OrderedDict, so this # recognises a real multi-endpoint fabric without mistaking a stand-in # attribute on a single-radio wrapper for one. if not isinstance(radios, Mapping) or len(radios) <= 1: return radio # one radio, and it is the one that transmits if not radio_id: # The fabric picks its endpoint at send time, so there is nothing here # to inspect. Say so rather than answer about whichever radio happens # to be the default -- a wrong "down" would suppress a healthy send. return None getter = getattr(fabric, "get_radio", None) if getter is None: return None try: return getter(str(radio_id)) except Exception: return None def _radio_link_down(self, radio_id: Optional[str] = None) -> bool: """True when the radio this send will use reports its link unusable. Only radios that track a link state answer this -- a KISS modem over serial does, a directly-driven SX1262 has none to lose -- so it reads False for the rest and leaves their timing untouched. """ radio = self._resolve_tx_radio(radio_id) if radio is None: return False # Identity checks, not truthiness: a radio that does not track link state # answers with something that is neither True nor False (or nothing at all), # and that has to read as "no opinion" rather than "down". if getattr(radio, "is_connected", None) is False: return True return getattr(radio, "is_degraded", None) is True async def _await_radio_link(self, timeout: float, radio_id: Optional[str] = None) -> bool: """Wait up to *timeout* seconds for the radio link, True if it is usable.""" if not self._radio_link_down(radio_id): return True if timeout <= 0: return False logger.info("Local TX: radio link down, waiting up to %.1fs for it", timeout) deadline = time.monotonic() + timeout while time.monotonic() < deadline: await asyncio.sleep(self.LOCAL_TX_LINK_POLL_S) if not self._radio_link_down(radio_id): logger.info("Local TX: radio link back, transmitting now") return True logger.warning("Local TX: radio link still down after %.1fs, giving up", timeout) return False async def schedule_retransmit( self, fwd_pkt: Packet, delay: float, airtime_ms: float = 0.0, local_transmission: bool = False, preferred_tx_radio_id: Optional[str] = None, ): """Schedule a packet retransmission with delay and return the task. If local_transmission is True and the first send fails, retry once after a short delay (handles transient radio/LBT failures). """ async def delayed_send(): await asyncio.sleep(delay) # Each attempt gets its own lock acquisition so the 1-second retry # backoff (local_transmission only) happens OUTSIDE the lock. # Holding _tx_lock across asyncio.sleep(1.0) would block every other # queued TX task for the full backoff period. # # Loop runs once for relayed packets, twice for local_transmission: # attempt 0 — initial try (no pre-sleep) # attempt 1 — retry after 1s backoff outside the lock for attempt in range(2 if local_transmission else 1): if attempt > 0 and not self._radio_link_down(preferred_tx_radio_id): # A transient failure with the link still up: back off OUTSIDE # the lock — other tasks can transmit here. logger.info("Retrying local TX in 1s (lock released during backoff)...") await asyncio.sleep(1.0) # A dropped link is not a transient failure to back off from. A KISS # modem that falls off USB re-enumerates and re-handshakes in ~2s, so # the flat backoff above would land inside the outage and spend the # retry on a link that is still down — which is exactly what turned # one modem hiccup into a failed message. Wait for the link instead. if local_transmission and not await self._await_radio_link( self._local_tx_link_wait_s(), preferred_tx_radio_id ): return False async with self._tx_lock: # ── Authoritative duty-cycle gate ────────────────────────── # The upfront can_transmit() call in __call__ is advisory: it # avoids scheduling packets obviously over budget, but cannot # prevent a race between tasks whose delay timers expire nearly # simultaneously. Both pass the advisory check before either # records airtime, then both attempt to transmit. # # Inside _tx_lock only one task runs at a time. The check and # record_tx() are effectively atomic — no TOCTOU window. # Re-checked every attempt because airtime state may change # while we wait for the lock or sleep through backoff. # Charged to the radio this egress goes out on, at that # radio's modulation: the airtime the caller computed is the # default radio's, which on a bridge is the wrong channel and # can be out by the ratio of their bandwidths. # # A named radio is gated on its own budget. Without one the # fabric picks inside send(), so every radio it could pick # has to have the budget -- a gate that passes because some # other channel is quiet is a gate that lets this one # transmit past its legal limit. if airtime_ms > 0: gate_ids = ( (preferred_tx_radio_id,) if preferred_tx_radio_id else self._egress_candidates() ) for gate_id in gate_ids: manager = self.airtime_budgets.for_radio(gate_id) try: gate_ms = self._egress_radio_airtime(fwd_pkt, gate_id) except Exception as exc: logger.debug(f"Falling back to the caller's airtime: {exc}") gate_ms = airtime_ms can_tx_now, _ = manager.can_transmit(gate_ms) if not can_tx_now: logger.warning( "Packet dropped at TX time: duty-cycle exceeded on radio %s " "(airtime=%.1fms)", gate_id or "default", gate_ms, ) return False # Registered as the send starts: another radio of this node can # hear it while it is still on the air. own_send = self._note_own_send_started(fwd_pkt, preferred_tx_radio_id) try: if preferred_tx_radio_id: try: sent = await self.dispatcher.send_packet( fwd_pkt, wait_for_ack=False, radio_id=preferred_tx_radio_id, ) except TypeError: sent = await self.dispatcher.send_packet( fwd_pkt, wait_for_ack=False ) else: sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) self._note_own_send_finished(own_send, fwd_pkt, sent=bool(sent)) if not sent: logger.warning( "Retransmit failed (attempt %d): dispatcher returned false", attempt + 1, ) if local_transmission and attempt == 0: continue return False self._record_packet_sent(fwd_pkt) charge_ms = airtime_ms if airtime_ms > 0: # Debited under the same lock the gate ran under, so # there is no window between deciding and recording. # Charged to the radio the send says carried it, # which is the only way to meter a fabric that chose # for itself -- and still the right answer when the # radio was named, since it is the radio confirming. try: charge_ms = self._charge_egress(fwd_pkt, preferred_tx_radio_id) except Exception as exc: logger.debug(f"Falling back to the caller's airtime: {exc}") self.airtime_budgets.for_radio(preferred_tx_radio_id).record_tx( airtime_ms ) packet_size = fwd_pkt.get_raw_length() logger.info( f"Retransmitted packet ({packet_size} bytes, {charge_ms:.1f}ms airtime)" ) return True except Exception as e: self._note_own_send_finished(own_send, fwd_pkt, sent=False) logger.error(f"Retransmit failed (attempt {attempt + 1}): {e}") if local_transmission and attempt == 0: pass # release lock, outer loop sleeps, then retries else: raise return False return asyncio.create_task(delayed_send()) async def schedule_retransmit_fanout( self, fwd_pkt: Packet, delay: float, airtime_ms: float, radio_ids, *, local_transmission: bool = False, ): """Schedule one logical packet as one physical TX per radio; return the task. The task resolves to a ``FanoutTxResult``. Each egress runs through ``schedule_retransmit`` — duty-cycle gate, TX lock, local retry and per-radio link wait — so the sends stay serialised and one radio's failure never stops another's. The first egress transmits ``fwd_pkt`` itself, so TX-time side effects (flood scope, ``_tx_metadata``) land where callers already look. Every other egress gets its own copy, so one send's metadata never overwrites another's. """ radio_ids = tuple(radio_ids) self._freeze_tx_normalisation(fwd_pkt) packets = [fwd_pkt] + [copy.deepcopy(fwd_pkt) for _ in radio_ids[1:]] async def run() -> FanoutTxResult: await asyncio.sleep(delay) # One shared delay, then the per-egress sends are created in order with # no further sleep. When every radio is ready they reach the TX lock in # egress order (asyncio runs ready tasks and grants the lock FIFO), so # the primary egress gets the first claim on the duty-cycle budget. A # local send whose radio link is down waits outside the lock, so a # healthy radio later in the order is not held up by it and may go first. tasks = [ await self.schedule_retransmit( pkt, 0.0, airtime_ms, local_transmission=local_transmission, preferred_tx_radio_id=radio_id, ) for pkt, radio_id in zip(packets, radio_ids) ] outcomes = await asyncio.gather(*tasks, return_exceptions=True) results = [] for radio_id, pkt, outcome in zip(radio_ids, packets, outcomes): if isinstance(outcome, BaseException): logger.warning("Fan-out TX via %s failed: %s", radio_id, outcome) results.append(RadioTxResult(radio_id, False, error=outcome)) continue # Read whether or not the send succeeded: a failed egress carries # the LBT figures that explain why, and the single-egress path in # _await_egress_tx has always read them unconditionally. metadata = getattr(pkt, "_tx_metadata", None) results.append( RadioTxResult( radio_id, bool(outcome), metadata=metadata if isinstance(metadata, dict) else None, ) ) return FanoutTxResult(results) return asyncio.create_task(run()) def _freeze_tx_normalisation(self, pkt: Packet) -> None: """Apply the dispatcher's send-time normalisation once and mark it decided. The dispatcher resolves flood scope and the default path-hash mode at send time. Fan-out sends the same logical packet several times, possibly seconds apart (link wait, retry backoff), so a live change to the node's default region or hash mode in between would put different bytes on each radio. Deciding both on the logical packet before it is copied keeps every egress, and the packet the router echoes, identical. """ for hook in ("_apply_flood_scope", "_apply_default_path_hash_mode"): apply = getattr(self.dispatcher, hook, None) if callable(apply): apply(pkt) pkt._flood_scope_applied = True pkt._path_hash_mode_applied = True async def _schedule_egress_tx( self, fwd_pkt: Packet, delay: float, airtime_ms: float, tx_radio_ids: Optional[Tuple[str, ...]], *, local_transmission: bool = False, ): """Schedule a TX over the resolved egress set. One egress (or none, leaving the choice to the fabric) keeps the plain ``schedule_retransmit`` path, whose task resolves to a bool and raises on a relay send error; more than one fans out. """ if tx_radio_ids and len(tx_radio_ids) > 1: return await self.schedule_retransmit_fanout( fwd_pkt, delay, airtime_ms, tx_radio_ids, local_transmission=local_transmission, ) return await self.schedule_retransmit( fwd_pkt, delay, airtime_ms, local_transmission=local_transmission, preferred_tx_radio_id=tx_radio_ids[0] if tx_radio_ids else None, ) def _egress_records(self, tx_result: FanoutTxResult) -> Optional[list]: """Per-radio egress rows for storage, or None on a node with one radio. With one radio the packet row already says everything these would. A bridge is different: one packet goes out on radios whose LBT outcomes differ, only the primary egress reaches the packet row, and a send that failed leaves no trace there at all. A failed egress carries whatever LBT metadata the driver reported. Where the radio raises instead of returning any (an SX1262 that gives up on a busy channel), the row records the failure with no attempts rather than inventing them: the driver has to report them before anything here can. """ _, radio_ids = self._fabric_endpoints() if not radio_ids or len(radio_ids) < 2: return None records = [] for result in tx_result.results: metadata = result.metadata if isinstance(result.metadata, dict) else {} backoffs = metadata.get("lbt_backoff_delays_ms") or [] records.append( { "radio_id": result.radio_id, "success": bool(result.success), "lbt_attempts": int(metadata.get("lbt_attempts", 0) or 0), "lbt_backoff_ms_total": float(sum(backoffs)), "lbt_channel_busy": bool(metadata.get("lbt_channel_busy", False)), } ) return records or None @staticmethod async def _await_egress_tx(tx_task, fwd_pkt: Packet) -> FanoutTxResult: """Await a task from ``_schedule_egress_tx`` as a ``FanoutTxResult``. A single-egress send error propagates, exactly as it did before fan-out. """ outcome = await tx_task if isinstance(outcome, FanoutTxResult): return outcome metadata = getattr(fwd_pkt, "_tx_metadata", None) metadata = metadata if isinstance(metadata, dict) else None radio_id = (metadata.get("radio_id") or metadata.get("tx_radio_id")) if metadata else None return FanoutTxResult([RadioTxResult(radio_id, bool(outcome), metadata=metadata)]) @staticmethod def _rx_radio_id_of(packet: Packet, metadata: Optional[dict]) -> Optional[str]: """The radio that heard this packet, however the caller carried the id.""" if metadata: radio_id = metadata.get("rx_radio_id") if radio_id: return radio_id return getattr(packet, "_rx_radio_id", None) or getattr(packet, "rx_radio_id", None) def _egress_radio_airtime(self, packet: Packet, radio_id: Optional[str]) -> float: """Time this packet will occupy the channel of the radio sending it. On a bridge the same bytes are a different length of transmission on each radio -- eight times over between 500 and 62.5 kHz -- so the figure the duty cycle is charged has to be the one for the radio that transmits. """ return self.airtime_budgets.for_radio(radio_id).calculate_airtime(packet.get_raw_length()) def airtime_stats(self) -> dict: """Node-wide airtime figures, in the shape AirtimeManager.get_stats returns. The single place for callers that report one set of numbers for the node. Reading ``airtime_mgr`` instead describes the default radio's channel only, which on a bridge is a quiet halving of figures people have been watching for months. """ return self.airtime_budgets.node_stats() def _egress_can_transmit( self, packet: Packet, tx_radio_ids: Optional[Tuple[str, ...]] ) -> Tuple[bool, float]: """Advisory gate before scheduling: is there budget for this forward? With a named egress set, each radio is sent to separately and re-gated under the TX lock, so one radio being over budget is no reason to drop a forward another can carry -- one radio with budget is enough, and the wait reported is the shortest of them. With no named set the fabric picks inside send() and the node cannot know which channel will carry the packet. Then every radio it could pick has to have the budget, and the wait is the longest of them. Asking whether *any* candidate has budget is exactly how a node transmits past a legal limit: it only has to guess the quiet channel. """ if tx_radio_ids: waits = [] for radio_id in tx_radio_ids: manager = self.airtime_budgets.for_radio(radio_id) can_tx, wait_time = manager.can_transmit( self._egress_radio_airtime(packet, radio_id) ) if can_tx: return True, 0.0 waits.append(wait_time) return False, min(waits) if waits else 1.0 waits = [] for radio_id in self._egress_candidates(): manager = self.airtime_budgets.for_radio(radio_id) can_tx, wait_time = manager.can_transmit(self._egress_radio_airtime(packet, radio_id)) if not can_tx: waits.append(wait_time) if waits: return False, max(waits) return True, 0.0 def _charge_egress(self, packet: Packet, preferred_tx_radio_id: Optional[str]) -> float: """Debit the radio that actually carried the packet, and say how much. The metadata a send returns names the endpoint that transmitted, and that is the authoritative answer: where the fabric chose inside send() it is the only way to find out, and where the radio was named explicitly it is the radio confirming it. Falls back to the radio that was asked for, and from there to the default budget, so a send always lands somewhere rather than going unmetered. """ metadata = getattr(packet, "_tx_metadata", None) radio_id = None if isinstance(metadata, dict): radio_id = metadata.get("radio_id") or metadata.get("tx_radio_id") radio_id = str(radio_id) if radio_id else preferred_tx_radio_id manager = self.airtime_budgets.for_radio(radio_id) charge_ms = self._egress_radio_airtime(packet, radio_id) manager.record_tx(charge_ms) return charge_ms def _planned_egress_radio_id(self, packet: Packet, rx_radio_id: Optional[str]) -> Optional[str]: """Ask the fabric which radio will carry this packet, before it is sent. Answerable only on a core that passes the ingress radio to the TX selector. There the choice is a function of the arguments, so the answer now is the answer send() reaches, and naming it changes nothing about which radio transmits -- it only lets the duty cycle be charged to the channel that carries the packet, and closes the window where a busy node's next arrival moves a reply onto the other band between the retransmit delay and the send. Returns None on a node with fewer than two radios, where there is nothing to choose, and against an older core, where the fabric picks inside send() and no caller can know in advance what it will pick. The bytes handed to the selector are this packet as it stands, a moment before the dispatcher applies flood scope and path-hash normalisation, so a selector that routed on payload content could in principle see a slightly different frame than the one sent. The tx_mode selectors route on the ingress radio and ignore the bytes entirely. """ fabric, ids = self._fabric_endpoints() if fabric is None or not ids or len(ids) < 2: return None if not fabric_selects_by_ingress_radio(fabric): return None try: raw = packet.write_to() except Exception as exc: logger.debug(f"Could not serialise a packet to resolve its egress radio: {exc}") return None try: radio_id = fabric.resolve_tx_radio_id(raw, None, rx_radio_id=rx_radio_id) except Exception as exc: logger.debug(f"Fabric could not name an egress radio: {exc}") return None return str(radio_id) if radio_id else None def _egress_candidates(self) -> Tuple[Optional[str], ...]: """Every radio the fabric could pick when nobody named one. The honest answer on a core whose selector sees only the frame bytes: it reads the node's most recent RX inside send(), so sticky and bridge can both come out on any radio. tx_mode default always picks default_radio, so that one is still exact. ``(None,)`` on a single-radio node, which resolves to the one budget it has always had. """ fabric, ids = self._fabric_endpoints() if fabric is None or not ids or len(ids) < 2: return (None,) mode = str(self._fabric_cfg().get("tx_mode", "default")).strip().lower() if mode in ("", "default"): default_id = getattr(fabric, "default_radio_id", None) if isinstance(default_id, str) and default_id in ids: return (default_id,) return (ids[0],) return tuple(ids) def _fabric_cfg(self) -> dict: fabric_cfg = self.config.get("fabric") return fabric_cfg if isinstance(fabric_cfg, dict) else {} def _fabric_endpoints(self): """Return ``(fabric, radio_ids)`` for a multi-radio fabric, else ``(None, None)``.""" radio = getattr(self.dispatcher, "radio", None) fabric = getattr(radio, "fabric", None) radios = getattr(fabric, "radios", None) # isinstance, not truthiness: RFFabric.radios is an OrderedDict, and a # stand-in attribute on a non-fabric radio must not read as endpoints. if not isinstance(radios, Mapping): return None, None return fabric, [str(rid) for rid in radios.keys()] def _resolve_relay_tx_radio_ids(self, rx_radio_id: Optional[str]) -> Optional[Tuple[str, ...]]: """Egress radios for a packet heard on ``rx_radio_id`` and being repeated. Applies only when fabric.tx_mode=bridge and exactly two radios are registered: the bridge target (the other radio) comes first, followed by the ingress radio itself when fabric.repeat_on_ingress is true. For all other layouts return None and let the dispatcher/fabric choose via existing logic. """ fabric_cfg = self._fabric_cfg() if str(fabric_cfg.get("tx_mode", "default")).strip().lower() != "bridge": return None if not rx_radio_id: return None _, ids = self._fabric_endpoints() if ids is None or len(ids) != 2: return None rx_id = str(rx_radio_id) if rx_id not in ids: return None bridge_id = ids[1] if ids[0] == rx_id else ids[0] if fabric_cfg.get("repeat_on_ingress") is True: return (bridge_id, rx_id) return (bridge_id,) def _resolve_bridge_pair_tx_radio(self, rx_radio_id: Optional[str]) -> Optional[str]: """Return deterministic pair TX radio for bridge mode (A->B, B->A).""" radio_ids = self._resolve_relay_tx_radio_ids(rx_radio_id) return radio_ids[0] if radio_ids else None def _resolve_origin_tx_radio_ids(self) -> Optional[Tuple[str, ...]]: """Egress radios for a packet this node originates. With fabric.origin_tx=all and exactly two radios, every radio, the default radio first. Otherwise None: the existing fabric selection (default radio / tx_mode selector) stays authoritative. """ if fabric_origin_tx(self._fabric_cfg()) != "all": return None fabric, ids = self._fabric_endpoints() if ids is None or len(ids) != 2: return None default_id = getattr(fabric, "default_radio_id", None) if isinstance(default_id, str) and default_id in ids: ids.remove(default_id) ids.insert(0, default_id) return tuple(ids) def _record_packet_sent(self, packet: Packet) -> None: """Record a packet send for flood/direct stats (forwarded and originated).""" route = getattr(packet, "header", 0) & PH_ROUTE_MASK if route in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.sent_flood_count += 1 elif route in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT): self.sent_direct_count += 1 @staticmethod def _read_noise_floor(radio) -> Optional[float]: """Read one radio's noise floor, or None when it cannot answer.""" try: if radio is not None and hasattr(radio, "get_noise_floor"): return radio.get_noise_floor() return None except Exception as e: logger.debug(f"Failed to get noise floor: {e}") return None def get_noise_floor(self) -> Optional[float]: """Noise floor of the default radio, which is what /stats reports.""" return self._read_noise_floor(self.dispatcher.radio if self.dispatcher else None) def _sampling_radios(self) -> list: """``(radio_id, radio)`` for every radio to sample, the default radio first. A single-radio node yields one entry whose id is None: one radio needs no label, so its stored rows and published records stay exactly as they were. The default radio comes first because callers publish only the first entry's sample — /stats, the MQTT status message and companion stats all read the default radio's figures. """ fabric, ids = self._fabric_endpoints() default_radio = self.dispatcher.radio if self.dispatcher else None if fabric is None or not ids or len(ids) < 2: return [(None, default_radio)] default_id = getattr(fabric, "default_radio_id", None) if isinstance(default_id, str) and default_id in ids: ids.remove(default_id) ids.insert(0, default_id) getter = getattr(fabric, "get_radio", None) radios = [] for radio_id in ids: radio = None if getter is not None: try: radio = getter(radio_id) except Exception as e: logger.debug(f"No endpoint for radio {radio_id}: {e}") radios.append((radio_id, radio)) return radios def get_cached_noise_floor(self) -> Optional[float]: """Return the last asynchronously-sampled noise floor value.""" return self._cached_noise_floor def get_stats(self) -> dict: runtime_config = normalize_modem_config(self.config, warn=False) redact_modem_tokens_in_place(runtime_config) uptime_seconds = time.time() - self.start_time # Get config sections repeater_config = self.config.get("repeater", {}) duty_cycle_config = self.config.get("duty_cycle", {}) delays_config = self.config.get("delays", {}) max_airtime_ms = duty_cycle_config.get("max_airtime_per_minute", 3600) max_duty_cycle_percent = (max_airtime_ms / 60000) * 100 # 60000ms = 1 minute # Calculate actual hourly rates (packets in last 3600 seconds) now = time.time() packets_last_hour = [p for p in self.recent_packets if now - p["timestamp"] < 3600] rx_per_hour = len(packets_last_hour) forwarded_per_hour = sum(1 for p in packets_last_hour if p.get("transmitted", False)) # Use cached value sampled by the background timer to avoid serial I/O on stats requests. noise_floor_dbm = self.get_cached_noise_floor() # Get CRC error count from radio hardware radio = self.dispatcher.radio if self.dispatcher else None crc_error_count = getattr(radio, "crc_error_count", 0) if radio else 0 # Get neighbors from database neighbors = self.storage.get_neighbors() if self.storage else {} # Format local_hash respecting path_hash_mode phm = self.config.get("mesh", {}).get("path_hash_mode", 0) _bc = {0: 1, 1: 2, 2: 3}.get(phm, 1) _hc = _bc * 2 _val = int.from_bytes(bytes(self.local_hash_bytes[:_bc]), "big") local_hash_str = f"0x{_val:0{_hc}x}" stats = { "local_hash": local_hash_str, "duplicate_cache_size": len(self.seen_packets), "cache_ttl": self.cache_ttl, "rx_count": self.rx_count, "forwarded_count": self.forwarded_count, "dropped_count": self.dropped_count, "recv_flood_count": self.recv_flood_count, "recv_direct_count": self.recv_direct_count, "sent_flood_count": self.sent_flood_count, "sent_direct_count": self.sent_direct_count, "flood_dup_count": self.flood_dup_count, "direct_dup_count": self.direct_dup_count, "rx_per_hour": rx_per_hour, "forwarded_per_hour": forwarded_per_hour, "recent_packets": list(self.recent_packets), "neighbors": neighbors, "uptime_seconds": uptime_seconds, "noise_floor_dbm": noise_floor_dbm, "crc_error_count": crc_error_count, # Add configuration data "config": { "node_name": repeater_config.get("node_name", "Unknown"), "repeater": { "mode": repeater_config.get("mode", "forward"), "use_score_for_tx": repeater_config.get("use_score_for_tx", False), "score_threshold": repeater_config.get("score_threshold", 0.3), "send_advert_interval_hours": repeater_config.get( "send_advert_interval_hours", 10 ), "direct_advert_interval_hours": repeater_config.get( "direct_advert_interval_hours", 0 ), "latitude": repeater_config.get("latitude", 0.0), "longitude": repeater_config.get("longitude", 0.0), "max_flood_hops": repeater_config.get("max_flood_hops", 64), "advert_interval_minutes": repeater_config.get("advert_interval_minutes", 120), "advert_rate_limit": repeater_config.get("advert_rate_limit", {}), "advert_penalty_box": repeater_config.get("advert_penalty_box", {}), "advert_adaptive": repeater_config.get("advert_adaptive", {}), }, "radio": self.config.get( "radio", {} ), # Read from live config, not cached radio_config "radios": runtime_config.get("radios") or [], "fabric": self.config.get("fabric") or {}, "radio_type": runtime_config.get("radio_type"), "sx1262": self.config.get("sx1262", {}), "ch341": self.config.get("ch341", {}), "kiss": self.config.get("kiss", {}), "modem_usb": runtime_config.get("modem_usb", {}), "modem_tcp": runtime_config.get("modem_tcp", {}), "duty_cycle": { "max_airtime_percent": max_duty_cycle_percent, "enforcement_enabled": duty_cycle_config.get("enforcement_enabled", True), }, "delays": { "tx_delay_factor": delays_config.get("tx_delay_factor", 1.0), "direct_tx_delay_factor": delays_config.get("direct_tx_delay_factor", 0.5), "rx_delay_base": delays_config.get("rx_delay_base", 0.0), }, "web": self.config.get("web", {}), # Include web configuration "mesh": { "loop_detect": self.config.get("mesh", {}).get("loop_detect", "off"), "unscoped_flood_allow": self.config.get("mesh", {}).get( "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True), ), "path_hash_mode": self.config.get("mesh", {}).get("path_hash_mode", 0), }, "mqtt_brokers": self.config.get("mqtt_brokers", {}), }, "public_key": None, } # Add airtime stats # The node-wide figures describe the whole node: totals summed across # channels, utilisation the highest of them. Each channel's own budget # is reported beside them, because on a bridge they are different # limits, differently spent. On a single-radio node every figure is the # one manager's, unchanged. stats.update(self.airtime_budgets.node_stats()) radios = self.airtime_budgets.per_radio_stats() if radios: stats["airtime_radios"] = radios return stats def _start_background_tasks(self): if self._background_task is None: self._background_task = asyncio.create_task(self._background_timer_loop()) logger.info("Background timer started for noise floor and adverts") async def _background_timer_loop(self): try: while True: current_time = time.time() # Check noise floor recording (every 30 seconds) if current_time - self.last_noise_measurement >= self.noise_floor_interval: await self._record_noise_floor_async() await self._record_crc_errors_async() self.last_noise_measurement = current_time # Check advert scheduling. # Prefer daemon-managed scheduling so repeater + room-server # intervals can share one loop and stagger transmissions. if self.periodic_advert_tick_func: await self.periodic_advert_tick_func() elif self.send_advert_interval_hours > 0 and self.send_advert_func: interval_seconds = self.send_advert_interval_hours * 3600 if current_time - self.last_advert_time >= interval_seconds: await self._send_periodic_advert_async() self.last_advert_time = current_time # Prune expired entries from duplicate detection cache (every 60s) if current_time - self.last_cache_cleanup >= 60.0: self.cleanup_cache() self.last_cache_cleanup = current_time # Prune old SQLite data (check every 6 hours) if current_time - self.last_db_cleanup >= 21600: if self.storage: try: retention_cfg = self.config.get("storage", {}).get("retention", {}) retention_days = retention_cfg.get("sqlite_cleanup_days", 31) companion_events_days = retention_cfg.get("companion_events_days", 31) self.storage.cleanup_old_data( days=retention_days, companion_events_days=companion_events_days, ) logger.info("Cleaned up SQLite data older than %d days", retention_days) except Exception as e: logger.warning(f"SQLite cleanup failed: {e}") self.last_db_cleanup = current_time # Sleep for 5 seconds before next check await asyncio.sleep(5.0) except asyncio.CancelledError: logger.info("Background timer loop cancelled") raise except Exception as e: logger.error(f"Error in background timer loop: {e}") # Restart the timer after a delay await asyncio.sleep(30) self._background_task = asyncio.create_task(self._background_timer_loop()) async def _record_noise_floor_async(self): """Sample every radio's noise floor and persist it against that radio. One radio at a time: a KISS modem's read blocks for up to its response timeout, and reading both at once would put two of those in flight while the radios are also carrying traffic. """ if not self.storage: return loop = asyncio.get_running_loop() for index, (radio_id, radio) in enumerate(self._sampling_radios()): is_default = index == 0 try: # Run in executor so KISS modem's blocking _send_command (up to 5s timeout) # does not block the event loop and hang the process / delay Ctrl+C. noise_floor = await loop.run_in_executor( None, self._read_noise_floor_for, radio_id, radio ) if noise_floor is None: logger.debug("Unable to read noise floor from radio %s", radio_id or "default") continue if is_default: self._cached_noise_floor = noise_floor self.storage.record_noise_floor(noise_floor, radio_id, publish=is_default) logger.debug(f"Recorded noise floor: {noise_floor} dBm ({radio_id or 'default'})") except Exception as e: logger.error(f"Error recording noise floor: {e}") def _read_noise_floor_for(self, radio_id: Optional[str], radio) -> Optional[float]: """Read the noise floor of one sampled radio. The default radio goes through ``get_noise_floor`` so everything that reports it reads the same path. """ if radio_id is None: return self.get_noise_floor() return self._read_noise_floor(radio) async def _record_crc_errors_async(self): """Persist each radio's CRC error delta from its hardware counter. Baselines are per radio: a shared one would charge every radio's errors to whichever was sampled last and report swings that never happened. """ if not self.storage: return for index, (radio_id, radio) in enumerate(self._sampling_radios()): try: current = int(getattr(radio, "crc_error_count", 0) or 0) if radio else 0 delta = current - self._crc_error_baselines.get(radio_id, 0) if delta > 0: self.storage.record_crc_errors(delta, radio_id, publish=index == 0) logger.debug( f"Recorded {delta} CRC errors (total: {current}, " f"radio {radio_id or 'default'})" ) self._crc_error_baselines[radio_id] = current except Exception as e: logger.error(f"Error recording CRC errors: {e}") async def _send_periodic_advert_async(self): logger.info( f"Periodic advert timer triggered (interval: {self.send_advert_interval_hours}h)" ) try: if self.send_advert_func: success = await self.send_advert_func() if success: logger.info("Periodic advert sent successfully") else: logger.warning("Failed to send periodic advert") else: logger.debug("No send_advert_func configured") except Exception as e: logger.error(f"Error sending periodic advert: {e}") def reload_runtime_config(self): """Reload runtime configuration from self.config (called after live config updates).""" try: # Refresh delay factors self.tx_delay_factor = self.config.get("delays", {}).get("tx_delay_factor", 1.0) self.direct_tx_delay_factor = self.config.get("delays", {}).get( "direct_tx_delay_factor", 0.5 ) # Refresh repeater settings repeater_config = self.config.get("repeater", {}) self.use_score_for_tx = repeater_config.get("use_score_for_tx", False) self.score_threshold = repeater_config.get("score_threshold", 0.3) self.multi_acks = self._normalize_multi_acks(self.config) self.send_advert_interval_hours = repeater_config.get("send_advert_interval_hours", 10) self.cache_ttl = repeater_config.get("cache_ttl", 60) self.max_flood_hops = repeater_config.get("max_flood_hops", 64) self.neighbour_link_tracker.refresh_config(self.config) self.loop_detect_mode = self._normalize_loop_detect_mode( self.config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF) ) with self.neighbour_link_tracker.lock: now_monotonic = time.monotonic() self.neighbour_link_tracker.purge_expired_locked(now_monotonic) while ( len(self.neighbour_link_tracker.links) > self.neighbour_link_tracker.max_entries ): self.neighbour_link_tracker.evict_stalest_locked() # Radio hardware apply and AirtimeManager modulation refresh are # handled by ConfigManager after a successful live radio update. logger.info("Runtime configuration reloaded successfully") except Exception as e: logger.error(f"Error reloading runtime config: {e}") def cleanup(self): if self._background_task and not self._background_task.done(): self._background_task.cancel() logger.info("Background timer task cancelled") if self.storage: try: self.storage.close() logger.info("StorageCollector closed successfully") except Exception as e: logger.error(f"Error closing StorageCollector: {e}") def __del__(self): try: self.cleanup() except Exception as exc: logger.debug(f"Engine cleanup during __del__ failed: {exc}")