adam/pyMC_Repeater
1
0 Fork 0
mirror of https://github.com/pyMC-dev/pyMC_Repeater.git synced 2026-11-03 20:02:24 +01:00
Code Issues 40 Packages Projects Releases Wiki Activity
Files
dcd972932a3f4959cd212878e58c1d1b9df2846c
pyMC_Repeater/repeater/engine.py
T
agessaman a439c2427c fix(stats): count CRC errors for the node, not the default radio 
The three readers of the CRC scalar all answered for one receiver. On a bridge
that is half the node: a backhaul going deaf moved no number an operator or an
observer could see, while the sum of radios[].errors in the same status message
already said so.

All three now read RepeaterHandler.get_crc_error_count(), which sums the
radios -- stats.errors in the MQTT status message, crc_error_count in /stats,
and n_recv_errors in the MeshCore wire response, which previously read
self.radio directly and so described the default endpoint. stats.errors and the
per-radio entries beside it now reconcile.

This matches what node_stats() already does with the airtime counters that sit
next to it: a node-level counter describes the node, and which radio
contributed what stays available per radio, in radios[] and in the crc_errors
table.

The sum prefers each radio's live hardware counter and falls back to the last
count the sampler read from a radio the fabric cannot hand back. These are
lifetime counters, so dropping a radio would walk the node's total backwards
and hand an observer reading deltas a negative interval. A single-radio node
still reports its one unlabelled counter, read live, exactly as before.
2026-09-21 16:31:30 -07:00

