From 631c670ba7e5bac19ef30c26babaa8b6bdd81ca5 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Wed, 21 Oct 2026 00:06:57 -0700 Subject: [PATCH] Announce ready OTA catalogs through the existing data connection --- examples/companion_radio/main.cpp | 2 +- src/Mesh.cpp | 9 +- src/helpers/esp32/WiFiOtaSeeder.cpp | 2 +- src/helpers/ota/OtaCli.cpp | 2 +- src/helpers/ota/OtaManager.cpp | 20 ++- src/helpers/ota/OtaManager.h | 8 + test/fixtures/ota_folder_ready.cpp | 233 ++++++++++++++++++++++++++++ test/test_ota_announce_cadence.py | 2 + test/test_ota_folder_ready.py | 62 ++++++++ 9 files changed, 334 insertions(+), 6 deletions(-) create mode 100644 test/fixtures/ota_folder_ready.cpp create mode 100644 test/test_ota_folder_ready.py diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index 1672fe9a0..eaa85937a 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -399,7 +399,7 @@ public: "; cleared %u oldest queued messages for OTA", (unsigned)drained); } - context.manager.announce(); + // The mesh loop announces the fully enumerated source with a bounded burst. mesh::usbLoggingPort().println("Bluetooth mOTA source attached"); return true; } diff --git a/src/Mesh.cpp b/src/Mesh.cpp index e9ae3af66..af5220707 100644 --- a/src/Mesh.cpp +++ b/src/Mesh.cpp @@ -537,6 +537,12 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { _ota_announce_count = 0; } const float ota_speed = getOtaSpeedFactor(); + if (ota::ota_ctx().manager.takeSourceReadyAnnouncement()) { + // COUNT/DESCRIBE completed over the existing host data session. Reuse the + // bounded discovery burst even when host preparation outlasted boot. + _ota_announce_count = 0; + _ota_announce_timer.arm(_ms->getMillis(), 0, 1.0f); + } if (_ota_timer_speed != ota_speed) { _ota_announce_timer.rescale(_ms->getMillis(), _ota_timer_speed, ota_speed); _ota_timer_speed = ota_speed; @@ -602,7 +608,8 @@ void __attribute__((noinline)) Mesh::serviceLoopMaintenance() { #if !defined(OTA_SEEDER_ONLY) if (oc.self_serve_supported && !oc.serving) oc.serving = ota::ota_serve_self(oc, 0); #endif - oc.manager.announce(); + if (oc.manager.sourceCatalogComplete() && oc.manager.servedCount()) + oc.manager.announce(); if (_ota_announce_count < 250) _ota_announce_count++; } // Re-arm: tight spacing during the boot burst; afterwards the fixed cadence (default 24h). When periodic diff --git a/src/helpers/esp32/WiFiOtaSeeder.cpp b/src/helpers/esp32/WiFiOtaSeeder.cpp index c4b18eef6..ee930ab45 100644 --- a/src/helpers/esp32/WiFiOtaSeeder.cpp +++ b/src/helpers/esp32/WiFiOtaSeeder.cpp @@ -133,7 +133,7 @@ void WiFiOtaSeeder::loop() { snprintf(link_info, sizeof(link_info), "tcp %s", seeder_client.remoteIP().toString().c_str()); context.set_folder_dest(&folder_store, link_info); - context.manager.announce(); + // The mesh loop announces the fully enumerated source with a bounded burst. mesh::usbLoggingPort().printf( "OTA seeder client connected (%s)\n", link_info); } diff --git a/src/helpers/ota/OtaCli.cpp b/src/helpers/ota/OtaCli.cpp index 81a2292cb..99c479745 100644 --- a/src/helpers/ota/OtaCli.cpp +++ b/src/helpers/ota/OtaCli.cpp @@ -1126,7 +1126,7 @@ static __attribute__((noinline)) bool handle_control_command( #if !defined(OTA_SEEDER_ONLY) if (c.self_serve_supported && !c.serving) c.serving = ota_serve_self(c, 0); #endif - char m2[120]; c.attach_folder(m2, sizeof(m2)); c.manager.announce(); + char m2[120]; c.attach_folder(m2, sizeof(m2)); strncpy(reply, m2, 159); reply[159] = 0; #else strcpy(reply, "ERR not built with OTA_FOLDER_SERIAL (set the seeder UART in platformio.ini)"); diff --git a/src/helpers/ota/OtaManager.cpp b/src/helpers/ota/OtaManager.cpp index 517a7d41a..e853d2691 100644 --- a/src/helpers/ota/OtaManager.cpp +++ b/src/helpers/ota/OtaManager.cpp @@ -181,6 +181,7 @@ void OtaManager::begin(uint32_t my_target_id, OtaSend send, void* ctx) { _resume_verify_idx = 0; _resume_invalidated = false; _resume_merkle.reset(); _n_serve = 0; _n_src_obj = 0; _view0.valid = false; _srcv.valid = false; + _sources_complete = true; _source_announce_pending = false; memset(_src_offered, 0, sizeof(_src_offered)); memset(_src_advertised, 0, sizeof(_src_advertised)); _n_src = 0; _n_cat = 0; @@ -297,6 +298,10 @@ bool OtaManager::remove_source(MotaSource* src) { } void OtaManager::refresh_sources() { + uint8_t previous_digest[4]; + setDigest(previous_digest); + const bool was_complete = _sources_complete; + bool complete = true; clearPendingEgress(); releaseColdBuffers(); _n_serve = 0; @@ -311,7 +316,7 @@ void OtaManager::refresh_sources() { for (uint8_t i = 0; i < cnt; i++) { if (_n_serve >= OTA_MAX_SERVE) break; MotaDesc d; - if (!src->describe(i, d)) continue; + if (!src->describe(i, d)) { complete = false; continue; } if (d.block_count == 0 || (uint64_t)d.block_count * 4 > OTA_PROOFGEN_SCRATCH) continue; // A legacy serial descriptor left the geometry byte zero. Inspect its fixed manifest once so it is // either proven servable before advertisement or excluded from the registry. @@ -320,7 +325,7 @@ void OtaManager::refresh_sources() { MotaManifest parsed; if (!src->read(i, 8, mf, sizeof(mf)) || !mota_parse_manifest(mf, sizeof(mf), parsed) || memcmp(parsed.merkle_root, d.mid, 4) != 0 || parsed.block_count != d.block_count || - parsed.payload_size != d.payload_size) continue; + parsed.payload_size != d.payload_size) { complete = false; continue; } d.block_size_log2 = parsed.block_size_log2; } if (d.block_size_log2 >= 32 || (1UL << d.block_size_log2) > OTA_MAX_BLOCK) continue; @@ -337,6 +342,15 @@ void OtaManager::refresh_sources() { } } _srcv.valid = false; // a loaded source view may now be stale; reloads on demand + _sources_complete = complete; + uint8_t digest[4]; + setDigest(digest); + // Host compression/attachment can finish after the TempRadio boot burst. + // Keep one signal until the normal mesh loop can announce the ready set; + // unchanged rescans coalesce, and failed metadata never starts a burst. + if (!complete || !_n_serve) _source_announce_pending = false; + else if (!was_complete || memcmp(previous_digest, digest, 4) != 0) + _source_announce_pending = true; } bool OtaManager::sourceStats(const MotaSource* src, uint16_t& offered, uint16_t& advertised) const { @@ -360,6 +374,8 @@ void OtaManager::clear_sources() { memset(_src_offered, 0, sizeof(_src_offered)); memset(_src_advertised, 0, sizeof(_src_advertised)); _n_serve = 0; + _sources_complete = true; + _source_announce_pending = false; if (_view0.valid) registerSelfEntry(); } diff --git a/src/helpers/ota/OtaManager.h b/src/helpers/ota/OtaManager.h index 198775e13..d7a32c5f1 100644 --- a/src/helpers/ota/OtaManager.h +++ b/src/helpers/ota/OtaManager.h @@ -374,6 +374,12 @@ public: return _view0.valid && _view0.m.manifest_start == manifest; } uint8_t servedCount() const { return _n_serve; } // total mOTAs we offer (own fw + folder) + bool sourceCatalogComplete() const { return _sources_complete; } + bool takeSourceReadyAnnouncement() { + const bool ready = _source_announce_pending; + _source_announce_pending = false; + return ready; + } // Per-source enumeration accounting. `offered` is what the attached host/source reported; `advertised` // is what fit in this build's serve registry after validation/deduplication/capacity limits. bool sourceStats(const MotaSource* src, uint16_t& offered, uint16_t& advertised) const; @@ -853,6 +859,8 @@ private: bool _enforce_auto_version = false; // OtaContext enables this even when EndF/version is unknown uint16_t _checkpoint_blocks = OTA_CHECKPOINT_BLOCKS; // resume checkpoint cadence (persisted) uint16_t _advert_mins = OTA_ADVERT_INTERVAL_MINS; // beacon re-advertise cadence, minutes; 0=off (persisted) + bool _sources_complete = true; + bool _source_announce_pending = false; uint8_t _max_hops = OTA_HOP_LIMIT_DEFAULT; // OTA flood reach in hops; 0=direct only (persisted) uint8_t _fflags = 0; // flags of the manifest currently being fetched bool _wire_v2_enabled = true; // advertise v2 in OTA_REQ while retaining transparent fallback diff --git a/test/fixtures/ota_folder_ready.cpp b/test/fixtures/ota_folder_ready.cpp new file mode 100644 index 000000000..61c263f43 --- /dev/null +++ b/test/fixtures/ota_folder_ready.cpp @@ -0,0 +1,233 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +using namespace mesh::ota; +namespace ota = mesh::ota; + +// One ordinary data stream answers the real device COUNT/DESCRIBE protocol. +// No text CLI, second socket or explicit announcement command exists here. +class FolderHost : public Stream { + std::vector response; + size_t position = 0; +public: + using Stream::write; + std::vector mids = {0x12345678}; + std::vector operations; + int failed_descriptor = -1; + bool fail_count = false; + bool invalid_geometry = false; + unsigned flushes = 0; + + size_t write(const uint8_t* request, size_t length) override { + assert(length >= 4 && request[0] == 'M' && request[1] == 'S'); + uint8_t checksum = 0; + for (size_t i = 2; i < length; ++i) checksum ^= request[i]; + assert(checksum == 0); + const uint8_t op = request[2]; + operations.push_back(op); + std::vector payload; + bool ok = true; + if (op == MS_OP_COUNT) { + assert(length == 4); + ok = !fail_count; + payload.push_back(static_cast(mids.size())); + } else { + assert(op == MS_OP_DESCRIBE && length == 5); + const uint8_t index = request[3]; + ok = index < mids.size() && index != failed_descriptor; + if (ok) { + payload.assign(MOTA_DESC_WIRE, 0); + wr_u32le(payload.data(), mids[index]); + wr_u32le(payload.data() + 4, 1234); + wr_u32le(payload.data() + 8, 0x01120001); + payload[12] = CODEC_FULL; payload[13] = MFLAG_FULL; + wr_u32le(payload.data() + 14, 8 + MOTA_MFL + 4 + 128 + 5); + wr_u32le(payload.data() + 18, 8 + MOTA_MFL); + wr_u32le(payload.data() + 22, invalid_geometry ? 0 : 1); + wr_u32le(payload.data() + 26, 8 + MOTA_MFL + 4); + wr_u32le(payload.data() + 30, 128); + payload[34] = 7; + } + } + response = {'m', 's', op, uint8_t(ok ? MS_STATUS_OK : MS_STATUS_ERR)}; + if (ok) response.insert(response.end(), payload.begin(), payload.end()); + checksum = 0; + for (uint8_t value : response) checksum ^= value; + response.push_back(checksum); + position = 0; + return length; + } + int available() override { return static_cast(response.size() - position); } + int read() override { + if (position < response.size()) return response[position++]; + ++g_mock_millis; + return -1; + } + void flush() override { ++flushes; } +}; + +struct Sent { + std::vector> frames; + std::vector times; + uint32_t now = 0; +}; +static bool capture(void* pointer, const uint8_t* data, uint16_t length, bool) { + auto& sent = *static_cast(pointer); + sent.frames.emplace_back(data, data + length); + sent.times.push_back(sent.now); + return true; +} + +namespace mesh { namespace ota { +struct OtaContext { + OtaManager manager; + bool self_serve_supported = false, serving = false; +}; +static OtaContext* active_context = nullptr; +OtaContext& ota_ctx() { assert(active_context); return *active_context; } +bool ota_serve_self(OtaContext&, unsigned) { return false; } +}} + +// @PRODUCTION_DEFINES@ +struct Clock { uint32_t now; uint32_t getMillis() const { return now; } }; +struct Fixture { + ota::OtaContext context; + Sent sent; + Clock clock; + Clock* _ms = &clock; + uint8_t _ota_announce_count = OTA_ANNOUNCE_BURST; + ota::LongTimer _ota_announce_timer; + float speed; + bool active = true; + Fixture(uint32_t start, float rate) : clock{start}, speed(rate) { + ota::active_context = &context; + context.manager.begin(1234, capture, &sent); + const uint8_t sid[4] = {1, 2, 3, 4}; + context.manager.set_seeder_id(sid); + context.manager.set_advert_mins(0); + _ota_announce_timer.arm(start, OTA_ANNOUNCE_DISABLED_POLL_MS, speed); + } + void step() { + if (!active) return; // Same admission boundary as the production Mesh loop. + sent.now = clock.now; + context.manager.set_clock(clock.now); + const float ota_speed = speed; + // @PRODUCTION_READY@ + // @PRODUCTION_ANNOUNCE@ + } + void advance(uint32_t milliseconds) { clock.now += milliseconds; step(); } +}; + +static void four_ready_beacons(MotaStreamWritePolicy policy, uint32_t start, float speed) { + Fixture f(start, speed); + FolderHost host; + SerialMotaSource source(host, policy, 5); + // Host preparation can outlast the entire existing boot burst. + f.advance(120000); + assert(f.sent.frames.empty()); + assert(f.context.manager.add_source(&source)); + assert((host.operations == std::vector{MS_OP_COUNT, MS_OP_DESCRIBE})); + assert(f.sent.frames.empty()); // Nothing is emitted from the data callback. + f.step(); + assert(f.sent.frames.size() == 1 && "ready catalog must discover without a control channel"); + AdvMsg advert{}; + assert(decode_adv(f.sent.frames[0].data(), f.sent.frames[0].size(), advert)); + assert(advert.n_motas == 1); + const uint32_t gap = scaleDelay(OTA_ANNOUNCE_BURST_MS, speed); + for (unsigned count = 1; count < OTA_ANNOUNCE_BURST; ++count) { + f.advance(gap - 1); + assert(f.sent.frames.size() == count); + f.context.manager.refresh_sources(); // Unchanged polls must not restart it. + f.advance(1); + assert(f.sent.frames.size() == count + 1); + assert(f.sent.times.back() - f.sent.times[f.sent.times.size() - 2] == gap); + } + f.context.manager.refresh_sources(); + f.advance(gap); + f.advance(OTA_ANNOUNCE_DISABLED_POLL_MS); + assert(f.sent.frames.size() == OTA_ANNOUNCE_BURST); + assert(host.flushes == (policy == MotaStreamWritePolicy::FlushTransmit ? host.operations.size() : 0)); + + // A fresh receiver with no known source learns and retrieves the exact MID. + OtaManager receiver; + Sent requests; + receiver.begin(1234, capture, &requests); + receiver.queryAll(); + assert(requests.frames.empty()); + receiver.on_message(f.sent.frames[0].data(), f.sent.frames[0].size()); + receiver.queryAll(); + assert(requests.frames.size() == 1); + const size_t before = f.sent.frames.size(); + f.context.manager.on_message(requests.frames[0].data(), requests.frames[0].size()); + assert(f.sent.frames.size() == before + 1); + receiver.on_message(f.sent.frames.back().data(), f.sent.frames.back().size()); + assert(receiver.catalogCount() == 1 && rd_u32le(receiver.catalogRow(0)->mid) == host.mids[0]); +} + +static void readiness_and_coalescing() { + Fixture f(1000, 1); + FolderHost host; + host.mids.clear(); + SerialMotaSource source(host, MotaStreamWritePolicy::NoFlush, 5); + assert(f.context.manager.add_source(&source)); + f.step(); + assert(f.sent.frames.empty()); + host.mids = {17, 18}; host.failed_descriptor = 1; + f.context.manager.refresh_sources(); + assert(f.context.manager.servedCount() == 1 && !f.context.manager.sourceCatalogComplete()); + f.step(); + assert(f.sent.frames.empty()); + // A complete rescan may have exactly the same deduplicated MID set. + host.mids = {17, 17}; host.failed_descriptor = -1; + f.context.manager.refresh_sources(); + assert(f.context.manager.sourceCatalogComplete()); + f.step(); + assert(f.sent.frames.size() == 1); + // Multiple changes before the next mesh tick collapse onto the newest set. + host.mids = {19}; f.context.manager.refresh_sources(); + host.mids = {20, 21}; f.context.manager.refresh_sources(); + f.step(); + assert(f.sent.frames.size() == 2); + std::reverse(host.mids.begin(), host.mids.end()); + f.context.manager.refresh_sources(); f.advance(1); + assert(f.sent.frames.size() == 2); // Canonical digest ignores enumeration order. + // Metadata failure cancels an unsent ready signal and automatic burst sends. + host.mids = {22, 23}; f.context.manager.refresh_sources(); + host.failed_descriptor = 1; f.context.manager.refresh_sources(); + f.advance(OTA_ANNOUNCE_BURST_MS); + assert(f.sent.frames.size() == 2); + host.failed_descriptor = -1; f.context.manager.refresh_sources(); + f.step(); + assert(f.sent.frames.size() == 3); + // Empty or invalid offerings never announce an empty ready catalog. + host.mids.clear(); f.context.manager.refresh_sources(); + f.advance(OTA_ANNOUNCE_BURST_MS); + assert(f.sent.frames.size() == 3); + host.mids = {24}; host.invalid_geometry = true; f.context.manager.refresh_sources(); + f.advance(OTA_ANNOUNCE_BURST_MS); + assert(f.sent.frames.size() == 3); + host.invalid_geometry = false; f.context.manager.refresh_sources(); + f.active = false; f.advance(120000); + assert(f.sent.frames.size() == 3); // Normal-radio admission retains readiness. + f.active = true; f.step(); + assert(f.sent.frames.size() == 4); + // Removing the source before admission must retire its pending signal. + host.mids = {25}; f.context.manager.refresh_sources(); + f.context.manager.clear_sources(); f.step(); + assert(f.sent.frames.size() == 4 && !f.context.manager.takeSourceReadyAnnouncement()); +} + +int main() { + for (auto policy : {MotaStreamWritePolicy::FlushTransmit, MotaStreamWritePolicy::NoFlush}) + for (uint32_t start : {0U, 0xffff0000U}) + for (float speed : {0.05f, 1.0f, 3.0f}) four_ready_beacons(policy, start, speed); + readiness_and_coalescing(); + puts("PASS: single-data-session folder readiness, exact four-beacon bursts, cold peer discovery and coalescing"); +} diff --git a/test/test_ota_announce_cadence.py b/test/test_ota_announce_cadence.py index 31da50ab9..31faee7ca 100644 --- a/test/test_ota_announce_cadence.py +++ b/test/test_ota_announce_cadence.py @@ -31,6 +31,8 @@ struct Manager { unsigned cleared = 0; std::vector sent; uint32_t advert_mins() const { return minutes; } + bool sourceCatalogComplete() const { return true; } + unsigned servedCount() const { return 1; } void announce() { sent.push_back(clock->now); } void clearPendingEgress() { ++cleared; } }; diff --git a/test/test_ota_folder_ready.py b/test/test_ota_folder_ready.py new file mode 100644 index 000000000..eaac88266 --- /dev/null +++ b/test/test_ota_folder_ready.py @@ -0,0 +1,62 @@ +"""Exercise real folder framing, catalog readiness and production burst code.""" +from pathlib import Path +import os +import re +import shutil +import subprocess +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + +ROOT = Path(__file__).resolve().parents[1] + + +class OtaFolderReadyTest(unittest.TestCase): + def run_fixture(self, missing_ready_trigger=False): + production = (ROOT / "src/Mesh.cpp").read_text(encoding="ascii") + defines = [] + for name in ("OTA_ANNOUNCE_BURST", "OTA_ANNOUNCE_BURST_MS", "OTA_ANNOUNCE_DISABLED_POLL_MS"): + match = re.search(r"^#define " + name + r"\s+(\d+(?:UL)?)\b", production, re.M) + self.assertIsNotNone(match) + defines.append("#define " + name + " " + match[1]) + source = (ROOT / "test/fixtures/ota_folder_ready.cpp").read_text(encoding="ascii") + source = source.replace("// @PRODUCTION_DEFINES@", "\n".join(defines)) + ready = extract_braced(production, "if (ota::ota_ctx().manager.takeSourceReadyAnnouncement())") + source = source.replace("// @PRODUCTION_READY@", "" if missing_ready_trigger else ready) + source = source.replace("// @PRODUCTION_ANNOUNCE@", extract_braced( + production, "if (_ota_announce_timer.ready(_ms->getMillis()))")) + with tempfile.TemporaryDirectory(prefix="ota-folder-ready-") as directory: + work = Path(directory) + cpp, executable = work / "ready.cpp", work / "ready" + cpp.write_text(source, encoding="ascii") + sanitizers = [] if os.name == "nt" else ["-fsanitize=address,undefined", "-fno-pie", "-no-pie"] + compiler = os.environ.get("CXX") or shutil.which("c++") or "g++" + variants = (("NRF52_PLATFORM=1",),) if missing_ready_trigger else ( + ("ESP32_PLATFORM=1", "COMPANION_RADIO_FULL=1"), ("NRF52_PLATFORM=1",)) + for definitions in variants: + with self.subTest(definitions=definitions): + sources = ("OtaManager.cpp", "OtaProtocol.cpp", "MotaContainer.cpp", "MerkleTree.cpp", "MotaSourceSerial.cpp") + result = subprocess.run([compiler, "-std=c++17", "-O1", *sanitizers, + "-DENABLE_OTA=1", "-DOTA_SEEDER_ONLY=1", *["-D" + value for value in definitions], + "-I", str(ROOT / "src"), "-I", str(ROOT / "test/mocks"), str(cpp), + *[str(ROOT / "src/helpers/ota" / name) for name in sources], + str(ROOT / "src/Utils.cpp"), "-o", str(executable)], capture_output=True, text=True, timeout=90) + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + result = subprocess.run([str(executable)], capture_output=True, text=True, timeout=20) + if missing_ready_trigger: + self.assertNotEqual(result.returncode, 0, "missing readiness trigger escaped the oracle") + self.assertIn("ready catalog must discover without a control channel", result.stderr) + else: + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + self.assertIn("PASS: single-data-session folder readiness", result.stdout) + + def test_single_host_data_session_announces_only_ready_changed_catalogs(self): + self.run_fixture() + + def test_missing_production_ready_trigger_is_detected(self): + self.run_fixture(missing_ready_trigger=True) + + +if __name__ == "__main__": + unittest.main()