From 5a83fce84c8d4bdf659b0a4d7cffc8bd23cc243e Mon Sep 17 00:00:00 2001 From: agessaman Date: Thu, 16 Jul 2026 08:48:34 -0700 Subject: [PATCH] refactor(router): centralize bridge fan-out and type drop reasons MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the eight hand-rolled companion-bridge delivery loops in the packet router with a single _fan_out_to_bridges helper that reads HandlerResult.authenticated directly, so a broken handler contract surfaces instead of being hidden by getattr hedges. Introduce a DropReason str-Enum in the engine for the seventeen fixed drop reasons (with __str__ returning the value, since str() of a str-Enum changed on Python 3.11+) and derive the router's expected-drop check from it, retiring the string-prefix tuple. Detailed variants keep their suffixes embedded in the reason string. Split the companion-delivery dedupe cache into a pure check and an explicit mark so PATH and protocol responses record delivery only after the fan-out runs — a copy where every bridge raised is retried on the next copy instead of being suppressed for the full TTL. --- repeater/engine.py | 102 ++++++++++----- repeater/handler_helpers/advert.py | 4 +- repeater/packet_router.py | 201 ++++++++++++++--------------- 3 files changed, 171 insertions(+), 136 deletions(-) diff --git a/repeater/engine.py b/repeater/engine.py index f69b428..cbcd04f 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -4,6 +4,7 @@ import logging import secrets import time from collections import OrderedDict, deque +from enum import Enum from typing import Optional, Tuple from openhop_core.node.handlers.base import BaseHandler @@ -47,6 +48,41 @@ LOOP_DETECT_MAX_COUNTERS = { } +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 RepeaterHandler(BaseHandler): @staticmethod def payload_type() -> int: @@ -359,7 +395,7 @@ class RepeaterHandler(BaseHandler): f"wait={wait_time:.1f}s before retry" ) self.dropped_count += 1 - drop_reason = "Duty cycle limit" + drop_reason = DropReason.DUTY_CYCLE_LIMIT else: tx_task = await self.schedule_retransmit( fwd_pkt, delay, airtime_ms, local_transmission=local_transmission @@ -394,9 +430,9 @@ class RepeaterHandler(BaseHandler): self.dropped_count += 1 # Determine drop reason if local_transmission and not allow_local_tx: - drop_reason = policy_reason or "No TX mode" + drop_reason = policy_reason or DropReason.NO_TX_MODE elif not allow_forward: - drop_reason = policy_reason or "Repeat disabled" + 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( @@ -424,7 +460,7 @@ class RepeaterHandler(BaseHandler): # Set drop reason for duplicates and count flood vs direct dups if is_dupe and drop_reason is None: - drop_reason = "Duplicate" + drop_reason = DropReason.DUPLICATE if is_dupe: if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.flood_dup_count += 1 @@ -473,7 +509,11 @@ class RepeaterHandler(BaseHandler): if self.storage: try: # Only skip mqtt for actual invalid/bad packets - invalid_reasons = ["Invalid advert packet", "Empty payload", "Path too long"] + invalid_reasons = ( + DropReason.INVALID_ADVERT, + DropReason.EMPTY_PAYLOAD, + DropReason.PATH_TOO_LONG, + ) skip_mqtt = drop_reason in invalid_reasons if drop_reason else False self.storage.record_packet(packet_record, skip_mqtt_if_invalid=skip_mqtt) except Exception as e: @@ -618,7 +658,7 @@ class RepeaterHandler(BaseHandler): src_hash, dst_hash, transmitted=False, - drop_reason="Duplicate", + drop_reason=DropReason.DUPLICATE, is_duplicate=True, packet_hash=pkt_hash_full, ) @@ -777,13 +817,13 @@ class RepeaterHandler(BaseHandler): def _get_drop_reason(self, packet: Packet, packet_hash: Optional[str] = None) -> str: if self.is_duplicate(packet, packet_hash=packet_hash): - return "Duplicate" + return DropReason.DUPLICATE if not packet or not packet.payload: - return "Empty payload" + return DropReason.EMPTY_PAYLOAD if len(packet.path or []) > MAX_PATH_SIZE: - return "Path too long" + return DropReason.PATH_TOO_LONG route_type = packet.header & PH_ROUTE_MASK @@ -793,15 +833,15 @@ class RepeaterHandler(BaseHandler): "unscoped_flood_allow", self.config.get("mesh", {}).get("global_flood_allow", True) ) if not unscoped_flood_allow: - return "Unscoped flood policy disabled" + 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 "Direct: no path" + return DropReason.DIRECT_NO_PATH next_hop = bytes(packet.path[:hash_size]) if next_hop != self.local_hash_bytes[:hash_size]: - return "Direct: not for us" + return DropReason.DIRECT_NOT_FOR_US # Default reason return "Unknown" @@ -832,7 +872,7 @@ class RepeaterHandler(BaseHandler): def validate_packet(self, packet: Packet) -> Tuple[bool, str]: if not packet or not packet.payload: - return False, "Empty payload" + return False, DropReason.EMPTY_PAYLOAD if packet.get_path_hash_size() > 3: return False, "Reserved path hash size is invalid" @@ -988,7 +1028,7 @@ class RepeaterHandler(BaseHandler): if packet.is_marked_do_not_retransmit(): # Check if packet has custom drop reason if not packet.drop_reason: - packet.drop_reason = "Marked do not retransmit" + packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None # Check unscoped flood policy @@ -998,24 +1038,24 @@ class RepeaterHandler(BaseHandler): route_type = packet.header & PH_ROUTE_MASK if route_type == ROUTE_TYPE_FLOOD: if not unscoped_flood_allow: - packet.drop_reason = "Unscoped flood policy disabled" + 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 = "Transport code not allowed to flood" + 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"FLOOD loop detected ({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 = "Duplicate" + packet.drop_reason = DropReason.DUPLICATE return None if packet.path is None: @@ -1027,17 +1067,17 @@ class RepeaterHandler(BaseHandler): hop_count = packet.get_path_hash_count() if self.max_flood_hops > 0 and hop_count >= self.max_flood_hops: - packet.drop_reason = f"Max flood hops limit reached ({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 = "Path hop count at maximum (63), cannot append" + 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 = "Path would exceed MAX_PATH_SIZE" + packet.drop_reason = DropReason.PATH_EXCEEDS_MAX_SIZE return None self.mark_seen(packet, packet_hash=packet_hash) @@ -1064,7 +1104,7 @@ class RepeaterHandler(BaseHandler): # Check if packet is marked do-not-retransmit if packet.is_marked_do_not_retransmit(): if not packet.drop_reason: - packet.drop_reason = "Marked do not retransmit" + packet.drop_reason = DropReason.MARKED_DO_NOT_RETRANSMIT return None hash_size = packet.get_path_hash_size() @@ -1072,17 +1112,17 @@ class RepeaterHandler(BaseHandler): # Check if we're the next hop if not packet.path or len(packet.path) < hash_size: - packet.drop_reason = "Direct: no path" + 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 = "Direct: not for us" + 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 = "Duplicate" + packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet, packet_hash=packet_hash) @@ -1116,7 +1156,7 @@ class RepeaterHandler(BaseHandler): # 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 = "Multipart: not direct-routed" + packet.drop_reason = f"{DropReason.MULTIPART}: not direct-routed" return None valid, reason = self.validate_packet(packet) @@ -1125,7 +1165,7 @@ class RepeaterHandler(BaseHandler): return None if packet.is_marked_do_not_retransmit(): - packet.drop_reason = packet.drop_reason or "Marked do not retransmit" + packet.drop_reason = packet.drop_reason or DropReason.MARKED_DO_NOT_RETRANSMIT return None payload = packet.payload or b"" @@ -1134,7 +1174,7 @@ class RepeaterHandler(BaseHandler): # 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 = "Multipart: unsupported embedded type" + packet.drop_reason = f"{DropReason.MULTIPART}: unsupported embedded type" return None hash_size = packet.get_path_hash_size() @@ -1143,10 +1183,10 @@ class RepeaterHandler(BaseHandler): # 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 = "Direct: no path" + packet.drop_reason = DropReason.DIRECT_NO_PATH return None if bytes(packet.path[:hash_size]) != self.local_hash_bytes[:hash_size]: - packet.drop_reason = "Direct: not for us" + packet.drop_reason = DropReason.DIRECT_NOT_FOR_US return None # Rebuild the received wrapper as the plain DIRECT ACK MeshCore would emit: @@ -1161,7 +1201,7 @@ class RepeaterHandler(BaseHandler): # so a repeated multipart ACK — or an equivalent plain ACK — is not relayed # twice. The pre-computed hash belongs to the original wrapper, so recompute. if self.is_duplicate(packet): - packet.drop_reason = "Duplicate" + packet.drop_reason = DropReason.DUPLICATE return None self.mark_seen(packet) diff --git a/repeater/handler_helpers/advert.py b/repeater/handler_helpers/advert.py index 263d31f..36bca19 100644 --- a/repeater/handler_helpers/advert.py +++ b/repeater/handler_helpers/advert.py @@ -15,6 +15,8 @@ from typing import Dict, Optional, Tuple from openhop_core.node.handlers.advert import AdvertHandler +from repeater.engine import DropReason + logger = logging.getLogger("AdvertHelper") @@ -551,7 +553,7 @@ class AdvertHelper: if not advert_data or not advert_data.get("valid"): logger.warning("Invalid advert packet received, dropping.") packet.mark_do_not_retransmit() - packet.drop_reason = "Invalid advert packet" + packet.drop_reason = DropReason.INVALID_ADVERT return # Extract data from parsed advert diff --git a/repeater/packet_router.py b/repeater/packet_router.py index 8e3029f..d08b9eb 100644 --- a/repeater/packet_router.py +++ b/repeater/packet_router.py @@ -5,7 +5,6 @@ import time from openhop_core.node.handlers.ack import AckHandler from openhop_core.node.handlers.advert import AdvertHandler from openhop_core.node.handlers.control import ControlHandler -from openhop_core.node.handlers.group_text import GroupTextHandler from openhop_core.node.handlers.login_response import LoginResponseHandler from openhop_core.node.handlers.login_server import LoginServerHandler from openhop_core.node.handlers.multipart import MultipartAckHandler @@ -16,11 +15,13 @@ from openhop_core.node.handlers.text import TextMessageHandler from openhop_core.node.handlers.trace import TraceHandler from openhop_core.protocol.constants import ( PAYLOAD_TYPE_GRP_DATA, + PAYLOAD_TYPE_GRP_TXT, PH_ROUTE_MASK, ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT, ) +from repeater.engine import DropReason from repeater.policy_engine import PolicyDecision, PolicyEngine logger = logging.getLogger("PacketRouter") @@ -29,27 +30,11 @@ logger = logging.getLogger("PacketRouter") # so the client is not spammed with duplicate telemetry when the mesh delivers multiple copies. _COMPANION_DEDUPE_TTL_SEC = 60.0 -# Drop reasons that are normal policy outcomes and should not be warning-level. -# TODO: create Enum in engine for drop reasons and use it here and in engine instead of string matching. -_EXPECTED_DROP_REASON_PREFIXES = ( - "Duplicate", - "Max flood hops limit reached", - "Path hop count at maximum", - "Path would exceed MAX_PATH_SIZE", - "Direct: no path", - "Direct: not for us", - "Unscoped flood policy disabled", - "Transport code not allowed to flood", - "FLOOD loop detected", - "Marked do not retransmit", - "Repeat disabled", - "No TX mode", - "Duty cycle limit", - "Empty payload", - "Path too long", - "Invalid advert packet", - "Multipart", -) +# Normal policy/forwarding drop outcomes (logged at debug, not warning). The engine +# is the single source of truth via DropReason; string-form acceptance is retained +# because records and the recent_packets fallback still carry detailed text variants +# (e.g. "Max flood hops limit reached (5/5)"). +_EXPECTED_DROP_REASONS = frozenset(DropReason) def _companion_dedup_key(packet) -> str | None: @@ -77,10 +62,14 @@ def _is_direct_intermediate_hop(packet) -> bool: ) -def _is_expected_drop_reason(reason: str | None) -> bool: +def _is_expected_drop_reason(reason) -> bool: + # Membership is the fast path for a DropReason member or its exact text; the + # startswith scan covers detailed variants that embed a suffix. + if reason in _EXPECTED_DROP_REASONS: + return True if not isinstance(reason, str) or not reason: return False - return any(reason.startswith(prefix) for prefix in _EXPECTED_DROP_REASON_PREFIXES) + return any(reason.startswith(expected) for expected in _EXPECTED_DROP_REASONS) def _drop_reason_from_recent_packets(handler, packet) -> str | None: @@ -181,11 +170,28 @@ class PacketRouter: if exc is not None: logger.error("_route_packet raised: %s", exc, exc_info=exc) - def _should_deliver_path_to_companions(self, packet) -> bool: - """Return True if this PATH/protocol-response should be delivered to companions (first of duplicates).""" + def _was_delivered_to_companions(self, packet) -> bool: + """Pure check: True if this PATH/protocol-response was already delivered (unexpired). + + Does not mutate delivery state, so callers can deliver first and only record + the packet as delivered once a bridge has actually received it. + """ key = _companion_dedup_key(packet) if not key: - return True + return False + expiry = self._companion_delivered.get(key) + return expiry is not None and expiry > time.time() + + def _mark_delivered_to_companions(self, packet) -> None: + """Record that this PATH/protocol-response reached companions. + + Called only after a bridge has received the packet, so a delivery that + raised for every bridge is retried on the next copy instead of being + suppressed for the full dedupe TTL. + """ + key = _companion_dedup_key(packet) + if not key: + return now = time.time() # Prune expired entries only when the dict grows large, avoiding a full # dict comprehension on every packet. 200 entries × 60 s TTL means a @@ -196,9 +202,13 @@ class PacketRouter: self._companion_delivered = { k: v for k, v in self._companion_delivered.items() if v > now } - if key in self._companion_delivered: - return False self._companion_delivered[key] = now + _COMPANION_DEDUPE_TTL_SEC + + def _should_deliver_path_to_companions(self, packet) -> bool: + """Atomic check-and-mark: True on the first of duplicate PATH/protocol-responses.""" + if self._was_delivered_to_companions(packet): + return False + self._mark_delivered_to_companions(packet) return True def _policy_companion_decision(self, packet, metadata: dict) -> PolicyDecision | None: @@ -263,6 +273,25 @@ class PacketRouter: return {} return companion_bridges + async def _fan_out_to_bridges(self, packet, bridges, *, context: str) -> bool: + """Offer packet to each bridge; True if any bridge authenticated it. + + Accepts a dict of bridges — pass a single-entry dict for targeted delivery + to the bridge that owns ``dest_hash``. A bridge that raises is logged and + skipped; ``result.authenticated`` is read directly (every bridge returns a + HandlerResult) so a broken contract surfaces instead of being hidden. + """ + authenticated = False + for bridge in bridges.values(): + try: + result = await bridge.process_received_packet(packet) + except Exception as e: + logger.debug("Companion bridge %s error: %s", context, e) + continue + if result.authenticated is True: + authenticated = True + return authenticated + async def _consume_via_local_candidates( self, packet, metadata: dict, dest_hash, helper, process_method_name: str ) -> bool: @@ -537,11 +566,7 @@ class PacketRouter: # Also feed adverts to companion bridges (for contact/path updates), # but keep policy drop final just like the other companion paths. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge advert error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="advert") elif payload_type == LoginServerHandler.payload_type(): # Route ANON_REQ/login to the local identity that owns it. The on-air @@ -566,11 +591,7 @@ class PacketRouter: # ACK has no dest in payload (4-byte CRC only); deliver to all bridges so sender sees send_confirmed. # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge ACK error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="ACK") elif payload_type == MultipartAckHandler.payload_type(): # MULTIPART ACK wrapper: low nibble of first byte is embedded payload type. @@ -610,18 +631,22 @@ class PacketRouter: dest_hash = packet.payload[0] if packet.payload else None companion_bridges = self._companion_bridges_for_packet(packet, metadata) if dest_hash is not None and dest_hash in companion_bridges: - if self._should_deliver_path_to_companions(packet): - result = await companion_bridges[dest_hash].process_received_packet(packet) - consumed = consumed or getattr(result, "authenticated", False) is True - elif companion_bridges and self._should_deliver_path_to_companions(packet): + if not self._was_delivered_to_companions(packet): + consumed = ( + await self._fan_out_to_bridges( + packet, {dest_hash: companion_bridges[dest_hash]}, context="PATH" + ) + or consumed + ) + self._mark_delivered_to_companions(packet) + elif companion_bridges and not self._was_delivered_to_companions(packet): # Dest not in bridges: path-return with ephemeral dest (e.g. multi-hop login). # Deliver to all bridges; each will try to decrypt and ignore if not relevant. - for bridge in companion_bridges.values(): - try: - result = await bridge.process_received_packet(packet) - consumed = consumed or getattr(result, "authenticated", False) is True - except Exception as e: - logger.debug(f"Companion bridge PATH error: {e}") + consumed = ( + await self._fan_out_to_bridges(packet, companion_bridges, context="PATH") + or consumed + ) + self._mark_delivered_to_companions(packet) logger.debug( "PATH dest=0x%02x (anon) delivered to %d bridge(s) for matching", dest_hash or 0, @@ -643,23 +668,15 @@ class PacketRouter: companion_bridges = self._companion_bridges_for_packet(packet, metadata) local_hash = getattr(self.daemon, "local_hash", None) if dest_hash is not None and dest_hash in companion_bridges: - try: - result = await companion_bridges[dest_hash].process_received_packet(packet) - consumed = consumed or getattr(result, "authenticated", False) is True - logger.info( - "RESPONSE dest=0x%02x delivered to companion bridge", - dest_hash, - ) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") + consumed = await self._fan_out_to_bridges( + packet, {dest_hash: companion_bridges[dest_hash]}, context="RESPONSE" + ) + logger.info("RESPONSE dest=0x%02x delivered to companion bridge", dest_hash) elif dest_hash == local_hash and companion_bridges: # Response addressed to this repeater (e.g. path-based reply to first hop) - for bridge in companion_bridges.values(): - try: - result = await bridge.process_received_packet(packet) - consumed = consumed or getattr(result, "authenticated", False) is True - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") + consumed = await self._fan_out_to_bridges( + packet, companion_bridges, context="RESPONSE" + ) logger.info( "RESPONSE dest=0x%02x (local) delivered to %d companion bridge(s)", dest_hash, @@ -669,12 +686,9 @@ class PacketRouter: # Dest not in bridges and not local: likely ANON_REQ response (dest = ephemeral # sender hash). Deliver to all bridges; each will try to decrypt and ignore if # not relevant (firmware-like behavior, works with multiple companion bridges). - for bridge in companion_bridges.values(): - try: - result = await bridge.process_received_packet(packet) - consumed = consumed or getattr(result, "authenticated", False) is True - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") + consumed = await self._fan_out_to_bridges( + packet, companion_bridges, context="RESPONSE" + ) logger.debug( "RESPONSE dest=0x%02x (anon) delivered to %d bridge(s) for matching", dest_hash or 0, @@ -691,23 +705,16 @@ class PacketRouter: elif payload_type == ProtocolResponseHandler.payload_type(): # PAYLOAD_TYPE_PATH (0x08): protocol responses (telemetry, binary, etc.). - # Deliver at most once per logical packet so the client is not spammed with duplicates. - # Do not set processed_by_injection so packet also reaches engine for DIRECT forwarding when we're a middle hop. + # Deliver at most once per logical packet so the client is not spammed with duplicates, + # but always deliver at a final hop (we are the destination). Do not set + # processed_by_injection for a middle hop so the packet still reaches engine forwarding. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges and self._should_deliver_path_to_companions(packet): - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE error: {e}") - if companion_bridges and _is_direct_final_hop(packet): - # DIRECT with empty path: we're the final hop; ensure delivery to all bridges (anon) - if not self._should_deliver_path_to_companions(packet): - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge RESPONSE (final hop) error: {e}") + final_hop = _is_direct_final_hop(packet) + if companion_bridges and (final_hop or not self._was_delivered_to_companions(packet)): + await self._fan_out_to_bridges(packet, companion_bridges, context="RESPONSE") + self._mark_delivered_to_companions(packet) + if companion_bridges and final_hop: + # DIRECT with empty path: we're the final hop, so consume after delivery. processed_by_injection = True self._record_for_ui(packet, metadata) @@ -732,36 +739,22 @@ class PacketRouter: if companion_bridges and _is_direct_final_hop(packet): # OpenHop release hygiene: an empty-path DIRECT has no next hop, # so consume after offering it to bridges even without MAC ownership. - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge REQ (final hop) error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="REQ") processed_by_injection = True self._record_for_ui(packet, metadata) - elif payload_type == GroupTextHandler.payload_type(): + elif payload_type == PAYLOAD_TYPE_GRP_TXT: # GRP_TXT: pass to all companions (they filter by channel); still forward. # Policy drop is final and blocks companion delivery. companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges: - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge GRP_TXT error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_TXT") elif payload_type == PAYLOAD_TYPE_GRP_DATA: # MeshCore forwards direct packets with remaining hops before payload # handling. Otherwise, companions authenticate and filter channels. if not _is_direct_intermediate_hop(packet): companion_bridges = self._companion_bridges_for_packet(packet, metadata) - if companion_bridges: - for bridge in companion_bridges.values(): - try: - await bridge.process_received_packet(packet) - except Exception as e: - logger.debug(f"Companion bridge GRP_DATA error: {e}") + await self._fan_out_to_bridges(packet, companion_bridges, context="GRP_DATA") # Only pass to repeater engine if not already processed by injection # Skip engine for packets we injected for TX (already sent; avoid double-send/double-count)