diff --git a/config.yaml.example b/config.yaml.example index 79b8333..3070db0 100644 --- a/config.yaml.example +++ b/config.yaml.example @@ -328,6 +328,20 @@ mesh: # Use null or "" via CLI command "region default " to clear. default_region: null + # Redundant flood retransmission suppression (rebroadcast cancellation). + # While our own flood retransmission waits out its collision-avoidance delay, + # other repeaters often rebroadcast the same packet first. When enabled, + # hearing flood_suppression_threshold rebroadcast copies cancels our pending + # TX; the packet is logged with drop reason + # "Redundant flood retransmission (rebroadcast heard)" instead of being sent. + # Reduces redundant airtime and collisions on busy networks. + flood_suppression_enabled: false + + # Rebroadcast copies heard (while our TX is pending) that trigger suppression. + # Minimum 1. Raise to 2+ if your repeater provides critical coverage and you + # only want to skip TX when the packet is clearly well-propagated. + flood_suppression_threshold: 1 + # Multiple Identity Configuration (Optional) # Define additional identities for the repeater to manage # Each identity operates independently with its own key pair and configuration diff --git a/repeater/engine.py b/repeater/engine.py index 5870870..368741f 100644 --- a/repeater/engine.py +++ b/repeater/engine.py @@ -43,6 +43,11 @@ LOOP_DETECT_MAX_COUNTERS = { LOOP_DETECT_STRICT: 1, } +# Sentinel returned by schedule_retransmit's task when a pending flood TX was +# cancelled by the redundant flood retransmission check (rebroadcasts of the +# same packet were heard before our own TX slot fired). +TX_RESULT_SUPPRESSED = "suppressed" + class RepeaterHandler(BaseHandler): @staticmethod @@ -85,6 +90,10 @@ class RepeaterHandler(BaseHandler): self.loop_detect_mode = self._normalize_loop_detect_mode( config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF) ) + # Redundant flood retransmission suppression (rebroadcast cancellation). + self._load_flood_suppression_config() + # pkt_hash (full upper hex) -> {"cancel_event": asyncio.Event, "dup_count": int} + self._pending_flood_tx = {} radio = dispatcher.radio if dispatcher else None if radio: @@ -118,6 +127,7 @@ class RepeaterHandler(BaseHandler): self.sent_direct_count = 0 self.flood_dup_count = 0 self.direct_dup_count = 0 + self.flood_suppressed_count = 0 # Storage collector for persistent packet logging try: @@ -238,6 +248,7 @@ class RepeaterHandler(BaseHandler): snr = metadata.get("snr", 0.0) rssi = metadata.get("rssi", 0) transmitted = False + tx_suppressed = False tx_delay_ms = 0.0 drop_reason = None lbt_attempts = 0 @@ -330,8 +341,17 @@ class RepeaterHandler(BaseHandler): self.dropped_count += 1 drop_reason = "Duty cycle limit" else: + suppressible = not local_transmission and route_type in ( + ROUTE_TYPE_FLOOD, + ROUTE_TYPE_TRANSPORT_FLOOD, + ) tx_task = await self.schedule_retransmit( - fwd_pkt, delay, airtime_ms, local_transmission=local_transmission + fwd_pkt, + delay, + airtime_ms, + local_transmission=local_transmission, + packet_hash=pkt_hash_full, + suppressible=suppressible, ) try: tx_success = await tx_task @@ -340,7 +360,13 @@ class RepeaterHandler(BaseHandler): drop_reason = "TX failed" logger.warning(f"Local TX failed: {e}") raise - if not tx_success: + if tx_success == TX_RESULT_SUPPRESSED: + transmitted = False + tx_suppressed = True + drop_reason = "Redundant flood retransmission (rebroadcast heard)" + self.flood_suppressed_count += 1 + self.dropped_count += 1 + elif not tx_success: transmitted = False drop_reason = "TX failed" self.dropped_count += 1 @@ -388,8 +414,8 @@ class RepeaterHandler(BaseHandler): 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 + # Check if this is a duplicate (a suppressed TX is not a duplicate of itself) + is_dupe = pkt_hash_full in self.seen_packets and not transmitted and not tx_suppressed # Set drop reason for duplicates and count flood vs direct dups if is_dupe and drop_reason is None: @@ -543,9 +569,12 @@ class RepeaterHandler(BaseHandler): """ self.rx_count += 1 route_type = packet.header & PH_ROUTE_MASK + pkt_hash_full = packet.calculate_packet_hash().hex().upper() if route_type in (ROUTE_TYPE_FLOOD, ROUTE_TYPE_TRANSPORT_FLOOD): self.recv_flood_count += 1 self.flood_dup_count += 1 + # Rebroadcast copy heard — may cancel our own pending flood TX. + self._note_flood_duplicate(pkt_hash_full) elif route_type in (ROUTE_TYPE_DIRECT, ROUTE_TYPE_TRANSPORT_DIRECT): self.recv_direct_count += 1 self.direct_dup_count += 1 @@ -573,7 +602,7 @@ class RepeaterHandler(BaseHandler): transmitted=False, drop_reason="Duplicate", is_duplicate=True, - packet_hash=packet.calculate_packet_hash().hex().upper(), + packet_hash=pkt_hash_full, ) if self.storage: @@ -773,6 +802,38 @@ class RepeaterHandler(BaseHandler): if len(self.seen_packets) > self.max_cache_size: self.seen_packets.popitem(last=False) + def _load_flood_suppression_config(self) -> None: + """Load redundant-flood-retransmission suppression settings from config.""" + mesh_cfg = self.config.get("mesh", {}) + self.flood_suppression_enabled = bool(mesh_cfg.get("flood_suppression_enabled", False)) + try: + threshold = int(mesh_cfg.get("flood_suppression_threshold", 1)) + except (TypeError, ValueError): + threshold = 1 + self.flood_suppression_threshold = max(1, threshold) + + def _note_flood_duplicate(self, pkt_hash: Optional[str]) -> None: + """Count a rebroadcast copy heard while our own flood TX is pending. + + Redundant flood retransmission check: once the configured number of + rebroadcast copies has been heard, the pending retransmission is + cancelled before it reaches the radio. + """ + if not self.flood_suppression_enabled or not pkt_hash: + return + entry = self._pending_flood_tx.get(pkt_hash) + if entry is None or entry["cancel_event"].is_set(): + return + entry["dup_count"] += 1 + if entry["dup_count"] >= self.flood_suppression_threshold: + entry["cancel_event"].set() + logger.info( + "Redundant flood retransmission check: heard %d rebroadcast(s) of " + "packet %s — cancelling our pending TX", + entry["dup_count"], + pkt_hash[:16], + ) + def validate_packet(self, packet: Packet) -> Tuple[bool, str]: if not packet or not packet.payload: @@ -952,6 +1013,9 @@ class RepeaterHandler(BaseHandler): # Suppress duplicates — pass pre-computed hash to avoid a second SHA-256. if self.is_duplicate(packet, packet_hash=packet_hash): + # Rebroadcast copy heard (dispatcher dedupe disabled path) — may + # cancel our own pending flood TX for this packet. + self._note_flood_duplicate(packet_hash or packet.calculate_packet_hash().hex().upper()) packet.drop_reason = "Duplicate" return None @@ -1143,77 +1207,123 @@ class RepeaterHandler(BaseHandler): delay: float, airtime_ms: float = 0.0, local_transmission: bool = False, + packet_hash: Optional[str] = None, + suppressible: bool = False, ): """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). + + When suppressible is True (relayed flood packets) and flood suppression + is enabled, the pending TX is registered under packet_hash so that + rebroadcast copies heard during the delay cancel it (redundant flood + retransmission check). The task then resolves to TX_RESULT_SUPPRESSED + instead of True/False. """ + suppression_entry = None + if suppressible and self.flood_suppression_enabled and packet_hash: + suppression_entry = {"cancel_event": asyncio.Event(), "dup_count": 0} + self._pending_flood_tx[packet_hash] = suppression_entry + + def _suppression_triggered() -> bool: + return suppression_entry is not None and suppression_entry["cancel_event"].is_set() + + def _log_suppressed(stage: str) -> None: + logger.info( + "TX prevented (%s): redundant flood retransmission check — " + "%d rebroadcast(s) of packet %s heard while TX was pending", + stage, + suppression_entry["dup_count"], + (packet_hash or "")[:16], + ) async def delayed_send(): - await asyncio.sleep(delay) + try: + 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: - # 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) + # Redundant flood retransmission check — rebroadcasts may have + # arrived while this task slept through its collision-avoidance + # delay. Skip the radio entirely if the packet is already covered. + if _suppression_triggered(): + _log_suppressed("pre-send") + return TX_RESULT_SUPPRESSED - 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. - if airtime_ms > 0: - can_tx_now, _ = self.airtime_mgr.can_transmit(airtime_ms) - if not can_tx_now: - logger.warning( - "Packet dropped at TX time: duty-cycle exceeded (airtime=%.1fms)", - airtime_ms, - ) - return False + # 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: + # 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) - try: - sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) - 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) + async with self._tx_lock: + # Re-check after acquiring the lock: rebroadcasts may have + # arrived while another task held the radio. + if _suppression_triggered(): + _log_suppressed("at-lock") + return TX_RESULT_SUPPRESSED + + # ── 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. if airtime_ms > 0: - self.airtime_mgr.record_tx(airtime_ms) - packet_size = fwd_pkt.get_raw_length() - logger.info( - f"Retransmitted packet ({packet_size} bytes, " - f"{airtime_ms:.1f}ms airtime)" - ) - return True - except Exception as e: - 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 + can_tx_now, _ = self.airtime_mgr.can_transmit(airtime_ms) + if not can_tx_now: + logger.warning( + "Packet dropped at TX time: duty-cycle exceeded (airtime=%.1fms)", + airtime_ms, + ) + return False + + try: + sent = await self.dispatcher.send_packet(fwd_pkt, wait_for_ack=False) + 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) + if airtime_ms > 0: + self.airtime_mgr.record_tx(airtime_ms) + packet_size = fwd_pkt.get_raw_length() + logger.info( + f"Retransmitted packet ({packet_size} bytes, " + f"{airtime_ms:.1f}ms airtime)" + ) + return True + except Exception as e: + 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 + finally: + # Deregister the pending TX regardless of outcome so the registry + # never grows beyond in-flight retransmissions. + if ( + suppression_entry is not None + and self._pending_flood_tx.get(packet_hash) is suppression_entry + ): + self._pending_flood_tx.pop(packet_hash, None) return asyncio.create_task(delayed_send()) @@ -1287,6 +1397,7 @@ class RepeaterHandler(BaseHandler): "sent_direct_count": self.sent_direct_count, "flood_dup_count": self.flood_dup_count, "direct_dup_count": self.direct_dup_count, + "flood_suppressed_count": self.flood_suppressed_count, "rx_per_hour": rx_per_hour, "forwarded_per_hour": forwarded_per_hour, "recent_packets": list(self.recent_packets), @@ -1332,6 +1443,8 @@ class RepeaterHandler(BaseHandler): self.config.get("mesh", {}).get("global_flood_allow", True), ), "path_hash_mode": self.config.get("mesh", {}).get("path_hash_mode", 0), + "flood_suppression_enabled": self.flood_suppression_enabled, + "flood_suppression_threshold": self.flood_suppression_threshold, }, "mqtt_brokers": self.config.get("mqtt_brokers", {}), }, @@ -1465,6 +1578,7 @@ class RepeaterHandler(BaseHandler): self.loop_detect_mode = self._normalize_loop_detect_mode( self.config.get("mesh", {}).get("loop_detect", LOOP_DETECT_OFF) ) + self._load_flood_suppression_config() # Note: Radio config changes require restart as they affect hardware # Note: Airtime manager has its own config reference that gets updated