Keep bridged OTA retries resumable across profile handoffs

Let the mesh timed OTA retry guard own byte-identical requests instead of the permanent bridge FIFO. Bind bridge ingress to its primary profile generation and preserve UART reflection/liveness protections. Verified a LoRa to ESP-NOW to LoRa update with outage recovery and install.

examples/simple_repeater/MyMesh.h

@@ -1239,6 +1239,7 @@ public:
 #if defined(WITH_BRIDGE)
 
   void configureBridgeFilter(AbstractBridge* active_bridge) {
+    active_bridge->setIngressProfiles(_radio->profiles());
 #if MESH_ENABLE_FLOOD_RULE_ENGINE
     active_bridge->setPacketFilter([](void* context, const mesh::Packet* packet) {
       return static_cast<MyMesh*>(context)->allowTransportPacket(packet, 2);  // RULE_MODE_BRIDGE

src/helpers/AbstractBridge.h

@@ -6,8 +6,25 @@ class AbstractBridge {
   using PacketFilter = bool (*)(void*, const mesh::Packet*);
   PacketFilter _packet_filter = nullptr;
   void* _packet_filter_context = nullptr;
+  const mesh::RadioProfiles* _ingress_profiles = nullptr;
 
 protected:
+  // Bridges enter the local primary MeshCore network, not whichever tuple
+  // the physical scanner currently visits. Snapshot its generation at queue
+  // admission so a delayed frame cannot acquire a newer temporary lease.
+  bool preparePacketIngress(mesh::Packet* packet) const {
+    if (!packet || (_ingress_profiles
+        && _ingress_profiles->params(0).protocol != mesh::RadioProfileProtocol::MeshCore)) return false;
+    packet->radio_profile = packet->radio_origin = 0;
+    packet->radio_generation = packet->radio_origin_generation =
+        _ingress_profiles ? _ingress_profiles->generation[0] : 0;
+    packet->radio_bound = false;
+    packet->radio_local = false;
+    packet->tx_radio = mesh::RADIO_TX_AUTO;
+    packet->radio_reply = packet->radio_reply_force = false;
+    return true;
+  }
+
   // Called on the mesh loop, before copying/marking/queuing a packet. Never
   // invoke the policy from a WiFi callback or the MQTT publishing task.
   bool allowsPacket(const mesh::Packet* packet) const {
@@ -21,6 +38,10 @@ protected:
 public:
   virtual ~AbstractBridge() {}
 
+  void setIngressProfiles(const mesh::RadioProfiles* profiles) {
+    _ingress_profiles = profiles;
+  }
+
   void setPacketFilter(PacketFilter filter, void* context) {
     _packet_filter = filter;
     _packet_filter_context = context;

src/helpers/bridges/BridgeBase.cpp

@@ -39,13 +39,17 @@ void BridgeBase::handleReceivedPacket(mesh::Packet *packet) {
     return;
   }
 
-  if (!allowsPacket(packet)) {
+  if (!preparePacketIngress(packet) || !allowsPacket(packet)) {
     _mgr->free(packet);
     return;
   }
 
-  if (!_seen_packets.wasSeen(packet)) {
-    _seen_packets.markSeen(packet);
+  // OTA retries deliberately reuse the same request/data bytes. The mesh
+  // owns their bounded retry/longer-path echo cache and hop ceiling; the
+  // ordinary bridge FIFO has no expiry and would pin a lost burst forever.
+  const bool ota = packet->getPayloadType() == PAYLOAD_TYPE_OTA;
+  if (ota || !_seen_packets.wasSeen(packet)) {
+    if (!ota) _seen_packets.markSeen(packet);
     // bridge_delay provides a buffer to prevent immediate processing conflicts in the mesh network.
     _mgr->queueInbound(packet, millis() + _prefs->bridge_delay);
   } else {

src/helpers/bridges/BridgeBase.h

@@ -110,9 +110,10 @@ protected:
    * @brief Common packet handling for received packets
    *
    * Implements the standard pattern used by all bridges:
-   * - Check if packet was seen before using _seen_packets.wasSeen()
-   * - Queue packet for mesh processing if not seen before
-   * - Free packet if already seen to prevent duplicates
+   * - Bind ingress to the current local primary-profile generation
+   * - Apply transport policy before queueing or marking a packet
+   * - Deduplicate ordinary packets using _seen_packets.wasSeen()
+   * - Leave resumable OTA retries to the mesh's timed retry/echo policy
    *
    * @param packet The received mesh packet
    */

src/helpers/bridges/ESPNowBridge.cpp

@@ -618,7 +618,10 @@ void ESPNowBridge::sendPacket(mesh::Packet *packet) {
 
   if (!allowsPacket(packet)) return;
 
-  if (!_seen_packets.wasSeen(packet)) {
+  // Core OTA forwarding already bounds repeated requests and rejects longer
+  // downstream echoes. Do not permanently suppress its byte-identical retry.
+  const bool ota = packet->getPayloadType() == PAYLOAD_TYPE_OTA;
+  if (ota || !_seen_packets.wasSeen(packet)) {
     // Check the serialized size before writing into the ESP-NOW-sized buffer.
     const int expectedMeshPacketLen = packet->getRawLength();
     const size_t maxPayload =
@@ -670,7 +673,7 @@ void ESPNowBridge::sendPacket(mesh::Packet *packet) {
     }
 
     if (encoded && queueTransmitFrames(frames, *packet)) {
-      _seen_packets.markSeen(packet);
+      if (!ota) _seen_packets.markSeen(packet);
       BRIDGE_DEBUG_PRINTLN("TX %s, len=%d, frames=%u\n",
                            mesh::bridge::espNowFormatName(_active_format),
                            meshPacketLen, (unsigned)frames.count);

src/helpers/bridges/RS232Bridge.cpp

@@ -293,9 +293,10 @@ void RS232Bridge::drainReceivedPackets() {
   _rx_packet_ready = false;
   if (valid) {
     // A cable loopback or replay of our own accepted mesh TX is not peer proof.
-    // Discard here to avoid testing/counting the same duplicate a second time
-    // inside BridgeBase::handleReceivedPacket().
-    if (_seen_packets.wasSeen(pkt)) {
+    // OTA must still reach the mesh's timed retry/echo policy, but it cannot
+    // refresh UART peer liveness merely by reflecting our own transmission.
+    const bool seen = _seen_packets.wasSeen(pkt);
+    if (pkt->getPayloadType() != PAYLOAD_TYPE_OTA && seen) {
       _mgr->free(pkt);
       return;
     }
@@ -304,18 +305,20 @@ void RS232Bridge::drainReceivedPackets() {
     // peer may have replaced the negotiated one. Unarm the safe default rather
     // than eventually sleeping that quiet legacy link. A later fresh ACK can
     // immediately restore negotiated capability; force mode remains explicit.
-    if (_have_ack && (!_ack_fresh
-        || static_cast<uint32_t>(_rx_packet_received_ms - _last_ack)
-            > mesh::bridge::rs232heartbeat::ACTIVE_PING_MS
-                + mesh::bridge::rs232heartbeat::ACK_MAX_AGE_MS)) {
-      _have_ack = false;
-      _peer_session = 0;
+    if (!seen) {
+      if (_have_ack && (!_ack_fresh
+          || static_cast<uint32_t>(_rx_packet_received_ms - _last_ack)
+              > mesh::bridge::rs232heartbeat::ACTIVE_PING_MS
+                  + mesh::bridge::rs232heartbeat::ACK_MAX_AGE_MS)) {
+        _have_ack = false;
+        _peer_session = 0;
+      }
+      _legacy_rx = true;
+      _legacy_fresh = true;
+      _last_legacy_rx = _rx_packet_received_ms;
+      _power.notePeer(_rx_packet_received_ms);
+      _partial_suspend_grace = false;
     }
-    _legacy_rx = true;
-    _legacy_fresh = true;
-    _last_legacy_rx = _rx_packet_received_ms;
-    _power.notePeer(_rx_packet_received_ms);
-    _partial_suspend_grace = false;
     onPacketReceived(pkt);
   }
   else _mgr->free(pkt);
@@ -431,7 +434,10 @@ void RS232Bridge::sendPacket(mesh::Packet *packet) {
   drainTransmit(millis(), true);
   if (_tx_size != 0) return; // Never interleave mesh traffic with a partial frame.
 
-  if (!_seen_packets.wasSeen(packet)) {
+  // The mesh's timed OTA retry cache, not the permanent bridge FIFO, owns
+  // duplicate suppression for resumable request/data packets.
+  const bool ota = packet->getPayloadType() == PAYLOAD_TYPE_OTA;
+  if (ota || !_seen_packets.wasSeen(packet)) {
     const int expected = packet->getRawLength();
     if (expected <= 0 || expected > MAX_TRANS_UNIT + 1) return;
     uint8_t* buffer = _tx_buffer;
@@ -460,6 +466,8 @@ void RS232Bridge::sendPacket(mesh::Packet *packet) {
     _tx_size = static_cast<uint16_t>(len + SERIAL_OVERHEAD);
     _tx_pos = 0;
     _tx_is_ping = false;
+    // Retain an outgoing OTA hash solely to reject reflected data as UART
+    // peer-liveness proof. It never suppresses OTA delivery in either direction.
     _seen_packets.markSeen(packet);
     drainTransmit(millis(), true);
 

test/fixtures/rs232_heartbeat/helpers/bridges/BridgeBase.h

@@ -4,6 +4,7 @@
 #include <vector>
 
 #define MAX_TRANS_UNIT 255
+#define PAYLOAD_TYPE_OTA 0x0c
 #define BRIDGE_DEBUG_PRINTLN(...) ((void)0)
 struct NodePrefs { uint32_t bridge_baud = 115200; };
 
@@ -11,6 +12,7 @@ namespace mesh {
 struct Packet {
   std::vector<uint8_t> payload{1, 2, 3};
   bool allowed = true;
+  uint8_t getPayloadType() const { return payload.empty() ? 0 : (payload[0] >> 2) & 0x0f; }
   int getRawLength() const { return static_cast<int>(payload.size()); }
   uint16_t writeTo(uint8_t* out) const {
     for (size_t i = 0; i < payload.size(); ++i) out[i] = payload[i];

test/fixtures/rs232_mqtt_runtime.cpp

@@ -9,6 +9,7 @@
 #include <helpers/bridges/ESPNowBridgeFormat.h>
 #include <helpers/bridges/RS232PowerSettings.h>
 struct FakeRng { void random(uint8_t* out, size_t size) { memset(out, 0x42, size); } };
+struct FakeRadio { const void* profiles() const { return nullptr; } };
 #define BRIDGE_MAX_BAUD 115200
 #define PUB_KEY_SIZE 32
 #define MESH_DEBUG_PRINTLN(...) ((void)0)
@@ -64,6 +65,7 @@ public:
   void setPacketFilter(bool (*check)(void*, const mesh::Packet*), void* context) {
     filter = check; filter_context = context; ++filters;
   }
+  void setIngressProfiles(const void*) {}
 };
 class RS232Bridge : public AbstractBridge {
   NodePrefs* prefs;
@@ -217,7 +219,8 @@ public:
   FakeCLI _cli;
   FakeAlerter _alerter;
   mesh::LocalIdentity self_id;
-  void* _mgr = nullptr; void* _radio = nullptr; void* _ms = nullptr;
+  FakeRadio radio; FakeRadio* _radio = &radio;
+  void* _mgr = nullptr; void* _ms = nullptr;
   void* getRTCClock() { return nullptr; }
   FakeRng rng;
   FakeRng* getRNG() { return &rng; }

test/test_bridge_ingress.py

@@ -0,0 +1,116 @@
+"""Execute real bridge admission and dispatcher/OTA lease guards without RF."""
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+from test_radio_receive_contract import method
+from test_transport_filter_wiring import HARNESS, PACKET
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class BridgeIngressTests(unittest.TestCase):
+    def test_primary_affinity_stale_queue_and_no_secondary_lease_appropriation(self):
+        base = (ROOT / "src/helpers/bridges/BridgeBase.cpp").read_text()
+        dispatcher = (ROOT / "src/Dispatcher.cpp").read_text()
+        mesh = (ROOT / "src/Mesh.cpp").read_text()
+        receive = method(dispatcher, "void Dispatcher::processRecvPacket(")
+        guard = receive[receive.index("  const auto* profiles ="):receive.index("  const uint8_t previous_profile")]
+        ota_start = mesh.index("    case PAYLOAD_TYPE_OTA: {")
+        ota_guard = mesh[mesh.index("      if (!isAnyTempRadioActive()", ota_start):
+                         mesh.index("      observeOtaRequestPressure(pkt);", ota_start)]
+        ota_guard = ota_guard.replace(" break;", " return false;")
+        code = HARNESS[:HARNESS.index("@METHODS@")]
+        code += method(base, "void BridgeBase::handleReceivedPacket(")
+        code += r'''
+using mesh::RadioCrossMode;
+using mesh::RadioProfileProtocol;
+struct Radio {
+ mesh::RadioProfiles* state;
+ const mesh::RadioProfiles* profiles() const {return state;}
+};
+struct Dispatcher {
+ Radio* _radio; unsigned accepted=0,released=0;
+ void releasePacket(mesh::Packet*) {++released;}
+ void receive(mesh::Packet* pkt) {
+@DISPATCH_GUARD@
+  ++accepted;
+ }
+ bool isAnyTempRadioActive() const {
+  return _radio->state && (_radio->state->primary_temporary || _radio->state->secondary_temporary);
+ }
+ bool otaEligible(mesh::Packet* pkt) {
+@OTA_GUARD@
+  return true;
+ }
+};
+int main() {
+ mesh::RadioProfiles profiles;
+ profiles.generation[0]=12;profiles.generation[1]=93;
+ profiles.primary_temporary=false;profiles.secondary_temporary=true;
+ profiles.cross=mesh::RadioCrossMode::Off;
+ BridgeBase bridge;bridge.setIngressProfiles(&profiles);
+ mesh::Packet packet;
+ packet.header=(PAYLOAD_TYPE_OTA<<2)|1;
+ packet.radio_profile=packet.radio_origin=1;
+ packet.radio_generation=packet.radio_origin_generation=93;
+ packet.radio_bound=true;packet.radio_local=true;packet.tx_radio=mesh::RADIO_TX_BOTH;
+ packet.radio_reply=packet.radio_reply_force=true;
+ bridge.handleReceivedPacket(&packet);
+ assert(bridge.manager.queued==1&&bridge.manager.freed==0);
+ assert(packet.radio_profile==0&&packet.radio_origin==0);
+ assert(packet.radio_generation==12&&packet.radio_origin_generation==12);
+ assert(!packet.radio_bound&&!packet.radio_local&&packet.tx_radio==mesh::RADIO_TX_AUTO);
+ assert(!packet.radio_reply&&!packet.radio_reply_force);
+ Radio radio{&profiles};Dispatcher dispatch{&radio};
+ dispatch.receive(&packet);assert(dispatch.accepted==1&&dispatch.released==0);
+ // A secondary-only update window is not permission to steal that lease.
+ assert(!dispatch.otaEligible(&packet));
+ profiles.primary_temporary=true;assert(dispatch.otaEligible(&packet));
+ // Frames already admitted with bridge.delay retain the old primary lease.
+ profiles.generation[0]=13;
+ dispatch.receive(&packet);assert(dispatch.accepted==1&&dispatch.released==1);
+ assert(packet.radio_generation==12&&packet.radio_origin_generation==12);
+ mesh::Packet fresh;fresh.header=packet.header;bridge.handleReceivedPacket(&fresh);
+ assert(fresh.radio_generation==13&&fresh.radio_profile==0);
+ dispatch.receive(&fresh);assert(dispatch.accepted==2&&dispatch.released==1);
+ // An explicit cross policy remains the sole opt-in to the other temp tuple.
+ profiles.primary_temporary=false;profiles.cross=mesh::RadioCrossMode::On;
+ assert(dispatch.otaEligible(&fresh));profiles.cross=mesh::RadioCrossMode::Off;
+ // A foreign primary tuple cannot reinterpret bridge MeshCore bytes.
+ profiles.primary.protocol=mesh::RadioProfileProtocol::Meshtext;
+ const auto queued=bridge.manager.queued;
+ bridge.handleReceivedPacket(&fresh);
+ assert(bridge.manager.queued==queued&&bridge.manager.freed==1);
+ // Legacy single-profile radios have no generation table; preserve that path.
+ bridge.setIngressProfiles(nullptr);radio.state=nullptr;
+ bridge.handleReceivedPacket(&fresh);assert(fresh.radio_generation==0&&!fresh.radio_local);
+ dispatch.receive(&fresh);assert(dispatch.accepted==3&&dispatch.released==1);
+}
+'''
+        code = code.replace("@DISPATCH_GUARD@", guard).replace("@OTA_GUARD@", ota_guard)
+        with tempfile.TemporaryDirectory(prefix="meshcore-bridge-ingress-") as directory:
+            work = Path(directory)
+            (work / "Mesh.h").write_text(PACKET)
+            source, binary = work / "main.cpp", work / "test"
+            source.write_text(code)
+            built = subprocess.run([
+                os.environ.get("CXX", "g++"), "-std=c++17", "-O1", "-Wall", "-Wextra", "-Werror",
+                "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie",
+                "-I", str(work), "-I", str(ROOT / "src"), str(source), "-o", str(binary),
+            ], capture_output=True, text=True, timeout=60)
+            self.assertEqual(built.returncode, 0, built.stdout + built.stderr)
+            checked = subprocess.run([str(binary)], capture_output=True, text=True, timeout=15)
+            self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+
+    def test_every_role_bridge_configuration_binds_the_primary_profiles(self):
+        header = (ROOT / "examples/simple_repeater/MyMesh.h").read_text()
+        configure = method(header, "void configureBridgeFilter(")
+        self.assertIn("active_bridge->setIngressProfiles(_radio->profiles());", configure)
+        self.assertLess(configure.index("setIngressProfiles"), configure.index("#if MESH_ENABLE_FLOOD_RULE_ENGINE"))
+
+
+if __name__ == "__main__":
+    unittest.main()

test/test_bridge_recovery.py

@@ -17,6 +17,7 @@ MOCKS = r'''
 #include <cstdio>
 #include <helpers/bridges/RS232Heartbeat.h>
 #define MAX_TRANS_UNIT 255
+#define PAYLOAD_TYPE_OTA 0x0c
 using portMUX_TYPE=int;
 #define portMUX_INITIALIZER_UNLOCKED 0
 struct esp_now_recv_info_t {const uint8_t* src_addr;};
@@ -41,6 +42,7 @@ struct Packet {
  }
  int getRawLength()const{return length;}
  uint16_t writeTo(uint8_t* out)const{memcpy(out,data,length);return length;}
+ uint8_t getPayloadType()const{return (data[0]>>2)&0x0f;}
  bool violatesRoutePolicy()const{return false;}
 };
 struct PacketManager {
@@ -78,6 +80,7 @@ public:
  virtual void begin()=0;virtual void end()=0;virtual void loop()=0;
  virtual void sendPacket(mesh::Packet*)=0;virtual void onPacketReceived(mesh::Packet*)=0;
  bool allowsPacket(const mesh::Packet*)const{return allowed;}
+ bool preparePacketIngress(mesh::Packet*)const{return true;}
  static uint16_t fletcher16(const uint8_t*,size_t);
  bool validateChecksum(const uint8_t*,size_t,uint16_t);
  void handleReceivedPacket(mesh::Packet*);
@@ -161,6 +164,42 @@ class BridgeRecoveryTests(unittest.TestCase):
                     result = subprocess.run([str(exe)], capture_output=True, text=True)
                     self.assertEqual(result.returncode, 0, result.stderr)
 
+    def test_ota_identical_retries_cross_espnow_and_uart_without_permanent_dedup(self):
+        self.run_fixture(r'''
+int main(){
+ Backend backend;backend.wifi=true;mesh::wireless::control().begin(backend);
+ sdk_initialized=sdk_started=true;sdk_mode=WIFI_STA;
+ NodePrefs prefs;mesh::PacketManager mgr;mesh::RTCClock rtc;
+ ESPNowBridge espnow(&prefs,&mgr,&rtc);espnow.begin();
+ const uint8_t mac[6]={1,2,3,4,5,6};
+ auto ota=packet(0x31); // flood-framed PAYLOAD_TYPE_OTA, byte-identical retry
+ espnow._seen_packets.markSeen(&ota); // even a preexisting FIFO row cannot pin it
+ espnow.sendPacket(&ota);espnow.loop();complete(espnow);
+ assert(sent.size()==1&&espnow._tx_count==0);
+ espnow.sendPacket(&ota);espnow.loop();complete(espnow);
+ assert(sent.size()==2&&sent[0]==sent[1]);
+ espnow.queueReceivedFrame(mac,ota.data,ota.length);espnow.loop();
+ espnow.queueReceivedFrame(mac,ota.data,ota.length);espnow.loop();
+ assert((mgr.received==std::vector<uint8_t>{0x31,0x31}));
+ // Ordinary message duplicates still stop in both bridge directions.
+ auto chat=packet(0x15);espnow.sendPacket(&chat);espnow.loop();complete(espnow);
+ const auto count=sent.size();espnow.sendPacket(&chat);espnow.loop();
+ assert(sent.size()==count);
+ espnow.queueReceivedFrame(mac,chat.data,chat.length);espnow.loop();
+ assert(mgr.received.size()==2);espnow.end();
+ HardwareSerial serial;RS232Bridge uart(&prefs,serial,1,2,&mgr,&rtc);uart.begin();
+ uart._seen_packets.markSeen(&ota);uart.sendPacket(&ota);uart.sendPacket(&ota);
+ auto doubled=uart_frame(ota);const auto tail=doubled;
+ doubled.insert(doubled.end(),tail.begin(),tail.end());assert(serial.output==doubled);
+ append(serial,uart_frame(ota));append(serial,uart_frame(ota));uart.loop();
+ assert((mgr.received==std::vector<uint8_t>{0x31,0x31,0x31,0x31}));
+ assert(!uart._legacy_rx); // reflected OTA is not proof of an external peer
+ uart.sendPacket(&chat);const auto length=serial.output.size();uart.sendPacket(&chat);
+ assert(serial.output.size()==length);append(serial,uart_frame(chat));uart.loop();
+ assert(mgr.received.size()==4&&mgr.allocs==mgr.frees);
+}
+''')
+
     def test_uart_pool_pressure_order_parser_bounds_and_restart(self):
         self.run_fixture(r'''
 int main(){

test/test_transport_filter_wiring.py

@@ -13,9 +13,14 @@ PACKET = r'''
 #include <cstdint>
 #include <cstddef>
 #include <cstring>
+#include <RadioProfiles.h>
+#include <RadioTxPolicy.h>
 namespace mesh {
 struct Packet {
  uint8_t payload[64] = {}, payload_len = 32, type = 5, header = (5 << 2);
+ uint8_t radio_profile=0,radio_origin=0,tx_radio=RADIO_TX_AUTO;
+ uint32_t radio_generation=0,radio_origin_generation=0;
+ bool radio_bound=false,radio_local=true,radio_reply=false,radio_reply_force=false;
  uint8_t getPayloadType() const { return (header >> 2) & 0x0f; }
  bool violatesRoutePolicy() const {
    const uint8_t route = header & 0x03;
@@ -38,6 +43,7 @@ HARNESS = r'''
 #define BRIDGE_DEBUG_PRINTLN(...) ((void)0)
 #define PUB_KEY_SIZE 32
 #define PAYLOAD_TYPE_ADVERT 4
+#define PAYLOAD_TYPE_OTA 0x0c
 #define MAX_SERIAL_PACKET_SIZE 300
 #define MAX_TRANS_UNIT 256
 #define SERIAL_OVERHEAD 6