From dde87f1d155d36b7b913e669affa4fbcf3a32709 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 15 Sep 2026 07:54:16 -0700 Subject: [PATCH] Preserve OTA block progress and channel checks during radio recovery Drain received packets in single-profile radio fault backoff as well as dual-profile mode. Share the normal channel-clear and bounded busy policy with driver retries so a failed send cannot retry blindly or bypass the existing single-retry limit. Retain an unadmitted OTA request flight and count v2 fallback evidence only after a sparse retry enters the transmit queue. Local queue congestion no longer disables v2 before a request was sent or discards received fragments. Include primary airtime when temporary RX-only radio2 crosses requests onto that channel, keeping response allowances conservative on slower profiles. Add six native regressions, including five failing-before cases, and document the queue-admission and crossover timing behavior. Validation: 1,502 native tests, 29 integration checks, and five firmware builds passed across ESP32-S3, nRF52, and STM32. --- docs/ota_user_guide.md | 4 ++ src/Dispatcher.cpp | 80 +++++++++++++---------- src/Dispatcher.h | 1 + src/Mesh.cpp | 6 +- src/helpers/ota/OtaManager.cpp | 25 +++++-- src/helpers/ota/OtaManager.h | 1 + test/README.md | 7 +- test/fixtures/radio_profiles/mesh_tests.h | 67 +++++++++++++++++++ test/test_ota/test_ota_core.cpp | 69 +++++++++++++++++++ 9 files changed, 216 insertions(+), 44 deletions(-) diff --git a/docs/ota_user_guide.md b/docs/ota_user_guide.md index efed5eb87..a88016e05 100644 --- a/docs/ota_user_guide.md +++ b/docs/ota_user_guide.md @@ -422,6 +422,10 @@ maintenance rate even when LoRa pacing is slow. If the local transmit queue cannot accept a catalog, manifest, or leaf request, OTA keeps it pending without consuming a network retry. Successfully queued requests that receive no answer still have bounded retries. +Blocked block requests also preserve the v2 transfer format and fragments +already received; queue congestion alone does not trigger legacy fallback. +Retry timing includes the primary channel when an RX-only temporary radio2 +uses crossover to send requests there. `ota config speed 0.5` is an equivalent setter. `ota config speed` and `ota speed` read the current factor; the `ota config` summary also includes it. diff --git a/src/Dispatcher.cpp b/src/Dispatcher.cpp index deb54b6ea..3f131d2e3 100644 --- a/src/Dispatcher.cpp +++ b/src/Dispatcher.cpp @@ -355,7 +355,7 @@ void Dispatcher::loop() { // The failed send has already returned the chip to RX. Drain a packet // arriving during backoff before asking to retune; otherwise BUSY can // keep the retry waiting forever on an unread RxDone interrupt. - if (isDualRadioActive()) checkRecv(); + checkRecv(); // A cancelled or expired packet must retire even if the chip never // becomes available for another retune (including an armed OTA reboot). if (!isPacketRadioCurrent(outbound) || !allowPacketTransmit(outbound)) { @@ -370,16 +370,20 @@ void Dispatcher::loop() { return; } - outbound_radio_retry_pending = false; if (prepared == RadioParamApplyResult::FAILED || !isPacketRadioCurrent(outbound) || !allowPacketTransmit(outbound)) { MESH_DEBUG_PRINTLN("%s Dispatcher::loop(): radio retry packet no longer allowed, type=%u", getLogDateTime(), (uint32_t)outbound->getPayloadType()); failOutboundTransmit(); - } else if (!startOutboundTransmit()) { - failOutboundTransmit(); } else { - return; + uint32_t retry_delay; + if (!isTransmitChannelReady(outbound, retry_delay)) { + outbound_radio_retry_at = futureMillis(retry_delay); + return; + } + outbound_radio_retry_pending = false; + if (!allowPacketTransmit(outbound) || !startOutboundTransmit()) failOutboundTransmit(); + else return; } } else if (_radio->isSendComplete()) { long t = _ms->getMillis() - outbound_start; @@ -716,6 +720,37 @@ void Dispatcher::processRecvPacket(Packet* pkt) { } } +bool Dispatcher::isTransmitChannelReady(const Packet* packet, uint32_t& retry_delay) { + const bool busy = packet != NULL && usePassiveChannelCheck(packet) + ? _radio->isReceivingPassive(getRetryInterferenceMargin()) + : _radio->isReceiving(); + const uint8_t profile = packet ? packet->radio_profile : 0; + uint32_t& profile_busy = profile_cad_busy[profile]; + if (busy) { + const uint32_t now = _ms->getMillis(); + // A driver retry owns its packet outside the queue, but remains one of + // the packets waiting for the channel's bounded busy allowance. + const int ready_count = _mgr->getOutboundCount(now) + (packet && outbound == packet ? 1 : 0); + cad_busy_start = profile_busy; + if (cad_busy_start == 0) { + cad_busy_start = now; + profile_busy = now; + } + const uint32_t maximum = scaleCADDelayForQueue( + getCADFailMaxDuration(), ready_count, MIN_CAD_FAIL_MAX_DURATION_MS); + if (now - cad_busy_start <= maximum) { + retry_delay = scaleCADDelayForQueue( + getCADFailRetryDelay(), ready_count, MIN_CAD_FAIL_RETRY_DELAY_MS); + return false; + } + _err_flags |= ERR_EVENT_CAD_TIMEOUT; + MESH_DEBUG_PRINTLN("%s Dispatcher: CAD busy max duration reached!", getLogDateTime()); + } + cad_busy_start = 0; + profile_busy = 0; + return true; +} + void Dispatcher::checkSend() { if (_radio->isCarrierWaveActive()) return; const uint32_t now = _ms->getMillis(); @@ -789,38 +824,13 @@ void Dispatcher::checkSend() { } } - bool channel_busy = pending != NULL && usePassiveChannelCheck(pending) - ? _radio->isReceivingPassive(getRetryInterferenceMargin()) - : _radio->isReceiving(); - if (channel_busy) { - const uint32_t cad_now = _ms->getMillis(); - const int ready_count = _mgr->getOutboundCount(cad_now); - uint32_t& profile_busy = profile_cad_busy[pending ? pending->radio_profile : 0]; - cad_busy_start = profile_busy; - if (cad_busy_start == 0) { - cad_busy_start = cad_now; // record when CAD busy state started - profile_busy = cad_now; - } - - const uint32_t max_busy_duration = scaleCADDelayForQueue( - getCADFailMaxDuration(), ready_count, MIN_CAD_FAIL_MAX_DURATION_MS); - if (cad_now - cad_busy_start > max_busy_duration) { - _err_flags |= ERR_EVENT_CAD_TIMEOUT; - - MESH_DEBUG_PRINTLN("%s Dispatcher::checkSend(): CAD busy max duration reached!", getLogDateTime()); - // channel activity has gone on too long... (Radio might be in a bad state) - // force the pending transmit below... - } else { - const uint32_t retry_delay = scaleCADDelayForQueue( - getCADFailRetryDelay(), ready_count, MIN_CAD_FAIL_RETRY_DELAY_MS); - if (!isDualRadioActive() || !pending || !_mgr->deferOutbound(pending, futureMillis(retry_delay))) { - next_tx_time = futureMillis(retry_delay); - } - return; + uint32_t retry_delay; + if (!isTransmitChannelReady(pending, retry_delay)) { + if (!isDualRadioActive() || !pending || !_mgr->deferOutbound(pending, futureMillis(retry_delay))) { + next_tx_time = futureMillis(retry_delay); } + return; } - cad_busy_start = 0; // reset busy state - profile_cad_busy[pending ? pending->radio_profile : 0] = 0; // Retuning and CAD can advance the clock. Keep the selection timestamp so // a newly due higher-priority packet cannot skip its own profile, airtime, diff --git a/src/Dispatcher.h b/src/Dispatcher.h index 2768ca440..84a628903 100644 --- a/src/Dispatcher.h +++ b/src/Dispatcher.h @@ -317,6 +317,7 @@ class Dispatcher { void releaseDroppedOutbound(); bool startOutboundTransmit(); bool scheduleOutboundRadioRetry(); + bool isTransmitChannelReady(const Packet* packet, uint32_t& retry_delay); void failOutboundTransmit(); void restoreOutboundTxOverrides(); void updateTxBudget(); diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 9b661d88a..6835cdad4 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -640,7 +640,11 @@ uint32_t Mesh::getOtaPacketAirtime() const { // Replies may occupy both profiles even with crossing off or radio2 RX-only. uint8_t mask = (profiles->primary_temporary ? 1 : 0) | (profiles->secondary_temporary ? 2 : 0); if (!mask) mask = 1; - if (profiles->canCross() && profiles->canTransmit(1)) mask |= 3; + // Each temporary origin can cross independently. In particular, an RX-only + // secondary can still send requests over the primary when crossing is on. + for (uint8_t origin = 0; origin < 2; ++origin) { + if (mask & (1U << origin)) mask |= profiles->transmitMask(origin); + } if (profiles->reply_tx != RADIO_TX_AUTO) { mask |= explicitRadioTxMask(profiles->reply_tx, profiles->canTransmit(1, profiles->reply_force)); } diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index d0ba9d3de..c4a9e1ca5 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -1092,6 +1092,7 @@ void OtaManager::clearReassembly() { _req_count = 0; _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; _flight_dirty = false; + _flight_request_pending = false; noteFetchActivity(); } @@ -1904,7 +1905,7 @@ bool OtaManager::requestFlight() { item.block_idx = (uint16_t)slot.block; item.want_mask = wireRequestMask(slot.block, slot.need, slot.mask, slot.encoded_len); } - if (request.n_items == 0) return false; + if (request.n_items == 0) { _flight_request_pending = false; return false; } _req_start = request.items[0].block_idx; _req_count = request.n_items; uint8_t b[5 + OTA_REQ_MAX_ITEMS * 4]; @@ -1912,6 +1913,7 @@ bool OtaManager::requestFlight() { (unsigned)_req_start, (unsigned)_req_count, (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); bool sent = emit(b, encode_req_window(b, sizeof(b), request), false); + _flight_request_pending = !sent; if (sent) noteFetchActivity(); return sent; } @@ -1942,6 +1944,9 @@ bool OtaManager::fillPipeline() { void OtaManager::requestMissing() { if (_fstate != FETCHING) return; if (activePipelineSlots() == 0) { fillPipeline(); return; } + // Admission backpressure is not evidence that the source ignored v2. + // Keep the original flight and retry it before evaluating network loss. + if (_flight_request_pending) { requestFlight(); return; } // No v2 DATA at all means the request may have reached only deployed sources which ignored its profile // bits (or the first response burst was lost). Retry once in the universal legacy profile instead of @@ -1961,8 +1966,8 @@ void OtaManager::requestMissing() { // Once v2 has been confirmed, the source can still disappear cleanly between blocks (or after a bad // representation is discarded), leaving every active slot empty. Give the v2 flight one sparse retry, // then clear the whole flight and ask again in the universal legacy profile so an old seeder can take over. - if (_wire_v2_session && _wire_v2_confirmed && !anyWireDataReceived() && - ++_wire_empty_stalls >= 2) { + const bool empty_v2 = _wire_v2_session && _wire_v2_confirmed && !anyWireDataReceived(); + if (empty_v2 && _wire_empty_stalls >= 1) { downgradeWireToLegacy(); requestFlight(); return; @@ -1972,14 +1977,21 @@ void OtaManager::requestMissing() { uint8_t slot = (uint8_t)((_retry_slot + offset) % OTA_FETCH_PIPELINE); if (_reasm[slot].block == NO_BLOCK) continue; ReassemblySlot& reassembly = _reasm[slot]; - if (_wire_v2_session && reassembly.wire_v2 && reassembly.mask != 0 && - !reassembly.awaiting_proof && ++reassembly.wire_stalls >= 2) { + const bool partial_v2 = _wire_v2_session && reassembly.wire_v2 && reassembly.mask != 0 && + !reassembly.awaiting_proof; + if (partial_v2 && reassembly.wire_stalls >= 1) { // A v2 seeder may disappear after establishing a representation while only deployed legacy seeders // remain. Keep one sparse retry, then fall back the whole flight to the universal profile. Clearing // every slot prevents geometry/representation mixing and guarantees that old sources can take over. downgradeWireToLegacy(); } - if (requestSlot(slot)) _retry_slot = (uint8_t)((slot + 1) % OTA_FETCH_PIPELINE); + if (requestSlot(slot)) { + // Advance fallback evidence only when the sparse retry enters the + // transmit queue. Repeated rejection must not erase received fragments. + if (empty_v2 && _wire_v2_session) ++_wire_empty_stalls; + if (partial_v2 && _wire_v2_session) ++reassembly.wire_stalls; + _retry_slot = (uint8_t)((slot + 1) % OTA_FETCH_PIPELINE); + } return; } } @@ -2060,6 +2072,7 @@ void OtaManager::loop() { return; } if (_fstate != FETCHING) return; + if (_flight_request_pending) { requestFlight(); return; } if (activePipelineSlots() == 0) { requestMissing(); return; diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index 81e384a80..ad8e71504 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -738,6 +738,7 @@ private: uint8_t _retry_slot = 0; // round-robin stalled-slot retry cursor uint8_t _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; // live request window, adaptive up to array capacity bool _flight_dirty = false; // a timeout/bad proof required recovery in this flight + bool _flight_request_pending = false; // local queue has not accepted this flight yet uint32_t _fetch_wait_since_ms = 0; // last new fragment/proof/request; retry deadline anchor uint32_t _radio_packet_airtime_ms = 0; // measured for MAX_TRANS_UNIT at active SF/BW float _speed = OTA_SPEED_DEFAULT; diff --git a/test/README.md b/test/README.md index adeea2601..fadc45232 100644 --- a/test/README.md +++ b/test/README.md @@ -178,8 +178,11 @@ does not reflect the GoogleTest count -- run the built binary directly ordinary traffic during OTA pacing; `test_packet_manager` checks that intentional pacing survives stale expiry without erasing prior congestion age. - OTA delivery: `test_ota` keeps rejected queue admissions out of metadata retry - limits; `test_trace_retry` advances the clock during CAD to catch selection and - pacing bypasses and cancels retries while the chip stays BUSY. + limits and preserves v2 negotiation/received fragments during blocked block + requests; `test_trace_retry` advances the clock during CAD to catch selection + and pacing bypasses, cancels retries while the chip stays BUSY, and checks + channel clearance, RX draining, and bounded recovery for radio fault retries. + It also accounts for the primary's airtime when an RX-only radio2 crosses TX. `python3 -B test/test_ota_apply_delivery.py -v` executes the production apply/TX guards with a reply on air, a busy queue, failed persistence, and clock rollover. It checks application and bootloader reboot paths. diff --git a/test/fixtures/radio_profiles/mesh_tests.h b/test/fixtures/radio_profiles/mesh_tests.h index 0417cf331..6a06e0603 100644 --- a/test/fixtures/radio_profiles/mesh_tests.h +++ b/test/fixtures/radio_profiles/mesh_tests.h @@ -182,6 +182,73 @@ TEST_F(DualProfileTest, OtaAirtimeFollowsParticipatingProfilesInsteadOfScannerVi EXPECT_EQ(node.getOtaPacketAirtime(), 900u); } +TEST_F(DualProfileTest, OtaAirtimeIncludesPrimaryWhenCrossingFromReceiveOnlySecondary) { + radio.distinguish_airtime = true; + radio.config.secondary_temporary = true; + radio.config.secondary.mode = mesh::RadioProfileMode::Rx; + radio.config.cross = mesh::RadioCrossMode::On; + auto packet = makeFloodPacket(PAYLOAD_TYPE_OTA); + EXPECT_EQ(node.getTransmitProfileMask(&packet), 1u); + // The secondary receives the update, but outgoing requests cross onto the + // much slower primary. Both contribute to the response/retry allowance. + EXPECT_EQ(node.getOtaPacketAirtime(), 970u); +} + +TEST_F(DualProfileTest, RadioFaultRetryWaitsForChannelClearInSingleAndDualMode) { + node.flood_attempts = 0; + radio.config.cross = mesh::RadioCrossMode::Off; + for (bool dual : {false, true}) { + radio.config.secondary.mode = dual ? mesh::RadioProfileMode::RxTx : mesh::RadioProfileMode::Off; + radio.fail_next_send = true; + ASSERT_NE(nullptr, queue()); + tick(); ASSERT_FALSE(radio.sending); + const auto before = radio.transmissions.size(); + radio.busy_profile = 0; + tick(500); + EXPECT_FALSE(radio.sending); + EXPECT_EQ(radio.transmissions.size(), before); + radio.busy_profile = -1; + tick(500); + EXPECT_TRUE(radio.sending); + radio.complete = true; tick(); + EXPECT_EQ(manager.getFreeCount(), 40); + } +} + +TEST_F(DualProfileTest, SingleProfileRadioRetryDrainsReceivedPacketBeforeTransmitting) { + node.flood_attempts = 0; + radio.config.secondary.mode = mesh::RadioProfileMode::Off; + radio.fail_next_send = true; + ASSERT_NE(nullptr, queue()); + tick(); ASSERT_FALSE(radio.sending); + radio.incoming = {uint8_t(ROUTE_TYPE_DIRECT | (PAYLOAD_TYPE_GRP_TXT << PH_TYPE_SHIFT)), 0, 0x42}; + tick(500); + EXPECT_TRUE(radio.incoming.empty()); + EXPECT_EQ(radio.config.rx_packets[0], 1u); + EXPECT_EQ((std::vector{0}), radio.transmissions); + radio.complete = true; tick(); + EXPECT_EQ(manager.getFreeCount(), 40); +} + +TEST_F(DualProfileTest, RadioFaultRetryRetainsBoundedBusyEscapeAndSingleRetryLimit) { + node.flood_attempts = 0; + radio.config.cross = mesh::RadioCrossMode::Off; + radio.fail_next_send = true; + ASSERT_NE(nullptr, queue()); + tick(); ASSERT_FALSE(radio.sending); + radio.busy_profile = 0; + tick(500); EXPECT_FALSE(radio.sending); + // The same bounded CAD policy used for first sends still permits recovery + // from a permanently busy detector. It does not grant another radio retry. + radio.fail_next_send = true; + tick(10000); + EXPECT_FALSE(radio.fail_next_send); + EXPECT_FALSE(radio.sending); + EXPECT_EQ(manager.getFreeCount(), 40); + tick(10000); + EXPECT_TRUE(radio.transmissions.empty()); +} + TEST_F(DualProfileTest, PacketBecomingReadyDuringCadCannotSkipItsOwnChannelCheck) { node.flood_attempts = 0; node.tempRadioActive = true; diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index 85c1154fa..3fee375d3 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -2963,6 +2963,75 @@ TEST(OtaTransfer, NewClientRequestsV2ThenRetriesLegacyWhenNoV2DataArrives) { EXPECT_EQ(fallback.want_mask, 0x007Fu); } +TEST(OtaTransfer, RejectedFirstFlightDoesNotDisableV2BeforeAnyRequestWasSent) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); + OtaManager client; + OtaStoreRam<4096> store; + GatedCapture sent; + client.begin(SIM_TARGET_ID, gated_capture_send, &sent); + client.set_fetch_store(&store); + client.set_clock(100); + client.pull(manifest.merkle_root, manifest.target_id); + 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); + for (uint32_t i = 1; i <= 10; ++i) { + client.set_clock(100 + i * client.fetchRetryTimeoutMs()); + client.loop(); + } + EXPECT_TRUE(sent.items.empty()); + sent.accept = true; + client.loop(); + ASSERT_EQ(sent.items.size(), 1u); + ReqMsg first{}; + ASSERT_TRUE(decode_req(sent.items[0].data(), (uint16_t)sent.items[0].size(), first)); + EXPECT_TRUE(ota_req_is_v2(first.want_mask)); +} + +TEST(OtaTransfer, RejectedSparseRetryKeepsReceivedV2Fragments) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); + OtaManager client; + OtaStoreRam<4096> store; + GatedCapture sent; + sent.accept = true; + client.begin(SIM_TARGET_ID, gated_capture_send, &sent); + client.set_fetch_store(&store); + client.set_clock(100); + client.pull(manifest.merkle_root, manifest.target_id); + 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)); + uint8_t body[OTA_DATA_V2_STREAM_ID_BYTES + OTA_FRAG_DATA_V2]; + mh4(body, manifest.payload, manifest.block_size()); + memcpy(body + OTA_DATA_V2_STREAM_ID_BYTES, manifest.payload, OTA_FRAG_DATA_V2); + DataMsg partial{}; + memcpy(partial.manifest_id, manifest.merkle_root, 4); + ASSERT_TRUE(ota_data_v2_pack(0, (uint16_t)manifest.block_size(), false, partial.frag_off)); + partial.data = body; + partial.data_len = sizeof body; + uint8_t wire[MAX_PACKET_PAYLOAD]; + ASSERT_TRUE(client.on_message(wire, encode_data(wire, sizeof wire, partial))); + sent.items.clear(); + sent.accept = false; + uint32_t now = 100; + for (unsigned i = 0; i < 10; ++i) { + now += client.fetchRetryTimeoutMs(); + client.set_clock(now); + client.loop(); + } + EXPECT_TRUE(sent.items.empty()); + sent.accept = true; + client.loop(); + ASSERT_EQ(sent.items.size(), 1u); + ReqMsg retry{}; + ASSERT_TRUE(decode_req(sent.items[0].data(), (uint16_t)sent.items[0].size(), retry)); + EXPECT_TRUE(ota_req_is_v2(retry.want_mask)); + EXPECT_EQ(ota_req_v2_fragments(retry.want_mask) & 1u, 0u); // keep fragment zero +} + TEST(OtaTransfer, TwoKilobyteClientAcceptsCanonicalLegacyFallbackThroughBitTwelve) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA_2K, SIM_MOTA_2K_LEN, manifest));