Pipeline OTA downloads and adapt relay timing

docs/ota_protocol.md

@@ -502,7 +502,9 @@ OTA_LEAVES:        manifest_id[4]  frag_idx(1)  frag_total(1)  bytes[]      # up
   removes the burst, so there is nothing to collide with. The block/manifest mask matches the 16-bit
   reassembly bitmap (<=16 fragments/block; 1 KB blocks = 7). `OTA_PROOF` is a single packet and needs no mask.
 - **Data and proof are separate phases.** `OTA_DATA` carries no proof; the proof is fetched once per block
-  via `OTA_REQ_PROOF`/`OTA_PROOF` after the block's data is complete.
+  via `OTA_REQ_PROOF`/`OTA_PROOF` after that block's data is complete. The receiver keeps a bounded two-block
+  window, so one block can await its proof or flash commit while DATA for the next block arrives. Slots retry
+  independently and only one stalled slot is retried per service tick.
 
 ### 8.5 Sizing against `MAX_PACKET_PAYLOAD = 184`
 
@@ -519,13 +521,20 @@ payload); larger self-images pass a bigger scratch buffer.
 ### 8.6 Temporary-radio and transfer boundary
 
 OTA packets may cross normal mesh relay hops, but each participating node processes or relays them only while
-its `tempradio` window is actually running. A receiver requests missing blocks in serial order from the offered
-firmware source. It never serves partial blocks. A normal install receiver never re-advertises its completed
+its `tempradio` window is actually running. A receiver selects missing blocks in serial order into a bounded
+two-block pipeline. It never serves partial blocks. A normal install receiver never re-advertises its completed
 download. An SD archive node is the deliberate exception: after a fully proof-verified container is published
 to its persistent archive, it registers that complete file as a MotaSource and advertises it as a new seeder.
 This keeps each active transfer as one transmitter and one receiver while still allowing active temporary-radio
 repeaters between them and persistent archive nodes to improve future availability.
 
+Because TempRadio is a bounded channel dedicated to this transfer, an accepted OTA flood normally relays after
+a randomized 0.25 to 0.5 packet-airtime delay. Repeated requests for the same manifest block or proof provide
+congestion feedback without changing the wire format. Frequent retries widen the local relay window through
+0.5-1.0, 0.75-2.0, and finally 1.0-3.0 airtimes. The maximum is hard-capped at three packet airtimes. Each
+30-second interval without another observed retry lowers the window one level. Other flood payloads retain the
+role's normal randomized delay, and CAD still applies when enabled.
+
 ---
 
 ## 9. Identity, trust & versioning

docs/ota_user_guide.md

@@ -149,7 +149,7 @@ re-running a `validate` pull re-begins fresh (it never resumes a stale partial).
 trusted - every kept block is checked against the target's own fingerprints, so a mismatched or missing seed
 just means those blocks are fetched over the radio (correct result, only slower).
 
-The node fetches from one source, **at low priority**, one block at a time. Mesh repeaters may carry the
+The node fetches from one source, **at low priority**, with a bounded two-block receive window. Mesh repeaters may carry the
 packets, but only while their temporary-radio windows are active. Check progress with `ota status`.
 
 If a `folder` pull loses its link mid-transfer, `ota status` shows **paused** - the host keeps the

src/Mesh.cpp

@@ -1,4 +1,5 @@
 #include "Mesh.h"
+#include "helpers/ota/OtaFormat.h"   // request types used by TempRadio relay-pressure tracking
 //#include <Arduino.h>
 #if defined(ENABLE_OTA)
 #include "helpers/ota/OtaContext.h"   // OTA mesh-integration is centralized here so every role gets it
@@ -37,6 +38,10 @@ static const uint8_t FLOOD_RETRY_MAX_PATH_DEFAULT = 1;
 static const uint32_t ORIGIN_ADVERT_RETRY_EXTRA_DELAY_MS = 60UL * 1000UL;
 static const uint32_t RECENT_ADVERT_MAX_AGE_SECONDS = 6UL * 60UL * 60UL;
 static const uint32_t FORWARDED_ADVERT_ECHO_WATCH_MS = 5UL * 60UL * 1000UL;
