From bb6089b313f6c24fed0a5afe272a3641786f72c9 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Wed, 30 Sep 2026 10:54:09 -0700 Subject: [PATCH] Fix relayed OTA retries and add guarded legacy recovery Use a bounded timed OTA forwarding cache while preserving echo, queue, TempRadio and forwarding-policy guards. Add opt-in manifest recovery for directly reachable legacy relays and interactive 125 kHz SF5/SF6/SF7 rehearsals, with regression tests and documentation. --- docs/lora_ota_automation.md | 44 +++ src/Mesh.cpp | 84 ++++- src/Mesh.h | 17 ++ .../test_simple_mesh_tables.cpp | 24 ++ test/test_trace_retry/test_trace_retry.cpp | 103 +++++++ tools/lora_ota/lora_ota.py | 287 +++++++++++++++++- tools/lora_ota/test_lora_ota.py | 198 ++++++++++++ tools/lora_ota/test_lora_ota_legacy_dedup.py | 163 ++++++++++ 8 files changed, 906 insertions(+), 14 deletions(-) create mode 100644 tools/lora_ota/test_lora_ota_legacy_dedup.py diff --git a/docs/lora_ota_automation.md b/docs/lora_ota_automation.md index f069f045..ed525f06 100644 --- a/docs/lora_ota_automation.md +++ b/docs/lora_ota_automation.md @@ -485,6 +485,19 @@ frequency, bandwidth, SF, and CR. The hard minimum is 20 minutes. The 120-minute default and the runner's worst-case timeout estimate are guidance. A window below the estimate prints a warning and continues; the radio lease can expire before a slow OTA finishes. +For an interactive run configured at 250 kHz, the runner offers a 125-kHz path +rehearsal at SF5 before the final update confirmation. If a 250-kHz rehearsal +cannot prove the temporary path but every short lease expires and the exact +normal path returns without active recovery, it offers the same 125-kHz SF5 +start. If the 125-kHz SF5 path fails in that same cleanly returned way, the +operator can approve a fresh SF6 rehearsal, then SF7 if SF6 fails. Each change +prints the revised plan and requires confirmation before its rehearsal; the +long transfer uses only the setting that passes. The selected long-window +duration is unchanged, and each attempt repeats the full short rehearsal. +`--yes` and non-interactive runs never change bandwidth or spreading factor +automatically. To start directly at 125 kHz SF5 without a prompt, pass +`--temp-radio 909.5,125,5,5,150` (choose a legal frequency and an appropriate +window for the actual route). The separate three-minute path rehearsal and its identity checks still run. Before that long window is allowed, the live runner performs a mandatory @@ -920,6 +933,37 @@ status for another manifest ID is an error, never permission to install it. ## Transmission loss and retries +### Legacy relay manifest workaround + +The deployed `2d03e098` firmware uses the ordinary 160-entry, non-expiring packet +hash FIFO for OTA forwarding. Identical manifest requests **and replies** can +therefore be suppressed after packet loss, even while CLI commands work. +The firmware fix replaces that forwarding decision with a small timed OTA +cache, suppresses downstream echoes and already-queued copies, and keeps +TempRadio, hop limits and forwarding filters authoritative. It applies to +discovery, manifest and block traffic, including opaque OTA relays. + +Until the relay is updated, add `--legacy-relay-dedup-workaround` to the same +runner command. This is opt-in because it adds airtime. It requires +`--source-shares-controller`, exactly one `--relay`, a saved **direct (zero-hop)** +route to that relay, and no extra `--ota-hops` beyond one. After an exact +`download: failed (manifest timeout) 0/0` for the requested MID, it sends up to +192 unique 13-byte **direct, non-forwarded** OTA no-operation/unknown-opcode +frames on the verified TempRadio channel. These replace legacy FIFO entries; +192 gives some loss headroom over 160 entries, not proof of reception. Then +it rechecks the destination and requests the same signed package once. A lost +pull reply is reconciled using status, not blindly replayed. + +There are at most two recovery rounds within the original transfer deadline. +Each round adds 2,496 origin Mesh bytes plus LoRa overhead, normally roughly +a minute at 125 kHz/SF7 before command/status latency. The companion's local +queue acceptance is not an RF acknowledgement. Recovery stops on insufficient +verified lease time, wrong radio settings, command errors, or changed session. +It never reboots a relay, cancels a partial image, changes package bytes, or +bypasses signature/install checks. It does **not** fix later block-retry +stalls on old firmware or relays beyond the controller's direct radio reach; +the firmware update is the permanent fix. + Read-only and replay-safe transmissions retry up to three times. Three retries or 90 seconds, whichever comes first, opens a 10-second stop-or-continue prompt when stdin is an interactive terminal. Continue remains the default on diff --git a/src/Mesh.cpp b/src/Mesh.cpp index 33560aaf..2bd13baf 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -235,6 +235,7 @@ uint8_t Mesh::getOtaHopLimit() const { return ota::ota_hop_limit(); } #endif void Mesh::begin() { + resetOtaForwardCache(); if (auto* limiter = getFloodAdvertLimiter()) limiter->reset(); _active_direct_retry_count = 0; _active_flood_retry_count = 0; @@ -315,6 +316,7 @@ void Mesh::loop() { const auto* p = _radio->profiles(); if (p && (_retry_radio_generations[0] != p->generation[0] || _retry_radio_generations[1] != p->generation[1] || _retry_cross_mode != p->cross)) { + resetOtaForwardCache(); // A changed session is not a failed radio link. Retire its ownership // without recording a failed final echo, and promptly return queued storage. for (int i = 0; i < TOTAL_DIRECT_RETRY_SLOTS; ++i) { @@ -347,6 +349,7 @@ void Mesh::loop() { _retry_radio_generations[1] = p->generation[1]; _retry_cross_mode = p->cross; } + if (_ota_forward_active && !isAnyTempRadioActive()) resetOtaForwardCache(); Dispatcher::loop(); serviceLoopMaintenance(); #if defined(ENABLE_OTA) && defined(ESP32_PLATFORM) && \ @@ -596,6 +599,67 @@ void Mesh::resetOtaRelayBackoff() { _ota_relay_backoff_level = 0; } +void Mesh::resetOtaForwardCache() { + memset(_ota_forwarded, 0, sizeof(_ota_forwarded)); + _ota_forward_next = 0; + _ota_forward_active = false; +} + +bool Mesh::allowOtaForwardRetry(const Packet* packet) { + // A deferred copy already owns the retry. Never fill the queue with more + // identical copies while CAD, airtime limits or relay timing delay it. + for (int i = 0; i < _mgr->getOutboundTotal(); ++i) { + const Packet* pending = _mgr->getOutboundByIdx(i); + if (pending && pending->getPayloadType() == PAYLOAD_TYPE_OTA + && pending->radio_profile == packet->radio_profile + && pending->payload_len == packet->payload_len + && memcmp(pending->payload, packet->payload, packet->payload_len) == 0) return false; + } + uint8_t hash[MAX_HASH_SIZE]; + packet->calculatePacketHash(hash); + const uint32_t now = _ms->getMillis(); + const uint32_t airtime = _radio->getProfileAirtime( + packet->radio_profile, packet->getRawLength(), packet->tx_cr); + const uint32_t quiet = airtime > 500 ? airtime * 2 : 1000; + for (const auto& entry : _ota_forwarded) { + if (!entry.valid || entry.profile != packet->radio_profile + || memcmp(entry.hash, hash, sizeof(hash)) != 0) continue; + // Age out route history as well, so a genuinely changed (longer) route + // cannot be pinned forever. Normal hop limits still bound old echoes. + const uint32_t age = now - entry.forwarded_at; + const uint32_t expiry = quiet > 7500 ? quiet * 4 : 30000; + if (age >= expiry) return true; + // A longer path within this window is a downstream echo. Suppressed + // echoes must not refresh the timer and starve the next genuine retry. + if (packet->getPathHashCount() > entry.min_hops) return false; + return age >= quiet; + } + return true; +} + +void Mesh::rememberOtaForward(const Packet* packet) { + _ota_forward_active = true; + uint8_t hash[MAX_HASH_SIZE]; + packet->calculatePacketHash(hash); + OtaForwardEntry* entry = nullptr; + for (auto& candidate : _ota_forwarded) { + if (candidate.valid && candidate.profile == packet->radio_profile + && memcmp(candidate.hash, hash, sizeof(hash)) == 0) { + entry = &candidate; + break; + } + } + if (!entry) { + entry = &_ota_forwarded[_ota_forward_next]; + _ota_forward_next = (_ota_forward_next + 1) % 16; + } + memcpy(entry->hash, hash, sizeof(hash)); + entry->forwarded_at = _ms->getMillis(); + entry->min_hops = packet->getPathHashCount(); + entry->profile = packet->radio_profile; + entry->valid = true; +} + void Mesh::observeOtaRequestPressure(const mesh::Packet* packet) { if (packet == NULL || packet->payload_len < 7) return; const uint8_t type = packet->payload[0]; @@ -887,6 +951,12 @@ uint8_t Mesh::getExtraAckTransmitCount() const { } void Mesh::onSendComplete(Packet* packet) { + // Profile selection/copying happens after sendOtaFlood(). Remember the + // actual transmitting profile as well, including locally injected OTA. + if (packet && packet->getPayloadType() == PAYLOAD_TYPE_OTA + && packet->isRouteFlood() && packet->getPathHashCount() == 0) { + rememberOtaForward(packet); + } watchForwardedAdvertEcho(packet); armDirectRetryOnSendComplete(packet); armFloodRetryOnSendComplete(packet); @@ -1315,10 +1385,10 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) { // relay it. 0 = only directly-received OTA. Runtime-tunable via `ota config hops`. if (n > getOtaHopLimit()) break; #endif - // ALWAYS process every accepted copy: OTA handlers are idempotent, and "eventually reliable" retries - // deliberately re-send IDENTICAL requests - if we gated processing on hasSeen(), the dedup would - // suppress those retries and the transfer could never recover from a lost reply. hasSeen() is used - // ONLY to avoid re-flooding the same packet more than once. + // ALWAYS process accepted copies locally. The ordinary seen FIFO has no + // expiry, so forwarding uses the bounded OTA retry cache below instead. + // Otherwise identical manifest/data requests AND replies can get stuck + // at a relay forever after just one lost transmission. bool seen = _tables->wasSeen(pkt); if (!seen) _tables->markSeen(pkt); const bool primary_ota = isPrimaryOtaTraffic(pkt->payload, pkt->payload_len); @@ -1339,13 +1409,14 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) { // Re-flood discovery at background priority, but keep an active transfer primary at every relay hop. // The free-pool reserve still sheds periodic discovery under pressure; requested transfer packets are // protected from that background-only gate. Ordinary hop and forwarding policy remain authoritative. - if (!terminal_ota && !seen && pkt->isRouteFlood() && !pkt->isMarkedDoNotRetransmit() + if (!terminal_ota && pkt->isRouteFlood() && !pkt->isMarkedDoNotRetransmit() #if defined(ENABLE_OTA) && n < getOtaHopLimit() #endif && (n + 1) * pkt->getPathHashSize() <= MAX_PATH_SIZE && (primary_ota || _mgr->getFreeCount() > OTA_FWD_MIN_FREE) - && allowPacketForward(pkt)) { + && allowPacketForward(pkt) && allowOtaForwardRetry(pkt)) { + rememberOtaForward(pkt); // record incoming hop count, before appending ourselves self_id.copyHashTo(&pkt->path[n * pkt->getPathHashSize()], pkt->getPathHashSize()); pkt->setPathHashCount(n + 1); action = ACTION_RETRANSMIT_DELAYED(ota_priority, getOtaRetransmitDelay(pkt)); @@ -2989,6 +3060,7 @@ bool Mesh::sendOtaFlood(Packet* packet, uint32_t delay_millis) { packet->header &= ~PH_ROUTE_MASK; packet->header |= ROUTE_TYPE_FLOOD; packet->setPathHashSizeAndCount(1, 0); + rememberOtaForward(packet); // downstream echoes of our own OTA must not be relayed _tables->markSeen(packet); // mark as sent, in case it floods back to us return sendPacket(packet, otaTrafficPriority(packet->payload, packet->payload_len), delay_millis); } diff --git a/src/Mesh.h b/src/Mesh.h index fb1891f6..2cf76f1b 100644 --- a/src/Mesh.h +++ b/src/Mesh.h @@ -145,6 +145,23 @@ class Mesh : public Dispatcher { bool valid; }; + // OTA retries deliberately reuse their payload/hash. Keep a small, timed + // forwarding cache instead of letting the ordinary FIFO suppress them for + // an entire transfer. 16 * 16 bytes; no per-packet allocation. + struct OtaForwardEntry { + uint8_t hash[MAX_HASH_SIZE]; + uint32_t forwarded_at; + uint8_t min_hops; + uint8_t profile; + bool valid; + }; + OtaForwardEntry _ota_forwarded[16] = {}; + uint8_t _ota_forward_next = 0; + bool _ota_forward_active = false; + bool allowOtaForwardRetry(const Packet* packet); + void rememberOtaForward(const Packet* packet); + void resetOtaForwardCache(); + RTCClock* _rtc; RNG* _rng; MeshTables* _tables; diff --git a/test/test_mesh_tables/test_simple_mesh_tables.cpp b/test/test_mesh_tables/test_simple_mesh_tables.cpp index 78976733..cd6aac0c 100644 --- a/test/test_mesh_tables/test_simple_mesh_tables.cpp +++ b/test/test_mesh_tables/test_simple_mesh_tables.cpp @@ -417,6 +417,30 @@ TEST(SimpleMeshTables, CooperativeRecentCursorTraverses2048Rows) { EXPECT_EQ(nullptr, t.getNextRecentRepeaterBySortKey(&cursor, cursor_index, cursor_index)); } +TEST(SimpleMeshTables, LegacyOtaDirectCacheTurnoverEvictsStaleHashWithoutFlooding) { + SimpleMeshTables table; + auto stale = makeFloodPacket(0x44); + stale.header = ROUTE_TYPE_FLOOD | (PAYLOAD_TYPE_OTA << PH_TYPE_SHIFT); + table.markSeen(&stale); + ASSERT_TRUE(table.wasSeen(&stale)); + int received = 0; + for (unsigned i = 0; i < 192; ++i) { + if (i % 6 == 0) continue; // 32 lost frames; remaining 160 replace the FIFO + const uint8_t raw[] = {0x32, 0, 0, 1, 2, 3, 4, 5, 6, 7, 8, + (uint8_t)i, (uint8_t)(i >> 8)}; + Packet packet; + ASSERT_TRUE(packet.readFrom(raw, sizeof(raw))); + ASSERT_TRUE(packet.isRouteDirect()); + ASSERT_EQ(PAYLOAD_TYPE_OTA, packet.getPayloadType()); + ASSERT_EQ(0, packet.getPathHashCount()); + ASSERT_EQ(0, packet.payload[0]); // unknown opcode; no manager state mutation + table.markSeen(&packet); + ++received; + } + EXPECT_EQ(MAX_PACKET_HASHES, received); + EXPECT_FALSE(table.wasSeen(&stale)); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/test/test_trace_retry/test_trace_retry.cpp b/test/test_trace_retry/test_trace_retry.cpp index c793da6f..be417475 100644 --- a/test/test_trace_retry/test_trace_retry.cpp +++ b/test/test_trace_retry/test_trace_retry.cpp @@ -528,6 +528,109 @@ TEST(RepeaterTransport, OtaTransferRelaysAsPrimaryTrafficWithoutOtaManager) { EXPECT_EQ(1, request.getPathHashCount()); } +class OtaRelayRetryTest : public testing::Test { + protected: + TraceTestClock clock; + TraceTestRTC rtc; + TraceTestRNG rng; + TraceTestRadio radio; + ForwardingTestTables tables; + StaticPoolPacketManager manager{12}; + TraceTestMesh node{radio, clock, rng, rtc, manager, tables}; + void SetUp() override { + node.begin(); + node.forwardFloods = true; + node.tempRadioActive = true; + } + mesh::DispatcherAction receive(const mesh::Packet& original) { + auto copy = original; // forwarding appends a hop to its owned packet + return node.receivePacket(©); + } +}; + +TEST_F(OtaRelayRetryTest, LostRequestsAndResponsesCanCrossRelayAgain) { + for (uint8_t type : {mesh::ota::OTA_GET_MANIFEST, mesh::ota::OTA_MANIFEST, + mesh::ota::OTA_REQ, mesh::ota::OTA_DATA, mesh::ota::OTA_PROOF, + mesh::ota::OTA_ADV, mesh::ota::OTA_QUERY, mesh::ota::OTA_HAVE}) { + auto packet = makeFloodPacket(PAYLOAD_TYPE_OTA); + packet.payload[0] = type; + ASSERT_NE(ACTION_RELEASE, receive(packet)); + // The general FIFO still reports seen, even after arbitrary elapsed time. + ASSERT_TRUE(tables.seen); + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + clock.now += 999; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + clock.now += 1; + EXPECT_NE(ACTION_RELEASE, receive(packet)); + } +} + +TEST_F(OtaRelayRetryTest, DownstreamEchoDoesNotRefreshOrReopenRetryWindow) { + auto packet = makeOtaManifestFragment(3); + ASSERT_NE(ACTION_RELEASE, receive(packet)); + auto echo = packet; + echo.setPathHashCount(1); + clock.now = 800; + EXPECT_EQ(ACTION_RELEASE, receive(echo)); + clock.now = 1000; + EXPECT_NE(ACTION_RELEASE, receive(packet)); + clock.now = 10000; + EXPECT_EQ(ACTION_RELEASE, receive(echo)); + clock.now = 31000; + EXPECT_NE(ACTION_RELEASE, receive(echo)); // expired route history permits a new longer route +} + +TEST_F(OtaRelayRetryTest, QueuedCopySuppressesRetriesUntilItLeavesQueue) { + auto packet = makeOtaManifestFragment(3); + ASSERT_NE(ACTION_RELEASE, receive(packet)); + auto* pending = node.obtainNewPacket(); + *pending = packet; + ASSERT_TRUE(manager.queueOutbound(pending, 0, 60000)); + clock.now = 10000; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + node.releasePacket(manager.removeOutboundByIdx(0)); + EXPECT_NE(ACTION_RELEASE, receive(packet)); +} + +TEST_F(OtaRelayRetryTest, TimerWrapAndRadioSessionResetDoNotStrandRetries) { + auto packet = makeOtaManifestFragment(3); + clock.now = 0xfffffe00u; + ASSERT_NE(ACTION_RELEASE, receive(packet)); + clock.now = 487; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + clock.now = 488; + EXPECT_NE(ACTION_RELEASE, receive(packet)); + node.tempRadioActive = false; + node.loop(); + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + node.tempRadioActive = true; + EXPECT_NE(ACTION_RELEASE, receive(packet)); +} + +TEST_F(OtaRelayRetryTest, RetryNeverBypassesForwardingOrReceiveFilter) { + auto packet = makeOtaManifestFragment(3); + ASSERT_NE(ACTION_RELEASE, receive(packet)); + clock.now = 2000; + node.forwardFloods = false; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + node.forwardFloods = true; + node.rejectFloods = true; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + node.rejectFloods = false; + EXPECT_NE(ACTION_RELEASE, receive(packet)); +} + +TEST_F(OtaRelayRetryTest, OriginEchoIsSuppressedOnActualTransmitProfile) { + auto packet = makeOtaManifestFragment(3); + packet.radio_profile = 1; + node.completePacketSend(&packet); + packet.setPathHashCount(1); + clock.now = 2000; + EXPECT_EQ(ACTION_RELEASE, receive(packet)); + packet.radio_profile = 0; + EXPECT_NE(ACTION_RELEASE, receive(packet)); // independent radio path +} + TEST(RepeaterTransport, OtaDiscoveryRelayKeepsCollisionJitter) { TraceTestClock clock; TraceTestRTC rtc; diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index 40bd6043..304c6753 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -146,6 +146,11 @@ RELAY_TIMING_COMMANDS_PER_RELAY = 12 RELAY_TIMING_RECOVERY_FILE = "relay-timing-settings.json" OTA_HOPS_RECOVERY_FILE = "ota-hop-settings.json" OTA_MAX_HOPS = 8 +# The deployed SimpleMeshTables FIFO holds 160 hashes. These short, direct +# (never forwarded) unknown-OTA frames provide 20% loss headroom. This is an +# explicit, best-effort legacy workaround, not an acknowledgement of reception. +LEGACY_OTA_CACHE_FRAMES = 192 +LEGACY_OTA_CACHE_ROUNDS = 2 TARGET_RXPS_RECOVERY_FILE = "target-rxps-settings.json" SOURCE_RXPS_RECOVERY_FILE = "source-rxps-settings.json" MIN_MESHCLI_VERSION = (1, 6, 0) @@ -190,6 +195,10 @@ class TransmissionError(OtaError): """A command may not have reached its destination or returned a reply.""" +class TempRadioPathUnproven(TransmissionError): + """The temporary path failed, but its short leases returned cleanly.""" + + class TransmissionStopped(OtaError): """The operator chose to stop an otherwise continuing retry loop.""" @@ -2116,6 +2125,19 @@ class Controller: ) raise OtaError("meshcli did not return the controller radio settings") + def send_legacy_ota_cache_probe(self, packet: bytes, timeout: float) -> None: + marker = f"OTA_CACHE_{secrets.token_hex(8)}" + result = self._execute( + ["send_raw", packet.hex(), "echo", marker], + "legacy OTA cache turnover", timeout=timeout, + ) + # meshcli 1.6.3 prints no JSON on send_raw success, only on failure. + # A marker proves the command completed; it is NOT an on-air ACK. + output = result.stdout + "\n" + result.stderr + if (result.returncode != 0 or marker not in result.stdout + or re.search(r"error|unknown command|unsupported|invalid", output, re.I)): + raise OtaError("legacy OTA cache probe was not accepted by the companion") + def get_public_key(self, timeout: float | None = None) -> str: objects = ( self._run(["infos"], "read controller identity") @@ -3042,6 +3064,82 @@ def warn_short_temp_radio_window( ) +def prompt_for_125khz_temp_radio( + args: argparse.Namespace, *, after_failed_rehearsal: bool = False, +) -> bool: + """Offer the interactive 125-kHz SF5-to-SF7 path rehearsal ladder.""" + if ( + args.temp_values[1] != 250 + or args.prepare_only + or args.yes + or not sys.stdin.isatty() + ): + return False + if after_failed_rehearsal: + question = ( + "The 250 kHz rehearsal could not prove the temporary path, but " + "all short leases expired and the normal path was verified. " + "Try 125 kHz at SF5, then SF6 and SF7 if needed, with a fresh " + "three-minute rehearsal and confirmation at each step? [y/N] " + ) + else: + question = ( + "Try TempRadio at 125 kHz SF5, then SF6 and SF7 if needed " + "(confirming each setting before rehearsal and transfer)? [y/N] " + ) + try: + approved = input(question).strip().lower() in ("y", "yes") + except EOFError: + return False + if not approved: + return False + freq, _bandwidth, _sf, cr, minutes = args.temp_values + args.temp_values = (freq, 125.0, 5, cr, minutes) + args.temp_radio = ( + f"{format_decimal(freq)},125,5,{cr},{minutes}" + ) + print( + f"[preflight] selected {args.temp_radio}; the transfer window remains " + f"{minutes} minutes" + ) + warn_short_temp_radio_window(args, include_install=not args.no_install) + return True + + +def prompt_for_next_125khz_sf(args: argparse.Namespace) -> bool: + """Advance one SF only after a cleanly returned failed path rehearsal.""" + freq, bandwidth, sf, cr, minutes = args.temp_values + if ( + bandwidth != 125 + or sf not in (5, 6) + or args.prepare_only + or args.yes + or not sys.stdin.isatty() + ): + return False + next_sf = sf + 1 + try: + approved = input( + f"125 kHz SF{sf} could not prove the temporary path. All short " + f"leases expired and the normal path was verified. Try a fresh " + f"three-minute rehearsal at SF{next_sf}? [y/N] " + ).strip().lower() in ("y", "yes") + except EOFError: + return False + if not approved: + return False + args.temp_values = (freq, bandwidth, next_sf, cr, minutes) + args.temp_radio = ( + f"{format_decimal(freq)},125,{next_sf},{cr},{minutes}" + ) + print( + f"[preflight] selected {args.temp_radio}; the transfer window remains " + f"{minutes} minutes" + ) + warn_short_temp_radio_window(args, include_install=not args.no_install) + return True + + def parse_source_terminal_banner(value: str) -> tuple[str, str]: """Read one Full/Companion welcome identity, not an advert or cached name.""" text = value.replace("\r", "") @@ -4662,6 +4760,8 @@ def confirm_update( print(f" boot target : {package.target_id:08X}; stage only, explicit install required") print(" boot safety : destination rechecks trust, continuity and upgrade-only policy at manual install") print(f" TempRadio : {args.temp_radio}") + if getattr(args, "legacy_relay_dedup_workaround", False): + print(" legacy retry: up to 2 x 192 short, non-forwarded TempRadio frames on manifest timeout") saved_rxps = getattr(args, "target_rxps_saved", None) rxps_profile = getattr(args, "target_rxps_profile", None) if isinstance(saved_rxps, RxpsSettings) and saved_rxps.enabled: @@ -5004,6 +5104,90 @@ def find_and_start_pull( ) +def recover_legacy_manifest_timeout( + controller: Controller, + args: argparse.Namespace, + package: MotaInfo, + seeder: SeederProcess | None, + status: str, + deadline: float, + attempt: int, +) -> bool: + """Bounded, opt-in turnover of a directly reachable legacy relay's FIFO. + + Only restart this exact MID after a confirmed manifest timeout with no + blocks. Never clear a partial/ready image, reboot a node, change a package, + or infer that an unacknowledged LoRa packet actually arrived. + """ + if (not getattr(args, "legacy_relay_dedup_workaround", False) + or attempt >= LEGACY_OTA_CACHE_ROUNDS + or download_manifest_id(status) != package.manifest_id + or not re.search(r"download: failed \(manifest timeout\) 0/0\b", status, re.I)): + return False + relays = getattr(args, "relay_values", []) + if len(relays) != 1 or not getattr(args, "source_shares_controller", False): + raise OtaError("legacy cache workaround requires one relay and a shared source/controller") + relay, _password = relays[0] + contacts = controller._run(["contact_info", relay], "check direct legacy relay route", timeout=5) + matching = [c for c in contacts if contact_matches_selector(c, relay)] + if (len(matching) != 1 or matching[0].get("out_path_len") != 0 + or not re.fullmatch(r"[0-9a-fA-F]{64}", relay)): + raise OtaError("legacy cache workaround requires the relay's full key and a direct (0-hop) contact route") + expected = RadioSettings(*args.temp_values[:4], False) + frame_prefix = bytes((0x32, 0, 0)) # OTA (0x0c), DIRECT, no path, unknown opcode 0 + nonce = secrets.token_bytes(8) + spacing = max(0.25, 4 * lora_airtime_seconds( + len(frame_prefix) + len(nonce) + 2, expected.bandwidth, + expected.spreading_factor, expected.coding_rate, preamble_symbols=32, + )) + print(f"[legacy] manifest timeout: bounded cache turnover {attempt + 1}/{LEGACY_OTA_CACHE_ROUNDS}; " + f"{LEGACY_OTA_CACHE_FRAMES} short direct frames, no relay reboot or package changes") + for index in range(LEGACY_OTA_CACHE_FRAMES): + ensure_seeder_running(seeder, "during legacy manifest recovery") + if index % 16 == 0: + # Use the existing binary controller connection, not the serial + # source that motatool owns. Fail closed on missing lease telemetry. + objects = controller._run(["cli", "tempradio"], "check legacy recovery TempRadio", timeout=5) + replies = [o["text"] for o in objects if isinstance(o.get("text"), str)] + if len(replies) != 1: + raise OtaError("cannot prove local TempRadio lease for legacy recovery") + state = parse_source_temp_radio_status(replies[0]) + remaining = re.search(r"\b(\d+)s left\b", replies[0]) + needed = 16 * (5 + spacing) + 10 + if (state is None or state[0] != "active" or not expected.matches(state[1]) + or remaining is None or int(remaining[1]) < needed + or deadline - time.monotonic() < needed): + raise OtaError("insufficient verified TempRadio/transfer time for legacy recovery") + controller.send_legacy_ota_cache_probe( + frame_prefix + nonce + struct.pack("= deadline: + raise OtaError("transfer deadline expired during legacy recovery; no pull resent") + try: + reply = controller.remote_command(args.target, + f"ota pull {package.manifest_id} flash", retry=False) + except TransmissionError: + reply = remote_command_with_seeder(controller, args.target, "ota status", seeder, + "resolving legacy recovery pull") + require_package_session(reply, package) + else: + if not re.match(rf"OK (?:pulling|resuming) mid={package.manifest_id}\b", reply, re.I): + raise OtaError(f"legacy recovery pull was not confirmed: {reply}") + print("[legacy] retried the same manifest; normal verification and transfer deadline remain in force") + return True + + def monitor_download( controller: Controller, args: argparse.Namespace, @@ -5011,6 +5195,7 @@ def monitor_download( seeder: SeederProcess | None = None, ) -> str: deadline = time.monotonic() + args.transfer_timeout_minutes * 60 + legacy_recoveries = 0 poll_seconds = float(args.poll_seconds) # A status command is ordinary half-duplex LoRa traffic. Asking immediately # after `ota pull` can occupy the link for a full reply timeout and was @@ -5065,6 +5250,12 @@ def monitor_download( controller, args.target, "ota status", seeder, "during transfer", ) + if recover_legacy_manifest_timeout( + controller, args, package, seeder, status, deadline, legacy_recoveries + ): + legacy_recoveries += 1 + last_progress = time.monotonic() + continue require_package_session(status, package) if "ready to install" in status.lower(): return status @@ -5088,6 +5279,12 @@ def monitor_download( controller, args.target, "ota status", seeder, "during transfer" ) query_seconds = time.monotonic() - query_started + if recover_legacy_manifest_timeout( + controller, args, package, seeder, status, deadline, legacy_recoveries + ): + legacy_recoveries += 1 + time.sleep(min(first_wait, max(0.0, deadline - time.monotonic()))) + continue require_package_session(status, package) lowered = status.lower() if "ready to install" in lowered: @@ -5855,6 +6052,7 @@ def run_temp_radio_preflight( controller_may_need_restore = False natural_expiry_waited = False normal_proven = False + normal_return_needed_recovery = False earliest_lease_expiry: float | None = None latest_lease_expiry: float | None = None primary_error: BaseException | None = None @@ -6513,7 +6711,12 @@ def run_temp_radio_preflight( ) prove_controller_identity("on TempRadio", deadline=proof_deadline) prove_source_identity("on TempRadio", deadline=proof_deadline) - prove_remote_identities("on TempRadio", deadline=proof_deadline) + try: + prove_remote_identities("on TempRadio", deadline=proof_deadline) + except TransmissionError as exc: + # The caller may offer a narrower bandwidth only after this + # rehearsal's finally block owns expiry and proves normal return. + raise TempRadioPathUnproven(str(exc)) from exc temp_proof_deadline("temporary path proof completion") if not shared_controller: @@ -6587,6 +6790,7 @@ def run_temp_radio_preflight( except (OtaError, OSError) as exc: first_normal_error = exc if first_normal_error is not None: + normal_return_needed_recovery = True recovery_error: BaseException | None = None try: recover_participants_on_temp() @@ -6633,6 +6837,13 @@ def run_temp_radio_preflight( f"three-minute TempRadio rehearsal failed: {primary_error}; " "cleanup also failed: " + "; ".join(cleanup_errors) ) from primary_error + if isinstance(primary_error, TempRadioPathUnproven) and ( + not normal_proven or normal_return_needed_recovery + ): + raise OtaError( + "temporary path proof failed and normal return required " + "recovery; no in-run bandwidth retry is safe" + ) from primary_error raise primary_error if cleanup_errors: raise OtaError( @@ -7395,6 +7606,12 @@ def build_parser() -> argparse.ArgumentParser: help=("minimum OTA reach for undeclared additional hops (0..8); otherwise " "infer from --relay count; only raise lower limits and restore afterward"), ) + parser.add_argument( + "--legacy-relay-dedup-workaround", action="store_true", + help=("on exact 0/0 manifest timeout, try up to two 192-frame direct cache turnovers; " + "requires one directly reachable relay and --source-shares-controller; " + "adds airtime, never reboots, does not fix legacy block retries"), + ) parser.add_argument( "--source-contact", metavar="NAME_OR_KEY", @@ -7588,6 +7805,12 @@ def validate_args(args: argparse.Namespace, parser: argparse.ArgumentParser) -> parser.error(f"at most {OTA_MAX_HOPS} --relay nodes are supported") if args.ota_hops is not None and not len(args.relay) <= args.ota_hops <= OTA_MAX_HOPS: parser.error("--ota-hops must be 0..8 and at least the declared --relay count") + if args.legacy_relay_dedup_workaround and ( + len(args.relay) != 1 or not args.source_shares_controller + or args.ota_hops not in (None, 1) + ): + parser.error("--legacy-relay-dedup-workaround requires one --relay, " + "--source-shares-controller, and no additional OTA hops") clear_manifest = args.clear_completed_manifest clear_body_hash = args.clear_completed_on_body_hash if bool(clear_manifest) != bool(clear_body_hash): @@ -8664,6 +8887,23 @@ def main( or not 1 <= target_rxps_saved.level <= 10 ): target_rxps_profile = None + # Let an interactive operator choose the narrower tuple before the + # update plan and any TempRadio lease are committed. + if prompt_for_125khz_temp_radio(args): + temp_command = f"tempradio {args.temp_radio}" + target_rxps_profile = select_rxps_temp_profile( + target.current_version, + args.temp_values, + all_participants_support_adaptive_preamble=( + all_participants_support_adaptive_preamble + ), + ) + if ( + target_rxps_saved is None + or target_rxps_saved.level is None + or not 1 <= target_rxps_saved.level <= 10 + ): + target_rxps_profile = None args.target_rxps_saved = target_rxps_saved args.target_rxps_profile = target_rxps_profile args.ota_hop_settings = prepare_ota_hop_settings(controller, args) @@ -8678,13 +8918,44 @@ def main( # long TempRadio window, relay timing, seeding, pull, or install can be # mutated. The rehearsal owns only independent three-minute leases and # proves their natural return before handing control back to this run. - run_temp_radio_preflight( - controller, - args, - target, - original_radio, - temp_radio, - ) + while True: + try: + run_temp_radio_preflight( + controller, args, target, original_radio, temp_radio, + ) + break + except TempRadioPathUnproven: + # The typed failure is emitted only after short-lease expiry + # and an unrecovered exact normal-path proof. Never reuse a + # live lease or silently change the approved radio plan. + if args.temp_values[1] == 250: + retry = prompt_for_125khz_temp_radio( + args, after_failed_rehearsal=True, + ) + else: + retry = prompt_for_next_125khz_sf(args) + if not retry: + raise + temp_command = f"tempradio {args.temp_radio}" + target_rxps_profile = select_rxps_temp_profile( + target.current_version, + args.temp_values, + all_participants_support_adaptive_preamble=( + all_participants_support_adaptive_preamble + ), + ) + if ( + target_rxps_saved is None + or target_rxps_saved.level is None + or not 1 <= target_rxps_saved.level <= 10 + ): + target_rxps_profile = None + args.target_rxps_profile = target_rxps_profile + freq, bandwidth, sf, cr, _minutes = args.temp_values + temp_radio = RadioSettings( + freq, bandwidth, sf, cr, original_radio.repeat + ) + confirm_update(args, target, package) apply_ota_hop_settings( controller, args, args.ota_hop_settings, owned_ota_hop_settings, work_dir, ) diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index c8106632..77cccfb5 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -3673,6 +3673,12 @@ class TempRadioPreflightTests(unittest.TestCase): return "OK - normal radio restore scheduled" if command == "get public.key": key = TempRadioPreflightTests.NODE_KEYS[name] + if ( + self.failure == "temp_remote_unreachable" + and on_temp + and name == "remote" + ): + raise ota.TransmissionError("no matching CLI reply") if self.failure == "temp_remote_identity" and on_temp and name == "far": return "> " + "44" * 32 if self.failure == "normal_remote_identity" and not on_temp and self.scheduled_temp and name == "remote": @@ -4542,6 +4548,98 @@ class TempRadioPreflightTests(unittest.TestCase): pull.assert_not_called() install.assert_not_called() + def test_temporary_path_timeout_allows_retry_only_after_clean_return(self) -> None: + controller, _calls, clock, source_state, error = self.run_rehearsal( + failure="temp_remote_unreachable", capture_error=True, + ) + self.assertIsInstance(error, ota.TempRadioPathUnproven) + self.assertGreaterEqual( + clock.now, + ota.TEMP_RADIO_PREFLIGHT_MINUTES * 60 + + ota.TEMP_RADIO_PREFLIGHT_MARGIN_SECONDS, + ) + self.assertFalse(bool(source_state["active"])) + self.assertTrue(controller.radio.matches(self.NORMAL)) + self.assertTrue(controller.all_nodes_normal()) + + def test_operator_reconfirms_each_125khz_sf_before_rehearsal(self) -> None: + image = firmware(b"bandwidth retry" * 300, VERSION_NEW) + package = ota.parse_mota(mota_blob(image)) + controller = mock.Mock() + controller.get_radio.return_value = self.NORMAL + controller.get_clock.return_value = int(ota.time.time()) + 1 + saved = ota.RxpsSettings(False, 18205, 20423, 8, 16) + with tempfile.TemporaryDirectory() as directory: + argv = [ + "release.mota", "remote", + "--controller-serial", "/dev/controller", + "--source-serial", "/dev/source", + "--password", "secret", + "--work-dir", str(Path(directory) / "work"), + "--temp-radio", "909.5,250,7,5,120", + ] + with contextlib.ExitStack() as stack: + stack.enter_context(mock.patch.object(ota, "preflight_inputs")) + stack.enter_context(mock.patch.object(ota, "bind_contact_selectors")) + stack.enter_context(mock.patch.object(ota, "preflight_source_cli")) + stack.enter_context(mock.patch.object( + ota, "ensure_source_clock_gate_safe", + return_value=(1_800_000_000, 1_800_000_059), + )) + stack.enter_context(mock.patch.object(ota, "read_source_rxps", return_value=saved)) + stack.enter_context(mock.patch.object(ota, "query_target", return_value=target())) + stack.enter_context(mock.patch.object( + ota, "prepare_package", + return_value=(Path("release.mota"), package, None), + )) + stack.enter_context(mock.patch.object( + ota, "read_lora_ota_participant_versions", + return_value={"destination": VERSION_NEW}, + )) + stack.enter_context(mock.patch.object( + ota, "read_remote_rxps", + return_value=ota.RxpsSettings(False, 18205, 20423, 8, 16), + )) + confirm = stack.enter_context(mock.patch.object(ota, "confirm_update")) + rehearsal = stack.enter_context(mock.patch.object( + ota, "run_temp_radio_preflight", + side_effect=[ + ota.TempRadioPathUnproven("temporary path unavailable"), + ota.TempRadioPathUnproven("SF5 path unavailable"), + ota.TempRadioPathUnproven("SF6 path unavailable"), + None, + ], + )) + disable_source = stack.enter_context(mock.patch.object(ota, "disable_source_rxps")) + arm_target = stack.enter_context(mock.patch.object( + ota, "arm_target_temp_radio", + side_effect=ota.OtaError("stop before long lease"), + )) + stack.enter_context(mock.patch.object( + ota, "switch_controller_to_temp_radio", + side_effect=ota.OtaError("mock cleanup cannot switch"), + )) + stack.enter_context(mock.patch.object( + ota.sys, "stdin", mock.Mock(isatty=lambda: True), + )) + prompts = stack.enter_context(mock.patch( + "builtins.input", side_effect=["n", "y", "y", "y"], + )) + stack.enter_context(contextlib.redirect_stdout(io.StringIO())) + stack.enter_context(contextlib.redirect_stderr(io.StringIO())) + result = ota.main(argv, controller_override=controller) + + self.assertEqual(result, 2) + self.assertEqual(prompts.call_count, 4) + self.assertEqual(confirm.call_count, 4) + self.assertEqual( + [(call.args[-1].bandwidth, call.args[-1].spreading_factor) + for call in rehearsal.call_args_list], + [(250.0, 7), (125.0, 5), (125.0, 6), (125.0, 7)], + ) + self.assertEqual(arm_target.call_args.args[2], "tempradio 909.5,125,7,5,120") + disable_source.assert_called_once() + def test_schedule_appearing_during_baseline_blocks_every_mutation( self, ) -> None: @@ -4688,6 +4786,106 @@ class TempRadioPreflightTests(unittest.TestCase): ) +class TempRadioBandwidthPromptTests(unittest.TestCase): + @staticmethod + def args() -> argparse.Namespace: + return argparse.Namespace( + temp_values=(909.5, 250.0, 7, 5, 120), + temp_radio="909.5,250,7,5,120", + prepare_only=False, + yes=False, + no_install=True, + ) + + def test_operator_can_select_125_before_rehearsal(self) -> None: + args = self.args() + with ( + mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)), + mock.patch("builtins.input", return_value="y") as prompt, + mock.patch.object(ota, "warn_short_temp_radio_window") as warning, + ): + self.assertTrue(ota.prompt_for_125khz_temp_radio(args)) + self.assertEqual(args.temp_values, (909.5, 125.0, 5, 5, 120)) + self.assertEqual(args.temp_radio, "909.5,125,5,5,120") + self.assertIn("SF5, then SF6 and SF7", prompt.call_args.args[0]) + warning.assert_called_once_with(args, include_install=False) + + def test_failed_rehearsal_prompt_requires_fresh_explicit_choice(self) -> None: + args = self.args() + with ( + mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)), + mock.patch("builtins.input", return_value="yes") as prompt, + mock.patch.object(ota, "warn_short_temp_radio_window"), + ): + self.assertTrue(ota.prompt_for_125khz_temp_radio( + args, after_failed_rehearsal=True, + )) + self.assertIn("all short leases expired", prompt.call_args.args[0]) + self.assertEqual(args.temp_values[1:3], (125.0, 5)) + self.assertFalse(ota.prompt_for_125khz_temp_radio(args)) + + def test_125khz_ladder_stops_after_sf7(self) -> None: + args = self.args() + args.temp_values = (909.5, 125.0, 5, 5, 120) + args.temp_radio = "909.5,125,5,5,120" + with ( + mock.patch.object(ota.sys, "stdin", mock.Mock(isatty=lambda: True)), + mock.patch("builtins.input", side_effect=["y", "yes"]) as prompt, + mock.patch.object(ota, "warn_short_temp_radio_window") as warning, + ): + self.assertTrue(ota.prompt_for_next_125khz_sf(args)) + self.assertEqual(args.temp_values[2], 6) + self.assertEqual(args.temp_radio, "909.5,125,6,5,120") + self.assertTrue(ota.prompt_for_next_125khz_sf(args)) + self.assertEqual(args.temp_values[2], 7) + self.assertEqual(args.temp_radio, "909.5,125,7,5,120") + self.assertFalse(ota.prompt_for_next_125khz_sf(args)) + self.assertEqual(prompt.call_count, 2) + self.assertEqual(warning.call_count, 2) + + def test_125khz_ladder_requires_interactive_approval(self) -> None: + for is_tty, yes, answer in ( + (False, False, "y"), + (True, True, "y"), + (True, False, "n"), + ): + with self.subTest(is_tty=is_tty, yes=yes, answer=answer): + args = self.args() + args.temp_values = (909.5, 125.0, 5, 5, 120) + args.temp_radio = "909.5,125,5,5,120" + args.yes = yes + with ( + mock.patch.object( + ota.sys, "stdin", mock.Mock(isatty=lambda: is_tty), + ), + mock.patch("builtins.input", return_value=answer) as prompt, + ): + self.assertFalse(ota.prompt_for_next_125khz_sf(args)) + self.assertEqual(args.temp_values[2], 5) + if not is_tty or yes: + prompt.assert_not_called() + + def test_noninteractive_or_declined_offer_keeps_original_tuple(self) -> None: + for is_tty, yes, answer in ( + (False, False, "y"), + (True, True, "y"), + (True, False, "n"), + ): + with self.subTest(is_tty=is_tty, yes=yes, answer=answer): + args = self.args() + args.yes = yes + with ( + mock.patch.object( + ota.sys, "stdin", mock.Mock(isatty=lambda: is_tty), + ), + mock.patch("builtins.input", return_value=answer) as prompt, + ): + self.assertFalse(ota.prompt_for_125khz_temp_radio(args)) + self.assertEqual(args.temp_values[1], 250.0) + if not is_tty or yes: + prompt.assert_not_called() + + class ReliabilityTests(unittest.TestCase): def test_controller_rejects_meshcli_zero_exit_no_response(self) -> None: """Do not trust meshcore-cli's exit status without protocol output.""" diff --git a/tools/lora_ota/test_lora_ota_legacy_dedup.py b/tools/lora_ota/test_lora_ota_legacy_dedup.py new file mode 100644 index 00000000..9a631d46 --- /dev/null +++ b/tools/lora_ota/test_lora_ota_legacy_dedup.py @@ -0,0 +1,163 @@ +"""Offline coverage for the bounded, explicit legacy relay workaround.""" +import argparse +import contextlib +import io +import subprocess +import unittest +from unittest import mock + +import lora_ota as ota +from test_lora_ota import firmware, mota_blob, VERSION_NEW + + +class LegacyManifestRecoveryTests(unittest.TestCase): + def setUp(self): + self.package = ota.parse_mota(mota_blob(firmware(b"test" * 100, VERSION_NEW))) + self.relay = "ab" * 32 + self.args = argparse.Namespace( + legacy_relay_dedup_workaround=True, source_shares_controller=True, + relay_values=[(self.relay, "secret")], target="cd" * 32, + temp_values=(909.5, 125, 7, 5, 120), + ) + self.status = ("OTA | download: failed (manifest timeout) 0/0 " + f"id={self.package.manifest_id}") + self.controller = mock.Mock() + self.contact = {"public_key": self.relay, "out_path_len": 0} + self.lease = "TempRadio active: 909.500,125.00,7,5 6000s left,preamble=32" + self.controller._run.side_effect = lambda cmd, *a, **kw: ( + [self.contact] if cmd[0] == "contact_info" else [{"text": self.lease}] + ) + self.controller.remote_command.side_effect = [ + self.status, f"OK pulling mid={self.package.manifest_id} -> flash (primary traffic)" + ] + self.addCleanup(mock.patch.stopall) + mock.patch.object(ota.time, "sleep").start() + mock.patch.object(ota.time, "monotonic", return_value=100).start() + self.output = contextlib.redirect_stdout(io.StringIO()) + self.output.__enter__() + self.addCleanup(self.output.__exit__, None, None, None) + + def recover(self, status=None, attempt=0, deadline=1000): + return ota.recover_legacy_manifest_timeout( + self.controller, self.args, self.package, None, + self.status if status is None else status, deadline, attempt, + ) + + def test_unique_small_direct_frames_and_single_exact_pull(self): + self.assertTrue(self.recover()) + frames = [c.args[0] for c in self.controller.send_legacy_ota_cache_probe.call_args_list] + self.assertEqual(len(frames), 192) + self.assertEqual(len(set(frames)), 192) + self.assertEqual({len(f) for f in frames}, {13}) + self.assertTrue(all(f[:3] == bytes.fromhex("320000") for f in frames)) + self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list], + ["ota status", f"ota pull {self.package.manifest_id} flash"]) + self.assertFalse(self.controller.remote_command.call_args_list[-1].kwargs["retry"]) + self.assertEqual(self.controller._run.call_count, 13) # route + 12 lease proofs + + def test_off_by_default_and_two_round_limit(self): + self.args.legacy_relay_dedup_workaround = False + self.assertFalse(self.recover()) + self.args.legacy_relay_dedup_workaround = True + self.assertFalse(self.recover(attempt=2)) + self.controller._run.assert_not_called() + + def test_wrong_mid_partial_ready_and_other_errors_never_trigger(self): + for status in ( + self.status.replace(self.package.manifest_id, "FFFFFFFF"), + self.status.replace("0/0", "1/40"), + self.status.replace("manifest timeout", "signature invalid"), + f"download: fetching 1/40 id={self.package.manifest_id}", + f"download: ready to install 40/40 id={self.package.manifest_id}", + ): + with self.subTest(status=status): + self.assertFalse(self.recover(status)) + self.controller.send_legacy_ota_cache_probe.assert_not_called() + + def test_not_direct_and_unproven_temp_lease_fail_before_transmit(self): + for path in (-1, 1, None): + self.contact["out_path_len"] = path + with self.assertRaisesRegex(ota.OtaError, "direct"): + self.recover() + self.contact["out_path_len"] = 0 + for lease in ("TempRadio inactive", self.lease.replace("6000s", "30s"), + self.lease.replace("125.00", "250.00"), + self.lease.replace(" 6000s left", "")): + self.lease = lease + with self.assertRaises(ota.OtaError): + self.recover() + self.controller.send_legacy_ota_cache_probe.assert_not_called() + + def test_global_deadline_is_not_extended(self): + with self.assertRaisesRegex(ota.OtaError, "insufficient"): + self.recover(deadline=101) + self.controller.send_legacy_ota_cache_probe.assert_not_called() + + def test_session_change_never_cancels_or_pulls(self): + self.controller.remote_command.side_effect = ["no download"] + with self.assertRaisesRegex(ota.OtaError, "session changed"): + self.recover() + self.assertEqual(self.controller.remote_command.call_count, 1) + + def test_completion_during_turnover_is_preserved(self): + self.controller.remote_command.side_effect = [ + f"download: ready to install 40/40 id={self.package.manifest_id}" + ] + self.assertTrue(self.recover()) + self.assertEqual(self.controller.remote_command.call_count, 1) + + def test_lost_pull_ack_reconciles_without_resending(self): + self.controller.remote_command.side_effect = [self.status, ota.TransmissionError("lost"), + f"download: fetching 0/40 id={self.package.manifest_id}"] + self.assertTrue(self.recover()) + self.assertEqual([c.args[1] for c in self.controller.remote_command.call_args_list], + ["ota status", f"ota pull {self.package.manifest_id} flash", "ota status"]) + + def test_probe_error_aborts_no_pull(self): + self.controller.send_legacy_ota_cache_probe.side_effect = ota.OtaError("queue full") + with self.assertRaisesRegex(ota.OtaError, "queue full"): + self.recover() + self.controller.remote_command.assert_not_called() + + def test_raw_send_error_or_missing_marker_is_not_success(self): + for failure in ("Error sending raw packet", "Unknown command", ""): + controller = object.__new__(ota.Controller) + def execute(commands, *a, **kw): + output = failure + (commands[-1] if failure else "") + return subprocess.CompletedProcess([], 0, stdout=output, stderr="") + with mock.patch.object(controller, "_execute", side_effect=execute): + with self.assertRaises(ota.OtaError): + controller.send_legacy_ota_cache_probe(b"\x32\0\0", timeout=5) + + def test_parser_requires_explicit_compatible_topology(self): + parser = ota.build_parser() + common = ["update.mota", "target", "--legacy-relay-dedup-workaround"] + for flags in ([], ["--relay", "relay"], ["--source-shares-controller"], + ["--relay", "a", "--relay", "b", "--source-shares-controller"]): + with contextlib.redirect_stderr(io.StringIO()), self.assertRaises(SystemExit): + ota.validate_args(parser.parse_args(common + flags), parser) + + def test_monitor_uses_same_deadline_and_bounded_attempt_counter(self): + self.args.transfer_timeout_minutes = 20 + self.args.poll_seconds = 1 + self.args.reply_timeout = 1 + ready = f"download: ready to install 1/1 id={self.package.manifest_id}" + for passive in (False, True): + with self.subTest(passive=passive): + self.controller.remote_command.side_effect = [self.status, self.status, ready] + seeder = mock.Mock() if passive else None + if seeder: + seeder.payload_read_progress.return_value = (0, 1, 0) + with mock.patch.object(ota, "initial_status_wait_seconds", return_value=0), \ + mock.patch.object(ota, "passive_progress_stall_seconds", return_value=0), \ + mock.patch.object(ota, "transfer_tail_guard_seconds", return_value=0), \ + mock.patch.object(ota, "recover_legacy_manifest_timeout", + side_effect=[True, True, False]) as recover: + self.assertEqual(ota.monitor_download(self.controller, self.args, + self.package, seeder), ready) + self.assertEqual([c.args[-1] for c in recover.call_args_list], [0, 1, 2]) + self.assertEqual({c.args[-2] for c in recover.call_args_list}, {1300}) + + +if __name__ == "__main__": + unittest.main()