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.
This commit is contained in:
mikecarper committed 2026-09-15 07:54:16 -07:00
1 parent 0582366527
commit dde87f1d15
9 files changed
+216 -44

No files matched your search

+4
View File
@@ -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.
+45 -35
View File
@@ -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,
+1
View File
@@ -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();
+5 -1
View File
@@ -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));
}
+19 -6
View File
@@ -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;
+1
View File
@@ -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;
+5 -2
View File
@@ -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.
+67
View File
@@ -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<uint8_t>{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;
+69
View File
@@ -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));