+static const uint32_t OTA_RETRY_OBSERVE_MIN_MS = 750UL;
+static const uint32_t OTA_RETRY_OBSERVE_MAX_MS = 30000UL;
+static const uint32_t OTA_RETRY_ESCALATE_MS = 6000UL;
+static const uint32_t OTA_RELAY_DECAY_MS = 30000UL;
 
 static bool hasValidEncryptedPayloadLength(uint16_t payload_len, uint16_t clear_prefix_len) {
   const uint16_t overhead = clear_prefix_len + CIPHER_MAC_SIZE;
@@ -214,6 +219,7 @@ void Mesh::begin() {
     _recent_advert_echoes[i].confirmed = false;
     _recent_advert_echoes[i].valid = false;
   }
+  resetOtaRelayBackoff();
   Dispatcher::begin();
 #if defined(ENABLE_OTA)
   uint32_t my_tid = 0;
@@ -398,6 +404,91 @@ uint32_t Mesh::getRetransmitDelay(const mesh::Packet* packet) {
 
   return _rng->nextInt(0, 5)*t;
 }
+void Mesh::resetOtaRelayBackoff() {
+  for (uint8_t i = 0; i < OTA_REQUEST_TRACK_SLOTS; i++) {
+    memset(_ota_request_track[i].key, 0, sizeof(_ota_request_track[i].key));
+    _ota_request_track[i].last_request_ms = 0;
+    _ota_request_track[i].min_path_hops = 0;
+    _ota_request_track[i].valid = false;
+  }
+  _ota_relay_decay_at = 0;
+  _ota_relay_backoff_level = 0;
+}
+
+void Mesh::observeOtaRequestPressure(const mesh::Packet* packet) {
+  if (packet == NULL || packet->payload_len < 7) return;
+  const uint8_t type = packet->payload[0];
+  if (type != ota::OTA_REQ && type != ota::OTA_REQ_PROOF) return;
+
+  const uint32_t now = _ms->getMillis();
+  int match = -1;
+  int replacement = 0;
+  uint32_t oldest_age = 0;
+  for (uint8_t i = 0; i < OTA_REQUEST_TRACK_SLOTS; i++) {
+    OtaRequestTrackEntry& entry = _ota_request_track[i];
+    if (entry.valid && memcmp(entry.key, packet->payload, sizeof(entry.key)) == 0) {
+      match = i;
+      break;
+    }
+    uint32_t age = entry.valid ? now - entry.last_request_ms : UINT32_MAX;
+    if (i == 0 || age > oldest_age) {
+      oldest_age = age;
+      replacement = i;
+    }
+  }
+
+  if (match < 0) {
+    OtaRequestTrackEntry& entry = _ota_request_track[replacement];
+    memcpy(entry.key, packet->payload, sizeof(entry.key));
+    entry.last_request_ms = now;
+    entry.min_path_hops = packet->getPathHashCount();
+    entry.valid = true;
+    return;
+  }
+
+  OtaRequestTrackEntry& entry = _ota_request_track[match];
+  const uint32_t elapsed = now - entry.last_request_ms;
+  const uint8_t path_hops = packet->getPathHashCount();
+  // A longer-path copy shortly after the first is a downstream forwarding echo, not client retry pressure.
+  if (path_hops > entry.min_path_hops && elapsed <= OTA_RETRY_OBSERVE_MAX_MS) return;
+
+  if (elapsed >= OTA_RETRY_OBSERVE_MIN_MS && elapsed <= OTA_RETRY_OBSERVE_MAX_MS) {
+    if (elapsed <= OTA_RETRY_ESCALATE_MS) {
+      if (_ota_relay_backoff_level < 3) _ota_relay_backoff_level++;
+    } else if (_ota_relay_backoff_level == 0) {
+      _ota_relay_backoff_level = 1;
+    }
+    _ota_relay_decay_at = now + OTA_RELAY_DECAY_MS;
+  }
+  entry.last_request_ms = now;
+  if (path_hops < entry.min_path_hops) entry.min_path_hops = path_hops;
+}
+
+void Mesh::decayOtaRelayBackoff() {
+  if (_ota_relay_backoff_level == 0 || _ota_relay_decay_at == 0) return;
+  const uint32_t now = _ms->getMillis();
+  while (_ota_relay_backoff_level > 0 && (int32_t)(now - _ota_relay_decay_at) >= 0) {
+    _ota_relay_backoff_level--;
+    _ota_relay_decay_at += OTA_RELAY_DECAY_MS;
+  }
+  if (_ota_relay_backoff_level == 0) _ota_relay_decay_at = 0;
+}
+
+uint32_t Mesh::getOtaRetransmitDelay(const mesh::Packet* packet) {
+  if (packet == NULL) return 0;
+  decayOtaRelayBackoff();
+  uint32_t airtime = _radio->getEstAirtimeFor(packet->getRawLength());
+  if (airtime == 0) return 0;
+  // Quarter-airtime units by pressure level: 0.25-0.5, 0.5-1.0, 0.75-2.0, 1.0-3.0.
+  // The maximum uses floor division, so level 3 can never exceed exactly three measured airtimes.
+  static const uint8_t min_quarters[4] = {1, 2, 3, 4};
+  static const uint8_t max_quarters[4] = {2, 4, 8, 12};
+  const uint8_t level = _ota_relay_backoff_level > 3 ? 3 : _ota_relay_backoff_level;
+  uint32_t min_delay = (airtime * min_quarters[level] + 3) / 4;
+  uint32_t max_delay = (airtime * max_quarters[level]) / 4;
+  if (max_delay < min_delay) max_delay = min_delay;
+  return _rng->nextInt(min_delay, max_delay + 1);
+}
 uint32_t Mesh::getDirectRetransmitDelay(const Packet* packet) {
   return 0;  // by default, no delay
 }
@@ -902,6 +993,7 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
       // OTA is invisible outside an actually-running temporary-radio window. In particular, do not add it
       // to the seen table: a copy heard on the normal channel must not suppress one received after temp radio starts.
       if (!isTempRadioActive()) break;
+      observeOtaRequestPressure(pkt);
       uint8_t n = pkt->getPathHashCount();   // hops travelled to reach us (flood path-hash count)
 #if defined(ENABLE_OTA)
       // Accept-gate (duty-cycle horizon): ignore OTA from further than our hop limit - neither process nor
@@ -934,7 +1026,7 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) {
           && allowPacketForward(pkt)) {
         self_id.copyHashTo(&pkt->path[n * pkt->getPathHashSize()], pkt->getPathHashSize());
         pkt->setPathHashCount(n + 1);
-        action = ACTION_RETRANSMIT_DELAYED(OTA_TX_PRIORITY, getRetransmitDelay(pkt));
+        action = ACTION_RETRANSMIT_DELAYED(OTA_TX_PRIORITY, getOtaRetransmitDelay(pkt));
       }
       break;
     }

src/Mesh.h

@@ -110,12 +110,23 @@ class Mesh : public Dispatcher {
     bool valid;
   };
 
+  static const uint8_t OTA_REQUEST_TRACK_SLOTS = 4;
+  struct OtaRequestTrackEntry {
+    uint8_t key[7];                    // message type + manifest_id + block_idx
+    uint32_t last_request_ms;
+    uint8_t min_path_hops;             // ignore a later downstream echo with a longer path
+    bool valid;
+  };
+
   RTCClock* _rtc;
   RNG* _rng;
   MeshTables* _tables;
   DirectRetryEntry _direct_retries[MAX_DIRECT_RETRY_SLOTS];
   FloodRetryEntry _flood_retries[MAX_FLOOD_RETRY_SLOTS];
   RecentAdvertEchoEntry _recent_advert_echoes[MAX_RECENT_ADVERT_ECHOS];
