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