Wait for nRF52 BLE client data before sending queued frames

src/helpers/nrf52/SerialBLEInterface.cpp

@@ -1033,6 +1033,10 @@ size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) {
 
   const uint16_t mtu = conn->getMtu();
   if (mtu <= 3) return 0;
+#if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE
+  // Record only public framing metadata, never characteristic payload data.
+  meshBleTrace(0xf036, handle, frame.buf[0] | (frame.len << 8), mtu);
+#endif
 
   // BLEUart::write() reports either the full requested length or zero, even
   // when its internal multi-notification loop queued an earlier fragment
@@ -1293,6 +1297,14 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) {
     return 0;
   }
 
+  // Restored bonded subscriptions can be enabled while the client is still
+  // discovering services and negotiating its MTU. An unsolicited push at the
+  // default MTU can exhaust the notification slots partway through a frame.
+  // The independent UART RX callback (or explicit mOTA activation) establishes
+  // application readiness. Keep queued frames and their TX watchdog idle until
+  // then; the secured-link no-data timer still bounds an abandoned connection.
+  if (!_companionDataSeen.load(std::memory_order_acquire)) return 0;
+
   if (send_queue_len > 0) {
     if (!isConnected()) {
       BLE_DEBUG_PRINTLN("writeBytes: connection invalid, clearing send queue");

test/fixtures/nrf52_ble_tx_session.cpp

@@ -63,6 +63,8 @@ static struct {
   unsigned connected() const { assert(!critical_depth); return 1; }
 } Bluefruit;
 
+struct SerialBLEInterface;
+static SerialBLEInterface* instance;
 struct SerialBLEInterface {
   struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; };
   using RxQueueGuard = Guard;
@@ -71,6 +73,7 @@ struct SerialBLEInterface {
   uint8_t send_queue_len = 0;
   Frame send_queue[FRAME_QUEUE_SIZE];
   std::atomic<bool> _tx_reset_pending{false};
+  std::atomic<bool> _companionDataSeen{true};
   uint32_t _send_queue_generation = 0;
   uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID;
   uint32_t _tx_recovery_generation = 0;
@@ -81,14 +84,26 @@ struct SerialBLEInterface {
   mesh::BleDisconnectRecovery _tx_disconnect_recovery;
   SecuritySessionTimer _companion_start_timer;
   SecuritySessionTimer _security_timer;
+  struct MotaStream {
+    bool active = false;
+    void setActive(bool value) { active = value; }
+  } _mota_stream;
   bool _peer_address_valid = true;
   struct Uart {
     struct Chunk { uint16_t handle; std::vector<uint8_t> bytes; };
     std::vector<Chunk> chunks;
+    std::vector<uint8_t> rx_bytes;
     std::function<void()> after_write;
     unsigned attempts = 0, fail_on_attempt = 0;
     bool notifyEnabled(uint16_t) const { assert(!critical_depth); return true; }
-    void flush() { assert(!critical_depth); }
+    void flush() { assert(!critical_depth); rx_bytes.clear(); }
+    int available() const { assert(!critical_depth); return rx_bytes.size(); }
+    int read(uint8_t* destination, size_t length) {
+      assert(!critical_depth && length <= rx_bytes.size());
+      std::memcpy(destination, rx_bytes.data(), length);
+      rx_bytes.erase(rx_bytes.begin(), rx_bytes.begin() + length);
+      return length;
+    }
     size_t write(uint16_t handle, const uint8_t* data, size_t length) {
       assert(!critical_depth);
       if (++attempts == fail_on_attempt) return 0;
@@ -101,7 +116,7 @@ struct SerialBLEInterface {
       return length;
     }
   } bleuart;
-  SerialBLEInterface() { open(1); }
+  SerialBLEInterface() { instance = this; open(1); }
   void open(uint16_t handle) {
     _conn_handle = handle;
     recv_queue.setEnabled(true);
@@ -127,6 +142,8 @@ struct SerialBLEInterface {
   bool hasRequiredReplyCapacity() const;
   size_t writeFrame(const uint8_t*, size_t);
   size_t checkRecvFrame(uint8_t*);
+  static void onBleUartRX(uint16_t);
+  void setMotaStreamActive(bool);
   void recoverStalledTx(const char*, uint16_t = BLE_CONN_HANDLE_INVALID, uint32_t = 0);
   bool isConnected() const;
   bool isWriteBusy() const;
@@ -134,6 +151,71 @@ struct SerialBLEInterface {
 };
 @METHODS@
 
+static void application_start() {
+  for (bool required : {false, true}) {
+    SerialBLEInterface port;
+    const size_t requests = disconnect_requests.size();
+    now_ms = 100;
+    Bluefruit.connection.mtu = 23;
+    port._companionDataSeen.store(false);
+    port._companion_start_timer.start(now_ms);
+    uint8_t frame[75] = {uint8_t(required ? 0x87 : 0x88)};
+    uint8_t received[MAX_FRAME_SIZE] = {};
+    port.bleuart.fail_on_attempt = 4; // Three default-MTU slots accept 60 bytes.
+    assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame));
+    for (uint32_t elapsed : {0U, 250U, 10001U}) {
+      now_ms = 100 + elapsed;
+      assert(port.checkRecvFrame(received) == 0);
+      assert(port.bleuart.chunks.empty() && !port.bleuart.attempts);
+      assert(port.send_queue_len == 1 && !port._last_retry_attempt);
+      assert(!port._tx_stall_watchdog.active() && !port._tx_disconnect_recovery.pending());
+      assert(port._companion_start_timer.pending());
+      assert(disconnect_requests.size() == requests);
+    }
+    // Even a terminal's first newline opens the gate through the real RX
+    // callback, independently of checkRecvFrame's application-task polling.
+    Bluefruit.connection.mtu = 247;
+    port.bleuart.rx_bytes = {'\n'};
+    SerialBLEInterface::onBleUartRX(1);
+    assert(port._companionDataSeen && !port._companion_start_timer.pending());
+    assert(port.checkRecvFrame(received) == 1 && received[0] == '\n');
+    assert(port.send_queue_len == 0 && port.bleuart.chunks.size() == 1);
+    assert(port.bleuart.chunks[0].bytes == std::vector<uint8_t>(frame, frame + sizeof(frame)));
+    assert(disconnect_requests.size() == requests);
+  }
+  // A Full Companion's explicit mOTA activation establishes readiness without
+  // requiring a Nordic UART command, retaining the separate mOTA transport.
+  SerialBLEInterface port;
+  port._companionDataSeen.store(false);
+  port._companion_start_timer.start(now_ms);
+  const uint8_t frame[] = {0x83};
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame));
+  assert(port.checkRecvFrame(received) == 0 && port.bleuart.chunks.empty());
+  port.setMotaStreamActive(true);
+  assert(port._mota_stream.active && port._companionDataSeen);
+  assert(!port._companion_start_timer.pending());
+  assert(port.checkRecvFrame(received) == 0 && port.send_queue_len == 0);
+  assert(port.bleuart.chunks.size() == 1);
+  Bluefruit.connection.mtu = 23;
+}
+
+static void previous_prestart_partial() {
+  SerialBLEInterface port;
+  const size_t requests = disconnect_requests.size();
+  Bluefruit.connection.mtu = 23;
+  port._companionDataSeen.store(false);
+  uint8_t frame[75] = {0x88}, received[MAX_FRAME_SIZE] = {};
+  port.bleuart.fail_on_attempt = 4;
+  assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame));
+  assert(port.checkRecvFrame(received) == 0);
+  assert(port.bleuart.chunks.size() == 3);
+  size_t accepted = 0;
+  for (const auto& chunk : port.bleuart.chunks) accepted += chunk.bytes.size();
+  assert(accepted == 60 && port._tx_disconnect_recovery.pending());
+  assert(disconnect_requests.size() == requests + 1);
+}
+
 static void enqueue_reset() {
   SerialBLEInterface port;
   const uint8_t old_reply[] = {1, 11}, interrupted[] = {2, 22}, fresh[] = {3, 33};
@@ -255,6 +337,8 @@ int main(int argc, char** argv) {
   assert(argc == 2);
   if (!std::strcmp(argv[1], "enqueue")) enqueue_reset();
   else if (!std::strcmp(argv[1], "connection")) connection_changes();
+  else if (!std::strcmp(argv[1], "application-start")) application_start();
+  else if (!std::strcmp(argv[1], "prestart-partial")) previous_prestart_partial();
   else {
     enqueue_reset(); callback_ownership(); connection_changes(); old_queue_and_recovery();
     recovery_connection_changes();

test/test_nrf52_ble_backpressure.py

@@ -42,6 +42,7 @@ struct SerialBLEInterface {
   Frame send_queue[FRAME_QUEUE_SIZE];
   uint16_t _conn_handle = 1;
   std::atomic<bool> _tx_reset_pending{false};
+  std::atomic<bool> _companionDataSeen{true};
   uint32_t _send_queue_generation = 0;
   uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID;
   uint32_t _tx_recovery_generation = 0;

test/test_nrf52_ble_tx_session.py

@@ -10,6 +10,7 @@ import unittest
 from cpp_source import body
 
 ROOT = Path(__file__).resolve().parents[1]
+START_GATE = "  if (!_companionDataSeen.load(std::memory_order_acquire)) return 0;\n"
 LEGACY_CLEAR = r'''
 void SerialBLEInterface::clearBuffers() {
   send_queue_len = 0;
@@ -53,14 +54,21 @@ class Nrf52BleTxSessionTests(unittest.TestCase):
             "bool SerialBLEInterface::servicePendingTxReset()", "void SerialBLEInterface::shiftSendQueueLeft()",
             "size_t SerialBLEInterface::writeBleUartFrame(", "bool SerialBLEInterface::hasRequiredReplyCapacity() const",
             "size_t SerialBLEInterface::writeFrame(", "size_t SerialBLEInterface::checkRecvFrame(",
+            "void SerialBLEInterface::onBleUartRX(",
             "void SerialBLEInterface::recoverStalledTx(", "void SerialBLEInterface::serviceTxRecovery(",
             "bool SerialBLEInterface::isConnected() const",
             "bool SerialBLEInterface::isWriteBusy() const", "bool SerialBLEInterface::hasPendingIO() const")
         methods = "\n".join(body(source, signature) for signature in signatures)
+        header = (ROOT / "src/helpers/nrf52/SerialBLEInterface.h").read_text()
+        methods += "\n" + body(header, "void setMotaStreamActive(bool active)").replace(
+            "void setMotaStreamActive(", "void SerialBLEInterface::setMotaStreamActive(", 1)
+        if methods.count(START_GATE) != 1:
+            raise AssertionError("Expected exactly one production application-start gate")
         variants = {"production": methods,
                     "legacy_enqueue": methods.replace(body(source, signatures[0]), LEGACY_CLEAR).replace(
                         body(source, "size_t SerialBLEInterface::writeFrame("), LEGACY_ENQUEUE),
-                    "legacy_write": methods.replace(body(source, "size_t SerialBLEInterface::writeBleUartFrame("), LEGACY_WRITE)}
+                    "legacy_write": methods.replace(body(source, "size_t SerialBLEInterface::writeBleUartFrame("), LEGACY_WRITE),
+                    "legacy_start": methods.replace(START_GATE, "", 1)}
         fixture = (ROOT / "test/fixtures/nrf52_ble_tx_session.cpp").read_text(encoding="ascii")
         cls.work = tempfile.TemporaryDirectory(prefix="meshcore-ble-tx-session-")
         cls.binaries = {}
@@ -99,6 +107,17 @@ class Nrf52BleTxSessionTests(unittest.TestCase):
                 self.assertNotEqual(result.returncode, 0)
                 self.assertIn(assertion, result.stderr)
 
+    def test_early_bonded_notifications_wait_for_uart_or_mota_application_start(self):
+        result = self.run_case("production", "application-start")
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+    def test_previous_start_path_reproduces_partial_sixty_byte_notification(self):
+        result = self.run_case("legacy_start", "prestart-partial")
+        self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+        result = self.run_case("legacy_start", "application-start")
+        self.assertNotEqual(result.returncode, 0)
+        self.assertIn("bleuart.chunks.empty()", result.stderr)
+
 
 if __name__ == "__main__":
     unittest.main()