2841 lines
128 KiB
Python
Raw Blame History

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
# Last sample per radio id, on a fabric only. /stats reports the default
# radio's figure as it always has; this carries the others so a bridge's
# sidebar curves all advance on the same beat.
self._cached_noise_floor_by_radio: Dict[str, float] = {}
# 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=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=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=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.
gated: Dict[Optional[str], float] = {}
if airtime_ms > 0:
gate_ids = (
(preferred_tx_radio_id,)
if preferred_tx_radio_id
else self._egress_gate_radio_ids(fwd_pkt)
)
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
gated[gate_id] = gate_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, gated
)
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 airtime_stats_by_radio(self) -> list:
"""``[{radio_id, ...}]`` per radio, empty on a single-radio node.
Radios sharing a channel report the same figures because they are the
same budget, which is the statement rather than a duplication.
"""
return self.airtime_budgets.per_radio_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(), so the fabric is asked
which radio that will be and only that one has to have the budget. Where
it will not say, every radio it could pick has to, and the wait is the
longest of them -- asking whether *any* candidate has budget is how a
node transmits past a legal limit, since it only has to guess the quiet
channel, and requiring all of them drops traffic the node could legally
have carried on another band.
"""
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_gate_radio_ids(packet):
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],
gated: Optional[Dict[Optional[str], float]] = None,
) -> 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.
The figure is the one the gate computed for that radio, not a fresh
calculation. A config reload runs on another thread and can retune a
radio while the packet is still on the air; recomputing afterwards then
charges the modulation the radio has now for a transmission that went
out on the modulation it had a moment ago -- on a 500 kHz-for-62.5 kHz
retune that is an eightfold undercharge, repeatable by holding the
reload against a busy node. Only a radio the gate never saw is computed
here, which is the fabric having chosen one nobody named.
"""
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
if radio_id is None and gated and len(gated) == 1:
# Nobody named a radio and the send reported none, but the gate had
# a single answer from the fabric -- charge that rather than the
# default, which on a bridge is the other band entirely.
radio_id = next(iter(gated))
manager = self.airtime_budgets.for_radio(radio_id)
if gated and radio_id in gated:
charge_ms = gated[radio_id]
else:
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):
# The answer would be about the node's most recent RX rather than
# about this packet, and the retransmit delay is long enough for
# that to change. Asking here would pin the wrong radio.
return None
return self._ask_fabric_tx_radio_id(packet, rx_radio_id)
def _ask_fabric_tx_radio_id(self, packet: Packet, rx_radio_id: Optional[str]) -> Optional[str]:
"""Put the question to the fabric: which radio carries this packet?
Passes the ingress radio when the core can take it, and asks the plain
question when it cannot -- an older core answers from its most recent
RX, which is the same thing its selector will read inside send().
"""
fabric, ids = self._fabric_endpoints()
if fabric is None or not ids or len(ids) < 2:
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:
if fabric_selects_by_ingress_radio(fabric):
radio_id = fabric.resolve_tx_radio_id(raw, None, rx_radio_id=rx_radio_id)
else:
radio_id = fabric.resolve_tx_radio_id(raw, None)
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_gate_radio_ids(self, packet: Packet) -> Tuple[Optional[str], ...]:
"""Which radios the TX-time gate must satisfy when nobody named one.
Asked of the fabric first, under the TX lock an instant before the send,
so an older core's selector is read at almost the moment it will be read
again inside send(). Holding *every* radio the fabric could pick to its
own budget is safe against overrun but is not the behaviour these nodes
had: an exhausted local radio would refuse a transmission the fabric was
about to put on an idle link radio, black-holing that band for the rest
of the window, and worse with three radios.
The candidate set remains the answer when the fabric will not say -- a
node with one radio, or a resolver that raised. What the gate cannot
close is the sliver between asking and sending, where a reception could
move an older core's choice; the debit still lands on the radio the send
reports, so that costs at most one packet charged to the wrong channel,
which is the exposure these nodes already had.
"""
resolved = self._ask_fabric_tx_radio_id(packet, getattr(packet, "_rx_radio_id", None))
if resolved is not None:
return (resolved,)
return self._egress_candidates()
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_cached_noise_floor_by_radio(self) -> Dict[str, float]:
"""Last sample per radio id. Empty on a single-radio node."""
return dict(self._cached_noise_floor_by_radio)
def get_crc_error_count(self) -> int:
"""The node's cumulative CRC error count, summed over its radios.
The node's, not the default radio's: on a bridge the default radio is
one of two receivers, and reporting only its errors leaves a deaf
backhaul invisible in every scalar the node publishes. Summing matches
what AirtimeBudgets.node_stats does with the airtime counters beside it.
Every reader of this figure sees the same number -- the MQTT status
message's ``stats.errors``, /stats' ``crc_error_count`` and the MeshCore
wire response's ``n_recv_errors`` -- so the sum of ``radios[].errors``
in a status message reconciles with the scalar. Which radio contributed
what stays available per radio, both there and in the crc_errors table.
"""
# Live counts win; a radio the fabric cannot hand back right now falls
# back to the last count the sampler read from it. These are lifetime
# counters, so dropping a radio from the sum would walk the node's total
# backwards and hand an observer reading deltas a negative interval --
# the same reason AirtimeBudgets keeps its retired channels' totals.
counts = {
radio_id: count
for radio_id, count in self._crc_error_baselines.items()
if radio_id is not None
}
counts.update(self.get_crc_error_count_by_radio())
if counts:
return sum(counts.values())
# Single-radio node: one unlabelled counter, read live.
radio = self.dispatcher.radio if self.dispatcher else None
return int(getattr(radio, "crc_error_count", 0) or 0) if radio else 0
def get_crc_error_count_by_radio(self) -> Dict[str, int]:
"""Each radio's cumulative CRC error count. Empty on a single-radio node.
Read from the hardware counters rather than the sampler's baselines, so
a status message is not pinned to whenever the CRC tick last ran. A
radio the fabric cannot hand back is omitted rather than reported as
zero: absent and error-free are not the same claim.
"""
counts = {}
for radio_id, radio in self._sampling_radios():
if radio_id is None or radio is None:
continue
counts[radio_id] = int(getattr(radio, "crc_error_count", 0) or 0)
return counts
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
crc_error_count = self.get_crc_error_count()
# 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
# Only a fabric has these, and only beside the unchanged scalar above:
# a single-radio node's response keeps exactly the shape it had.
noise_by_radio = self.get_cached_noise_floor_by_radio()
if noise_by_radio:
stats["noise_floor_radios"] = [
{"radio_id": radio_id, "noise_floor_dbm": dbm}
for radio_id, dbm in noise_by_radio.items()
]
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()
sampled = self._sampling_radios()
# A radio that has left the fabric keeps no stale figure in /stats.
live_ids = {radio_id for radio_id, _ in sampled if radio_id is not None}
for retired in set(self._cached_noise_floor_by_radio) - live_ids:
self._cached_noise_floor_by_radio.pop(retired, None)
for index, (radio_id, radio) in enumerate(sampled):
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
if radio_id is not None:
self._cached_noise_floor_by_radio[radio_id] = 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}")
Reference in new issue View Git Blame Copy Permalink