From b2f2b5bf02f8bf77b8a3ae3dc7aed71cba12eb69 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Tue, 15 Sep 2026 06:16:29 -0700 Subject: [PATCH] Add persistent LoRa OTA speed control and harden timing Add get/set ota.speed and OTA CLI aliases for factors from 0.05 to 3, defaulting to the existing pace. Persist the setting independently of the OTA workspace and apply it to OTA packet pacing, relays, discovery and adverts while retaining the configured airtime budget. Preserve physical proof/retry allowances, account for both participating radio profiles, keep optional deadlines armed across millis rollover, and exclude intentional pacing from observer stale-packet age. Keep local verification at its normal rate and retain a valid saved speed if backup recovery cannot finish. Add documentation and CI regression coverage. Validation: 1,491 native tests and 26 CLI/integration tests passed. Built ESP32 full repeater, ESP32 full Companion, nRF52 full Companion and STM32 USB Companion successfully, including their memory/size checks. --- .github/workflows/run-unit-tests.yml | 1 + docs/cli_commands.md | 23 ++ docs/ota_easy.md | 7 + docs/ota_user_guide.md | 46 +++ examples/companion_radio/MyMesh.cpp | 9 + src/Dispatcher.cpp | 32 ++ src/Dispatcher.h | 9 + src/Mesh.cpp | 75 ++++- src/Mesh.h | 9 +- src/helpers/CommonCLI.cpp | 7 + src/helpers/RxReservePacketManager.h | 9 + src/helpers/ota/OtaCli.cpp | 11 +- src/helpers/ota/OtaManager.cpp | 67 +++- src/helpers/ota/OtaManager.h | 13 + src/helpers/ota/OtaSpeedConfig.cpp | 132 ++++++++ src/helpers/ota/OtaSpeedConfig.h | 15 + src/helpers/ota/OtaTiming.h | 71 +++++ test/README.md | 7 + test/fixtures/radio_profiles/mesh_tests.h | 71 +++++ test/test_ota/test_ota_core.cpp | 290 ++++++++++++++++++ test/test_ota_speed.py | 109 +++++++ .../test_rx_reserve_packet_manager.cpp | 55 ++++ 22 files changed, 1032 insertions(+), 36 deletions(-) create mode 100644 src/helpers/ota/OtaSpeedConfig.cpp create mode 100644 src/helpers/ota/OtaSpeedConfig.h create mode 100644 src/helpers/ota/OtaTiming.h create mode 100644 test/test_ota_speed.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index a6b8f89c8..951f67056 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -106,6 +106,7 @@ jobs: python3 -B test/test_repeater_radio_timing_integration.py 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 - name: Verify dual-profile switching and chirp timing working-directory: test diff --git a/docs/cli_commands.md b/docs/cli_commands.md index fd752fe67..fe3d76ab7 100644 --- a/docs/cli_commands.md +++ b/docs/cli_commands.md @@ -177,6 +177,29 @@ See [LoRa CLI host service](host_cli_service.md) for the complete --- +### LoRa OTA speed + +```text +get ota.speed +set ota.speed <0.05..3> +``` + +`1` preserves the current OTA timing. Values below `1` slow OTA traffic; +values above `1` shorten adjustable delays. For example, `0.5` doubles those +delays and `3` divides them by three. Decimal values such as `.05` work. +`ota config speed ` is an equivalent setter, and `ota config` shows +the value. The setting is saved and can be changed during a transfer. + +Apply it on the source, destination, and relays. It controls all LoRa OTA +packet types on both profiles, relay timing, discovery, and OTA adverts. +Slower settings extend recovery waits and space queued OTA packets while +allowing ordinary traffic through. Faster settings retain the physical retry +allowance and the configured airtime budget, so throughput is not guaranteed +to rise in direct proportion. Proof turnaround keeps its physical allowance. +Ordinary messages, node adverts, and local staged-image verification keep their +timing. Available in OTA-enabled builds for all roles, including Companions. +See the [OTA guide](ota_user_guide.md#adjust-lora-ota-speed). + ### Start or stop an Over-The-Air (OTA) firmware update **Search terms:** WiFi OTA, wireless firmware update, OTA uploader, update firmware. diff --git a/docs/ota_easy.md b/docs/ota_easy.md index 88e966e2a..87c6e8519 100644 --- a/docs/ota_easy.md +++ b/docs/ota_easy.md @@ -457,6 +457,13 @@ Monitor the transfer: ota status ``` +To adjust LoRa OTA pacing, use `get ota.speed` and `set ota.speed <0.05..3>` on +the source, destination, and OTA relays. `1` preserves the existing timing; +lower values slow OTA down, and higher values shorten adjustable delays. +This saved setting covers both profiles and OTA adverts as well as the transfer. +It does not change radio modulation or the configured airtime budget. See +[LoRa OTA speed](ota_user_guide.md#adjust-lora-ota-speed) for details. + For a powered bench update where restart-resume is not needed, `ota config checkpoint 0` on the destination removes periodic progress writes. It is a smaller optimization than selecting the correct hop count and it trades away persisted mid-download resume; restore the previous checkpoint cadence afterward. diff --git a/docs/ota_user_guide.md b/docs/ota_user_guide.md index 55a507280..58f36f68e 100644 --- a/docs/ota_user_guide.md +++ b/docs/ota_user_guide.md @@ -380,6 +380,52 @@ complete target inventory, storage layouts, and safety contract. --- +## Adjust LoRa OTA speed + +Use one saved speed factor for this node's LoRa OTA traffic: + +```text +get ota.speed +set ota.speed 0.5 +set ota.speed .05 +set ota.speed 3 +set ota.speed 1 +``` + +The range is **0.05 through 3**, including decimals. **1 is the existing timing**. +At `0.5`, adjustable OTA delays are twice as long; at `0.05`, they are twenty +times as long. At `3`, adjustable delays are divided by three. Actual transfer +speed also depends on packet airtime, the configured airtime budget, relays, +and packet loss; `3` does not guarantee three times the throughput. + +The factor covers OTA requests and responses (catalog, manifest, leaves, DATA, +and proofs), OTA relay delays, discovery jitter, and automatic OTA adverts. +Below `1`, the transmit queue also spaces out **every OTA packet**, including +both profile copies, while letting ordinary messages run. Recovery waits grow +for slower transfers; faster settings retain the existing retry allowance so +they do not retry before a physical response train can finish. The radio's +frequency, bandwidth, SF, coding rate, preamble, and airtime budget stay as set. +The proof turnaround allowance also keeps its existing physical minimum. +Dual-profile timing accounts for the participating profiles and reply copies; +it does not vary with the scanner's current visit. Deliberate slow pacing is +excluded from observer builds' stale-packet age, while congestion still ages +queued packets normally. + +Set the same factor on the **source, destination, and OTA relays**. It is a +local setting, not a value negotiated with peers. A slow source paired with a +receiver at `1` can cause unnecessary retries. The setting takes effect during +an active session, survives reboot and temporary-radio expiry, and applies to +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. + +`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. +The setting is available on OTA-enabled repeaters, rooms, sensors, Companions, +and seeder-only builds. Its separate `/ota_speed` settings file preserves the +existing preference layouts and does not need an active OTA workspace. + ## Optional: let it update automatically By default your node only *discovers* updates - it won't download or install on its own. If you want more diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 40307f3cf..2afdb4fc9 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -21,6 +21,9 @@ #if defined(ENABLE_OTA) && defined(OTA_HEAP_CONTEXT) #include #endif +#if defined(ENABLE_OTA) +#include +#endif #include "helpers/radiolib/RXPowerSaving.h" #include "helpers/radiolib/RxBoostedGainDefaults.h" #include "helpers/radiolib/CadTiming.h" @@ -1699,6 +1702,9 @@ void MyMesh::begin(bool has_display, bool radio_available) { initializeOfflineQueue(); BaseChatMesh::begin(); _radio_profiles.begin(_store->getPrimaryFS(), _radio, getRTCClock()); +#if defined(ENABLE_OTA) + mesh::ota::beginSpeedConfig(_store->getPrimaryFS()); +#endif const bool identity_loaded = _store->loadMainIdentity(self_id); const bool is_new_install = !identity_loaded @@ -2399,6 +2405,9 @@ bool MyMesh::handleLocalControlCommand(const char* command, char* reply, while (*command == ' ') command++; if (handleTxRoutingCommand(command, reply, reply_size)) return true; +#if COMPANION_FEATURE_OTA_CLI + if (mesh::ota::handleSpeedCommand(command, reply, reply_size)) return true; +#endif if (handleCompanionBluetoothCommand(command, reply, reply_size)) return true; if (handleCompanionWirelessCommand(command, reply, reply_size)) return true; diff --git a/src/Dispatcher.cpp b/src/Dispatcher.cpp index 78838dc3d..86da2c7a4 100644 --- a/src/Dispatcher.cpp +++ b/src/Dispatcher.cpp @@ -1,4 +1,5 @@ #include "Dispatcher.h" +#include "helpers/ota/OtaTiming.h" #if MESH_PACKET_LOGGING #include @@ -66,6 +67,7 @@ void Dispatcher::begin() { outbound_radio_retry_at = 0; outbound_radio_retry_pending = false; outbound_radio_retry_used = false; + ota_tx_airtime = 0; radio_nonrx_start = _ms->getMillis(); duty_cycle_window_ms = getDutyCycleWindowMs(); @@ -376,6 +378,10 @@ void Dispatcher::loop() { } else if (_radio->isSendComplete()) { long t = _ms->getMillis() - outbound_start; total_air_time += t; + if (outbound->getPayloadType() == PAYLOAD_TYPE_OTA) { + ota_tx_finished_at = _ms->getMillis(); + ota_tx_airtime = t > 0 ? (uint32_t)t : 0; + } //Serial.print(" airtime="); Serial.println(t); updateTxBudget(); @@ -707,6 +713,9 @@ void Dispatcher::processRecvPacket(Packet* pkt) { void Dispatcher::checkSend() { if (_radio->isCarrierWaveActive()) return; const uint32_t now = _ms->getMillis(); + if (ota_tx_airtime && now - ota_tx_finished_at >= ota::packetQuietTime(ota_tx_airtime, getOtaSpeedFactor())) { + ota_tx_airtime = 0; + } uint32_t next_outbound; if (_mgr->getNextOutboundTime(now, next_outbound)) { if ((int32_t)(next_outbound - now) > 0) return; @@ -724,6 +733,29 @@ void Dispatcher::checkSend() { failOutboundTransmit(); return; } + if (ota_tx_airtime) { + const uint32_t gap = ota::packetQuietTime(ota_tx_airtime, getOtaSpeedFactor()); + const uint32_t elapsed = now - ota_tx_finished_at; + if (elapsed >= gap) ota_tx_airtime = 0; + else if (pending && pending->getPayloadType() == PAYLOAD_TYPE_OTA) { + // Preserve this packet's priority and retry ownership. A short deferral + // lets ordinary traffic run and makes live speed changes take effect + // within 100 ms, even when the previous speed was much slower. + const uint32_t remaining = gap - elapsed; + int candidates = _mgr->getOutboundTotal(); + do { + if (candidates-- <= 0 || !_mgr->deferOutboundForPacing(pending, now, + now + (remaining > 100 ? 100 : remaining))) return; + pending = _mgr->peekNextOutbound(now); + } while (pending && pending->getPayloadType() == PAYLOAD_TYPE_OTA); + if (!pending) return; + if (!isPacketRadioCurrent(pending)) { + outbound = _mgr->getNextOutbound(now); + failOutboundTransmit(); + return; + } + } + } updateTxBudget(); uint32_t est_airtime = pending && _radio->profiles() diff --git a/src/Dispatcher.h b/src/Dispatcher.h index bc07f37a4..2768ca440 100644 --- a/src/Dispatcher.h +++ b/src/Dispatcher.h @@ -228,6 +228,12 @@ public: // lifecycle hook before returning each packet to the pool. virtual Packet* getNextDroppedOutbound() { return NULL; } virtual bool deferOutbound(Packet*, uint32_t) { return false; } + // Deliberate local pacing is not congestion. Managers with stale-packet + // expiry can pause that age for this interval, preserving any prior backlog. + virtual bool deferOutboundForPacing(Packet* packet, uint32_t now, uint32_t scheduled_for) { + (void)now; + return deferOutbound(packet, scheduled_for); + } virtual int getOutboundCount(uint32_t now) const = 0; virtual int getOutboundTotal() const = 0; // Returns the earliest runnable time in the queue. A queue with any overdue @@ -281,6 +287,8 @@ class Dispatcher { #endif uint8_t outbound_restore_cr; unsigned long next_tx_time; + uint32_t ota_tx_finished_at = 0; + uint32_t ota_tx_airtime = 0; unsigned long cad_busy_start; unsigned long radio_nonrx_start; unsigned long next_floor_calib_time, next_agc_reset_time; @@ -389,6 +397,7 @@ protected: virtual const char* getLogDateTime() { return ""; } virtual float getAirtimeBudgetFactor() const; + virtual float getOtaSpeedFactor() const { return 1.0f; } virtual int calcRxDelay(float score, uint32_t air_time) const; virtual bool shouldBypassRxDelay(const Packet* packet) { (void)packet; diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 46b8e8603..9793bc9fc 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -3,6 +3,7 @@ //#include #if defined(ENABLE_OTA) #include "helpers/ota/OtaContext.h" // OTA mesh-integration is centralized here so every role gets it +#include "helpers/ota/OtaSpeedConfig.h" #include "helpers/ota/OtaProtocol.h" // decode_adv -> the `ota neighbors` discovery table #if !defined(OTA_SEEDER_ONLY) #include "helpers/ota/OtaSelf.h" // ota_self_firmware -> auto-advertise our own image @@ -442,18 +443,25 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { } if (!_ota_temp_was_active) { _ota_temp_was_active = true; - _next_ota_tick = 0; - _next_ota_announce = 0; + _next_ota_tick = _ms->getMillis(); + _ota_announce_timer.arm(_ms->getMillis(), 0, 1.0f); _ota_announce_count = 0; } - ota::ota_ctx().manager.set_clock(_ms->getMillis()); - float ota_spacing = 1.0f + getAirtimeBudgetFactor(); - if (ota_spacing < 1.0f) ota_spacing = 1.0f; - if (ota_spacing > 10.0f) ota_spacing = 10.0f; - ota::ota_ctx().manager.set_link_timing( - _radio->getEstAirtimeFor(MAX_TRANS_UNIT), - (uint16_t)(ota_spacing * 1000.0f + 0.5f)); + const float ota_speed = getOtaSpeedFactor(); + if (_ota_timer_speed != ota_speed) { + _ota_announce_timer.rescale(_ms->getMillis(), _ota_timer_speed, ota_speed); + _ota_timer_speed = ota_speed; + // Re-arm from now; changing speed must not immediately issue a retry + // for a response that is still queued at the previous pace. + _next_ota_tick = futureMillis(ota::scaleDelay(OTA_RETRY_TICK_MS, ota_speed < 1.0f ? ota_speed : 1.0f)); + } + syncOtaTiming(ota::ota_ctx().manager); ota::ota_ctx().manager.serviceEgress(); + const uint32_t ota_loop_interval = ota::ota_ctx().manager.loopIntervalMs(OTA_RETRY_TICK_MS); + if ((int32_t)(_next_ota_tick - _ms->getMillis()) > (int32_t)ota_loop_interval) { + // Entering local verification must not inherit a long radio retry wait. + _next_ota_tick = futureMillis(ota_loop_interval); + } if (millisHasNowPassed(_next_ota_tick)) { // one-shot on first tick: resume an interrupted fetch left staged in flash before a reboot. Only adopt // a PARTIAL container (continue fetching the holes); a COMPLETE one is left for manual/auto-install, @@ -472,9 +480,9 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { #if defined(NRF52_PLATFORM) && defined(OTA_SD_STORE) ota::ota_ctx().serviceSdCache(_ms->getMillis()); // archive unseen mOTAs only while the receive slot is idle #endif - _next_ota_tick = futureMillis(OTA_RETRY_TICK_MS); + _next_ota_tick = futureMillis(ota::ota_ctx().manager.loopIntervalMs(OTA_RETRY_TICK_MS)); } - if (millisHasNowPassed(_next_ota_announce)) { // auto-advertise so peers discover us (tiny beacon) + if (_ota_announce_timer.ready(_ms->getMillis())) { // auto-advertise so peers discover us (tiny beacon) ota::OtaContext& oc = ota::ota_ctx(); bool in_burst = _ota_announce_count < OTA_ANNOUNCE_BURST; uint32_t mins = oc.manager.advert_mins(); // periodic cadence in minutes; 0 = disabled (boot burst only) @@ -492,7 +500,7 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { uint32_t gap = in_burst ? OTA_ANNOUNCE_BURST_MS : (mins != 0) ? mins * 60000UL : OTA_ANNOUNCE_DISABLED_POLL_MS; - _next_ota_announce = futureMillis(gap); + _ota_announce_timer.arm(_ms->getMillis(), gap, ota_speed); } #if !defined(OTA_SEEDER_ONLY) { // auto-install (once per COMPLETE fetch): only signed images, and apply_fetched enforces trust @@ -610,13 +618,50 @@ void Mesh::decayOtaRelayBackoff() { if (_ota_relay_backoff_level == 0) _ota_relay_decay_at = 0; } +float Mesh::getOtaSpeedFactor() const { +#if defined(ENABLE_OTA) + return ota::speedFactor(); +#else + return ota::OTA_SPEED_DEFAULT; +#endif +} + +uint32_t Mesh::getOtaPacketAirtime() const { + const auto* profiles = _radio->profiles(); + if (!profiles || !profiles->enabled()) return _radio->getProfileAirtime(0, MAX_TRANS_UNIT); + // Use the participating profiles, never the scanner's instantaneous visit. + // 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; + if (profiles->reply_tx != RADIO_TX_AUTO) { + mask |= explicitRadioTxMask(profiles->reply_tx, profiles->canTransmit(1, profiles->reply_force)); + } + uint32_t airtime = 0; + for (uint8_t profile = 0; profile < 2; ++profile) { + if (mask & (1U << profile)) airtime += _radio->getProfileAirtime(profile, MAX_TRANS_UNIT); + } + return airtime; +} + +#if defined(ENABLE_OTA) +void Mesh::syncOtaTiming(ota::OtaManager& manager) { + manager.set_clock(_ms->getMillis()); + float spacing = 1.0f + getAirtimeBudgetFactor(); + if (spacing < 1.0f) spacing = 1.0f; + if (spacing > 10.0f) spacing = 10.0f; + manager.set_link_timing(getOtaPacketAirtime(), (uint16_t)(spacing * 1000.0f + 0.5f)); + manager.set_speed(getOtaSpeedFactor()); +} +#endif + uint32_t Mesh::getOtaRetransmitDelay(const mesh::Packet* packet) { if (packet == NULL) return 0; // Active transfer packets use the role's configured flood collision window. // Deployment tooling can temporarily select txdelay 0.3 on managed relays; // the value is airtime-scaled by the role and therefore follows SF/BW. if (isPrimaryOtaTraffic(packet->payload, packet->payload_len)) { - return getRetransmitDelay(packet); + return ota::relayDelay(getRetransmitDelay(packet), getOtaSpeedFactor()); } decayOtaRelayBackoff(); uint32_t airtime = _radio->getProfileAirtime(packet->radio_profile, packet->getRawLength(), packet->tx_cr); @@ -629,7 +674,7 @@ uint32_t Mesh::getOtaRetransmitDelay(const mesh::Packet* packet) { 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); + return ota::relayDelay(_rng->nextInt(min_delay, max_delay + 1), getOtaSpeedFactor()); } int Mesh::calcRxDelayForPacket(const Packet* packet, float score, uint32_t air_time) { @@ -1212,7 +1257,7 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) { bool terminal_ota = false; #if defined(ENABLE_OTA) if (ota::OtaContext* context = ota::ota_context_if_active()) { - context->manager.set_clock(_ms->getMillis()); + syncOtaTiming(context->manager); context->manager.note_rx_path_hops(n); terminal_ota = context->manager.on_message(pkt->payload, pkt->payload_len); context->track_session(context->manager.fetchState(), _ms->getMillis()); diff --git a/src/Mesh.h b/src/Mesh.h index 6d32c3736..112467f6c 100644 --- a/src/Mesh.h +++ b/src/Mesh.h @@ -1,6 +1,7 @@ #pragma once #include +#include #include // OTA-over-LoRa transport is understood by every Mesh role even when the OTA manager/installer is not @@ -34,6 +35,8 @@ namespace mesh { +namespace ota { class OtaManager; } + #ifndef MAX_DIRECT_RETRY_SLOTS #define MAX_DIRECT_RETRY_SLOTS 6 #endif @@ -529,13 +532,17 @@ protected: // OTA mesh-integration is centralized in Mesh::begin()/loop()/dispatch, so every role (repeater, // companion, room, sensor, ...) gets fetch/serve/apply without per-example wiring. static bool otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool flood); + void syncOtaTiming(ota::OtaManager& manager); unsigned long _next_ota_tick = 0; - unsigned long _next_ota_announce = 0; // advertisements are scheduled only while temp radio is active + ota::LongTimer _ota_announce_timer; // supports long scaled intervals across millis rollovers + float _ota_timer_speed = ota::OTA_SPEED_DEFAULT; uint8_t _ota_announce_count = 0; // adverts sent so far (boot burst before settling to daily) bool _ota_resumed = false; // one-shot: resumed an interrupted fetch staged in flash on boot bool _ota_autoinstall_tried = false; // attempted auto-install for the current COMPLETE fetch bool _ota_temp_was_active = false; // detects entry into a temporary-radio window #endif + float getOtaSpeedFactor() const override; + uint32_t getOtaPacketAirtime() const; /** * \brief Perform search of local DB of matching GroupChannels. diff --git a/src/helpers/CommonCLI.cpp b/src/helpers/CommonCLI.cpp index 4e21ba2b6..25b99b90f 100644 --- a/src/helpers/CommonCLI.cpp +++ b/src/helpers/CommonCLI.cpp @@ -25,6 +25,7 @@ #if defined(ENABLE_OTA) #include "ota/OtaCli.h" #include "ota/OtaContext.h" // persist/sync OTA policy + signer allowlist with NodePrefs + #include "ota/OtaSpeedConfig.h" #endif #ifndef BRIDGE_MAX_BAUD @@ -734,6 +735,9 @@ static void formatSnrDbX4Short(char* dest, size_t dest_len, int16_t snr_x4) { void CommonCLI::loadPrefs(FILESYSTEM* fs) { _radio_profiles.begin(fs, _callbacks->getProfileRadio(), _rtc, true); +#if defined(ENABLE_OTA) + mesh::ota::beginSpeedConfig(fs); +#endif const bool display_settings_loaded = mesh::ui::loadDisplayPowerSettings(fs, false); (void)display_settings_loaded; bool is_fresh_install = false; @@ -2467,6 +2471,9 @@ void CommonCLI::handleCommand(uint32_t sender_timestamp, char* command, char* re if (mesh::wireless::control().handle(command, reply, 160, millis(), _callbacks->wirelessCommandSource(sender_timestamp))) return; if (_radio_profiles.handle(command, reply)) return; +#if defined(ENABLE_OTA) + if (mesh::ota::handleSpeedCommand(command, reply, 160)) return; +#endif if (strncmp(command, "set tempradio ", 14) == 0) { handleCommand(sender_timestamp, command + 4, reply); return; diff --git a/src/helpers/RxReservePacketManager.h b/src/helpers/RxReservePacketManager.h index fce0672a6..331a1c940 100644 --- a/src/helpers/RxReservePacketManager.h +++ b/src/helpers/RxReservePacketManager.h @@ -104,6 +104,15 @@ public: return StaticPoolPacketManager::peekNextOutbound(now); } + bool deferOutboundForPacing(mesh::Packet* packet, uint32_t now, uint32_t scheduled_for) override { + if (!StaticPoolPacketManager::deferOutbound(packet, scheduled_for)) return false; + uint32_t previous; + if ((int32_t)(scheduled_for - now) > 0 && lookupAge(packet, &previous)) { + recordAge(packet, previous + (scheduled_for - now)); + } + return true; + } + mesh::Packet* getNextDroppedOutbound() override { return _dropped.removeByIdx(0); } diff --git a/src/helpers/ota/OtaCli.cpp b/src/helpers/ota/OtaCli.cpp index 5500ecb6d..f18d75e5e 100644 --- a/src/helpers/ota/OtaCli.cpp +++ b/src/helpers/ota/OtaCli.cpp @@ -1,5 +1,6 @@ #include "OtaCli.h" #include "OtaContext.h" +#include "OtaSpeedConfig.h" #include "FolderMotaStore.h" // `ota pull folder` destination (set_mid on the connected folder store) #include "OtaVerify.h" #include "OtaSelf.h" @@ -152,11 +153,14 @@ static bool is_cmd(const char* a, const char* names, const char** rest) { static bool handle_dev(const char* d, char* reply, OtaContext& c); bool handle_ota_command(const char* command, char* reply, mesh::MainBoard& board) { + if (handleSpeedCommand(command, reply, 160)) return true; const char* a = command + 3; if (*a != 0 && *a != ' ') return false; while (*a == ' ') a++; if (!ota_acquire_context(reply, 160)) return true; OtaContext& c = ota_ctx(); + c.manager.set_clock(millis()); + c.manager.set_speed(speedFactor()); const char* rest = a; // ---- raw / internal primitives, tucked under `ota dev ...` ---- @@ -1003,9 +1007,10 @@ bool handle_ota_command(const char* command, char* reply, mesh::MainBoard& board sprintf(reply, "OK OTA reach = %ld hop%s (saved)%s", h, h == 1 ? "" : "s", h == 0 ? " - direct only" : ""); } else { // show current policy uint8_t af = c.manager.autofetch(); + char speed[16]; formatSpeed(speed, sizeof(speed)); #if defined(NRF52_PLATFORM) && defined(OTA_SD_STORE) bool cache_ready = c.ensureSdCache(); - sprintf(reply, "ota config: cache=%s/%u autofetch=%s autoinstall=%s checkpoint=%u advert=%umin hops=%u keys=%u", + snprintf(reply, 160, "ota config: speed=%sx cache=%s/%u autofetch=%s autoinstall=%s checkpoint=%u advert=%umin hops=%u keys=%u", speed, cache_ready ? (c.sd_cache.autoCaptureEnabled() ? "on" : "off") : "unavailable", cache_ready ? (unsigned)c.sd_cache.capturedCount() : 0, af == OtaManager::AUTOFETCH_ANY ? "any" : af == OtaManager::AUTOFETCH_SIGNED ? "signed" : "off", @@ -1014,11 +1019,11 @@ bool handle_ota_command(const char* command, char* reply, mesh::MainBoard& board (unsigned)c.manager.max_hops(), (unsigned)c.allow.count()); #else #if defined(OTA_SEEDER_ONLY) - sprintf(reply, "ota config: mode=seeder-only autofetch=off autoinstall=off checkpoint=%u advert=%umin hops=%u", + snprintf(reply, 160, "ota config: speed=%sx mode=seeder-only autofetch=off autoinstall=off checkpoint=%u advert=%umin hops=%u", speed, (unsigned)c.manager.checkpoint_blocks(), (unsigned)c.manager.advert_mins(), (unsigned)c.manager.max_hops()); #else - sprintf(reply, "ota config: autofetch=%s autoinstall=%s checkpoint=%u advert=%umin hops=%u keys=%u (persisted)", + snprintf(reply, 160, "ota config: speed=%sx autofetch=%s autoinstall=%s checkpoint=%u advert=%umin hops=%u keys=%u (persisted)", speed, af == OtaManager::AUTOFETCH_ANY ? "any" : af == OtaManager::AUTOFETCH_SIGNED ? "signed" : "off", c.autoinstall == OtaContext::AUTOINSTALL_TRUSTED ? "trusted" : "off", (unsigned)c.manager.checkpoint_blocks(), (unsigned)c.manager.advert_mins(), diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index 5b5fef67f..498a0a824 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -21,6 +21,37 @@ OtaManager::~OtaManager() { #endif } +bool OtaManager::set_speed(float speed) { + if (!validSpeed(speed)) return false; + if (_speed == speed) return true; + const float ratio = speed / _speed; + // Retry and proof-turnaround deadlines retain their physical allowance. + const float retry_ratio = (speed < 1.0f ? speed : 1.0f) / (_speed < 1.0f ? _speed : 1.0f); + auto rescale = [&](uint32_t& deadline, float change, bool optional = false) { + if ((int32_t)(deadline - _now_ms) > 0) { + const double remaining = ceil((double)(deadline - _now_ms) / change); + const uint32_t delay = remaining >= INT32_MAX ? INT32_MAX : (uint32_t)remaining; + deadline = optional ? armedDeadline(_now_ms, delay) : _now_ms + delay; + } + }; + for (uint8_t i = 0; i < _n_manifest_jobs; ++i) rescale(_manifest_jobs[i].ready_at, ratio); + for (uint8_t i = 0; i < _n_serve_jobs; ++i) { + if (_serve_jobs[i].proof_ready_at) rescale(_serve_jobs[i].proof_ready_at, retry_ratio, true); + } + for (uint8_t i = 0; i < _n_src; ++i) { + if (_sources[i].query_pending) rescale(_sources[i].query_at, ratio); + if (_sources[i].query_retry_at) rescale(_sources[i].query_retry_at, retry_ratio, true); + } + _speed = speed; + // A live change must not treat a response train paced at the old speed as + // lost, or immediately exhaust manifest/leaf retries on the next tick. + noteFetchActivity(); + for (auto& slot : _reasm) { + if (slot.awaiting_proof) slot.proof_request_at = armedDeadline(_now_ms, proofGraceMs()); + } + return true; +} + bool OtaManager::expandCatalog() { if (_catalog_heap || OTA_INLINE_CATALOG >= OTA_MAX_CATALOG) return false; CatRow* expanded = static_cast(malloc(sizeof(CatRow) * OTA_MAX_CATALOG)); @@ -336,7 +367,7 @@ void OtaManager::handleQuery(const uint8_t* m, uint16_t n) { } if (q.want_fragments == 0 || (missing != 0 && (q.want_fragments & missing) == missing)) { s.query_pending = false; - s.query_retry_at = _now_ms + OTA_CATALOG_RETRY_MS; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); } } if (_n_serve == 0 || memcmp(q.seeder_id, _seeder_id, 4) != 0) return; // (2) only WE answer queries to us @@ -480,16 +511,16 @@ bool OtaManager::queueManifestJob(const uint8_t* mid, uint16_t want_mask) { } uint32_t OtaManager::manifestEgressGapMs() const { - if (_radio_packet_airtime_ms == 0) return OTA_MANIFEST_EGRESS_MIN_GAP_MS; + if (_radio_packet_airtime_ms == 0) return pacedDelay(OTA_MANIFEST_EGRESS_MIN_GAP_MS); uint64_t gap = (uint64_t)_radio_packet_airtime_ms * _tx_spacing_permille; gap = (gap + 999u) / 1000u; if (gap < OTA_MANIFEST_EGRESS_MIN_GAP_MS) gap = OTA_MANIFEST_EGRESS_MIN_GAP_MS; if (gap > OTA_MANIFEST_EGRESS_MAX_GAP_MS) gap = OTA_MANIFEST_EGRESS_MAX_GAP_MS; - return (uint32_t)gap; + return pacedDelay((uint32_t)gap); } uint32_t OtaManager::proofEgressGapMs() const { - if (_radio_packet_airtime_ms == 0) return OTA_PROOF_EGRESS_MIN_GAP_MS; + if (_radio_packet_airtime_ms == 0) return retryDelay(OTA_PROOF_EGRESS_MIN_GAP_MS); // The adapter may have one response transmitting and two more admitted. Start this timer when the final // DATA is accepted, so cover all three packet-service intervals before opening the legacy-request turn. uint64_t gap = (uint64_t)_radio_packet_airtime_ms * _tx_spacing_permille @@ -497,7 +528,9 @@ uint32_t OtaManager::proofEgressGapMs() const { gap = (gap + 999u) / 1000u; if (gap < OTA_PROOF_EGRESS_MIN_GAP_MS) gap = OTA_PROOF_EGRESS_MIN_GAP_MS; if (gap > OTA_PROOF_EGRESS_MAX_GAP_MS) gap = OTA_PROOF_EGRESS_MAX_GAP_MS; - return (uint32_t)gap; + // This is a physical queue-drain/legacy-response allowance, not an optional + // transmit delay. Faster settings cannot make those packets finish earlier. + return retryDelay((uint32_t)gap); } void OtaManager::clearPendingEgress() { @@ -709,7 +742,7 @@ void OtaManager::serviceEgress() { (_serve_wire_len + fragment_data - 1u) / fragment_data); job.pending_mask &= valid_mask; if (job.pending_mask == 0) { - job.proof_ready_at = _now_ms + proofEgressGapMs(); + job.proof_ready_at = armedDeadline(_now_ms, proofEgressGapMs()); return; } uint8_t fragment = 0; @@ -752,7 +785,7 @@ void OtaManager::serviceEgress() { // proactive proof into the radio queue at the same instant: the two half-duplex transmissions would // collide and force the receiver onto its multi-second retry tick. Include every response that the // adapter can already have admitted ahead of this proof, then retain a bounded fast-link floor. - job.proof_ready_at = _now_ms + proofEgressGapMs(); + job.proof_ready_at = armedDeadline(_now_ms, proofEgressGapMs()); } } return; @@ -884,7 +917,7 @@ void OtaManager::scheduleQuery(const uint8_t* seeder, const uint8_t* digest) { if (s.query_pending || s.have_catalog || (s.query_owned && s.query_retry_at && (int32_t)(_now_ms - s.query_retry_at) < 0)) return; uint32_t j = (rd_u32le(seeder) ^ rd_u32le(digest) ^ rd_u32le(_seeder_id)) % OTA_QUERY_SPREAD_MS; - s.query_at = _now_ms + OTA_QUERY_MIN_MS + j; + s.query_at = _now_ms + pacedDelay(OTA_QUERY_MIN_MS + j); s.query_pending = true; s.query_owned = true; return; @@ -909,7 +942,7 @@ void OtaManager::queryAll() { s.query_pending = false; s.query_owned = true; s.query_retries = 0; - s.query_retry_at = _now_ms + OTA_CATALOG_RETRY_MS; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); uint32_t want = 0; if (s.have_total) { uint32_t full = s.have_total >= 32 ? UINT32_MAX : ((1UL << s.have_total) - 1); @@ -953,7 +986,7 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) { s.query_retries = 0; s.query_retry_at = 0; } else { - s.query_retry_at = _now_ms + OTA_CATALOG_RETRY_MS; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); } } for (uint8_t r = 0; r < hv.n_rows && hv.rows; r++) { @@ -1123,7 +1156,7 @@ static uint8_t count_mask_bits(uint16_t mask) { } uint32_t OtaManager::fetchRetryTimeoutMs() const { - if (_radio_packet_airtime_ms == 0) return OTA_FETCH_RETRY_FALLBACK_MS; + if (_radio_packet_airtime_ms == 0) return retryDelay(OTA_FETCH_RETRY_FALLBACK_MS); // Estimate every packet still owed in the current flight, plus the recovery request itself. DATA uses // the measured maximum packet airtime; smaller REQ/PROOF packets are deliberately overestimated. Each @@ -1150,11 +1183,11 @@ uint32_t OtaManager::fetchRetryTimeoutMs() const { service += OTA_FETCH_RETRY_GUARD_MS; if (service < OTA_FETCH_RETRY_MIN_MS) service = OTA_FETCH_RETRY_MIN_MS; if (service > OTA_FETCH_RETRY_MAX_MS) service = OTA_FETCH_RETRY_MAX_MS; - return (uint32_t)service; + return retryDelay((uint32_t)service); } uint32_t OtaManager::proofGraceMs() const { - if (_radio_packet_airtime_ms == 0) return OTA_PROOF_GRACE_MS; + if (_radio_packet_airtime_ms == 0) return retryDelay(OTA_PROOF_GRACE_MS); uint32_t path_transmissions = _observed_path_transmissions; if (path_transmissions == 0) path_transmissions = (uint32_t)_max_hops + 1u; uint64_t grace = (uint64_t)_radio_packet_airtime_ms * path_transmissions; @@ -1162,7 +1195,7 @@ uint32_t OtaManager::proofGraceMs() const { grace += 100u; if (grace < OTA_PROOF_GRACE_MS) grace = OTA_PROOF_GRACE_MS; if (grace > OTA_FETCH_RETRY_MAX_MS) grace = OTA_FETCH_RETRY_MAX_MS; - return (uint32_t)grace; + return retryDelay((uint32_t)grace); } void OtaManager::finishFlight() { @@ -1756,7 +1789,7 @@ bool OtaManager::handleData(const uint8_t* m, uint16_t n) { // can arrive without another request/response turn. An older server, or a lost proof, falls back through // serviceEgress() to the existing REQ_PROOF wire message. slot.awaiting_proof = true; - slot.proof_request_at = _now_ms + proofGraceMs(); + slot.proof_request_at = armedDeadline(_now_ms, proofGraceMs()); return true; } @@ -1969,7 +2002,7 @@ void OtaManager::loop() { want = full & ~s.have_mask; } sendQuery(s.seeder, s.digest, 0, want); - s.query_retry_at = _now_ms + OTA_CATALOG_RETRY_MS; + 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; @@ -1979,7 +2012,7 @@ void OtaManager::loop() { } s.query_retries++; sendQuery(s.seeder, s.digest, 0, want); - s.query_retry_at = _now_ms + OTA_CATALOG_RETRY_MS; + s.query_retry_at = armedDeadline(_now_ms, retryDelay(OTA_CATALOG_RETRY_MS)); } } if (_fstate == VERIFYING_STAGED) { diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index fbd6c9627..f4622af62 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -4,6 +4,7 @@ #include #include "OtaFormat.h" #include "OtaMemoryPolicy.h" +#include "OtaTiming.h" #include "OtaProtocol.h" #include "OtaByteIO.h" #include "OtaStore.h" @@ -386,6 +387,17 @@ public: void queryAll(); // Coarse clock for source/catalog ages + LRU (the Mesh adapter feeds millis; 0 in host tests is fine). void set_clock(uint32_t ms) { _now_ms = ms; } + bool set_speed(float speed); + float speed() const { return _speed; } + uint32_t pacedDelay(uint32_t ms) const { return scaleDelay(ms, _speed); } + // Faster pacing cannot shorten a loss-recovery deadline below the existing + // physical packet-flight allowance. Slower pacing expands that allowance. + uint32_t retryDelay(uint32_t ms) const { return scaleDelay(ms, _speed < 1.0f ? _speed : 1.0f); } + uint32_t loopIntervalMs(uint32_t normal_ms) const { + // Resume verification and local leaf comparisons do no radio waiting. + return _fstate == VERIFYING_STAGED || (_fstate == WANT_LEAVES && _diffing) + ? normal_ms : retryDelay(normal_ms); + } // Feed the actual TempRadio packet airtime and dispatcher transmit-spacing factor. The fetch retry // deadline then follows SF/BW, configured airtime budget, flight width, and observed/configured path. void set_link_timing(uint32_t max_packet_airtime_ms, uint16_t tx_spacing_permille) { @@ -728,6 +740,7 @@ private: bool _flight_dirty = false; // a timeout/bad proof required recovery in this flight 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; uint16_t _tx_spacing_permille = 2000; // 1/duty-cycle; default airtime factor 1 => 50% TX uint8_t _observed_path_transmissions = 0; // source + relays; 0 falls back to configured max_hops+1 uint32_t _packets_sent = 0; // OTA packets accepted by the radio adapter (wrap-safe) diff --git a/src/helpers/ota/OtaSpeedConfig.cpp b/src/helpers/ota/OtaSpeedConfig.cpp new file mode 100644 index 000000000..0e8466b86 --- /dev/null +++ b/src/helpers/ota/OtaSpeedConfig.cpp @@ -0,0 +1,132 @@ +#if defined(ENABLE_OTA) +#include "OtaSpeedConfig.h" +#include +#include +#include +#include + +namespace mesh { namespace ota { +namespace { +FILESYSTEM* settings_fs = nullptr; +float factor = OTA_SPEED_DEFAULT; +bool held = false; +constexpr size_t ImageSize = 12; +const char* const Path = "/ota_speed"; +const char* const Temp = "/ota_speed.tmp"; +const char* const Backup = "/ota_speed.bak"; + +bool readImage(const char* path, uint8_t* image) { +#if defined(NRF52_PLATFORM) + File file(*settings_fs); + if (!file.open(path, FILE_O_READ)) return false; +#elif defined(STM32_PLATFORM) + File file = settings_fs->open(path, FILE_O_READ); +#else + File file = settings_fs->open(path, "r"); +#endif + if (!file) return false; + const bool complete = file.size() == ImageSize && file.read(image, ImageSize) == (int)ImageSize; + file.close(); + if (!complete) return false; + float value; + memcpy(&value, image + 4, sizeof(value)); + return !memcmp(image, "OS\1\0", 4) && validSpeed(value) + && storage::readLE32(image + 8) == storage::updateCRC32(0xffffffffU, image, 8); +} + +bool saveSpeed(float value) { + if (!settings_fs || held) return false; + uint8_t image[ImageSize] = {'O', 'S', 1, 0}; + memcpy(image + 4, &value, sizeof(value)); + storage::writeLE32(image + 8, storage::updateCRC32(0xffffffffU, image, 8)); + if (settings_fs->exists(Temp) && !settings_fs->remove(Temp)) return false; +#if defined(NRF52_PLATFORM) + File file(*settings_fs); + if (!file.open(Temp, FILE_O_WRITE)) return false; +#elif defined(STM32_PLATFORM) + File file = settings_fs->open(Temp, FILE_O_WRITE); +#else + File file = settings_fs->open(Temp, "w"); +#endif + if (!file) return false; + bool ok = file.write(image, sizeof(image)) == sizeof(image); + file.flush(); file.close(); + uint8_t verify[ImageSize]; + if (!ok || !readImage(Temp, verify) || memcmp(image, verify, sizeof(image))) return false; + if (settings_fs->exists(Backup) && !settings_fs->remove(Backup)) return false; + const bool had_primary = settings_fs->exists(Path); + if (had_primary && !settings_fs->rename(Path, Backup)) return false; + if (!settings_fs->rename(Temp, Path)) { + if (had_primary && !settings_fs->rename(Backup, Path)) held = true; + return false; + } + if (had_primary) settings_fs->remove(Backup); + factor = value; // publish only after the complete image commits + return true; +} +} + +void beginSpeedConfig(FILESYSTEM* fs) { + settings_fs = fs; factor = OTA_SPEED_DEFAULT; held = false; + if (!fs) return; + uint8_t image[ImageSize]; + bool loaded = readImage(Path, image); + if (!loaded && readImage(Backup, image)) { + // A verified backup is still authoritative when storage cannot repair the + // rename gap. Keep its pace instead of silently reverting to the faster 1x. + memcpy(&factor, image + 4, sizeof(factor)); + if (fs->exists(Path) && !fs->remove(Path)) { held = true; return; } + if (!fs->rename(Backup, Path)) { held = true; return; } + loaded = true; + } + if (loaded) memcpy(&factor, image + 4, sizeof(factor)); + else held = fs->exists(Path) || fs->exists(Backup); +} + +float speedFactor() { return factor; } + +void formatSpeed(char* text, size_t capacity) { + const uint32_t scaled = (uint32_t)(factor * 1000000.0f + 0.5f); + snprintf(text, capacity, "%u.%06u", (unsigned)(scaled / 1000000), (unsigned)(scaled % 1000000)); + if (!capacity) return; + size_t n = strlen(text); + while (n && text[n - 1] == '0') text[--n] = 0; + if (n && text[n - 1] == '.') text[n - 1] = 0; +} + +bool handleSpeedCommand(const char* command, char* reply, size_t capacity) { + const char* value = nullptr; + bool require_value = false, read_only = false; + const char* const forms[] = {"set ota.speed", "get ota.speed", "ota config speed", "ota cfg speed", "ota set speed", "ota speed"}; + for (size_t i = 0; i < sizeof(forms) / sizeof(forms[0]); ++i) { + const size_t n = strlen(forms[i]); + if (!strncmp(command, forms[i], n) && (command[n] == 0 || command[n] == ' ')) { + value = command + n; + while (*value == ' ') ++value; + require_value = i == 0; read_only = i == 1; + break; + } + } + if (!value) return false; + if ((require_value && !*value) || (read_only && *value)) { + snprintf(reply, capacity, "ERR usage: set ota.speed <0.05..3> | get ota.speed"); + return true; + } + if (*value) { + float parsed; + if (!cli::parseDecimalStrict(value, parsed) || !validSpeed(parsed)) { + snprintf(reply, capacity, "ERR ota.speed must be 0.05..3 (1=current speed)"); + return true; + } + if (!saveSpeed(parsed)) { + snprintf(reply, capacity, "ERR ota.speed save failed; speed unchanged"); + return true; + } + } + char number[16]; formatSpeed(number, sizeof(number)); + snprintf(reply, capacity, *value ? "OK ota.speed=%sx (saved)" : "> ota.speed=%sx", number); + return true; +} + +} } +#endif diff --git a/src/helpers/ota/OtaSpeedConfig.h b/src/helpers/ota/OtaSpeedConfig.h new file mode 100644 index 000000000..36e8e3498 --- /dev/null +++ b/src/helpers/ota/OtaSpeedConfig.h @@ -0,0 +1,15 @@ +#pragma once + +#include +#include "OtaTiming.h" + +namespace mesh { namespace ota { + +// A small, separate settings image shared by every OTA-capable role. Reading +// or changing speed never borrows the Companion's OTA workspace. +void beginSpeedConfig(FILESYSTEM* fs); +float speedFactor(); +void formatSpeed(char* text, size_t capacity); +bool handleSpeedCommand(const char* command, char* reply, size_t capacity); + +} } diff --git a/src/helpers/ota/OtaTiming.h b/src/helpers/ota/OtaTiming.h new file mode 100644 index 000000000..fbe3c3ffb --- /dev/null +++ b/src/helpers/ota/OtaTiming.h @@ -0,0 +1,71 @@ +#pragma once + +#include +#include + +namespace mesh { +namespace ota { + +constexpr float OTA_SPEED_DEFAULT = 1.0f; +constexpr float OTA_SPEED_MIN = 0.05f; +constexpr float OTA_SPEED_MAX = 3.0f; + +inline bool validSpeed(float speed) { + return isfinite(speed) && speed >= OTA_SPEED_MIN && speed <= OTA_SPEED_MAX; +} + +// Relative deadlines must fit the signed half of millis(), including after +// rollover. Keep zero immediate and preserve every existing delay at 1x. +inline uint32_t scaleDelay(uint32_t delay_ms, float speed) { + if (!validSpeed(speed)) speed = OTA_SPEED_DEFAULT; + const double scaled = ceil((double)delay_ms / speed); + return scaled >= INT32_MAX ? INT32_MAX : (uint32_t)scaled; +} + +// For optional deadlines whose owners use zero to mean "not armed". A timer +// that lands exactly on millis rollover must still fire (one millisecond late). +inline uint32_t armedDeadline(uint32_t now, uint32_t delay_ms) { + const uint32_t deadline = now + delay_ms; + return deadline ? deadline : 1; +} + +// Extra quiet time between OTA packets when slowing an otherwise immediately +// drainable queue. Actual RF airtime and the dispatcher's airtime budget remain +// unchanged. At 1x and above, the existing queue can drain at its normal rate. +inline uint32_t packetQuietTime(uint32_t airtime_ms, float speed) { + if (!validSpeed(speed) || speed >= OTA_SPEED_DEFAULT) return 0; + const uint32_t cycle = scaleDelay(airtime_ms, speed); + return cycle > airtime_ms ? cycle - airtime_ms : 0; +} + +inline uint32_t relayDelay(uint32_t delay_ms, float speed) { + // DispatcherAction packs the priority above its 24-bit delay. An extreme + // slow setting must not spill into those priority bits. + const uint32_t scaled = scaleDelay(delay_ms, speed); + return scaled > 0x00ffffffU ? 0x00ffffffU : scaled; +} + +// Advert intervals can reach seven days; at 0.05x that is 140 days. Count +// down in real elapsed milliseconds instead of overflowing a millis deadline. +class LongTimer { + uint64_t remaining_ = 0; + uint32_t last_ = 0; +public: + bool ready(uint32_t now) { + const uint32_t elapsed = now - last_; + last_ = now; + remaining_ = remaining_ > elapsed ? remaining_ - elapsed : 0; + return remaining_ == 0; + } + void arm(uint32_t now, uint32_t delay_ms, float speed) { + last_ = now; + remaining_ = (uint64_t)ceil((double)delay_ms / (validSpeed(speed) ? speed : 1.0f)); + } + void rescale(uint32_t now, float old_speed, float new_speed) { + ready(now); + remaining_ = (uint64_t)ceil((double)remaining_ * old_speed / new_speed); + } +}; + +} // namespace ota +} // namespace mesh diff --git a/test/README.md b/test/README.md index f5c51e184..e72e5f988 100644 --- a/test/README.md +++ b/test/README.md @@ -170,6 +170,13 @@ does not reflect the GoogleTest count -- run the built binary directly suite may need shims **before** the include -- e.g. `test_mqtt_presets` does `#define WITH_MQTT_BRIDGE 1` (the preset table is behind that flag) and `#define PROGMEM` (the embedded CA-cert strings are PROGMEM-qualified). +- OTA speed: `python3 -B test/test_ota_speed.py -v` exercises the shared CLI, + saved settings, failed writes, and failed backup recovery. Native `test_ota` + covers complete transfers with loss, live changes, physical proof allowances, + local verification, discovery, and deadlines landing exactly on clock rollover. + `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. - 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 1384cd2cc..3e6205e5f 100644 --- a/test/fixtures/radio_profiles/mesh_tests.h +++ b/test/fixtures/radio_profiles/mesh_tests.h @@ -62,8 +62,12 @@ class DualProfileTestMesh : public RetryCodingRateMesh { using RetryCodingRateMesh::isPacketRadioCurrent; using RetryCodingRateMesh::getTransmitAirtime; using RetryCodingRateMesh::getTransmitProfileMask; + using RetryCodingRateMesh::getRetransmitDelay; + using RetryCodingRateMesh::getOtaPacketAirtime; using RetryCodingRateMesh::tryParsePacket; bool cross_filter_allows = true; + float ota_speed = 1.0f; + float getOtaSpeedFactor() const override { return ota_speed; } unsigned cross_filter_calls = 0; bool allowRadioProfileCross(const mesh::Packet*) override { ++cross_filter_calls; @@ -100,6 +104,73 @@ class DualProfileTest : public testing::Test { void tick(uint32_t ms = 1) { clock.now += ms; node.loop(); } }; +TEST_F(DualProfileTest, OtaSpeedPacesBothCopiesAndYieldsToOrdinaryTraffic) { + node.flood_attempts = 0; + node.tempRadioActive = true; + node.ota_speed = 0.05f; + radio.config.cross = mesh::RadioCrossMode::On; + auto* first = queue(PAYLOAD_TYPE_OTA); + ASSERT_NE(nullptr, first); + tick(); ASSERT_TRUE(radio.sending); + radio.complete = true; + tick(100); // actual OTA TX airtime: 100 ms; quiet allowance: 1900 ms + EXPECT_EQ(radio.transmissions.size(), 1u); + auto* normal = queue(PAYLOAD_TYPE_GRP_TXT); + ASSERT_NE(nullptr, normal); + tick(); + ASSERT_TRUE(radio.sending); // the OTA copy yields to normal messages + EXPECT_EQ(radio.transmissions.size(), 2u); + radio.complete = true; tick(100); + tick(); // the normal dispatcher's next-TX deadline opens on the following loop + ASSERT_TRUE(radio.sending); + radio.complete = true; tick(100); + EXPECT_EQ(radio.transmissions.size(), 3u); // both normal-message profiles sent + tick(1000); + EXPECT_EQ(radio.transmissions.size(), 3u); + node.ota_speed = 1.0f; + tick(101); // discard the old slow wait without losing the secondary copy + EXPECT_EQ(radio.transmissions.size(), 4u); + EXPECT_EQ(radio.transmissions.back(), 1u); + radio.complete = true; tick(100); + EXPECT_EQ(manager.getFreeCount(), 40); +} + +TEST_F(DualProfileTest, OtaSpeedScalesRelayWindowButLeavesOrdinaryRelaysAlone) { + rng.value = 2; + auto ota_packet = makeFloodPacket(PAYLOAD_TYPE_OTA); + ota_packet.payload[0] = mesh::ota::OTA_DATA; + auto ordinary_packet = makeFloodPacket(PAYLOAD_TYPE_GRP_TXT); + const auto ordinary_delay = node.getRetransmitDelay(&ordinary_packet); + const auto normal = node.otaRelayDelay(&ota_packet); + ASSERT_GT(normal, 0u); + node.ota_speed = 0.5f; + EXPECT_EQ(node.otaRelayDelay(&ota_packet), normal * 2); + EXPECT_EQ(node.getRetransmitDelay(&ordinary_packet), ordinary_delay); + node.ota_speed = 3.0f; + EXPECT_EQ(node.otaRelayDelay(&ota_packet), (normal + 2) / 3); + EXPECT_EQ(node.getRetransmitDelay(&ordinary_packet), ordinary_delay); +} + +TEST_F(DualProfileTest, OtaAirtimeFollowsParticipatingProfilesInsteadOfScannerVisit) { + radio.distinguish_airtime = true; // primary 900 ms, secondary 70 ms + radio.config.secondary_temporary = true; + radio.config.cross = mesh::RadioCrossMode::Off; + EXPECT_EQ(node.getOtaPacketAirtime(), 70u); + radio.selected = 1; + radio.estimated_airtime = 12345; + EXPECT_EQ(node.getOtaPacketAirtime(), 70u); + radio.config.reply_tx = mesh::RADIO_TX_BOTH; + EXPECT_EQ(node.getOtaPacketAirtime(), 970u); + radio.config.secondary_temporary = false; + radio.config.primary_temporary = true; + radio.config.secondary.mode = mesh::RadioProfileMode::Rx; + EXPECT_EQ(node.getOtaPacketAirtime(), 900u); + radio.config.reply_force = true; + EXPECT_EQ(node.getOtaPacketAirtime(), 970u); + radio.config.secondary.mode = mesh::RadioProfileMode::Off; + EXPECT_EQ(node.getOtaPacketAirtime(), 900u); +} + TEST_F(DualProfileTest, CrossFilterKeepsPrimaryTransmissionAndAvoidsCopyAllocation) { node.cross_filter_allows = false; ASSERT_NE(nullptr, queue()); diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index cffdafa97..cbc9913f4 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -4676,6 +4676,296 @@ TEST(OtaWarmStart, LeafDiffAuthenticatesAndFindsDifferingBlocks) { EXPECT_FALSE(miss[0]); EXPECT_FALSE(miss[2]); EXPECT_FALSE(miss[4]); } +TEST(OtaSpeed, BoundsScalingAndLongIntervals) { + EXPECT_TRUE(validSpeed(0.05f)); + EXPECT_TRUE(validSpeed(3.0f)); + EXPECT_FALSE(validSpeed(0.049f)); + EXPECT_FALSE(validSpeed(3.001f)); + EXPECT_FALSE(validSpeed(NAN)); + EXPECT_FALSE(validSpeed(INFINITY)); + for (uint32_t delay : {0u, 1u, 100u, 1000u, 86400000u}) { + EXPECT_EQ(scaleDelay(delay, 1.0f), delay); + } + EXPECT_EQ(scaleDelay(1000, 0.05f), 20000u); + EXPECT_EQ(scaleDelay(1000, 0.5f), 2000u); + EXPECT_EQ(scaleDelay(1000, 3.0f), 334u); + EXPECT_EQ(scaleDelay(UINT32_MAX, 0.05f), (uint32_t)INT32_MAX); + EXPECT_EQ(relayDelay(2000000, 0.05f), 0x00ffffffU); + EXPECT_EQ(relayDelay(100, 1.0f), 100u); + EXPECT_EQ(packetQuietTime(1000, 0.05f), 19000u); + EXPECT_EQ(packetQuietTime(1000, 1.0f), 0u); + EXPECT_EQ(packetQuietTime(1000, 3.0f), 0u); + + LongTimer timer; + uint32_t now = UINT32_MAX - 100; + timer.arm(now, 7u * 86400000u, 0.05f); + for (unsigned day = 1; day < 140; ++day) { + now += 86400000u; + EXPECT_FALSE(timer.ready(now)) << day; + } + now += 86400000u; + EXPECT_TRUE(timer.ready(now)); + timer.arm(now, 1000, 0.5f); + timer.rescale(now + 1000, 0.5f, 1.0f); + EXPECT_FALSE(timer.ready(now + 1499)); + EXPECT_TRUE(timer.ready(now + 1500)); +} + +TEST(OtaSpeed, ManifestPacingAndRetryAllowancesFollowFactor) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest)); + for (float speed : {0.05f, 0.5f, 1.0f, 3.0f}) { + OtaManager server; + CapturedMessages sent; + server.begin(0, capture_send, &sent); + ASSERT_TRUE(server.set_speed(speed)); + EXPECT_FALSE(server.set_speed(NAN)); + EXPECT_FLOAT_EQ(server.speed(), speed); + server.set_link_timing(80, 2000); + ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN)); + GetManifestMsg request{}; + memcpy(request.manifest_id, manifest.merkle_root, 4); + request.want_mask = 0xffff; + uint8_t wire[MAX_PACKET_PAYLOAD]; + const uint16_t len = encode_get_manifest(wire, sizeof wire, request); + server.set_clock(UINT32_MAX - 40); // egress crosses millis rollover + ASSERT_TRUE(server.on_message(wire, len)); + const uint32_t gap = speed == 0.05f ? 3200 : speed == 0.5f ? 320 : speed == 1.0f ? 160 : 54; + server.set_clock(UINT32_MAX - 40 + gap - 1); + server.serviceEgress(); + EXPECT_TRUE(sent.items.empty()); + server.set_clock(UINT32_MAX - 40 + gap); + server.serviceEgress(); + ASSERT_EQ(sent.items.size(), 1u); + server.set_clock(UINT32_MAX - 40 + 2 * gap); + server.serviceEgress(); + ASSERT_EQ(sent.items.size(), 2u); + EXPECT_EQ(server.pendingManifestJobs(), 0u); + const uint32_t timeout = server.fetchRetryTimeoutMs(); + ASSERT_TRUE(server.set_speed(1.0f)); + const uint32_t normal = server.fetchRetryTimeoutMs(); + EXPECT_GE(timeout, normal); + if (speed < 1.0f) EXPECT_GT(timeout, normal); + } +} + +TEST(OtaSpeed, LiveChangeRescalesQueuedMetadataWithoutDroppingIt) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest)); + OtaManager server; + CapturedMessages sent; + server.begin(0, capture_send, &sent); + server.set_link_timing(80, 2000); + ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN)); + GetManifestMsg request{}; + memcpy(request.manifest_id, manifest.merkle_root, 4); + request.want_mask = 0xffff; + uint8_t wire[MAX_PACKET_PAYLOAD]; + server.set_clock(UINT32_MAX - 159); // deadline is exactly zero + ASSERT_TRUE(server.on_message(wire, encode_get_manifest(wire, sizeof wire, request))); + server.set_clock(UINT32_MAX - 79); // 80 ms remain at 1x + ASSERT_TRUE(server.set_speed(0.5f)); + server.set_clock(79); // 159 ms later + server.serviceEgress(); + EXPECT_TRUE(sent.items.empty()); + server.set_clock(80); + server.serviceEgress(); + ASSERT_EQ(sent.items.size(), 1u); + server.set_clock(160); + ASSERT_TRUE(server.set_speed(3.0f)); // 240 ms of 0.5x waiting become 40 ms + server.set_clock(199); + server.serviceEgress(); + EXPECT_EQ(sent.items.size(), 1u); + server.set_clock(200); + server.serviceEgress(); + EXPECT_EQ(sent.items.size(), 2u); +} + +TEST(OtaSpeed, WholeTransferRecoversLostDataAtEverySpeed) { + struct Link { + OtaManager* destination; + std::vector> queue; + uint32_t next = 0; + bool drop_data = false; + static bool send(void* context, const uint8_t* bytes, uint16_t len, bool) { + auto& link = *static_cast(context); + if (link.queue.size() >= 2) return false; + link.queue.emplace_back(bytes, bytes + len); + return true; + } + void deliver(uint32_t now, float speed) { + if (queue.empty() || (int32_t)(now - next) < 0) return; + auto bytes = std::move(queue.front()); queue.erase(queue.begin()); + next = now + 100 + packetQuietTime(100, speed); + if (drop_data && ota_msg_type(bytes.data(), bytes.size()) == OTA_DATA) drop_data = false; + else destination->on_message(bytes.data(), bytes.size()); + } + }; + for (float speed : {0.05f, 0.5f, 1.0f, 3.0f}) { + SCOPED_TRACE(speed); + OtaManager server, client; + OtaStoreRam<4096> store; + Link to_client{&client, {}, 0, true}, to_server{&server}; + server.begin(0, Link::send, &to_client); + client.begin(SIM_TARGET_ID, Link::send, &to_server); + client.set_fetch_store(&store); + client.set_autofetch(OtaManager::AUTOFETCH_ANY); + for (auto* manager : {&server, &client}) { + ASSERT_TRUE(manager->set_speed(speed)); + manager->set_link_timing(100, 2000); + manager->set_max_hops(0); + } + ASSERT_TRUE(server.serve(SIM_MOTA_1K, SIM_MOTA_1K_LEN)); + server.announce(); + uint32_t next_loop = client.retryDelay(1000); + for (uint32_t now = 0; now < 600000 && client.fetchState() != OtaManager::COMPLETE; now += 10) { + server.set_clock(now); client.set_clock(now); + server.serviceEgress(); client.serviceEgress(); + to_client.deliver(now, speed); to_server.deliver(now, speed); + if (now >= next_loop) { + server.loop(); client.loop(); + next_loop = now + client.retryDelay(1000); + } + } + EXPECT_FALSE(to_client.drop_data); + ASSERT_EQ(client.fetchState(), OtaManager::COMPLETE); + ASSERT_EQ(store.staged_size(), SIM_MOTA_1K_LEN); + EXPECT_EQ(memcmp(store.data(), SIM_MOTA_1K, SIM_MOTA_1K_LEN), 0); + } +} + +TEST(OtaSpeed, DiscoveryAndCatalogRecoveryRespectSlowerTiming) { + OtaManager client; + CapturedMessages sent; + client.begin(SIM_TARGET_ID, capture_send, &sent); + client.set_archive_interest(true); + ASSERT_TRUE(client.set_speed(0.05f)); + AdvMsg adv{}; + adv.seeder_id[0] = 1; // deterministic jitter of 1 ms before scaling + adv.n_motas = 1; + uint8_t wire[MAX_PACKET_PAYLOAD]; + client.set_clock(100); + client.on_message(wire, encode_adv(wire, sizeof wire, adv)); + client.set_clock(6119); + client.loop(); + EXPECT_TRUE(sent.items.empty()); + client.set_clock(6120); + client.loop(); + ASSERT_EQ(sent.items.size(), 1u); + client.set_clock(6120 + 299999); + client.loop(); + EXPECT_EQ(sent.items.size(), 1u); + client.set_clock(6120 + 300000); + client.loop(); + EXPECT_EQ(sent.items.size(), 2u); +} + +TEST(OtaSpeed, CatalogRetryStaysArmedAtMillisRollover) { + for (float speed : {0.05f, 1.0f, 3.0f}) { + OtaManager client; + CapturedMessages sent; + client.begin(SIM_TARGET_ID, capture_send, &sent); + ASSERT_TRUE(client.set_speed(speed)); + 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)); + client.set_clock(0u - client.retryDelay(OTA_CATALOG_RETRY_MS)); + client.queryAll(); // retry deadline would be zero, the old disarmed sentinel + ASSERT_EQ(sent.items.size(), 1u); + client.set_clock(0); client.loop(); + EXPECT_EQ(sent.items.size(), 1u); + client.set_clock(1); client.loop(); + EXPECT_EQ(sent.items.size(), 2u); + } +} + +TEST(OtaSpeed, SlowRadioSettingDoesNotDelayLocalResumeVerification) { + OtaStoreRam<4096> store; + ASSERT_TRUE(store.begin(SIM_MOTA_LEN)); + ASSERT_TRUE(store.write(0, SIM_MOTA, SIM_MOTA_LEN)); + OtaManager manager; + manager.begin(SIM_TARGET_ID, nullptr, nullptr); + manager.set_fetch_store(&store); + manager.set_autofetch(OtaManager::AUTOFETCH_ANY); + ASSERT_TRUE(manager.set_speed(0.05f)); + EXPECT_EQ(manager.loopIntervalMs(1000), 20000u); + ASSERT_TRUE(manager.resumeStaged(nullptr)); + ASSERT_EQ(manager.fetchState(), OtaManager::VERIFYING_STAGED); + EXPECT_EQ(manager.loopIntervalMs(1000), 1000u); + manager.loop(); + EXPECT_EQ(manager.fetchState(), OtaManager::COMPLETE); +} + +TEST(OtaSpeed, ProactiveProofPreservesDrainTimeAndSurvivesRollover) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); + const unsigned fragments = (manifest.block_size() + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA; + for (uint32_t start : {0u, UINT32_MAX - 599}) { + OtaManager server; + CapturedMessages sent; + server.begin(0, capture_send, &sent); + server.set_link_timing(100, 2000); // three packet-service intervals = 600 ms + ASSERT_TRUE(server.serve(SIM_MOTA_1K, SIM_MOTA_1K_LEN)); + ReqMsg request{}; + memcpy(request.manifest_id, manifest.merkle_root, 4); + request.want_mask = 0xffff; + uint8_t wire[MAX_PACKET_PAYLOAD]; + server.set_clock(start); + ASSERT_TRUE(server.on_message(wire, encode_req(wire, sizeof wire, request))); + for (unsigned i = 0; i < fragments; ++i) server.serviceEgress(); + ASSERT_EQ(sent.items.size(), fragments); + server.set_clock(start + 300); + ASSERT_TRUE(server.set_speed(3)); // cannot shorten a physical drain already pending + server.set_clock(start + 599); server.serviceEgress(); + EXPECT_EQ(sent.items.size(), fragments); + server.set_clock(start + 601); server.serviceEgress(); + ASSERT_EQ(sent.items.size(), fragments + 1); + EXPECT_EQ(ota_msg_type(sent.items.back().data(), sent.items.back().size()), OTA_PROOF); + // Also preserve the drain when the response job starts at 3x. + sent.items.clear(); + server.set_clock(1000); + ASSERT_TRUE(server.on_message(wire, encode_req(wire, sizeof wire, request))); + for (unsigned i = 0; i < fragments; ++i) server.serviceEgress(); + server.set_clock(1599); server.serviceEgress(); + EXPECT_EQ(sent.items.size(), fragments); + server.set_clock(1600); server.serviceEgress(); + EXPECT_EQ(sent.items.size(), fragments + 1); + } +} + +TEST(OtaSpeed, MissingProofFallbackStaysArmedAtMillisRollover) { + 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); + ASSERT_TRUE(client.set_speed(0.05f)); + 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); + sent.items.clear(); + client.set_clock(0u - client.retryDelay(OTA_PROOF_GRACE_MS)); + uint8_t wire[MAX_PACKET_PAYLOAD]; + for (uint32_t offset = 0; offset < manifest.block_size(); offset += OTA_FRAG_DATA) { + DataMsg data{}; + memcpy(data.manifest_id, manifest.merkle_root, 4); + data.frag_off = offset; + data.data = manifest.payload + offset; + data.data_len = std::min(OTA_FRAG_DATA, manifest.block_size() - offset); + ASSERT_TRUE(client.on_message(wire, encode_data(wire, sizeof wire, data))); + } + client.set_clock(0); client.serviceEgress(); + EXPECT_TRUE(sent.items.empty()); + client.set_clock(1); client.serviceEgress(); + ASSERT_EQ(sent.items.size(), 1u); + EXPECT_EQ(ota_msg_type(sent.items.back().data(), sent.items.back().size()), OTA_REQ_PROOF); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/test/test_ota_speed.py b/test/test_ota_speed.py new file mode 100644 index 000000000..9bc36cf44 --- /dev/null +++ b/test/test_ota_speed.py @@ -0,0 +1,109 @@ +"""Exercise production OTA speed CLI/persistence without an OTA workspace.""" +from pathlib import Path +import shutil +import subprocess +import tempfile +import unittest + +ROOT = Path(__file__).resolve().parents[1] +HARNESS = r''' +#include +#include +#include +using namespace mesh::ota; +int main() { + MemoryFS fs; + char reply[160]; + auto command = [&](const char* text, const char* prefix) { + memset(reply, 0, sizeof reply); + assert(handleSpeedCommand(text, reply, sizeof reply)); + assert(!strncmp(reply, prefix, strlen(prefix))); + }; + beginSpeedConfig(&fs); + command("get ota.speed", "> ota.speed=1x"); + for (const auto* bad : {"", "0", "-1", "0.049", "3.01", "nan", "inf", "1e0", "1abc", "1 2"}) { + const std::string text = std::string("set ota.speed ") + bad; + command(text.c_str(), "ERR"); + assert(speedFactor() == 1.0f && fs.files.empty()); + } + command("set ota.speed .05", "OK ota.speed=0.05x (saved)"); + assert(validSpeed(speedFactor())); + beginSpeedConfig(&fs); + command("get ota.speed", "> ota.speed=0.05x"); + command("ota config speed 3", "OK ota.speed=3x (saved)"); + command("ota config speed", "> ota.speed=3x"); + command("ota cfg speed 0.5", "OK ota.speed=0.5x"); + command("ota speed", "> ota.speed=0.5x"); + command("get ota.speed 1", "ERR"); + assert(!handleSpeedCommand("get ota.speed.extra", reply, sizeof reply)); + assert(!handleSpeedCommand("ota config advert 10", reply, sizeof reply)); + const auto good = fs.files["/ota_speed"]; + fs.fail_write = true; + command("set ota.speed 2", "ERR"); + assert(speedFactor() == 0.5f && fs.files["/ota_speed"] == good); + fs.fail_write = false; + fs.fail_rename = 2; // cannot promote temp; restore primary from backup + command("set ota.speed 2", "ERR"); + assert(speedFactor() == 0.5f && fs.files["/ota_speed"] == good); + fs.files["/ota_speed.bak"] = good; + fs.files.erase("/ota_speed"); // power cut in rename gap + beginSpeedConfig(&fs); + assert(speedFactor() == 0.5f && fs.files["/ota_speed"] == good); + fs.files["/ota_speed.bak"] = good; + fs.files.erase("/ota_speed"); + fs.fail_rename = 1; // valid backup remains authoritative if repair fails + beginSpeedConfig(&fs); + assert(speedFactor() == 0.5f); + command("set ota.speed 2", "ERR"); + assert(fs.files["/ota_speed.bak"] == good); + beginSpeedConfig(&fs); + assert(speedFactor() == 0.5f && fs.files["/ota_speed"] == good); + fs.files["/ota_speed"][4] ^= 1; // damaged image is not partly adopted + beginSpeedConfig(&fs); + assert(speedFactor() == 1.0f); + command("set ota.speed 2", "ERR"); + assert(speedFactor() == 1.0f); + fs.files.clear(); + beginSpeedConfig(&fs); + command("ota set speed 1.125", "OK ota.speed=1.125x"); + beginSpeedConfig(&fs); + assert(speedFactor() == 1.125f); +} +''' + + +class OtaSpeedTest(unittest.TestCase): + def test_commands_range_storage_and_failure_recovery(self): + compiler = shutil.which("g++") or shutil.which("clang++") + self.assertIsNotNone(compiler) + with tempfile.TemporaryDirectory() as directory: + source = Path(directory) / "speed.cpp" + source.write_text(HARNESS) + binary = Path(directory) / "speed" + result = subprocess.run([ + compiler, "-std=c++17", "-Wall", "-Wextra", "-DENABLE_OTA=1", + "-I", str(ROOT / "test/fixtures/radio_profiles/mocks"), + "-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"), + str(ROOT / "src/helpers/ota/OtaSpeedConfig.cpp"), str(source), + "-o", str(binary), + ], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + subprocess.run([str(binary)], check=True) + + def test_shared_role_wiring_and_idle_context_access(self): + common = (ROOT / "src/helpers/CommonCLI.cpp").read_text() + companion = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() + for text in (common, companion): + self.assertIn("mesh::ota::beginSpeedConfig(", text) + self.assertIn("mesh::ota::handleSpeedCommand(command, reply,", text) + cli = (ROOT / "src/helpers/ota/OtaCli.cpp").read_text() + self.assertLess(cli.index("handleSpeedCommand(command"), cli.index("ota_acquire_context(reply")) + mesh = (ROOT / "src/Mesh.cpp").read_text() + self.assertIn("syncOtaTiming(context->manager)", mesh) + self.assertIn("syncOtaTiming(ota::ota_ctx().manager)", mesh) + self.assertIn("manager.set_speed(getOtaSpeedFactor())", mesh) + self.assertIn("_ota_announce_timer.arm(_ms->getMillis(), gap, ota_speed)", mesh) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_packet_manager/test_rx_reserve_packet_manager.cpp b/test/test_packet_manager/test_rx_reserve_packet_manager.cpp index 50dfa856c..df902a7c0 100644 --- a/test/test_packet_manager/test_rx_reserve_packet_manager.cpp +++ b/test/test_packet_manager/test_rx_reserve_packet_manager.cpp @@ -113,6 +113,8 @@ public: bool bypass_rx_delay = false; bool configured_cad_enabled = false; int configured_agc_interval = 0; + float ota_speed = 1.0f; + float getOtaSpeedFactor() const override { return ota_speed; } int getAGCResetInterval() const override { return configured_agc_interval; @@ -1170,6 +1172,59 @@ TEST(RxReservePacketManager, DispatcherNotifiesBeforeReleasingStaleOutbound) { EXPECT_EQ(0, radio.send_starts); } +TEST(RxReservePacketManager, PacingPausesExpiryWithoutErasingPriorCongestion) { + for (uint32_t start : {0u, UINT32_MAX - 20000}) { + RxReservePacketManager manager(8, 4); + auto* paced = manager.allocNew(); + auto* stale = manager.allocNew(); + ASSERT_TRUE(manager.queueOutbound(paced, 0, start)); + ASSERT_TRUE(manager.queueOutbound(stale, 1, start)); + // Already spent 29 seconds waiting. One minute of deliberate pacing must + // not expire this packet, or grant another full 30 seconds afterwards. + for (uint32_t elapsed = 29000; elapsed < 89000; elapsed += 100) { + ASSERT_EQ(manager.peekNextOutbound(start + elapsed), paced); + ASSERT_TRUE(manager.deferOutboundForPacing(paced, start + elapsed, start + elapsed + 100)); + } + EXPECT_EQ(manager.getNextDroppedOutbound(), stale); + manager.free(stale); + EXPECT_EQ(manager.peekNextOutbound(start + 90000), paced); + EXPECT_EQ(manager.peekNextOutbound(start + 90001), nullptr); + EXPECT_EQ(manager.getNextDroppedOutbound(), paced); + manager.free(paced); + EXPECT_EQ(manager.getFreeCount(), 8); + } +} + +TEST(RxReservePacketManager, DispatcherKeepsOtaQueuedAcrossSlowQuietWindow) { + RxReservePacketManager manager(8, 4); + TestClock clock; + TestRadio radio; + TestDispatcher dispatcher(radio, clock, manager); + dispatcher.begin(); + dispatcher.ota_speed = 0.05f; + for (unsigned i = 0; i < 2; ++i) { + auto* packet = dispatcher.obtainNewPacket(); + ASSERT_NE(packet, nullptr); + packet->header = ROUTE_TYPE_FLOOD | (PAYLOAD_TYPE_OTA << PH_TYPE_SHIFT); + packet->payload_len = 1; packet->payload[0] = i; + ASSERT_TRUE(dispatcher.sendPacket(packet, 0)); + } + clock.now = 1; dispatcher.loop(); + ASSERT_EQ(radio.send_starts, 1); + clock.now = 1801; radio.send_complete = true; dispatcher.loop(); + radio.send_complete = false; + for (clock.now = 1901; clock.now < 36001; clock.now += 100) { + dispatcher.loop(); + ASSERT_EQ(radio.send_starts, 1); + ASSERT_EQ(dispatcher.failed_packet, nullptr); + } + clock.now = 36001; dispatcher.loop(); // 34.2 seconds of intentional quiet + EXPECT_EQ(radio.send_starts, 2); + EXPECT_EQ(dispatcher.failed_packet, nullptr); + clock.now++; radio.send_complete = true; dispatcher.loop(); + EXPECT_EQ(manager.getFreeCount(), 8); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS();