diff --git a/docs/ota_protocol.md b/docs/ota_protocol.md index f730b73c..8811ee35 100644 --- a/docs/ota_protocol.md +++ b/docs/ota_protocol.md @@ -359,7 +359,7 @@ Message types: | `OTA_HAVE` | 0x03 | discovery | the catalog reply (fragmented, digest-tagged) | | `OTA_GET_MANIFEST` | 0x04 | transfer | request a manifest's fragments (`want_mask`) by `manifest_id` | | `OTA_MANIFEST` | 0x05 | transfer | the manifest-minus-leaves, fragmented | -| `OTA_REQ` | 0x06 | transfer | request specific DATA fragments of one block (`want_mask`) | +| `OTA_REQ` | 0x06 | transfer | request fragments from an adaptive flight of 1-4 blocks (`want_mask` per block) | | `OTA_DATA` | 0x07 | transfer | one self-describing fragment of a block's data | | `OTA_REQ_PROOF` | 0x08 | transfer | request/re-request a missing proof | | `OTA_PROOF` | 0x09 | transfer | the merkle proof for one block | @@ -371,8 +371,10 @@ Message types: - **Priority:** `OTA_ADV`, `OTA_QUERY`, and `OTA_HAVE` enqueue at background priority 250. Once a fetch is active, manifest, block request, data, and proof messages enqueue at primary priority 0. Relay-only nodes classify the wire message identically, so a transfer stays primary across the complete path. -- **Reliability is *eventual*:** the fetcher re-requests only missing fragments after a one-second stalled - service tick. No hard ACKs or global ordering are required. +- **Reliability is *eventual*:** the fetcher re-requests only missing fragments after an adaptive deadline + derived from packet airtime, outstanding response packets, duty pacing, and path length. The manager may + still run a one-second maintenance tick, but that tick is not itself a retry timer. No hard ACKs or global + ordering are required. - **Relay envelope:** OTA still uses a bounded flood-shaped mesh header so the same packets can cross the configured number of hops without first discovering an addressed return path. During TempRadio each node forwards one copy; active OTA packets do not use the generic flood-retry subsystem. The fetcher verifies @@ -454,13 +456,13 @@ fetcher server (any node that has the mid) OTA_GET_MANIFEST(mid, want_mask) > (want_mask=0xFFFF first; only missing fragments on retry) <------- OTA_MANIFEST(mid, frag_idx, frag_total, bytes) x requested frags (reassemble manifest, verify, compute geometry: BC, block_size, payload_size) - for each missing block: - OTA_REQ(mid, block_idx, want_mask) > (want_mask=all fragments first; only the holes on retry) - <------- OTA_DATA(mid, block_idx, frag_off, data) x requested frags - (reassemble block from frag_off slices) - <------- OTA_PROOF(mid, block_idx, n_proof, proof) (proactive) - [if proof is absent after grace: OTA_REQ_PROOF(mid, block_idx) -> legacy/loss fallback] - (verify proof vs merkle_root -> write block -> write leaves[i]) + for each adaptive flight of missing blocks (starts at 1, grows on clean flights): + OTA_REQ(mid, {block_idx, want_mask}[]) > (one packet; all fragments first, only holes on recovery) + <------- OTA_DATA(mid, block_idx, frag_off, data) x requested frags/block + <------- OTA_PROOF(mid, block_idx, n_proof, proof) x requested blocks + (independently reassemble + verify each block, but remain RX-silent until the flight drains) + [after adaptive deadline: recover one block's holes, or OTA_REQ_PROOF for a missing proof] + (clean flight grows by one block; recovered flight halves the next width) when all blocks present: verify full merkle_root + image_hash -> COMPLETE ``` @@ -471,7 +473,8 @@ All offsets after the 1-byte type. Encoders/decoders in `OtaProtocol.cpp`; const ``` OTA_GET_MANIFEST: manifest_id[4] want_mask(uint16) # bit k = send manifest fragment k; 0xFFFF = all OTA_MANIFEST: manifest_id[4] frag_idx(1) frag_total(1) bytes[] # up to OTA_MF_FRAG=176 B/frag -OTA_REQ: manifest_id[4] block_idx(uint16) want_mask(uint16) # bit k = send fragment k of block +OTA_REQ: manifest_id[4] { block_idx(uint16) want_mask(uint16) }[1..4] + # one or more rows; bit k = send fragment k of that block OTA_DATA: manifest_id[4] block_idx(uint16) frag_off(uint16) data[] # up to OTA_FRAG_DATA=160 B OTA_REQ_PROOF: manifest_id[4] block_idx(uint16) OTA_PROOF: manifest_id[4] block_idx(uint16) n_proof(1) proof[] # n_proof x 4 bytes @@ -496,6 +499,17 @@ OTA_LEAVES: manifest_id[4] frag_idx(1) frag_total(1) bytes[] # up byte offset of `data` within the block, so the global position is `block_idx*block_size + frag_off` - a fragment is self-placing when returned by the source. The fetcher tracks a per-block slice bitmap and reassembles before requesting the proof. +- **Adaptive flight size is not signed block size.** The container continues to use 1 KiB Merkle leaves and + each slot is one existing 1 KiB block. A clean link changes how many of those blocks one `OTA_REQ` names: + 1, then 2, then 3, then the compiled cap. The manifest stores `block_size_log2`, so 3 KiB is not a valid + logical geometry; enabling 2 KiB would require larger device reassembly buffers and new-package/bootloader + validation while saving only one request/proof pair per 2 KiB. It does not address premature retries, which + were the dominant packet multiplier. +- **Append-only request-window compatibility:** the first four-byte request row is exactly the original + `block_idx + want_mask` body. Old sources decode that row and ignore appended bytes. New sources queue all + rows. If a new fetcher meets an old source, the unserved tail rows eventually time out and are recovered as + ordinary single-row requests; the dirty flight then contracts. Old fetchers continue sending nine-byte + single-row requests, which new sources accept normally. - **Fragment-level requests (anti-deadlock + anti-congestion):** `OTA_REQ`, `OTA_GET_MANIFEST`, and `OTA_QUERY` carry fragment masks. For catalog discovery, `want_fragments` is a 32-bit bitmap and covers the protocol maximum 255-row catalog (26 fragments at the current packet size). For block and manifest transfer, the @@ -508,10 +522,10 @@ OTA_LEAVES: manifest_id[4] frag_idx(1) frag_total(1) bytes[] # up reassembly bitmap (<=16 fragments/block; 1 KB blocks = 7). `OTA_PROOF` is a single packet and needs no mask. - **Data and proof remain separate packets, without a normal extra round trip.** A server retains requested blocks in a bounded descriptor queue, admits at most one response per main-loop pass, and sends one - `OTA_PROOF` after the requested fragments. The receiver waits 500 ms after completing a block, then uses - `OTA_REQ_PROOF` only as a loss/legacy-server fallback. It keeps a bounded multi-block window, so one block - can await proof or flash commit while DATA for the next block arrives. Slots retry independently and only - one stalled slot is retried per service tick. + `OTA_PROOF` after each block's requested fragments. The receiver uses an airtime/path-aware proof grace + (never less than 500 ms) and defers `OTA_REQ_PROOF` while any flight slot still expects DATA. Thus a missing + early proof cannot make the receiver transmit into the rest of a legitimate half-duplex response train. + After the complete-flight deadline, only one slot's missing fragments/proof is requested at a time. ### 8.5 Sizing against `MAX_PACKET_PAYLOAD = 184` @@ -529,9 +543,11 @@ payload); larger self-images pass a bigger scratch buffer. OTA packets may cross normal mesh relay hops, but each participating node processes or relays them only while its `tempradio` window is actually running. A receiver selects missing blocks in serial order into a bounded -request pipeline. The default capacity is two blocks. Targets with a larger compiled capacity begin at two, -grow the live window by one after four clean verified blocks, and shrink it after two consecutive no-progress -service ticks; the RAK3401 LoRa-OTA target currently adapts from two through four blocks. It never serves +request flight. Every session starts with one block. A clean completed flight increases the next request by +one block; a flight requiring fragment/proof recovery halves the next width (`4 -> 2`, `3 -> 2`, `2 -> 1`). +The default compiled cap is two blocks; the RAK3401 LoRa-OTA target caps at four, so it probes +`1 -> 2 -> 3 -> 4`. All rows are sent in one backward-compatible `OTA_REQ`, and no freed slot is refilled +until the current flight is finished. It never serves partial blocks. A normal install receiver never re-advertises its completed download. An SD archive node is the deliberate exception: after a fully proof-verified container is published to its persistent archive, it registers that complete file as a MotaSource and advertises it as a new seeder. @@ -543,8 +559,12 @@ public-flood receive holdoff and relay jitter, and do not schedule generic flood queue admits at most two DATA/PROOF packets ahead of the radio while preserving at least four free packet-pool entries. CAD remains enabled to arbitrate the half-duplex channel, but its busy retry is scaled to one-quarter of a packet airtime and clamped to 5-50 ms instead of the ordinary 120-360 ms cadence. Discovery traffic keeps -collision jitter and background priority. These changes remove software waits and duplicate bursts; they do -not remove the one required forwarding transmission per hop. +collision jitter and background priority. The fetch deadline uses the active radio's measured maximum-packet +airtime, remaining DATA/PROOF packet count, dispatcher airtime factor, and longest observed path (falling back +to the configured hop horizon before one is observed), with bounded guard time. Faster SF/BW settings +therefore recover loss sooner; slower or multi-hop settings do not spuriously re-request a response still on +air. These changes remove software waits and duplicate bursts; they do not remove the one required forwarding +transmission per hop. --- diff --git a/docs/ota_user_guide.md b/docs/ota_user_guide.md index 6cb3e292..2f46a490 100644 --- a/docs/ota_user_guide.md +++ b/docs/ota_user_guide.md @@ -149,9 +149,13 @@ re-running a `validate` pull re-begins fresh (it never resumes a stale partial). trusted - every kept block is checked against the target's own fingerprints, so a mismatched or missing seed just means those blocks are fetched over the radio (correct result, only slower). -The node fetches from one source as **primary traffic**, with a bounded adaptive block-request window. -RAK3401 OTA builds start with two concurrent block requests, grow to four on a clean link, and contract on -repeated stalls. Mesh repeaters carry the packets only while their temporary-radio windows are active. +The node fetches from one source as **primary traffic**, with bounded adaptive request flights. Every session +probes with one 1 KiB block, then clean flights grow `1 -> 2 -> 3 -> 4` blocks on RAK3401 OTA builds. All blocks +in a flight share one request packet, and the receiver stays silent until the source/relays finish returning +them. A recovery halves the next flight. Retry timing follows the active SF/BW airtime, duty budget, and path +length, so faster settings recover sooner without a fixed one-second request colliding with a valid multi-hop +response. The signed block size itself remains 1 KiB. Mesh repeaters carry the packets only while their +temporary-radio windows are active. Check progress with `ota status`. If a `folder` pull loses its link mid-transfer, `ota status` shows **paused** - the host keeps the diff --git a/examples/companion_radio/MyMesh.cpp b/examples/companion_radio/MyMesh.cpp index 60b8a0ec..1cf3fe49 100644 --- a/examples/companion_radio/MyMesh.cpp +++ b/examples/companion_radio/MyMesh.cpp @@ -359,6 +359,12 @@ int MyMesh::getFromOfflineQueue(uint8_t frame[]) { } float MyMesh::getAirtimeBudgetFactor() const { + // TempRadio is a short-lived, explicitly coordinated OTA channel. Let its + // primary transfer use the full dispatcher budget; the persisted public + // channel duty factor becomes authoritative again as soon as TempRadio ends. +#if defined(COMPANION_RADIO_FULL) + if (isTempRadioActive()) return 0.0f; +#endif return _prefs.airtime_factor; } diff --git a/examples/simple_repeater/MyMesh.h b/examples/simple_repeater/MyMesh.h index 09bc2a0e..82173eee 100644 --- a/examples/simple_repeater/MyMesh.h +++ b/examples/simple_repeater/MyMesh.h @@ -725,7 +725,10 @@ class MyMesh : public mesh::Mesh, public CommonCLICallbacks protected: bool isTempRadioActive() const override; float getAirtimeBudgetFactor() const override { - return _prefs.airtime_factor; + // A bounded TempRadio window is an explicitly coordinated private OTA + // channel. Use its full TX budget without changing the saved normal-radio + // duty factor; restoring the ordinary tuple restores ordinary pacing too. + return isTempRadioActive() ? 0.0f : _prefs.airtime_factor; } bool getCADEnabled() const override { return _prefs.cad_enabled; diff --git a/src/Mesh.cpp b/src/Mesh.cpp index fef475ee..dc2c28fe 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -356,6 +356,12 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { _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)); ota::ota_ctx().manager.serviceEgress(); if (millisHasNowPassed(_next_ota_tick)) { // one-shot on first tick: resume an interrupted fetch left staged in flash before a reboot. Only adopt @@ -1061,17 +1067,19 @@ DispatcherAction Mesh::onRecvPacket(Packet* pkt) { if (!seen) _tables->markSeen(pkt); const bool primary_ota = isPrimaryOtaTraffic(pkt->payload, pkt->payload_len); const uint8_t ota_priority = otaTrafficPriority(pkt->payload, pkt->payload_len); + bool terminal_ota = false; #if defined(ENABLE_OTA) ota::ota_ctx().manager.set_clock(_ms->getMillis()); // discovery jitter/ages - ota::ota_ctx().manager.on_message(pkt->payload, pkt->payload_len); // central OTA receive (beacon/query/ - // have/manifest/data/proof; all roles) + ota::ota_ctx().manager.note_rx_path_hops(n); // adaptive fetch timing + terminal_ota = ota::ota_ctx().manager.on_message(pkt->payload, pkt->payload_len); + // central OTA receive (all roles) ota::ota_ctx().track_session(ota::ota_ctx().manager.fetchState(), _ms->getMillis()); onOtaRecv(pkt); // optional per-example hook #endif // 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 (!seen && pkt->isRouteFlood() && !pkt->isMarkedDoNotRetransmit() + if (!terminal_ota && !seen && pkt->isRouteFlood() && !pkt->isMarkedDoNotRetransmit() #if defined(ENABLE_OTA) && n < getOtaHopLimit() #endif diff --git a/src/helpers/esp32/WiFiOtaSeeder.cpp b/src/helpers/esp32/WiFiOtaSeeder.cpp index 002ae4ca..515f5fec 100644 --- a/src/helpers/esp32/WiFiOtaSeeder.cpp +++ b/src/helpers/esp32/WiFiOtaSeeder.cpp @@ -19,8 +19,11 @@ namespace { WiFiServer seeder_server(OTA_SEEDER_TCP_PORT); WiFiClient seeder_client; -SerialMotaSource seeder_source(seeder_client, 3000); -FolderMotaStore folder_store(seeder_client, 3000); +// WiFiClient::flush() clears received bytes rather than flushing TX. Disabling +// the serial-only flush prevents a fast host reply from being discarded and +// turning every block read into a three-second timeout/retry. +SerialMotaSource seeder_source(seeder_client, 3000, false); +FolderMotaStore folder_store(seeder_client, 3000, false); bool listener_active = false; bool tcp_folder_attached = false; @@ -80,6 +83,8 @@ void WiFiOtaSeeder::loop() { WiFiClient incoming = seeder_server.available(); if (!incoming) return; + incoming.setNoDelay(true); // tiny framed requests should leave immediately + if (!WiFiOtaSeederPolicy::canAttachTcpFolder(context.folder_active, tcp_folder_attached)) { incoming.stop(); diff --git a/src/helpers/ota/FolderMotaStore.cpp b/src/helpers/ota/FolderMotaStore.cpp index 139edcb2..ba5cfdd9 100644 --- a/src/helpers/ota/FolderMotaStore.cpp +++ b/src/helpers/ota/FolderMotaStore.cpp @@ -27,11 +27,18 @@ bool FolderMotaStore::txn(uint8_t op, const uint8_t* args, uint16_t arglen, while (_io.read() >= 0) {} // drop any stale/partial bytes before a fresh request uint8_t xs = op; for (uint16_t i = 0; i < arglen; i++) xs ^= args[i]; - _io.write(MOTA_SEEDER_REQ_MAGIC0); _io.write(MOTA_SEEDER_REQ_MAGIC1); - _io.write(op); - if (arglen) _io.write(args, arglen); - _io.write(xs); - _io.flush(); + // Keep one contiguous write for TCP (and fewer calls on UART). The largest + // request is OP_WRITE's fixed header plus one bounded data chunk. + if (arglen > 10 + MOTA_SEEDER_WRITE_MAX) return false; + uint8_t frame[4 + 10 + MOTA_SEEDER_WRITE_MAX]; + uint16_t frame_len = 0; + frame[frame_len++] = MOTA_SEEDER_REQ_MAGIC0; + frame[frame_len++] = MOTA_SEEDER_REQ_MAGIC1; + frame[frame_len++] = op; + if (arglen) { memcpy(frame + frame_len, args, arglen); frame_len += arglen; } + frame[frame_len++] = xs; + if (_io.write(frame, frame_len) != frame_len) return false; + if (_flush_after_write) _io.flush(); uint32_t t0 = millis(); bool got = false; uint8_t prev = 0; while ((millis() - t0) < _to) { // scan for response magic 'm' 's' (tolerate noise) diff --git a/src/helpers/ota/FolderMotaStore.h b/src/helpers/ota/FolderMotaStore.h index 6e60ffba..88d20abd 100644 --- a/src/helpers/ota/FolderMotaStore.h +++ b/src/helpers/ota/FolderMotaStore.h @@ -22,7 +22,9 @@ namespace ota { class FolderMotaStore : public OtaStore { public: - explicit FolderMotaStore(Stream& io, uint32_t timeout_ms = 3000) : _io(io), _to(timeout_ms) {} + explicit FolderMotaStore(Stream& io, uint32_t timeout_ms = 3000, + bool flush_after_write = true) + : _io(io), _to(timeout_ms), _flush_after_write(flush_after_write) {} // The container being pulled - set from the chosen `ota pull` mid before begin()/reopen(). void set_mid(const uint8_t mid[4]) { memcpy(_mid, mid, 4); } @@ -49,6 +51,7 @@ private: Stream& _io; uint32_t _to; + bool _flush_after_write; uint8_t _mid[4] = {0, 0, 0, 0}; uint32_t _total = 0; }; diff --git a/src/helpers/ota/MotaSourceSerial.cpp b/src/helpers/ota/MotaSourceSerial.cpp index 0ceea474..98526316 100644 --- a/src/helpers/ota/MotaSourceSerial.cpp +++ b/src/helpers/ota/MotaSourceSerial.cpp @@ -25,13 +25,18 @@ bool SerialMotaSource::readExact(uint8_t* b, uint32_t n) { bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len) { while (_io.read() >= 0) {} // drop any stale/partial bytes before a fresh request + if (arglen > 7) return false; // largest source request is READ(idx,off,len) uint8_t xs = op; for (uint8_t i = 0; i < arglen; i++) xs ^= args[i]; - _io.write(MOTA_SEEDER_REQ_MAGIC0); _io.write(MOTA_SEEDER_REQ_MAGIC1); - _io.write(op); - if (arglen) _io.write(args, arglen); - _io.write(xs); - _io.flush(); + uint8_t frame[11]; + uint8_t frame_len = 0; + frame[frame_len++] = MOTA_SEEDER_REQ_MAGIC0; + frame[frame_len++] = MOTA_SEEDER_REQ_MAGIC1; + frame[frame_len++] = op; + if (arglen) { memcpy(frame + frame_len, args, arglen); frame_len += arglen; } + frame[frame_len++] = xs; + if (_io.write(frame, frame_len) != frame_len) return false; + if (_flush_after_write) _io.flush(); // scan for response magic 'm' 's' (tolerate leading noise) uint32_t t0 = millis(); bool got = false; diff --git a/src/helpers/ota/MotaSourceSerial.h b/src/helpers/ota/MotaSourceSerial.h index 5a8d1134..4dc3142c 100644 --- a/src/helpers/ota/MotaSourceSerial.h +++ b/src/helpers/ota/MotaSourceSerial.h @@ -14,7 +14,12 @@ namespace ota { class SerialMotaSource : public MotaSource { public: - explicit SerialMotaSource(Stream& io, uint32_t timeout_ms = 400) : _io(io), _to(timeout_ms) {} + // Hardware serial needs flush() to finish transmitting before waiting for a + // reply. NetworkClient/WiFiClient gives flush() receive-side semantics and + // discards bytes, so TCP users must pass flush_after_write=false. + explicit SerialMotaSource(Stream& io, uint32_t timeout_ms = 400, + bool flush_after_write = true) + : _io(io), _to(timeout_ms), _flush_after_write(flush_after_write) {} uint8_t count() override; bool describe(uint8_t idx, MotaDesc& out) override; @@ -29,6 +34,7 @@ private: Stream& _io; uint32_t _to; + bool _flush_after_write; }; } // namespace ota diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index f1b644f8..9e27050c 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -435,33 +435,42 @@ bool OtaManager::loadActiveServeBlock() { return true; } -void OtaManager::handleReq(const uint8_t* m, uint16_t n) { - ReqMsg rq; - if (!decode_req(m, n, rq)) return; - uint32_t idx = rq.block_idx; +bool OtaManager::handleReq(const uint8_t* m, uint16_t n) { + ReqWindowMsg rq; + if (!decode_req_window(m, n, rq)) return false; ServeView* v = resolve(rq.manifest_id); - if (!v || idx >= v->m.block_count || idx > 0xFFFFu) return; - uint32_t blen = v->m.block_size(); - uint32_t off = idx * blen; - if (off + blen > v->m.payload_size) blen = v->m.payload_size - off; - uint16_t valid_mask = frag_full_mask((blen + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA); - uint16_t want = (uint16_t)(rq.want_mask & valid_mask); - if (want != 0) queueServeJob(v->m.merkle_root, (uint16_t)idx, want); + if (!v) return false; + bool accepted = false; + const uint32_t block_size = v->m.block_size(); + for (uint8_t i = 0; i < rq.n_items; i++) { + const uint32_t idx = rq.items[i].block_idx; + if (idx >= v->m.block_count) continue; + uint32_t blen = block_size; + const uint32_t off = idx * block_size; + if (off + blen > v->m.payload_size) blen = v->m.payload_size - off; + const uint16_t valid_mask = frag_full_mask((blen + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA); + const uint16_t want = (uint16_t)(rq.items[i].want_mask & valid_mask); + if (want != 0) accepted |= queueServeJob(v->m.merkle_root, (uint16_t)idx, want); + } + return accepted; } -void OtaManager::handleReqProof(const uint8_t* m, uint16_t n) { +bool OtaManager::handleReqProof(const uint8_t* m, uint16_t n) { ReqProofMsg rp; - if (!decode_req_proof(m, n, rp)) return; + if (!decode_req_proof(m, n, rp)) return false; ServeView* v = resolve(rp.manifest_id); - if (!v || rp.block_idx >= v->m.block_count) return; - if ((uint64_t)v->m.block_count * 4 > v->scratch_sz) return; - queueServeJob(v->m.merkle_root, rp.block_idx, 0); // proof-only, or merge with queued DATA + if (!v || rp.block_idx >= v->m.block_count) return false; + if ((uint64_t)v->m.block_count * 4 > v->scratch_sz) return false; + return queueServeJob(v->m.merkle_root, rp.block_idx, 0); // proof-only, or merge with queued DATA } void OtaManager::serviceEgress() { // A proactive proof normally follows the final DATA fragment. If it was lost, or the source is older and - // never sent one, issue the legacy proof request after a short grace. Admit at most one fallback per call. - if (_fstate == FETCHING) { + // never sent one, issue the legacy proof request after a short grace. Stay RX-silent while another slot + // still expects DATA, and restart the grace on every new packet, so this fallback cannot collide with a + // legitimate multi-block response train. Admit at most one fallback per call. + if (_fstate == FETCHING && !flightHasPendingData() + && (uint32_t)(_now_ms - _fetch_wait_since_ms) >= proofGraceMs()) { for (uint8_t i = 0; i < OTA_FETCH_PIPELINE; i++) { ReassemblySlot& slot = _reasm[i]; if (!slot.awaiting_proof || slot.proof_request_at == 0 @@ -714,8 +723,8 @@ void OtaManager::clearReassembly() { _retry_slot = 0; _req_count = 0; _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; - _pipeline_success_streak = 0; - _pipeline_stall_streak = 0; + _flight_dirty = false; + noteFetchActivity(); } int OtaManager::findReassemblySlot(uint32_t block) const { @@ -732,44 +741,98 @@ int OtaManager::findEmptyReassemblySlot() const { return -1; } -void OtaManager::notePipelineBlockVerified() { - _pipeline_stall_streak = 0; - if (_pipeline_width >= OTA_FETCH_PIPELINE) return; - if (++_pipeline_success_streak < OTA_FETCH_PIPELINE_GROW_BLOCKS) return; - _pipeline_success_streak = 0; - _pipeline_width++; - OTA_DBG("OTA: request pipeline grew to %u/%u after clean block progress\n", - (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); +uint8_t OtaManager::activePipelineSlots() const { + uint8_t active = 0; + for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) { + if (_reasm[slot].block != NO_BLOCK) active++; + } + return active; +} + +bool OtaManager::flightHasPendingData() const { + for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) { + if (_reasm[slot].block != NO_BLOCK && !_reasm[slot].awaiting_proof) return true; + } + return false; +} + +static uint8_t count_mask_bits(uint16_t mask) { + uint8_t count = 0; + while (mask) { count += (uint8_t)(mask & 1u); mask >>= 1; } + return count; +} + +uint32_t OtaManager::fetchRetryTimeoutMs() const { + if (_radio_packet_airtime_ms == 0) return 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 + // packet must traverse the source+relay path and each transmitter is paced by its configured duty budget. + uint32_t packets = 1; + uint8_t active = 0; + for (uint8_t slot_index = 0; slot_index < OTA_FETCH_PIPELINE; slot_index++) { + const ReassemblySlot& slot = _reasm[slot_index]; + if (slot.block == NO_BLOCK) continue; + active++; + packets += slot.awaiting_proof ? 1u + : (uint32_t)count_mask_bits((uint16_t)(slot.need & ~slot.mask)) + 1u; + } + if (active == 0) { + const uint32_t fragments = (_fbs + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA; + packets += (fragments + 1) * _pipeline_width; + } + uint32_t path_transmissions = _observed_path_transmissions; + if (path_transmissions == 0) path_transmissions = (uint32_t)_max_hops + 1u; + uint64_t service = (uint64_t)packets * _radio_packet_airtime_ms * path_transmissions; + service = (service * _tx_spacing_permille + 999u) / 1000u; + service = (service * 5u + 3u) / 4u; // 25% CAD/relay contention allowance + 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; +} + +uint32_t OtaManager::proofGraceMs() const { + if (_radio_packet_airtime_ms == 0) return 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; + grace = (grace * _tx_spacing_permille + 999u) / 1000u; + 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; +} + +void OtaManager::finishFlight() { + const uint8_t old_width = _pipeline_width; + if (_flight_dirty) { + if (_pipeline_width > OTA_FETCH_PIPELINE_INITIAL) { + _pipeline_width = (uint8_t)((_pipeline_width + 1u) / 2u); + if (_pipeline_width < OTA_FETCH_PIPELINE_INITIAL) _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; + } + } else if (_pipeline_width < OTA_FETCH_PIPELINE) { + _pipeline_width++; + } + if (_pipeline_width != old_width) { + OTA_DBG("OTA: adaptive flight %s to %u/%u\n", _flight_dirty ? "shrunk" : "grew", + (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); + } + _flight_dirty = false; } void OtaManager::notePipelineStall() { - _pipeline_success_streak = 0; - if (_pipeline_width <= OTA_FETCH_PIPELINE_INITIAL) { - _pipeline_stall_streak = 0; - return; + if (!_flight_dirty) { + OTA_DBG("OTA: adaptive flight recovery at width %u/%u\n", + (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); } - if (++_pipeline_stall_streak < OTA_FETCH_PIPELINE_SHRINK_TICKS) return; - _pipeline_stall_streak = 0; - _pipeline_width--; - OTA_DBG("OTA: request pipeline shrank to %u/%u after stalled ticks\n", - (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); + _flight_dirty = true; } bool OtaManager::blockInPipeline(uint32_t block) const { return findReassemblySlot(block) >= 0; } -uint32_t OtaManager::pipelineProgress() const { - // A small stable fingerprint is sufficient here: equality only suppresses a retry for one service tick. - uint32_t progress = 2166136261u; - for (uint8_t slot = 0; slot < OTA_FETCH_PIPELINE; slot++) { - progress = (progress ^ _reasm[slot].block) * 16777619u; - progress = (progress ^ _reasm[slot].mask) * 16777619u; - progress = (progress ^ (_reasm[slot].awaiting_proof ? 1u : 0u)) * 16777619u; - } - return progress; -} - // Begin (or resume) fetching a chosen mid: try a staged-partial resume first, else request the manifest. void OtaManager::startFetch(const uint8_t* mid, uint32_t target, bool validate) { (void)target; @@ -781,6 +844,7 @@ void OtaManager::startFetch(const uint8_t* mid, uint32_t target, bool validate) // real partial), so never adopt it via resumeStaged - always re-begin and run the manifest->leaves->diff. if (!validate && resumeStaged(mid)) return; // (non-validate) resume a partial container left in flash memcpy(_fid, mid, 4); + _observed_path_transmissions = 0; // learn the actual source/relay path from this fetch's replies _fstate = WANT_MANIFEST; _mf_total = 0; _mf_mask = 0; _mf_len = 0; _mf_retries = 0; _loop_last_mfmask = 0; // fresh manifest reassembly memset(_mf_buf, 0, sizeof(_mf_buf)); @@ -859,7 +923,6 @@ void OtaManager::handleManifest(const uint8_t* m, uint16_t n) { _fpoff = payload_off; _floff = leaves_off; _fpsize = payload_size; _fbc = bc; _fbs = bs; _ftotal = total; _have = 0; clearReassembly(); // fresh transfer: drop any prior per-block state - _loop_last_have = 0; _loop_last_progress = pipelineProgress(); // Warm-start (folder-capture): if requested and the image is small enough for the fixed leaves bitmap, // bulk-fetch the target leaves and diff the seed already staged in the store; else just transfer normally. if (_validate && beginLeafDiff()) { @@ -868,6 +931,7 @@ void OtaManager::handleManifest(const uint8_t* m, uint16_t n) { } _fstate = FETCHING; OTA_DBG("OTA: FETCHING bc=%u bs=%u total=%u\n", (unsigned)bc, (unsigned)bs, (unsigned)total); + requestMissing(); // final manifest fragment arrived: open flight 1 immediately } // --- leaf-diff warm-start (fetcher side; motatool folder-capture only) ----------------------------- @@ -980,7 +1044,7 @@ bool OtaManager::resumeStaged(const uint8_t* want_mid) { _fpoff = payload_off; _floff = leaves_off; _fpsize = m.payload_size; _fbc = bc; _fbs = bs; _ftotal = total; clearReassembly(); - _loop_last_have = 0; _loop_last_progress = 0; + _observed_path_transmissions = 0; beginStagedVerification(); OTA_DBG("OTA: RESUME verifying %u blocks total=%u\n", (unsigned)bc, (unsigned)total); @@ -1077,53 +1141,58 @@ void OtaManager::verifyStagedStep() { ? COMPLETE : FAILED; } else { _fstate = FETCHING; - _loop_last_have = _have; - _loop_last_progress = pipelineProgress(); requestMissing(); } _resume_merkle.reset(); } -void OtaManager::handleData(const uint8_t* m, uint16_t n) { +bool OtaManager::handleData(const uint8_t* m, uint16_t n) { DataMsg dm; - if (!decode_data(m, n, dm) || !_fetch) return; - if (_fstate != FETCHING || memcmp(dm.manifest_id, _fid, 4) != 0) return; - if (dm.block_idx >= _fbc) return; - if (blockPresent(dm.block_idx)) return; // already stored + verified + if (!decode_data(m, n, dm) || !_fetch) return false; + if (_fstate != FETCHING || memcmp(dm.manifest_id, _fid, 4) != 0) return false; + if (dm.block_idx >= _fbc) return false; + if (blockPresent(dm.block_idx)) return true; // late duplicate for this terminal int slot_index = findReassemblySlot(dm.block_idx); - if (slot_index < 0) return; // accept only blocks inside our bounded window + if (slot_index < 0) return false; // not one of this node's requested blocks ReassemblySlot& slot = _reasm[slot_index]; uint32_t blen = blockLen(dm.block_idx); - if (dm.frag_off % OTA_FRAG_DATA != 0) return; // canonical FRAG_DATA-aligned slices only - if ((uint32_t)dm.frag_off + dm.data_len > blen) return; // slice out of the block + if (dm.frag_off % OTA_FRAG_DATA != 0) return false; // canonical FRAG_DATA-aligned slices only + if ((uint32_t)dm.frag_off + dm.data_len > blen) return false; // slice out of the block uint32_t expected_len = blen - dm.frag_off; if (expected_len > OTA_FRAG_DATA) expected_len = OTA_FRAG_DATA; - if (dm.data_len != expected_len) return; // do not mark a short slice as complete + if (dm.data_len != expected_len) return false; // do not mark a short slice as complete uint32_t kf = dm.frag_off / OTA_FRAG_DATA; - if (kf >= 16) return; + if (kf >= 16) return false; + const uint16_t bit = (uint16_t)(1u << kf); + if (slot.mask & bit) return true; // duplicate flood/retry copy is terminal too memcpy(slot.buf + dm.frag_off, dm.data, dm.data_len); - slot.mask |= (uint16_t)(1u << kf); - if (slot.mask != slot.need || slot.awaiting_proof) return; + slot.mask |= bit; + noteFetchActivity(); + if (slot.mask != slot.need || slot.awaiting_proof) return true; // The paced server sends a proof immediately after this block's requested DATA. Wait briefly so that proof // 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 + OTA_PROOF_GRACE_MS; + slot.proof_request_at = _now_ms + proofGraceMs(); + return true; } -void OtaManager::handleProof(const uint8_t* m, uint16_t n) { +bool OtaManager::handleProof(const uint8_t* m, uint16_t n) { ProofMsg pm; - if (!decode_proof(m, n, pm) || !_fetch) return; - if (_fstate != FETCHING || memcmp(pm.manifest_id, _fid, 4) != 0) return; + if (!decode_proof(m, n, pm) || !_fetch) return false; + if (_fstate != FETCHING || memcmp(pm.manifest_id, _fid, 4) != 0) return false; + if (pm.block_idx >= _fbc) return false; int slot_index = findReassemblySlot(pm.block_idx); - if (slot_index < 0 || !_reasm[slot_index].awaiting_proof) return; + if (slot_index < 0 || !_reasm[slot_index].awaiting_proof) return blockPresent(pm.block_idx); ReassemblySlot& slot = _reasm[slot_index]; uint32_t block = slot.block; uint32_t blen = blockLen(block); + noteFetchActivity(); if (!merkle_verify(slot.buf, blen, block, pm.proof, pm.n_proof, _froot, _fbc)) { + notePipelineStall(); clearReassemblySlot((uint8_t)slot_index); // bad -> drop and re-fetch only this block - requestMissing(); - return; + if (activePipelineSlots() == 0) { finishFlight(); requestMissing(); } + return true; } // verified -> commit the payload block, then its leaf (the present marker). A write failure here means a // FOLDER destination's seeder link dropped mid-transfer: PAUSE (hold progress on the host, stop @@ -1132,22 +1201,27 @@ void OtaManager::handleProof(const uint8_t* m, uint16_t n) { uint8_t leaf[4]; merkle_leaf(leaf, slot.buf, blen); if (!_fetch->write(_fpoff + block * _fbs, slot.buf, blen) || !_fetch->write(_floff + block * 4, leaf, 4)) { - _fstate = PAUSED; clearReassembly(); return; + _fstate = PAUSED; clearReassembly(); return true; } _have++; OTA_DBG("OTA: block %u OK have=%u/%u\n", (unsigned)block, (unsigned)_have, (unsigned)_fbc); - notePipelineBlockVerified(); clearReassemblySlot((uint8_t)slot_index); // periodically persist progress (meta/leaf page + open payload) so a reboot can resume (no-op for RAM); // cadence is runtime-tunable via `ota config checkpoint ` (0 = never) if (_checkpoint_blocks && _have % _checkpoint_blocks == 0) _fetch->checkpoint(); - if (_have < _fbc) { requestMissing(); return; } // keep the client window full + if (_have < _fbc) { + // Do not continuously refill a partially completed flight: the remaining response packets are already + // queued at the source/relays. Staying silent here is what prevents the old threefold request amplification. + if (activePipelineSlots() == 0) { finishFlight(); requestMissing(); } + return true; + } // Every leaf must be readable and collectively match the manifest root. // A scratch allocation/read failure is an integrity failure, never success. clearReassembly(); _fstate = storedLeavesRootMatches() ? COMPLETE : FAILED; if (_fstate == COMPLETE && !_fetch->finalize()) _fstate = FAILED; OTA_DBG("OTA: transfer %s\n", _fstate == COMPLETE ? "COMPLETE" : "FAILED(integrity/storage)"); + return true; } bool OtaManager::requestSlot(uint8_t slot_index) { @@ -1155,16 +1229,14 @@ bool OtaManager::requestSlot(uint8_t slot_index) { ReassemblySlot& slot = _reasm[slot_index]; if (slot.block >= _fbc) return false; _req_start = slot.block; - _req_count = 0; - for (uint8_t i = 0; i < OTA_FETCH_PIPELINE; i++) { - if (_reasm[i].block != NO_BLOCK) _req_count++; - } + _req_count = activePipelineSlots(); if (slot.awaiting_proof) { ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)slot.block; uint8_t b[16]; bool sent = emit(b, encode_req_proof(b, sizeof(b), rp), false); if (sent) { slot.proof_request_at = 0; + noteFetchActivity(); OTA_DBG("OTA: REQ_PROOF fallback block=%u (have=%u/%u window=%u)\n", (unsigned)slot.block, (unsigned)_have, (unsigned)_fbc, (unsigned)_req_count); } @@ -1178,11 +1250,38 @@ bool OtaManager::requestSlot(uint8_t slot_index) { OTA_DBG("OTA: REQ block=%u want=%04x (have=%u/%u mask=%04x window=%u)\n", (unsigned)slot.block, (unsigned)want, (unsigned)_have, (unsigned)_fbc, (unsigned)slot.mask, (unsigned)_req_count); - return emit(b, encode_req(b, sizeof(b), rq), false); + bool sent = emit(b, encode_req(b, sizeof(b), rq), false); + if (sent) noteFetchActivity(); + return sent; +} + +bool OtaManager::requestFlight() { + if (_fstate != FETCHING) return false; + ReqWindowMsg request = {}; + memcpy(request.manifest_id, _fid, 4); + for (uint8_t slot_index = 0; slot_index < OTA_FETCH_PIPELINE; slot_index++) { + const ReassemblySlot& slot = _reasm[slot_index]; + if (slot.block == NO_BLOCK || slot.awaiting_proof) continue; + ReqItem& item = request.items[request.n_items++]; + item.block_idx = (uint16_t)slot.block; + item.want_mask = (uint16_t)(slot.need & ~slot.mask); + if (item.want_mask == 0) item.want_mask = slot.need; + } + if (request.n_items == 0) return false; + _req_start = request.items[0].block_idx; + _req_count = request.n_items; + uint8_t b[5 + OTA_REQ_MAX_ITEMS * 4]; + OTA_DBG("OTA: REQ flight first=%u blocks=%u width=%u/%u\n", + (unsigned)_req_start, (unsigned)_req_count, + (unsigned)_pipeline_width, (unsigned)OTA_FETCH_PIPELINE); + bool sent = emit(b, encode_req_window(b, sizeof(b), request), false); + if (sent) noteFetchActivity(); + return sent; } bool OtaManager::fillPipeline() { - bool requested = false; + if (activePipelineSlots() != 0) return false; // flights never overlap/refill + uint8_t assigned = 0; while (true) { int slot_index = findEmptyReassemblySlot(); if (slot_index < 0) break; @@ -1194,24 +1293,26 @@ bool OtaManager::fillPipeline() { slot.need = frag_full_mask((blockLen(block) + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA); slot.awaiting_proof = false; slot.proof_request_at = 0; - requestSlot((uint8_t)slot_index); - requested = true; + assigned++; } - return requested; + if (assigned == 0) return false; + _flight_dirty = false; + requestFlight(); + return true; } void OtaManager::requestMissing() { if (_fstate != FETCHING) return; - if (fillPipeline()) return; + if (activePipelineSlots() == 0) { fillPipeline(); return; } - // A full window with no progress is stalled. Retry one slot per service tick, round-robin, so a weak - // block cannot trigger a simultaneous multi-block re-burst on a half-duplex channel. + // The airtime-aware deadline expired. Retry one slot's holes per deadline, round-robin, so a weak block + // cannot trigger a simultaneous multi-block re-burst on a half-duplex channel. Recovery dirties the + // flight; its successor is reduced (4->2, 3->2, 2->1) after all current slots finish. notePipelineStall(); for (uint8_t offset = 0; offset < OTA_FETCH_PIPELINE; offset++) { uint8_t slot = (uint8_t)((_retry_slot + offset) % OTA_FETCH_PIPELINE); if (_reasm[slot].block == NO_BLOCK) continue; - requestSlot(slot); - _retry_slot = (uint8_t)((slot + 1) % OTA_FETCH_PIPELINE); + if (requestSlot(slot)) _retry_slot = (uint8_t)((slot + 1) % OTA_FETCH_PIPELINE); return; } } @@ -1290,30 +1391,32 @@ void OtaManager::loop() { return; } if (_fstate != FETCHING) return; - // Retry only when a whole tick passed with no committed block, fragment, proof transition, or window - // assignment. This avoids re-request spam while any pipelined block is making progress. - uint32_t progress = pipelineProgress(); - if (_have == _loop_last_have && progress == _loop_last_progress) requestMissing(); - _loop_last_have = _have; - _loop_last_progress = pipelineProgress(); + if (activePipelineSlots() == 0) { + requestMissing(); + return; + } + // The main loop may still tick each second for discovery, but block recovery waits for the calculated + // response train: current SF/BW airtime * outstanding packets * source/relay transmissions * duty pacing. + // Every newly accepted fragment/proof restarts the deadline; duplicate flood copies do not. + if ((uint32_t)(_now_ms - _fetch_wait_since_ms) >= fetchRetryTimeoutMs()) requestMissing(); } // ---------------- dispatch ---------------- -void OtaManager::on_message(const uint8_t* msg, uint16_t len) { +bool OtaManager::on_message(const uint8_t* msg, uint16_t len) { switch (ota_msg_type(msg, len)) { - case OTA_ADV: handleAdv(msg, len); break; - case OTA_QUERY: handleQuery(msg, len); break; - case OTA_HAVE: handleHave(msg, len); break; - case OTA_GET_MANIFEST: handleGetManifest(msg, len); break; - case OTA_MANIFEST: handleManifest(msg, len); break; - case OTA_GET_LEAVES: handleGetLeaves(msg, len); break; - case OTA_LEAVES: handleLeaves(msg, len); break; - case OTA_REQ: handleReq(msg, len); break; - case OTA_DATA: handleData(msg, len); break; - case OTA_REQ_PROOF: handleReqProof(msg, len); break; - case OTA_PROOF: handleProof(msg, len); break; - default: break; + case OTA_ADV: handleAdv(msg, len); return false; + case OTA_QUERY: handleQuery(msg, len); return false; + case OTA_HAVE: handleHave(msg, len); return false; + case OTA_GET_MANIFEST: handleGetManifest(msg, len); return false; + case OTA_MANIFEST: handleManifest(msg, len); return false; + case OTA_GET_LEAVES: handleGetLeaves(msg, len); return false; + case OTA_LEAVES: handleLeaves(msg, len); return false; + case OTA_REQ: return handleReq(msg, len); + case OTA_DATA: return handleData(msg, len); + case OTA_REQ_PROOF: return handleReqProof(msg, len); + case OTA_PROOF: return handleProof(msg, len); + default: return false; } } diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index f9c145c8..745a3b5e 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -3,6 +3,7 @@ #include #include #include "OtaFormat.h" +#include "OtaProtocol.h" #include "OtaByteIO.h" #include "OtaStore.h" #include "MotaContainer.h" @@ -139,32 +140,31 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len) #define OTA_PROOF_GRACE_MS 500 // wait for the server's proactive proof before legacy REQ_PROOF fallback #endif #ifndef OTA_FETCH_PIPELINE -#define OTA_FETCH_PIPELINE 2 // concurrent client block slots; keeps one block ready behind proof/flash work +#define OTA_FETCH_PIPELINE 2 // max blocks in one adaptive request flight #endif #if OTA_FETCH_PIPELINE < 1 #error "OTA_FETCH_PIPELINE must be at least 1" #endif +#if OTA_FETCH_PIPELINE > OTA_REQ_MAX_ITEMS +#error "OTA_FETCH_PIPELINE exceeds the OTA_REQ window wire limit" +#endif #ifndef OTA_FETCH_PIPELINE_INITIAL - #if OTA_FETCH_PIPELINE >= 2 - #define OTA_FETCH_PIPELINE_INITIAL 2 - #else - #define OTA_FETCH_PIPELINE_INITIAL 1 - #endif +#define OTA_FETCH_PIPELINE_INITIAL 1 // probe every session conservatively, then grow one block per clean flight #endif #if OTA_FETCH_PIPELINE_INITIAL < 1 || OTA_FETCH_PIPELINE_INITIAL > OTA_FETCH_PIPELINE #error "OTA_FETCH_PIPELINE_INITIAL must be between 1 and OTA_FETCH_PIPELINE" #endif -#ifndef OTA_FETCH_PIPELINE_GROW_BLOCKS -#define OTA_FETCH_PIPELINE_GROW_BLOCKS 4 // clean verified blocks before adding one concurrent request slot +#ifndef OTA_FETCH_RETRY_MIN_MS +#define OTA_FETCH_RETRY_MIN_MS 5000 // quiet-time floor: covers host/queue/turnaround latency on fast links #endif -#if OTA_FETCH_PIPELINE_GROW_BLOCKS < 1 -#error "OTA_FETCH_PIPELINE_GROW_BLOCKS must be at least 1" +#ifndef OTA_FETCH_RETRY_MAX_MS +#define OTA_FETCH_RETRY_MAX_MS 60000 // bound loss recovery when configured radio settings are impractical #endif -#ifndef OTA_FETCH_PIPELINE_SHRINK_TICKS -#define OTA_FETCH_PIPELINE_SHRINK_TICKS 2 // consecutive no-progress retry ticks before removing one slot +#ifndef OTA_FETCH_RETRY_GUARD_MS +#define OTA_FETCH_RETRY_GUARD_MS 500 // processing/CAD/queue jitter beyond calculated on-air service #endif -#if OTA_FETCH_PIPELINE_SHRINK_TICKS < 1 -#error "OTA_FETCH_PIPELINE_SHRINK_TICKS must be at least 1" +#ifndef OTA_FETCH_RETRY_FALLBACK_MS +#define OTA_FETCH_RETRY_FALLBACK_MS 5000 // portable-host fallback until the Mesh adapter supplies airtime #endif // nRF52 note: a flash page-erase halts the CPU (~85 ms, code runs from flash) and starves the LoRa RX, // so writing to flash on every received packet drops in-flight DATA and the transfer stalls. The SD-safe @@ -301,6 +301,18 @@ 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; } + // 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) { + _radio_packet_airtime_ms = max_packet_airtime_ms; + _tx_spacing_permille = tx_spacing_permille < 1000 ? 1000 : tx_spacing_permille; + } + // `relay_hops` is the received packet's path-hash count; one more radio transmission originated it. + // Keep the longest path seen in this fetch so a shorter duplicate cannot make recovery premature. + void note_rx_path_hops(uint8_t relay_hops) { + uint8_t transmissions = relay_hops == 0xFF ? 0xFF : (uint8_t)(relay_hops + 1); + if (transmissions > _observed_path_transmissions) _observed_path_transmissions = transmissions; + } // Codec compatibility: a node only fetches/accepts firmware it can actually apply. ESP32 A/B accepts // full images; nRF52 single-slot does not and disables them (it requires an in-place delta). A manual @@ -346,7 +358,10 @@ public: // This node's id (pubkey[0:4]), stamped into adverts we send so receivers can count distinct seeders. void set_seeder_id(const uint8_t* id4) { if (id4) for (int i = 0; i < 4; i++) _seeder_id[i] = id4[i]; } - void on_message(const uint8_t* msg, uint16_t len); // feed one received OTA message + // Feed one received OTA message. True means this node terminally consumed a + // bulk request/response, so Mesh must not echo it as if this node were an + // intermediate relay. + bool on_message(const uint8_t* msg, uint16_t len); void loop(); // drive fetch (re-request missing blocks) // Fast, non-blocking service called from the main radio loop. It admits at most one retained server // response or proof fallback per call; the send callback provides packet-pool/queue backpressure. @@ -358,7 +373,7 @@ public: void reset_session() { _fstate = IDLE; _have = 0; _req_count = 0; _mf_retries = 0; clearReassembly(); - _loop_last_have = 0; _loop_last_progress = pipelineProgress(); + _observed_path_transmissions = 0; _mf_total = 0; _mf_mask = 0; _mf_len = 0; _loop_last_mfmask = 0; freeLeaves(); _validate = false; _archive_fetch = false; _lv_retries = 0; _loop_last_lvmask = 0; @@ -371,6 +386,8 @@ public: uint32_t blocksTotal() const { return _fbc; } uint8_t fetchPipelineWidth() const { return _pipeline_width; } static constexpr uint8_t fetchPipelineCapacity() { return OTA_FETCH_PIPELINE; } + uint32_t fetchRetryTimeoutMs() const; + uint32_t proofGraceMs() const; const uint8_t* fetchManifestId() const { return _fid; } // --- discovery catalog (for `ota neighbors`): mOTAs heard around us via OTA_HAVE, deduped by mid --- @@ -412,10 +429,10 @@ private: bool beginLeafDiff(); // enter WANT_LEAVES if a seed diff is viable void diffStep(); // diff one batch of seed blocks per loop tick void freeLeaves(); // release the heap leaves buffer + diff state - void handleReq(const uint8_t* m, uint16_t n); - void handleData(const uint8_t* m, uint16_t n); - void handleReqProof(const uint8_t* m, uint16_t n); - void handleProof(const uint8_t* m, uint16_t n); + bool handleReq(const uint8_t* m, uint16_t n); + bool handleData(const uint8_t* m, uint16_t n); + bool handleReqProof(const uint8_t* m, uint16_t n); + bool handleProof(const uint8_t* m, uint16_t n); void startFetch(const uint8_t* mid, uint32_t target, bool validate = false); // begin/resume a fetch bool wantRow(const uint8_t* mid, uint32_t target, uint8_t codec, uint8_t flags) const; // fetch this row? void clearReassembly(); // forget every in-flight pipeline slot @@ -423,9 +440,11 @@ private: int findReassemblySlot(uint32_t block) const; int findEmptyReassemblySlot() const; bool blockInPipeline(uint32_t block) const; - void notePipelineBlockVerified(); void notePipelineStall(); - uint32_t pipelineProgress() const; + void finishFlight(); + uint8_t activePipelineSlots() const; + bool flightHasPendingData() const; + void noteFetchActivity() { _fetch_wait_since_ms = _now_ms; } uint32_t pickMissingBlock(); // choose the next missing block not already in flight int serveEntryIndex(const uint8_t* mid) const; // registry slot serving this mid (-1 if none) ServeView* resolve(const uint8_t* mid); // pick/load the ServeView for this mid (nullptr) @@ -455,8 +474,9 @@ private: void beginStagedVerification(); void verifyStagedStep(); bool requestSlot(uint8_t slot); // request DATA holes or the proof for one slot - bool fillPipeline(); // assign and request blocks until the window is full - void requestMissing(); // fill the window, or retry one stalled slot + bool requestFlight(); // send all newly assigned blocks in one OTA_REQ + bool fillPipeline(); // assign + send a new flight only when the old one ended + void requestMissing(); // start a flight, or recover one stalled slot uint32_t blockLen(uint32_t i) const; uint32_t _target = 0; @@ -507,8 +527,7 @@ private: uint32_t _resume_verify_idx = 0; bool _resume_invalidated = false; MerkleAccumulator _resume_merkle; - uint32_t _req_start = 0, _req_count = 0; // most recently requested block + active window size - uint32_t _loop_last_have = 0; // for stall detection in loop() + uint32_t _req_start = 0, _req_count = 0; // most recently requested block + active flight size uint32_t _desired_target = 0; // manual cross-target override (0 = auto / own target) uint8_t _desired_mid[4] = {0,0,0,0}; // pull a specific manifest_id (see want_mid) bool _have_desired_mid = false; @@ -523,9 +542,10 @@ private: uint16_t _advert_mins = OTA_ADVERT_INTERVAL_MINS; // beacon re-advertise cadence, minutes; 0=off (persisted) uint8_t _max_hops = OTA_HOP_LIMIT_DEFAULT; // OTA flood reach in hops; 0=direct only (persisted) uint8_t _fflags = 0; // flags of the manifest currently being fetched - // Bounded adaptive multi-block client pipeline. Each slot independently reassembles DATA and awaits its - // Merkle proof, so proof/flash work for one block can overlap radio delivery of the next. The live window - // grows on clean verified blocks and contracts after stalled service ticks; the wire protocol is unchanged. + // Bounded adaptive multi-block client flight. Each slot independently reassembles DATA and awaits its + // Merkle proof. One append-only OTA_REQ packet opens the whole flight; the receiver then stays quiet until + // every slot completes or an airtime-aware deadline expires. Clean flights grow 1->2->3->4; any recovery + // makes the next flight smaller. Logical/signed block geometry remains 1 KB. struct ReassemblySlot { uint32_t block = NO_BLOCK; uint16_t mask = 0; // received FRAG_DATA-slice bitmap @@ -535,11 +555,13 @@ private: uint8_t buf[OTA_MAX_BLOCK]; }; ReassemblySlot _reasm[OTA_FETCH_PIPELINE]; - uint32_t _loop_last_progress = 0; // aggregate pipeline progress for stall detection uint8_t _retry_slot = 0; // round-robin stalled-slot retry cursor uint8_t _pipeline_width = OTA_FETCH_PIPELINE_INITIAL; // live request window, adaptive up to array capacity - uint8_t _pipeline_success_streak = 0; // verified blocks since the last grow/stall decision - uint8_t _pipeline_stall_streak = 0; // consecutive no-progress service ticks + 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 + 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 // multi-fragment manifest reassembly (a signed v2 manifest exceeds one packet) uint8_t _mf_buf[OTA_MF_MAXFRAG * OTA_MF_FRAG]; // sized to the fragment cap so no valid manifest is silently dropped uint16_t _mf_retries = 0; // GET_MANIFEST retries while WANT_MANIFEST (give up after a cap) diff --git a/src/helpers/ota/OtaProtocol.cpp b/src/helpers/ota/OtaProtocol.cpp index 6a5d7afc..e2120bdd 100644 --- a/src/helpers/ota/OtaProtocol.cpp +++ b/src/helpers/ota/OtaProtocol.cpp @@ -110,6 +110,30 @@ bool decode_req(const uint8_t* buf, uint16_t len, ReqMsg& m) { return r.ok; } +uint16_t encode_req_window(uint8_t* buf, uint16_t cap, const ReqWindowMsg& m) { + if (m.n_items == 0 || m.n_items > OTA_REQ_MAX_ITEMS) return 0; + W w(buf, cap); w.u8(OTA_REQ); w.raw(m.manifest_id, 4); + for (uint8_t i = 0; i < m.n_items; i++) { + w.u16(m.items[i].block_idx); w.u16(m.items[i].want_mask); + } + return w.ok ? w.n : 0; +} + +bool decode_req_window(const uint8_t* buf, uint16_t len, ReqWindowMsg& m) { + // Five-byte prefix plus one or more complete four-byte rows. Requiring exact row framing keeps a + // truncated request from silently turning into a request for the wrong fragments. + if (len < 9 || ((len - 5) % 4) != 0) return false; + uint16_t count = (len - 5) / 4; + if (count == 0 || count > OTA_REQ_MAX_ITEMS) return false; + R r(buf, len); if (r.u8() != OTA_REQ) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + m.n_items = (uint8_t)count; + for (uint8_t i = 0; i < m.n_items; i++) { + m.items[i].block_idx = r.u16(); m.items[i].want_mask = r.u16(); + } + return r.ok && r.remaining() == 0; +} + uint16_t encode_data(uint8_t* buf, uint16_t cap, const DataMsg& m) { W w(buf, cap); w.u8(OTA_DATA); w.raw(m.manifest_id, 4); w.u16(m.block_idx); w.u16(m.frag_off); w.raw(m.data, m.data_len); diff --git a/src/helpers/ota/OtaProtocol.h b/src/helpers/ota/OtaProtocol.h index 0786c283..ccfd2d4d 100644 --- a/src/helpers/ota/OtaProtocol.h +++ b/src/helpers/ota/OtaProtocol.h @@ -63,12 +63,24 @@ struct ManifestMsg { const uint8_t* bytes; uint16_t len; }; -// ---- OTA_REQ: request specific fragments of ONE block (direct) ---- -// body: manifest_id(4) block_idx(2) want_mask(2). want_mask bit k = "send fragment k" (the slice at byte -// offset k*OTA_FRAG_DATA). A fetcher requests the full mask ((1< // AUTO-GENERATED by tools/mota/gen_targets.py - do not edit by hand. -// 540 OTA-capable PlatformIO envs. Maps target_id (= sha2-256:4 of the env name, LE uint32) +// 552 OTA-capable PlatformIO envs. Maps target_id (= sha2-256:4 of the env name, LE uint32) // to the human-readable env name, so a node/tool can name a target seen over the air WITHOUT // transmitting the string in the .mota / LoRa protocol. Regenerate when the OTA env set changes. +// Size-constrained receivers can set OTA_TARGET_NAME_TABLE=0. They still match targets by ID; +// OTA_LOCAL_TARGET_ID + OTA_LOCAL_TARGET_NAME optionally retain the local human-readable name. namespace mesh { namespace ota { -// target_id -> env name, or nullptr if unknown. Linear scan (table is small; lookups are rare). +#ifndef OTA_TARGET_NAME_TABLE + #define OTA_TARGET_NAME_TABLE 1 +#endif + +// target_id -> env name, or nullptr if unknown. Linear scan; lookups are rare. inline const char* ota_target_env_name(uint32_t target_id) { +#if OTA_TARGET_NAME_TABLE static const struct { uint32_t id; const char* env; } T[] = { { 0x149bcb23, "Ebyte_EoRa-S3_companion_radio_ble" }, { 0x30d4e442, "Ebyte_EoRa-S3_companion_radio_usb" }, @@ -264,6 +271,7 @@ inline const char* ota_target_env_name(uint32_t target_id) { { 0xed73ac03, "LilyGo_T-Echo-Lite_room_server" }, { 0xab0d550a, "LilyGo_T-Echo_Card_companion_radio_ble" }, { 0x1bd2a266, "LilyGo_T-Echo_Card_companion_radio_usb" }, + { 0x3655c748, "LilyGo_T-Echo_Card_kiss_modem" }, { 0x9a5ae8fc, "LilyGo_T-Echo_Card_repeater" }, { 0x1e9eb97a, "LilyGo_T-Echo_Card_room_server" }, { 0x139536f4, "LilyGo_T-Echo_companion_radio_ble" }, @@ -305,6 +313,7 @@ inline const char* ota_target_env_name(uint32_t target_id) { { 0xd0e2e616, "LilyGo_TDeck_repeater" }, { 0x3f9edfe3, "LilyGo_TETH_Elite_sx1262_companion_radio_ble" }, { 0xc23f8082, "LilyGo_TETH_Elite_sx1262_companion_radio_usb" }, + { 0x0e74ae16, "LilyGo_TETH_Elite_sx1262_kiss_modem" }, { 0x8f256e56, "LilyGo_TETH_Elite_sx1262_repeater" }, { 0xf1407f39, "LilyGo_TETH_Elite_sx1262_room_server" }, { 0xe90a3181, "LilyGo_Tlora_C6_companion_radio_ble_" }, @@ -341,6 +350,15 @@ inline const char* ota_target_env_name(uint32_t target_id) { { 0xcea95e20, "Meshadventurer_sx1268_terminal_chat" }, { 0xd45277ff, "Meshimi_companion_radio_ble_" }, { 0x2cad9a9d, "Meshimi_repeater_" }, + { 0x02a4a8c1, "meshnology_w12_companion_radio_ble" }, + { 0x35d960bd, "meshnology_w12_companion_radio_usb" }, + { 0x62977023, "meshnology_w12_companion_radio_wifi" }, + { 0x17a7ed8c, "meshnology_w12_kiss_modem" }, + { 0x07f5f5b5, "meshnology_w12_repeater" }, + { 0xfd0d4b8f, "meshnology_w12_repeater_bridge_espnow" }, + { 0xf5afb7dd, "meshnology_w12_room_server" }, + { 0x6ebf28cb, "meshnology_w12_sensor" }, + { 0x2d589527, "meshnology_w12_terminal_chat" }, { 0x77d0810c, "Minewsemi_me25ls01_repeater" }, { 0xf8259880, "Nano_G2_Ultra_repeater" }, { 0x063f1f00, "nibble_screen_connect_companion_radio_ble_" }, @@ -494,6 +512,7 @@ inline const char* ota_target_env_name(uint32_t target_id) { { 0x236f605c, "ThinkNode_M6_repeater" }, { 0x93994fed, "ThinkNode_M6_room_server" }, { 0x66c47cfd, "ThinkNode_M7_companion_radio_ble" }, + { 0x48061a01, "ThinkNode_M7_companion_radio_ethernet" }, { 0xa01688a7, "ThinkNode_M7_companion_radio_usb" }, { 0xa5f4a562, "ThinkNode_M7_companion_radio_wifi" }, { 0xef83222b, "ThinkNode_M7_kiss_modem" }, @@ -554,6 +573,11 @@ inline const char* ota_target_env_name(uint32_t target_id) { }; for (unsigned i = 0; i < sizeof(T) / sizeof(T[0]); i++) if (T[i].id == target_id) return T[i].env; +#elif defined(OTA_LOCAL_TARGET_ID) && defined(OTA_LOCAL_TARGET_NAME) + if (target_id == (uint32_t)(OTA_LOCAL_TARGET_ID)) return OTA_LOCAL_TARGET_NAME; +#else + (void)target_id; +#endif return nullptr; } diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index 64141440..b38948ed 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -484,6 +484,23 @@ TEST(OtaProtocol, CodecRoundTrips) { ReqMsg r2; ASSERT_TRUE(decode_req(buf, n, r2)); EXPECT_EQ(r2.block_idx, 7); EXPECT_EQ(r2.want_mask, 0x005f); + // A request window is an append-only extension: its first row is byte-for-byte a legacy ReqMsg, while + // new decoders see every requested block. This is the rolling-upgrade fallback for adaptive flights. + ReqWindowMsg rw{}; + memcpy(rw.manifest_id, rq.manifest_id, 4); + rw.n_items = 3; + rw.items[0] = {7, 0x005f}; rw.items[1] = {8, 0x007f}; rw.items[2] = {9, 0x0003}; + n = encode_req_window(buf, sizeof(buf), rw); + ASSERT_EQ(n, 17u); + ReqMsg legacy; + ASSERT_TRUE(decode_req(buf, n, legacy)); // an old source serves row zero + EXPECT_EQ(legacy.block_idx, 7); EXPECT_EQ(legacy.want_mask, 0x005f); + ReqWindowMsg rw2{}; + ASSERT_TRUE(decode_req_window(buf, n, rw2)); + ASSERT_EQ(rw2.n_items, 3u); + EXPECT_EQ(rw2.items[1].block_idx, 8); EXPECT_EQ(rw2.items[2].want_mask, 0x0003); + EXPECT_FALSE(decode_req_window(buf, n - 1, rw2)); // reject a truncated final row + // GET_LEAVES: bulk-fetch the target leaves[] with a fragment want_mask (motatool warm-start) GetLeavesMsg gl{{5,6,7,8}, 0x0007}; // want leaves fragments {0,1,2} n = encode_get_leaves(buf, sizeof(buf), gl); @@ -735,6 +752,44 @@ static void deliver_manifest_fragment(OtaManager& client, const uint8_t mid[4], client.on_message(wire, wire_len); } +static void deliver_verified_block(OtaManager& client, const MotaManifest& manifest, + uint32_t block, bool reverse_fragments = false) { + ASSERT_LT(block, manifest.block_count); + const uint32_t block_size = manifest.block_size(); + const uint32_t block_off = block * block_size; + const uint32_t block_len = block_off + block_size <= manifest.payload_size + ? block_size : manifest.payload_size - block_off; + const uint32_t fragments = (block_len + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA; + uint8_t wire[MAX_PACKET_PAYLOAD]; + for (uint32_t position = 0; position < fragments; position++) { + const uint32_t fragment = reverse_fragments ? fragments - position - 1 : position; + const uint32_t offset = fragment * OTA_FRAG_DATA; + uint32_t length = block_len - offset; + if (length > OTA_FRAG_DATA) length = OTA_FRAG_DATA; + DataMsg data; + memcpy(data.manifest_id, manifest.merkle_root, 4); + data.block_idx = (uint16_t)block; + data.frag_off = (uint16_t)offset; + data.data = manifest.payload + block_off + offset; + data.data_len = (uint16_t)length; + uint16_t wire_len = encode_data(wire, sizeof(wire), data); + ASSERT_GT(wire_len, 0); + ASSERT_TRUE(client.on_message(wire, wire_len)); + } + std::vector scratch(manifest.block_count * 4); + uint8_t siblings[32 * 4]; + uint8_t sibling_count = merkle_gen_proof( + manifest.leaves, manifest.block_count, block, scratch.data(), siblings); + ProofMsg proof; + memcpy(proof.manifest_id, manifest.merkle_root, 4); + proof.block_idx = (uint16_t)block; + proof.n_proof = sibling_count; + proof.proof = siblings; + uint16_t wire_len = encode_proof(wire, sizeof(wire), proof); + ASSERT_GT(wire_len, 0); + ASSERT_TRUE(client.on_message(wire, wire_len)); +} + TEST(OtaTransfer, ServerPacesOneKilobyteBlockAndProactiveProofWithBackpressure) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); @@ -752,8 +807,11 @@ TEST(OtaTransfer, ServerPacesOneKilobyteBlockAndProactiveProofWithBackpressure) uint8_t wire[MAX_PACKET_PAYLOAD]; uint16_t wire_len = encode_req(wire, sizeof(wire), request); ASSERT_GT(wire_len, 0); - server.on_message(wire, wire_len); - server.on_message(wire, wire_len); // identical in-flight retry merges + OtaManager relay; + relay.begin(0, capture_send, &sent); + EXPECT_FALSE(relay.on_message(wire, wire_len)); // an intermediate still forwards this request + EXPECT_TRUE(server.on_message(wire, wire_len)); + EXPECT_TRUE(server.on_message(wire, wire_len)); // identical in-flight retry merges EXPECT_EQ(server.pendingServeJobs(), 1u); EXPECT_TRUE(sent.items.empty()); // receive handler never allocates a packet burst @@ -782,6 +840,41 @@ TEST(OtaTransfer, ServerPacesOneKilobyteBlockAndProactiveProofWithBackpressure) EXPECT_EQ(server.pendingServeJobs(), 0u); } +TEST(OtaTransfer, ServerQueuesEveryBlockInOneRequestFlight) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); + ASSERT_EQ(manifest.block_count, 3u); + + OtaManager server; + CapturedMessages sent; + server.begin(0, capture_send, &sent); + ASSERT_TRUE(server.serve(SIM_MOTA_1K, SIM_MOTA_1K_LEN)); + + ReqWindowMsg request{}; + memcpy(request.manifest_id, manifest.merkle_root, 4); + request.n_items = 3; + for (uint8_t i = 0; i < request.n_items; i++) { + request.items[i].block_idx = i; + request.items[i].want_mask = 0xFFFF; + } + uint8_t wire[MAX_PACKET_PAYLOAD]; + uint16_t wire_len = encode_req_window(wire, sizeof(wire), request); + ASSERT_GT(wire_len, 0); + server.on_message(wire, wire_len); + EXPECT_EQ(server.pendingServeJobs(), 3u); + + // The bounded sender remains paced and drains the jobs in request order, including each proactive proof. + const uint32_t packets_per_full_block = + (manifest.block_size() + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA + 1; + for (uint32_t i = 0; i < packets_per_full_block; i++) server.serviceEgress(); + ASSERT_EQ(sent.items.size(), packets_per_full_block); + ProofMsg proof; + ASSERT_TRUE(decode_proof(sent.items.back().data(), + (uint16_t)sent.items.back().size(), proof)); + EXPECT_EQ(proof.block_idx, 0); + EXPECT_EQ(server.pendingServeJobs(), 2u); +} + TEST(OtaTransfer, MissingProactiveProofFallsBackAfterGrace) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); @@ -799,8 +892,7 @@ TEST(OtaTransfer, MissingProactiveProofFallsBackAfterGrace) { deliver_manifest_fragment(client, manifest.merkle_root, 1, manifest.manifest_start + OTA_MF_FRAG, (uint16_t)(MOTA_MFL - OTA_MF_FRAG)); - client.loop(); - ASSERT_EQ(sent.items.size(), 2u); + ASSERT_EQ(sent.items.size(), 1u); // first one-block flight opens immediately const uint32_t block_len = manifest.block_size(); uint8_t wire[MAX_PACKET_PAYLOAD]; @@ -817,24 +909,24 @@ TEST(OtaTransfer, MissingProactiveProofFallsBackAfterGrace) { ASSERT_GT(length_on_wire, 0); client.on_message(wire, length_on_wire); } - EXPECT_EQ(sent.items.size(), 2u); + EXPECT_EQ(sent.items.size(), 1u); client.set_clock(100 + OTA_PROOF_GRACE_MS - 1); client.serviceEgress(); - EXPECT_EQ(sent.items.size(), 2u); + EXPECT_EQ(sent.items.size(), 1u); client.set_clock(100 + OTA_PROOF_GRACE_MS); client.serviceEgress(); - ASSERT_EQ(sent.items.size(), 3u); + ASSERT_EQ(sent.items.size(), 2u); ReqProofMsg fallback; ASSERT_TRUE(decode_req_proof(sent.items.back().data(), (uint16_t)sent.items.back().size(), fallback)); EXPECT_EQ(fallback.block_idx, 0); } -TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) { +TEST(OtaTransfer, ClientUsesQuietBatchedFlightsAndAcceptsOutOfOrderBlocks) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); - ASSERT_GE(manifest.block_count, 3u); + ASSERT_EQ(manifest.block_count, 3u); OtaManager client; OtaStoreRam<4096> store; @@ -850,63 +942,85 @@ TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) { manifest.manifest_start + OTA_MF_FRAG, (uint16_t)(MOTA_MFL - OTA_MF_FRAG)); ASSERT_EQ(client.fetchState(), OtaManager::FETCHING); - client.set_clock(5000); - client.loop(); // opens the bounded request window + ASSERT_EQ(sent.items.size(), 1u); + ReqWindowMsg first{}; + ASSERT_TRUE(decode_req_window(sent.items[0].data(), + (uint16_t)sent.items[0].size(), first)); + ASSERT_EQ(first.n_items, 1u); + EXPECT_EQ(first.items[0].block_idx, 0); + EXPECT_EQ(client.fetchPipelineWidth(), 1u); + // A clean one-block probe grows the next flight to two blocks, both named in ONE request packet. + deliver_verified_block(client, manifest, 0); + ASSERT_EQ(client.blocksHave(), 1u); + ASSERT_EQ(client.fetchPipelineWidth(), 2u); ASSERT_EQ(sent.items.size(), 2u); - ReqMsg request0, request1; - ASSERT_TRUE(decode_req(sent.items[0].data(), (uint16_t)sent.items[0].size(), request0)); - ASSERT_TRUE(decode_req(sent.items[1].data(), (uint16_t)sent.items[1].size(), request1)); - EXPECT_EQ(request0.block_idx, 0); - EXPECT_EQ(request1.block_idx, 1); + ReqWindowMsg second{}; + ASSERT_TRUE(decode_req_window(sent.items[1].data(), + (uint16_t)sent.items[1].size(), second)); + ASSERT_EQ(second.n_items, 2u); + EXPECT_EQ(second.items[0].block_idx, 1); + EXPECT_EQ(second.items[1].block_idx, 2); - // Deliver block 1 first and its fragments in reverse order. Its independent slot must accept the - // proactive proof without spending a REQ_PROOF round trip or disturbing block 0's reserved slot. - const uint32_t block = 1; - const uint32_t block_size = manifest.block_size(); - const uint32_t block_len = block + 1 < manifest.block_count - ? block_size : manifest.payload_size - block * block_size; - const uint32_t fragments = (block_len + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA; - uint8_t wire[MAX_PACKET_PAYLOAD]; - for (uint32_t k = fragments; k-- > 0;) { - const uint32_t offset = k * OTA_FRAG_DATA; - uint32_t length = block_len - offset; - if (length > OTA_FRAG_DATA) length = OTA_FRAG_DATA; - DataMsg data; - memcpy(data.manifest_id, manifest.merkle_root, 4); - data.block_idx = block; - data.frag_off = (uint16_t)offset; - data.data = manifest.payload + block * block_size + offset; - data.data_len = (uint16_t)length; - uint16_t wire_len = encode_data(wire, sizeof(wire), data); - ASSERT_GT(wire_len, 0); - client.on_message(wire, wire_len); - } + // Complete block 2 first, with reversed fragments. Its slot verifies independently, but the client does + // not refill or transmit anything while block 1's response is still expected from the same flight. + deliver_verified_block(client, manifest, 2, true); + EXPECT_EQ(client.blocksHave(), 2u); + EXPECT_EQ(sent.items.size(), 2u); - ASSERT_EQ(sent.items.size(), 2u); // no immediate proof request + deliver_verified_block(client, manifest, 1); + EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE); + EXPECT_EQ(client.blocksHave(), 3u); + EXPECT_EQ(sent.items.size(), 2u); // no continuous-refill request was emitted +} - uint8_t scratch[SIM_MOTA_1K_BLOCKS * 4]; - uint8_t siblings[32 * 4]; - uint8_t sibling_count = merkle_gen_proof( - manifest.leaves, manifest.block_count, block, scratch, siblings); - ProofMsg proof; - memcpy(proof.manifest_id, manifest.merkle_root, 4); - proof.block_idx = block; - proof.n_proof = sibling_count; - proof.proof = siblings; - uint16_t wire_len = encode_proof(wire, sizeof(wire), proof); - ASSERT_GT(wire_len, 0); - client.on_message(wire, wire_len); +TEST(OtaTransfer, NewClientFallsBackWhenLegacySourceServesOnlyFirstWindowRow) { + MotaManifest manifest; + ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); + ASSERT_EQ(manifest.block_count, 3u); - EXPECT_EQ(client.blocksHave(), 1u); - ASSERT_EQ(sent.items.size(), 3u); // freed slot was immediately refilled - ReqMsg refill; - ASSERT_TRUE(decode_req(sent.items[2].data(), (uint16_t)sent.items[2].size(), refill)); - EXPECT_EQ(refill.block_idx, 2); + OtaManager client; + OtaStoreRam<4096> store; + CapturedMessages sent; + client.begin(SIM_TARGET_ID, capture_send, &sent); + client.set_fetch_store(&store); + client.set_clock(100); + client.pull(manifest.merkle_root, manifest.target_id); + sent.items.clear(); + 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)); + deliver_verified_block(client, manifest, 0); + ASSERT_EQ(sent.items.size(), 2u); + + // Legacy decode sees and serves only row zero (block 1) from the new two-row request. + ReqMsg legacy{}; + ASSERT_TRUE(decode_req(sent.items.back().data(), + (uint16_t)sent.items.back().size(), legacy)); + ASSERT_EQ(legacy.block_idx, 1); + deliver_verified_block(client, manifest, legacy.block_idx); + ASSERT_EQ(client.blocksHave(), 2u); + ASSERT_EQ(sent.items.size(), 2u); // block 2 remains reserved; no immediate refill + + const uint32_t timeout = client.fetchRetryTimeoutMs(); + client.set_clock(100 + timeout); + client.loop(); // tail row was not served: recover it conventionally + ASSERT_EQ(sent.items.size(), 3u); + EXPECT_EQ(sent.items.back().size(), 9u); // exact legacy single-row OTA_REQ + ReqMsg fallback{}; + ASSERT_TRUE(decode_req(sent.items.back().data(), + (uint16_t)sent.items.back().size(), fallback)); + EXPECT_EQ(fallback.block_idx, 2); + + deliver_verified_block(client, manifest, 2); + EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE); + EXPECT_EQ(client.blocksHave(), 3u); } #if OTA_FETCH_PIPELINE >= 3 -TEST(OtaTransfer, RequestWindowGrowsOnCleanBlocksAndShrinksOnStalls) { +TEST(OtaTransfer, RequestFlightGrowsOnCleanFlightsAndHalvesAfterRecovery) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, manifest)); ASSERT_GE(manifest.block_count, 7u); @@ -916,6 +1030,7 @@ TEST(OtaTransfer, RequestWindowGrowsOnCleanBlocksAndShrinksOnStalls) { CapturedMessages sent; client.begin(SIM_TARGET_ID, capture_send, &sent); client.set_fetch_store(&store); + client.set_clock(100); client.pull(manifest.merkle_root, manifest.target_id); ASSERT_EQ(client.fetchPipelineCapacity(), 4u); sent.items.clear(); @@ -924,59 +1039,53 @@ TEST(OtaTransfer, RequestWindowGrowsOnCleanBlocksAndShrinksOnStalls) { deliver_manifest_fragment(client, manifest.merkle_root, 1, manifest.manifest_start + OTA_MF_FRAG, (uint16_t)(MOTA_MFL - OTA_MF_FRAG)); - client.loop(); + ASSERT_EQ(client.fetchPipelineWidth(), 1u); + ASSERT_EQ(sent.items.size(), 1u); + + deliver_verified_block(client, manifest, 0); // clean width-1 flight -> width 2 ASSERT_EQ(client.fetchPipelineWidth(), 2u); + ASSERT_EQ(sent.items.size(), 2u); + ReqWindowMsg width2{}; + ASSERT_TRUE(decode_req_window(sent.items.back().data(), + (uint16_t)sent.items.back().size(), width2)); + ASSERT_EQ(width2.n_items, 2u); - auto deliver_verified_block = [&](uint32_t block) { - const uint32_t block_size = manifest.block_size(); - const uint32_t block_len = block + 1 < manifest.block_count - ? block_size : manifest.payload_size - block * block_size; - uint8_t wire[MAX_PACKET_PAYLOAD]; - for (uint32_t offset = 0; offset < block_len; offset += OTA_FRAG_DATA) { - uint32_t length = block_len - offset; - if (length > OTA_FRAG_DATA) length = OTA_FRAG_DATA; - DataMsg data; - memcpy(data.manifest_id, manifest.merkle_root, 4); - data.block_idx = block; - data.frag_off = (uint16_t)offset; - data.data = manifest.payload + block * block_size + offset; - data.data_len = (uint16_t)length; - uint16_t wire_len = encode_data(wire, sizeof(wire), data); - ASSERT_GT(wire_len, 0); - client.on_message(wire, wire_len); - } + deliver_verified_block(client, manifest, 1); + EXPECT_EQ(client.fetchPipelineWidth(), 2u); // still waiting for the same flight's block 2 + EXPECT_EQ(sent.items.size(), 2u); // receiver remains silent + deliver_verified_block(client, manifest, 2); // clean width-2 flight -> width 3 + EXPECT_EQ(client.fetchPipelineWidth(), 3u); + ASSERT_EQ(sent.items.size(), 3u); + ReqWindowMsg width3{}; + ASSERT_TRUE(decode_req_window(sent.items.back().data(), + (uint16_t)sent.items.back().size(), width3)); + ASSERT_EQ(width3.n_items, 3u); + EXPECT_EQ(width3.items[0].block_idx, 3); - std::vector scratch(manifest.block_count * 4); - uint8_t siblings[32 * 4]; - uint8_t sibling_count = merkle_gen_proof( - manifest.leaves, manifest.block_count, block, scratch.data(), siblings); - ProofMsg proof; - memcpy(proof.manifest_id, manifest.merkle_root, 4); - proof.block_idx = block; - proof.n_proof = sibling_count; - proof.proof = siblings; - uint16_t wire_len = encode_proof(wire, sizeof(wire), proof); - ASSERT_GT(wire_len, 0); - client.on_message(wire, wire_len); - }; - - for (uint32_t block = 0; block < 3; block++) { - deliver_verified_block(block); - EXPECT_EQ(client.fetchPipelineWidth(), 2u); - } - deliver_verified_block(3); - EXPECT_EQ(client.blocksHave(), 4u); + // Ordinary one-second maintenance ticks do nothing until the flight's calculated service time expires. + const uint32_t timeout = client.fetchRetryTimeoutMs(); + client.set_clock(100 + timeout - 1); + client.loop(); + EXPECT_EQ(sent.items.size(), 3u); EXPECT_EQ(client.fetchPipelineWidth(), 3u); - client.loop(); // establish the new committed-block/progress baseline - client.loop(); // first no-progress service tick - EXPECT_EQ(client.fetchPipelineWidth(), 3u); - client.loop(); // second no-progress service tick contracts the request window + client.set_clock(100 + timeout); + client.loop(); // recover one slot with a legacy single-row REQ + ASSERT_EQ(sent.items.size(), 4u); + ReqMsg recovery{}; + ASSERT_TRUE(decode_req(sent.items.back().data(), + (uint16_t)sent.items.back().size(), recovery)); + EXPECT_EQ(recovery.block_idx, 3); + EXPECT_EQ(client.fetchPipelineWidth(), 3u); // resize only after this flight drains + + deliver_verified_block(client, manifest, 3); + deliver_verified_block(client, manifest, 4); + deliver_verified_block(client, manifest, 5); EXPECT_EQ(client.fetchPipelineWidth(), 2u); } #endif -TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) { +TEST(OtaTransfer, FlightRetriesOnlyMissingFragmentsAfterItsDeadline) { MotaManifest manifest; ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest)); @@ -985,6 +1094,7 @@ TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) { CapturedMessages sent; client.begin(SIM_TARGET_ID, capture_send, &sent); client.set_fetch_store(&store); + client.set_clock(100); client.pull(manifest.merkle_root, manifest.target_id); sent.items.clear(); deliver_manifest_fragment(client, manifest.merkle_root, 0, @@ -992,8 +1102,7 @@ TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) { deliver_manifest_fragment(client, manifest.merkle_root, 1, manifest.manifest_start + OTA_MF_FRAG, (uint16_t)(MOTA_MFL - OTA_MF_FRAG)); - client.loop(); - ASSERT_EQ(sent.items.size(), 2u); // initial requests for blocks 0 and 1 + ASSERT_EQ(sent.items.size(), 1u); // conservative one-block probe flight DataMsg first_fragment; memcpy(first_fragment.manifest_id, manifest.merkle_root, 4); @@ -1006,21 +1115,56 @@ TEST(OtaTransfer, PipelineRetriesOneSlotAndOnlyItsMissingFragments) { ASSERT_GT(wire_len, 0); client.on_message(wire, wire_len); - client.loop(); // observes progress; no premature retry - EXPECT_EQ(sent.items.size(), 2u); - client.loop(); // stalled: retry slot 0 only - ASSERT_EQ(sent.items.size(), 3u); + const uint32_t timeout = client.fetchRetryTimeoutMs(); + client.set_clock(100 + timeout - 1); + client.loop(); // no premature fixed-tick retry + EXPECT_EQ(sent.items.size(), 1u); + client.set_clock(100 + timeout); + client.loop(); // deadline: retry only block 0's holes + ASSERT_EQ(sent.items.size(), 2u); ReqMsg retry0; - ASSERT_TRUE(decode_req(sent.items[2].data(), (uint16_t)sent.items[2].size(), retry0)); + ASSERT_TRUE(decode_req(sent.items[1].data(), (uint16_t)sent.items[1].size(), retry0)); EXPECT_EQ(retry0.block_idx, 0); EXPECT_EQ(retry0.want_mask, (uint16_t)(0x007F & ~0x0001)); - client.loop(); // next stalled tick rotates to slot 1 - ASSERT_EQ(sent.items.size(), 4u); - ReqMsg retry1; - ASSERT_TRUE(decode_req(sent.items[3].data(), (uint16_t)sent.items[3].size(), retry1)); - EXPECT_EQ(retry1.block_idx, 1); - EXPECT_EQ(retry1.want_mask, 0x007F); + client.set_clock(100 + timeout * 2 - 1); + client.loop(); // the recovery request restarted the deadline + EXPECT_EQ(sent.items.size(), 2u); +} + +TEST(OtaTransfer, RetryTimingAdaptsToRadioPathAndAirtimeBudget) { + 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); + client.set_clock(100); + client.set_max_hops(0); + client.set_link_timing(100, 2000); // direct response fits the conservative quiet floor + client.pull(manifest.merkle_root, manifest.target_id); + sent.items.clear(); + deliver_manifest_fragment(client, manifest.merkle_root, 0, + manifest.manifest_start, OTA_MF_FRAG); + deliver_manifest_fragment(client, manifest.merkle_root, 1, + manifest.manifest_start + OTA_MF_FRAG, + (uint16_t)(MOTA_MFL - OTA_MF_FRAG)); + const uint32_t fast_direct = client.fetchRetryTimeoutMs(); + EXPECT_GE(fast_direct, (uint32_t)OTA_FETCH_RETRY_MIN_MS); + + client.set_max_hops(2); // no observed reply path: budget source + 2 relays + const uint32_t fast_relayed = client.fetchRetryTimeoutMs(); + EXPECT_GT(fast_relayed, fast_direct); + + client.set_link_timing(200, 2000); // half the bandwidth doubles packet airtime + const uint32_t slower_radio = client.fetchRetryTimeoutMs(); + EXPECT_GT(slower_radio, fast_relayed); + + client.set_link_timing(200, 3000); // a 1/3-duty node needs more service time + EXPECT_GT(client.fetchRetryTimeoutMs(), slower_radio); + EXPECT_GT(client.proofGraceMs(), (uint32_t)OTA_PROOF_GRACE_MS); } TEST(OtaTransfer, RejectsShortNonFinalManifestFragment) { diff --git a/tools/lora_ota/lora_ota.py b/tools/lora_ota/lora_ota.py index c899585c..9a59d224 100755 --- a/tools/lora_ota/lora_ota.py +++ b/tools/lora_ota/lora_ota.py @@ -290,6 +290,106 @@ def adaptive_poll_ceiling(baseline_seconds: float) -> float: return max(baseline_seconds, baseline_seconds * ADAPTIVE_POLL_MAX_FACTOR) +def lora_airtime_seconds( + payload_bytes: int, + bandwidth_khz: float, + spreading_factor: int, + coding_rate: int, + *, + preamble_symbols: int = 16, +) -> float: + """Estimate explicit-header LoRa airtime for one Mesh packet.""" + symbol_seconds = (2 ** spreading_factor) / (bandwidth_khz * 1000.0) + low_data_rate = 1 if symbol_seconds >= 0.016 else 0 + numerator = ( + 8 * payload_bytes - 4 * spreading_factor + 28 + 16 + ) + denominator = 4 * (spreading_factor - 2 * low_data_rate) + payload_symbols = 8 + max( + math.ceil(numerator / denominator) * coding_rate, + 0, + ) + return (preamble_symbols + 4.25 + payload_symbols) * symbol_seconds + + +def ota_path_transmissions(args: argparse.Namespace) -> int: + # The source always transmits once; every explicitly managed relay adds + # one forwarding transmission to the response train. + return 1 + len(getattr(args, "relay", []) or []) + + +def initial_status_wait_seconds( + args: argparse.Namespace, + package: MotaInfo, +) -> float: + """Choose a quiet first-poll window from package and TempRadio airtime.""" + baseline = float(args.poll_seconds) + temp_values = getattr(args, "temp_values", None) + if not temp_values: + return baseline + _frequency, bandwidth, sf, cr, _minutes = temp_values + blocks = math.ceil(package.payload_size / package.block_size) + fragments = math.ceil(package.block_size / 160) + # DATA dominates. Treat proofs and the roughly one request per three + # adaptive blocks as maximum-size packets, then apply a measured 4x guard + # for CAD, RX/TX turnarounds, flash commits, and scheduler latency. + packets = blocks * (fragments + 1) + math.ceil(blocks / 3) + estimated = ( + packets + * lora_airtime_seconds(184, bandwidth, sf, cr) + * ota_path_transmissions(args) + * 4.0 + ) + return min(adaptive_poll_ceiling(baseline), max(baseline, estimated)) + + +def transfer_tail_guard_seconds(args: argparse.Namespace) -> float: + """Drain the last loaded block after the host-read progress signal.""" + temp_values = getattr(args, "temp_values", None) + if not temp_values: + return 3.0 + _frequency, bandwidth, sf, cr, _minutes = temp_values + return max( + 2.0, + 10 + * lora_airtime_seconds(184, bandwidth, sf, cr) + * ota_path_transmissions(args) + * 4.0, + ) + + +def passive_progress_stall_seconds( + args: argparse.Namespace, + package: MotaInfo, +) -> float: + """Return the no-progress interval that justifies an OTA status packet. + + A healthy transfer can legitimately spend a long time transmitting the + fragments for one block before it asks the host for another. Scale that + interval with the selected LoRa modulation and managed relay count so a + slow/two-hop deployment is not mistaken for a stall. Fast TempRadio links + retain a 20-second floor, which is long enough to cover scheduler and flash + jitter without hiding a genuinely lost pull indefinitely. + """ + baseline = float(args.poll_seconds) + temp_values = getattr(args, "temp_values", None) + if not temp_values: + return max(20.0, baseline) + _frequency, bandwidth, sf, cr, _minutes = temp_values + fragments = math.ceil(package.block_size / 160) + one_block_packets = fragments + 2 # request/proof plus DATA fragments + estimated = ( + one_block_packets + * lora_airtime_seconds(184, bandwidth, sf, cr) + * ota_path_transmissions(args) + * 6.0 + ) + return min( + max(20.0, adaptive_poll_ceiling(baseline)), + max(20.0, estimated), + ) + + def format_decimal(value: float) -> str: return f"{value:.6f}".rstrip("0").rstrip(".") @@ -1710,6 +1810,33 @@ class SeederProcess: f"{self._log_tail()}" ) + def payload_read_progress(self, package: MotaInfo) -> tuple[int, int, int]: + """Return unique, total, and aggregate payload-block host reads. + + ``motatool serve -v`` logs each successful random-access read. A read + at a payload block boundary means the source has reached that block's + response job. This is a passive progress signal: unlike remote + ``ota status``, reading it consumes no LoRa airtime and cannot collide + with the transfer it is observing. + """ + if self.log_file is not None and not self.log_file.closed: + self.log_file.flush() + try: + detail = self.log_path.read_text(encoding="utf-8", errors="replace") + except OSError: + return 0, math.ceil(package.payload_size / package.block_size) + offsets = [ + int(match) + for match in re.findall(r"\bREAD\s+\d+\s+@(\d+)\s+OK\b", detail) + ] + total = math.ceil(package.payload_size / package.block_size) + boundaries = { + package.payload_offset + index * package.block_size + for index in range(total) + } + payload_reads = [offset for offset in offsets if offset in boundaries] + return len(set(payload_reads)), total, len(payload_reads) + def stop(self) -> None: if self.process is None: return @@ -1977,6 +2104,76 @@ def monitor_download( ) -> str: deadline = time.monotonic() + args.transfer_timeout_minutes * 60 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 + # measured adding tens of seconds to an otherwise short transfer. First + # watch the local seeder log, which is airtime-free, and query only after + # every payload block has reached the source (plus a bounded final-packet + # drain). If verbose progress is unavailable, fall back to an airtime-based + # quiet window before the first query. + first_wait = initial_status_wait_seconds(args, package) + ensure_seeder_running(seeder, "before first transfer status check") + if seeder is not None and hasattr(seeder, "payload_read_progress"): + stall_wait = passive_progress_stall_seconds(args, package) + tail_wait = transfer_tail_guard_seconds(args) + last_seen = -1 + last_activity = -1 + last_progress = time.monotonic() + last_status_activity = -1 + last_reported_quarter = -1 + while time.monotonic() < deadline: + ensure_seeder_running(seeder, "before first transfer status check") + seen, total, activity = seeder.payload_read_progress(package) + now = time.monotonic() + if activity > last_activity: + last_activity = activity + last_progress = now + if seen > last_seen: + last_seen = seen + quarter = (seen * 4 // total) if total else 0 + if seen and quarter > last_reported_quarter: + print(f"[download] passive source progress {seen}/{total}") + last_reported_quarter = quarter + quiet_for = now - last_progress + query_reason: str | None = None + if ( + total > 0 + and seen >= total + and activity > last_status_activity + and quiet_for >= tail_wait + ): + query_reason = ( + f"source read all {total} payload blocks and has been quiet " + f"for {quiet_for:.1f}s" + ) + elif quiet_for >= stall_wait: + query_reason = f"no new source block read for {quiet_for:.0f}s" + + if query_reason is not None: + print( + f"[download] {query_reason}; checking destination status" + ) + status = remote_command_with_seeder( + controller, args.target, "ota status", seeder, + "during transfer", + ) + require_package_session(status, package) + if "ready to install" in status.lower(): + return status + # A status exchange can overlap a useful retry read. Snapshot + # it on the next loop and remain passive until that activity + # drains. If nothing else moves, use the radio-scaled stall + # interval before the next diagnostic query. + last_status_activity = activity + last_progress = time.monotonic() + continue + time.sleep(min(1.0, max(0.0, deadline - time.monotonic()))) + raise OtaError( + "transfer timeout; the partial download remains staged and can " + "resume when the same mOTA is served again" + ) + + time.sleep(min(first_wait, max(0.0, deadline - time.monotonic()))) while time.monotonic() < deadline: query_started = time.monotonic() status = remote_command_with_seeder( @@ -2199,15 +2396,17 @@ def request_install( def shorten_relay_temp_windows( controller: Controller, args: argparse.Namespace, + relay_values: list[tuple[str, str]] | None = None, ) -> None: - if not args.relay_values: + values = args.relay_values if relay_values is None else relay_values + if not values: return command = temp_radio_command_for_minutes(args, TEMP_RADIO_RETURN_MINUTES) print( f"[relays] scheduling return to the normal channel in " f"{TEMP_RADIO_RETURN_MINUTES} minute" ) - for relay_name, relay_password in args.relay_values: + for relay_name, relay_password in values: reply = controller.remote_command( relay_name, command, password=relay_password ) @@ -2744,10 +2943,13 @@ def main( work_dir: Path | None = None controller: Controller | None = controller_override original_radio: RadioSettings | None = None + temp_radio: RadioSettings | None = None controller_changed = False seeder: SeederProcess | None = None seeder_attempted = False source_temp_owned = False + target_temp_owned = False + armed_relay_values: list[tuple[str, str]] = [] password = args.password or os.environ.get("MESHCORE_ADMIN_PASSWORD", "") try: preflight_inputs(args) @@ -2815,11 +3017,13 @@ def main( arm_target_temp_radio( controller, args, temp_command, temp_radio, original_radio ) + target_temp_owned = True for relay_name, relay_password in args.relay_values: temp_reply = controller.remote_command( relay_name, temp_command, password=relay_password ) require_temp_radio_reply(relay_name, temp_reply) + armed_relay_values.append((relay_name, relay_password)) if not args.source_already_temp and not args.source_shares_controller: source_cli_command(args, temp_command) source_temp_owned = True @@ -2850,7 +3054,9 @@ def main( ) else: shorten_target_temp_window(controller, args) + target_temp_owned = False shorten_relay_temp_windows(controller, args) + armed_relay_values.clear() seeder.stop() seeder = None if source_temp_owned: @@ -2859,6 +3065,9 @@ def main( return 0 install_confirmed = request_install(controller, args, package) + # The install path replaces the long download lease with a bounded + # three-minute window, then either reboots or is safely self-restoring. + target_temp_owned = False if install_confirmed: print(f"[install] {args.target} accepted the image and is rebooting") else: @@ -2871,6 +3080,7 @@ def main( # the relays are no longer needed on TempRadio. Give each a short, # bounded window, then verify through the restored normal route. shorten_relay_temp_windows(controller, args) + armed_relay_values.clear() # Stop seeding before returning the controller to its ordinary channel. seeder.stop() @@ -2922,6 +3132,51 @@ def main( # this is harmless on TCP and explicitly cleans the serial case. if seeder_attempted and args.source_serial: source_cli_command(args, "ota folder off", check=False) + needs_remote_cleanup = target_temp_owned or bool(armed_relay_values) + if ( + needs_remote_cleanup + and not args.leave_controller_radio + and controller is not None + and original_radio is not None + and temp_radio is not None + ): + if not controller_changed: + try: + controller.set_radio( + temp_radio, + "return controller to TempRadio for failure cleanup", + ) + controller_changed = True + time.sleep(TEMP_RADIO_SWITCH_DELAY_SECONDS) + except (OtaError, OSError) as exc: + print( + "WARNING: could not reach TempRadio to shorten remote " + f"leases; their configured windows remain bounded: {exc}", + file=sys.stderr, + ) + if controller_changed: + if target_temp_owned: + try: + shorten_target_temp_window(controller, args) + target_temp_owned = False + except (OtaError, OSError) as exc: + print( + "WARNING: could not shorten the destination " + f"TempRadio lease: {exc}", + file=sys.stderr, + ) + if armed_relay_values: + try: + shorten_relay_temp_windows( + controller, args, armed_relay_values + ) + armed_relay_values.clear() + except (OtaError, OSError) as exc: + print( + "WARNING: could not shorten every relay TempRadio " + f"lease: {exc}", + file=sys.stderr, + ) if source_temp_owned and not args.leave_controller_radio: shorten_source_temp_window(args, check=False) if ( diff --git a/tools/lora_ota/test_lora_ota.py b/tools/lora_ota/test_lora_ota.py index 2cba5150..d367350b 100644 --- a/tools/lora_ota/test_lora_ota.py +++ b/tools/lora_ota/test_lora_ota.py @@ -490,7 +490,10 @@ class DownloadSessionTests(unittest.TestCase): args = argparse.Namespace( target="remote", transfer_timeout_minutes=1, poll_seconds=1, ) - with self.assertRaisesRegex(ota.OtaError, "switched to mOTA"): + with ( + mock.patch.object(ota.time, "sleep"), + self.assertRaisesRegex(ota.OtaError, "switched to mOTA"), + ): ota.monitor_download(controller, args, self.package) def test_monitor_accepts_only_the_expected_ready_session(self) -> None: @@ -500,8 +503,10 @@ class DownloadSessionTests(unittest.TestCase): args = argparse.Namespace( target="remote", transfer_timeout_minutes=1, poll_seconds=1, ) - status = ota.monitor_download(controller, args, self.package) + with mock.patch.object(ota.time, "sleep") as sleep: + status = ota.monitor_download(controller, args, self.package) self.assertIn(self.package.manifest_id, status) + sleep.assert_called_once_with(1.0) def test_monitor_stops_immediately_when_seeder_exits(self) -> None: controller = self.Controller([]) @@ -517,6 +522,40 @@ class DownloadSessionTests(unittest.TestCase): ota.monitor_download(controller, args, self.package, Seeder()) self.assertEqual(controller.commands, []) + def test_monitor_uses_advancing_source_log_without_airtime_queries(self) -> None: + controller = self.Controller([ + f"OTA | download: ready to install 9/9 id={self.package.manifest_id} 2s" + ]) + args = argparse.Namespace( + target="remote", transfer_timeout_minutes=1, poll_seconds=1, + ) + total = ( + self.package.payload_size + self.package.block_size - 1 + ) // self.package.block_size + + class Seeder: + calls = 0 + + def ensure_running(self, _context: str) -> None: + return None + + def payload_read_progress( + self, _package: ota.MotaInfo + ) -> tuple[int, int, int]: + self.calls += 1 + progress = min(self.calls, total) + return progress, total, progress + + seeder = Seeder() + with ( + mock.patch.object(ota.time, "sleep"), + mock.patch.object(ota, "transfer_tail_guard_seconds", return_value=0), + ): + status = ota.monitor_download(controller, args, self.package, seeder) + self.assertIn(self.package.manifest_id, status) + self.assertEqual(controller.commands, ["ota status"]) + self.assertEqual(seeder.calls, total) + def test_lost_pull_reply_is_resolved_without_blind_replay(self) -> None: controller = self.Controller([ "OTA | no download", @@ -639,6 +678,51 @@ class ReliabilityTests(unittest.TestCase): ) self.assertEqual(ota.adaptive_poll_ceiling(60), 180) + def test_first_status_wait_scales_with_radio_and_relay_airtime(self) -> None: + package = ota.parse_mota(mota_blob(firmware(b"A" * 50_000, VERSION_NEW))) + direct = argparse.Namespace( + poll_seconds=30, + temp_values=(909.95, 500.0, 5, 5, 120), + relay=[], + ) + relayed = argparse.Namespace( + poll_seconds=30, + temp_values=(909.95, 62.5, 7, 5, 120), + relay=["relay"], + ) + self.assertGreaterEqual(ota.initial_status_wait_seconds(direct, package), 30) + self.assertGreater( + ota.initial_status_wait_seconds(relayed, package), + ota.initial_status_wait_seconds(direct, package), + ) + self.assertLessEqual( + ota.initial_status_wait_seconds(relayed, package), + ota.adaptive_poll_ceiling(30), + ) + self.assertGreater( + ota.passive_progress_stall_seconds(relayed, package), + ota.passive_progress_stall_seconds(direct, package), + ) + + def test_seeder_progress_counts_only_payload_block_boundaries(self) -> None: + package = ota.parse_mota(mota_blob(firmware(b"B" * 2500, VERSION_NEW))) + with tempfile.TemporaryDirectory() as directory: + log = Path(directory) / "serve.log" + expected = [ + package.payload_offset + index * package.block_size + for index in range(3) + ] + log.write_text( + "[dev] READ 0 @8 OK\n" + f"[dev] READ 0 @{expected[0]} OK\n" + f"[dev] READ 0 @{expected[2]} OK\n", + encoding="utf-8", + ) + seeder = object.__new__(ota.SeederProcess) + seeder.log_path = log + seeder.log_file = None + self.assertEqual(seeder.payload_read_progress(package), (2, 3, 2)) + def test_marked_output_excludes_queued_messages(self) -> None: controller = object.__new__(ota.Controller) controller._execute = lambda _commands, _label: subprocess.CompletedProcess( diff --git a/tools/mota/gen_targets.py b/tools/mota/gen_targets.py index d30267fc..e9e3c387 100644 --- a/tools/mota/gen_targets.py +++ b/tools/mota/gen_targets.py @@ -65,11 +65,18 @@ def main(): f"// {len(rows)} OTA-capable PlatformIO envs. Maps target_id (= sha2-256:4 of the env name, LE uint32)", "// to the human-readable env name, so a node/tool can name a target seen over the air WITHOUT", "// transmitting the string in the .mota / LoRa protocol. Regenerate when the OTA env set changes.", + "// Size-constrained receivers can set OTA_TARGET_NAME_TABLE=0. They still match targets by ID;", + "// OTA_LOCAL_TARGET_ID + OTA_LOCAL_TARGET_NAME optionally retain the local human-readable name.", "", "namespace mesh { namespace ota {", "", - "// target_id -> env name, or nullptr if unknown. Linear scan (table is small; lookups are rare).", + "#ifndef OTA_TARGET_NAME_TABLE", + " #define OTA_TARGET_NAME_TABLE 1", + "#endif", + "", + "// target_id -> env name, or nullptr if unknown. Linear scan; lookups are rare.", "inline const char* ota_target_env_name(uint32_t target_id) {", + "#if OTA_TARGET_NAME_TABLE", " static const struct { uint32_t id; const char* env; } T[] = {", ] out += [f' {{ 0x{tid:08x}, "{e}" }},' for tid, e in rows] @@ -77,6 +84,11 @@ def main(): " };", " for (unsigned i = 0; i < sizeof(T) / sizeof(T[0]); i++)", " if (T[i].id == target_id) return T[i].env;", + "#elif defined(OTA_LOCAL_TARGET_ID) && defined(OTA_LOCAL_TARGET_NAME)", + " if (target_id == (uint32_t)(OTA_LOCAL_TARGET_ID)) return OTA_LOCAL_TARGET_NAME;", + "#else", + " (void)target_id;", + "#endif", " return nullptr;", "}", "", diff --git a/variants/rak3401/platformio.ini b/variants/rak3401/platformio.ini index 877a1b6a..aa1cd100 100644 --- a/variants/rak3401/platformio.ini +++ b/variants/rak3401/platformio.ini @@ -64,7 +64,10 @@ build_flags = -D ENABLE_OTA=1 -D OTA_FLASH_STORE=1 -D OTA_FOLDER_SERIAL - -D OTA_FETCH_PIPELINE=4 ; adapt from 2 to 4 concurrent block requests on a clean OTA link + -D OTA_TARGET_NAME_TABLE=0 ; IDs remain authoritative; omit the 540-entry display-name table + -D OTA_LOCAL_TARGET_ID=0x2fa509c1 + -D OTA_LOCAL_TARGET_NAME='"RAK_3401_repeater_lora_ota_no_external_sensors"' + -D OTA_FETCH_PIPELINE=4 ; adaptive one-packet flights grow 1 -> 2 -> 3 -> 4 on a clean link build_src_filter = ${rak3401.build_src_filter} + +