dandri/HaloKeymind
1
0 Fork 0
mirror of https://github.com/mikecarper/MeshCore.git synced 2026-10-22 15:23:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Prioritize and pace private TempRadio OTA

Browse Source
This commit is contained in:
mikecarper committed 2026-08-13 17:53:33 -07:00
1 parent f7c2d382c3
commit cd824765b4
15 files changed
+851 -232
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

docs/ota_protocol.md
+38 -33
View File
@@ -312,9 +312,10 @@ while n > 1:
accept iff h == merkle_root and p == len(proof)
```
Over LoRa, `leaves[]` are **omitted** from the manifest transfer; a serving node computes a block's proof
on demand from its stored `leaves[]` (`OTA_REQ_PROOF`/`OTA_PROOF`, Section 8.5), and the fetcher fills its own
`leaves[i]` as each verified block lands.
Over LoRa, `leaves[]` are **omitted** from the manifest transfer. A serving node computes a block's proof
on demand from its stored `leaves[]` and normally sends `OTA_PROOF` immediately after that block's paced
`OTA_DATA`. `OTA_REQ_PROOF` remains the fallback for an older source or a lost proactive proof. The fetcher
fills its own `leaves[i]` as each verified block lands.
---
@@ -353,29 +354,30 @@ Message types:
| `ota_msg_type` | val | routing | purpose |
|---|---|---|---|
| `OTA_ADV` | 0x01 | flood | tiny per-node beacon (discovery tier 1) |
| `OTA_QUERY` | 0x02 | flood | ask a source for its catalog (discovery tier 2) |
| `OTA_HAVE` | 0x03 | flood | the catalog reply (fragmented, digest-tagged) |
| `OTA_GET_MANIFEST` | 0x04 | direct | request a manifest's fragments (`want_mask`) by `manifest_id` |
| `OTA_MANIFEST` | 0x05 | direct | the manifest-minus-leaves, fragmented |
| `OTA_REQ` | 0x06 | direct | request specific DATA fragments of one block (`want_mask`) |
| `OTA_DATA` | 0x07 | direct | one self-describing fragment of a block's data |
| `OTA_REQ_PROOF` | 0x08 | direct | request the merkle proof for one block |
| `OTA_PROOF` | 0x09 | direct | the merkle proof for one block |
| `OTA_GET_LEAVES` | 0x0A | direct | request the target's `leaves[]` fragments (`want_mask`) - warm-start only |
| `OTA_LEAVES` | 0x0B | direct | a fragment of the `leaves[]` array (for host-side seed leaf-diff) |
| `OTA_ADV` | 0x01 | discovery | tiny per-node beacon (discovery tier 1) |
| `OTA_QUERY` | 0x02 | discovery | ask a source for its catalog (discovery tier 2) |
| `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_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 |
| `OTA_GET_LEAVES` | 0x0A | transfer | request the target's `leaves[]` fragments (`want_mask`) - warm-start only |
| `OTA_LEAVES` | 0x0B | transfer | a fragment of the `leaves[]` array (for host-side seed leaf-diff) |
- **`manifest_id`** = the manifest's `merkle_root` (4 bytes) - a compact content id present in every
transfer message, so a multi-mota server dispatches each request to the right image.
- **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 missing fragments/blocks after a timeout, possibly
from a different peer. No hard ACKs, no global ordering.
- **Relay:** replies are flooded, so transparent relay needs no per-requester addressing, and the transfer
is trustless (the fetcher verifies every block against the signed root). Any neighbor may serve any
fragment it has. A repeater without `ENABLE_OTA` transports `PAYLOAD_TYPE_OTA` opaquely and does not need
the manager, staging store, installer, or destination bootloader.
- **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.
- **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
every block against the signed root, and a repeater without `ENABLE_OTA` can transport
`PAYLOAD_TYPE_OTA` opaquely without the manager, staging store, installer, or destination bootloader.
- **Hop limit + duty cycle:** OTA floods accumulate one path-hash per relay (the mesh's flood routing). A
node with the OTA manager *accepts* a packet only if it arrived within `ota config hops` hops (default 3;
`0` = direct only) and *relays* it only while still under that limit, appending its own hash. Relay-only
@@ -456,8 +458,8 @@ fetcher server (any node that has the mid)
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_REQ_PROOF(mid, block_idx) ---->
<------- OTA_PROOF(mid, block_idx, n_proof, proof)
<------- 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])
when all blocks present: verify full merkle_root + image_hash -> COMPLETE
```
@@ -490,7 +492,7 @@ OTA_LEAVES: manifest_id[4] frag_idx(1) frag_total(1) bytes[] # up
fetch. **Normal P2P nodes never use this** - they target only the blocks they want; the only always-on part
is answering `OTA_GET_LEAVES` with leaves the node already holds, so any node's firmware can be captured.
- **Block <-> fragments:** a 1 KB block is split into self-describing `OTA_DATA` fragments. `frag_off` is the
- **Block <-> fragments:** a 1 KB block remains split into self-describing `OTA_DATA` fragments. `frag_off` is the
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.
@@ -504,10 +506,12 @@ OTA_LEAVES: manifest_id[4] frag_idx(1) frag_total(1) bytes[] # up
with the tail of the in-flight burst and drop the same fragment forever - a hang. Requesting only the hole
removes the burst, so there is nothing to collide with. The block/manifest mask matches the 16-bit
reassembly bitmap (<=16 fragments/block; 1 KB blocks = 7). `OTA_PROOF` is a single packet and needs no mask.
- **Data and proof are separate phases.** `OTA_DATA` carries no proof; the proof is fetched once per block
via `OTA_REQ_PROOF`/`OTA_PROOF` after that block's data is complete. The receiver keeps a bounded two-block
window, so one block can await its proof or flash commit while DATA for the next block arrives. Slots retry
independently and only one stalled slot is retried per service tick.
- **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.
### 8.5 Sizing against `MAX_PACKET_PAYLOAD = 184`
@@ -534,12 +538,13 @@ to its persistent archive, it registers that complete file as a MotaSource and a
This keeps each active transfer as one transmitter and one receiver while still allowing active temporary-radio
repeaters between them and persistent archive nodes to improve future availability.
Because TempRadio is a bounded channel dedicated to this transfer, an accepted OTA flood normally relays after
a randomized 0.25 to 0.5 packet-airtime delay. Repeated requests for the same manifest block or proof provide
congestion feedback without changing the wire format. Frequent retries widen the local relay window through
0.5-1.0, 0.75-2.0, and finally 1.0-3.0 airtimes. The maximum is hard-capped at three packet airtimes. Each
30-second interval without another observed retry lowers the window one level. Other flood payloads retain the
role's normal randomized delay, and CAD still applies when enabled.
TempRadio is treated as a private maintenance network. Active transfer packets use priority 0, bypass the
public-flood receive holdoff and relay jitter, and do not schedule generic flood retries. The bounded serving
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.
---
examples/simple_repeater/MyMesh.cpp
+4
View File
@@ -1147,12 +1147,16 @@ bool MyMesh::evaluateScopeRewriteTiming(const mesh::Packet* packet,
}
bool MyMesh::shouldBypassRxDelay(const mesh::Packet* packet) {
if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA
&& isTempRadioActive()) return true;
bool fast_track = false;
return evaluateScopeRewriteTiming(packet, fast_track) && fast_track;
}
int MyMesh::calcRxDelayForPacket(const mesh::Packet* packet, float score,
uint32_t air_time) {
if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA
&& isTempRadioActive()) return 0;
bool fast_track = false;
if (!evaluateScopeRewriteTiming(packet, fast_track)) {
return calcRxDelay(score, air_time);
examples/simple_room_server/MyMesh.cpp
+2
View File
@@ -384,6 +384,8 @@ bool MyMesh::isLooped(const mesh::Packet* packet,
int MyMesh::calcRxDelayForPacket(const mesh::Packet* packet, float score,
uint32_t air_time) {
if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA
&& isTempRadioActive()) return 0;
bool fast_track = false;
if (!evaluateFloodRuleTiming(packet, fast_track)) {
return calcRxDelay(score, air_time);
src/Mesh.cpp
+55 -11
View File
@@ -153,13 +153,37 @@ static uint8_t otaTrafficPriority(const uint8_t* msg, uint16_t len) {
return isPrimaryOtaTraffic(msg, len) ? OTA_TRANSFER_TX_PRIORITY : OTA_TX_PRIORITY;
}
static bool isPacedOtaResponse(const uint8_t* msg, uint16_t len) {
if (!msg || len == 0) return false;
return msg[0] == ota::OTA_DATA || msg[0] == ota::OTA_PROOF;
}
#if defined(ENABLE_OTA)
static int queuedPacedOtaResponses(PacketManager* manager) {
int paced = 0;
const int total = manager->getOutboundTotal();
for (int i = 0; i < total; i++) {
Packet* packet = manager->getOutboundByIdx(i);
if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA
&& isPacedOtaResponse(packet->payload, packet->payload_len)) {
paced++;
}
}
return paced;
}
// Adapter so the portable OtaManager can emit packets through the mesh (message-priority, hop-capped).
void Mesh::otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
// DATA/PROOF admission is credit-limited; false leaves the manager's response descriptor intact.
bool Mesh::otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
Mesh* m = (Mesh*)ctx;
if (!m->isTempRadioActive()) return;
if (!m->isTempRadioActive()) return false;
if (isPacedOtaResponse(msg, len)
&& (queuedPacedOtaResponses(m->_mgr) >= OTA_EGRESS_QUEUE_CREDIT
|| m->_mgr->getFreeCount() <= OTA_EGRESS_MIN_FREE)) {
return false;
}
Packet* p = m->createOtaPacket(msg, len);
if (p) m->sendOtaFlood(p);
return p && m->sendOtaFlood(p);
}
// Runtime OTA flood reach (`ota config hops`, persisted in NodePrefs): accept packets up to N hops away and
@@ -318,6 +342,10 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() {
#endif
const bool ota_active = isTempRadioActive();
if (!ota_active) {
if (_ota_temp_was_active) {
ota::ota_ctx().manager.clearPendingEgress();
resetOtaRelayBackoff();
}
_ota_temp_was_active = false;
return;
}
@@ -327,6 +355,8 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() {
_next_ota_announce = 0;
_ota_announce_count = 0;
}
ota::ota_ctx().manager.set_clock(_ms->getMillis());
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
// a PARTIAL container (continue fetching the holes); a COMPLETE one is left for manual/auto-install,
@@ -341,12 +371,11 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() {
}
}
#endif
ota::ota_ctx().manager.set_clock(_ms->getMillis()); // for discovery jitter/ages + the pending-query timer
ota::ota_ctx().manager.loop(); // re-request still-missing OTA blocks + fire scheduled queries
#if defined(NRF52_PLATFORM) && defined(OTA_SD_STORE)
ota::ota_ctx().serviceSdCache(_ms->getMillis()); // archive unseen mOTAs only while the receive slot is idle
#endif
_next_ota_tick = futureMillis(3000);
_next_ota_tick = futureMillis(OTA_RETRY_TICK_MS);
}
if (millisHasNowPassed(_next_ota_announce)) { // auto-advertise so peers discover us (tiny beacon)
ota::OtaContext& oc = ota::ota_ctx();
@@ -485,6 +514,7 @@ void Mesh::decayOtaRelayBackoff() {
uint32_t Mesh::getOtaRetransmitDelay(const mesh::Packet* packet) {
if (packet == NULL) return 0;
if (isPrimaryOtaTraffic(packet->payload, packet->payload_len)) return 0;
decayOtaRelayBackoff();
uint32_t airtime = _radio->getEstAirtimeFor(packet->getRawLength());
if (airtime == 0) return 0;
@@ -498,6 +528,23 @@ uint32_t Mesh::getOtaRetransmitDelay(const mesh::Packet* packet) {
if (max_delay < min_delay) max_delay = min_delay;
return _rng->nextInt(min_delay, max_delay + 1);
}
int Mesh::calcRxDelayForPacket(const Packet* packet, float score, uint32_t air_time) {
if (packet != NULL && packet->getPayloadType() == PAYLOAD_TYPE_OTA
&& isTempRadioActive()) {
return 0;
}
return Dispatcher::calcRxDelayForPacket(packet, score, air_time);
}
uint32_t Mesh::getCADFailRetryDelay() const {
if (!isTempRadioActive()) return _rng->nextInt(1, 4) * 120;
uint32_t airtime = _radio->getEstAirtimeFor(MAX_TRANS_UNIT);
uint32_t retry = airtime / 4;
if (retry < 5) retry = 5;
if (retry > 50) retry = 50;
return retry;
}
uint32_t Mesh::getDirectRetransmitDelay(const Packet* packet) {
return 0; // by default, no delay
}
@@ -582,6 +629,7 @@ uint8_t Mesh::applyFloodRetryAttemptPolicy(const Packet* packet,
switch (packet->getPayloadType()) {
case PAYLOAD_TYPE_REQ:
case PAYLOAD_TYPE_OTA:
return 0;
case PAYLOAD_TYPE_GRP_TXT:
return attempts;
@@ -654,10 +702,6 @@ void Mesh::onSendFail(Packet* packet) {
clearPendingFloodRetryOnSendFail(packet);
}
uint32_t Mesh::getCADFailRetryDelay() const {
return _rng->nextInt(1, 4)*120;
}
int Mesh::searchPeersByHash(const uint8_t* hash) {
return 0; // not found
}
@@ -2639,12 +2683,12 @@ Packet* Mesh::createOtaPacket(const uint8_t* data, size_t len) {
return packet;
}
void Mesh::sendOtaFlood(Packet* packet, uint32_t delay_millis) {
bool Mesh::sendOtaFlood(Packet* packet, uint32_t delay_millis) {
packet->header &= ~PH_ROUTE_MASK;
packet->header |= ROUTE_TYPE_FLOOD;
packet->setPathHashSizeAndCount(1, 0);
_tables->markSeen(packet); // mark as sent, in case it floods back to us
sendPacket(packet, otaTrafficPriority(packet->payload, packet->payload_len), delay_millis);
return sendPacket(packet, otaTrafficPriority(packet->payload, packet->payload_len), delay_millis);
}
#endif
src/Mesh.h
+23 -5
View File
@@ -17,6 +17,19 @@
#ifndef OTA_FWD_MIN_FREE
#define OTA_FWD_MIN_FREE 4
#endif
// The OTA manager emits one retained response at a time, and this adapter keeps only a small number of OTA
// responses queued behind the packet currently on air. Unrelated queued traffic does not block the primary
// transfer. The free-pool floor preserves RX capacity and gives private-network relays a chance to forward
// each fragment before the origin fills the channel again.
#ifndef OTA_EGRESS_QUEUE_CREDIT
#define OTA_EGRESS_QUEUE_CREDIT 2
#endif
#ifndef OTA_EGRESS_MIN_FREE
#define OTA_EGRESS_MIN_FREE 4
#endif
#ifndef OTA_RETRY_TICK_MS
#define OTA_RETRY_TICK_MS 1000
#endif
namespace mesh {
@@ -214,10 +227,15 @@ protected:
virtual uint32_t getRetransmitDelay(const Packet* packet);
/**
* \returns Adaptive TempRadio-only OTA relay delay, from 0.25-0.5 airtimes up to a 3-airtime cap.
* \returns TempRadio OTA relay delay. Primary transfer packets forward immediately on the private
* network; discovery retains collision jitter.
*/
virtual uint32_t getOtaRetransmitDelay(const Packet* packet);
// Private TempRadio traffic has one intended path, so do not pay the public-flood receive holdoff. Keep
// CAD enabled, but retry it at radio-time scale instead of the ordinary 200 ms control-traffic cadence.
int calcRxDelayForPacket(const Packet* packet, float score, uint32_t air_time) override;
/**
* \returns number of milliseconds delay to apply to retransmitting the given packet, for DIRECT mode.
*/
@@ -481,7 +499,7 @@ protected:
// OTA mesh-integration is centralized in Mesh::begin()/loop()/dispatch, so every role (repeater,
// companion, room, sensor, ...) gets fetch/serve/apply without per-example wiring.
static void otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool flood);
static bool otaSendAdapter(void* ctx, const uint8_t* msg, uint16_t len, bool flood);
unsigned long _next_ota_tick = 0;
unsigned long _next_ota_announce = 0; // advertisements are scheduled only while temp radio is active
uint8_t _ota_announce_count = 0; // adverts sent so far (boot burst before settling to daily)
@@ -549,9 +567,9 @@ public:
#if defined(ENABLE_OTA)
// Build a PAYLOAD_TYPE_OTA packet from raw OTA message bytes (route set by sendOtaFlood).
Packet* createOtaPacket(const uint8_t* data, size_t len);
// Flood-send at the priority derived from the OTA message: discovery is background, active transfer is
// primary. Relay-only nodes use the same classification.
void sendOtaFlood(Packet* packet, uint32_t delay_millis = 0);
// The flood-shaped header is only the bounded multi-hop/dedup envelope inside TempRadio. Active transfer
// packets are primary and do not use generic flood retries or public-network relay backoff.
bool sendOtaFlood(Packet* packet, uint32_t delay_millis = 0);
#endif
Packet* createTrace(uint32_t tag, uint32_t auth_code, uint8_t flags = 0);
Packet* createControlData(const uint8_t* data, size_t len);
src/helpers/ota/OtaFormat.h
+2 -2
View File
@@ -81,8 +81,8 @@ enum OtaMsgType : uint8_t {
OTA_MANIFEST = 0x05,
OTA_REQ = 0x06, // request a window of blocks' DATA fragments
OTA_DATA = 0x07, // one fragment of a block's data (self-describing by frag_off; no proof)
OTA_REQ_PROOF = 0x08, // request the merkle proof for one block (data + proof are fetched separately)
OTA_PROOF = 0x09, // the merkle proof for one block
OTA_REQ_PROOF = 0x08, // request/re-request proof (legacy-source or lost proactive-proof fallback)
OTA_PROOF = 0x09, // the merkle proof for one block (normally follows its paced DATA)
OTA_GET_LEAVES = 0x0A, // request the target's merkle leaves[] (motatool warm-start; bitmap-fragmented)
OTA_LEAVES = 0x0B, // a fragment of the leaves[] array (for host-side seed leaf-diff)
};
src/helpers/ota/OtaManager.cpp
+143 -43
View File
@@ -48,6 +48,7 @@ void OtaManager::begin(uint32_t my_target_id, OtaSend send, void* ctx) {
_resume_merkle.reset();
_n_serve = 0; _n_src_obj = 0; _view0.valid = false; _srcv.valid = false;
_n_src = 0; _n_cat = 0;
clearPendingEgress();
}
// ---------------- serve (multi-mota registry) ----------------
@@ -65,6 +66,7 @@ bool OtaManager::serve(const uint8_t* mota, uint32_t len) {
_view0.read = nullptr; _view0.read_ctx = nullptr; // payload is contiguous _view0.m.payload
_view0.scratch = scratch; _view0.scratch_sz = OTA_PROOFGEN_SCRATCH; // <=1024 blocks (RAM .mota is small)
_view0.valid = true;
clearPendingEgress();
registerSelfEntry();
return true;
}
@@ -81,6 +83,7 @@ bool OtaManager::serve_self(const uint8_t* manifest, uint16_t mfl, const uint8_t
_view0.mfl = mfl; _view0.read = read; _view0.read_ctx = ctx;
_view0.scratch = proof_scratch; _view0.scratch_sz = proof_scratch_sz; // sized for our (large) image
_view0.valid = true;
clearPendingEgress();
registerSelfEntry();
return true;
}
@@ -127,6 +130,7 @@ bool OtaManager::remove_source(MotaSource* src) {
}
void OtaManager::refresh_sources() {
clearPendingEgress();
_n_serve = 0;
if (_view0.valid) registerSelfEntry();
for (uint8_t s = 0; s < _n_src_obj; s++) {
@@ -163,12 +167,14 @@ void OtaManager::refresh_sources() {
}
void OtaManager::clear_sources() {
clearPendingEgress();
_n_src_obj = 0; _srcv.valid = false;
_n_serve = 0;
if (_view0.valid) registerSelfEntry();
}
void OtaManager::clear_primary() {
clearPendingEgress();
_view0.valid = false;
if (_n_serve > 0 && _serve[0].is_self) {
for (uint8_t i = 1; i < _n_serve; i++) {
@@ -377,21 +383,56 @@ static inline uint16_t frag_full_mask(uint32_t nf) {
return (nf >= 16) ? 0xFFFFu : (uint16_t)((1u << nf) - 1);
}
// Emit one block's data as self-describing DATA fragments (frag_off); only the fragments whose bit is set
// in `want_mask` are sent (bit k = the slice at k*OTA_FRAG_DATA). The proof is fetched separately.
void OtaManager::emitBlockData(const uint8_t* mid, uint32_t idx, const uint8_t* data, uint32_t blen,
uint16_t want_mask) {
uint32_t k = 0;
for (uint32_t fo = 0; fo < blen; fo += OTA_FRAG_DATA, k++) {
if (!(want_mask & (1u << k))) continue; // fetcher didn't ask for this fragment
uint32_t fl = (fo + OTA_FRAG_DATA <= blen) ? OTA_FRAG_DATA : (blen - fo);
DataMsg dm;
memcpy(dm.manifest_id, mid, 4);
dm.block_idx = (uint16_t)idx; dm.frag_off = (uint16_t)fo;
dm.data = data + fo; dm.data_len = (uint16_t)fl;
uint8_t b[MAX_PACKET_PAYLOAD];
emit(b, encode_data(b, sizeof(b), dm), false);
// Retain a bounded descriptor instead of allocating every DATA packet inline. Repeated requests that arrive
// while a block is already queued merge only fragments not yet admitted to the radio queue; a later retry can
// enqueue the block again if one of those admitted packets was actually lost over the air.
bool OtaManager::queueServeJob(const uint8_t* mid, uint16_t block, uint16_t want_mask) {
for (uint8_t i = 0; i < _n_serve_jobs; i++) {
ServeJob& job = _serve_jobs[i];
if (job.block != block || memcmp(job.mid, mid, 4) != 0) continue;
job.pending_mask |= (uint16_t)(want_mask & ~job.emitted_mask);
return true;
}
if (_n_serve_jobs >= OTA_SERVE_QUEUE) return false;
ServeJob& job = _serve_jobs[_n_serve_jobs++];
memcpy(job.mid, mid, 4);
job.block = block;
job.pending_mask = want_mask;
job.emitted_mask = 0;
return true;
}
void OtaManager::clearPendingEgress() {
_n_serve_jobs = 0;
_serve_block_len = 0;
_serve_block_loaded = false;
}
void OtaManager::popServeJob() {
if (_n_serve_jobs == 0) return;
for (uint8_t i = 1; i < _n_serve_jobs; i++) _serve_jobs[i - 1] = _serve_jobs[i];
_n_serve_jobs--;
_serve_block_len = 0;
_serve_block_loaded = false;
}
bool OtaManager::loadActiveServeBlock() {
if (_n_serve_jobs == 0) return false;
ServeJob& job = _serve_jobs[0];
ServeView* v = resolve(job.mid);
if (!v || job.block >= v->m.block_count) return false;
uint32_t bs = v->m.block_size();
if (bs == 0 || bs > OTA_MAX_BLOCK) return false;
uint32_t off = (uint32_t)job.block * bs;
uint32_t blen = (off + bs <= v->m.payload_size) ? bs : (v->m.payload_size - off);
if (v->read) {
if (!v->read(v->read_ctx, off, _serve_block, blen)) return false;
} else {
memcpy(_serve_block, v->m.payload + off, blen);
}
_serve_block_len = (uint16_t)blen;
_serve_block_loaded = true;
return true;
}
void OtaManager::handleReq(const uint8_t* m, uint16_t n) {
@@ -399,33 +440,84 @@ void OtaManager::handleReq(const uint8_t* m, uint16_t n) {
if (!decode_req(m, n, rq)) return;
uint32_t idx = rq.block_idx;
ServeView* v = resolve(rq.manifest_id);
if (v) { // serve a fully-held mota (own fw or attached folder)
if (idx >= v->m.block_count) return;
uint32_t bs = v->m.block_size();
uint32_t off = idx * bs;
uint32_t blen = (off + bs <= v->m.payload_size) ? bs : (v->m.payload_size - off);
uint8_t blk[OTA_MAX_BLOCK];
const uint8_t* data;
if (v->read) { if (!v->read(v->read_ctx, off, blk, blen)) return; data = blk; }
else { data = v->m.payload + off; }
emitBlockData(v->m.merkle_root, idx, data, blen, rq.want_mask);
}
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);
}
void OtaManager::handleReqProof(const uint8_t* m, uint16_t n) {
ReqProofMsg rp;
if (!decode_req_proof(m, n, rp)) return;
ServeView* v = resolve(rp.manifest_id);
if (!v) return;
if (rp.block_idx >= v->m.block_count) return;
if ((uint64_t)v->m.block_count * 4 > v->scratch_sz) return; // proof-gen needs block_count*4 scratch
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
}
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) {
for (uint8_t i = 0; i < OTA_FETCH_PIPELINE; i++) {
ReassemblySlot& slot = _reasm[i];
if (!slot.awaiting_proof || slot.proof_request_at == 0
|| (int32_t)(_now_ms - slot.proof_request_at) < 0) continue;
if (requestSlot(i)) return;
break;
}
}
if (_n_serve_jobs == 0) return;
ServeJob& job = _serve_jobs[0];
ServeView* v = resolve(job.mid);
if (!v || job.block >= v->m.block_count
|| (uint64_t)v->m.block_count * 4 > v->scratch_sz) {
popServeJob();
return;
}
if (job.pending_mask != 0) {
if (!_serve_block_loaded && !loadActiveServeBlock()) {
popServeJob();
return;
}
uint8_t fragment = 0;
while (fragment < 16 && !(job.pending_mask & (1u << fragment))) fragment++;
uint32_t frag_off = (uint32_t)fragment * OTA_FRAG_DATA;
if (fragment >= 16 || frag_off >= _serve_block_len) {
job.pending_mask = 0;
return;
}
uint32_t frag_len = _serve_block_len - frag_off;
if (frag_len > OTA_FRAG_DATA) frag_len = OTA_FRAG_DATA;
DataMsg dm;
memcpy(dm.manifest_id, job.mid, 4);
dm.block_idx = job.block;
dm.frag_off = (uint16_t)frag_off;
dm.data = _serve_block + frag_off;
dm.data_len = (uint16_t)frag_len;
uint8_t b[MAX_PACKET_PAYLOAD];
if (emit(b, encode_data(b, sizeof(b), dm), false)) {
const uint16_t bit = (uint16_t)(1u << fragment);
job.pending_mask &= (uint16_t)~bit;
job.emitted_mask |= bit;
}
return;
}
uint8_t proof[32 * 4];
uint8_t np = merkle_gen_proof(v->m.leaves, v->m.block_count, rp.block_idx, v->scratch, proof);
uint8_t np = merkle_gen_proof(v->m.leaves, v->m.block_count, job.block, v->scratch, proof);
ProofMsg pm;
memcpy(pm.manifest_id, v->m.merkle_root, 4);
pm.block_idx = rp.block_idx; pm.n_proof = np; pm.proof = proof;
memcpy(pm.manifest_id, job.mid, 4);
pm.block_idx = job.block;
pm.n_proof = np;
pm.proof = proof;
uint8_t b[MAX_PACKET_PAYLOAD];
emit(b, encode_proof(b, sizeof(b), pm), false);
if (emit(b, encode_proof(b, sizeof(b), pm), false)) popServeJob();
}
// ---------------- fetch ----------------
@@ -613,6 +705,7 @@ void OtaManager::clearReassemblySlot(uint8_t slot) {
_reasm[slot].mask = 0;
_reasm[slot].need = 0;
_reasm[slot].awaiting_proof = false;
_reasm[slot].proof_request_at = 0;
}
// Forget all blocks currently being reassembled or awaiting proofs.
@@ -1010,10 +1103,12 @@ void OtaManager::handleData(const uint8_t* m, uint16_t n) {
if (kf >= 16) return;
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; // wait for all slices (or proof already asked)
// block fully reassembled -> request its proof (data + proof are fetched separately)
if (slot.mask != slot.need || slot.awaiting_proof) return;
// 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;
requestSlot((uint8_t)slot_index);
slot.proof_request_at = _now_ms + OTA_PROOF_GRACE_MS;
}
void OtaManager::handleProof(const uint8_t* m, uint16_t n) {
@@ -1055,10 +1150,10 @@ void OtaManager::handleProof(const uint8_t* m, uint16_t n) {
OTA_DBG("OTA: transfer %s\n", _fstate == COMPLETE ? "COMPLETE" : "FAILED(integrity/storage)");
}
void OtaManager::requestSlot(uint8_t slot_index) {
if (_fstate != FETCHING || slot_index >= OTA_FETCH_PIPELINE) return;
bool OtaManager::requestSlot(uint8_t slot_index) {
if (_fstate != FETCHING || slot_index >= OTA_FETCH_PIPELINE) return false;
ReassemblySlot& slot = _reasm[slot_index];
if (slot.block >= _fbc) return;
if (slot.block >= _fbc) return false;
_req_start = slot.block;
_req_count = 0;
for (uint8_t i = 0; i < OTA_FETCH_PIPELINE; i++) {
@@ -1066,10 +1161,14 @@ void OtaManager::requestSlot(uint8_t slot_index) {
}
if (slot.awaiting_proof) {
ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)slot.block;
uint8_t b[16]; emit(b, encode_req_proof(b, sizeof(b), rp), false);
OTA_DBG("OTA: REQ_PROOF block=%u (have=%u/%u window=%u)\n",
(unsigned)slot.block, (unsigned)_have, (unsigned)_fbc, (unsigned)_req_count);
return;
uint8_t b[16];
bool sent = emit(b, encode_req_proof(b, sizeof(b), rp), false);
if (sent) {
slot.proof_request_at = 0;
OTA_DBG("OTA: REQ_PROOF fallback block=%u (have=%u/%u window=%u)\n",
(unsigned)slot.block, (unsigned)_have, (unsigned)_fbc, (unsigned)_req_count);
}
return sent;
}
uint16_t want = (uint16_t)(slot.need & ~slot.mask);
if (want == 0) want = slot.need; // safety: never send an empty request
@@ -1079,7 +1178,7 @@ void 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);
emit(b, encode_req(b, sizeof(b), rq), false);
return emit(b, encode_req(b, sizeof(b), rq), false);
}
bool OtaManager::fillPipeline() {
@@ -1094,6 +1193,7 @@ bool OtaManager::fillPipeline() {
slot.mask = 0;
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;
}
src/helpers/ota/OtaManager.h
+43 -10
View File
@@ -14,15 +14,16 @@
// the signed merkle root via proofs). It is portable (no Arduino / radio / Ed25519) so it can be
// driven by a host simulation; a thin Mesh adapter wires it to PAYLOAD_TYPE_OTA on device.
//
// Transfer is per-block and 2-phase: a 1 KB logical block is fetched as self-describing DATA fragments
// (frag_off), reassembled, then its merkle PROOF
// is requested separately and verified against the signed root before the block is committed.
// Transfer keeps 1 KB logical blocks. A server paces each block's self-describing DATA fragments through a
// bounded response queue and follows them with the Merkle PROOF. A newer fetcher waits briefly for that
// proactive proof; the existing REQ_PROOF message remains as a loss/legacy-server fallback.
namespace mesh {
namespace ota {
// Emit an OTA message (one packet payload). `flood`=true for announce/query, false for direct replies.
typedef void (*OtaSend)(void* ctx, const uint8_t* msg, uint16_t len, bool flood);
// Try to emit one OTA message (one packet payload). `flood`=true for announce/query, false for direct
// replies. False applies backpressure: the manager retains paced DATA/PROOF egress and tries it again.
typedef bool (*OtaSend)(void* ctx, const uint8_t* msg, uint16_t len, bool flood);
// Read `len` payload bytes at offset `off` from the serve source (flash-backed self-serve); false on
// error. nullptr means the payload is a contiguous RAM buffer (the staged `.mota`).
@@ -128,6 +129,15 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len)
#ifndef OTA_FRAG_DATA
#define OTA_FRAG_DATA 160 // data bytes per DATA fragment (<= MAX_PACKET_PAYLOAD - 9-byte header)
#endif
#ifndef OTA_SERVE_QUEUE
#define OTA_SERVE_QUEUE 4 // requested blocks retained without allocating one packet per fragment
#endif
#if OTA_SERVE_QUEUE < 1
#error "OTA_SERVE_QUEUE must be at least 1"
#endif
#ifndef OTA_PROOF_GRACE_MS
#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
#endif
@@ -151,7 +161,7 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len)
#error "OTA_FETCH_PIPELINE_GROW_BLOCKS must be at least 1"
#endif
#ifndef OTA_FETCH_PIPELINE_SHRINK_TICKS
#define OTA_FETCH_PIPELINE_SHRINK_TICKS 2 // consecutive 3-second no-progress ticks before removing one slot
#define OTA_FETCH_PIPELINE_SHRINK_TICKS 2 // consecutive no-progress retry ticks before removing one slot
#endif
#if OTA_FETCH_PIPELINE_SHRINK_TICKS < 1
#error "OTA_FETCH_PIPELINE_SHRINK_TICKS must be at least 1"
@@ -338,6 +348,11 @@ public:
void on_message(const uint8_t* msg, uint16_t len); // feed one received OTA message
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.
void serviceEgress();
void clearPendingEgress();
uint8_t pendingServeJobs() const { return _n_serve_jobs; }
// Drop the current fetch session back to IDLE so a fresh `ota pull` / advert starts a new one.
void reset_session() {
@@ -384,7 +399,9 @@ private:
uint16_t catalogCapacity() const { return _catalog_heap ? OTA_MAX_CATALOG : OTA_INLINE_CATALOG; }
bool expandCatalog();
void emit(const uint8_t* b, uint16_t n, bool flood) { if (_send && n) _send(_ctx, b, n, flood); }
bool emit(const uint8_t* b, uint16_t n, bool flood) {
return _send && n && _send(_ctx, b, n, flood);
}
void handleAdv(const uint8_t* m, uint16_t n); // beacon -> sources table (+ query if interested)
void handleQuery(const uint8_t* m, uint16_t n); // serve: reply OTA_HAVE catalog
void handleHave(const uint8_t* m, uint16_t n); // peer: catalog rows (+ startFetch if a row matches)
@@ -415,8 +432,9 @@ private:
bool loadSource(const ServeEntry& e); // load an external mota into _srcv (head+leaves)
void registerSelfEntry(); // (re)build entry[0] from view0
static bool srcReadTramp(void* c, uint32_t off, uint8_t* buf, uint32_t len); // source payload reader
void emitBlockData(const uint8_t* mid, uint32_t idx, const uint8_t* data, uint32_t blen,
uint16_t want_mask); // emit only the requested DATA fragments of one block
bool queueServeJob(const uint8_t* mid, uint16_t block, uint16_t want_mask);
bool loadActiveServeBlock();
void popServeJob();
void sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target,
uint32_t want_fragments); // ask a source for all/selected catalog fragments
void scheduleQuery(const uint8_t* seeder, const uint8_t* digest); // jittered + suppressible
@@ -436,7 +454,7 @@ private:
bool storedLeavesRootMatches() const;
void beginStagedVerification();
void verifyStagedStep();
void requestSlot(uint8_t slot); // request DATA holes or the proof for one slot
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
uint32_t blockLen(uint32_t i) const;
@@ -465,6 +483,20 @@ private:
uint8_t _scratch[OTA_PROOFGEN_SCRATCH]; // proof-gen / fetch root-check working buffer
#endif
// Server-side response descriptors are tiny. The active 1 KB block has its own buffer so proof generation
// or a simultaneous fetch cannot overwrite DATA retained behind radio-queue backpressure.
struct ServeJob {
uint8_t mid[4] = {0};
uint16_t block = 0;
uint16_t pending_mask = 0;
uint16_t emitted_mask = 0;
};
ServeJob _serve_jobs[OTA_SERVE_QUEUE];
uint8_t _n_serve_jobs = 0;
uint8_t _serve_block[OTA_MAX_BLOCK];
uint16_t _serve_block_len = 0;
bool _serve_block_loaded = false;
// fetch
OtaStore* _fetch = nullptr;
FetchState _fstate = IDLE;
@@ -499,6 +531,7 @@ private:
uint16_t mask = 0; // received FRAG_DATA-slice bitmap
uint16_t need = 0; // full bitmap for this block
bool awaiting_proof = false;
uint32_t proof_request_at = 0; // proactive-proof grace deadline; 0 after fallback is sent
uint8_t buf[OTA_MAX_BLOCK];
};
ReassemblySlot _reasm[OTA_FETCH_PIPELINE];
src/helpers/ota/OtaProtocol.h
+2 -2
View File
@@ -70,7 +70,7 @@ struct ManifestMsg {
// (saves airtime and avoids the re-REQ colliding with a multi-fragment burst on half-duplex radios).
struct ReqMsg { uint8_t manifest_id[4]; uint16_t block_idx; uint16_t want_mask; };
// ---- OTA_DATA: one self-describing fragment of a block's data (proof is fetched separately) ----
// ---- OTA_DATA: one self-describing fragment of a block's data (proof stays in a separate packet) ----
// body: manifest_id(4) block_idx(2) frag_off(2) data[]
// `frag_off` is the byte offset of `data` within block `block_idx` (global position = block_idx*block_size
// + frag_off), so a fragment is self-placing when received from the selected source.
@@ -96,7 +96,7 @@ struct LeavesMsg {
const uint8_t* bytes; uint16_t len;
};
// ---- OTA_REQ_PROOF: request the merkle proof for one (reassembled) block (direct) ----
// ---- OTA_REQ_PROOF: proof fallback for an older source or a lost proactive proof (direct) ----
struct ReqProofMsg { uint8_t manifest_id[4]; uint16_t block_idx; };
// ---- OTA_PROOF: the merkle proof (ordered sibling digests) for one block (direct) ----
test/README.md
+1 -1
View File
@@ -49,7 +49,7 @@ does not reflect the GoogleTest count -- run the built binary directly
| `test_serial_mode_switch` | `src/helpers/ArduinoSerialInterface.cpp`, `src/helpers/MultiSerialInterface.h` | independent terminal/seeder control-sequence recognition across reads and binary-frame boundaries; passthrough ownership of USB input and suppression of binary output; Bluetooth-only connection and pairing-request routing |
| `test_ble_tx_stall_watchdog` | `src/helpers/BleTxStallWatchdog.h` | exact BLE fragment progress; blocked-reply timeout; rollover-safe elapsed time; disconnect recovery retry and completion |
| `test_ota` | `src/helpers/ota/` | container and EndF integrity; protocol codecs; transfer, resume, and apply safety; adaptive 2-to-4 block-request window growth and stall contraction; active-transfer priority classification |
| `test_trace_retry` | `src/Mesh.cpp` retry and relay policy | opaque OTA relay behavior during TempRadio; background discovery priority; primary active-transfer priority and packet-reserve bypass; trace and flood retry timing |
| `test_trace_retry` | `src/Mesh.cpp` retry and relay policy | opaque OTA relay behavior during TempRadio; background discovery priority; immediate primary transfer relay, receive-delay bypass, fast CAD retry, and no generic flood retry; trace and non-OTA flood retry timing |
| `test_utils` | `src/Utils.cpp` | `Utils::toHex` (upstream) |
## Conventions (and how to add a suite)
test/test_ota/test_ota_core.cpp
+145 -26
View File
@@ -538,26 +538,48 @@ namespace {
struct SimMsg { OtaManager* dest; std::vector<uint8_t> bytes; };
static std::vector<SimMsg> g_q;
struct SendTo { OtaManager* dest; };
static void sim_send(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
static bool sim_send(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
g_q.push_back({((SendTo*)ctx)->dest, std::vector<uint8_t>(msg, msg + len)});
return true;
}
struct CapturedMessages { std::vector<std::vector<uint8_t>> items; };
static void capture_send(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
static bool capture_send(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
((CapturedMessages*)ctx)->items.emplace_back(msg, msg + len);
return true;
}
struct GatedCapture {
bool accept = false;
std::vector<std::vector<uint8_t>> items;
};
static bool gated_capture_send(void* ctx, const uint8_t* msg, uint16_t len, bool /*flood*/) {
GatedCapture* capture = (GatedCapture*)ctx;
if (!capture->accept) return false;
capture->items.emplace_back(msg, msg + len);
return true;
}
// Drive the bus to quiescence: deliver queued messages; when idle, advance the client's clock (monotonic
// across calls, so a jittered query scheduled in a prior pump still comes due) and call loop() (fires the
// scheduled catalog query / block re-requests). Two idle ticks in a row = quiescent.
static uint32_t g_clk = 0;
static void pump(OtaManager& client, int guard_max = 200000) {
static void pump(OtaManager& client, OtaManager* server = nullptr, int guard_max = 200000) {
int idle = 0, guard = 0;
while (guard++ < guard_max) {
if (server) server->serviceEgress();
client.serviceEgress();
if (!g_q.empty()) {
SimMsg m = std::move(g_q.front()); g_q.erase(g_q.begin());
m.dest->on_message(m.bytes.data(), (uint16_t)m.bytes.size());
idle = 0;
} else {
g_clk += 5000; client.set_clock(g_clk); client.loop();
g_clk += 5000;
client.set_clock(g_clk);
client.loop();
if (server) {
server->set_clock(g_clk);
server->loop();
server->serviceEgress();
}
client.serviceEgress();
if (!g_q.empty()) { idle = 0; continue; }
if (++idle >= 2) break;
}
@@ -682,7 +704,7 @@ TEST(OtaTransfer, TwoManagersFullTransfer) {
ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
server.announce(); // -> client hears the beacon, queries, catalogs, then fetches
pump(client); // beacon -> query -> have -> startFetch -> full transfer
pump(client, &server); // beacon -> query -> have -> startFetch -> full transfer
EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE);
EXPECT_EQ(client.blocksHave(), client.blocksTotal());
@@ -713,6 +735,102 @@ static void deliver_manifest_fragment(OtaManager& client, const uint8_t mid[4],
client.on_message(wire, wire_len);
}
TEST(OtaTransfer, ServerPacesOneKilobyteBlockAndProactiveProofWithBackpressure) {
MotaManifest manifest;
ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
ASSERT_EQ(manifest.block_size(), 1024u);
OtaManager server;
GatedCapture sent;
server.begin(0, gated_capture_send, &sent);
ASSERT_TRUE(server.serve(SIM_MOTA_1K, SIM_MOTA_1K_LEN));
ReqMsg request;
memcpy(request.manifest_id, manifest.merkle_root, 4);
request.block_idx = 0;
request.want_mask = 0xFFFF;
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
EXPECT_EQ(server.pendingServeJobs(), 1u);
EXPECT_TRUE(sent.items.empty()); // receive handler never allocates a packet burst
server.serviceEgress(); // callback applies packet-queue backpressure
EXPECT_TRUE(sent.items.empty());
EXPECT_EQ(server.pendingServeJobs(), 1u);
sent.accept = true;
const uint32_t fragment_count = (1024 + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA;
for (uint32_t i = 0; i < fragment_count; i++) {
server.serviceEgress();
ASSERT_EQ(sent.items.size(), i + 1);
DataMsg data;
ASSERT_TRUE(decode_data(sent.items.back().data(),
(uint16_t)sent.items.back().size(), data));
EXPECT_EQ(data.block_idx, 0);
EXPECT_EQ(data.frag_off, i * OTA_FRAG_DATA);
}
EXPECT_EQ(server.pendingServeJobs(), 1u); // proof is retained behind DATA
server.serviceEgress();
ASSERT_EQ(sent.items.size(), fragment_count + 1);
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(), 0u);
}
TEST(OtaTransfer, MissingProactiveProofFallsBackAfterGrace) {
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.pull(manifest.merkle_root, manifest.target_id);
sent.items.clear();
client.set_clock(100);
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));
client.loop();
ASSERT_EQ(sent.items.size(), 2u);
const uint32_t block_len = manifest.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 = 0;
data.frag_off = (uint16_t)offset;
data.data = manifest.payload + offset;
data.data_len = (uint16_t)length;
uint16_t length_on_wire = encode_data(wire, sizeof(wire), data);
ASSERT_GT(length_on_wire, 0);
client.on_message(wire, length_on_wire);
}
EXPECT_EQ(sent.items.size(), 2u);
client.set_clock(100 + OTA_PROOF_GRACE_MS - 1);
client.serviceEgress();
EXPECT_EQ(sent.items.size(), 2u);
client.set_clock(100 + OTA_PROOF_GRACE_MS);
client.serviceEgress();
ASSERT_EQ(sent.items.size(), 3u);
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) {
MotaManifest manifest;
ASSERT_TRUE(mota_parse(SIM_MOTA_1K, SIM_MOTA_1K_LEN, manifest));
@@ -742,8 +860,8 @@ TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) {
EXPECT_EQ(request0.block_idx, 0);
EXPECT_EQ(request1.block_idx, 1);
// Deliver block 1 first and its fragments in reverse order. Its independent slot must request and
// accept a proof without disturbing the still-empty slot reserved for block 0.
// 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
@@ -765,10 +883,7 @@ TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) {
client.on_message(wire, wire_len);
}
ASSERT_EQ(sent.items.size(), 3u);
ReqProofMsg proof_request;
ASSERT_TRUE(decode_req_proof(sent.items[2].data(), (uint16_t)sent.items[2].size(), proof_request));
EXPECT_EQ(proof_request.block_idx, block);
ASSERT_EQ(sent.items.size(), 2u); // no immediate proof request
uint8_t scratch[SIM_MOTA_1K_BLOCKS * 4];
uint8_t siblings[32 * 4];
@@ -784,9 +899,9 @@ TEST(OtaTransfer, ClientPipelinesAndRefillsOutOfOrderBlocks) {
client.on_message(wire, wire_len);
EXPECT_EQ(client.blocksHave(), 1u);
ASSERT_EQ(sent.items.size(), 4u); // freed slot was immediately refilled
ASSERT_EQ(sent.items.size(), 3u); // freed slot was immediately refilled
ReqMsg refill;
ASSERT_TRUE(decode_req(sent.items[3].data(), (uint16_t)sent.items[3].size(), refill));
ASSERT_TRUE(decode_req(sent.items[2].data(), (uint16_t)sent.items[2].size(), refill));
EXPECT_EQ(refill.block_idx, 2);
}
@@ -992,7 +1107,7 @@ TEST(OtaTransfer, MultiFragmentBlocks) {
ASSERT_TRUE(server.serve(SIM_MOTA_1K, SIM_MOTA_1K_LEN));
server.announce();
pump(client);
pump(client, &server);
EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE);
EXPECT_EQ(client.blocksTotal(), SIM_MOTA_1K_BLOCKS); // 1 KB blocks => fewer, larger blocks
@@ -1036,14 +1151,14 @@ TEST(OtaFolder, ServesSelfPlusFolderAndFetchesExternal) {
// discovery: beacon -> the client catalogs the source, queries it, and the broadcast HAVE fills the
// catalog with BOTH served mids.
server.announce();
pump(client);
pump(client, &server);
client.queryAll();
pump(client);
pump(client, &server);
EXPECT_EQ(client.catalogCount(), 2);
// fetch the EXTERNAL (folder) mota by mid -> served via the source, relayed block-by-block.
client.pull(mExt.merkle_root, mExt.target_id);
pump(client);
pump(client, &server);
EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE);
EXPECT_EQ(client.blocksHave(), client.blocksTotal());
@@ -1189,12 +1304,16 @@ TEST(OtaTransfer, ResumeAfterReboot) {
// drive only until the first block commits, then "crash"
int idle = 0, guard = 0;
while (guard++ < 100000) {
server.serviceEgress();
client.serviceEgress();
if (!g_q.empty()) {
SimMsg msg = std::move(g_q.front()); g_q.erase(g_q.begin());
msg.dest->on_message(msg.bytes.data(), (uint16_t)msg.bytes.size());
idle = 0;
} else {
g_clk += 5000; client.set_clock(g_clk); client.loop();
g_clk += 5000;
client.set_clock(g_clk); client.loop(); client.serviceEgress();
server.set_clock(g_clk); server.loop(); server.serviceEgress();
if (!g_q.empty()) { idle = 0; } else if (++idle >= 2) break;
}
if (client.blocksHave() >= 1) break;
@@ -1217,7 +1336,7 @@ TEST(OtaTransfer, ResumeAfterReboot) {
EXPECT_EQ(client2.fetchState(), OtaManager::FETCHING);
EXPECT_EQ(client2.blocksTotal(), SIM_MOTA_1K_BLOCKS);
pump(client2);
pump(client2, &server);
EXPECT_EQ(client2.fetchState(), OtaManager::COMPLETE);
ASSERT_EQ(store.staged_size(), SIM_MOTA_1K_LEN);
EXPECT_EQ(0, std::memcmp(store.data(), SIM_MOTA_1K, SIM_MOTA_1K_LEN)); // byte-identical to the original
@@ -1234,7 +1353,7 @@ TEST(OtaTransfer, ResumeRehashesPayloadBeforeTrustingPresentLeaf) {
client.set_autofetch(OtaManager::AUTOFETCH_ANY);
ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
server.announce();
pump(client);
pump(client, &server);
ASSERT_EQ(client.fetchState(), OtaManager::COMPLETE);
MotaManifest staged;
@@ -1270,7 +1389,7 @@ TEST(OtaTransfer, ResumeReadFailureCanNeverBecomeComplete) {
client.set_autofetch(OtaManager::AUTOFETCH_ANY);
ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
server.announce();
pump(client);
pump(client, &server);
ASSERT_EQ(client.fetchState(), OtaManager::COMPLETE);
MotaManifest staged;
@@ -1296,7 +1415,7 @@ TEST(OtaTransfer, ClientRejectsWrongTarget) {
client.set_autofetch(OtaManager::AUTOFETCH_ANY); // tests exercise fetch-on-advert; policy default is OFF
ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
server.announce();
pump(client); // catalogs the row but wantRow rejects it (wrong target) -> never fetches
pump(client, &server); // catalogs the row but wantRow rejects it (wrong target) -> never fetches
EXPECT_EQ(client.fetchState(), OtaManager::IDLE); // never started
}
@@ -1315,13 +1434,13 @@ TEST(OtaTransfer, ManualCrossTargetFetch) {
// without the override: catalogs the row but won't fetch (wrong target)
server.announce();
pump(client);
pump(client, &server);
EXPECT_EQ(client.fetchState(), OtaManager::IDLE);
// with want(): deliberately fetch the different-target firmware to completion
client.want(SIM_TARGET_ID);
server.announce();
pump(client);
pump(client, &server);
EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE);
ASSERT_EQ(store.staged_size(), SIM_MOTA_LEN);
EXPECT_EQ(0, std::memcmp(store.data(), SIM_MOTA, SIM_MOTA_LEN));
@@ -1387,7 +1506,7 @@ TEST(OtaTransfer, ArchivePullAcceptsCrossTargetUnsupportedCodec) {
MotaManifest m;
ASSERT_TRUE(mota_parse(SIM_MOTA, SIM_MOTA_LEN, m));
client.pull_archive(m.merkle_root, m.target_id); // capture ignores local target + install codec
pump(client);
pump(client, &server);
EXPECT_EQ(client.fetchState(), OtaManager::COMPLETE);
ASSERT_EQ(store.staged_size(), SIM_MOTA_LEN);
@@ -1451,7 +1570,7 @@ TEST(OtaTransfer, ReceiverDoesNotReSeed) {
client.set_autofetch(OtaManager::AUTOFETCH_ANY);
ASSERT_TRUE(server.serve(SIM_MOTA, SIM_MOTA_LEN));
server.announce();
pump(client);
pump(client, &server);
ASSERT_EQ(client.fetchState(), OtaManager::COMPLETE);
EXPECT_EQ(client.servedCount(), 0);
client.reset_session();
test/test_trace_retry/test_trace_retry.cpp
+36 -8
View File
@@ -104,6 +104,14 @@ public:
return getOtaRetransmitDelay(packet);
}
int receiveDelay(const mesh::Packet* packet, float score, uint32_t air_time) {
return calcRxDelayForPacket(packet, score, air_time);
}
uint32_t cadRetryDelay() const {
return getCADFailRetryDelay();
}
void completePacketSend(mesh::Packet* packet) {
onSendComplete(packet);
}
@@ -246,7 +254,7 @@ TEST(RepeaterTransport, OtaTransferRelaysAsPrimaryTrafficWithoutOtaManager) {
EXPECT_EQ(1, request.getPathHashCount());
}
TEST(RepeaterTransport, OtaRelayDelayIsQuarterToHalfAirtime) {
TEST(RepeaterTransport, OtaDiscoveryRelayKeepsCollisionJitter) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
@@ -255,6 +263,7 @@ TEST(RepeaterTransport, OtaRelayDelayIsQuarterToHalfAirtime) {
StaticPoolPacketManager manager(12);
TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
mesh::Packet packet = makeFloodPacket(PAYLOAD_TYPE_OTA);
packet.payload[0] = mesh::ota::OTA_ADV;
rng.value = 0;
EXPECT_EQ(node.otaRelayDelay(&packet), 3u); // ceil(0.25 * 10 ms)
@@ -262,7 +271,7 @@ TEST(RepeaterTransport, OtaRelayDelayIsQuarterToHalfAirtime) {
EXPECT_EQ(node.otaRelayDelay(&packet), 5u); // 0.5 * 10 ms
}
TEST(RepeaterTransport, OtaRelayDelayEscalatesAndCapsAtThreeAirtimes) {
TEST(RepeaterTransport, OtaTransferRelayIsImmediateEvenUnderRequestPressure) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
@@ -282,17 +291,36 @@ TEST(RepeaterTransport, OtaRelayDelayEscalatesAndCapsAtThreeAirtimes) {
mesh::Packet request = make_request();
node.receivePacket(&request); // first request establishes the key
EXPECT_EQ(node.otaRelayDelay(&request), 0u);
for (uint32_t retry = 1; retry <= 3; retry++) {
clock.now = retry * 3000;
request = make_request(); // a fresh origin copy has the same zero-hop request key
node.receivePacket(&request); // frequent repeats raise one level each
}
rng.value = 20; // last value in inclusive [10, 30]
EXPECT_EQ(node.otaRelayDelay(&request), 30u); // hard maximum = 3 * 10 ms
rng.value = 20;
EXPECT_EQ(node.otaRelayDelay(&request), 0u); // private transfer never inherits flood backoff
clock.now = 100000; // three clean 30-second decay periods
clock.now = 100000;
rng.value = 2;
EXPECT_EQ(node.otaRelayDelay(&request), 5u); // returned to the 0.25-0.5 baseline
EXPECT_EQ(node.otaRelayDelay(&request), 0u);
}
TEST(RepeaterTransport, TempRadioOtaBypassesReceiveHoldoffAndUsesFastCadRetry) {
TraceTestClock clock;
TraceTestRTC rtc;
TraceTestRNG rng;
TraceTestRadio radio; // fixed 10 ms full-packet airtime
ForwardingTestTables tables;
StaticPoolPacketManager manager(12);
TraceTestMesh node(radio, clock, rng, rtc, manager, tables);
mesh::Packet packet = makeFloodPacket(PAYLOAD_TYPE_OTA);
packet.payload[0] = mesh::ota::OTA_DATA;
EXPECT_GT(node.receiveDelay(&packet, 0.0f, 10), 0);
EXPECT_EQ(node.cadRetryDelay(), 120u);
node.tempRadioActive = true;
EXPECT_EQ(node.receiveDelay(&packet, 0.0f, 10), 0);
EXPECT_EQ(node.cadRetryDelay(), 5u);
}
TEST(RepeaterTransport, OpaqueKnownFloodPayloadsAreRelayed) {
@@ -517,7 +545,7 @@ TEST(FloodRetry, PayloadAndPathPolicyCapsEveryFloodType) {
packet.setPathHashSizeAndCount(1, 0);
uint8_t origin_limit;
if (type == PAYLOAD_TYPE_REQ) {
if (type == PAYLOAD_TYPE_REQ || type == PAYLOAD_TYPE_OTA) {
origin_limit = 0;
} else if (type == PAYLOAD_TYPE_GRP_TXT || type == PAYLOAD_TYPE_RESPONSE
|| type == PAYLOAD_TYPE_TXT_MSG || type == PAYLOAD_TYPE_ANON_REQ
@@ -531,7 +559,7 @@ TEST(FloodRetry, PayloadAndPathPolicyCapsEveryFloodType) {
packet.setPathHashSizeAndCount(1, 1);
uint8_t transit_limit;
if (type == PAYLOAD_TYPE_REQ) {
if (type == PAYLOAD_TYPE_REQ || type == PAYLOAD_TYPE_OTA) {
transit_limit = 0;
} else if (type == PAYLOAD_TYPE_GRP_TXT) {
transit_limit = 15;
tools/lora_ota/lora_ota.py
+65 -27
View File
@@ -2034,38 +2034,76 @@ def arm_target_temp_radio(
normal_radio: RadioSettings,
) -> None:
"""Resolve a lost TempRadio reply without blindly replaying the command."""
try:
reply = controller.remote_command(args.target, command, retry=False)
except TransmissionError as command_error:
print(
"[destination] TempRadio reply was lost; probing the declared "
"temporary channel"
)
controller.set_radio(temp_radio, "probe destination TempRadio state")
retries = 0
while True:
try:
identity = controller.remote_command(
args.target, "ota self", retry=False
reply = controller.remote_command(args.target, command, retry=False)
except TransmissionError as command_error:
print(
"[destination] TempRadio reply was lost; probing the declared "
"temporary channel"
)
if re.search(r"\bbase_hash=[0-9A-Fa-f]{16}\b", identity) is None:
raise OtaError(
"destination replied on TempRadio but did not return a valid "
"running EndF identity"
controller.set_radio(temp_radio, "probe destination TempRadio state")
found_on_temp = False
try:
identity = controller.remote_command(
args.target, "ota self", retry=False
)
except (OtaError, TransmissionError) as probe_error:
raise OtaError(
"destination TempRadio outcome is ambiguous; the controller was "
"restored to its normal channel and the command was not replayed"
) from probe_error
finally:
controller.set_radio(
normal_radio, "restore controller after TempRadio probe"
if re.search(r"\bbase_hash=[0-9A-Fa-f]{16}\b", identity) is None:
raise OtaError(
"destination replied on TempRadio but did not return a valid "
"running EndF identity"
)
found_on_temp = True
except (OtaError, TransmissionError):
pass
finally:
controller.set_radio(
normal_radio, "restore controller after TempRadio probe"
)
if found_on_temp:
print(
"[destination] resolved lost TempRadio reply from the exact "
"target identity; continuing"
)
return
# The 2-second scheduled handoff is long past by the time the
# temporary-channel probe times out. An exact identity reply back
# on the normal channel therefore proves that the target did not
# remain on TempRadio, making a bounded replay safe.
try:
identity = controller.remote_command(
args.target, "ota self", retry=False
)
if re.search(r"\bbase_hash=[0-9A-Fa-f]{16}\b", identity) is None:
raise OtaError(
"destination replied on the normal channel but did not "
"return a valid running EndF identity"
)
except (OtaError, TransmissionError) as normal_probe_error:
raise OtaError(
"destination TempRadio outcome is ambiguous; the controller "
"was restored to its normal channel and the command was not "
"replayed"
) from normal_probe_error
if retries >= TRANSMISSION_RETRY_LIMIT:
raise OtaError(
"destination remained on the normal channel after "
f"{retries + 1} verified TempRadio delivery attempts"
) from command_error
retries += 1
delay = transmission_retry_delay(retries)
print(
"[destination] exact normal-channel identity proves TempRadio is "
f"inactive; safe retry {retries}/{TRANSMISSION_RETRY_LIMIT} "
f"in {format_decimal(delay)}s"
)
print(
"[destination] resolved lost TempRadio reply from the exact target "
"identity; continuing"
)
time.sleep(delay)
continue
require_temp_radio_reply(args.target, reply)
return
require_temp_radio_reply(args.target, reply)
def temp_radio_command_for_minutes(
tools/lora_ota/rak3401_mota_chain.py
+180 -56
View File
@@ -30,14 +30,37 @@ except ImportError:
RELEASE_TAG = "rak3401-mota-v1.16.07-c1caa5ad-to-v1.17.01-c96bdd6e"
RELEASE_URL = f"https://github.com/mikecarper/MeshCore/releases/tag/{RELEASE_TAG}"
ASSET_NAME = "RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-c96bdd6e.zip"
ASSET_NAME = "RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-f7c2d382.zip"
ASSET_URL = (
f"https://github.com/mikecarper/MeshCore/releases/download/{RELEASE_TAG}/"
f"{ASSET_NAME}"
)
ASSET_SHA256 = "46c7480ed6bdc2aa01fb23a0f70e34c4012ffdd42b616d07bde66cf66d594630"
CHECKSUM_LIST_SHA256 = "8c74bbf9ae244f8c32041c68791b2780650c8a9321c23d32206614009739be5f"
BUNDLE_ROOT_NAME = "RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-c96bdd6e"
ASSET_SHA256 = "7dbfc603056a05e7e0a7db6d16bf21fa37d758f1fdd74766dc0adc79d6a85dbb"
CHECKSUM_LIST_SHA256 = "1223d01863d52e55774af0e44efd77730e860183fa80741488eecd0e0efe803a"
BUNDLE_ROOT_NAME = "RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-f7c2d382"
# This 27-step candidate passed steps 1 and 2, then its test was deliberately
# stopped so the adaptive primary requester could be backported into every
# historical source family. It remains valid for offline diagnosis only.
SUPERSEDED_27_ASSET_SHA256 = (
"2f81eec1f6cc2f3ab0fc246d7254766582118400f7b7de5e66ef4af582b0c337"
)
SUPERSEDED_27_CHECKSUM_LIST_SHA256 = (
"8861c6ecb7f9c805f5e87fb422a057f0b9d4b21099ee5dbc8aa96c557f7d87f9"
)
# The second v1.17.01 candidate reached step 16 before its checked CC310 path
# returned an incorrect digest for application flash. Preserve its exact pins
# for offline diagnosis, but never permit it to reach a live device.
KNOWN_FAILED_V11701_STEP16_ASSET_SHA256 = (
"46c7480ed6bdc2aa01fb23a0f70e34c4012ffdd42b616d07bde66cf66d594630"
)
KNOWN_FAILED_V11701_STEP16_CHECKSUM_LIST_SHA256 = (
"8c74bbf9ae244f8c32041c68791b2780650c8a9321c23d32206614009739be5f"
)
KNOWN_FAILED_V11701_ROOT_NAME = (
"RAK3401-update-chain-v1.16.7-c1caa5ad-to-v1.17.01-c96bdd6e"
)
# The first v1.17.01 candidate reused an unsafe source-transition bridge at
# step 15. Keep its hashes recognizable for offline diagnosis, but never allow
@@ -70,7 +93,7 @@ DEFAULT_TARGET_KEY = (
)
EXPECTED_TARGET_ID = 0x2FA509C1
EXPECTED_HARDWARE = "RAK_3401"
EXPECTED_STEP_COUNT = 26
EXPECTED_STEP_COUNT = 28
EXPECTED_START_VERSION = "1.16.7.0"
EXPECTED_FINAL_VERSION = "1.17.1.0"
MIN_MESHCLI_VERSION = (1, 6, 0)
@@ -89,22 +112,48 @@ KNOWN_UNSAFE_VERSION = "1.16.8.7"
KNOWN_UNSAFE_IMAGE_SHA256 = (
"61ced8b63953c614748c2fa1b04c2c01e8eb6626a604f6ef95fd2594d6d8ce71"
)
SAFE_CANDIDATE_STEP6_IMAGE_SHA256 = (
KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256 = (
"4bab3d2d6f6d3a033713d2db87565cb5f7fabe29b2b902b911724dd602fb7df8"
)
SUPERSEDED_27_STEP6_IMAGE_SHA256 = (
"16cd12a3cbbb563baf546191dc841b1255f11d3957d8c636366145430e31b617"
)
KNOWN_FAILED_V11701_STEP = 15
KNOWN_FAILED_V11701_VERSION = "1.16.9.111"
KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256 = (
"e8d9d1bd06217c7fd8d7fd333c6fcfde79000838550aa149b10be12fdc64fccb"
)
SAFE_CANDIDATE_STEP15_IMAGE_SHA256 = (
KNOWN_FAILED_V11701_SAFE_STEP15_IMAGE_SHA256 = (
"1124247f65772f11f9527408e51971eb9633ed656206276fa95019275bb8fdd2"
)
SUPERSEDED_27_STEP15_IMAGE_SHA256 = (
"82880e72dbbae41847ade865da06ed26f90d239a7716b1415cfe729b2d35bd94"
)
KNOWN_FAILED_V11701_STEP16 = 16
KNOWN_FAILED_V11701_STEP16_VERSION = "1.16.9.112"
KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256 = (
"35211ea70be635376b366a90afc74fcd8a7695f744f9c21a50bc872455ec5b21"
)
SUPERSEDED_27_STEP16_IMAGE_SHA256 = (
"5fd19be1cda7a0e9bdc9464be78abb238a3db5966538b1a1efdb949e8510e1d9"
)
SUPERSEDED_27_FINAL_IMAGE_SHA256 = (
"e1376869da043c05792b3458e505e67e818f58d94d3fc050d22ab68578a2f2e8"
)
# Exact anchors for the fresh 28-step candidate. The pinned checksum list
# covers every byte; these anchors also make the live gate fail closed if its
# structural checks are ever reused independently.
CURRENT_CANDIDATE_ANCHORS = (
(1, "09638e0e67faabdba97c621178294a5279c2d379540593738e2abad7afdc40cc"),
(6, "f5908b1e823b85268de8986871e10a78d288014a584e604254477520386daeaf"),
(11, "fa3e15db6a1818b5f4399a795d78f9274f709b58ccbd3268fda4e5973fa554af"),
(12, "959c4a98bf5d54c7aa17bad014830c0549b8439c3de6075ee97fea56849acb06"),
(13, "03b401f2752b9ed2aa442e86a49049aecd614fcdf9941beef0f77d3005b65471"),
(16, "ef06aad907d5822663bd0b7b46e189bc69e3b70e6b4474e8c004afb49ebdd828"),
(17, "c5defba43c561c4c73d49f557f5bc063b5db41ebcf5495934a64c003a652c009"),
(28, "c1d165a50ef2447d857f24af624e927a4f2e59978b6d072dbfc4a16189ea40da"),
)
KNOWN_UNSAFE_RELEASE_MESSAGE = (
f"live installation of {KNOWN_UNSAFE_RELEASE_TAG} is disabled: a physical RAK3401 "
f"test reached step {KNOWN_UNSAFE_STEP} (v{KNOWN_UNSAFE_VERSION}) but "
@@ -128,6 +177,12 @@ KNOWN_FAILED_V11701_STEP16_MESSAGE = (
"flash, so return-code fallback is not sufficient. Use --verify-only for "
"artifact inspection. Do not deploy this replaced candidate."
)
SUPERSEDED_27_MESSAGE = (
"live installation of the superseded 27-step candidate is disabled: its "
"physical test was stopped after step 2 so the adaptive primary requester "
"could be moved into every historical bridge. Use --verify-only for that "
"bundle and use the fresh 28-step candidate for the lab rerun."
)
class KnownUnsafeReleaseError(ota.OtaError):
@@ -138,28 +193,65 @@ def require_live_release_safe(
args: argparse.Namespace,
steps: list[ChainStep],
) -> None:
if len(steps) == EXPECTED_STEP_COUNT:
for number, expected_sha256 in CURRENT_CANDIDATE_ANCHORS:
if steps[number - 1].target_sha256 != expected_sha256:
raise KnownUnsafeReleaseError(
"live installation is disabled: the 28-step candidate has "
f"an unrecognized step-{number} image"
)
if not args.accept_test_candidate:
raise KnownUnsafeReleaseError(
"live installation of the fresh 28-step chain requires "
"--accept-test-candidate until its complete physical-board "
"test passes"
)
return
step6_sha256 = steps[KNOWN_UNSAFE_STEP - 1].target_sha256
if step6_sha256 == KNOWN_UNSAFE_IMAGE_SHA256:
raise KnownUnsafeReleaseError(KNOWN_UNSAFE_RELEASE_MESSAGE)
if step6_sha256 != SAFE_CANDIDATE_STEP6_IMAGE_SHA256:
if step6_sha256 == KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256:
if len(steps) < KNOWN_FAILED_V11701_STEP:
raise KnownUnsafeReleaseError(
"live installation is disabled: the failed v1.17.01 chain is "
"missing its audited step 15"
)
step15_sha256 = steps[KNOWN_FAILED_V11701_STEP - 1].target_sha256
if step15_sha256 == KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256:
raise KnownUnsafeReleaseError(KNOWN_FAILED_V11701_MESSAGE)
if step15_sha256 != KNOWN_FAILED_V11701_SAFE_STEP15_IMAGE_SHA256:
raise KnownUnsafeReleaseError(
"live installation is disabled: step 15 is not a recognized, "
"audited RAK3401 SHA-safe bridge image"
)
if len(steps) < KNOWN_FAILED_V11701_STEP16:
raise KnownUnsafeReleaseError(
"live installation is disabled: the failed v1.17.01 chain is "
"missing its audited step 16"
)
step16_sha256 = steps[KNOWN_FAILED_V11701_STEP16 - 1].target_sha256
if step16_sha256 == KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256:
raise KnownUnsafeReleaseError(KNOWN_FAILED_V11701_STEP16_MESSAGE)
raise KnownUnsafeReleaseError(
"live installation is disabled: step 6 is not a recognized, "
"audited RAK3401 bridge image"
"live installation is disabled: this is an unrecognized replacement "
"for a physically failed v1.17.01 chain"
)
step15_sha256 = steps[KNOWN_FAILED_V11701_STEP - 1].target_sha256
if step15_sha256 == KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256:
raise KnownUnsafeReleaseError(KNOWN_FAILED_V11701_MESSAGE)
if step15_sha256 != SAFE_CANDIDATE_STEP15_IMAGE_SHA256:
if step6_sha256 == SUPERSEDED_27_STEP6_IMAGE_SHA256:
if (
len(steps) == 27
and steps[14].target_sha256 == SUPERSEDED_27_STEP15_IMAGE_SHA256
and steps[15].target_sha256 == SUPERSEDED_27_STEP16_IMAGE_SHA256
and steps[-1].target_sha256 == SUPERSEDED_27_FINAL_IMAGE_SHA256
):
raise KnownUnsafeReleaseError(SUPERSEDED_27_MESSAGE)
raise KnownUnsafeReleaseError(
"live installation is disabled: step 15 is not a recognized, "
"audited RAK3401 SHA-safe bridge image"
"live installation is disabled: this is an unrecognized variant "
"of the superseded 27-step candidate"
)
step16_sha256 = steps[KNOWN_FAILED_V11701_STEP16 - 1].target_sha256
if step16_sha256 == KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256:
raise KnownUnsafeReleaseError(KNOWN_FAILED_V11701_STEP16_MESSAGE)
raise KnownUnsafeReleaseError(
"live installation is disabled: step 16 is not a recognized, "
"physically tested software-SHA bridge image"
"live installation is disabled: step 6 is not a recognized, audited "
"RAK3401 bridge image"
)
@@ -236,7 +328,11 @@ def download_release_asset(destination: Path) -> None:
def extract_bundle(archive_path: Path, destination: Path) -> Path:
root_names = (BUNDLE_ROOT_NAME, KNOWN_UNSAFE_ROOT_NAME)
root_names = (
BUNDLE_ROOT_NAME,
KNOWN_FAILED_V11701_ROOT_NAME,
KNOWN_UNSAFE_ROOT_NAME,
)
existing_roots = [destination / name for name in root_names if (destination / name).is_dir()]
if len(existing_roots) == 1:
return existing_roots[0]
@@ -284,7 +380,11 @@ def locate_bundle(args: argparse.Namespace, work_dir: Path) -> Path:
if supplied.is_dir():
direct = supplied / "CHAIN.csv"
nested_roots = [
supplied / name for name in (BUNDLE_ROOT_NAME, KNOWN_UNSAFE_ROOT_NAME)
supplied / name for name in (
BUNDLE_ROOT_NAME,
KNOWN_FAILED_V11701_ROOT_NAME,
KNOWN_UNSAFE_ROOT_NAME,
)
if (supplied / name / "CHAIN.csv").is_file()
]
if direct.is_file():
@@ -297,7 +397,9 @@ def locate_bundle(args: argparse.Namespace, work_dir: Path) -> Path:
actual = sha256_file(supplied)
if actual not in {
ASSET_SHA256,
SUPERSEDED_27_ASSET_SHA256,
KNOWN_FAILED_V11701_ASSET_SHA256,
KNOWN_FAILED_V11701_STEP16_ASSET_SHA256,
KNOWN_UNSAFE_ASSET_SHA256,
}:
raise ota.OtaError(
@@ -313,7 +415,9 @@ def verify_checksum_list(bundle_root: Path) -> None:
checksum_digest = sha256_file(checksum_path)
expected_lists = {
CHECKSUM_LIST_SHA256,
SUPERSEDED_27_CHECKSUM_LIST_SHA256,
KNOWN_FAILED_V11701_CHECKSUM_LIST_SHA256,
KNOWN_FAILED_V11701_STEP16_CHECKSUM_LIST_SHA256,
KNOWN_UNSAFE_CHECKSUM_LIST_SHA256,
}
if checksum_digest not in expected_lists:
@@ -367,9 +471,10 @@ def parse_chain(bundle_root: Path) -> tuple[list[ChainStep], bytes]:
rows = list(csv.DictReader(source))
except OSError as exc:
raise ota.OtaError(f"cannot read {chain_path}: {exc}") from exc
if len(rows) != EXPECTED_STEP_COUNT:
if len(rows) not in (26, 27, EXPECTED_STEP_COUNT):
raise ota.OtaError(
f"chain contains {len(rows)} steps, expected {EXPECTED_STEP_COUNT}"
f"chain contains {len(rows)} steps, expected 26, 27, or "
f"{EXPECTED_STEP_COUNT}"
)
steps: list[ChainStep] = []
@@ -419,38 +524,52 @@ def parse_chain(bundle_root: Path) -> tuple[list[ChainStep], bytes]:
if steps[0].from_version != EXPECTED_START_VERSION:
raise ota.OtaError("chain has an unexpected starting version")
step6 = steps[KNOWN_UNSAFE_STEP - 1]
recognized_step6_hashes = {
KNOWN_UNSAFE_IMAGE_SHA256,
SAFE_CANDIDATE_STEP6_IMAGE_SHA256,
}
if (
step6.to_version != KNOWN_UNSAFE_VERSION
or step6.target_sha256 not in recognized_step6_hashes
):
raise ota.OtaError(
"bundle does not contain a recognized audited step-6 image; "
"review and repin the runner before using it"
)
if step6.target_sha256 == SAFE_CANDIDATE_STEP6_IMAGE_SHA256:
step15 = steps[KNOWN_FAILED_V11701_STEP - 1]
recognized_step15_hashes = {
KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256,
SAFE_CANDIDATE_STEP15_IMAGE_SHA256,
if len(steps) == EXPECTED_STEP_COUNT:
for number, expected_sha256 in CURRENT_CANDIDATE_ANCHORS:
if steps[number - 1].target_sha256 != expected_sha256:
raise ota.OtaError(
f"fresh candidate step {number} does not match its audited "
"image pin"
)
expected_final_version = EXPECTED_FINAL_VERSION
else:
step6 = steps[KNOWN_UNSAFE_STEP - 1]
recognized_step6_hashes = {
KNOWN_UNSAFE_IMAGE_SHA256,
KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256,
SUPERSEDED_27_STEP6_IMAGE_SHA256,
}
if (
step15.to_version != KNOWN_FAILED_V11701_VERSION
or step15.target_sha256 not in recognized_step15_hashes
step6.to_version != KNOWN_UNSAFE_VERSION
or step6.target_sha256 not in recognized_step6_hashes
):
raise ota.OtaError(
"bundle does not contain a recognized audited step-15 image; "
"bundle does not contain a recognized audited step-6 image; "
"review and repin the runner before using it"
)
expected_final_version = (
"1.17.2.0"
if step6.target_sha256 == KNOWN_UNSAFE_IMAGE_SHA256
else EXPECTED_FINAL_VERSION
)
if step6.target_sha256 in {
KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256,
SUPERSEDED_27_STEP6_IMAGE_SHA256,
}:
step15 = steps[KNOWN_FAILED_V11701_STEP - 1]
recognized_step15_hashes = {
KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256,
KNOWN_FAILED_V11701_SAFE_STEP15_IMAGE_SHA256,
SUPERSEDED_27_STEP15_IMAGE_SHA256,
}
if (
step15.to_version != KNOWN_FAILED_V11701_VERSION
or step15.target_sha256 not in recognized_step15_hashes
):
raise ota.OtaError(
"bundle does not contain a recognized audited step-15 image; "
"review and repin the runner before using it"
)
expected_final_version = (
"1.17.2.0"
if step6.target_sha256 == KNOWN_UNSAFE_IMAGE_SHA256
else EXPECTED_FINAL_VERSION
)
if steps[-1].to_version != expected_final_version:
raise ota.OtaError(
f"chain ends at {steps[-1].to_version}, expected {expected_final_version}"
@@ -854,8 +973,8 @@ def run_step(
def build_parser() -> argparse.ArgumentParser:
parser = argparse.ArgumentParser(
description=(
"Download, verify, resume, and install the exact 26-step RAK3401 "
"mOTA chain from c1caa5ad to c96bdd6e."
"Download, verify, resume, and install the exact 28-step RAK3401 "
"mOTA chain from c1caa5ad to f7c2d382."
)
)
parser.add_argument(
@@ -874,8 +993,8 @@ def build_parser() -> argparse.ArgumentParser:
"--accept-test-candidate",
action="store_true",
help=(
"reserved lab gate for a future pinned candidate; it never overrides "
"a bundle known to have failed physical testing"
"allow the pinned fresh 28-step chain for its direct physical lab test; "
"it never overrides a bundle known to have failed physical testing"
),
)
@@ -1058,6 +1177,11 @@ def main(argv: list[str] | None = None) -> int:
f"WARNING: {KNOWN_FAILED_V11701_STEP16_MESSAGE}",
file=sys.stderr,
)
elif (
len(steps) == 27
and steps[-1].target_sha256 == SUPERSEDED_27_FINAL_IMAGE_SHA256
):
print(f"WARNING: {SUPERSEDED_27_MESSAGE}", file=sys.stderr)
else:
print(
"Unfailed candidate verified offline; complete physical-board testing "
@@ -1168,7 +1292,7 @@ def main(argv: list[str] | None = None) -> int:
if not enabled.lower().startswith("ok - system watchdog enabled"):
raise ota.OtaError(f"could not enable system watchdog: {enabled}")
require_watchdog_state(controller, target_name, "on")
print("[watchdog] re-enabled after verified step 26 boot")
print(f"[watchdog] re-enabled after verified step {len(steps)} boot")
print(
f"RAK3401 update complete: {final_target.current_version} "
f"{final_target.base_hash.hex().upper()}"
tools/lora_ota/test_lora_ota.py
+112 -8
View File
@@ -835,10 +835,60 @@ class ReliabilityTests(unittest.TestCase):
)
self.assertEqual(
controller.commands,
["tempradio 909.95,250,5,5,120", "ota self"],
["tempradio 909.95,250,5,5,120", "ota self", "ota self"],
)
self.assertEqual(controller.radios, [temporary, normal])
@mock.patch.object(ota.time, "sleep")
def test_target_temp_retries_only_after_exact_normal_identity(
self, sleep: mock.Mock
) -> None:
normal = ota.RadioSettings(910.525, 62.5, 7, 5, False)
temporary = ota.RadioSettings(909.95, 500.0, 5, 5, False)
class Controller:
def __init__(self) -> None:
self.commands: list[str] = []
self.radios: list[ota.RadioSettings] = []
self.replies = iter([
ota.TransmissionError("lost command reply"),
ota.TransmissionError("not on temporary channel"),
"self body=1 image=2 base_hash=0011223344556677",
"OK - temp params for 120 mins",
])
def remote_command(
self, _target: str, command: str, **_kwargs: object
) -> str:
self.commands.append(command)
reply = next(self.replies)
if isinstance(reply, Exception):
raise reply
return reply
def set_radio(self, radio: ota.RadioSettings, _label: str) -> None:
self.radios.append(radio)
controller = Controller()
ota.arm_target_temp_radio(
controller,
argparse.Namespace(target="remote"),
"tempradio 909.95,500,5,5,120",
temporary,
normal,
)
self.assertEqual(
controller.commands,
[
"tempradio 909.95,500,5,5,120",
"ota self",
"ota self",
"tempradio 909.95,500,5,5,120",
],
)
self.assertEqual(controller.radios, [temporary, normal])
sleep.assert_called_once_with(ota.transmission_retry_delay(1))
def test_install_retries_only_after_still_ready_is_confirmed(self) -> None:
image = firmware(b"install" * 900, VERSION_NEW)
package = ota.parse_mota(mota_blob(image))
@@ -1191,10 +1241,14 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
argparse.Namespace(accept_test_candidate=False), steps
)
def test_corrected_chain_requires_explicit_lab_gate(self) -> None:
def test_second_failed_chain_is_always_blocked(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(16)]
steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
steps[14].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP15_IMAGE_SHA256
steps[5].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256
)
steps[14].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP15_IMAGE_SHA256
)
steps[15].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256
)
@@ -1215,8 +1269,12 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
def test_checked_cc310_step16_is_always_blocked(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(16)]
steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
steps[14].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP15_IMAGE_SHA256
steps[5].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256
)
steps[14].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP15_IMAGE_SHA256
)
steps[15].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_STEP16_IMAGE_SHA256
)
@@ -1230,7 +1288,9 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
def test_first_v11701_candidate_is_blocked_after_failed_step15(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(15)]
steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
steps[5].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256
)
steps[14].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_STEP15_IMAGE_SHA256
)
@@ -1244,7 +1304,9 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
def test_unrecognized_step15_is_blocked(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(15)]
steps[5].target_sha256 = rak_chain.SAFE_CANDIDATE_STEP6_IMAGE_SHA256
steps[5].target_sha256 = (
rak_chain.KNOWN_FAILED_V11701_SAFE_STEP6_IMAGE_SHA256
)
steps[14].target_sha256 = "00" * 32
with self.assertRaisesRegex(
rak_chain.KnownUnsafeReleaseError,
@@ -1254,6 +1316,48 @@ class Rak3401KnownUnsafeReleaseTests(unittest.TestCase):
argparse.Namespace(accept_test_candidate=True), steps
)
def test_fresh_28_step_candidate_requires_explicit_lab_gate(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(28)]
for number, image_sha256 in rak_chain.CURRENT_CANDIDATE_ANCHORS:
steps[number - 1].target_sha256 = image_sha256
with self.assertRaisesRegex(
rak_chain.KnownUnsafeReleaseError,
"requires --accept-test-candidate",
):
rak_chain.require_live_release_safe(
argparse.Namespace(accept_test_candidate=False), steps
)
rak_chain.require_live_release_safe(
argparse.Namespace(accept_test_candidate=True), steps
)
def test_fresh_28_step_candidate_rejects_changed_anchor(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(28)]
for number, image_sha256 in rak_chain.CURRENT_CANDIDATE_ANCHORS:
steps[number - 1].target_sha256 = image_sha256
steps[10].target_sha256 = "00" * 32
with self.assertRaisesRegex(
rak_chain.KnownUnsafeReleaseError,
"unrecognized step-11 image",
):
rak_chain.require_live_release_safe(
argparse.Namespace(accept_test_candidate=True), steps
)
def test_superseded_27_step_candidate_is_always_blocked(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(27)]
steps[5].target_sha256 = rak_chain.SUPERSEDED_27_STEP6_IMAGE_SHA256
steps[14].target_sha256 = rak_chain.SUPERSEDED_27_STEP15_IMAGE_SHA256
steps[15].target_sha256 = rak_chain.SUPERSEDED_27_STEP16_IMAGE_SHA256
steps[-1].target_sha256 = rak_chain.SUPERSEDED_27_FINAL_IMAGE_SHA256
with self.assertRaisesRegex(
rak_chain.KnownUnsafeReleaseError,
"superseded 27-step candidate",
):
rak_chain.require_live_release_safe(
argparse.Namespace(accept_test_candidate=True), steps
)
def test_unrecognized_step6_is_blocked(self) -> None:
steps = [mock.Mock(target_sha256="") for _ in range(6)]
steps[5].target_sha256 = "00" * 32