Announce ready OTA catalogs through the existing data connection

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;
   }

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

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);
 }

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)");

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();
 }
 

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

test/fixtures/ota_folder_ready.cpp

@@ -0,0 +1,233 @@
+#include <helpers/ota/OtaManager.h>
+#include <helpers/ota/MotaSourceSerial.h>
+#include <helpers/ota/MotaSeederProto.h>
+#include <helpers/ota/OtaByteIO.h>
+#include <helpers/ota/OtaTiming.h>
+#include <algorithm>
+#include <cassert>
+#include <cstdio>
+#include <vector>
+
+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<uint8_t> response;
+  size_t position = 0;
+public:
+  using Stream::write;
+  std::vector<uint32_t> mids = {0x12345678};
+  std::vector<uint8_t> 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<uint8_t> payload;
+    bool ok = true;
+    if (op == MS_OP_COUNT) {
+      assert(length == 4);
+      ok = !fail_count;
+      payload.push_back(static_cast<uint8_t>(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<int>(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<std::vector<uint8_t>> frames;
+  std::vector<uint32_t> times;
+  uint32_t now = 0;
+};
+static bool capture(void* pointer, const uint8_t* data, uint16_t length, bool) {
+  auto& sent = *static_cast<Sent*>(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<uint8_t>{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");
+}

test/test_ota_announce_cadence.py

@@ -31,6 +31,8 @@ struct Manager {
   unsigned cleared = 0;
   std::vector<uint32_t> 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; }
 };

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()