Improve repeater alerts and delivery reliability

src/Dispatcher.cpp

@@ -344,7 +344,12 @@ void Dispatcher::checkSend() {
   }
   
   if (!millisHasNowPassed(next_tx_time)) return;
-  if (_radio->isReceiving()) {
+
+  Packet* pending = _mgr->peekNextOutbound(_ms->getMillis());
+  bool channel_busy = pending != NULL && usePassiveChannelCheck(pending)
+    ? _radio->isReceivingPassive(getRetryInterferenceMargin())
+    : _radio->isReceiving();
+  if (channel_busy) {
     if (cad_busy_start == 0) {
       cad_busy_start = _ms->getMillis();   // record when CAD busy state started
     }

src/Dispatcher.h

@@ -94,6 +94,18 @@ public:
   */
   virtual bool isReceiving() { return false; }
 
+  /**
+   * \brief Non-invasive channel check for a queued retry.
+   *
+   * Implementations should avoid CAD or any operation that restarts RX, because
+   * doing so can hide the downstream forwarding echo that would cancel the
+   * retry. Radios without a passive implementation retain the legacy check.
+   */
+  virtual bool isReceivingPassive(int interference_margin_db) {
+    (void)interference_margin_db;
+    return isReceiving();
+  }
+
   virtual float getLastRSSI() const { return 0; }
   virtual float getLastSNR() const { return 0; }
 
@@ -114,6 +126,10 @@ public:
 
   virtual bool queueOutbound(Packet* packet, uint8_t priority, uint32_t scheduled_for) = 0;
   virtual Packet* getNextOutbound(uint32_t now) = 0;    // by priority
+  virtual Packet* peekNextOutbound(uint32_t now) {
+    (void)now;
+    return NULL;
+  }
   virtual int getOutboundCount(uint32_t now) const = 0;
   virtual int getOutboundTotal() const = 0;
   virtual int getFreeCount() const = 0;
@@ -208,6 +224,11 @@ protected:
   virtual uint32_t getCADFailMaxDuration() const;
   virtual uint8_t getDefaultTxCodingRate() const { return 0; }
   virtual bool allowPacketTransmit(const Packet* packet) const { return true; }
+  virtual bool usePassiveChannelCheck(const Packet* packet) const {
+    (void)packet;
+    return false;
+  }
+  virtual int getRetryInterferenceMargin() const { return 12; }
   virtual int getInterferenceThreshold() const { return 0; }    // disabled by default
   virtual bool getCADEnabled() const { return false; }    // hardware CAD disabled by default
   virtual int getAGCResetInterval() const { return 0; }    // disabled by default

src/Mesh.cpp

@@ -154,6 +154,7 @@ void Mesh::begin() {
     _direct_retries[i].priority = 0;
     _direct_retries[i].progress_marker = 0;
     _direct_retries[i].expect_path_growth = false;
+    _direct_retries[i].final_hop_retry = false;
     _direct_retries[i].waiting_final_echo = false;
     _direct_retries[i].queued = false;
     _direct_retries[i].active = false;
@@ -547,10 +548,13 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
 
         if (!_tables->wasSeen(pkt)) {
           _tables->markSeen(pkt);
+          bool final_hop_retry = pkt->getPathHashCount() == 1
+              && pkt->getPayloadType() == PAYLOAD_TYPE_TXT_MSG
+              && pkt->payload_len >= 2 + CIPHER_MAC_SIZE;
           removePathPrefix(pkt, 1);
 
           uint32_t d = getDirectRetransmitDelay(pkt);
-          maybeScheduleDirectRetry(pkt, 0);
+          maybeScheduleDirectRetry(pkt, 0, final_hop_retry);
           return ACTION_RETRANSMIT_DELAYED(0, d);  // Routed traffic is HIGHEST priority
         }
       }
@@ -908,6 +912,7 @@ void Mesh::clearDirectRetrySlot(int idx) {
   _direct_retries[idx].priority = 0;
   _direct_retries[idx].progress_marker = 0;
   _direct_retries[idx].expect_path_growth = false;
+  _direct_retries[idx].final_hop_retry = false;
   _direct_retries[idx].waiting_final_echo = false;
   _direct_retries[idx].queued = false;
   _direct_retries[idx].active = false;
@@ -922,6 +927,29 @@ bool Mesh::isDirectRetryQueued(const Packet* packet) const {
   return false;
 }
 
+bool Mesh::usePassiveChannelCheck(const Packet* packet) const {
+  for (int i = 0; i < MAX_DIRECT_RETRY_SLOTS; i++) {
+    if (_direct_retries[i].active && _direct_retries[i].queued
+        && _direct_retries[i].packet == packet) {
+      return true;
+    }
+  }
+
+  // Flood retries use the same receive-side cancellation as direct retries:
+  // an overheard downstream forwarding echo removes the queued retry. Avoid
+  // CAD here too, since restarting RX can hide that echo. The initial flood
+  // has trigger_packet set but queued=false, so ordinary flood forwarding
+  // continues to use the normal CAD check.
+  for (int i = 0; i < MAX_FLOOD_RETRY_SLOTS; i++) {
+    if (_flood_retries[i].active && _flood_retries[i].queued
+        && _flood_retries[i].packet == packet) {
+      return true;
+    }
+  }
+
+  return false;
+}
+
 void Mesh::calculateDirectRetryKey(const Packet* packet, uint8_t* dest_key) const {
   uint8_t type = packet->getPayloadType();
   Utils::sha256(dest_key, MAX_HASH_SIZE, &type, 1, packet->payload, packet->payload_len);
@@ -1005,6 +1033,13 @@ void Mesh::armDirectRetryOnSendComplete(const Packet* packet) {
                            _direct_retries[i].next_hop_hash, _direct_retries[i].next_hop_hash_len);
         _direct_retries[i].echo_wait_started_at = _ms->getMillis();
         _direct_retries[i].retry_attempts_sent++;
+        if (_direct_retries[i].final_hop_retry) {
+          // The destination does not forward the packet, so no downstream echo
+          // can confirm this hop. Send exactly one duplicate and finish without
+          // treating the lack of an echo as a link failure.
+          clearDirectRetrySlot(i);
+          continue;
+        }
         uint8_t max_attempts = getDirectRetryMaxAttempts(packet);
         if (max_attempts < 1) {
           max_attempts = 1;
@@ -1235,13 +1270,26 @@ bool Mesh::canDecodeDirectPayloadForSelf(const Packet* packet) {
   }
 }
 
-void Mesh::maybeScheduleDirectRetry(const Packet* packet, uint8_t priority) {
-  const uint8_t* next_hop_hash;
-  uint8_t next_hop_hash_len;
-  uint8_t progress_marker;
-  bool expect_path_growth;
-  if (!getDirectRetryTarget(packet, next_hop_hash, next_hop_hash_len, progress_marker, expect_path_growth)
-      || !allowDirectRetry(packet, next_hop_hash, next_hop_hash_len)) {
+void Mesh::maybeScheduleDirectRetry(const Packet* packet, uint8_t priority, bool final_hop_retry) {
+  const uint8_t* next_hop_hash = NULL;
+  uint8_t next_hop_hash_len = 0;
+  uint8_t progress_marker = 0;
+  bool expect_path_growth = false;
+  if (final_hop_retry) {
+    if (packet == NULL || !packet->isRouteDirect()
+        || packet->getPayloadType() != PAYLOAD_TYPE_TXT_MSG
+        || packet->getPathHashCount() != 0 || packet->payload_len < 2 + CIPHER_MAC_SIZE
+        || !allowDirectRetry(packet, NULL, 0)) {
+      return;
+    }
+    // The encrypted payload exposes only the destination hash to a relay. Keep
+    // it for diagnostics, but do not apply recent-repeater/SNR eligibility to
+    // the destination itself.
+    next_hop_hash = packet->payload;
+    next_hop_hash_len = 1;
+  } else if (!getDirectRetryTarget(packet, next_hop_hash, next_hop_hash_len,
+                                   progress_marker, expect_path_growth)
+             || !allowDirectRetry(packet, next_hop_hash, next_hop_hash_len)) {
     return;
   }
 
@@ -1275,6 +1323,7 @@ void Mesh::maybeScheduleDirectRetry(const Packet* packet, uint8_t priority) {
   _direct_retries[slot_idx].priority = priority;
   _direct_retries[slot_idx].progress_marker = progress_marker;
   _direct_retries[slot_idx].expect_path_growth = expect_path_growth;
+  _direct_retries[slot_idx].final_hop_retry = final_hop_retry;
   _direct_retries[slot_idx].waiting_final_echo = false;
   _direct_retries[slot_idx].queued = false;
   _direct_retries[slot_idx].active = true;

src/Mesh.h

@@ -66,6 +66,7 @@ class Mesh : public Dispatcher {
     uint8_t priority;
     uint8_t progress_marker;
     bool expect_path_growth;
+    bool final_hop_retry;
     bool waiting_final_echo;
     bool queued;
     bool active;
@@ -103,7 +104,7 @@ class Mesh : public Dispatcher {
   bool getDirectRetryTarget(const Packet* packet, const uint8_t*& next_hop_hash, uint8_t& next_hop_hash_len,
                             uint8_t& progress_marker, bool& expect_path_growth) const;
   bool canDecodeDirectPayloadForSelf(const Packet* packet);
-  void maybeScheduleDirectRetry(const Packet* packet, uint8_t priority);
+  void maybeScheduleDirectRetry(const Packet* packet, uint8_t priority, bool final_hop_retry = false);
   void clearFloodRetrySlot(int idx);
   bool isFloodRetryQueued(const Packet* packet) const;
   bool cancelFloodRetryOnEcho(const Packet* packet);
@@ -118,6 +119,7 @@ protected:
   void onSendComplete(Packet* packet) override;
   void onSendFail(Packet* packet) override;
   bool allowPacketTransmit(const Packet* packet) const override;
+  bool usePassiveChannelCheck(const Packet* packet) const override;
 
   virtual uint32_t getCADFailRetryDelay() const override;
 

src/helpers/RxReservePacketManager.h

@@ -51,6 +51,19 @@ class RxReservePacketManager : public StaticPoolPacketManager {
     return false;
   }
 
+  void expireStaleOutbound(uint32_t now) {
+    for (int i = getOutboundTotal() - 1; i >= 0; i--) {
+      mesh::Packet* pkt = getOutboundByIdx(i);
+      uint32_t scheduled_for;
+      if (pkt && lookupAge(pkt, &scheduled_for)
+          && (int32_t)(now - scheduled_for) > (int32_t)STALE_OUTBOUND_MS) {
+        MESH_DEBUG_PRINTLN("RxReservePacketManager: dropping stale queued outbound");
+        removeOutboundByIdx(i);
+        free(pkt);
+      }
+    }
+  }
+
 public:
   RxReservePacketManager(int pool_size, int rx_reserve)
     : StaticPoolPacketManager(pool_size), _rx_reserve(rx_reserve),
@@ -73,19 +86,17 @@ public:
   }
 
   mesh::Packet* getNextOutbound(uint32_t now) override {
-    // Expire queued packets that have waited too long past their scheduled time.
-    for (int i = getOutboundTotal() - 1; i >= 0; i--) {
-      mesh::Packet* pkt = getOutboundByIdx(i);
-      uint32_t scheduled_for;
-      if (pkt && lookupAge(pkt, &scheduled_for)
-          && (int32_t)(now - scheduled_for) > (int32_t)STALE_OUTBOUND_MS) {
-        MESH_DEBUG_PRINTLN("RxReservePacketManager: dropping stale queued outbound");
-        removeOutboundByIdx(i);
-        free(pkt);
-      }
-    }
+    expireStaleOutbound(now);
     return StaticPoolPacketManager::getNextOutbound(now);
   }
+
+
+  mesh::Packet* peekNextOutbound(uint32_t now) override {
+    // Dispatcher chooses active/passive channel sensing from this pointer, so
+    // apply the same expiry policy as getNextOutbound() before exposing it.
+    expireStaleOutbound(now);
+    return StaticPoolPacketManager::peekNextOutbound(now);
+  }
 };
 
 // The packet manager for an app build: observer builds reserve a quarter of the pool

src/helpers/SimpleMeshTables.h

@@ -10,6 +10,10 @@
 #endif
 
 #define MAX_PACKET_HASHES  (128+32)
+#ifndef MAX_PACKET_ACKS
+  #define MAX_PACKET_ACKS  64
+#endif
+#define ACK_VALID_BYTES  ((MAX_PACKET_ACKS + 7) / 8)
 #ifndef MESH_ENABLE_RECENT_REPEATERS
   #define MESH_ENABLE_RECENT_REPEATERS  0
 #endif
@@ -52,6 +56,9 @@ public:
 private:
   uint8_t _hashes[MAX_PACKET_HASHES*MAX_HASH_SIZE];
   int _next_idx;
+  uint32_t _acks[MAX_PACKET_ACKS];
+  uint8_t _ack_valid[ACK_VALID_BYTES];
+  uint8_t _next_ack_idx;
   uint32_t _direct_dups, _flood_dups;
   RecentRepeaterInfo _recent_repeaters[RECENT_REPEATER_STORAGE_SLOTS];
 
@@ -70,6 +77,48 @@ private:
     _next_idx = (_next_idx + 1) % MAX_PACKET_HASHES;
   }
 
+  bool isAckPacket(const mesh::Packet* packet) const {
+    return packet->getPayloadType() == PAYLOAD_TYPE_ACK && packet->payload_len >= sizeof(uint32_t);
+  }
+
+  bool isAckSlotValid(uint8_t idx) const {
+    return (_ack_valid[idx >> 3] & (uint8_t)(1U << (idx & 7))) != 0;
+  }
+
+  void setAckSlotValid(uint8_t idx, bool valid) {
+    uint8_t mask = (uint8_t)(1U << (idx & 7));
+    if (valid) {
+      _ack_valid[idx >> 3] |= mask;
+    } else {
+      _ack_valid[idx >> 3] &= (uint8_t)~mask;
+    }
+  }
+
+  bool hasSeenAck(uint32_t ack) const {
+    for (uint8_t i = 0; i < MAX_PACKET_ACKS; i++) {
+      if (isAckSlotValid(i) && _acks[i] == ack) {
+        return true;
+      }
+    }
+    return false;
+  }
+
+  void storeAck(uint32_t ack) {
+    if (hasSeenAck(ack)) return;
+    _acks[_next_ack_idx] = ack;
+    setAckSlotValid(_next_ack_idx, true);
+    _next_ack_idx = (uint8_t)((_next_ack_idx + 1) % MAX_PACKET_ACKS);
+  }
+
+  void clearAck(uint32_t ack) {
+    for (uint8_t i = 0; i < MAX_PACKET_ACKS; i++) {
+      if (isAckSlotValid(i) && _acks[i] == ack) {
+        setAckSlotValid(i, false);
+        return;
+      }
+    }
+  }
+
   int8_t weightedSnrX4RoundUp(int8_t curr_snr_x4, int8_t new_snr_x4) const {
     // Keep existing SNR heavier than a single new sample: 75% existing + 25% new.
     int32_t weighted_sum = ((int32_t)curr_snr_x4 * 3) + (int32_t)new_snr_x4;
@@ -152,6 +201,9 @@ public:
   SimpleMeshTables() { 
     memset(_hashes, 0, sizeof(_hashes));
     _next_idx = 0;
+    memset(_acks, 0, sizeof(_acks));
+    memset(_ack_valid, 0, sizeof(_ack_valid));
+    _next_ack_idx = 0;
     _direct_dups = _flood_dups = 0;
     memset(_recent_repeaters, 0, sizeof(_recent_repeaters));
   }
@@ -160,6 +212,10 @@ public:
   void restoreFrom(File f) {
     f.read(_hashes, sizeof(_hashes));
     f.read((uint8_t *) &_next_idx, sizeof(_next_idx));
+    // ACKs are short-lived transport state and are intentionally not persisted.
+    memset(_acks, 0, sizeof(_acks));
+    memset(_ack_valid, 0, sizeof(_ack_valid));
+    _next_ack_idx = 0;
     // Recent repeater entries are intentionally not restored across boots.
     // This avoids struct-layout migration issues and keeps stale path quality
     // stats from persisting indefinitely.
@@ -172,6 +228,20 @@ public:
 #endif
 
   bool wasSeen(const mesh::Packet* packet) override {
+    if (isAckPacket(packet)) {
+      uint32_t ack;
+      memcpy(&ack, packet->payload, sizeof(ack));
+      if (hasSeenAck(ack)) {
+        if (packet->isRouteDirect()) {
+          _direct_dups++;
+        } else {
+          _flood_dups++;
+        }
+        return true;
+      }
+      return false;
+    }
+
     uint8_t hash[MAX_HASH_SIZE];
     packet->calculatePacketHash(hash);
 
@@ -188,6 +258,13 @@ public:
   }
 
   void markSeen(const mesh::Packet* packet) override {
+    if (isAckPacket(packet)) {
+      uint32_t ack;
+      memcpy(&ack, packet->payload, sizeof(ack));
+      storeAck(ack);
+      return;
+    }
+
     uint8_t hash[MAX_HASH_SIZE];
     packet->calculatePacketHash(hash);
     if (!hasSeenHash(hash)) {
@@ -197,6 +274,13 @@ public:
   }
 
   void markSent(const mesh::Packet* packet) override {
+    if (isAckPacket(packet)) {
+      uint32_t ack;
+      memcpy(&ack, packet->payload, sizeof(ack));
+      storeAck(ack);
+      return;
+    }
+
     // Outbound packets must be marked as already-sent without teaching the recent-heard cache about ourselves.
     uint8_t hash[MAX_HASH_SIZE];
     packet->calculatePacketHash(hash);
@@ -206,6 +290,13 @@ public:
   }
 
   void clear(const mesh::Packet* packet) override {
+    if (isAckPacket(packet)) {
+      uint32_t ack;
+      memcpy(&ack, packet->payload, sizeof(ack));
+      clearAck(ack);
+      return;
+    }
+
     uint8_t hash[MAX_HASH_SIZE];
     packet->calculatePacketHash(hash);
 

src/helpers/StaticPoolPacketManager.cpp

@@ -43,6 +43,19 @@ mesh::Packet* PacketQueue::get(uint32_t now) {
   return top;
 }
 
+mesh::Packet* PacketQueue::peek(uint32_t now) const {
+  uint8_t min_pri = 0xFF;
+  int best_idx = -1;
+  for (int j = 0; j < _num; j++) {
+    if ((int32_t)(_schedule_table[j] - now) > 0) continue;
+    if (_pri_table[j] < min_pri) {
+      min_pri = _pri_table[j];
+      best_idx = j;
+    }
+  }
+  return best_idx < 0 ? NULL : _table[best_idx];
+}
+
 mesh::Packet* PacketQueue::removeByIdx(int i) {
   if (i >= _num) return NULL;  // invalid index
 
@@ -96,6 +109,10 @@ mesh::Packet* StaticPoolPacketManager::getNextOutbound(uint32_t now) {
   return send_queue.get(now);
 }
 
+mesh::Packet* StaticPoolPacketManager::peekNextOutbound(uint32_t now) {
+  return send_queue.peek(now);
+}
+
 int  StaticPoolPacketManager::getOutboundCount(uint32_t now) const {
   return send_queue.countBefore(now);
 }

src/helpers/StaticPoolPacketManager.h

@@ -11,6 +11,7 @@ class PacketQueue {
 public:
   PacketQueue(int max_entries);
   mesh::Packet* get(uint32_t now);
+  mesh::Packet* peek(uint32_t now) const;
   bool add(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for);
   int count() const { return _num; }
   int countBefore(uint32_t now) const;
@@ -28,6 +29,7 @@ public:
   void free(mesh::Packet* packet) override;
   bool queueOutbound(mesh::Packet* packet, uint8_t priority, uint32_t scheduled_for) override;
   mesh::Packet* getNextOutbound(uint32_t now) override;
+  mesh::Packet* peekNextOutbound(uint32_t now) override;
   int getOutboundCount(uint32_t now) const override;
   int getOutboundTotal() const override;
   int getFreeCount() const override;

src/helpers/radiolib/RadioLibWrappers.cpp

@@ -404,6 +404,21 @@ bool RadioLibWrapper::isChannelActive() {
   return false;
 }
 
+bool RadioLibWrapper::isReceivingPassive(int interference_margin_db) {
+  // RX duty-cycle radios hold BUSY high while asleep. Do not attempt an SPI
+  // IRQ/RSSI read until the next listen window. Treat the unknown channel as
+  // busy so the retry is deferred instead of transmitting blind; Dispatcher
+  // retains its bounded busy timeout as a last-resort escape.
+  if (isChipBusy()) return true;
+  if (isReceivingPacket()) return true;
+
+  // Use a fixed margin for retries even when the normal interference threshold
+  // is disabled. This is passive (RSSI only): unlike CAD it does not restart RX
+  // and cannot erase the forwarding echo that would cancel the retry.
+  return interference_margin_db > 0
+      && getCurrentRSSI() - _noise_floor >= interference_margin_db;
+}
+
 uint8_t RadioLibWrapper::getRadioState() const {
   return state;
 }

src/helpers/radiolib/RadioLibWrappers.h

@@ -112,6 +112,7 @@ public:
 
     return isChannelActive();
   }
+  bool isReceivingPassive(int interference_margin_db) override;
 
   virtual void setParams(float freq, float bw, uint8_t sf, uint8_t cr) = 0;
   uint16_t getDefaultPreambleLength() const override { return preambleLengthForSF(_preamble_sf); }