From 5d69aa461fa24965a86bd8570d86ccf0bc1137b2 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 15 Sep 2026 06:45:28 -0700 Subject: [PATCH] Fix OTA retry accounting and radio delivery races Keep catalog, manifest and leaf requests pending when the local transmit queue rejects them instead of spending their network retry allowance. Keep the packet selected before CAD so work becoming due during sensing cannot bypass its own channel or pacing checks. Retire cancelled radio retries before waiting for BUSY hardware or their retry deadline. Wait for active transmission completion before applying firmware. Stop admitting new transmissions after the apply drain deadline so busy queues cannot postpone reboot indefinitely, and keep those deadlines armed across millis rollover. Retain the existing persistence-flush backoff. Add failing-before/fixed-after regression coverage for all four issues, including the production OTA apply and TX guards in CI. Validation: 1,496 native tests, 42 Python integration tests, and five firmware builds passed across ESP32-S3, nRF52, and STM32 targets. --- .github/workflows/run-unit-tests.yml | 1 + docs/ota_user_guide.md | 3 + src/Dispatcher.cpp | 12 +- src/Mesh.cpp | 17 ++- src/helpers/ota/OtaManager.cpp | 36 +++--- src/helpers/ota/OtaManager.h | 2 +- test/README.md | 6 + test/fixtures/radio_profiles/mesh_tests.h | 80 +++++++++++- test/test_companion_contact_sleep_contract.py | 2 +- test/test_ota/test_ota_core.cpp | 55 ++++++++ test/test_ota_apply_delivery.py | 122 ++++++++++++++++++ 11 files changed, 312 insertions(+), 24 deletions(-) create mode 100644 test/test_ota_apply_delivery.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 951f67056..9ebfa1f80 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -107,6 +107,7 @@ jobs: python3 -B test/test_temp_radio_reply_delivery_contract.py python3 -B test/test_reply_tx_integration.py python3 -B test/test_ota_speed.py -v + python3 -B test/test_ota_apply_delivery.py -v - name: Verify dual-profile switching and chirp timing working-directory: test diff --git a/docs/ota_user_guide.md b/docs/ota_user_guide.md index 58f36f68e..efed5eb87 100644 --- a/docs/ota_user_guide.md +++ b/docs/ota_user_guide.md @@ -419,6 +419,9 @@ OTA on either radio profile. `set ota.speed 1` restores the previous pace. Ordinary chat, node adverts, and USB/WiFi source transfers keep their timing. Local verification of a staged image and leaf comparisons keep their normal 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. `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 86da2c7a4..deb54b6ea 100644 --- a/src/Dispatcher.cpp +++ b/src/Dispatcher.cpp @@ -356,6 +356,12 @@ void Dispatcher::loop() { // arriving during backoff before asking to retune; otherwise BUSY can // keep the retry waiting forever on an unread RxDone interrupt. if (isDualRadioActive()) 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)) { + failOutboundTransmit(); + return; + } if (!millisHasNowPassed(outbound_radio_retry_at)) return; const auto prepared = _radio->prepareTransmitProfile(outbound->radio_profile, outbound->radio_reply && outbound->radio_reply_force); @@ -816,7 +822,11 @@ void Dispatcher::checkSend() { cad_busy_start = 0; // reset busy state profile_cad_busy[pending ? pending->radio_profile : 0] = 0; - outbound = _mgr->getNextOutbound(_ms->getMillis()); + // Retuning and CAD can advance the clock. Keep the selection timestamp so + // a newly due higher-priority packet cannot skip its own profile, airtime, + // and OTA pacing checks. A manager may also expire/reorder work on peek. + if (pending && _mgr->peekNextOutbound(now) != pending) return; + outbound = _mgr->getNextOutbound(now); if (outbound) { outbound_radio_retry_pending = false; outbound_radio_retry_used = false; diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 9793bc9fc..9b661d88a 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -413,16 +413,16 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { ota::OtaContext& oc = ota::ota_ctx(); if (oc.apply_pending) { if (oc.apply_at == 0) { - oc.apply_at = futureMillis(1500); - oc.apply_hard = futureMillis(15000); - } else if (millisHasNowPassed(oc.apply_at) && + oc.apply_at = ota::armedDeadline(_ms->getMillis(), 1500); + oc.apply_hard = ota::armedDeadline(_ms->getMillis(), 15000); + } else if (!hasOutbound() && millisHasNowPassed(oc.apply_at) && (_mgr->getOutboundTotal() == 0 || millisHasNowPassed(oc.apply_hard))) { if (!prepareForOtaReboot()) { // A role-specific persistence flush failed or is in bounded // backoff. Keep the verified apply armed without hammering storage // on every pass through the event loop. - oc.apply_at = futureMillis(1000); - oc.apply_hard = futureMillis(15000); + oc.apply_at = ota::armedDeadline(_ms->getMillis(), 1000); + oc.apply_hard = ota::armedDeadline(_ms->getMillis(), 15000); } else if (oc.bootloader_apply_pending) { ota::ota_reboot_to_bootloader_update(); } else { @@ -521,6 +521,13 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { } bool Mesh::allowPacketTransmit(const Packet* packet) const { +#if defined(ENABLE_OTA) && !defined(OTA_SEEDER_ONLY) + // Once the apply drain limit expires, let the active TX finish but decline + // new work. Otherwise a continuous queue can either prevent reboot forever + // or make the hard deadline cut off a packet that is still on air. + const ota::OtaContext* oc = ota::ota_context_if_active(); + if (oc && oc->apply_pending && oc->apply_at != 0 && millisHasNowPassed(oc->apply_hard)) return false; +#endif // This is an egress guard, separate from the receive-side TempRadio check below. A relay can queue an OTA // packet just before its temporary window closes; never let that delayed packet leak onto the normal channel. if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index 498a0a824..d0ba9d3de 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -924,13 +924,13 @@ void OtaManager::scheduleQuery(const uint8_t* seeder, const uint8_t* digest) { } } -void OtaManager::sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target, +bool OtaManager::sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target, uint32_t want_fragments) { QueryMsg q; memcpy(q.seeder_id, seeder, 4); memcpy(q.set_digest, digest, 4); q.filter_target = filter_target; q.want_fragments = want_fragments; uint8_t b[24]; - emit(b, encode_query(b, sizeof(b), q), true); // FLOODED so neighbours overhear it and suppress + return emit(b, encode_query(b, sizeof(b), q), true); // FLOODED so neighbours overhear it and suppress } // User-initiated browse (`ota neighbors`): immediately ask every incomplete/changed source. A complete @@ -939,16 +939,20 @@ void OtaManager::queryAll() { for (uint8_t i = 0; i < _n_src; i++) { Source& s = _sources[i]; if (s.have_catalog) continue; - s.query_pending = false; + s.query_pending = true; + s.query_at = _now_ms; s.query_owned = true; s.query_retries = 0; - s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + s.query_retry_at = 0; uint32_t want = 0; if (s.have_total) { uint32_t full = s.have_total >= 32 ? UINT32_MAX : ((1UL << s.have_total) - 1); want = full & ~s.have_mask; } - sendQuery(s.seeder, s.digest, 0, want); + if (sendQuery(s.seeder, s.digest, 0, want)) { + s.query_pending = false; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + } } } @@ -1995,14 +1999,15 @@ void OtaManager::loop() { for (uint8_t i = 0; i < _n_src; i++) { Source& s = _sources[i]; if (s.query_pending && (int32_t)(_now_ms - s.query_at) >= 0) { - s.query_pending = false; uint32_t want = 0; if (s.have_total) { uint32_t full = s.have_total >= 32 ? UINT32_MAX : ((1UL << s.have_total) - 1); want = full & ~s.have_mask; } - sendQuery(s.seeder, s.digest, 0, want); - s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + if (sendQuery(s.seeder, s.digest, 0, want)) { + s.query_pending = false; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + } } else if (s.query_owned && !s.have_catalog && !s.query_pending && s.query_retry_at && (int32_t)(_now_ms - s.query_retry_at) >= 0 && s.query_retries < OTA_CATALOG_MAX_RETRY) { uint32_t want = 0; @@ -2010,9 +2015,10 @@ void OtaManager::loop() { uint32_t full = s.have_total >= 32 ? UINT32_MAX : ((1UL << s.have_total) - 1); want = full & ~s.have_mask; } - s.query_retries++; - sendQuery(s.seeder, s.digest, 0, want); - s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + if (sendQuery(s.seeder, s.digest, 0, want)) { + s.query_retries++; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); + } } } if (_fstate == VERIFYING_STAGED) { @@ -2024,13 +2030,13 @@ void OtaManager::loop() { // the link and burn the retry cap while fragments are still arriving (mirrors FETCHING + WANT_LEAVES). // Give up after a cap of stalled retries so an unreachable mid doesn't pin the single fetch slot forever. if (_mf_mask == _loop_last_mfmask) { - if (++_mf_retries > OTA_MANIFEST_MAX_RETRY) { failFetch(FETCH_ERROR_MANIFEST_TIMEOUT); return; } + if (_mf_retries >= OTA_MANIFEST_MAX_RETRY) { failFetch(FETCH_ERROR_MANIFEST_TIMEOUT); return; } GetManifestMsg gm; memcpy(gm.manifest_id, _fid, 4); // request only the manifest fragments still missing; 0xFFFF ("send all") until we know frag_total gm.want_mask = (_mf_total > 0) ? (uint16_t)(frag_full_mask(_mf_total) & ~_mf_mask) : 0xFFFF; if (gm.want_mask == 0) gm.want_mask = 0xFFFF; // safety: never send an empty request uint8_t b[16]; - emit(b, encode_get_manifest(b, sizeof(b), gm), false); + if (emit(b, encode_get_manifest(b, sizeof(b), gm), false)) ++_mf_retries; } _loop_last_mfmask = _mf_mask; return; @@ -2041,14 +2047,14 @@ void OtaManager::loop() { // tick congests the link (and burns the retry cap) while fragments are still streaming in. On a stall, // ask for just the missing bitmap (anti-burst); give up (FAILED) after a cap of stalled retries. if (_lv_mask == _loop_last_lvmask) { - if (++_lv_retries > OTA_LEAVES_MAX_RETRY) { failFetch(FETCH_ERROR_LEAVES_TIMEOUT); return; } + if (_lv_retries >= OTA_LEAVES_MAX_RETRY) { failFetch(FETCH_ERROR_LEAVES_TIMEOUT); return; } GetLeavesMsg gl; memcpy(gl.manifest_id, _fid, 4); gl.want_mask = (_lv_total > 0) ? (uint16_t)(frag_full_mask(_lv_total) & ~_lv_mask) : 0xFFFF; if (gl.want_mask == 0) gl.want_mask = 0xFFFF; // safety: never send an empty request OTA_DBG("OTA: GET_LEAVES retry=%u want=%04x (mask=%04x/%u)\n", (unsigned)_lv_retries, (unsigned)gl.want_mask, (unsigned)_lv_mask, (unsigned)_lv_total); uint8_t b[16]; - emit(b, encode_get_leaves(b, sizeof(b), gl), false); + if (emit(b, encode_get_leaves(b, sizeof(b), gl), false)) ++_lv_retries; } _loop_last_lvmask = _lv_mask; return; diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index f4622af62..81e384a80 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -591,7 +591,7 @@ private: void popManifestJob(); bool loadActiveServeBlock(); void popServeJob(); - void sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target, + bool sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target, uint32_t want_fragments); // ask a source for all/selected catalog fragments void scheduleQuery(const uint8_t* seeder, const uint8_t* digest); // jittered + suppressible void reDiscover() { diff --git a/test/README.md b/test/README.md index e72e5f988..adeea2601 100644 --- a/test/README.md +++ b/test/README.md @@ -177,6 +177,12 @@ does not reflect the GoogleTest count -- run the built binary directly `test_trace_retry` checks both profile copies, stable dual-profile airtime, and 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. + `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. - To add a suite: create `test/test_/test_.cpp` with a `main()`, and add any host-only source it links to the `native` env's `build_src_filter` in `platformio.ini` (header-only code needs no source entry). No other wiring. diff --git a/test/fixtures/radio_profiles/mesh_tests.h b/test/fixtures/radio_profiles/mesh_tests.h index 3e6205e5f..0417cf331 100644 --- a/test/fixtures/radio_profiles/mesh_tests.h +++ b/test/fixtures/radio_profiles/mesh_tests.h @@ -5,6 +5,7 @@ class DualProfileTestRadio : public RetryCodingRateRadio { uint8_t selected = 0; int busy_profile = -1; bool fail_next_send = false; + bool hold_prepare_busy = false; bool carrier = false, carrier_on_receive = false; unsigned carrier_services = 0, recoveries = 0; bool isCarrierWaveActive() const override { return carrier; } @@ -12,6 +13,8 @@ class DualProfileTestRadio : public RetryCodingRateRadio { void loop() override { ++carrier_services; } bool recoverRadio(bool) override { ++recoveries; return true; } uint32_t estimated_airtime = 10; + TraceTestClock* sensing_clock = nullptr; + uint32_t sensing_delay = 0; bool distinguish_airtime = false; uint32_t getProfileAirtime(uint8_t profile, int bytes, uint8_t cr = 0) override { return distinguish_airtime ? (profile ? 70 : 900) @@ -35,12 +38,16 @@ class DualProfileTestRadio : public RetryCodingRateRadio { return prepareTransmitProfile(p, false); } mesh::RadioParamApplyResult prepareTransmitProfile(uint8_t p, bool reply_rx_override) override { + if (hold_prepare_busy) return mesh::RadioParamApplyResult::BUSY; if (!incoming.empty()) return mesh::RadioParamApplyResult::BUSY; if (!config.canTransmit(p, reply_rx_override)) return mesh::RadioParamApplyResult::FAILED; selected = p; cr = config.params(p).cr; return mesh::RadioParamApplyResult::APPLIED; } - bool isReceiving() override { return selected == busy_profile; } + bool isReceiving() override { + if (sensing_clock) sensing_clock->now += sensing_delay; + return selected == busy_profile; + } bool isReceivingPassive(int) override { return isReceiving(); } bool startSendRaw(const uint8_t* bytes, int size) override { if (fail_next_send) { fail_next_send = false; return false; } @@ -67,6 +74,10 @@ class DualProfileTestMesh : public RetryCodingRateMesh { using RetryCodingRateMesh::tryParsePacket; bool cross_filter_allows = true; float ota_speed = 1.0f; + bool suppress_tx = false; + bool allowPacketTransmit(const mesh::Packet* packet) const override { + return !suppress_tx && RetryCodingRateMesh::allowPacketTransmit(packet); + } float getOtaSpeedFactor() const override { return ota_speed; } unsigned cross_filter_calls = 0; bool allowRadioProfileCross(const mesh::Packet*) override { @@ -171,6 +182,73 @@ TEST_F(DualProfileTest, OtaAirtimeFollowsParticipatingProfilesInsteadOfScannerVi EXPECT_EQ(node.getOtaPacketAirtime(), 900u); } +TEST_F(DualProfileTest, PacketBecomingReadyDuringCadCannotSkipItsOwnChannelCheck) { + node.flood_attempts = 0; + node.tempRadioActive = true; + radio.config.cross = mesh::RadioCrossMode::Off; + auto* normal = node.obtainNewPacket(); + *normal = makeFloodPacket(PAYLOAD_TYPE_GRP_TXT); + normal->tx_radio = mesh::RADIO_TX_PRIMARY; + ASSERT_TRUE(node.sendPacket(normal, 1)); + auto* ota = node.obtainNewPacket(); + *ota = makeFloodPacket(PAYLOAD_TYPE_OTA); + ota->tx_radio = mesh::RADIO_TX_SECONDARY; + ASSERT_TRUE(node.sendPacket(ota, 0, 20)); + radio.busy_profile = 1; + radio.sensing_clock = &clock; + radio.sensing_delay = 50; // secondary becomes ready during primary's clear check + tick(); + ASSERT_EQ(radio.transmissions.size(), 1u); + EXPECT_EQ(radio.transmissions.back(), 0u); + radio.complete = true; + tick(); tick(); + EXPECT_EQ(radio.transmissions.size(), 1u); // busy secondary still must wait + radio.busy_profile = -1; + tick(1000); + ASSERT_EQ(radio.transmissions.size(), 2u); + EXPECT_EQ(radio.transmissions.back(), 1u); +} + +TEST_F(DualProfileTest, CadCannotPullPacedOtaAheadOfItsQuietDeadline) { + node.flood_attempts = 0; + node.tempRadioActive = true; + node.ota_speed = 0.05f; + radio.config.cross = mesh::RadioCrossMode::Off; + ASSERT_NE(queue(PAYLOAD_TYPE_OTA), nullptr); + tick(); radio.complete = true; tick(100); // 1900 ms quiet, ending at 2001 + auto* ota = node.obtainNewPacket(); + *ota = makeFloodPacket(PAYLOAD_TYPE_OTA); + ota->tx_radio = mesh::RADIO_TX_SECONDARY; + ASSERT_TRUE(node.sendPacket(ota, 0)); + auto* normal = node.obtainNewPacket(); + *normal = makeFloodPacket(PAYLOAD_TYPE_GRP_TXT); + normal->tx_radio = mesh::RADIO_TX_PRIMARY; + ASSERT_TRUE(node.sendPacket(normal, 1)); + radio.sensing_clock = &clock; + radio.sensing_delay = 150; // longer than OTA's short queue deferral + tick(); + ASSERT_EQ(radio.transmissions.size(), 2u); + EXPECT_EQ(radio.transmissions.back(), 0u); // ordinary packet, not early OTA + radio.complete = true; tick(); + radio.sensing_delay = 0; + tick(100); + EXPECT_EQ(radio.transmissions.size(), 2u); +} + +TEST_F(DualProfileTest, CancelledRadioRetryDoesNotWaitForBusyHardware) { + node.flood_attempts = 0; + radio.config.cross = mesh::RadioCrossMode::Off; + radio.fail_next_send = true; + ASSERT_NE(queue(), nullptr); + tick(); // failed start retains the packet for one radio retry + ASSERT_EQ(manager.getFreeCount(), 39); + node.suppress_tx = true; // OTA apply or a permission change cancels pending work + radio.hold_prepare_busy = true; + tick(1000); + EXPECT_EQ(manager.getFreeCount(), 40); + EXPECT_TRUE(radio.transmissions.empty()); +} + TEST_F(DualProfileTest, CrossFilterKeepsPrimaryTransmissionAndAvoidsCopyAllocation) { node.cross_filter_allows = false; ASSERT_NE(nullptr, queue()); diff --git a/test/test_companion_contact_sleep_contract.py b/test/test_companion_contact_sleep_contract.py index 2c5754ab5..769bb5f0a 100644 --- a/test/test_companion_contact_sleep_contract.py +++ b/test/test_companion_contact_sleep_contract.py @@ -168,7 +168,7 @@ class CompanionContactSleepContractTest(unittest.TestCase): guard_at = maintenance.index("prepareForOtaReboot()") apply_at = maintenance.index("ota::ota_reboot_to_apply()") self.assertLess(guard_at, apply_at) - self.assertIn("oc.apply_at = futureMillis(1000)", maintenance) + self.assertIn("oc.apply_at = ota::armedDeadline(_ms->getMillis(), 1000)", maintenance) self.assertIn("bool prepareForOtaReboot() override;", header) guard = function_body(mesh, "bool MyMesh::prepareForOtaReboot()") diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index cbc9913f4..85c1154fa 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -4533,6 +4533,61 @@ TEST(OtaTransfer, ManifestGiveUpAfterRetries) { EXPECT_EQ(client.wanted(), 0u); } +TEST(OtaTransfer, MetadataRetryLimitsDoNotCountRejectedQueueAdmission) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest)); + for (bool leaves : {false, true}) { + OtaManager client; + OtaStoreRam<4096> store; + GatedCapture sent; + client.begin(SIM_TARGET_ID, gated_capture_send, &sent); + client.set_fetch_store(&store); + ASSERT_EQ(client.pull(manifest.merkle_root, manifest.target_id, leaves), OtaManager::PULL_STARTED); + if (leaves) { + 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)); + } + const auto waiting = leaves ? OtaManager::WANT_LEAVES : OtaManager::WANT_MANIFEST; + ASSERT_EQ(client.fetchState(), waiting); + for (unsigned i = 0; i < 100; ++i) { client.set_clock(i * 1000); client.loop(); } + ASSERT_EQ(client.fetchState(), waiting); + EXPECT_TRUE(sent.items.empty()); + sent.accept = true; + client.loop(); + ASSERT_EQ(sent.items.size(), 1u); + EXPECT_EQ(ota_msg_type(sent.items[0].data(), sent.items[0].size()), leaves ? OTA_GET_LEAVES : OTA_GET_MANIFEST); + for (unsigned i = 0; i < 21; ++i) client.loop(); // actual unanswered sends remain bounded + EXPECT_EQ(client.fetchState(), OtaManager::FAILED); + } +} + +TEST(OtaTransfer, CatalogRetriesRemainAvailableAfterQueueBackpressure) { + for (bool manual : {false, true}) { + OtaManager client; + GatedCapture sent; + client.begin(SIM_TARGET_ID, gated_capture_send, &sent); + client.set_archive_interest(true); + AdvMsg adv{}; + adv.seeder_id[0] = 1; adv.n_motas = 1; + uint8_t wire[MAX_PACKET_PAYLOAD]; + client.on_message(wire, encode_adv(wire, sizeof wire, adv)); + if (manual) client.queryAll(); + for (unsigned i = 1; i <= 10; ++i) { client.set_clock(i * OTA_CATALOG_RETRY_MS); client.loop(); } + EXPECT_TRUE(sent.items.empty()); + sent.accept = true; + client.loop(); + ASSERT_EQ(sent.items.size(), 1u); + sent.accept = false; + for (unsigned i = 11; i <= 20; ++i) { client.set_clock(i * OTA_CATALOG_RETRY_MS); client.loop(); } + EXPECT_EQ(sent.items.size(), 1u); + sent.accept = true; + client.loop(); + ASSERT_EQ(sent.items.size(), 2u); + EXPECT_EQ(ota_msg_type(sent.items.back().data(), sent.items.back().size()), OTA_QUERY); + } +} + // A receiver never becomes a source, either while fetching or after completion. TEST(OtaTransfer, ReceiverDoesNotReSeed) { g_q.clear(); diff --git a/test/test_ota_apply_delivery.py b/test/test_ota_apply_delivery.py new file mode 100644 index 000000000..3747b808d --- /dev/null +++ b/test/test_ota_apply_delivery.py @@ -0,0 +1,122 @@ +"""Run the production deferred OTA apply and TX guard with an active radio.""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +from test_ota_dev_reboot_guard import function_body + +ROOT = Path(__file__).resolve().parents[1] + +HARNESS = r''' +#include +#include +#include +#define PAYLOAD_TYPE_OTA 0x0c +#define TOTAL_FLOOD_RETRY_SLOTS 1 +namespace mesh { +namespace ota { +struct OtaContext { + bool apply_pending=false, bootloader_apply_pending=false; + uint32_t apply_at=0, apply_hard=0; +}; +OtaContext context; +OtaContext* active=&context; +int applications=0, bootloaders=0; +OtaContext& ota_ctx() { return context; } +OtaContext* ota_context_if_active() { return active; } +void ota_reboot_to_apply() { ++applications; } +void ota_reboot_to_bootloader_update() { ++bootloaders; } +} +struct Packet { uint8_t getPayloadType() const { return 0; } }; +struct Clock { uint32_t now=0; uint32_t getMillis() const { return now; } }; +struct Manager { int queued=0; int getOutboundTotal() const { return queued; } }; +struct Mesh { + Clock clock; Clock* _ms=&clock; + Manager manager; Manager* _mgr=&manager; + Packet* outbound=nullptr; + bool flush_ok=true; + int _active_flood_retry_count=0; + struct { bool active=false, queued=false; Packet* packet=nullptr; uint8_t retry_attempts_sent=0; } _flood_retries[1]; + uint8_t getEligibleFloodRetryMaxAttempts(const Packet*) const { return 0; } + bool isAnyTempRadioActive() const { return true; } + bool hasOutbound() const { return outbound != nullptr; } + bool millisHasNowPassed(uint32_t deadline) const { return (int32_t)(clock.now-deadline)>0; } + uint32_t futureMillis(uint32_t ms) const { return clock.now+ms; } + bool prepareForOtaReboot() { return flush_ok; } + bool allowPacketTransmit(const Packet* packet) const { @ALLOW@ } + void serviceApply() { + ota::OtaContext& oc=ota::ota_ctx(); + if (oc.apply_pending) { @APPLY@ } + } +}; +} +int main() { + using namespace mesh; + Mesh node; + Packet packet; + auto reset = [&]() { + ota::context={}; ota::active=&ota::context; + ota::applications=ota::bootloaders=0; + node.clock.now=0; node.outbound=nullptr; + node.manager.queued=0; node.flush_ok=true; + ota::context.apply_pending=true; + }; + reset(); node.serviceApply(); + node.clock.now=2000; node.outbound=&packet; + node.serviceApply(); + assert(ota::applications==0); // dequeued is not the same as finished transmitting + node.outbound=nullptr; node.serviceApply(); + assert(ota::applications==1); + + reset(); node.serviceApply(); + node.clock.now=15001; node.outbound=&packet; node.manager.queued=99; + assert(!node.allowPacketTransmit(&packet)); // busy traffic cannot keep extending reboot + node.serviceApply(); assert(ota::applications==0); + node.outbound=nullptr; node.serviceApply(); + assert(ota::applications==1); + + reset(); node.serviceApply(); + node.clock.now=15001; node.flush_ok=false; + node.serviceApply(); assert(ota::applications==0); + assert(node.allowPacketTransmit(&packet)); // storage backoff releases the TX gate + node.flush_ok=true; node.clock.now+=1001; node.serviceApply(); + assert(ota::applications==1); + + reset(); node.clock.now=UINT32_MAX-1499; node.serviceApply(); + assert(ota::context.apply_at!=0); // exact rollover must not re-arm forever + node.clock.now=1; node.outbound=&packet; node.serviceApply(); + assert(ota::applications==0); + node.clock.now=2; node.outbound=nullptr; + ota::context.bootloader_apply_pending=true; node.serviceApply(); + assert(ota::bootloaders==1 && ota::applications==0); + + ota::context.apply_pending=false; + assert(node.allowPacketTransmit(&packet)); + ota::active=nullptr; + assert(node.allowPacketTransmit(&packet)); + assert(ota::active==nullptr); // idle inspection must not borrow an OTA workspace +} +''' + + +class OtaApplyDeliveryTest(unittest.TestCase): + def test_reply_completion_busy_queue_flush_failure_and_rollover(self): + source = (ROOT / "src/Mesh.cpp").read_text() + maintenance = function_body(source, "void __attribute__((noinline)) Mesh::serviceLoopMaintenance()") + apply = function_body(maintenance, "if (oc.apply_pending)") + allow = function_body(source, "bool Mesh::allowPacketTransmit(") + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler) + with tempfile.TemporaryDirectory() as directory: + cpp = Path(directory) / "apply.cpp" + binary = Path(directory) / "apply" + cpp.write_text(HARNESS.replace("@APPLY@", apply).replace("@ALLOW@", allow)) + subprocess.run([compiler, "-std=c++17", "-Wall", "-Wextra", "-DENABLE_OTA=1", + "-I", str(ROOT / "src"), str(cpp), "-o", str(binary)], check=True) + subprocess.run([str(binary)], check=True) + + +if __name__ == "__main__": + unittest.main()