+  OtaRequestTrackEntry _ota_request_track[OTA_REQUEST_TRACK_SLOTS] = {};
+  uint32_t _ota_relay_decay_at = 0;
+  uint8_t _ota_relay_backoff_level = 0;
   uint8_t _active_direct_retry_count;
   uint8_t _active_flood_retry_count;
   uint8_t _waiting_direct_retry_count;
@@ -154,6 +165,9 @@ class Mesh : public Dispatcher {
   void armFloodRetryOnSendComplete(const Packet* packet);
   void clearPendingFloodRetryOnSendFail(const Packet* packet);
   void maybeScheduleFloodRetry(const Packet* packet, uint8_t priority);
+  void resetOtaRelayBackoff();
+  void observeOtaRequestPressure(const Packet* packet);
+  void decayOtaRelayBackoff();
   void serviceLoopMaintenance();
   //void routeRecvAcks(Packet* packet, uint32_t delay_millis);
   DispatcherAction forwardMultipartDirect(Packet* pkt);
@@ -195,6 +209,11 @@ protected:
    */
   virtual uint32_t getRetransmitDelay(const Packet* packet);
 
+  /**
+   * \returns  Adaptive TempRadio-only OTA relay delay, from 0.25-0.5 airtimes up to a 3-airtime cap.
+   */
+  virtual uint32_t getOtaRetransmitDelay(const Packet* packet);
+
   /**
    * \returns  number of milliseconds delay to apply to retransmitting the given packet, for DIRECT mode.
    */

src/helpers/ota/OtaManager.cpp

@@ -371,8 +371,8 @@ void OtaManager::handleGetLeaves(const uint8_t* m, uint16_t n) {
   }
 }
 
-// Smallest mask covering `nf` fragments: bit k set for k in [0, nf). Caps at 16 (matches _reasm_mask and
-// the manifest reassembly), which bounds a block at 16 fragments - our 1 KB blocks are 7.
+// Smallest mask covering `nf` fragments: bit k set for k in [0, nf). Caps at 16 (matches each pipeline
+// slot mask and the manifest reassembly), which bounds a block at 16 fragments - our 1 KB blocks are 7.
 static inline uint16_t frag_full_mask(uint32_t nf) {
   return (nf >= 16) ? 0xFFFFu : (uint16_t)((1u << nf) - 1);
 }
@@ -607,10 +607,48 @@ bool OtaManager::wantRow(const uint8_t* mid, uint32_t target, uint8_t codec, uin
   return true;
 }
 
-// Forget the block currently being reassembled / awaited (back to NO_BLOCK). Safe to call between blocks:
-// the next DATA fragment re-derives the slice mask for whatever block it belongs to.
+void OtaManager::clearReassemblySlot(uint8_t slot) {
+  if (slot >= OTA_FETCH_PIPELINE) return;
+  _reasm[slot].block = NO_BLOCK;
+  _reasm[slot].mask = 0;
+  _reasm[slot].need = 0;
+  _reasm[slot].awaiting_proof = false;
+}
+
+// Forget all blocks currently being reassembled or awaiting proofs.
 void OtaManager::clearReassembly() {
-  _reasm_block = NO_BLOCK; _reasm_mask = 0; _reasm_need = 0; _awaiting_proof = false;
+  for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) clearReassemblySlot(slot);
+  _retry_slot = 0;
+  _req_count = 0;
+}
+
+int OtaManager::findReassemblySlot(uint32_t block) const {
+  for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) {
+    if (_reasm[slot].block == block) return slot;
+  }
+  return -1;
+}
+
+int OtaManager::findEmptyReassemblySlot() const {
+  for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) {
+    if (_reasm[slot].block == NO_BLOCK) return slot;
+  }
+  return -1;
+}
+
+bool OtaManager::blockInPipeline(uint32_t block) const {
+  return findReassemblySlot(block) >= 0;
+}
+
+uint32_t OtaManager::pipelineProgress() const {
+  // A small stable fingerprint is sufficient here: equality only suppresses a retry for one service tick.
+  uint32_t progress = 2166136261u;
+  for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) {
+    progress = (progress ^ _reasm[slot].block) * 16777619u;
+    progress = (progress ^ _reasm[slot].mask) * 16777619u;
+    progress = (progress ^ (_reasm[slot].awaiting_proof ? 1u : 0u)) * 16777619u;
+  }
+  return progress;
 }
 
 // Begin (or resume) fetching a chosen mid: try a staged-partial resume first, else request the manifest.
