From f15772d90ca10a5359ea73eb5825896e727da27b Mon Sep 17 00:00:00 2001 From: Valentin Kivachuk Burda Date: Sat, 27 Jun 2026 16:28:03 +0200 Subject: [PATCH] Fix ota req proto --- .github/workflows/dev-firmware-rolling.yml | 161 ++++++++++------ docs/ota_protocol.md | 33 +++- src/helpers/ota/MotaContainer.cpp | 127 +++++------- src/helpers/ota/OtaByteIO.h | 42 ++++ src/helpers/ota/OtaManager.cpp | 214 +++++++++++++++++---- src/helpers/ota/OtaManager.h | 38 +++- src/helpers/ota/OtaProtocol.h | 5 +- test/test_ota/test_ota_core.cpp | 73 ++++++- tools/mota/dev_motas.py | 92 +++++++++ 9 files changed, 594 insertions(+), 191 deletions(-) create mode 100644 src/helpers/ota/OtaByteIO.h create mode 100644 tools/mota/dev_motas.py diff --git a/.github/workflows/dev-firmware-rolling.yml b/.github/workflows/dev-firmware-rolling.yml index 4e8c6a07..421e8431 100644 --- a/.github/workflows/dev-firmware-rolling.yml +++ b/.github/workflows/dev-firmware-rolling.yml @@ -1,11 +1,15 @@ # Rolling DEV firmware release (fork convenience). # -# On every push, rebuild ALL device firmwares and replace the assets of ONE rolling prerelease -# (tag `dev-latest`) so other devs can always grab the current binaries from the same place. Also packages -# a demo `.mota` (full + a same-image delta, which is tiny) so the OTA format can be exercised. +# On every push, rebuild a representative set of OTA-capable firmwares (ESP32 + nRF52, one per board family +# plus the RAK4631 / Heltec V3 test boards in every role) and replace the assets of ONE rolling prerelease +# (tag `dev-latest`). Each firmware ships with a full + same-image-delta `.mota` so the OTA format/transport +# can be exercised per board. UNSIGNED dev builds for testing only. # -# These are UNSIGNED development builds for testing only — not official releases. -# To limit which branches trigger this, add a `branches:` filter under `push:` below. +# `prepare` (re)creates the empty release, then each matrix build uploads its own assets to it — so one +# board that fails to compile only loses its own cell, and there is no cross-job artifact plumbing. +# +# To change which boards build, edit the `matrix.include` list below (env + its platform). Only OTA-capable +# (ENABLE_OTA) envs make sense here. The full OTA set is ~308 envs; this is a curated ~one-per-board subset. name: Dev Firmware (rolling) @@ -16,7 +20,6 @@ on: permissions: contents: write -# Only the latest push matters — cancel any in-flight run so the release tracks HEAD. concurrency: group: dev-firmware-rolling cancel-in-progress: true @@ -24,11 +27,80 @@ concurrency: env: RELEASE_TAG: dev-latest FIRMWARE_VERSION: dev - MOTA_ENV: Heltec_v3_repeater # representative ESP32 env whose built .bin carries EndF, used for the demo .mota jobs: - build-and-release: + prepare: runs-on: ubuntu-latest + steps: + - name: Clone Repo + uses: actions/checkout@v6 + - name: (Re)create the empty rolling release + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + gh release delete "$RELEASE_TAG" --yes --cleanup-tag 2>/dev/null || true + NOTES=$(cat <\`. + + A representative set of OTA-capable boards (ESP32 + nRF52). Each firmware ships a \`.full.mota\` + (the flashable image) and a \`.delta.mota\` (a same-image patch — intentionally tiny — to exercise + the delta path: sequential on ESP32, in-place on nRF52/RAK4631). + EOF + ) + gh release create "$RELEASE_TAG" \ + --title "Dev firmware (latest commit)" \ + --notes "$NOTES" \ + --prerelease \ + --target "$GITHUB_SHA" + + build: + needs: prepare + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + include: + # --- nRF52 (OTA in-place; RAK4631 is the validated target, all roles) --- + - { env: RAK_4631_repeater, platform: NRF52 } + - { env: RAK_4631_room_server, platform: NRF52 } + - { env: RAK_4631_companion_radio_ble, platform: NRF52 } + - { env: RAK_4631_companion_radio_usb, platform: NRF52 } + # --- ESP32 test board (Heltec V3, all roles) --- + - { env: Heltec_v3_repeater, platform: ESP32 } + - { env: Heltec_v3_room_server, platform: ESP32 } + - { env: Heltec_v3_companion_radio_ble, platform: ESP32 } + - { env: Heltec_v3_companion_radio_usb, platform: ESP32 } + # --- ESP32, one representative per board family --- + - { env: Ebyte_EoRa-S3_room_server, platform: ESP32 } + - { env: Generic_E22_sx1262_repeater, platform: ESP32 } + - { env: Heltec_E213_repeater, platform: ESP32 } + - { env: Heltec_E290_repeater, platform: ESP32 } + - { env: Heltec_T190_repeater_, platform: ESP32 } + - { env: Heltec_ct62_repeater, platform: ESP32 } + - { env: Heltec_v2_repeater, platform: ESP32 } + - { env: LilyGo_T3S3_sx1262_repeater, platform: ESP32 } + - { env: LilyGo_TBeam_1W_repeater, platform: ESP32 } + - { env: LilyGo_TDeck_repeater, platform: ESP32 } + - { env: LilyGo_TETH_Elite_sx1262_repeater, platform: ESP32 } + - { env: LilyGo_TLora_V2_1_1_6_repeater, platform: ESP32 } + - { env: M5Stack_Unit_C6L_repeater, platform: ESP32 } + - { env: Station_G2_logging_repeater, platform: ESP32 } + - { env: Station_G3_ESP32_logging_repeater, platform: ESP32 } + - { env: T_Beam_S3_Supreme_SX1262_repeater, platform: ESP32 } + - { env: Tbeam_SX1262_repeater, platform: ESP32 } + - { env: ThinkNode_M2_room_server, platform: ESP32 } + - { env: Xiao_C3_repeater, platform: ESP32 } + - { env: Xiao_S3_WIO_repeater, platform: ESP32 } + - { env: heltec_tracker_v2_repeater, platform: ESP32 } + - { env: heltec_v4_expansionkit_repeater, platform: ESP32 } + - { env: nibble_screen_connect_repeater, platform: ESP32 } steps: - name: Clone Repo uses: actions/checkout@v6 @@ -36,66 +108,31 @@ jobs: - name: Setup Build Environment uses: ./.github/actions/setup-build-environment - # build.sh is best-effort: a board that fails to compile is skipped, the rest still land in out/. - # All three roles append to the same out/ directory. - - name: Build all firmwares (companion + repeater + room-server) - run: | - /usr/bin/env bash build.sh build-companion-firmwares - /usr/bin/env bash build.sh build-repeater-firmwares - /usr/bin/env bash build.sh build-room-server-firmwares - echo "Built artifacts:"; ls -la out + - name: Build ${{ matrix.env }} + run: /usr/bin/env bash build.sh build-firmware ${{ matrix.env }} - - name: Package demo .mota (full + same-image delta) + - name: Package .mota (full + same-image delta) run: | set -euo pipefail - pip install --quiet detools # delta codec only (a full .mota needs no extra deps) - BIN="$(ls out/${MOTA_ENV}-*.bin 2>/dev/null | grep -v -- '-merged' | head -1 || true)" - if [ -z "$BIN" ]; then - echo "::warning::no ${MOTA_ENV} .bin built — skipping demo .mota" - else - echo "Using firmware image: $BIN" - # full image .mota (codec 0): payload = the flashable image (BODY||EndF) - python3 tools/mota/mota.py build --fw "$BIN" --target-env "$MOTA_ENV" \ - --fw-version 0.0.0 --codec full --out "out/${MOTA_ENV}-demo.full.mota" - # delta .mota (codec 1, sequential+crle): base == target == same image -> near-empty patch - python3 tools/mota/mota.py build --fw "$BIN" --base "$BIN" --target-env "$MOTA_ENV" \ - --fw-version 0.0.1 --codec sequential --compression crle --out "out/${MOTA_ENV}-demo.delta.mota" - echo "Demo .mota sizes:"; ls -l out/${MOTA_ENV}-demo.*.mota - fi + pip install --quiet detools intelhex + SHA=$(git rev-parse --short HEAD) + python3 tools/mota/dev_motas.py \ + --env "${{ matrix.env }}" --platform "${{ matrix.platform }}" \ + --build-dir ".pio/build/${{ matrix.env }}" --target-env "${{ matrix.env }}" \ + --out-prefix "out/${{ matrix.env }}-${FIRMWARE_VERSION}-${SHA}" --work /tmp \ + || echo "::warning::.mota packaging failed for ${{ matrix.env }}" - - name: Upload workflow artifacts - uses: actions/upload-artifact@v7 - with: - name: dev-firmwares - path: out - if-no-files-found: warn - retention-days: 5 - - - name: Replace the rolling dev release + - name: Upload assets to the rolling release env: GH_TOKEN: ${{ github.token }} run: | set -euo pipefail - if [ -z "$(ls -A out 2>/dev/null)" ]; then - echo "::error::nothing built — not touching the release"; exit 1 + shopt -s nullglob + files=(out/*) + if [ ${#files[@]} -eq 0 ]; then + echo "::warning::no assets for ${{ matrix.env }}"; exit 0 fi - # delete the previous release + its tag so old binaries disappear, then recreate at HEAD - gh release delete "$RELEASE_TAG" --yes --cleanup-tag 2>/dev/null || true - NOTES=$(cat <\`. - - \`${MOTA_ENV}-demo.full.mota\` / \`*.delta.mota\` are OTA format demos built from this run's - \`${MOTA_ENV}\` image (the delta is a same-image patch, so it is intentionally tiny). - EOF - ) - gh release create "$RELEASE_TAG" out/* \ - --title "Dev firmware (latest commit)" \ - --notes "$NOTES" \ - --prerelease \ - --target "$GITHUB_SHA" + echo "Uploading: ${files[*]}" + # retry once for transient API hiccups when many matrix jobs upload concurrently + gh release upload "$RELEASE_TAG" "${files[@]}" --clobber \ + || (sleep 5 && gh release upload "$RELEASE_TAG" "${files[@]}" --clobber) diff --git a/docs/ota_protocol.md b/docs/ota_protocol.md index 1f99c228..9c8a0fe9 100644 --- a/docs/ota_protocol.md +++ b/docs/ota_protocol.md @@ -312,9 +312,13 @@ the beacon (steady state is query-free). For a single served mota, `set_digest = ``` OTA_QUERY (flood): seeder_id[4] set_digest[4] filter_target(uint32) # filter_target 0 = everything OTA_HAVE (flood): seeder_id[4] set_digest[4] frag_idx(1) frag_total(1) n_rows(1) rows[] - HaveRow (14 bytes, OTA_HAVE_ROW_BYTES): mid[4] target_id(4) fw_version(4) codec_id(1) flags(1) + HaveRow (16 bytes, OTA_HAVE_ROW_BYTES): mid[4] target_id(4) fw_version(4) codec_id(1) flags(1) have_count(2) ``` +`have_count` is how many blocks the advertiser currently holds (`== block_count` for a full copy, less for a +partial/in-progress source). It lets a fetcher see, per mid, **how many peers have it and at what progress** +— so it knows the firmware is on multiple peers and can trust the swarm (§8.6) rather than depend on one. + A node interested in a source's offering schedules a QUERY; the source replies with its full catalog as `OTA_HAVE` rows (fragmented if they exceed one packet — up to 12 rows per fragment). The heavy manifest is fetched per-mid only on commit (§8.3). @@ -382,6 +386,33 @@ OTA_PROOF: manifest_id[4] block_idx(uint16) n_proof(1) proof[] # n_ A served mota supports up to `OTA_MAX_BLOCK/4` leaves in the default 4 KB proof scratch (≤1024 blocks ≈ 1 MB payload); larger self-images pass a bigger scratch buffer. +### 8.6 Swarm load distribution (don't hammer one seeder) + +The discovery anti-storm (§8.2) stops 50 neighbours all *querying* one node. The same hazard exists for the +*transfer*: if one node has new firmware and 50 want it, naïve fetchers would all REQ the same blocks from +the same seeder. Because OTA is always lowest-priority (§8) the mesh won't collapse, but the transfer would +be needlessly slow and centralized. Mitigations (all in `OtaManager`, reusing the §8.2 jitter/suppress idea): + +- **Overhearing fills holes for free.** Every fetcher accepts any *broadcast* `OTA_DATA` for its mid, not + just data it requested. So within a broadcast neighbourhood, one peer's request serves everyone who hears it. +- **De-correlated requests.** A fetcher picks a **random** missing block (not lowest-first), so N fetchers + don't lockstep on the same block; collectively they pull different blocks and everyone overhears them all. + Each fetch also holds its first REQ a random `OTA_REQ_SPREAD_MS` so simultaneous starters don't burst together. +- **Request suppression.** Overhearing a peer's `OTA_REQ` for a block makes a fetcher spend its next REQ on a + *different* block (`OTA_REQ_SUPPRESS_MS`) — the broadcast DATA will fill the overheard one anyway. +- **Sources multiply (the key to "don't pull one node"):** + - **Re-seed after COMPLETE (epidemic).** A node that finishes a download advertises + serves it (it now has + all blocks *and* leaves, so it serves DATA and proofs). The origin seeds a few peers, they seed the next + ring, etc. — load on the origin drops from O(N) to ~O(log N). (Default `autoinstall=off` means a completed + node lingers as a seeder until the operator applies.) + - **Partial re-serve during the transfer.** A still-fetching node serves the **DATA** of blocks it already + holds (not proofs — it may lack sibling leaves), so peers can source bytes from it, not only the origin. +- **Serve de-dup.** A holder about to serve a block it just overheard *another* holder broadcast suppresses + its own send (`OTA_SERVE_SUPPRESS_MS`), so multiple sources of one mota don't duplicate-broadcast it. + +All serving stays reactive and lowest-priority, so seeding never competes with real traffic — the system is +"eventually upgradable": a busy node simply delays OTA until it has spare airtime. + --- ## 9. Identity, trust & versioning diff --git a/src/helpers/ota/MotaContainer.cpp b/src/helpers/ota/MotaContainer.cpp index 859337ce..c27c084a 100644 --- a/src/helpers/ota/MotaContainer.cpp +++ b/src/helpers/ota/MotaContainer.cpp @@ -1,6 +1,7 @@ #include "MotaContainer.h" #include "MerkleTree.h" #include "Multihash.h" +#include "OtaByteIO.h" #include namespace mesh { @@ -14,99 +15,61 @@ bool MotaManifest::is_approved() const { return approval && memcmp(approval, APPROVAL_YES, 4) == 0; } +// Read the manifest's fixed head + conditional/variable fields from a cursor (shared by the full-container +// and standalone-manifest parsers). Reads each field by name in declaration order (docs/ota_protocol.md §4); +// `signed_off` is the cursor base the signature is measured from (manifest_start). Leaves/payload (only in +// a full container) are read by the caller. Returns false on any over-read or bad format_ver. +static bool parse_manifest_fields(ByteReader& r, uint32_t signed_off, MotaManifest& out) { + out.format_ver = r.u8(); + if (out.format_ver != MOTA_FORMAT_VER) return false; + out.flags = r.u8(); + out.hash_algo = r.u8(); + out.target_id = r.u32(); + out.fw_version = r.u32(); + out.image_size = r.u32(); + out.payload_size = r.u32(); + out.block_size_log2 = r.u8(); + out.merkle_root = r.take(4); + out.image_hash = r.take(32); + out.codec_id = r.u8(); + out.hw_id = r.take(32); // 32-byte NUL-padded hardware tag (signed) + if (!out.is_full()) out.base_hash = r.take(8); + if (out.is_signed()) { + out.signer_pubkey = r.take(32); + out.signed_len = r.pos() - signed_off; // signature covers manifest_start .. here (exclusive) + out.signature = r.take(64); + } else { + out.signed_len = r.pos() - signed_off; + } + out.approval = r.take(4); + if (!r.ok) return false; + if (out.block_size_log2 == 0 || out.block_size_log2 > 24 || out.payload_size == 0) return false; + out.block_count = (out.payload_size + out.block_size() - 1) / out.block_size(); + return out.block_count != 0; +} + bool mota_parse(const uint8_t* buf, uint32_t len, MotaManifest& out) { out = MotaManifest(); if (len < 4 + 4 + 5) return false; if (memcmp(buf, MOTA_MAGIC, 4) != 0) return false; if (memcmp(buf + len - 5, MOTA_TRAILER, 5) != 0) return false; - uint32_t total = rd_u32(buf + 4); - if (total != len) return false; + if (rd_u32(buf + 4) != len) return false; // MOTA_TOTAL_SIZE must equal the actual length - const uint8_t* p = buf + 8; // start of manifest - const uint8_t* end = buf + len - 5; // start of trailer - out.manifest_start = p; - // helper bounds check - #define NEED(n) do { if ((uint32_t)(end - p) < (uint32_t)(n)) return false; } while (0) - - NEED(3 + 16 + 1 + 4 + 32 + 1 + 32); // fixed head incl. hw_id[32] - out.format_ver = p[0]; - if (out.format_ver != MOTA_FORMAT_VER) return false; - out.flags = p[1]; - out.hash_algo = p[2]; - out.target_id = rd_u32(p + 3); - out.fw_version = rd_u32(p + 7); - out.image_size = rd_u32(p + 11); - out.payload_size = rd_u32(p + 15); - out.block_size_log2 = p[19]; - out.merkle_root = p + 20; - out.image_hash = p + 24; - out.codec_id = p[56]; - out.hw_id = p + 57; // 32-byte NUL-padded hardware tag (signed) - p += 89; - - if (out.block_size_log2 == 0 || out.block_size_log2 > 24) return false; - uint32_t bs = out.block_size(); - out.block_count = (out.payload_size + bs - 1) / bs; - if (out.payload_size == 0 || out.block_count == 0) return false; - - if (!out.is_full()) { NEED(8); out.base_hash = p; p += 8; } - - if (out.is_signed()) { - NEED(32); out.signer_pubkey = p; p += 32; - out.signed_len = (uint32_t)(p - (buf + 8)); // signature covers everything up to here - NEED(64); out.signature = p; p += 64; - } else { - out.signed_len = (uint32_t)(p - (buf + 8)); - } - - NEED(4); out.approval = p; p += 4; - - uint32_t leaves_bytes = out.block_count * 4; - NEED(leaves_bytes); out.leaves = p; p += leaves_bytes; - - NEED(out.payload_size); out.payload = p; p += out.payload_size; - - // payload must end exactly at the trailer - if (p != end) return false; - #undef NEED - return true; + ByteReader r(buf, len - 5); // everything up to (not incl.) the trailer + r.skip(4 + 4); // MAGIC + MOTA_TOTAL_SIZE (already validated) + out.manifest_start = buf + 8; + if (!parse_manifest_fields(r, 8, out)) return false; + out.leaves = r.take(out.block_count * 4); + out.payload = r.take(out.payload_size); + if (!r.ok) return false; + return r.pos() == len - 5; // payload must end exactly at the trailer } bool mota_parse_manifest(const uint8_t* mf, uint32_t len, MotaManifest& out) { out = MotaManifest(); - const uint8_t* p = mf; - const uint8_t* end = mf + len; - #define NEEDM(n) do { if ((uint32_t)(end - p) < (uint32_t)(n)) return false; } while (0) - - NEEDM(89); // fixed head incl. hw_id[32] out.manifest_start = mf; - out.format_ver = p[0]; - if (out.format_ver != MOTA_FORMAT_VER) return false; - out.flags = p[1]; - out.hash_algo = p[2]; - out.target_id = rd_u32(p + 3); - out.fw_version = rd_u32(p + 7); - out.image_size = rd_u32(p + 11); - out.payload_size = rd_u32(p + 15); - out.block_size_log2 = p[19]; - out.merkle_root = p + 20; - out.image_hash = p + 24; - out.codec_id = p[56]; - out.hw_id = p + 57; // 32-byte NUL-padded hardware tag (signed) - p += 89; - if (!out.is_full()) { NEEDM(8); out.base_hash = p; p += 8; } - if (out.is_signed()) { - NEEDM(32); out.signer_pubkey = p; p += 32; - out.signed_len = (uint32_t)(p - mf); - NEEDM(64); out.signature = p; p += 64; - } else { - out.signed_len = (uint32_t)(p - mf); - } - NEEDM(4); out.approval = p; p += 4; - if (out.block_size_log2 == 0 || out.block_size_log2 > 24 || out.payload_size == 0) return false; - out.block_count = (out.payload_size + out.block_size() - 1) / out.block_size(); - #undef NEEDM - return true; + ByteReader r(mf, len); // a standalone manifest = container bytes [8, leaves) + return parse_manifest_fields(r, 0, out); } bool mota_check_root(const MotaManifest& m) { diff --git a/src/helpers/ota/OtaByteIO.h b/src/helpers/ota/OtaByteIO.h new file mode 100644 index 00000000..5d201ada --- /dev/null +++ b/src/helpers/ota/OtaByteIO.h @@ -0,0 +1,42 @@ +#pragma once + +#include +#include + +// A tiny bounds-checked little-endian cursor for reading the `.mota` container (docs/ota_protocol.md §3-§4) +// in a self-documenting way: each field is read by name in order, instead of hand-computed byte offsets +// (`p[0]`, `rd_u32(p+3)`, `p += 89`, `NEED(n)` ...). Any over-read flips `ok` false and yields zero/null, so +// callers parse the whole struct then check `r.ok` once. 32-bit offsets (a container can be >64 KB; the +// 16-byte LoRa wire messages keep their own uint16 cursor in OtaProtocol.cpp). No allocation; `take()` +// returns a pointer INTO the caller's buffer (zero-copy), matching the manifest's by-pointer fields. + +namespace mesh { +namespace ota { + +struct ByteReader { + const uint8_t* p; + uint32_t len; + uint32_t n = 0; + bool ok = true; + + ByteReader(const uint8_t* buf, uint32_t length) : p(buf), len(length) {} + + uint32_t pos() const { return n; } + bool fits(uint32_t k) const { return ok && (uint64_t)n + k <= len; } + + uint8_t u8() { if (!fits(1)) { ok = false; return 0; } return p[n++]; } + uint32_t u32() { // little-endian + if (!fits(4)) { ok = false; return 0; } + uint32_t v = (uint32_t)p[n] | ((uint32_t)p[n+1] << 8) | ((uint32_t)p[n+2] << 16) | ((uint32_t)p[n+3] << 24); + n += 4; return v; + } + // Borrow `k` bytes at the cursor (e.g. merkle_root[4], leaves[4*BC]) and advance; null on overflow. + const uint8_t* take(uint32_t k) { + if (!fits(k)) { ok = false; return nullptr; } + const uint8_t* r = p + n; n += k; return r; + } + void skip(uint32_t k) { if (!fits(k)) { ok = false; return; } n += k; } +}; + +} // namespace ota +} // namespace mesh diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index f38c0b7d..7793ef91 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -16,7 +16,8 @@ static void wr_u32(uint8_t* p, uint32_t v) { p[0]=v; p[1]=v>>8; p[2]=v>>16; p[3] 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; + _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] = 0xFFFFFFFFu; // 0xFFFFFFFF = empty (never a real block) } // ---------------- serve (multi-mota registry) ---------------- @@ -58,8 +59,8 @@ void OtaManager::registerSelfEntry() { 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.is_self = true; e.src = nullptr; e.src_idx = 0; + 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; } @@ -85,10 +86,18 @@ void OtaManager::refresh_sources() { 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.is_self = false; e.src = src; e.src_idx = i; e.desc = d; + 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 } @@ -118,20 +127,28 @@ OtaManager::ServeView* OtaManager::resolve(const uint8_t* mid) { // 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 (!e.src || d.leaves_off < 8) return false; + 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; - if (!e.src->read(e.src_idx, 8, _src_manifest, mfl)) return false; - if (!mota_parse_manifest(_src_manifest, mfl, _srcv.m)) 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; - if (!e.src->read(e.src_idx, d.leaves_off, _src_leaves, d.block_count * 4)) 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.src; _srcv_rdctx.idx = e.src_idx; _srcv_rdctx.payload_off = d.payload_off; + _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); @@ -139,12 +156,42 @@ bool OtaManager::loadSource(const ServeEntry& e) { return true; } -// ServeReadFn trampoline: payload-relative offset -> absolute source read. +// 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 { @@ -190,6 +237,8 @@ void OtaManager::handleQuery(const uint8_t* m, uint16_t n) { memcpy(row, e.mid, 4); wr_u32(row + 4, e.target_id); wr_u32(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 @@ -227,30 +276,66 @@ void OtaManager::handleGetManifest(const uint8_t* m, uint16_t n) { } } +// 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) return; - 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; - 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; } - // send the block's data as self-describing fragments (frag_off); the proof is fetched separately - 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, v->m.merkle_root, 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); + 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); } } } @@ -338,6 +423,7 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) { const uint8_t* mid = row; uint32_t target = rd_u32(row + 4), fwver = rd_u32(row + 8); uint8_t codec = row[12], flags = row[13]; + uint32_t have_count = (uint32_t)row[14] | ((uint32_t)row[15] << 8); // 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; } @@ -353,6 +439,7 @@ void OtaManager::handleHave(const uint8_t* m, uint16_t n) { } 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); } } @@ -376,6 +463,7 @@ void OtaManager::startFetch(const uint8_t* mid, uint32_t target) { if (!_fetch || _fstate == FETCHING || _fstate == WANT_MANIFEST) return; if (resumeStaged(mid)) return; // resume a partial container left in flash memcpy(_fid, mid, 4); + _rng = (rd_u32(_seeder_id) ^ rd_u32(_fid)) | 1u; // per-node block-pick/jitter sequence (distinct per node) _fstate = WANT_MANIFEST; _mf_total = 0; _mf_mask = 0; _mf_len = 0; // fresh manifest reassembly GetManifestMsg gm; memcpy(gm.manifest_id, _fid, 4); @@ -420,6 +508,7 @@ void OtaManager::handleManifest(const uint8_t* m, uint16_t n) { // 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_u32(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; } @@ -436,8 +525,12 @@ void OtaManager::handleManifest(const uint8_t* m, uint16_t n) { // fresh transfer: clear any per-block reassembly state from a prior session _reasm_block = 0xFFFFFFFFu; _reasm_mask = 0; _reasm_need = 0; _awaiting_proof = false; _loop_last_have = 0; _loop_last_mask = 0; + // Swarm: hold the first REQ a random fraction of OTA_REQ_SPREAD_MS so N nodes that just discovered the + // same mid don't all burst-request block 0 in lockstep. loop() fires the first REQ once the hold elapses. + if (_rng == 0) _rng = (rd_u32(_seeder_id) ^ rd_u32(_fid)) | 1u; + _req_hold_at = _now_ms + (rngNext() % OTA_REQ_SPREAD_MS); + _peer_req_block = 0xFFFFFFFFu; _peer_req_at = 0; OTA_DBG("OTA: FETCHING bc=%u bs=%u total=%u\n", (unsigned)bc, (unsigned)bs, (unsigned)total); - requestMissing(); } bool OtaManager::resumeStaged(const uint8_t* want_mid) { @@ -479,7 +572,7 @@ bool OtaManager::resumeStaged(const uint8_t* want_mid) { } else { _fstate = COMPLETE; } - if (_fstate == COMPLETE) _fetch->finalize(); + if (_fstate == COMPLETE) { _fetch->finalize(); serveFetched(); } return true; } _fstate = FETCHING; // resume fetching the holes @@ -551,7 +644,7 @@ void OtaManager::handleProof(const uint8_t* m, uint16_t n) { } else { _fstate = COMPLETE; // per-block proofs already guaranteed integrity vs the root } - if (_fstate == COMPLETE) _fetch->finalize(); // commit the staged container to persistent storage + if (_fstate == COMPLETE) { _fetch->finalize(); serveFetched(); } // commit + re-seed (epidemic spread) OTA_DBG("OTA: transfer %s\n", _fstate == COMPLETE ? "COMPLETE" : "FAILED(root)"); } @@ -570,8 +663,7 @@ void OtaManager::requestMissing() { // 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 = 0; - while (start < _fbc && blockPresent(start)) start++; + uint32_t start = pickMissingBlock(); if (start >= _fbc) return; _req_start = start; _req_count = 1; ReqMsg rq; memcpy(rq.manifest_id, _fid, 4); @@ -582,6 +674,51 @@ void OtaManager::requestMissing() { 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() const { + 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 = ((OtaManager*)this)->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) { @@ -596,6 +733,7 @@ void OtaManager::loop() { 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(); @@ -612,8 +750,8 @@ void OtaManager::on_message(const uint8_t* msg, uint16_t len) { 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: handleReq(msg, len); break; - case OTA_DATA: handleData(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; diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index 7a77e2cc..43a9569e 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -63,6 +63,15 @@ typedef bool (*ServeReadFn)(void* ctx, uint32_t off, uint8_t* buf, uint32_t len) #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 @@ -97,13 +106,16 @@ public: 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 entries) + MotaDesc desc; // cached region offsets (source / fetch entries) }; - // Context for the source-payload reader trampoline (maps a payload-relative offset to a source read). - struct SrcReadCtx { MotaSource* src; uint8_t idx; uint32_t payload_off; }; + // 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); @@ -212,6 +224,7 @@ public: uint8_t codec, flags; uint8_t seeder0[4]; // first source that advertised this mid uint8_t n_seeders; // distinct sources advertising it (saturates) — "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; } @@ -231,11 +244,19 @@ private: 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) + uint32_t pickMissingBlock() const; // 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; } @@ -257,6 +278,8 @@ private: 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 @@ -270,6 +293,15 @@ private: 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 = 0xFFFFFFFFu; // 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; diff --git a/src/helpers/ota/OtaProtocol.h b/src/helpers/ota/OtaProtocol.h index 5f2fc96c..d8385353 100644 --- a/src/helpers/ota/OtaProtocol.h +++ b/src/helpers/ota/OtaProtocol.h @@ -34,7 +34,8 @@ struct QueryMsg { // ---- 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; }; +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) @@ -42,7 +43,7 @@ struct HaveMsg { 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 = 14; // mid4 + target4 + fwver4 + codec1 + flags1 +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]; }; diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index 5b77ec40..32e7899b 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -325,8 +325,8 @@ TEST(OtaProtocol, CodecRoundTrips) { EXPECT_EQ(q2.filter_target, 0x11223344u); // OTA_HAVE: a 2-row catalog (mid, target, fwver, codec, flags per row) tagged with the offering digest - uint8_t rows[2 * 14]; - for (int i = 0; i < 2 * 14; i++) rows[i] = (uint8_t)(i + 1); + uint8_t rows[2 * OTA_HAVE_ROW_BYTES]; + for (int i = 0; i < 2 * OTA_HAVE_ROW_BYTES; i++) rows[i] = (uint8_t)(i + 1); HaveMsg hv{{0x29,0x17,0xe4,0xf7}, {0xd1,0xd2,0xd3,0xd4}, 0, 1, 2, rows}; n = encode_have(buf, sizeof(buf), hv); ASSERT_GT(n, 0); EXPECT_EQ(ota_msg_type(buf, n), OTA_HAVE); @@ -334,7 +334,7 @@ TEST(OtaProtocol, CodecRoundTrips) { EXPECT_EQ(0, memcmp(h2.seeder_id, hv.seeder_id, 4)); EXPECT_EQ(0, memcmp(h2.set_digest, hv.set_digest, 4)); EXPECT_EQ(h2.frag_total, 1); EXPECT_EQ(h2.n_rows, 2); - EXPECT_EQ(0, memcmp(h2.rows, rows, 2 * 14)); + EXPECT_EQ(0, memcmp(h2.rows, rows, 2 * OTA_HAVE_ROW_BYTES)); GetManifestMsg gm{{1,2,3,4}}; n = encode_get_manifest(buf, sizeof(buf), gm); @@ -555,6 +555,72 @@ TEST(OtaFolder, ServesSelfPlusFolderAndFetchesExternal) { EXPECT_TRUE(mota_check_image_hash_full(got)); } +// --- swarm: a completed peer re-seeds beyond the origin (epidemic spread) --------------------- +namespace { +// A topology-aware bus: each node delivers only to its listed neighbours (so we can build multi-hop chains +// the origin can't reach directly). DATA emits are counted per node to observe who actually sources blocks. +struct TopoNode { int idx; }; +static std::vector g_tn; +static std::vector> g_adj; +static std::vector g_tdata; +static std::vector>> g_tq; +static size_t g_th = 0; +static uint32_t g_tclk = 0; +static void topo_send(void* ctx, const uint8_t* msg, uint16_t len, bool) { + int from = ((TopoNode*)ctx)->idx; + if (len && msg[0] == OTA_DATA) g_tdata[from]++; + for (int nb : g_adj[from]) g_tq.push_back({nb, std::vector(msg, msg + len)}); +} +static void topo_pump(int guard = 2000000) { + int idle = 0, g = 0; + while (g++ < guard) { + if (g_th < g_tq.size()) { + auto m = g_tq[g_th++]; + g_tn[m.first]->on_message(m.second.data(), (uint16_t)m.second.size()); + idle = 0; + if (g_th > 8192) { g_tq.erase(g_tq.begin(), g_tq.begin() + g_th); g_th = 0; } + } else { + g_tclk += 1000; + for (auto* nd : g_tn) { nd->set_clock(g_tclk); nd->loop(); } + if (g_th < g_tq.size()) { idle = 0; continue; } + if (++idle >= 3) break; + } + } +} +} + +// Line topology: origin <-> relay <-> leaf, with origin and leaf NOT connected. The relay fetches from the +// origin, COMPLETEs, and re-seeds; the leaf — which can ONLY hear the relay — must then obtain the whole +// firmware from the relay. If the leaf completes, the load provably spread off the origin (the origin never +// served the leaf). Validates re-serve-after-complete (and the partial-re-serve serve path it shares). +TEST(OtaSwarm, CompletedPeerReSeedsBeyondOrigin) { + g_tn.clear(); g_adj.clear(); g_tdata.clear(); g_tq.clear(); g_th = 0; g_tclk = 0; + static OtaManager origin, relay, leaf; + static OtaStoreRam<4096> rstore, lstore; + g_tn = {&origin, &relay, &leaf}; + g_adj = {{1}, {0, 2}, {1}}; // origin<->relay<->leaf + g_tdata = {0, 0, 0}; + static TopoNode t0{0}, t1{1}, t2{2}; + uint8_t id0[4] = {1,1,1,1}, id1[4] = {2,2,2,2}, id2[4] = {3,3,3,3}; + origin.begin(0, topo_send, &t0); origin.set_seeder_id(id0); + relay.begin(SIM_TARGET_ID, topo_send, &t1); relay.set_seeder_id(id1); + relay.set_fetch_store(&rstore); relay.set_autofetch(OtaManager::AUTOFETCH_ANY); + leaf.begin(SIM_TARGET_ID, topo_send, &t2); leaf.set_seeder_id(id2); + leaf.set_fetch_store(&lstore); leaf.set_autofetch(OtaManager::AUTOFETCH_ANY); + + ASSERT_TRUE(origin.serve(SIM_MOTA, SIM_MOTA_LEN)); + origin.announce(); + topo_pump(); + ASSERT_EQ(relay.fetchState(), OtaManager::COMPLETE); // relay sourced it from the origin + + relay.announce(); // relay now beacons its catalog (incl. the re-seeded mota) + topo_pump(); + EXPECT_EQ(leaf.fetchState(), OtaManager::COMPLETE); // leaf got it ONLY via the relay (re-serve) + ASSERT_EQ(lstore.staged_size(), SIM_MOTA_LEN); + EXPECT_EQ(0, std::memcmp(lstore.data(), SIM_MOTA, SIM_MOTA_LEN)); // byte-exact through the relay + EXPECT_GT(g_tdata[1], 0); // the relay actually served DATA (re-seeded) +} + // Fetch-resume across a reboot: a client commits some blocks, "reboots" (a fresh OtaManager on the SAME // persisted store), and resumeStaged() re-adopts the partial container and finishes the remaining blocks — // without re-fetching the manifest or the blocks already present. @@ -657,6 +723,7 @@ static uint16_t make_have1(uint8_t* buf, uint16_t cap, const uint8_t mid[4], row[4]=target; row[5]=target>>8; row[6]=target>>16; row[7]=target>>24; row[8]=fwver; row[9]=fwver>>8; row[10]=fwver>>16; row[11]=fwver>>24; row[12]=codec; row[13]=flags; + row[14]=0; row[15]=0; // have_count (unused in this 1-row discovery test) HaveMsg hv{{0xAA,0xBB,0xCC,0xDD}, {0,0,0,0}, 0, 1, 1, row}; return encode_have(buf, cap, hv); } diff --git a/tools/mota/dev_motas.py b/tools/mota/dev_motas.py new file mode 100644 index 00000000..b3e2e098 --- /dev/null +++ b/tools/mota/dev_motas.py @@ -0,0 +1,92 @@ +#!/usr/bin/env python3 +"""Package a full + same-image-delta .mota for ONE built firmware (dev rolling release). + +Extracts the OTA image (BODY||EndF) from a PlatformIO build and runs `mota build` twice: + - full .mota (codec full) — the flashable image, universally applicable + - delta .mota (same-image) — base == target, so the patch is tiny (a format/transport demo). + ESP32 -> sequential+crle ; nRF52 -> in-place (RAK4631 flash layout). + +Image source per platform: + ESP32 : /firmware.bin (pio_endf has already appended EndF) + nRF52 : app region of /firmware.hex, extracted via intelhex (EndF already appended to the .hex) + +Usage: + dev_motas.py --env RAK_4631_repeater --platform NRF52 --build-dir .pio/build/RAK_4631_repeater \ + --target-env RAK_4631_repeater --out-prefix out/RAK_4631_repeater-dev-abc1234 +Writes .full.mota and (best-effort) .delta.mota. +""" +import argparse, os, subprocess, sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +MOTA = os.path.join(HERE, "mota.py") + +# nRF52 (RAK4631) flash layout — keep in sync with src/helpers/ota/OtaFlashLayout_nrf52.h +NRF52_APP_BASE = 0x26000 +NRF52_FS_START = 0xD4000 +NRF52_INPLACE_MEMORY = NRF52_FS_START - NRF52_APP_BASE # 0xAE000 working size for in-place apply +NRF52_INPLACE_SEGMENT = 4096 + + +def extract_image(platform, build_dir, work): + """Return the path to the OTA image (BODY||EndF) as a raw .bin, or None if not found.""" + if platform == "ESP32": + bin_path = os.path.join(build_dir, "firmware.bin") + return bin_path if os.path.isfile(bin_path) else None + if platform == "NRF52": + hex_path = os.path.join(build_dir, "firmware.hex") + if not os.path.isfile(hex_path): + return None + from intelhex import IntelHex + ih = IntelHex(hex_path) + # the app .hex is a single contiguous segment [APP_BASE .. app_end]; take it verbatim + start = ih.minaddr() + data = ih.tobinarray(start=start, end=ih.maxaddr()) + out = os.path.join(work, "app_image.bin") + with open(out, "wb") as f: + f.write(bytes(data)) + return out + return None + + +def run_mota(args): + print(" +", "mota.py", " ".join(args)) + subprocess.run([sys.executable, MOTA, "build", *args], check=True) + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--env", required=True) + ap.add_argument("--platform", required=True, choices=["ESP32", "NRF52"]) + ap.add_argument("--build-dir", required=True) + ap.add_argument("--target-env", required=True) + ap.add_argument("--out-prefix", required=True) + ap.add_argument("--work", default=".") + a = ap.parse_args() + + img = extract_image(a.platform, a.build_dir, a.work) + if not img: + print(f"::warning::no OTA image for {a.env} ({a.platform}) — skipping .mota") + return 0 + print(f"OTA image for {a.env}: {img} ({os.path.getsize(img)} bytes)") + + # full .mota (always) + run_mota(["--fw", img, "--target-env", a.target_env, "--fw-version", "0.0.0", + "--codec", "full", "--out", a.out_prefix + ".full.mota"]) + + # same-image delta .mota (best-effort; codec per platform's applier) + try: + if a.platform == "ESP32": + run_mota(["--fw", img, "--base", img, "--target-env", a.target_env, "--fw-version", "0.0.1", + "--codec", "sequential", "--compression", "crle", "--out", a.out_prefix + ".delta.mota"]) + else: # NRF52 in-place + run_mota(["--fw", img, "--base", img, "--target-env", a.target_env, "--fw-version", "0.0.1", + "--codec", "inplace", "--compression", "crle", + "--inplace-memory", str(NRF52_INPLACE_MEMORY), "--inplace-segment", str(NRF52_INPLACE_SEGMENT), + "--out", a.out_prefix + ".delta.mota"]) + except subprocess.CalledProcessError as e: + print(f"::warning::delta .mota for {a.env} failed ({e}); full .mota still produced") + return 0 + + +if __name__ == "__main__": + sys.exit(main())