From 5da783040f67007d515b4cab6d44fde2a4f1d20f Mon Sep 17 00:00:00 2001 From: Valentin Kivachuk Burda Date: Mon, 29 Jun 2026 13:03:05 +0200 Subject: [PATCH] ota: LoRa transfer protocol + session manager (discovery, block fetch/serve) --- src/helpers/ota/MotaSeederProto.h | 44 ++ src/helpers/ota/OtaContext.cpp | 12 + src/helpers/ota/OtaContext.h | 191 ++++++++ src/helpers/ota/OtaDebug.h | 11 + src/helpers/ota/OtaManager.cpp | 780 ++++++++++++++++++++++++++++++ src/helpers/ota/OtaManager.h | 358 ++++++++++++++ src/helpers/ota/OtaProtocol.cpp | 137 ++++++ src/helpers/ota/OtaProtocol.h | 112 +++++ src/helpers/ota/OtaSource.h | 55 +++ src/helpers/ota/OtaStore.h | 103 ++++ 10 files changed, 1803 insertions(+) create mode 100644 src/helpers/ota/MotaSeederProto.h create mode 100644 src/helpers/ota/OtaContext.cpp create mode 100644 src/helpers/ota/OtaContext.h create mode 100644 src/helpers/ota/OtaDebug.h create mode 100644 src/helpers/ota/OtaManager.cpp create mode 100644 src/helpers/ota/OtaManager.h create mode 100644 src/helpers/ota/OtaProtocol.cpp create mode 100644 src/helpers/ota/OtaProtocol.h create mode 100644 src/helpers/ota/OtaSource.h create mode 100644 src/helpers/ota/OtaStore.h diff --git a/src/helpers/ota/MotaSeederProto.h b/src/helpers/ota/MotaSeederProto.h new file mode 100644 index 00000000..51767c9c --- /dev/null +++ b/src/helpers/ota/MotaSeederProto.h @@ -0,0 +1,44 @@ +#pragma once + +#include + +// Wire contract for the "mota-seeder" link: a device (CLIENT) pulls `.mota` bytes on demand from a host +// daemon (SERVER) that owns a folder of `.mota` files. This is the FIRST concrete MotaSource transport +// (docs/ota_protocol.md §9) — the device speaks it over a dedicated Stream (a spare UART / USB-UART), so +// it never contends with the line-based text CLI on the main console. +// +// The device always initiates; every request gets exactly one response. Framing is resync-safe: the +// reader scans for the 2-byte magic, so line noise / a half-read frame just times out and is retried +// (OTA is lowest priority — eventually-upgradable). All multi-byte fields are little-endian. +// +// request (device -> host): 'M' 'S' op(1) args... xsum(1 = XOR of op+args) +// response (host -> device): 'm' 's' op(1) status(1) payload... xsum(1 = XOR of all prior) +// +// OP_COUNT 0x01 args: - resp payload: count(1) +// OP_DESCRIBE 0x02 args: idx(1) resp payload: MotaDesc wire (38 B, see below) [status OK] +// OP_READ 0x03 args: idx(1) off(4) len(2) resp payload: len bytes [status OK] +// +// MotaDesc wire (38 B): mid[4] target_id(4) fw_version(4) codec(1) flags(1) total_size(4) +// leaves_off(4) block_count(4) payload_off(4) payload_size(4) +// +// status: 0 = OK, non-zero = error (idx out of range, read past EOF, ...). On error the response carries +// no payload (just magic+op+status+xsum). + +namespace mesh { +namespace ota { + +static const uint8_t MOTA_SEEDER_REQ_MAGIC0 = 'M'; +static const uint8_t MOTA_SEEDER_REQ_MAGIC1 = 'S'; +static const uint8_t MOTA_SEEDER_RSP_MAGIC0 = 'm'; +static const uint8_t MOTA_SEEDER_RSP_MAGIC1 = 's'; + +static const uint8_t MS_OP_COUNT = 0x01; +static const uint8_t MS_OP_DESCRIBE = 0x02; +static const uint8_t MS_OP_READ = 0x03; + +static const uint8_t MS_STATUS_OK = 0x00; + +static const uint16_t MOTA_DESC_WIRE = 38; // bytes of a MotaDesc on the wire (see layout above) + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaContext.cpp b/src/helpers/ota/OtaContext.cpp new file mode 100644 index 00000000..10f2bba4 --- /dev/null +++ b/src/helpers/ota/OtaContext.cpp @@ -0,0 +1,12 @@ +#include "OtaContext.h" + +namespace mesh { +namespace ota { + +OtaContext& ota_ctx() { + static OtaContext ctx; + return ctx; +} + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaContext.h b/src/helpers/ota/OtaContext.h new file mode 100644 index 00000000..2a67c238 --- /dev/null +++ b/src/helpers/ota/OtaContext.h @@ -0,0 +1,191 @@ +#pragma once + +#include // snprintf (hw_id mismatch message) +#include // strncmp/strncpy (hw_id) +#include "OtaManager.h" +#include "OtaStore.h" +#include "SignerAllowlist.h" +#include "OtaApply.h" +#include "OtaFormat.h" +#include "OtaSelf.h" // ota_self_firmware() — prefer self-describing EndF identity at begin() +#include "OtaBlInfo.h" // bootloader OTA-apply capability marker (nRF52); cached after first read +#if defined(NRF52_PLATFORM) && defined(OTA_FLASH_STORE) + #include "OtaStoreFlashNrf52.h" +#elif defined(ESP32_PLATFORM) && defined(OTA_FLASH_STORE) + #include "OtaStoreFlashEsp32.h" +#endif +#if defined(OTA_FOLDER_SERIAL) + #include "MotaSourceSerial.h" // relay an external folder served by a host daemon over the USB serial + #ifndef OTA_FOLDER_SERIAL_STREAM + #define OTA_FOLDER_SERIAL_STREAM Serial // default: the same USB console the CLI uses (no extra HW) + #endif + #ifndef OTA_FOLDER_SERIAL_BAUD + #define OTA_FOLDER_SERIAL_BAUD 115200 + #endif + // The console Serial is already begun by the example; a DEDICATED UART (override the stream) needs init, + // so define OTA_FOLDER_SERIAL_BEGIN to have attach_folder() call .begin(baud) on it. +#endif + +// Per-device OTA singleton shared by the CLI (OtaCli) and the mesh adapter (the example's MyMesh). +// Holds the session engine, a staging store (fetch), a RAM serve buffer, and the signer allowlist. +// nRF52 stages into FLASH (OtaStoreFlashNrf52): a delta can be 100 KB+, too big to hold in RAM, and the +// COMPLETE container must persist so the bootloader can apply it after reboot. A flash page-erase halts +// the CPU (~85 ms) and starves the LoRa RX, so the store COALESCES writes to the 4 KB page (the erase +// unit) and commits each page once, off the per-packet path (see OtaManager.h) — RAM stays O(one page). +// (v1 has no mid-transfer resume; an interrupted fetch simply restarts.) ESP32/native use the RAM store. + +namespace mesh { +namespace ota { + +#ifndef OTA_SERVE_BUF_SIZE +#define OTA_SERVE_BUF_SIZE 16384 +#endif +#ifndef OTA_FETCH_BUF_SIZE +#define OTA_FETCH_BUF_SIZE 16384 +#endif + +struct OtaContext { + OtaManager manager; +#if defined(NRF52_PLATFORM) && defined(OTA_FLASH_STORE) + OtaStoreFlashNrf52 fetch_store; // persistent flash staging (survives reboot; large deltas) +#elif defined(ESP32_PLATFORM) && defined(OTA_FLASH_STORE) + OtaStoreFlashEsp32 fetch_store; // stages in the inactive A/B slot (delta + full, RX-safe) +#else + OtaStoreRam fetch_store; +#endif + SignerAllowlist allow; + uint8_t serve_buf[OTA_SERVE_BUF_SIZE]; + uint32_t serve_expected = 0; // size declared by `ota stage` + bool serving = false; // manager.serve() succeeded + // flash-backed self-serve: cached merkle leaves (heap, freed on re-serve) + assembled manifest of our + // own running firmware. The payload is read from flash per block; only the metadata is held in RAM. + // serve_self_proof is the proof-gen working buffer (>= block_count*4) — sized to OUR image's block + // count (the manager's fixed 4 KB scratch only covers <=1024 blocks; a >1 MB image needs more). + uint8_t* serve_self_leaves = nullptr; + uint8_t* serve_self_proof = nullptr; + uint8_t serve_self_manifest[MOTA_MFL]; // fixed-layout full+unsigned manifest-minus-leaves (197 B) + ApplyState apply_st; // pending apply (P6) + + // OTA policy (persisted via NodePrefs; autofetch lives in the manager). Conservative defaults: a fresh + // node discovers + announces but never fetches/installs without operator intent. + static const uint8_t AUTOINSTALL_OFF = 0, AUTOINSTALL_TRUSTED = 1; + uint8_t autoinstall = AUTOINSTALL_OFF; // 1 = auto-apply a COMPLETE fetch IF signed + allowlisted + bool config_dirty = false; // CLI set a policy/key -> CommonCLI persists + clears + char hw_id[33] = {0}; // this device's hardware tag (from board.getOtaHwId(), set in begin) + + // True if the staged .mota's hw_id is compatible with this device: equal tags, or either side empty + // ("unknown" -> can't enforce -> permissive). Brick-safety gate for apply (esp. manual cross-target). + bool hwMatches(const uint8_t* mhw /*32B, may be null*/) const { + if (!hw_id[0] || !mhw) return true; + bool declared = false; for (int i = 0; i < 32; i++) if (mhw[i]) { declared = true; break; } + if (!declared) return true; + return strncmp((const char*)mhw, hw_id, 32) == 0; + } + + // Apply the COMPLETE fetched .mota (platform dispatch) and arm the slot; sets apply_pending on success + // so the deferred-reboot path (mesh loop) takes over. Caller ensures the fetch is COMPLETE. Shared by + // manual `ota applydelta` and the auto-install path. + bool apply_fetched(char* msg) { + // hardware-compatibility gate (brick-safety) — refuse a .mota whose hw_id is for different hardware, + // independent of signature; covers a manual cross-target `ota dev want` onto an incompatible board. + { + uint8_t hdr[8], mb[256]; + uint32_t total = fetch_store.staged_size(); + if (total >= 13 && fetch_store.read(0, hdr, 8) && memcmp(hdr, MOTA_MAGIC, 4) == 0) { + uint32_t mr = total - 8; if (mr > sizeof(mb)) mr = sizeof(mb); + MotaManifest mm; + if (fetch_store.read(8, mb, mr) && mota_parse_manifest(mb, mr, mm) && !hwMatches(mm.hw_id)) { + char want[33] = {0}; memcpy(want, mm.hw_id, 32); + snprintf(msg, 96, "refused: .mota hw_id '%.32s' != this device '%s' (incompatible hardware)", want, hw_id); + return false; + } + } + } + bool ok; +#if defined(NRF52_PLATFORM) + ok = ota_apply_mota_nrf52(fetch_store.data(), fetch_store.staged_size(), allow, apply_st, msg); +#elif defined(ESP32_PLATFORM) && defined(OTA_FLASH_STORE) + ok = ota_apply_detools_mota(fetch_store, allow, apply_st, msg); +#else + ok = ota_apply_detools_mota(fetch_store.data(), fetch_store.staged_size(), allow, apply_st, msg); +#endif + if (ok) apply_pending = true; + return ok; + } + + // Deferred apply-reboot: a verified `ota applydelta` approves the update but does NOT reboot inline, + // so the CLI can first deliver the "verified; applying" reply (over LoRa it's the only way the + // operator learns the apply started). The mesh loop then calls ota_reboot_to_apply() once that reply + // has actually been transmitted. apply_at/apply_hard are mesh-clock deadlines the loop fills in. + bool apply_pending = false; + uint32_t apply_at = 0; // earliest reboot time (lets the reply get queued + start sending) + uint32_t apply_hard = 0; // hard cap, in case the TX queue never idles on a busy node + + // Bootloader OTA-apply capability (nRF52): can THIS device's bootloader apply a .mota? Read from flash + // ONCE (the scan is ~40 KB) and cached in RAM. On other platforms present=false (apply is in-app). + OtaBlCaps _bl_caps; + bool _bl_caps_read = false; + const OtaBlCaps& bootloaderCaps() { + if (!_bl_caps_read) { _bl_caps = ota_bootloader_caps(); _bl_caps_read = true; } + return _bl_caps; + } + + // --- discovery: the "what mOTAs are available around me" view ---------------------------------- + // The catalog (heard mOTAs) + the heard-sources table now live in OtaManager (built from beacons + + // OTA_HAVE catalog replies, the two-tier discovery). `ota neighbors` renders manager.catalogRow(); + // `ota pull` acts on a mid. Here we only keep the fetch-session age stamp. + uint32_t session_started_ms = 0; // when the fetch session last left IDLE (for the age display) + uint8_t prev_fstate = OtaManager::IDLE; + bool folder_active = false; // an external `.mota` folder is attached + being relayed + + // Attach/detach an external folder of `.mota` served by a host daemon over the seeder UART (the node + // then advertises + relays them alongside its own fw). Only built when OTA_FOLDER_SERIAL is configured. +#if defined(OTA_FOLDER_SERIAL) + bool attach_folder(char* msg, size_t cap) { + static SerialMotaSource src(OTA_FOLDER_SERIAL_STREAM, 600); +#ifdef OTA_FOLDER_SERIAL_BEGIN + OTA_FOLDER_SERIAL_STREAM.begin(OTA_FOLDER_SERIAL_BAUD); // dedicated UART; console is already up +#endif + manager.clear_sources(); // idempotent re-attach + if (!manager.add_source(&src)) { strncpy(msg, "ERR no free source slot", cap); return false; } + folder_active = true; + snprintf(msg, cap, "OK folder attached (serial) — serving %u mOTA total (own fw + folder)", + (unsigned)manager.servedCount()); + return true; + } +#endif + void detach_folder() { manager.clear_sources(); folder_active = false; } + + void track_session(uint8_t fstate, uint32_t now) { // stamp the session start (age display) + if (fstate != prev_fstate) { + if (prev_fstate == OtaManager::IDLE && fstate != OtaManager::IDLE) session_started_ms = now; + prev_fstate = fstate; + } + } + + void begin(uint32_t target_id, OtaSend send, void* ctx, const char* hw = nullptr) { + // Prefer the firmware's SELF-DESCRIBING EndF identity (docs §2) over the build-flag values the caller + // passed — it's correct on any build (build.sh injection, bare IDE build, ...), so `ota ls`/`status` + // and fetch-routing show the right hardware/role instead of 0 / "". + SelfFwInfo _fi; + if (ota_self_firmware(_fi) && _fi.valid) { + if (_fi.target_id) target_id = _fi.target_id; + if (_fi.hw_id[0]) hw = _fi.hw_id; + } + manager.begin(target_id, send, ctx); + if (hw) { strncpy(hw_id, hw, sizeof(hw_id) - 1); hw_id[sizeof(hw_id) - 1] = 0; } + // a node only fetches firmware it can apply: ESP32 A/B -> sequential, nRF52 single-slot -> in-place +#if defined(NRF52_PLATFORM) + manager.set_apply_codec(CODEC_DETOOLS_INPLACE); +#elif defined(ESP32_PLATFORM) + manager.set_apply_codec(CODEC_DETOOLS_SEQUENTIAL); // preferred (streams straight to the slot) + manager.set_apply_codec2(CODEC_DETOOLS_INPLACE); // also accepted -> a single in-place .mota fits both +#endif + manager.set_fetch_store(&fetch_store); + } +}; + +OtaContext& ota_ctx(); // process-wide singleton + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaDebug.h b/src/helpers/ota/OtaDebug.h new file mode 100644 index 00000000..a126cdec --- /dev/null +++ b/src/helpers/ota/OtaDebug.h @@ -0,0 +1,11 @@ +#pragma once + +// Opt-in OTA tracing over Serial: build with -D OTA_DEBUG to watch the fetch (ADV/REQ/block/page-flush) +// during bring-up. Compiles to nothing otherwise, and on the native host (no Arduino), so it never +// touches a non-debug or test build. +#if defined(OTA_DEBUG) && defined(ARDUINO) + #include + #define OTA_DBG(...) do { Serial.printf(__VA_ARGS__); } while (0) +#else + #define OTA_DBG(...) do {} while (0) +#endif diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp new file mode 100644 index 00000000..e204a057 --- /dev/null +++ b/src/helpers/ota/OtaManager.cpp @@ -0,0 +1,780 @@ +#include "OtaManager.h" +#include "OtaProtocol.h" +#include "MerkleTree.h" +#include "Multihash.h" +#include "OtaByteIO.h" +#include "OtaDebug.h" +#include + +namespace mesh { +namespace ota { + +void OtaManager::begin(uint32_t my_target_id, OtaSend send, void* ctx) { + _target = my_target_id; _send = send; _ctx = ctx; + _fstate = IDLE; _have = 0; _fbc = 0; + _n_serve = 0; _n_src_obj = 0; _view0.valid = false; _srcv.valid = false; _fetch_served = false; + for (uint8_t i = 0; i < 8; i++) _recent_blk[i] = NO_BLOCK; // empty slot (never a real block index) +} + +// ---------------- serve (multi-mota registry) ---------------- +// +// A node offers a SET of mOTAs: its own firmware (view0) plus any external "folder" sources (OtaSource). +// Every fetch message carries the manifest_id, so a request dispatches to the matching ServeView via +// resolve() — view0 is resident; an external mota is (re)loaded on demand into _srcv. The catalog (what +// we advertise / answer OTA_QUERY with) is the lightweight _serve[] registry. + +bool OtaManager::serve(const uint8_t* mota, uint32_t len) { + if (!mota_parse(mota, len, _view0.m)) return false; + _view0.mfl = (uint16_t)(_view0.m.leaves - _view0.m.manifest_start); // contiguous container + _view0.read = nullptr; _view0.read_ctx = nullptr; // payload is contiguous _view0.m.payload + _view0.scratch = _scratch; _view0.scratch_sz = sizeof(_scratch); // <=1024 blocks (RAM .mota is small) + _view0.valid = true; + registerSelfEntry(); + return true; +} + +bool OtaManager::serve_self(const uint8_t* manifest, uint16_t mfl, const uint8_t* leaves, + uint32_t block_count, uint8_t* proof_scratch, uint32_t proof_scratch_sz, + ServeReadFn read, void* ctx) { + if (proof_scratch_sz < (uint64_t)block_count * 4) return false; // proof-gen needs count*4 working bytes + if (!mota_parse_manifest(manifest, mfl, _view0.m)) return false; // fixed fields: root, image_hash, sizes + _view0.m.manifest_start = manifest; + _view0.m.leaves = leaves; // pre-computed, caller-owned (heap) + _view0.m.payload = nullptr; // read on demand via `read` + _view0.m.block_count = block_count; + _view0.mfl = mfl; _view0.read = read; _view0.read_ctx = ctx; + _view0.scratch = proof_scratch; _view0.scratch_sz = proof_scratch_sz; // sized for our (large) image + _view0.valid = true; + registerSelfEntry(); + return true; +} + +// (Re)build registry slot 0 from view0 (our own fw / RAM mota). Keeps any source entries in [1..]. +void OtaManager::registerSelfEntry() { + if (!_view0.valid) return; + ServeEntry& e = _serve[0]; + memcpy(e.mid, _view0.m.merkle_root, 4); + e.target_id = _view0.m.target_id; e.fw_version = _view0.m.fw_version; + e.codec_id = _view0.m.codec_id; e.flags = _view0.m.flags; e.have_count = _view0.m.block_count; + e.is_self = true; e.is_fetch = false; e.src = nullptr; e.src_idx = 0; + if (_n_serve == 0) _n_serve = 1; +} + +bool OtaManager::add_source(MotaSource* src) { + if (!src || _n_src_obj >= OTA_MAX_SOURCE_OBJ) return false; + _src_list[_n_src_obj++] = src; + refresh_sources(); + return true; +} + +void OtaManager::refresh_sources() { + uint8_t base = _view0.valid ? 1 : 0; // entry 0 stays our own fw + if (_view0.valid) registerSelfEntry(); + _n_serve = base; + for (uint8_t s = 0; s < _n_src_obj; s++) { + MotaSource* src = _src_list[s]; + if (!src) continue; + uint8_t cnt = src->count(); + for (uint8_t i = 0; i < cnt && _n_serve < OTA_MAX_SERVE; i++) { + MotaDesc d; + if (!src->describe(i, d)) continue; + if (serveEntryIndex(d.mid) >= 0) continue; // already offered (e.g. our own fw in the folder) + ServeEntry& e = _serve[_n_serve++]; + memcpy(e.mid, d.mid, 4); + e.target_id = d.target_id; e.fw_version = d.fw_version; + e.codec_id = d.codec_id; e.flags = d.flags; e.have_count = d.block_count; // a folder mota is fully held + e.is_self = false; e.is_fetch = false; e.src = src; e.src_idx = i; e.desc = d; + } + } + // re-seed a completed download (epidemic spread) as one more served mota, backed by the fetch store + if (_fetch_served && _n_serve < OTA_MAX_SERVE && serveEntryIndex(_fetch_desc.mid) < 0) { + ServeEntry& e = _serve[_n_serve++]; + memcpy(e.mid, _fetch_desc.mid, 4); + e.target_id = _fetch_desc.target_id; e.fw_version = _fetch_desc.fw_version; + e.codec_id = _fetch_desc.codec_id; e.flags = _fetch_desc.flags; e.have_count = _fetch_desc.block_count; + e.is_self = false; e.is_fetch = true; e.src = nullptr; e.src_idx = 0; e.desc = _fetch_desc; + } + _srcv.valid = false; // a loaded source view may now be stale; reloads on demand +} + +void OtaManager::clear_sources() { + _n_src_obj = 0; _srcv.valid = false; + _n_serve = _view0.valid ? 1 : 0; + if (_view0.valid) registerSelfEntry(); +} + +int OtaManager::serveEntryIndex(const uint8_t* mid) const { + for (uint8_t i = 0; i < _n_serve; i++) + if (memcmp(_serve[i].mid, mid, 4) == 0) return i; + return -1; +} + +OtaManager::ServeView* OtaManager::resolve(const uint8_t* mid) { + if (_view0.valid && memcmp(mid, _view0.m.merkle_root, 4) == 0) return &_view0; + if (_srcv.valid && memcmp(mid, _srcv_mid, 4) == 0) return &_srcv; + int i = serveEntryIndex(mid); + if (i < 0) return nullptr; + if (_serve[i].is_self) return _view0.valid ? &_view0 : nullptr; + return loadSource(_serve[i]) ? &_srcv : nullptr; +} + +// Load an external mota into the on-demand _srcv: read its manifest-minus-leaves + leaves[] from the +// source into RAM, parse, and wire a payload reader that streams blocks from the source on REQ. (The +// payload itself is NOT held in RAM — only the small head + the leaves, <=4 KB for <=1024 blocks.) +bool OtaManager::loadSource(const ServeEntry& e) { + const MotaDesc& d = e.desc; + if (d.leaves_off < 8) return false; + if (e.is_fetch ? (_fetch == nullptr) : (e.src == nullptr)) return false; + uint16_t mfl = (uint16_t)(d.leaves_off - 8); + if (mfl == 0 || mfl > sizeof(_src_manifest)) return false; + if (d.block_count == 0 || (uint64_t)d.block_count * 4 > sizeof(_src_leaves)) return false; + // read the manifest-minus-leaves + leaves[] from the backing — an external folder MotaSource, or (for a + // completed download we re-seed) our own fetch store. Container offsets are absolute, so a store read + // at the same offsets works identically. + bool ok = e.is_fetch ? _fetch->read(8, _src_manifest, mfl) + : e.src->read(e.src_idx, 8, _src_manifest, mfl); + if (!ok || !mota_parse_manifest(_src_manifest, mfl, _srcv.m)) return false; + if (memcmp(_srcv.m.merkle_root, d.mid, 4) != 0) return false; // descriptor/bytes disagree + if (_srcv.m.block_count != d.block_count) return false; + ok = e.is_fetch ? _fetch->read(d.leaves_off, _src_leaves, d.block_count * 4) + : e.src->read(e.src_idx, d.leaves_off, _src_leaves, d.block_count * 4); + if (!ok) return false; + _srcv.m.manifest_start = _src_manifest; + _srcv.m.leaves = _src_leaves; + _srcv.m.payload = nullptr; + _srcv.mfl = mfl; + _srcv_rdctx.src = e.is_fetch ? nullptr : e.src; _srcv_rdctx.idx = e.src_idx; + _srcv_rdctx.payload_off = d.payload_off; _srcv_rdctx.store = e.is_fetch ? _fetch : nullptr; + _srcv.read = srcReadTramp; _srcv.read_ctx = &_srcv_rdctx; + _srcv.scratch = _scratch; _srcv.scratch_sz = sizeof(_scratch); + memcpy(_srcv_mid, d.mid, 4); + _srcv.valid = true; + return true; +} + +// ServeReadFn trampoline: payload-relative offset -> absolute read of the backing (external source or fetch store). +bool OtaManager::srcReadTramp(void* c, uint32_t off, uint8_t* buf, uint32_t len) { + SrcReadCtx* x = (SrcReadCtx*)c; + if (x->store) return x->store->read(x->payload_off + off, buf, len); + return x->src->read(x->idx, x->payload_off + off, buf, len); +} + +// After a download COMPLETEs, advertise + serve the staged container so this node re-seeds it to peers +// (epidemic spread: the origin seeds a few, they seed the next ring -> load on the origin is O(log N), not +// O(N)). The completed container has ALL blocks + leaves, so it serves DATA *and* proofs correctly. Re-uses +// the on-demand source view; serving is reactive + lowest-priority, so it never competes with real traffic. +void OtaManager::serveFetched() { + if (!_fetch || _fstate != COMPLETE || _fbc == 0 || _floff < 8) return; + uint16_t mfl = (uint16_t)(_floff - 8); + if (mfl == 0 || mfl > sizeof(_src_manifest)) return; + if ((uint64_t)_fbc * 4 > sizeof(_src_leaves)) return; // proof-gen scratch caps re-seed at <=1024 blocks + uint8_t head[OTA_SRC_MANIFEST_MAX]; + if (!_fetch->read(8, head, mfl)) return; + MotaManifest m; + if (!mota_parse_manifest(head, mfl, m)) return; + MotaDesc& d = _fetch_desc; + memcpy(d.mid, _fid, 4); + d.target_id = m.target_id; d.fw_version = m.fw_version; d.codec_id = m.codec_id; d.flags = m.flags; + d.total_size = _ftotal; d.leaves_off = _floff; d.block_count = _fbc; + d.payload_off = _fpoff; d.payload_size = _fpsize; + _fetch_served = true; + refresh_sources(); // add the fetch entry to the catalog; the set-digest change makes the next beacon advertise it +} + +void OtaManager::unserveFetched() { + if (!_fetch_served) return; + _fetch_served = false; + _srcv.valid = false; // the loaded source view may be the fetch we're dropping + refresh_sources(); +} + +// sha2-256:4 over the SORTED set of mids we serve — peers use it to tell if our offering changed. Sorting +// makes it canonical across nodes regardless of insert order; for a single mota it is mh4(mid) (unchanged). +void OtaManager::setDigest(uint8_t out[4]) const { + if (_n_serve == 0) { memset(out, 0, 4); return; } + uint8_t order[OTA_MAX_SERVE]; + for (uint8_t i = 0; i < _n_serve; i++) order[i] = i; + for (uint8_t i = 1; i < _n_serve; i++) { // insertion sort by mid (n <= 12) + uint8_t v = order[i]; int j = (int)i - 1; + while (j >= 0 && memcmp(_serve[order[j]].mid, _serve[v].mid, 4) > 0) { order[j+1] = order[j]; j--; } + order[j+1] = v; + } + uint8_t cat[OTA_MAX_SERVE * 4]; + for (uint8_t i = 0; i < _n_serve; i++) memcpy(cat + (uint32_t)i * 4, _serve[order[i]].mid, 4); + mh4(out, cat, (size_t)_n_serve * 4); +} + +void OtaManager::announce() { // tiny per-node beacon (constant size, independent of how many mOTAs) + AdvMsg a; + memcpy(a.seeder_id, _seeder_id, 4); + a.n_motas = _n_serve; + setDigest(a.set_digest); + uint8_t b[16]; + emit(b, encode_adv(b, sizeof(b), a), true); +} + +// OTA_QUERY: two roles. (1) OVERHEAR-SUPPRESSION — any node that has a pending query for the same +// {source,digest} cancels it (someone else already asked; the broadcast HAVE is coming). (2) If the query +// is addressed to US, reply with our catalog (broadcast, tagged with our digest so every overhearer caches +// it). All served mOTAs matching filter_target are returned, fragmented if they exceed one packet. +void OtaManager::handleQuery(const uint8_t* m, uint16_t n) { + QueryMsg q; + if (!decode_query(m, n, q)) return; + if (_pq_active && memcmp(_pq_seeder, q.seeder_id, 4) == 0 && memcmp(_pq_digest, q.set_digest, 4) == 0) + _pq_active = false; // (1) suppress our own pending query + if (_n_serve == 0 || memcmp(q.seeder_id, _seeder_id, 4) != 0) return; // (2) only WE answer queries to us + uint8_t dg[4]; setDigest(dg); + uint8_t rowbuf[OTA_MAX_SERVE * OTA_HAVE_ROW_BYTES]; + uint8_t nm = 0; + for (uint8_t i = 0; i < _n_serve; i++) { + const ServeEntry& e = _serve[i]; + if (q.filter_target != 0 && q.filter_target != e.target_id) continue; + uint8_t* row = rowbuf + (uint32_t)nm * OTA_HAVE_ROW_BYTES; + memcpy(row, e.mid, 4); + wr_u32le(row + 4, e.target_id); wr_u32le(row + 8, e.fw_version); + row[12] = e.codec_id; row[13] = e.flags; + uint32_t hc = e.have_count > 0xFFFFu ? 0xFFFFu : e.have_count; // blocks we hold (awareness for fetchers) + row[14] = (uint8_t)(hc & 0xFF); row[15] = (uint8_t)(hc >> 8); + nm++; + } + const uint8_t per = (uint8_t)((MAX_PACKET_PAYLOAD - 12) / OTA_HAVE_ROW_BYTES); // rows per HAVE fragment + uint8_t ftotal = (uint8_t)((nm + per - 1) / per); if (ftotal == 0) ftotal = 1; + for (uint8_t fi = 0; fi < ftotal; fi++) { + uint8_t bse = (uint8_t)(fi * per); + uint8_t cnt = (uint8_t)((nm - bse > per) ? per : (nm - bse)); + HaveMsg hv; memcpy(hv.seeder_id, _seeder_id, 4); memcpy(hv.set_digest, dg, 4); + hv.frag_idx = fi; hv.frag_total = ftotal; hv.n_rows = cnt; + hv.rows = cnt ? (rowbuf + (uint32_t)bse * OTA_HAVE_ROW_BYTES) : nullptr; + uint8_t b[MAX_PACKET_PAYLOAD]; + emit(b, encode_have(b, sizeof(b), hv), true); // broadcast: all neighbours cache it + } +} + +void OtaManager::handleGetManifest(const uint8_t* m, uint16_t n) { + GetManifestMsg gm; + if (!decode_get_manifest(m, n, gm)) return; + ServeView* v = resolve(gm.manifest_id); + if (!v) return; + // A signed v2 manifest (with hw_id[32]) exceeds one LoRa packet, so send it as fragments. Each carries + // up to OTA_MF_FRAG manifest bytes; the client reassembles by frag_idx. (Re-sent in full on a retry.) + uint32_t mfl = v->mfl; + const uint8_t* src = v->m.manifest_start; + uint8_t ftotal = (uint8_t)((mfl + OTA_MF_FRAG - 1) / OTA_MF_FRAG); if (ftotal == 0) ftotal = 1; + for (uint8_t fi = 0; fi < ftotal; fi++) { + uint32_t off = (uint32_t)fi * OTA_MF_FRAG; + uint32_t fl = mfl - off; if (fl > OTA_MF_FRAG) fl = OTA_MF_FRAG; + ManifestMsg mm; + memcpy(mm.manifest_id, v->m.merkle_root, 4); + mm.frag_idx = fi; mm.frag_total = ftotal; + mm.bytes = src + off; mm.len = (uint16_t)fl; + uint8_t b[MAX_PACKET_PAYLOAD]; + emit(b, encode_manifest(b, sizeof(b), mm), false); + } +} + +// Emit one block's data as self-describing DATA fragments (frag_off); the proof is fetched separately. +void OtaManager::emitBlockData(const uint8_t* mid, uint32_t idx, const uint8_t* data, uint32_t blen) { + for (uint32_t fo = 0; fo < blen; fo += OTA_FRAG_DATA) { + uint32_t fl = (fo + OTA_FRAG_DATA <= blen) ? OTA_FRAG_DATA : (blen - fo); + DataMsg dm; + memcpy(dm.manifest_id, mid, 4); + dm.block_idx = (uint16_t)idx; dm.frag_off = (uint16_t)fo; + dm.data = data + fo; dm.data_len = (uint16_t)fl; + uint8_t b[MAX_PACKET_PAYLOAD]; + emit(b, encode_data(b, sizeof(b), dm), false); + } +} + +// True if we recently overheard ANOTHER holder broadcast this block's DATA — so we should not re-serve it +// (avoids N sources duplicate-broadcasting one block; keeps OTA airtime minimal). See noteOverheardData(). +bool OtaManager::recentlyServed(uint32_t blk) const { + for (uint8_t i = 0; i < 8; i++) + if (_recent_blk[i] == blk && (uint32_t)(_now_ms - _recent_at[i]) < OTA_SERVE_SUPPRESS_MS) return true; + return false; +} + +void OtaManager::noteOverheardData(const uint8_t* m, uint16_t n) { + DataMsg dm; + if (!decode_data(m, n, dm)) return; + _recent_blk[_recent_i] = dm.block_idx; _recent_at[_recent_i] = _now_ms; + _recent_i = (uint8_t)((_recent_i + 1) & 7); +} + +void OtaManager::handleReq(const uint8_t* m, uint16_t n) { + ReqMsg rq; + if (!decode_req(m, n, rq)) return; + ServeView* v = resolve(rq.manifest_id); + if (v) { // serve a fully-held mota (own fw / folder / completed fetch) + uint32_t bs = v->m.block_size(); + for (uint32_t k = 0; k < rq.count; k++) { + uint32_t idx = rq.start_block + k; + if (idx >= v->m.block_count) break; + if (recentlyServed(idx)) continue; // another holder just broadcast it — don't duplicate + uint32_t off = idx * bs; + uint32_t blen = (off + bs <= v->m.payload_size) ? bs : (v->m.payload_size - off); + uint8_t blk[OTA_MAX_BLOCK]; + const uint8_t* data; + if (v->read) { if (!v->read(v->read_ctx, off, blk, blen)) break; data = blk; } + else { data = v->m.payload + off; } + emitBlockData(v->m.merkle_root, idx, data, blen); + } + return; + } + // Partial re-serve (swarm DURING the transfer): we're fetching this mid and already hold some of these + // blocks — serve their DATA (not proofs; we may lack sibling leaves) from our staging store, so peers can + // source from us, not only the origin. Reactive + lowest-priority, so real traffic is never impacted. + if (_fetch && _fstate == FETCHING && memcmp(rq.manifest_id, _fid, 4) == 0) { + for (uint32_t k = 0; k < rq.count; k++) { + uint32_t idx = rq.start_block + k; + if (idx >= _fbc) break; + if (!blockPresent(idx) || recentlyServed(idx)) continue; + uint8_t blk[OTA_MAX_BLOCK]; + uint32_t blen = blockLen(idx); + if (!_fetch->read(_fpoff + idx * _fbs, blk, blen)) continue; + emitBlockData(_fid, idx, blk, blen); + } + } +} + +void OtaManager::handleReqProof(const uint8_t* m, uint16_t n) { + ReqProofMsg rp; + if (!decode_req_proof(m, n, rp)) return; + ServeView* v = resolve(rp.manifest_id); + if (!v) return; + if (rp.block_idx >= v->m.block_count) return; + if ((uint64_t)v->m.block_count * 4 > v->scratch_sz) return; // proof-gen needs block_count*4 scratch + uint8_t proof[32 * 4]; + uint8_t np = merkle_gen_proof(v->m.leaves, v->m.block_count, rp.block_idx, v->scratch, proof); + ProofMsg pm; + memcpy(pm.manifest_id, v->m.merkle_root, 4); + pm.block_idx = rp.block_idx; pm.n_proof = np; pm.proof = proof; + uint8_t b[MAX_PACKET_PAYLOAD]; + emit(b, encode_proof(b, sizeof(b), pm), false); +} + +// ---------------- fetch ---------------- + +// A tiny per-node BEACON: record the source; ask it for its catalog (OTA_QUERY) only when we're +// interested AND its set-digest is one we haven't catalogued yet (so a stable mesh is query-free). +void OtaManager::handleAdv(const uint8_t* m, uint16_t n) { + AdvMsg a; + if (!decode_adv(m, n, a)) return; + bool have_sid = (_seeder_id[0] | _seeder_id[1] | _seeder_id[2] | _seeder_id[3]) != 0; + if (have_sid && memcmp(a.seeder_id, _seeder_id, 4) == 0) return; // our own beacon, re-flooded + if (a.n_motas == 0) return; // source offers nothing + + int slot = -1, lru = 0; // find/insert the source (LRU evict) + for (int i = 0; i < _n_src; i++) { + if (memcmp(_sources[i].seeder, a.seeder_id, 4) == 0) { slot = i; break; } + if (_sources[i].last_ms < _sources[lru].last_ms) lru = i; + } + bool fresh = (slot < 0); + if (fresh) { slot = (_n_src < OTA_MAX_SOURCES) ? _n_src++ : lru; _sources[slot] = Source{}; } + Source& s = _sources[slot]; + bool changed = fresh || memcmp(s.digest, a.set_digest, 4) != 0; + memcpy(s.seeder, a.seeder_id, 4); memcpy(s.digest, a.set_digest, 4); + s.n_motas = a.n_motas; s.last_ms = _now_ms; + if (changed) s.have_catalog = false; + + // interested = auto-fetch enabled, or a manual pull/want is pending. (Browsing queries via queryAll().) + bool interested = (_autofetch != AUTOFETCH_OFF) || _have_desired_mid || _desired_target; + if (interested && !s.have_catalog) scheduleQuery(a.seeder_id, a.set_digest); // jittered + suppressible +} + +// Schedule a catalog query after a random jitter (id ⊕ digest, so neighbours pick different delays). The +// node with the shortest jitter sends; the rest overhear that QUERY (or the broadcast HAVE) and suppress. +void OtaManager::scheduleQuery(const uint8_t* seeder, const uint8_t* digest) { + if (_pq_active && memcmp(_pq_seeder, seeder, 4) == 0 && memcmp(_pq_digest, digest, 4) == 0) return; // already pending + memcpy(_pq_seeder, seeder, 4); memcpy(_pq_digest, digest, 4); + uint32_t j = (rd_u32le(seeder) ^ rd_u32le(digest) ^ rd_u32le(_seeder_id)) % OTA_QUERY_SPREAD_MS; + _pq_at = _now_ms + OTA_QUERY_MIN_MS + j; + _pq_active = true; +} + +void OtaManager::sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target) { + QueryMsg q; memcpy(q.seeder_id, seeder, 4); memcpy(q.set_digest, digest, 4); q.filter_target = filter_target; + uint8_t b[16]; + emit(b, encode_query(b, sizeof(b), q), true); // FLOODED so neighbours overhear it and suppress +} + +// User-initiated browse (`ota neighbors`): ask every known source now (no jitter — infrequent + explicit). +void OtaManager::queryAll() { for (uint8_t i = 0; i < _n_src; i++) sendQuery(_sources[i].seeder, _sources[i].digest, 0); } + +// A catalog reply: record each mOTA (deduped by mid; distinct-source count for the UI), and if a row +// matches our fetch interest (auto-fetch own-target, or a pending pull/want), begin fetching it. +void OtaManager::handleHave(const uint8_t* m, uint16_t n) { + HaveMsg hv; + if (!decode_have(m, n, hv)) return; + bool have_sid = (_seeder_id[0] | _seeder_id[1] | _seeder_id[2] | _seeder_id[3]) != 0; + if (have_sid && memcmp(hv.seeder_id, _seeder_id, 4) == 0) return; // our own catalog + // PASSIVE: any overheard HAVE marks its source catalogued + cancels a pending query for it (storm + // suppression) — every node caches the rows below, even one that never queried. + for (uint8_t i = 0; i < _n_src; i++) + if (memcmp(_sources[i].seeder, hv.seeder_id, 4) == 0 && memcmp(_sources[i].digest, hv.set_digest, 4) == 0) + _sources[i].have_catalog = true; + if (_pq_active && memcmp(_pq_seeder, hv.seeder_id, 4) == 0 && memcmp(_pq_digest, hv.set_digest, 4) == 0) + _pq_active = false; + for (uint8_t r = 0; r < hv.n_rows && hv.rows; r++) { + const uint8_t* row = hv.rows + (uint32_t)r * OTA_HAVE_ROW_BYTES; + const uint8_t* mid = row; + uint32_t target = rd_u32le(row + 4), fwver = rd_u32le(row + 8); + uint8_t codec = row[12], flags = row[13]; + uint32_t have_count = rd_u16le(row + 14); // this source's progress + int slot = -1, lru = 0; // upsert into the catalog (dedup by mid) + for (int i = 0; i < _n_cat; i++) { + if (memcmp(_catalog[i].mid, mid, 4) == 0) { slot = i; break; } + if (_catalog[i].last_ms < _catalog[lru].last_ms) lru = i; + } + if (slot < 0) { + slot = (_n_cat < OTA_MAX_CATALOG) ? _n_cat++ : lru; + _catalog[slot] = CatRow{}; + memcpy(_catalog[slot].mid, mid, 4); + memcpy(_catalog[slot].seeders[0], hv.seeder_id, 4); + _catalog[slot].n_seeders = 1; + } else { + CatRow& cc = _catalog[slot]; // count DISTINCT sources (no double-count) + bool known = false; + for (uint8_t k = 0; k < cc.n_seeders; k++) + if (memcmp(cc.seeders[k], hv.seeder_id, 4) == 0) { known = true; break; } + if (!known && cc.n_seeders < OTA_CAT_SEEDERS) memcpy(cc.seeders[cc.n_seeders++], hv.seeder_id, 4); + } + CatRow& c = _catalog[slot]; + c.target_id = target; c.fw_version = fwver; c.codec = codec; c.flags = flags; c.last_ms = _now_ms; + if (have_count > c.have_max) c.have_max = have_count; // best-known progress among sources + if (wantRow(mid, target, codec, flags)) startFetch(mid, target); + } +} + +bool OtaManager::wantRow(const uint8_t* mid, uint32_t target, uint8_t codec, uint8_t flags) const { + if (!_fetch || _fstate == FETCHING || _fstate == WANT_MANIFEST) return false; // busy with a session + if (_fstate == COMPLETE && memcmp(mid, _fid, 4) == 0) return false; // already have it + if (!codecOk(codec)) return false; // can't apply this codec + if (_have_desired_mid) // manual pull of a specific mid + return memcmp(mid, _desired_mid, 4) == 0 && (_desired_target == 0 || target == _desired_target); + if (_desired_target) return target == _desired_target; // cross-target want (role switch) + if (_autofetch == AUTOFETCH_OFF) return false; // discover only + if (target != _target) return false; // auto-fetch = our own target + if (_autofetch == AUTOFETCH_SIGNED && !(flags & MFLAG_SIGNED)) return false; // signed-only policy + return true; +} + +// Forget the block currently being reassembled / awaited (back to NO_BLOCK). Safe to call between blocks: +// the next DATA fragment re-derives the slice mask for whatever block it belongs to. +void OtaManager::clearReassembly() { + _reasm_block = NO_BLOCK; _reasm_mask = 0; _reasm_need = 0; _awaiting_proof = false; +} + +// De-sync the first REQ across the swarm: hold it a random fraction of OTA_REQ_SPREAD_MS so N nodes that +// just discovered the same mid don't burst-request block 0 in lockstep (loop() fires it once the hold +// elapses). Assumes the per-node RNG is already seeded; also forgets any peer-REQ note from a prior session. +void OtaManager::armFirstReqHold() { + _req_hold_at = _now_ms + (rngNext() % OTA_REQ_SPREAD_MS); + _peer_req_block = NO_BLOCK; _peer_req_at = 0; +} + +// 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) { + (void)target; + if (!_fetch || _fstate == FETCHING || _fstate == WANT_MANIFEST) return; + if (resumeStaged(mid)) return; // resume a partial container left in flash + memcpy(_fid, mid, 4); + seedBlockRng(); // per-node block-pick/jitter sequence (distinct per node) + _fstate = WANT_MANIFEST; + _mf_total = 0; _mf_mask = 0; _mf_len = 0; _mf_retries = 0; // fresh manifest reassembly + GetManifestMsg gm; memcpy(gm.manifest_id, _fid, 4); + uint8_t b[16]; + emit(b, encode_get_manifest(b, sizeof(b), gm), false); +} + +void OtaManager::handleManifest(const uint8_t* m, uint16_t n) { + ManifestMsg mm; + if (!decode_manifest(m, n, mm) || !_fetch) return; + if (_fstate != WANT_MANIFEST || memcmp(mm.manifest_id, _fid, 4) != 0) return; + if (mm.frag_total == 0 || mm.frag_total > OTA_MF_MAXFRAG || mm.frag_idx >= mm.frag_total) return; + + // reassemble the (possibly multi-fragment) manifest into _mf_buf; place fragment frag_idx at its offset + uint32_t foff = (uint32_t)mm.frag_idx * OTA_MF_FRAG; + if (foff + mm.len > sizeof(_mf_buf)) return; + if (mm.frag_total != _mf_total) { _mf_total = mm.frag_total; _mf_mask = 0; _mf_len = 0; } // (re)start + memcpy(_mf_buf + foff, mm.bytes, mm.len); + _mf_mask |= (uint16_t)(1u << mm.frag_idx); + if (mm.frag_idx == mm.frag_total - 1) _mf_len = foff + mm.len; // last fragment fixes the length + uint16_t full = (mm.frag_total >= 16) ? 0xFFFF : (uint16_t)((1u << mm.frag_total) - 1); + if (_mf_mask != full || _mf_len == 0) return; // wait until every fragment is in + + const uint8_t* mf = _mf_buf; // fully reassembled manifest-minus-leaves + uint32_t mfl = _mf_len; + if (mfl != MOTA_MFL) { _fstate = FAILED; return; } // manifest-minus-leaves is a fixed 197 bytes + if (!codecOk(mf[56])) { _fstate = IDLE; return; } // codec we can't apply (lying/stale ADV) — abort + uint32_t payload_size = rd_u32le(mf + 15); + uint8_t bsl = mf[19]; + uint32_t bs = 1u << bsl; + // a block must fit our reassembly buffer (and be non-empty) — reject an oversized block_size up front + if (bs == 0 || bs > OTA_MAX_BLOCK || payload_size == 0) { _fstate = FAILED; return; } + uint32_t bc = (payload_size + bs - 1) / bs; + if (bc > 0xFFFFu) { _fstate = FAILED; return; } // block_idx is uint16 on the wire — can't address more + memcpy(_froot, mf + 20, 4); + + uint32_t leaves_off = 8 + mfl; + uint32_t payload_off = leaves_off + bc * 4; + uint32_t total = payload_off + payload_size + 5; + + // Hand the store the parsed layout BEFORE begin(), so a partition-backed store (ESP32) can choose + // placement and refuse an unfittable fetch up front: a FULL payload streams to the inactive slot, + // a delta's whole container is staged together. (image_size at mf+11, is_full from flags at mf+1.) + bool is_full = (mf[1] & MFLAG_FULL) != 0; + if (!_fetch->plan_layout(is_full, rd_u32le(mf + 11), payload_off, payload_size)) { _fstate = FAILED; return; } + unserveFetched(); // the store is about to be overwritten by this new fetch — stop re-seeding the old one + if (!_fetch->begin(total)) { _fstate = FAILED; return; } + // declare the metadata extent so a flash store can pin it (leaves are written all transfer long) + if (!_fetch->set_meta_size(payload_off)) { _fstate = FAILED; return; } + uint8_t hdr[8]; + memcpy(hdr, MOTA_MAGIC, 4); + wr_u32le(hdr + 4, total); + if (!_fetch->write(0, hdr, 8) || + !_fetch->write(8, mf, mfl) || + !_fetch->write(total - 5, MOTA_TRAILER, 5)) { _fstate = FAILED; return; } + + _fflags = mf[1]; // manifest flags (FULL/SIGNED) of the fetch in progress (auto-install gate) + _fpoff = payload_off; _floff = leaves_off; _fpsize = payload_size; _fbc = bc; _fbs = bs; + _ftotal = total; _have = 0; _fstate = FETCHING; + clearReassembly(); // fresh transfer: drop any prior per-block state + _loop_last_have = 0; _loop_last_mask = 0; + if (_rng == 0) seedBlockRng(); + armFirstReqHold(); + OTA_DBG("OTA: FETCHING bc=%u bs=%u total=%u\n", (unsigned)bc, (unsigned)bs, (unsigned)total); +} + +bool OtaManager::resumeStaged(const uint8_t* want_mid) { + if (!_fetch || _fstate == FETCHING || _fstate == WANT_MANIFEST) return false; + if (!_fetch->reopen()) return false; // nothing persisted in the store + uint32_t total = _fetch->staged_size(); + uint8_t hdr[8]; + if (total < 13 || !_fetch->read(0, hdr, 8) || memcmp(hdr, MOTA_MAGIC, 4) != 0) return false; + // read + parse the stored manifest (everything before leaves[]) to recompute the geometry + uint8_t mbuf[256]; + uint32_t mread = total - 8; if (mread > sizeof(mbuf)) mread = sizeof(mbuf); + MotaManifest m; + if (!_fetch->read(8, mbuf, mread) || !mota_parse_manifest(mbuf, mread, m)) return false; + if (want_mid && memcmp(m.merkle_root, want_mid, 4) != 0) return false; // a different fw is staged + if (!codecOk(m.codec_id)) return false; + uint32_t mfl = (uint32_t)(m.approval - m.manifest_start) + 4; // manifest-minus-leaves length + uint32_t bs = m.block_size(); + if (bs == 0 || bs > OTA_MAX_BLOCK) return false; + uint32_t bc = m.block_count; + uint32_t leaves_off = 8 + mfl; + uint32_t payload_off = leaves_off + bc * 4; + if ((uint64_t)payload_off + m.payload_size + 5 != total) return false; // geometry must match the header + + memcpy(_fid, m.merkle_root, 4); + memcpy(_froot, m.merkle_root, 4); + _fflags = m.flags; + _fpoff = payload_off; _floff = leaves_off; _fpsize = m.payload_size; _fbc = bc; _fbs = bs; + _ftotal = total; + _have = 0; + for (uint32_t i = 0; i < bc; i++) if (blockPresent(i)) _have++; // count blocks whose leaf survived + clearReassembly(); + _loop_last_have = 0; _loop_last_mask = 0; + OTA_DBG("OTA: RESUME have=%u/%u total=%u\n", (unsigned)_have, (unsigned)bc, (unsigned)total); + + if (_have >= bc) { // already complete -> verify root + finalize + if (bc * 4 <= sizeof(_scratch) && _fetch->read(_floff, _scratch, bc * 4)) { + uint8_t root[4]; merkle_root(root, _scratch, bc); + _fstate = (memcmp(root, _froot, 4) == 0) ? COMPLETE : FAILED; + } else { + _fstate = COMPLETE; + } + if (_fstate == COMPLETE) { _fetch->finalize(); serveFetched(); } + return true; + } + _fstate = FETCHING; // resume fetching the holes + // De-sync the first REQ exactly like a fresh fetch, so a coordinated reboot (whole-site power-cycle) + // doesn't make every resuming node REQ in lockstep. + seedBlockRng(); + armFirstReqHold(); + _loop_last_have = _have; _loop_last_mask = _reasm_mask; // "no progress yet" -> loop will request after the hold + return true; +} + +uint32_t OtaManager::blockLen(uint32_t i) const { + uint32_t off = i * _fbs; + return (off + _fbs <= _fpsize) ? _fbs : (_fpsize - off); +} + +bool OtaManager::blockPresent(uint32_t i) const { + uint8_t leaf[4]; + if (!_fetch->read(_floff + i * 4, leaf, 4)) return false; + return !(leaf[0]==0xFF && leaf[1]==0xFF && leaf[2]==0xFF && leaf[3]==0xFF); +} + +void 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 + 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.block_idx != _reasm_block) { // (re)start reassembly for this block + _reasm_block = dm.block_idx; _reasm_mask = 0; _awaiting_proof = false; + uint32_t nf = (blen + OTA_FRAG_DATA - 1) / OTA_FRAG_DATA; + _reasm_need = (nf >= 16) ? 0xFFFF : (uint16_t)((1u << nf) - 1); + } + uint32_t kf = dm.frag_off / OTA_FRAG_DATA; + if (kf >= 16) return; + memcpy(_reasm_buf + dm.frag_off, dm.data, dm.data_len); + _reasm_mask |= (uint16_t)(1u << kf); + if (_reasm_mask != _reasm_need || _awaiting_proof) return; // wait for all slices (or proof already asked) + // block fully reassembled -> request its proof (data + proof are fetched separately) + _awaiting_proof = true; + ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)_reasm_block; + uint8_t b[16]; emit(b, encode_req_proof(b, sizeof(b), rp), false); +} + +void 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 (!_awaiting_proof || pm.block_idx != _reasm_block) return; // not the block we're verifying + uint32_t blen = blockLen(_reasm_block); + if (!merkle_verify(_reasm_buf, blen, _reasm_block, pm.proof, pm.n_proof, _froot, _fbc)) { + clearReassembly(); // bad -> drop, re-fetch the block + return; + } + // verified -> commit the payload block, then its leaf (the present marker) + if (!_fetch->write(_fpoff + (uint32_t)_reasm_block * _fbs, _reasm_buf, blen)) return; + uint8_t leaf[4]; merkle_leaf(leaf, _reasm_buf, blen); + if (!_fetch->write(_floff + (uint32_t)_reasm_block * 4, leaf, 4)) return; + _have++; + OTA_DBG("OTA: block %u OK have=%u/%u\n", (unsigned)_reasm_block, (unsigned)_have, (unsigned)_fbc); + clearReassembly(); + // periodically persist progress (meta/leaf page + open payload) so a reboot can resume (no-op for RAM); + // cadence is runtime-tunable via `ota config checkpoint ` (0 = never) + if (_checkpoint_blocks && _have % _checkpoint_blocks == 0) _fetch->checkpoint(); + if (_have < _fbc) { requestMissing(); return; } // next block + // all blocks present -> final root cross-check + finalize + if (_fbc * 4 <= sizeof(_scratch) && _fetch->read(_floff, _scratch, _fbc * 4)) { + uint8_t root[4]; merkle_root(root, _scratch, _fbc); + _fstate = (memcmp(root, _froot, 4) == 0) ? COMPLETE : FAILED; + } else { + _fstate = COMPLETE; // per-block proofs already guaranteed integrity vs the root + } + if (_fstate == COMPLETE) { _fetch->finalize(); serveFetched(); } // commit + re-seed (epidemic spread) + OTA_DBG("OTA: transfer %s\n", _fstate == COMPLETE ? "COMPLETE" : "FAILED(root)"); +} + +void OtaManager::requestMissing() { + if (_fstate != FETCHING) return; + // Per-block serial flow (split data/proof). If the current block's data is fully reassembled and we + // are waiting on its proof, (re)send the proof request rather than re-fetching the data — this also + // recovers from a lost PROOF reply. + if (_awaiting_proof && _reasm_block != NO_BLOCK) { + ReqProofMsg rp; memcpy(rp.manifest_id, _fid, 4); rp.block_idx = (uint16_t)_reasm_block; + uint8_t b[16]; emit(b, encode_req_proof(b, sizeof(b), rp), false); + OTA_DBG("OTA: REQ_PROOF block=%u (have=%u/%u)\n", + (unsigned)_reasm_block, (unsigned)_have, (unsigned)_fbc); + return; + } + // Otherwise request the DATA fragments of the next missing block. One block at a time keeps the + // server's TX queue tiny so OTA never floods the mesh (docs/ota_protocol.md §8); a block's fragments + // are self-describing (frag_off) so they may be served by ANY peer, BitTorrent-style. + uint32_t start = pickMissingBlock(); + if (start >= _fbc) return; + _req_start = start; _req_count = 1; + ReqMsg rq; memcpy(rq.manifest_id, _fid, 4); + rq.start_block = (uint16_t)start; rq.count = 1; + uint8_t b[16]; + OTA_DBG("OTA: REQ block=%u (have=%u/%u mask=%04x)\n", + (unsigned)start, (unsigned)_have, (unsigned)_fbc, (unsigned)_reasm_mask); + emit(b, encode_req(b, sizeof(b), rq), false); +} + +// Choose which block to request next. Swarm-aware so N fetchers of the same mid spread their load instead +// of marching in lockstep on the same block: +// - finish an in-flight partially-reassembled block first (don't waste received fragments); +// - otherwise pick a RANDOM missing block (de-correlates fetchers -> they collectively pull different +// blocks, and every broadcast DATA fills everyone's hole); +// - skip a block a peer just REQ'd (its DATA is already coming over the air) unless it's all that's left. +// Returns _fbc if nothing to request. +uint32_t OtaManager::pickMissingBlock() { + if (_fbc == 0) return _fbc; + // (1) keep finishing a block we've already started reassembling (recover its lost fragments) + if (_reasm_block < _fbc && !blockPresent(_reasm_block) && _reasm_mask != 0) return _reasm_block; + // (2) count missing blocks + uint32_t miss = 0; + for (uint32_t i = 0; i < _fbc; i++) if (!blockPresent(i)) miss++; + if (miss == 0) return _fbc; + bool suppress = (_peer_req_block < _fbc) && ((uint32_t)(_now_ms - _peer_req_at) < OTA_REQ_SUPPRESS_MS); + // (3) pick the k-th missing block (k from the per-node RNG), optionally skipping the peer-REQ'd one + uint32_t k = rngNext() % miss; + uint32_t seen = 0, chosen = _fbc, firstAny = _fbc; + for (uint32_t i = 0; i < _fbc; i++) { + if (blockPresent(i)) continue; + if (firstAny == _fbc) firstAny = i; + if (seen == k) { chosen = i; } + seen++; + } + if (suppress && chosen == _peer_req_block) { // pick a different missing block than the one in flight elsewhere + for (uint32_t i = 0; i < _fbc; i++) { + uint32_t j = (chosen + 1 + i) % _fbc; + if (!blockPresent(j) && j != _peer_req_block) { chosen = j; break; } + } + // if the suppressed block is the ONLY one left, chosen stays == it (we still need it eventually) + } + return (chosen < _fbc) ? chosen : firstAny; +} + +// Observe a peer's OTA_REQ for the mid we're fetching: its block's DATA is broadcast, so it will fill our +// hole too — note it so pickMissingBlock() spends our next REQ on a DIFFERENT block (swarm de-dup). +void OtaManager::noteOverheardReq(const uint8_t* m, uint16_t n) { + if (_fstate != FETCHING) return; + ReqMsg rq; + if (!decode_req(m, n, rq) || memcmp(rq.manifest_id, _fid, 4) != 0) return; + _peer_req_block = rq.start_block; + _peer_req_at = _now_ms; +} + +void OtaManager::loop() { + // fire a scheduled catalog query once its jitter has elapsed (unless overhearing already suppressed it) + if (_pq_active && (int32_t)(_now_ms - _pq_at) >= 0) { + _pq_active = false; + sendQuery(_pq_seeder, _pq_digest, 0); // unfiltered: one broadcast HAVE serves everyone + } + if (_fstate == WANT_MANIFEST) { + // the MANIFEST reply may have been lost on a marginal link — retry GET_MANIFEST, but give up after a + // cap so an unreachable mid doesn't pin the single fetch slot (or emit) forever. + if (++_mf_retries > OTA_MANIFEST_MAX_RETRY) { _fstate = FAILED; return; } + GetManifestMsg gm; memcpy(gm.manifest_id, _fid, 4); + uint8_t b[16]; + emit(b, encode_get_manifest(b, sizeof(b), gm), false); + return; + } + if (_fstate != FETCHING) return; + if ((int32_t)(_now_ms - _req_hold_at) < 0) return; // swarm: initial random hold (de-sync N fetchers) + // retry only when a whole tick passed with NO progress — neither a committed block nor a new fragment + // of the in-flight block. This avoids re-request spam while a block's fragments are still streaming in. + if (_have == _loop_last_have && _reasm_mask == _loop_last_mask) requestMissing(); + _loop_last_have = _have; + _loop_last_mask = _reasm_mask; +} + +// ---------------- dispatch ---------------- + +void 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_REQ: noteOverheardReq(msg, len); handleReq(msg, len); break; + case OTA_DATA: handleData(msg, len); noteOverheardData(msg, len); break; + case OTA_REQ_PROOF: handleReqProof(msg, len); break; + case OTA_PROOF: handleProof(msg, len); break; + default: break; + } +} + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h new file mode 100644 index 00000000..3428b453 --- /dev/null +++ b/src/helpers/ota/OtaManager.h @@ -0,0 +1,358 @@ +#pragma once + +#include +#include +#include "OtaFormat.h" +#include "OtaByteIO.h" +#include "OtaStore.h" +#include "MotaContainer.h" +#include "OtaSource.h" + +// Transport-agnostic OTA session engine (docs/ota_protocol.md §5/§8). It SERVES a complete `.mota` +// (answering GET_MANIFEST / REQ) and/or FETCHES one into an OtaStore (verifying every block against +// the signed merkle root via proofs). It is portable (no Arduino / radio / Ed25519) so it can be +// driven by a host simulation; a thin Mesh adapter wires it to PAYLOAD_TYPE_OTA on device. +// +// Transfer is per-block and 2-phase: a 1 KB logical block is fetched as self-describing DATA fragments +// (frag_off, so any peer can serve any fragment — BitTorrent-style), reassembled, then its merkle PROOF +// is requested separately and verified against the signed root before the block is committed. + +namespace mesh { +namespace ota { + +// Emit an OTA message (one packet payload). `flood`=true for announce/query, false for direct replies. +typedef void (*OtaSend)(void* ctx, const uint8_t* msg, uint16_t len, bool flood); + +// Read `len` payload bytes at offset `off` from the serve source (flash-backed self-serve); false on +// error. nullptr means the payload is a contiguous RAM buffer (the staged `.mota`). +typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len); + +#ifndef OTA_PROOFGEN_SCRATCH +#define OTA_PROOFGEN_SCRATCH 4096 // server proof-gen working buffer (supports up to 1024 blocks) +#endif + +#ifndef OTA_MAX_BLOCK +#define OTA_MAX_BLOCK 1024 // largest logical block (merkle leaf unit) = reassembly buffer size +#endif +#ifndef OTA_CHECKPOINT_BLOCKS +#define OTA_CHECKPOINT_BLOCKS 4 // persist progress (store.checkpoint) every N committed blocks (resume) +#endif +#ifndef OTA_MF_FRAG +#define OTA_MF_FRAG 176 // manifest bytes per OTA_MANIFEST fragment (<= MAX_PACKET_PAYLOAD - header) +#endif +#ifndef OTA_MF_MAXFRAG +#define OTA_MF_MAXFRAG 4 // max manifest fragments (the fixed 197 B manifest is always 2) +#endif +#ifndef OTA_MANIFEST_MAX_RETRY +#define OTA_MANIFEST_MAX_RETRY 20 // give up (FAILED) after this many GET_MANIFEST retries — frees the slot +#endif +#ifndef OTA_MAX_SOURCES +#define OTA_MAX_SOURCES 12 // heard OTA sources (beacon senders) tracked (LRU); ~12 B each +#endif +#ifndef OTA_MAX_SERVE +#define OTA_MAX_SERVE 12 // mOTAs THIS node offers (own fw + external folder); == one HAVE fragment +#endif +#ifndef OTA_MAX_SOURCE_OBJ +#define OTA_MAX_SOURCE_OBJ 4 // external MotaSource objects (folders/transports) attached at once +#endif +#ifndef OTA_SRC_MANIFEST_MAX +#define OTA_SRC_MANIFEST_MAX 256 // manifest-minus-leaves buffer for the loaded source mota (head+sig+approval) +#endif +#ifndef OTA_MAX_CATALOG +#define OTA_MAX_CATALOG 12 // distinct mOTAs catalogued from OTA_HAVE replies (LRU) +#endif +#ifndef OTA_QUERY_MIN_MS +#define OTA_QUERY_MIN_MS 300 // min delay before sending a catalog query (overhear-suppression window) +#endif +#ifndef OTA_QUERY_SPREAD_MS +#define OTA_QUERY_SPREAD_MS 4000 // random jitter span so 50 neighbours don't all query at once (storm) +#endif +#ifndef OTA_REQ_SPREAD_MS +#define OTA_REQ_SPREAD_MS 3000 // initial random hold before a fetch's first REQ (de-sync N fetchers) +#endif +#ifndef OTA_REQ_SUPPRESS_MS +#define OTA_REQ_SUPPRESS_MS 2500 // after overhearing a peer's REQ for a block, don't also request it — +#endif // its DATA is broadcast and will fill our hole too (swarm de-dup) +#ifndef OTA_SERVE_SUPPRESS_MS +#define OTA_SERVE_SUPPRESS_MS 1500 // don't re-serve a block whose DATA we just overheard another holder send +#endif // (so multiple sources of the same mota don't duplicate-broadcast it) +#ifndef OTA_FRAG_DATA +#define OTA_FRAG_DATA 160 // data bytes per DATA fragment (<= MAX_PACKET_PAYLOAD - 9-byte header) +#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 +// driver (Adafruit flash_nrf5x) always erases on flush, so there is no erase-free write; instead +// OtaStoreFlashNrf52 COALESCES to the 4 KB page (the erase unit) and writes each page once — RAM stays +// O(one page), never O(mota). It pins flash page 0 (header+manifest+merkle leaves, which update all +// transfer long) in RAM and streams the payload through one sliding page buffer, flushing page 0 and the +// last page at finalize() (radio idle). Flash is then touched ~once per 4 KB (≈1 per 4 blocks), not per +// packet; a small delta that fits page 0 does ZERO flash I/O until COMPLETE. (Pacing alone is not enough.) + +class OtaManager { +public: + enum FetchState : uint8_t { IDLE, WANT_MANIFEST, FETCHING, COMPLETE, FAILED }; + + // Sentinel for "no block" in the reassembly / peer-REQ / recently-served slots (a real block index is + // a small uint16, so 0xFFFFFFFF is never valid). + static const uint32_t NO_BLOCK = 0xFFFFFFFFu; + + // --- multi-mota serve --- A ServeView is everything a serve handler needs for ONE mota. Two can be + // resident: view0 = our own fw / a RAM `.mota` (always loaded), plus one on-demand source view that is + // (re)loaded from a MotaSource when a request targets a different external mota. Requests dispatch by + // manifest_id (carried in every fetch message) -> the matching ServeView via resolve(). + struct ServeView { + bool valid = false; + MotaManifest m; // parsed manifest (fields/pointers into the backing buffers) + uint16_t mfl = 0; // manifest-minus-leaves length (the OTA_MANIFEST payload) + ServeReadFn read = nullptr; // payload reader (nullptr => m.payload is contiguous in RAM) + void* read_ctx = nullptr; + uint8_t* scratch = nullptr; // proof-gen working buffer (>= block_count*4) + uint32_t scratch_sz = 0; + }; + // A lightweight catalog entry: what we advertise per mota + how to load its ServeView on demand. + struct ServeEntry { + uint8_t mid[4]; + uint32_t target_id, fw_version; + uint8_t codec_id, flags; + uint32_t have_count; // blocks we currently hold (== block_count when complete) + bool is_self; // true => entry is view0 (our own fw / RAM mota) + bool is_fetch; // true => load from our own fetch store (a completed download we re-seed) + MotaSource* src; // else: load from this external source ... + uint8_t src_idx; // ... at this index + MotaDesc desc; // cached region offsets (source / fetch entries) + }; + // Context for the source-payload reader trampoline (maps a payload-relative offset to a backing read: + // an external MotaSource, or — when `store` is set — our own fetch store, for re-seeding a completed mota). + struct SrcReadCtx { MotaSource* src; uint8_t idx; uint32_t payload_off; OtaStore* store; }; + + void begin(uint32_t my_target_id, OtaSend send, void* ctx); + + // --- serve --- Provide a complete, contiguous `.mota` to serve (caller keeps it alive). + bool serve(const uint8_t* mota, uint32_t len); + // Serve from a non-contiguous source (e.g. our own firmware in flash): a pre-assembled manifest + // (manifest-minus-leaves, `mfl` bytes), the pre-computed merkle `leaves` (kept alive by caller), and a + // `read` callback for payload blocks. Lets a node host its own image without holding it in RAM. + bool serve_self(const uint8_t* manifest, uint16_t mfl, const uint8_t* leaves, uint32_t block_count, + uint8_t* proof_scratch, uint32_t proof_scratch_sz, ServeReadFn read, void* ctx); + // Attach an external "folder" of `.mota` images (USB-serial daemon, BLE, WiFi URLs, NFS/samba, ...). + // The node then advertises + RELAYS them transparently alongside its own fw — peers just see more mOTAs. + // Re-enumerates the source into the serve registry. Returns false if no slots remain. (Trustless: the + // fetcher verifies merkle+signature, so the source is never trusted — see OtaSource.h.) + bool add_source(MotaSource* src); + // Re-read every attached source's catalog (call when the folder's contents change). Rebuilds entries + // [1..] from the sources; entry 0 (our own fw) is preserved. + void refresh_sources(); + // Drop all external sources (keep serving our own fw). + void clear_sources(); + uint8_t servedCount() const { return _n_serve; } // total mOTAs we offer (own fw + folder) + // Read-only view of one served entry (for `ota serve` listing): mid/target/fwver/codec/flags + is_self. + const ServeEntry* servedEntry(uint8_t i) const { return i < _n_serve ? &_serve[i] : nullptr; } + + // Broadcast the tiny per-node BEACON (OTA_ADV): seeder_id + count + set-digest of everything we serve. + // Constant size regardless of how many mOTAs — peers ask for the catalog via OTA_QUERY only on interest. + void announce(); + + // --- fetch --- Provide the staging store; fetching starts on a matching OTA_ADV. + void set_fetch_store(OtaStore* s) { _fetch = s; } + + // Resume a fetch from a container already persisted in the store (after a reboot). want_mid=nullptr + // accepts whatever is staged; otherwise only resumes if the staged manifest_id matches. Re-parses the + // stored manifest, recomputes geometry, counts present blocks, and continues FETCHING the holes (or goes + // straight to COMPLETE if all blocks are present). Returns true if it adopted a staged container. + bool resumeStaged(const uint8_t* want_mid); + + // Manual cross-target override (decision: deliberate role switch, e.g. companion -> repeater on the + // same hardware). Normally a node only auto-fetches its OWN target_id; `want(T)` makes it accept an + // ADV for target T instead (T=0 restores auto). The user takes responsibility for HW compatibility; + // a hw_id brick-safety check is the planned safety layer (see docs/ota_protocol.md / plan). + void want(uint32_t target_id) { _desired_target = target_id; reDiscover(); } + uint32_t wanted() const { return _desired_target; } + uint32_t target() const { return _target; } // this node's own OTA target_id (set in begin) + + // Pull a SPECIFIC advertised mOTA by manifest_id (e.g. `ota pull <#>` picks the one more peers have), + // not just any firmware for the target. mid=nullptr clears the filter (accept any mid for the target). + void want_mid(const uint8_t* mid) { + if (mid) { for (int i = 0; i < 4; i++) _desired_mid[i] = mid[i]; _have_desired_mid = true; } + else _have_desired_mid = false; + reDiscover(); + } + + // Begin fetching a chosen mid now (sets want + starts the manifest fetch / resume). Used by `ota pull` + // once the user picks a catalogued mOTA (the source is reached via the flooded GET_MANIFEST). + void pull(const uint8_t* mid, uint32_t target) { want(target); want_mid(mid); startFetch(mid, target); } + // Ask every known source for its catalog (populates `ota neighbors`). Async — rows arrive via OTA_HAVE. + 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; } + + // Codec compatibility: a node only fetches/accepts fw it can actually apply. CODEC_FULL is always + // acceptable; the platform's single delta codec is set here (ESP32 A/B -> sequential, nRF52 single- + // slot -> in-place). A mismatching `.mota` is rejected at OTA_ADV time, before fetching anything. + void set_apply_codec(uint8_t c) { _apply_codec = c; } + // A platform may apply MORE than one delta codec (ESP32 does both sequential AND in-place, so a single + // in-place `.mota` can target both ESP32 and nRF52). 0xFF = unset. + void set_apply_codec2(uint8_t c) { _apply_codec2 = c; } + bool codecOk(uint8_t c) const { return c == CODEC_FULL || c == _apply_codec || c == _apply_codec2; } + + // Auto-fetch policy (manual `ota pull` always works regardless): 0=off (discover only), 1=any + // compatible own-target advert, 2=only signed adverts. Conservative default = off. + static const uint8_t AUTOFETCH_OFF = 0, AUTOFETCH_ANY = 1, AUTOFETCH_SIGNED = 2; + void set_autofetch(uint8_t p) { _autofetch = p; reDiscover(); } + uint8_t autofetch() const { return _autofetch; } + + // Resume checkpoint cadence (runtime-tunable, persisted in NodePrefs): persist progress every N + // committed blocks. 0 = never (resume only from a finalized container). Default OTA_CHECKPOINT_BLOCKS. + void set_checkpoint_blocks(uint16_t n) { _checkpoint_blocks = n; } + uint16_t checkpoint_blocks() const { return _checkpoint_blocks; } + bool fetched_is_signed() const { return (_fflags & MFLAG_SIGNED) != 0; } // flags of the fetched manifest + + // 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 + void loop(); // drive fetch (re-request missing blocks) + + // Drop the current fetch session back to IDLE (so a fresh `ota pull` / advert starts a new one). Also + // stops re-seeding a previously-completed download — callers clear the staging store right after, so the + // re-seed view would otherwise advertise a mota we can no longer serve. + void reset_session() { + _fstate = IDLE; _have = 0; _req_count = 0; _mf_retries = 0; + clearReassembly(); + _loop_last_have = 0; _loop_last_mask = 0; + _mf_total = 0; _mf_mask = 0; _mf_len = 0; + unserveFetched(); + } + + FetchState fetchState() const { return _fstate; } + uint32_t blocksHave() const { return _have; } + uint32_t blocksTotal() const { return _fbc; } + const uint8_t* fetchManifestId() const { return _fid; } + + // --- discovery catalog (for `ota neighbors`): mOTAs heard around us via OTA_HAVE, deduped by mid --- + static const uint8_t OTA_CAT_SEEDERS = 4; // distinct sources tracked per catalog row (for "N nodes have it") + struct CatRow { + uint8_t mid[4]; + uint32_t target_id, fw_version; + uint8_t codec, flags; + uint8_t seeders[OTA_CAT_SEEDERS][4]; // distinct sources advertising this mid (deduped; capped) + uint8_t n_seeders; // count of the above (capped at OTA_CAT_SEEDERS) — "N+ nodes have it" + uint32_t have_max; // best block-count any source reported (== total when a full copy exists) + uint32_t last_ms; + }; + uint8_t catalogCount() const { return _n_cat; } + const CatRow* catalogRow(uint8_t i) const { return i < _n_cat ? &_catalog[i] : nullptr; } + uint8_t sourceCount() const { return _n_src; } // distinct OTA sources (beacon senders) heard + +private: + void emit(const uint8_t* b, uint16_t n, bool flood) { if (_send && n) _send(_ctx, b, n, flood); } + void handleAdv(const uint8_t* m, uint16_t n); // beacon -> sources table (+ query if interested) + void handleQuery(const uint8_t* m, uint16_t n); // serve: reply OTA_HAVE catalog + void handleHave(const uint8_t* m, uint16_t n); // peer: catalog rows (+ startFetch if a row matches) + void handleGetManifest(const uint8_t* m, uint16_t n); + void handleManifest(const uint8_t* m, uint16_t n); + 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); + void startFetch(const uint8_t* mid, uint32_t target); // begin/resume a fetch of a chosen mid + bool wantRow(const uint8_t* mid, uint32_t target, uint8_t codec, uint8_t flags) const; // fetch this row? + void noteOverheardReq(const uint8_t* m, uint16_t n); // observe a peer's OTA_REQ (swarm de-dup) + uint32_t rngNext() { _rng = _rng * 1664525u + 1013904223u; return _rng; } // per-node LCG (block pick/jitter) + void seedBlockRng() { _rng = (rd_u32le(_seeder_id) ^ rd_u32le(_fid)) | 1u; } // distinct per node (id^mid) + void clearReassembly(); // forget the in-flight block (reset to NO_BLOCK) + void armFirstReqHold(); // de-sync the first REQ across swarm peers (jitter) + uint32_t pickMissingBlock(); // choose the next block to request (swarm-aware) + 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) + bool loadSource(const ServeEntry& e); // load an external mota into _srcv (head+leaves) + void registerSelfEntry(); // (re)build entry[0] from view0 + static bool srcReadTramp(void* c, uint32_t off, uint8_t* buf, uint32_t len); // source payload reader + void serveFetched(); // after COMPLETE: re-seed the staged mota (epidemic) + void unserveFetched(); // stop re-seeding (store about to be overwritten) + void emitBlockData(const uint8_t* mid, uint32_t idx, const uint8_t* data, uint32_t blen); // DATA fragments + bool recentlyServed(uint32_t blk) const; // a peer just broadcast this block's DATA? + void noteOverheardData(const uint8_t* m, uint16_t n); // remember overheard DATA (serve de-dup) + void sendQuery(const uint8_t* seeder, const uint8_t* digest, uint32_t filter_target); // ask a source for its catalog + void scheduleQuery(const uint8_t* seeder, const uint8_t* digest); // jittered + suppressible + void reDiscover() { for (uint8_t i = 0; i < _n_src; i++) _sources[i].have_catalog = false; _pq_active = false; } + void setDigest(uint8_t out[4]) const; // sha2-256:4 over our served mids + bool blockPresent(uint32_t i) const; + void requestMissing(); + uint32_t blockLen(uint32_t i) const; + + uint32_t _target = 0; + OtaSend _send = nullptr; + void* _ctx = nullptr; + + // serve (multi-mota): view0 = our own fw / a RAM `.mota`; _srcv = the currently-loaded external mota. + ServeView _view0; + ServeView _srcv; + uint8_t _srcv_mid[4] = {0}; + SrcReadCtx _srcv_rdctx = {nullptr, 0, 0}; + ServeEntry _serve[OTA_MAX_SERVE]; // catalog (what we advertise) — entry 0 is view0 + uint8_t _n_serve = 0; + MotaSource* _src_list[OTA_MAX_SOURCE_OBJ] = {nullptr}; + uint8_t _n_src_obj = 0; + bool _fetch_served = false; // we re-seed our last completed download (epidemic spread) + MotaDesc _fetch_desc; // its catalog descriptor (mid + region offsets) + uint8_t _src_manifest[OTA_SRC_MANIFEST_MAX]; // manifest-minus-leaves of the loaded source mota + uint8_t _src_leaves[OTA_PROOFGEN_SCRATCH]; // leaves[] of the loaded source mota (<=1024 blocks) + uint8_t _scratch[OTA_PROOFGEN_SCRATCH]; // proof-gen / fetch root-check working buffer + + // fetch + OtaStore* _fetch = nullptr; + FetchState _fstate = IDLE; + uint8_t _fid[4] = {0}; + uint8_t _froot[4] = {0}; + uint32_t _ftotal = 0, _fpoff = 0, _floff = 0, _fpsize = 0, _fbc = 0, _fbs = 0; + uint32_t _have = 0; + uint32_t _req_start = 0, _req_count = 0; // last block requested (per-block serial flow; telemetry) + uint32_t _loop_last_have = 0; // for stall detection in loop() + // swarm load-spreading (so 50 fetchers don't all hammer the seeder for the same block in lockstep) + uint32_t _rng = 0; // per-node LCG state (seeded from seeder_id^fid) + uint32_t _req_hold_at = 0; // _now_ms before which we hold the first REQ (startup jitter) + uint32_t _peer_req_block = NO_BLOCK; // a block a peer just REQ'd (its broadcast DATA will fill us) + uint32_t _peer_req_at = 0; // when we overheard it (suppression window) + // serve-side de-dup: blocks whose DATA we recently overheard ANOTHER holder broadcast (don't re-serve) + uint32_t _recent_blk[8]; + uint32_t _recent_at[8] = {0}; + uint8_t _recent_i = 0; + 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; + uint8_t _apply_codec = CODEC_DETOOLS_SEQUENTIAL; // platform's delta codec (OtaContext sets it) + uint8_t _apply_codec2 = 0xFF; // optional 2nd accepted delta codec (ESP32: in-place) + uint8_t _seeder_id[4] = {0,0,0,0}; // our node id (pubkey[0:4]) for advert seeder counting + uint8_t _autofetch = AUTOFETCH_OFF; // auto-fetch policy (persisted in NodePrefs) + uint16_t _checkpoint_blocks = OTA_CHECKPOINT_BLOCKS; // resume checkpoint cadence (persisted) + uint8_t _fflags = 0; // flags of the manifest currently being fetched + // multi-fragment reassembly of the current block (per-block 2-phase: fetch data, then its proof) + uint32_t _reasm_block = NO_BLOCK; // block being reassembled / awaiting proof (none) + uint16_t _reasm_mask = 0; // received FRAG_DATA-slice bitmap (bit k = slice @ k*FRAG_DATA) + uint16_t _reasm_need = 0; // full mask once all slices of the current block are in + bool _awaiting_proof = false; // data complete; REQ_PROOF sent, verify on PROOF + uint16_t _loop_last_mask = 0; // fragment-level stall detection in loop() + uint8_t _reasm_buf[OTA_MAX_BLOCK]; + // 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) + uint8_t _mf_total = 0; // frag_total of the manifest being reassembled (0 = none) + uint16_t _mf_mask = 0; // received manifest-fragment bitmap + uint32_t _mf_len = 0; // assembled manifest length (set by the last fragment) + + // discovery: heard sources (beacon senders) + the catalog assembled from their OTA_HAVE replies + struct Source { uint8_t seeder[4]; uint8_t digest[4]; uint8_t n_motas; uint32_t last_ms; bool have_catalog; }; + Source _sources[OTA_MAX_SOURCES]; + uint8_t _n_src = 0; + CatRow _catalog[OTA_MAX_CATALOG]; + uint8_t _n_cat = 0; + uint32_t _now_ms = 0; // coarse clock (fed by set_clock; for ages/LRU/jitter) + // pending catalog query (jittered + suppressed on overhearing a matching QUERY/HAVE — anti-storm) + bool _pq_active = false; + uint8_t _pq_seeder[4] = {0}; + uint8_t _pq_digest[4] = {0}; + uint32_t _pq_at = 0; // _now_ms deadline to actually send the query +}; + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaProtocol.cpp b/src/helpers/ota/OtaProtocol.cpp new file mode 100644 index 00000000..50e61476 --- /dev/null +++ b/src/helpers/ota/OtaProtocol.cpp @@ -0,0 +1,137 @@ +#include "OtaProtocol.h" +#include + +namespace mesh { +namespace ota { + +// Little-endian cursor helpers. +namespace { +struct W { + uint8_t* p; uint16_t cap; uint16_t n; bool ok; + W(uint8_t* b, uint16_t c) : p(b), cap(c), n(0), ok(true) {} + void u8(uint8_t v) { if (n + 1 > cap) { ok = false; return; } p[n++] = v; } + void u16(uint16_t v){ u8(v & 0xFF); u8(v >> 8); } + void u32(uint32_t v){ u8(v); u8(v >> 8); u8(v >> 16); u8(v >> 24); } + void raw(const uint8_t* d, uint16_t l) { if (n + l > cap) { ok = false; return; } memcpy(p + n, d, l); n += l; } +}; +struct R { + const uint8_t* p; uint16_t len; uint16_t n; bool ok; + R(const uint8_t* b, uint16_t l) : p(b), len(l), n(0), ok(true) {} + uint8_t u8() { if (n + 1 > len) { ok = false; return 0; } return p[n++]; } + uint16_t u16() { uint16_t a = u8(); return a | ((uint16_t)u8() << 8); } + uint32_t u32() { uint32_t a = u8(); a |= (uint32_t)u8() << 8; a |= (uint32_t)u8() << 16; a |= (uint32_t)u8() << 24; return a; } + const uint8_t* raw(uint16_t l) { if (n + l > len) { ok = false; return nullptr; } const uint8_t* r = p + n; n += l; return r; } + uint16_t remaining() const { return len - n; } +}; +} // namespace + +uint16_t encode_adv(uint8_t* buf, uint16_t cap, const AdvMsg& m) { // tiny per-node beacon + W w(buf, cap); w.u8(OTA_ADV); w.raw(m.seeder_id, 4); w.u8(m.n_motas); w.raw(m.set_digest, 4); + return w.ok ? w.n : 0; +} +bool decode_adv(const uint8_t* buf, uint16_t len, AdvMsg& m) { + R r(buf, len); if (r.u8() != OTA_ADV) return false; + const uint8_t* sid = r.raw(4); if (sid) memcpy(m.seeder_id, sid, 4); + m.n_motas = r.u8(); + const uint8_t* d = r.raw(4); if (d) memcpy(m.set_digest, d, 4); + return r.ok; +} + +uint16_t encode_query(uint8_t* buf, uint16_t cap, const QueryMsg& m) { + W w(buf, cap); w.u8(OTA_QUERY); w.raw(m.seeder_id, 4); w.raw(m.set_digest, 4); w.u32(m.filter_target); + return w.ok ? w.n : 0; +} +bool decode_query(const uint8_t* buf, uint16_t len, QueryMsg& m) { + R r(buf, len); if (r.u8() != OTA_QUERY) return false; + const uint8_t* sid = r.raw(4); if (sid) memcpy(m.seeder_id, sid, 4); + const uint8_t* dg = r.raw(4); if (dg) memcpy(m.set_digest, dg, 4); + m.filter_target = r.u32(); + return r.ok; +} + +uint16_t encode_have(uint8_t* buf, uint16_t cap, const HaveMsg& m) { + W w(buf, cap); w.u8(OTA_HAVE); w.raw(m.seeder_id, 4); w.raw(m.set_digest, 4); + w.u8(m.frag_idx); w.u8(m.frag_total); w.u8(m.n_rows); w.raw(m.rows, (uint16_t)m.n_rows * OTA_HAVE_ROW_BYTES); + return w.ok ? w.n : 0; +} +bool decode_have(const uint8_t* buf, uint16_t len, HaveMsg& m) { + R r(buf, len); if (r.u8() != OTA_HAVE) return false; + const uint8_t* sid = r.raw(4); if (sid) memcpy(m.seeder_id, sid, 4); + const uint8_t* dg = r.raw(4); if (dg) memcpy(m.set_digest, dg, 4); + m.frag_idx = r.u8(); m.frag_total = r.u8(); m.n_rows = r.u8(); + m.rows = r.raw((uint16_t)m.n_rows * OTA_HAVE_ROW_BYTES); + return r.ok; +} + +uint16_t encode_get_manifest(uint8_t* buf, uint16_t cap, const GetManifestMsg& m) { + W w(buf, cap); w.u8(OTA_GET_MANIFEST); w.raw(m.manifest_id, 4); + return w.ok ? w.n : 0; +} +bool decode_get_manifest(const uint8_t* buf, uint16_t len, GetManifestMsg& m) { + R r(buf, len); if (r.u8() != OTA_GET_MANIFEST) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + return r.ok; +} + +uint16_t encode_manifest(uint8_t* buf, uint16_t cap, const ManifestMsg& m) { + W w(buf, cap); w.u8(OTA_MANIFEST); w.raw(m.manifest_id, 4); w.u8(m.frag_idx); w.u8(m.frag_total); + w.raw(m.bytes, m.len); + return w.ok ? w.n : 0; +} +bool decode_manifest(const uint8_t* buf, uint16_t len, ManifestMsg& m) { + R r(buf, len); if (r.u8() != OTA_MANIFEST) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + m.frag_idx = r.u8(); m.frag_total = r.u8(); + m.len = r.remaining(); m.bytes = r.raw(m.len); + return r.ok; +} + +uint16_t encode_req(uint8_t* buf, uint16_t cap, const ReqMsg& m) { + W w(buf, cap); w.u8(OTA_REQ); w.raw(m.manifest_id, 4); w.u16(m.start_block); w.u8(m.count); + return w.ok ? w.n : 0; +} +bool decode_req(const uint8_t* buf, uint16_t len, ReqMsg& m) { + 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.start_block = r.u16(); m.count = r.u8(); + return r.ok; +} + +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); + return w.ok ? w.n : 0; +} +bool decode_data(const uint8_t* buf, uint16_t len, DataMsg& m) { + R r(buf, len); if (r.u8() != OTA_DATA) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + m.block_idx = r.u16(); m.frag_off = r.u16(); + m.data_len = r.remaining(); m.data = r.raw(m.data_len); + return r.ok; +} + +uint16_t encode_req_proof(uint8_t* buf, uint16_t cap, const ReqProofMsg& m) { + W w(buf, cap); w.u8(OTA_REQ_PROOF); w.raw(m.manifest_id, 4); w.u16(m.block_idx); + return w.ok ? w.n : 0; +} +bool decode_req_proof(const uint8_t* buf, uint16_t len, ReqProofMsg& m) { + R r(buf, len); if (r.u8() != OTA_REQ_PROOF) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + m.block_idx = r.u16(); + return r.ok; +} + +uint16_t encode_proof(uint8_t* buf, uint16_t cap, const ProofMsg& m) { + W w(buf, cap); w.u8(OTA_PROOF); w.raw(m.manifest_id, 4); w.u16(m.block_idx); + w.u8(m.n_proof); w.raw(m.proof, (uint16_t)m.n_proof * 4); + return w.ok ? w.n : 0; +} +bool decode_proof(const uint8_t* buf, uint16_t len, ProofMsg& m) { + R r(buf, len); if (r.u8() != OTA_PROOF) return false; + const uint8_t* id = r.raw(4); if (id) memcpy(m.manifest_id, id, 4); + m.block_idx = r.u16(); m.n_proof = r.u8(); m.proof = r.raw((uint16_t)m.n_proof * 4); + return r.ok; +} + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaProtocol.h b/src/helpers/ota/OtaProtocol.h new file mode 100644 index 00000000..d8385353 --- /dev/null +++ b/src/helpers/ota/OtaProtocol.h @@ -0,0 +1,112 @@ +#pragma once + +#include +#include +#include "OtaFormat.h" + +// Encode/decode for the OTA LoRa messages (docs/ota_protocol.md §8). Each message is a packet payload: +// [0]=ota_msg_type, then a fixed body. Portable + allocation-free; unit-tested on the host. +// +// manifest_id == the manifest's merkle_root (4 bytes), a compact content id. + +namespace mesh { +namespace ota { + +// ---- OTA_ADV: tiny per-NODE beacon (flood, periodic). CONSTANT size regardless of how many mOTAs a +// node serves — it just says "I'm a source, here's how many + a digest of my set". A peer that's +// interested asks for the catalog via OTA_QUERY. (Replaces the old per-mOTA advert so a folder node with +// N images costs one 10-byte beacon, not N adverts.) ---- +struct AdvMsg { + uint8_t seeder_id[4]; // advertiser's node id = pubkey[0:4]; the QUERY address + distinct-source id + uint8_t n_motas; // # of complete, servable mOTAs (saturates at 255) + uint8_t set_digest[4]; // sha2-256:4 over the sorted set of served mids; "did my offering change?" +}; + +// ---- OTA_QUERY: "list what you serve" — addressed to a source by seeder_id, FLOODED so neighbours +// overhear it (storm suppression). set_digest identifies the offering being asked about (so an overhearer +// can suppress its own pending query for the same {source,digest}). filter_target=0 = everything. ---- +struct QueryMsg { + uint8_t seeder_id[4]; // which source this query is for (the source matches its own id) + uint8_t set_digest[4]; // the offering digest we're asking about (for overhear-suppression) + uint32_t filter_target; // 0 = all (the scalable default); else only mOTAs for this target_id +}; + +// ---- OTA_HAVE: the compact catalog (source -> mesh), FLOODED + tagged with set_digest so EVERY node +// that overhears it caches the rows (passive, no query needed). Fragmented. ---- +// body: seeder_id(4) set_digest(4) frag_idx(1) frag_total(1) n_rows(1) rows[ mid(4) target(4) fwver(4) codec(1) flags(1) ] +struct HaveRow { uint8_t mid[4]; uint32_t target_id; uint32_t fw_version; uint8_t codec_id; uint8_t flags; + uint16_t have_count; }; // blocks the advertiser holds (== block_count if complete; less => partial source) +struct HaveMsg { + uint8_t seeder_id[4]; + uint8_t set_digest[4]; // the offering this catalog describes (overhearers cache by it) + uint8_t frag_idx, frag_total; + uint8_t n_rows; // rows in THIS fragment + const uint8_t* rows; // points into buf: n_rows * OTA_HAVE_ROW_BYTES +}; +static const uint8_t OTA_HAVE_ROW_BYTES = 16; // mid4 + target4 + fwver4 + codec1 + flags1 + have_count2 + +// ---- OTA_GET_MANIFEST: request the manifest for a content id (direct) ---- +struct GetManifestMsg { uint8_t manifest_id[4]; }; + +// ---- OTA_MANIFEST: the manifest-minus-leaves[], fragmented (direct) ---- +// body: manifest_id(4) frag_idx(1) frag_total(1) bytes[] +struct ManifestMsg { + uint8_t manifest_id[4]; + uint8_t frag_idx, frag_total; + const uint8_t* bytes; uint16_t len; +}; + +// ---- OTA_REQ: request a window of blocks (direct) ---- +struct ReqMsg { uint8_t manifest_id[4]; uint16_t start_block; uint8_t count; }; + +// ---- OTA_DATA: one self-describing fragment of a block's data (proof is fetched separately) ---- +// body: manifest_id(4) block_idx(2) frag_off(2) data[] +// `frag_off` is the byte offset of `data` within block `block_idx` (global position = block_idx*block_size +// + frag_off), so a fragment is self-placing and can be requested from ANY peer (BitTorrent-style). +struct DataMsg { + uint8_t manifest_id[4]; + uint16_t block_idx; + uint16_t frag_off; + const uint8_t* data; uint16_t data_len; +}; + +// ---- OTA_REQ_PROOF: request the merkle proof for one (reassembled) block (direct) ---- +struct ReqProofMsg { uint8_t manifest_id[4]; uint16_t block_idx; }; + +// ---- OTA_PROOF: the merkle proof (ordered sibling digests) for one block (direct) ---- +struct ProofMsg { uint8_t manifest_id[4]; uint16_t block_idx; uint8_t n_proof; const uint8_t* proof; }; + +// Each encode_* returns the total payload length (incl. the leading msg-type byte), 0 on overflow. +// Each decode_* returns true on success (and points struct fields into `buf`). + +uint16_t encode_adv(uint8_t* buf, uint16_t cap, const AdvMsg& m); +bool decode_adv(const uint8_t* buf, uint16_t len, AdvMsg& m); + +uint16_t encode_query(uint8_t* buf, uint16_t cap, const QueryMsg& m); +bool decode_query(const uint8_t* buf, uint16_t len, QueryMsg& m); + +uint16_t encode_have(uint8_t* buf, uint16_t cap, const HaveMsg& m); +bool decode_have(const uint8_t* buf, uint16_t len, HaveMsg& m); + +uint16_t encode_get_manifest(uint8_t* buf, uint16_t cap, const GetManifestMsg& m); +bool decode_get_manifest(const uint8_t* buf, uint16_t len, GetManifestMsg& m); + +uint16_t encode_manifest(uint8_t* buf, uint16_t cap, const ManifestMsg& m); +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_data(uint8_t* buf, uint16_t cap, const DataMsg& m); +bool decode_data(const uint8_t* buf, uint16_t len, DataMsg& m); + +uint16_t encode_req_proof(uint8_t* buf, uint16_t cap, const ReqProofMsg& m); +bool decode_req_proof(const uint8_t* buf, uint16_t len, ReqProofMsg& m); + +uint16_t encode_proof(uint8_t* buf, uint16_t cap, const ProofMsg& m); +bool decode_proof(const uint8_t* buf, uint16_t len, ProofMsg& m); + +inline uint8_t ota_msg_type(const uint8_t* buf, uint16_t len) { return len ? buf[0] : 0xFF; } + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaSource.h b/src/helpers/ota/OtaSource.h new file mode 100644 index 00000000..64c8fce6 --- /dev/null +++ b/src/helpers/ota/OtaSource.h @@ -0,0 +1,55 @@ +#pragma once + +#include +#include +#include "OtaFormat.h" + +// Transport-agnostic "folder of firmware" abstraction (docs/ota_protocol.md §9). A node can RELAY +// `.mota` images it does not hold in flash — a user drops several `.mota` (different architectures) into +// some external store and the node advertises + serves them as if it held them. Peers just see "this node +// has N mOTAs"; the node knows they are external. The store is reached through a MotaSource, so the SAME +// serve code drives a USB-serial host daemon, BLE, a WiFi URL list, an NFS/samba mount, ... — only the +// `read()` plumbing differs per transport. +// +// The relay is TRUSTLESS: the fetcher verifies every block against the signed merkle root, so a source is +// never trusted. A wrong descriptor or wrong bytes simply makes the fetch fail its merkle/signature check +// — a malicious or buggy source cannot forge firmware, only deny it. + +namespace mesh { +namespace ota { + +// A parsed top-level descriptor of one `.mota` a source provides: enough to advertise it in the catalog +// AND to locate every region for serving, WITHOUT holding the whole image in RAM. Offsets are absolute +// byte positions within the `.mota` container (which always begins MAGIC(4) total(4) manifest...). +struct MotaDesc { + uint8_t mid[4] = {0}; // merkle_root (the content id peers fetch by) + uint32_t target_id = 0; + uint32_t fw_version = 0; + uint8_t codec_id = 0; + uint8_t flags = 0; + uint32_t total_size = 0; // full `.mota` length (bytes) + uint32_t leaves_off = 0; // byte offset of the merkle leaves[] (manifest-minus-leaves = [8, leaves_off)) + uint32_t block_count = 0; // == number of leaves (== number of payload blocks) + uint32_t payload_off = 0; // byte offset of the payload + uint32_t payload_size = 0; +}; + +// One or more complete `.mota` images, reachable as random-access bytes. Implementations are device-side +// and transport-specific (the engine in OtaManager is portable and never includes this directly for I/O; +// it only calls through the interface). +class MotaSource { +public: + virtual ~MotaSource() {} + // Number of complete, servable mOTAs this source currently offers (may change as the folder changes; + // the manager re-enumerates on add_source / refresh). + virtual uint8_t count() = 0; + // Cheap metadata + region offsets for mota `idx`. False if idx is out of range or unparsable. + virtual bool describe(uint8_t idx, MotaDesc& out) = 0; + // Random-access read of `len` bytes at absolute offset `off` of mota `idx` into `buf`. Returns true iff + // exactly `len` bytes were produced. May block on the transport (serial round-trip); OTA is lowest + // priority so latency here is acceptable. + virtual bool read(uint8_t idx, uint32_t off, uint8_t* buf, uint32_t len) = 0; +}; + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaStore.h b/src/helpers/ota/OtaStore.h new file mode 100644 index 00000000..27f2fb29 --- /dev/null +++ b/src/helpers/ota/OtaStore.h @@ -0,0 +1,103 @@ +#pragma once + +#include +#include +#include +#include "OtaFormat.h" // MOTA_MAGIC (resume: detect a persisted partial container) + +// Staging backend for an in-transit `.mota` (docs/ota_protocol.md §7). Blocks may arrive out of order +// and progress must survive reboots, so the store is random-access. The transfer/verify logic is +// written against this interface; concrete impls are per-platform (RAM for tests/bring-up; persistent +// flash — ESP32 OTA slot / nRF52 raw region — for production, dropped in behind the same interface). + +namespace mesh { +namespace ota { + +class OtaStore { +public: + virtual ~OtaStore() {} + // Prepare staging for a container of `total_size` bytes (erases/clears). false if it won't fit. + virtual bool begin(uint32_t total_size) = 0; + virtual bool write(uint32_t offset, const uint8_t* data, uint32_t len) = 0; + virtual bool read(uint32_t offset, uint8_t* buf, uint32_t len) const = 0; + virtual uint32_t capacity() const = 0; + virtual uint32_t staged_size() const = 0; // total_size from begin(), 0 if none + virtual void clear() = 0; + + // Optional: declare the size of the leading metadata (header + manifest + merkle leaves, i.e. + // everything before the payload). A flash-backed store keeps that region — which is updated + // throughout the transfer (a leaf is committed per block) — pinned in one RAM page, so it can + // flush the bulk payload page-by-page without re-erasing the leaves' page on every block. + // Returns false if the metadata won't fit the store's pinned region (transfer is then refused). + virtual bool set_meta_size(uint32_t meta_bytes) { (void)meta_bytes; return true; } + + // Optional: commit any RAM-buffered data to persistent storage. Called once when the transfer + // reaches COMPLETE (radio idle), so a flash store does its page writes off the RX critical path. + // After this returns, a flash store's data() view is coherent. No-op for purely in-RAM stores. + virtual void finalize() {} + + // Optional: persist in-progress state (the metadata/leaf-progress page + any open payload buffer) so a + // reboot mid-transfer can resume. Called by OtaManager every OTA_CHECKPOINT_BLOCKS committed blocks. + // A flash store flushes its pinned meta page + the open payload page (consistency: payload-before-leaves) + // so every block whose leaf is persisted also has its payload in flash. No-op for RAM stores. Infrequent + // (at LoRa block rates, ~once per many minutes) so the extra erases don't matter. + virtual void checkpoint() {} + + // Optional: re-attach to a container ALREADY persisted in the backing store (after a reboot), WITHOUT + // erasing. Returns true if a syntactically valid container (MOTA_MAGIC header + plausible total) is + // present and the store is now set up to read/continue-writing it; false if none (caller starts fresh). + // OtaManager then reads + parses the stored manifest to recompute geometry and resume the fetch. + virtual bool reopen() { return false; } + + // Optional: declare the container's logical layout once the manifest is parsed, BEFORE begin(), so a + // store backed by a single spare A/B partition (ESP32) can choose placement and reject an unfittable + // fetch up front. A FULL image's payload IS the final firmware (no decode), so it can stream straight + // to the inactive slot's offset 0 while the small meta/leaves/trailer persist elsewhere; a delta's + // whole container is staged together (the decoder reads the patch from it at apply). image_size is the + // reconstructed image; [payload_off, payload_off+payload_size) is the payload region in the container. + // Return false if it cannot fit the backing store (the transfer is then refused before any block). + virtual bool plan_layout(bool is_full, uint32_t image_size, uint32_t payload_off, uint32_t payload_size) { + (void)is_full; (void)image_size; (void)payload_off; (void)payload_size; return true; + } +}; + +// Fixed-capacity RAM store — for native tests and device bring-up of the transfer/verify path. +// (Does NOT survive reboot; a persistent flash store replaces it for production — see D1.) +template +class OtaStoreRam : public OtaStore { + uint8_t _buf[CAP] = {}; // zero-init so a never-written store's reopen() finds no MOTA_MAGIC + uint32_t _total = 0; +public: + bool begin(uint32_t total_size) override { + if (total_size > CAP) return false; + _total = total_size; + memset(_buf, 0xFF, total_size); // mimic erased flash (so unfilled leaf slots read as 'missing') + return true; + } + bool write(uint32_t off, const uint8_t* d, uint32_t len) override { + if ((uint64_t)off + len > _total) return false; + memcpy(_buf + off, d, len); + return true; + } + bool read(uint32_t off, uint8_t* b, uint32_t len) const override { + if ((uint64_t)off + len > _total) return false; + memcpy(b, _buf + off, len); + return true; + } + uint32_t capacity() const override { return CAP; } + uint32_t staged_size() const override { return _total; } + void clear() override { _total = 0; } + // RAM doesn't survive a real reboot, but the buffer persists within a process — enough to exercise the + // manager's resume path in native tests. Recover `total` from the stored header so read() bounds work. + bool reopen() override { + if (memcmp(_buf, MOTA_MAGIC, 4) != 0) return false; + uint32_t t = (uint32_t)_buf[4] | ((uint32_t)_buf[5] << 8) | ((uint32_t)_buf[6] << 16) | ((uint32_t)_buf[7] << 24); + if (t < 13 || t > CAP) return false; + _total = t; + return true; + } + const uint8_t* data() const { return _buf; } // contiguous view (RAM store only) +}; + +} // namespace ota +} // namespace mesh