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()