@@ -702,7 +740,7 @@ void OtaManager::handleManifest(const uint8_t* m, uint16_t n) {
   _fpoff = payload_off; _floff = leaves_off; _fpsize = payload_size; _fbc = bc; _fbs = bs;
   _ftotal = total; _have = 0;
   clearReassembly();                             // fresh transfer: drop any prior per-block state
-  _loop_last_have = 0; _loop_last_mask = 0;
+  _loop_last_have = 0; _loop_last_progress = pipelineProgress();
   // Warm-start (folder-capture): if requested and the image is small enough for the fixed leaves bitmap,
   // bulk-fetch the target leaves and diff the seed already staged in the store; else just transfer normally.
   if (_validate && beginLeafDiff()) {
@@ -823,7 +861,7 @@ bool OtaManager::resumeStaged(const uint8_t* want_mid) {
   _fpoff = payload_off; _floff = leaves_off; _fpsize = m.payload_size; _fbc = bc; _fbs = bs;
   _ftotal = total;
   clearReassembly();
-  _loop_last_have = 0; _loop_last_mask = 0;
+  _loop_last_have = 0; _loop_last_progress = 0;
   beginStagedVerification();
   OTA_DBG("OTA: RESUME verifying %u blocks total=%u\n",
           (unsigned)bc, (unsigned)total);
@@ -884,12 +922,12 @@ void OtaManager::verifyStagedStep() {
     }
 
     const uint32_t length = blockLen(index);
-    if (!_fetch->read(_fpoff + index * _fbs, _reasm_buf, length)) {
+    if (!_fetch->read(_fpoff + index * _fbs, _reasm[0].buf, length)) {
       _fstate = FAILED;
       return;
     }
     uint8_t computed_leaf[4];
-    merkle_leaf(computed_leaf, _reasm_buf, length);
+    merkle_leaf(computed_leaf, _reasm[0].buf, length);
     if (memcmp(computed_leaf, stored_leaf, sizeof(stored_leaf)) != 0) {
       // Payload and marker disagree. Clear the marker so the normal fetch path
       // requests this block again; a stale marker must never bless bad bytes.
@@ -921,7 +959,7 @@ void OtaManager::verifyStagedStep() {
   } else {
     _fstate = FETCHING;
     _loop_last_have = _have;
-    _loop_last_mask = 0;
+    _loop_last_progress = pipelineProgress();
     requestMissing();
   }
   _resume_merkle.reset();
@@ -933,97 +971,129 @@ void OtaManager::handleData(const uint8_t* m, uint16_t n) {
   if (_fstate != FETCHING || memcmp(dm.manifest_id, _fid, 4) != 0) return;
   if (dm.block_idx >= _fbc) return;
   if (blockPresent(dm.block_idx)) return;                   // already stored + verified
+  int slot_index = findReassemblySlot(dm.block_idx);
+  if (slot_index < 0) return;                               // accept only blocks inside our bounded window
+  ReassemblySlot& slot = _reasm[slot_index];
   uint32_t blen = blockLen(dm.block_idx);
   if (dm.frag_off % OTA_FRAG_DATA != 0) return;             // canonical FRAG_DATA-aligned slices only
   if ((uint32_t)dm.frag_off + dm.data_len > blen) return;   // slice out of the block
-  if (dm.block_idx != _reasm_block) {                       // (re)start reassembly for this block
-    _reasm_block = dm.block_idx; _reasm_mask = 0; _awaiting_proof = false;
-    uint32_t nf = (blen + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA;
-    _reasm_need = (nf >= 16) ? 0xFFFF : (uint16_t)((1u << nf) - 1);
-  }
+  uint32_t expected_len = blen - dm.frag_off;
+  if (expected_len > OTA_FRAG_DATA) expected_len = OTA_FRAG_DATA;
+  if (dm.data_len != expected_len) return;                  // do not mark a short slice as complete
   uint32_t kf = dm.frag_off / OTA_FRAG_DATA;
   if (kf >= 16) return;
-  memcpy(_reasm_buf + dm.frag_off, dm.data, dm.data_len);
-  _reasm_mask |= (uint16_t)(1u << kf);
-  if (_reasm_mask != _reasm_need || _awaiting_proof) return;  // wait for all slices (or proof already asked)
+  memcpy(slot.buf + dm.frag_off, dm.data, dm.data_len);
+  slot.mask |= (uint16_t)(1u << kf);
+  if (slot.mask != slot.need || slot.awaiting_proof) return;  // wait for all slices (or proof already asked)
   // block fully reassembled -> request its proof (data + proof are fetched separately)
-  _awaiting_proof = true;
-  ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)_reasm_block;
-  uint8_t b[16]; emit(b, encode_req_proof(b, sizeof(b), rp), false);
+  slot.awaiting_proof = true;
+  requestSlot((uint8_t)slot_index);
 }
 
 void OtaManager::handleProof(const uint8_t* m, uint16_t n) {
   ProofMsg pm;
   if (!decode_proof(m, n, pm) || !_fetch) return;
   if (_fstate != FETCHING || memcmp(pm.manifest_id, _fid, 4) != 0) return;
-  if (!_awaiting_proof || pm.block_idx != _reasm_block) return;   // not the block we're verifying
-  uint32_t blen = blockLen(_reasm_block);
-  if (!merkle_verify(_reasm_buf, blen, _reasm_block, pm.proof, pm.n_proof, _froot, _fbc)) {
-    clearReassembly();                                                      // bad -> drop, re-fetch the block
+  int slot_index = findReassemblySlot(pm.block_idx);
+  if (slot_index < 0 || !_reasm[slot_index].awaiting_proof) return;
+  ReassemblySlot& slot = _reasm[slot_index];
+  uint32_t block = slot.block;
+  uint32_t blen = blockLen(block);
+  if (!merkle_verify(slot.buf, blen, block, pm.proof, pm.n_proof, _froot, _fbc)) {
+    clearReassemblySlot((uint8_t)slot_index);                   // bad -> drop and re-fetch only this block
+    requestMissing();
     return;
   }
   // verified -> commit the payload block, then its leaf (the present marker). A write failure here means a
   // FOLDER destination's seeder link dropped mid-transfer: PAUSE (hold progress on the host, stop
   // requesting, do NOT fall back to RAM/flash). The block is left uncommitted (its leaf stays 0xFF), so on
   // reconnect resumeStaged() re-requests exactly it. Flash failures also pause instead of being ignored.
-  uint8_t leaf[4]; merkle_leaf(leaf, _reasm_buf, blen);
-  if (!_fetch->write(_fpoff + (uint32_t)_reasm_block * _fbs, _reasm_buf, blen) ||
-      !_fetch->write(_floff + (uint32_t)_reasm_block * 4, leaf, 4)) {
+  uint8_t leaf[4]; merkle_leaf(leaf, slot.buf, blen);
+  if (!_fetch->write(_fpoff + block * _fbs, slot.buf, blen) ||
+      !_fetch->write(_floff + block * 4, leaf, 4)) {
     _fstate = PAUSED; clearReassembly(); return;
   }
   _have++;
-  OTA_DBG("OTA: block %u OK  have=%u/%u\n", (unsigned)_reasm_block, (unsigned)_have, (unsigned)_fbc);
-  clearReassembly();
+  OTA_DBG("OTA: block %u OK  have=%u/%u\n", (unsigned)block, (unsigned)_have, (unsigned)_fbc);
+  clearReassemblySlot((uint8_t)slot_index);
   // periodically persist progress (meta/leaf page + open payload) so a reboot can resume (no-op for RAM);
   // cadence is runtime-tunable via `ota config checkpoint <N>` (0 = never)
   if (_checkpoint_blocks && _have % _checkpoint_blocks == 0) _fetch->checkpoint();
-  if (_have < _fbc) { requestMissing(); return; }            // next block
+  if (_have < _fbc) { requestMissing(); return; }            // keep the client window full
   // Every leaf must be readable and collectively match the manifest root.
   // A scratch allocation/read failure is an integrity failure, never success.
+  clearReassembly();
   _fstate = storedLeavesRootMatches() ? COMPLETE : FAILED;
   if (_fstate == COMPLETE && !_fetch->finalize()) _fstate = FAILED;
   OTA_DBG("OTA: transfer %s\n", _fstate == COMPLETE ? "COMPLETE" : "FAILED(integrity/storage)");
 }
 
-void OtaManager::requestMissing() {
-  if (_fstate != FETCHING) return;
-  // Per-block serial flow (split data/proof). If the current block's data is fully reassembled and we
-  // are waiting on its proof, (re)send the proof request rather than re-fetching the data - this also
-  // recovers from a lost PROOF reply.
-  if (_awaiting_proof && _reasm_block != NO_BLOCK) {
-    ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)_reasm_block;
+void OtaManager::requestSlot(uint8_t slot_index) {
+  if (_fstate != FETCHING || slot_index >= OTA_FETCH_PIPELINE) return;
+  ReassemblySlot& slot = _reasm[slot_index];
+  if (slot.block >= _fbc) return;
+  _req_start = slot.block;
+  _req_count = 0;
+  for (uint8_t i = 0; i < OTA_FETCH_PIPELINE; i++) {
+    if (_reasm[i].block != NO_BLOCK) _req_count++;
+  }
+  if (slot.awaiting_proof) {
+    ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)slot.block;
     uint8_t b[16]; emit(b, encode_req_proof(b, sizeof(b), rp), false);
-    OTA_DBG("OTA: REQ_PROOF block=%u (have=%u/%u)\n",
-            (unsigned)_reasm_block, (unsigned)_have, (unsigned)_fbc);
+    OTA_DBG("OTA: REQ_PROOF block=%u (have=%u/%u window=%u)\n",
+            (unsigned)slot.block, (unsigned)_have, (unsigned)_fbc, (unsigned)_req_count);
     return;
   }
-  // Otherwise request the DATA fragments of the next missing block. One block at a time keeps the
-  // server's TX queue tiny so OTA never floods the mesh (docs/ota_protocol.md Section 8); a block's fragments
-  // are self-describing (frag_off) so missing slices can be retried without re-sending a full block.
-  uint32_t start = pickMissingBlock();
-  if (start >= _fbc) return;
-  _req_start = start; _req_count = 1;
-  // Ask only for fragments we still lack: the holes of the in-flight partial block (_reasm_need & ~mask),
-  // or all fragments of a fresh block. A lost fragment then costs one fragment to re-fetch, not the whole
-  // block - and the re-REQ can't collide with a multi-fragment burst (half-duplex) as before.
-  uint32_t nf = (blockLen(start) + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA;
-  uint16_t need = frag_full_mask(nf);
-  uint16_t have = (start == _reasm_block) ? _reasm_mask : 0;
-  uint16_t want = (uint16_t)(need & ~have);
-  if (want == 0) want = need;                       // safety: never send an empty request
+  uint16_t want = (uint16_t)(slot.need & ~slot.mask);
+  if (want == 0) want = slot.need;                  // safety: never send an empty request
   ReqMsg rq; memcpy(rq.manifest_id, _fid, 4);
-  rq.block_idx = (uint16_t)start; rq.want_mask = want;
+  rq.block_idx = (uint16_t)slot.block; rq.want_mask = want;
   uint8_t b[16];
-  OTA_DBG("OTA: REQ block=%u want=%04x (have=%u/%u mask=%04x)\n",
-          (unsigned)start, (unsigned)want, (unsigned)_have, (unsigned)_fbc, (unsigned)_reasm_mask);
+  OTA_DBG("OTA: REQ block=%u want=%04x (have=%u/%u mask=%04x window=%u)\n",
+          (unsigned)slot.block, (unsigned)want, (unsigned)_have, (unsigned)_fbc,
+          (unsigned)slot.mask, (unsigned)_req_count);
   emit(b, encode_req(b, sizeof(b), rq), false);
 }
 
-// One receiver walks missing blocks in serial order. A partially-reassembled block is always finished first.
+bool OtaManager::fillPipeline() {
+  bool requested = false;
+  while (true) {
+    int slot_index = findEmptyReassemblySlot();
+    if (slot_index < 0) break;
+    uint32_t block = pickMissingBlock();
+    if (block >= _fbc) break;
+    ReassemblySlot& slot = _reasm[slot_index];
+    slot.block = block;
+    slot.mask = 0;
+    slot.need = frag_full_mask((blockLen(block) + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA);
+    slot.awaiting_proof = false;
+    requestSlot((uint8_t)slot_index);
+    requested = true;
+  }
+  return requested;
+}
+
+void OtaManager::requestMissing() {
+  if (_fstate != FETCHING) return;
+  if (fillPipeline()) return;
+
+  // A full window with no progress is stalled. Retry one slot per service tick, round-robin, so a weak
+  // block cannot trigger a simultaneous multi-block re-burst on a half-duplex channel.
+  for (uint8_t offset = 0; offset < OTA_FETCH_PIPELINE; offset++) {
+    uint8_t slot = (uint8_t)((_retry_slot + offset) % OTA_FETCH_PIPELINE);
+    if (_reasm[slot].block == NO_BLOCK) continue;
+    requestSlot(slot);
+    _retry_slot = (uint8_t)((slot + 1) % OTA_FETCH_PIPELINE);
+    return;
+  }
+}
+
+// Walk missing blocks in serial order, excluding blocks already assigned to a pipeline slot.
 uint32_t OtaManager::pickMissingBlock() {
   if (_fbc == 0) return _fbc;
-  if (_reasm_block < _fbc && !blockPresent(_reasm_block) && _reasm_mask != 0) return _reasm_block;
-  for (uint32_t i = 0; i < _fbc; i++) if (!blockPresent(i)) return i;
+  for (uint32_t i = 0; i < _fbc; i++) {
+    if (!blockInPipeline(i) && !blockPresent(i)) return i;
+  }
   return _fbc;
 }
 
@@ -1092,11 +1162,12 @@ void OtaManager::loop() {
     return;
   }
   if (_fstate != FETCHING) return;
-  // retry only when a whole tick passed with NO progress - neither a committed block nor a new fragment
-  // of the in-flight block. This avoids re-request spam while a block's fragments are still streaming in.
-  if (_have == _loop_last_have && _reasm_mask == _loop_last_mask) requestMissing();
+  // Retry only when a whole tick passed with no committed block, fragment, proof transition, or window
+  // assignment. This avoids re-request spam while any pipelined block is making progress.
+  uint32_t progress = pipelineProgress();
+  if (_have == _loop_last_have && progress == _loop_last_progress) requestMissing();
   _loop_last_have = _have;
-  _loop_last_mask = _reasm_mask;
+  _loop_last_progress = pipelineProgress();
 }
 
 // ---------------- dispatch ----------------

src/helpers/ota/OtaManager.h

@@ -128,6 +128,12 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len)
 #ifndef OTA_FRAG_DATA
 #define OTA_FRAG_DATA 160           // data bytes per DATA fragment (<= MAX_PACKET_PAYLOAD - 9-byte header)
 #endif
+#ifndef OTA_FETCH_PIPELINE
+#define OTA_FETCH_PIPELINE 2        // concurrent client block slots; keeps one block ready behind proof/flash work
+#endif
+#if OTA_FETCH_PIPELINE < 1
+#error "OTA_FETCH_PIPELINE must be at least 1"
+#endif
 // nRF52 note: a flash page-erase halts the CPU (~85 ms, code runs from flash) and starves the LoRa RX,
 // so writing to flash on every received packet drops in-flight DATA and the transfer stalls. The SD-safe
 // driver (Adafruit flash_nrf5x) always erases on flush, so there is no erase-free write; instead
@@ -315,7 +321,7 @@ public:
   void reset_session() {
     _fstate = IDLE; _have = 0; _req_count = 0; _mf_retries = 0;
     clearReassembly();
-    _loop_last_have = 0; _loop_last_mask = 0;
+    _loop_last_have = 0; _loop_last_progress = pipelineProgress();
     _mf_total = 0; _mf_mask = 0; _mf_len = 0; _loop_last_mfmask = 0;
     freeLeaves(); _validate = false; _archive_fetch = false;
     _lv_retries = 0; _loop_last_lvmask = 0;
@@ -371,8 +377,13 @@ private:
   void handleProof(const uint8_t* m, uint16_t n);
   void startFetch(const uint8_t* mid, uint32_t target, bool validate = false);   // begin/resume a fetch
   bool wantRow(const uint8_t* mid, uint32_t target, uint8_t codec, uint8_t flags) const;  // fetch this row?
-  void clearReassembly();                                 // forget the in-flight block (reset to NO_BLOCK)
-  uint32_t pickMissingBlock();                            // choose the next missing block in serial order
+  void clearReassembly();                                 // forget every in-flight pipeline slot
+  void clearReassemblySlot(uint8_t slot);
+  int findReassemblySlot(uint32_t block) const;
+  int findEmptyReassemblySlot() const;
+  bool blockInPipeline(uint32_t block) const;
+  uint32_t pipelineProgress() const;
+  uint32_t pickMissingBlock();                            // choose the next missing block not already in flight
   int  serveEntryIndex(const uint8_t* mid) const;         // registry slot serving this mid (-1 if none)
   ServeView* resolve(const uint8_t* mid);                 // pick/load the ServeView for this mid (nullptr)
   bool loadSource(const ServeEntry& e);                   // load an external mota into _srcv (head+leaves)
@@ -399,7 +410,9 @@ private:
   bool storedLeavesRootMatches() const;
   void beginStagedVerification();
   void verifyStagedStep();
-  void requestMissing();
+  void requestSlot(uint8_t slot);                         // request DATA holes or the proof for one slot
+  bool fillPipeline();                                    // assign and request blocks until the window is full
+  void requestMissing();                                  // fill the window, or retry one stalled slot
   uint32_t blockLen(uint32_t i) const;
 
   uint32_t _target = 0;
@@ -436,7 +449,7 @@ private:
   uint32_t   _resume_verify_idx = 0;
   bool       _resume_invalidated = false;
   MerkleAccumulator _resume_merkle;
-  uint32_t   _req_start = 0, _req_count = 0;   // last block requested (per-block serial flow; telemetry)
+  uint32_t   _req_start = 0, _req_count = 0;   // most recently requested block + active window size
   uint32_t   _loop_last_have = 0;              // for stall detection in loop()
   uint32_t   _desired_target = 0;              // manual cross-target override (0 = auto / own target)
   uint8_t    _desired_mid[4] = {0,0,0,0};      // pull a specific manifest_id (see want_mid)
@@ -452,13 +465,19 @@ private:
   uint16_t   _advert_mins = OTA_ADVERT_INTERVAL_MINS;    // beacon re-advertise cadence, minutes; 0=off (persisted)
   uint8_t    _max_hops = OTA_HOP_LIMIT_DEFAULT;          // OTA flood reach in hops; 0=direct only (persisted)
   uint8_t    _fflags = 0;                       // flags of the manifest currently being fetched
-  // multi-fragment reassembly of the current block (per-block 2-phase: fetch data, then its proof)
-  uint32_t   _reasm_block = NO_BLOCK;          // block being reassembled / awaiting proof (none)
-  uint16_t   _reasm_mask = 0;                  // received FRAG_DATA-slice bitmap (bit k = slice @ k*FRAG_DATA)
-  uint16_t   _reasm_need = 0;                  // full mask once all slices of the current block are in
-  bool       _awaiting_proof = false;          // data complete; REQ_PROOF sent, verify on PROOF
-  uint16_t   _loop_last_mask = 0;              // fragment-level stall detection in loop()
-  uint8_t    _reasm_buf[OTA_MAX_BLOCK];
+  // Bounded multi-block client pipeline. Each slot independently reassembles DATA and awaits its Merkle
+  // proof, so proof/flash work for one block can overlap radio delivery of the next. The wire protocol and
+  // opaque repeater behavior are unchanged.
+  struct ReassemblySlot {
+    uint32_t block = NO_BLOCK;
+    uint16_t mask = 0;                         // received FRAG_DATA-slice bitmap
+    uint16_t need = 0;                         // full bitmap for this block
+    bool awaiting_proof = false;
+    uint8_t buf[OTA_MAX_BLOCK];
+  };
+  ReassemblySlot _reasm[OTA_FETCH_PIPELINE];
+  uint32_t   _loop_last_progress = 0;           // aggregate pipeline progress for stall detection
+  uint8_t    _retry_slot = 0;                   // round-robin stalled-slot retry cursor
   // multi-fragment manifest reassembly (a signed v2 manifest exceeds one packet)
   uint8_t    _mf_buf[OTA_MF_MAXFRAG * OTA_MF_FRAG];   // sized to the fragment cap so no valid manifest is silently dropped
   uint16_t   _mf_retries = 0;                          // GET_MANIFEST retries while WANT_MANIFEST (give up after a cap)

test/test_ota/test_ota_core.cpp

@@ -677,6 +677,130 @@ static void deliver_manifest_fragment(OtaManager& client, const uint8_t mid[4],
   client.on_message(wire, wire_len);
 }
 
+TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) {
+  MotaManifest manifest;
+  ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
+  ASSERT_GE(manifest.block_count, 3u);
+
+  OtaManager client;
+  OtaStoreRam<4096> store;
+  CapturedMessages sent;
+  client.begin(SIM_TARGET_ID, capture_send, &sent);
+  client.set_fetch_store(&store);
+  client.pull(manifest.merkle_root, manifest.target_id);
+  sent.items.clear();                                      // discard the initial GET_MANIFEST
+
+  deliver_manifest_fragment(client, manifest.merkle_root, 0,
+                            manifest.manifest_start, OTA_MF_FRAG);
+  deliver_manifest_fragment(client, manifest.merkle_root, 1,
+                            manifest.manifest_start + OTA_MF_FRAG,
+                            (uint16_t)(MOTA_MFL - OTA_MF_FRAG));
+  ASSERT_EQ(client.fetchState(), OtaManager::FETCHING);
+  client.set_clock(5000);
+  client.loop();                                           // opens the bounded request window
+
+  ASSERT_EQ(sent.items.size(), 2u);
+  ReqMsg request0, request1;
+  ASSERT_TRUE(decode_req(sent.items[0].data(), (uint16_t)sent.items[0].size(), request0));
+  ASSERT_TRUE(decode_req(sent.items[1].data(), (uint16_t)sent.items[1].size(), request1));
+  EXPECT_EQ(request0.block_idx, 0);
+  EXPECT_EQ(request1.block_idx, 1);
+
+  // Deliver block 1 first and its fragments in reverse order. Its independent slot must request and
+  // accept a proof without disturbing the still-empty slot reserved for block 0.
+  const uint32_t block = 1;
+  const uint32_t block_size = manifest.block_size();
+  const uint32_t block_len = block + 1 < manifest.block_count
+      ? block_size : manifest.payload_size - block * block_size;
+  const uint32_t fragments = (block_len + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA;
+  uint8_t wire[MAX_PACKET_PAYLOAD];
+  for (uint32_t k = fragments; k-- > 0;) {
+    const uint32_t offset = k * OTA_FRAG_DATA;
+    uint32_t length = block_len - offset;
+    if (length > OTA_FRAG_DATA) length = OTA_FRAG_DATA;
+    DataMsg data;
+    memcpy(data.manifest_id, manifest.merkle_root, 4);
+    data.block_idx = block;
+    data.frag_off = (uint16_t)offset;
+    data.data = manifest.payload + block * block_size + offset;
+    data.data_len = (uint16_t)length;
+    uint16_t wire_len = encode_data(wire, sizeof(wire), data);
+    ASSERT_GT(wire_len, 0);
+    client.on_message(wire, wire_len);
+  }
+
+  ASSERT_EQ(sent.items.size(), 3u);
+  ReqProofMsg proof_request;
+  ASSERT_TRUE(decode_req_proof(sent.items[2].data(), (uint16_t)sent.items[2].size(), proof_request));
+  EXPECT_EQ(proof_request.block_idx, block);
+
+  uint8_t scratch[SIM_MOTA_1K_BLOCKS * 4];
+  uint8_t siblings[32 * 4];
+  uint8_t sibling_count = merkle_gen_proof(
+      manifest.leaves, manifest.block_count, block, scratch, siblings);
+  ProofMsg proof;
+  memcpy(proof.manifest_id, manifest.merkle_root, 4);
+  proof.block_idx = block;
+  proof.n_proof = sibling_count;
+  proof.proof = siblings;
+  uint16_t wire_len = encode_proof(wire, sizeof(wire), proof);
+  ASSERT_GT(wire_len, 0);
+  client.on_message(wire, wire_len);
+
+  EXPECT_EQ(client.blocksHave(), 1u);
+  ASSERT_EQ(sent.items.size(), 4u);                       // freed slot was immediately refilled
+  ReqMsg refill;
+  ASSERT_TRUE(decode_req(sent.items[3].data(), (uint16_t)sent.items[3].size(), refill));
+  EXPECT_EQ(refill.block_idx, 2);
+}
+
+TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) {
+  MotaManifest manifest;
+  ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
+
+  OtaManager client;
+  OtaStoreRam<4096> store;
+  CapturedMessages sent;
+  client.begin(SIM_TARGET_ID, capture_send, &sent);
+  client.set_fetch_store(&store);
+  client.pull(manifest.merkle_root, manifest.target_id);
+  sent.items.clear();
+  deliver_manifest_fragment(client, manifest.merkle_root, 0,
+                            manifest.manifest_start, OTA_MF_FRAG);
+  deliver_manifest_fragment(client, manifest.merkle_root, 1,
+                            manifest.manifest_start + OTA_MF_FRAG,
+                            (uint16_t)(MOTA_MFL - OTA_MF_FRAG));
+  client.loop();
+  ASSERT_EQ(sent.items.size(), 2u);                       // initial requests for blocks 0 and 1
+
+  DataMsg first_fragment;
+  memcpy(first_fragment.manifest_id, manifest.merkle_root, 4);
+  first_fragment.block_idx = 0;
+  first_fragment.frag_off = 0;
+  first_fragment.data = manifest.payload;
+  first_fragment.data_len = OTA_FRAG_DATA;
+  uint8_t wire[MAX_PACKET_PAYLOAD];
+  uint16_t wire_len = encode_data(wire, sizeof(wire), first_fragment);
+  ASSERT_GT(wire_len, 0);
+  client.on_message(wire, wire_len);
+
+  client.loop();                                         // observes progress; no premature retry
+  EXPECT_EQ(sent.items.size(), 2u);
+  client.loop();                                         // stalled: retry slot 0 only
+  ASSERT_EQ(sent.items.size(), 3u);
+  ReqMsg retry0;
+  ASSERT_TRUE(decode_req(sent.items[2].data(), (uint16_t)sent.items[2].size(), retry0));
+  EXPECT_EQ(retry0.block_idx, 0);
+  EXPECT_EQ(retry0.want_mask, (uint16_t)(0x007F & ~0x0001));
+
+  client.loop();                                         // next stalled tick rotates to slot 1
+  ASSERT_EQ(sent.items.size(), 4u);
+  ReqMsg retry1;
+  ASSERT_TRUE(decode_req(sent.items[3].data(), (uint16_t)sent.items[3].size(), retry1));
+  EXPECT_EQ(retry1.block_idx, 1);
+  EXPECT_EQ(retry1.want_mask, 0x007F);
+}
+
 TEST(OtaTransfer, RejectsShortNonFinalManifestFragment) {
   g_q.clear();
   MotaManifest manifest;

test/test_trace_retry/test_trace_retry.cpp

@@ -20,7 +20,12 @@ public:
 
 class TraceTestRNG : public mesh::RNG {
 public:
-  void random(uint8_t* dest, size_t sz) override { memset(dest, 0, sz); }
+  uint32_t value = 0;
+  void random(uint8_t* dest, size_t sz) override {
+    for (size_t offset = 0; offset < sz; offset++) {
+      dest[offset] = (uint8_t)(value >> (8 * (offset % sizeof(value))));
+    }
+  }
 };
 
 class TraceTestRadio : public mesh::Radio {
@@ -94,6 +99,10 @@ public:
     return getFloodRetryAttemptDelay(packet, attempt_idx);
   }
 
+  uint32_t otaRelayDelay(const mesh::Packet* packet) {
+    return getOtaRetransmitDelay(packet);
+  }
+
   void completePacketSend(mesh::Packet* packet) {
     onSendComplete(packet);
   }
@@ -214,6 +223,55 @@ TEST(RepeaterTransport, OtaFloodRelaysWithoutOtaManagerOnlyDuringTempRadio) {
   EXPECT_FALSE(node.canTransmit(&packet));  // queued-near-expiry packets cannot leak onto the normal channel
 }
 
+TEST(RepeaterTransport, OtaRelayDelayIsQuarterToHalfAirtime) {
+  TraceTestClock clock;
+  TraceTestRTC rtc;
+  TraceTestRNG rng;
+  TraceTestRadio radio;                     // fixed 10 ms packet airtime
+  ForwardingTestTables tables;
+  StaticPoolPacketManager manager(12);
+  TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
+  mesh::Packet packet = makeFloodPacket(PAYLOAD_TYPE_OTA);
+
+  rng.value = 0;
+  EXPECT_EQ(node.otaRelayDelay(&packet), 3u);  // ceil(0.25 * 10 ms)
+  rng.value = 2;                              // selects the last value in [3, 5]
+  EXPECT_EQ(node.otaRelayDelay(&packet), 5u);  // 0.5 * 10 ms
+}
+
+TEST(RepeaterTransport, OtaRelayDelayEscalatesAndCapsAtThreeAirtimes) {
+  TraceTestClock clock;
+  TraceTestRTC rtc;
+  TraceTestRNG rng;
+  TraceTestRadio radio;                         // fixed 10 ms packet airtime
+  ForwardingTestTables tables;
+  StaticPoolPacketManager manager(12);
+  TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
+  node.tempRadioActive = true;
+  node.forwardFloods = true;
+  auto make_request = []() {
+    mesh::Packet request = makeFloodPacket(PAYLOAD_TYPE_OTA);
+    request.payload_len = 9;
+    request.payload[0] = 0x06;                  // OTA_REQ
+    for (uint8_t i = 1; i < request.payload_len; i++) request.payload[i] = i;
+    return request;
+  };
+
+  mesh::Packet request = make_request();
+  node.receivePacket(&request);                  // first request establishes the key
+  for (uint32_t retry = 1; retry <= 3; retry++) {
+    clock.now = retry * 3000;
+    request = make_request();                    // a fresh origin copy has the same zero-hop request key
+    node.receivePacket(&request);                // frequent repeats raise one level each
+  }
+  rng.value = 20;                                // last value in inclusive [10, 30]
+  EXPECT_EQ(node.otaRelayDelay(&request), 30u);  // hard maximum = 3 * 10 ms
+
+  clock.now = 100000;                            // three clean 30-second decay periods
+  rng.value = 2;
+  EXPECT_EQ(node.otaRelayDelay(&request), 5u);   // returned to the 0.25-0.5 baseline
+}
+
 TEST(RepeaterTransport, OpaqueKnownFloodPayloadsAreRelayed) {
   TraceTestClock clock;
   TraceTestRTC rtc;