diff --git a/docs/ota_protocol.md b/docs/ota_protocol.md index ac4f2a69..880d0ffc 100644 --- a/docs/ota_protocol.md +++ b/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 diff --git a/docs/ota_user_guide.md b/docs/ota_user_guide.md index a32ee906..7d7d3640 100644 --- a/docs/ota_user_guide.md +++ b/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 diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 103801a6..5a5c132e 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -1,4 +1,5 @@ #include "Mesh.h" +#include "helpers/ota/OtaFormat.h" // request types used by TempRadio relay-pressure tracking //#include #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; } diff --git a/src/Mesh.h b/src/Mesh.h index f220c08a..90e5a28f 100644 --- a/src/Mesh.h +++ b/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. */ diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index f57ec4ce..c44f7f72 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/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 ` (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 ---------------- diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index 69a7698e..03f20976 100644 --- a/src/helpers/ota/OtaManager.h +++ b/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) diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index aa5f8943..3cf44576 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/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; diff --git a/test/test_trace_retry/test_trace_retry.cpp b/test/test_trace_retry/test_trace_retry.cpp index 6ab9a046..89901770 100644 --- a/test/test_trace_retry/test_trace_retry.cpp +++ b/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;