Fix OTA resume and radio state recovery

Preserve resumed nRF52 flash pages and keep filtered OTA catalogs incomplete until full pages arrive. Bound stuck radio preparation, report companion setting failures, retain safe rollback, and codify conservative advert penalties with regression tests.

src/Dispatcher.cpp

@@ -27,6 +27,13 @@ namespace mesh {
 
 #define MIN_CAD_FAIL_RETRY_DELAY_MS       50UL
 #define MIN_CAD_FAIL_MAX_DURATION_MS     500UL
+#define RADIO_PREPARE_BUSY_GRACE_MS      8000UL
+
+static uint32_t validPrepareBusyAirtime(uint32_t airtime) {
+  // RadioLib estimates originate as 32-bit microseconds. An impossible or
+  // error-sentinel estimate must not retain a failed packet for days.
+  return airtime <= UINT32_MAX / 1000 ? airtime : 0;
+}
 
 static uint32_t scaleCADDelayForQueue(uint32_t normal_delay, int ready_count,
                                      uint32_t minimum_delay) {
@@ -65,6 +72,7 @@ void Dispatcher::begin() {
   last_meshcore_recv_millis = 0;
   _err_flags = 0;
   outbound_radio_retry_at = 0;
+  outbound_radio_prepare_deadline = 0;
   outbound_radio_retry_pending = false;
   outbound_radio_retry_used = false;
   outbound_cancellation = OutboundCancellation::None;
@@ -192,6 +200,7 @@ bool Dispatcher::scheduleOutboundRadioRetry() {
 
   outbound_radio_retry_used = true;
   outbound_radio_retry_pending = true;
+  outbound_radio_prepare_deadline = 0;
   outbound_radio_retry_at = futureMillis(getCADFailRetryDelay());
   MESH_DEBUG_PRINTLN("%s Dispatcher: retrying packet once after radio fault",
                      getLogDateTime());
@@ -216,6 +225,7 @@ void Dispatcher::failOutboundTransmit() {
   outbound = NULL;
   outbound_radio_retry_pending = false;
   outbound_radio_retry_used = false;
+  outbound_radio_prepare_deadline = 0;
   outbound_cancellation = OutboundCancellation::None;
 }
 
@@ -377,9 +387,33 @@ void Dispatcher::loop() {
       const auto prepared = _radio->prepareTransmitProfile(outbound->radio_profile,
           outbound->radio_reply && outbound->radio_reply_force);
       if (prepared == RadioParamApplyResult::BUSY) {
+        if (outbound_radio_prepare_deadline == 0) {
+          // A frame already being received may legitimately own either scan
+          // profile. Allow its longest full-packet airtime plus recovery grace.
+          // This timer covers only prepare BUSY, not ordinary CAD backoff.
+          uint32_t airtime = validPrepareBusyAirtime(_radio->getProfileAirtime(0, MAX_TRANS_UNIT));
+          if (isDualRadioActive()) {
+            const uint32_t other = validPrepareBusyAirtime(_radio->getProfileAirtime(1, MAX_TRANS_UNIT));
+            if (other > airtime) airtime = other;
+          }
+          const uint32_t timeout = airtime * 3 / 2 + RADIO_PREPARE_BUSY_GRACE_MS;
+          outbound_radio_prepare_deadline = _ms->getMillis() + timeout;
+          if (outbound_radio_prepare_deadline == 0) outbound_radio_prepare_deadline = 1;
+        } else if (millisHasNowPassed(outbound_radio_prepare_deadline)) {
+          // Retaining outbound disables the non-RX watchdog, and this backoff
+          // bypasses the liveness watchdog. Release ownership before recovery
+          // so even a failed reset cannot pin all later queue entries forever.
+          _err_flags |= ERR_EVENT_RADIO_WATCHDOG;
+          MESH_DEBUG_PRINTLN("%s Dispatcher: radio retry profile remained busy", getLogDateTime());
+          failOutboundTransmit();
+          _radio->recoverRadio(true);
+          radio_nonrx_start = _ms->getMillis();
+          return;
+        }
         outbound_radio_retry_at = futureMillis(10);
         return;
       }
+      outbound_radio_prepare_deadline = 0;
 
       if (prepared == RadioParamApplyResult::FAILED || !isPacketRadioCurrent(outbound)
           || !allowPacketTransmit(outbound)) {
@@ -1027,7 +1061,8 @@ bool Dispatcher::sendPacket(Packet* packet, uint8_t priority, uint32_t delay_mil
 // Utility function -- handles the case where millis() wraps around back to zero
 //   2's complement arithmetic will handle any unsigned subtraction up to HALF the word size (32-bits in this case)
 bool Dispatcher::millisHasNowPassed(unsigned long timestamp) const {
-  return (long)(_ms->getMillis() - timestamp) > 0;
+  // millis is a 32-bit counter even on native hosts where long is 64 bits.
+  return (int32_t)((uint32_t)_ms->getMillis() - (uint32_t)timestamp) > 0;
 }
 
 unsigned long Dispatcher::futureMillis(int millis_from_now) const {

src/Dispatcher.h

@@ -281,6 +281,7 @@ class Dispatcher {
   Packet* outbound;  // current outbound packet
   unsigned long outbound_expiry, outbound_start, total_air_time, rx_air_time;
   unsigned long outbound_radio_retry_at;
+  uint32_t outbound_radio_prepare_deadline = 0;
 #ifdef RADIO_LIVENESS_SOFT_ONLY
   unsigned long rx_watchdog_window_start;
 #else

src/Mesh.cpp

@@ -328,8 +328,11 @@ void Mesh::loop() {
     for (int i = 0; i < TOTAL_FLOOD_RETRY_SLOTS; ++i) {
       if (!_flood_retries[i].active) continue;
       const auto* packet = _flood_retries[i].queued ? _flood_retries[i].packet : _flood_retries[i].trigger_packet;
-      if ((_flood_retries[i].waiting_final_echo && _flood_retries[i].packet
-            && !isPacketRadioCurrent(_flood_retries[i].packet))
+      // Final flood echoes also retain metadata only. Release the old
+      // session's retry key even though there is no packet left to inspect.
+      const uint8_t profile = i / MAX_FLOOD_RETRY_SLOTS;
+      if ((_flood_retries[i].waiting_final_echo
+              && _retry_radio_generations[profile] != p->generation[profile])
           || (packet && !isPacketRadioCurrent(packet))) retireFloodRetrySlot(i);
     }
     for (int i = _mgr->getOutboundTotal() - 1; i >= 0; --i) {

src/helpers/FloodAdvertLimiter.cpp

@@ -130,6 +130,9 @@ void FloodAdvertLimiter::observe(const uint8_t* key, const uint8_t* hash, uint8_
   // The normal level belongs to the prefix, including its shortest RX path.
   // Keep that minimum on every colliding full key for consistent window-close
   // abuse accounting, but never combine their received counts into a strike.
+  // Be conservative once BAD is set: a later shorter path can raise the normal
+  // quota, but cannot retract an already-triggered penalty or recorded violation.
+  // Only seven-day recovery, an explicit clear, or reboot removes that penalty.
   if (hops < entry->min_hops) entry->min_hops = hops;
   for (size_t i = 0; i < _capacity; ++i) {
     if ((_entries[i].flags & ACTIVE) && memcmp(_entries[i].key, key, PREFIX_BYTES) == 0

src/helpers/ota/OtaManager.cpp

@@ -861,7 +861,8 @@ void OtaManager::invalidateCatalogSeeder(const uint8_t* seeder) {
     row.last_ms = 0;
     for (uint8_t k = 0; k < row.n_seeders; k++) {
       if (row.seeder_have[k] > row.have_max) row.have_max = row.seeder_have[k];
-      if (row.seeder_last_ms[k] > row.last_ms) row.last_ms = row.seeder_last_ms[k];
+      if (k == 0 || (uint32_t)(_now_ms - row.seeder_last_ms[k])
+          < (uint32_t)(_now_ms - row.last_ms)) row.last_ms = row.seeder_last_ms[k];
     }
     i++;
   }
@@ -888,7 +889,9 @@ void OtaManager::handleAdv(const uint8_t* m, uint16_t n) {
   int slot = -1, lru = 0;                                            // find/insert the source (LRU evict)
   for (int i = 0; i < _n_src; i++) {
     if (memcmp(_sources[i].seeder, a.seeder_id, 4) == 0) { slot = i; break; }
-    if (_sources[i].last_ms < _sources[lru].last_ms) lru = i;
+    // Compare elapsed ages: raw millis ordering reverses when uptime wraps.
+    if ((uint32_t)(_now_ms - _sources[i].last_ms)
+        > (uint32_t)(_now_ms - _sources[lru].last_ms)) lru = i;
   }
   bool fresh = (slot < 0);
   if (fresh) {
@@ -994,18 +997,25 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) {
     if (memcmp(_sources[i].digest, hv.set_digest, 4) != 0) return;
     break;
   }
-  // PASSIVE: every node caches rows it overhears. A source is catalogued only after EVERY advertised
-  // fragment arrived; otherwise a timed recovery QUERY asks for just the missing bitmap.
+  // PASSIVE: every node caches rows it overhears. Only complete canonical pages
+  // count toward discovery: filtered replies have the same fragment positions,
+  // but omit rows and cannot suppress a later unfiltered query for those pages.
+  const uint8_t per = (uint8_t)((MAX_PACKET_PAYLOAD - 12) / OTA_HAVE_ROW_BYTES);
   for (uint8_t i = 0; i < _n_src; i++) {
     Source& s = _sources[i];
     if (memcmp(s.seeder, hv.seeder_id, 4) != 0 || memcmp(s.digest, hv.set_digest, 4) != 0) continue;
-    if (s.have_total != hv.frag_total) {
-      s.have_total = hv.frag_total;
+    const uint8_t total = (uint8_t)((s.n_motas + per - 1) / per);
+    if (s.have_total != total) {
+      s.have_total = total;
       s.have_mask = 0;
       s.have_catalog = false;
     }
-    s.have_mask |= 1UL << hv.frag_idx;
-    uint32_t full = hv.frag_total >= 32 ? UINT32_MAX : ((1UL << hv.frag_total) - 1);
+    if (hv.frag_total == total && hv.frag_idx < total) {
+      const uint16_t remaining = s.n_motas - (uint16_t)hv.frag_idx * per;
+      const uint8_t expected_rows = remaining < per ? (uint8_t)remaining : per;
+      if (hv.n_rows == expected_rows) s.have_mask |= 1UL << hv.frag_idx;
+    }
+    uint32_t full = total >= 32 ? UINT32_MAX : ((1UL << total) - 1);
     s.have_catalog = s.have_mask == full;
     if (s.have_catalog) {
       s.query_pending = false;
@@ -1026,7 +1036,8 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) {
     int slot = -1, lru = 0;                                           // upsert into the catalog (dedup by mid)
     for (int i = 0; i < _n_cat; i++) {
       if (memcmp(catalog[i].mid, mid, 4) == 0) { slot = i; break; }
-      if (catalog[i].last_ms < catalog[lru].last_ms) lru = i;
+      if ((uint32_t)(_now_ms - catalog[i].last_ms)
+          > (uint32_t)(_now_ms - catalog[lru].last_ms)) lru = i;
     }
     if (slot < 0) {
       if (_n_cat >= catalogCapacity() && expandCatalog()) catalog = catalogData();
@@ -1056,7 +1067,8 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) {
     c.have_max = 0; c.last_ms = 0;
     for (uint8_t k = 0; k < c.n_seeders; k++) {
       if (c.seeder_have[k] > c.have_max) c.have_max = c.seeder_have[k];
-      if (c.seeder_last_ms[k] > c.last_ms) c.last_ms = c.seeder_last_ms[k];
+      if (k == 0 || (uint32_t)(_now_ms - c.seeder_last_ms[k])
+          < (uint32_t)(_now_ms - c.last_ms)) c.last_ms = c.seeder_last_ms[k];
     }
     if (wantRow(mid, target, fwver, codec, flags)) startFetch(mid, target);
   }

src/helpers/ota/OtaStoreFlashNrf52.cpp

@@ -69,6 +69,7 @@ void OtaStoreFlashNrf52::reset_session() {
   _ram_len = 0;
   _total = 0;
   _hybrid = false;
+  _preserve_payload_pages = false;
   _pay_idx = 0;
   _flushed = false;
   _io_ok = true;
@@ -309,7 +310,14 @@ uint8_t* OtaStoreFlashNrf52::write_slot(uint32_t pos) {
   if (page == 0)             return _meta_page + pos;
   if (page > _pay_idx) {
     if (!flush_pay()) return nullptr;
-    _pay_idx = page; memset(_pay_page, 0xFF, PG);             // advance, fresh page
+    _pay_idx = page;
+    // A resumed page can contain verified blocks on either side of this hole.
+    // Those blocks retain their leaf markers, so erasing their cached bytes here
+    // would silently corrupt them when this page is checkpointed or finalized.
+    if (_preserve_payload_pages)
+      memcpy(_pay_page, (const uint8_t*)(uintptr_t)(_write_start + page * PG), PG);
+    else
+      memset(_pay_page, 0xFF, PG);                         // fresh capture, no committed bytes yet
   }
   if (page == _pay_idx)      return _pay_page + (pos - page * PG);
   return nullptr;                                                      // page < _pay_idx: already flushed
@@ -598,6 +606,7 @@ bool OtaStoreFlashNrf52::reopen() {
     _ram_len = 0;
     _total = total;
     _hybrid = false;
+    _preserve_payload_pages = true;
     memcpy(_meta_page, p, PG);                  // load page 0 (header+manifest+leaves) into RAM to continue
     // The tail is deferred until finalize and can still hold old flash bytes.
     // Reconstruct only fixed framing; the manager must verify the manifest and

src/helpers/ota/OtaStoreFlashNrf52.h

@@ -22,6 +22,8 @@
 //     (set_meta_size enforces it; true for <= ~979 blocks, i.e. any realistic MeshCore image).
 //   - `_pay_page` is a single sliding buffer for one payload page (index >= 1). It advances
 //     monotonically with the (mostly in-order) block stream and flushes the page it leaves behind.
+//     After reopen, newly opened pages load their existing flash bytes to preserve committed blocks
+//     which share a page with missing blocks; only fresh captures initialize pages to erased bytes.
 //     Rare out-of-order writes to an already-flushed page go straight to flash as a safe read-modify-
 //     write (flash_nrf5x erases before programming, so re-touching a page never violates the
 //     writes-per-word limit - it just costs one extra erase).
@@ -42,6 +44,7 @@ class OtaStoreFlashNrf52 : public OtaStore {
   uint32_t _ram_len = 0;            // logical suffix backed by the fixed SRAM arena
   uint32_t _total = 0;              // staged container size (0 = none)
   bool     _hybrid = false;          // flash prefix + reset-retained SRAM suffix
+  bool     _preserve_payload_pages = false; // reopened pages can contain committed blocks around later holes
   bool     _flushed = false;        // finalize() committed everything to flash
   bool     _io_ok = true;           // cleared on a failed/out-of-bounds flash write or readback mismatch
   bool     _planned_bootloader = false; // manifest-kind decision made before begin()/any erase