Accelerate adaptive LoRa OTA transfers
This commit is contained in:
mikecarper committed
1 parent
e92794eabc
commit
8c5262c6f8
20 files changed
+1059
-307
No files matched your search
+40
-20
@@ -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.
|
||||
|
||||
---
|
||||
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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;
|
||||
}
|
||||
|
||||
|
||||
@@ -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;
|
||||
|
||||
+11
-3
@@ -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
|
||||
|
||||
@@ -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();
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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;
|
||||
};
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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
|
||||
|
||||
+210
-107
@@ -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 <N>` (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;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
#include <stdint.h>
|
||||
#include <stddef.h>
|
||||
#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)
|
||||
|
||||
@@ -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);
|
||||
|
||||
@@ -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<<nf)-1) for a fresh block and only the
|
||||
// missing bits to recover holes, so a lost fragment costs one fragment to re-send - not the whole block
|
||||
// (saves airtime and avoids the re-REQ colliding with a multi-fragment burst on half-duplex radios).
|
||||
// ---- OTA_REQ: request specific fragments of one or more blocks (direct) ----
|
||||
// body: manifest_id(4) { block_idx(2) want_mask(2) }[1..4]. The first row is exactly the original
|
||||
// single-block OTA_REQ body, so an older source safely serves that row and ignores appended rows. A newer
|
||||
// source queues every row. This lets a fetcher open one adaptive flight with one packet, then remain silent
|
||||
// while the source and relays return the requested DATA/PROOF train.
|
||||
//
|
||||
// want_mask bit k = "send fragment k" (the slice at byte offset k*OTA_FRAG_DATA). A recovery request uses
|
||||
// only the missing bits, so one lost fragment costs one fragment to re-send rather than the whole block.
|
||||
struct ReqMsg { uint8_t manifest_id[4]; uint16_t block_idx; uint16_t want_mask; };
|
||||
struct ReqItem { uint16_t block_idx; uint16_t want_mask; };
|
||||
#ifndef OTA_REQ_MAX_ITEMS
|
||||
#define OTA_REQ_MAX_ITEMS 4
|
||||
#endif
|
||||
struct ReqWindowMsg {
|
||||
uint8_t manifest_id[4];
|
||||
uint8_t n_items;
|
||||
ReqItem items[OTA_REQ_MAX_ITEMS];
|
||||
};
|
||||
|
||||
// ---- 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[]
|
||||
@@ -122,6 +134,8 @@ bool decode_manifest(const uint8_t* buf, uint16_t len, ManifestMsg& m);
|
||||
|
||||
uint16_t encode_req(uint8_t* buf, uint16_t cap, const ReqMsg& m);
|
||||
bool decode_req(const uint8_t* buf, uint16_t len, ReqMsg& m);
|
||||
uint16_t encode_req_window(uint8_t* buf, uint16_t cap, const ReqWindowMsg& m);
|
||||
bool decode_req_window(const uint8_t* buf, uint16_t len, ReqWindowMsg& m);
|
||||
|
||||
uint16_t encode_data(uint8_t* buf, uint16_t cap, const DataMsg& m);
|
||||
bool decode_data(const uint8_t* buf, uint16_t len, DataMsg& m);
|
||||
|
||||
@@ -2,14 +2,21 @@
|
||||
#include <stdint.h>
|
||||
|
||||
// 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;
|
||||
}
|
||||
|
||||
|
||||
+258
-114
@@ -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<uint8_t> 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<uint8_t> 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) {
|
||||
|
||||
+257
-2
@@ -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 (
|
||||
|
||||
@@ -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(
|
||||
|
||||
@@ -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;",
|
||||
"}",
|
||||
"",
|
||||
|
||||
@@ -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}
|
||||
+<helpers/ota/*.cpp>
|
||||
+<helpers/ui/SSD1306Display.cpp>
|
||||
|
||||
Reference in new issue
Block a user