Speed up authenticated modern NimBLE Companion transfers

src/helpers/esp32/SerialBLEInterface.cpp

@@ -538,11 +538,20 @@ uint32_t SerialBLEInterface::onPassKeyDisplay() {
 }
 
 void SerialBLEInterface::onAuthenticationComplete(NimBLEConnInfo& info) {
+  if (pServer == NULL || pServer->getConnectedCount() == 0
+      || info.getConnHandle() != last_conn_id) return;
   if (info.isEncrypted() && info.isAuthenticated() && info.isBonded()) {
+    const bool first_authentication = !deviceConnected;
     deviceConnected = true;
     mesh::companion::BluetoothPeerIdentity peer;
     bluetoothPeerFromNimbleAddress(*info.getIdAddress().getBase(), peer);
     noteSuccessfulConnection(peer);
+    if (first_authentication) {
+      // Ask once per authenticated link for 15-30 ms connection events. The
+      // central chooses the actual interval; never renegotiate on every send.
+      // Keep peripheral latency zero so active sync does not skip events.
+      pServer->updateConnParams(info.getConnHandle(), 12, 24, 0, 400);
+    }
   } else {
     BLEDevice::deleteBond(info.getIdAddress());
     deviceConnected = false;
@@ -985,9 +994,12 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) {
 #error "MESH_HIL_BLE_INTERVAL_MS must be in 7..60 milliseconds"
 #endif
 #define BLE_WRITE_MIN_INTERVAL MESH_HIL_BLE_INTERVAL_MS
+#elif defined(MESH_USE_NIMBLE_ARDUINO)
+// Modern NimBLE returns queue acceptance synchronously, so failed sends retain
+// their frame and accepted sends cannot be duplicated by a late status callback.
+#define BLE_WRITE_MIN_INTERVAL 7
 #else
-// Preserve the shipped pacing until a separate lab image has measured a safe
-// replacement. One value governs admission and both successful/failed sends.
+// Preserve legacy NimBLE/Bluedroid pacing, whose status callbacks are async.
 #define BLE_WRITE_MIN_INTERVAL 60
 #endif
 

test/fixtures/nimble_companion/mocks/NimBLEDevice.h

@@ -62,6 +62,10 @@ inline bool bonding = false, mitm = false, sc = false;
 inline std::vector<std::vector<uint8_t>> transmitted;
 inline unsigned disconnects = 0;
 inline unsigned notification_attempts = 0;
+struct ConnectionUpdate {
+  uint16_t handle, minimum, maximum, latency, timeout;
+};
+inline std::vector<ConnectionUpdate> connection_updates;
 }
 inline int ble_gap_wl_set(const ble_addr_t* peers, uint8_t count) {
   assert(count == 1);
@@ -125,6 +129,10 @@ struct NimBLEServer {
   bool start() { return true; }
   uint8_t getConnectedCount() const { return connected; }
   uint16_t getPeerMTU(uint16_t) const { return fake_nimble::mtu; }
+  void updateConnParams(uint16_t handle, uint16_t minimum, uint16_t maximum,
+                        uint16_t latency, uint16_t timeout) const {
+    fake_nimble::connection_updates.push_back({handle, minimum, maximum, latency, timeout});
+  }
   bool disconnect(uint16_t) const { ++fake_nimble::disconnects; return true; }
   NimBLEAdvertising* getAdvertising() { return &advertising; }
   NimBLEService* createService(const char*) {

test/fixtures/nimble_companion/test_nimble_companion.cpp

@@ -10,6 +10,7 @@ using namespace fake_nimble;
 static void reset() {
   NimBLEDevice::deinit(true);
   failure.clear(); calls.clear(); transmitted.clear();
+  connection_updates.clear();
   persisted = true; accept_notification = true; disconnects = 0; notification_attempts = 0;
   test_millis = 100;
 }
@@ -111,6 +112,7 @@ static void test_authentication_and_bond_lifecycle() {
     authenticate(interface, info);
     assert(!interface.isConnected() && !interface.takeSuccessfulConnection());
     assert(disconnects == 1);
+    assert(connection_updates.empty());
     NimBLEDevice::deinit(true);
   }
 
@@ -149,6 +151,47 @@ static void test_authentication_and_bond_lifecycle() {
   NimBLEDevice::deinit(true);
 }
 
+static void test_connection_parameters_only_on_current_authenticated_transition() {
+  reset(); SerialBLEInterface interface;
+  assert(interface.begin("MeshCore-", "connection", 123456));
+  interface.enable();
+  auto* server = NimBLEDevice::server.get();
+  NimBLEConnInfo info;
+  info.handle = 29;
+  // A callback after the physical link has closed must not request an update.
+  server->callbacks->onAuthenticationComplete(info);
+  assert(connection_updates.empty() && !interface.isConnected());
+  server->connected = true;
+  server->callbacks->onConnect(server, info);
+  assert(connection_updates.empty() && !interface.isConnected());
+  NimBLEConnInfo stale = info;
+  stale.handle = 7;
+  server->callbacks->onAuthenticationComplete(stale);
+  assert(connection_updates.empty() && !interface.isConnected());
+  server->callbacks->onAuthenticationComplete(info);
+  assert(connection_updates.size() == 1 && interface.isConnected());
+  const auto& first = connection_updates[0];
+  assert(first.handle == 29 && first.minimum == 12 && first.maximum == 24
+         && first.latency == 0 && first.timeout == 400);
+  server->callbacks->onAuthenticationComplete(info);
+  assert(connection_updates.size() == 1); // No repeated update on the same link.
+  server->connected = false;
+  server->callbacks->onDisconnect(server, info, 0);
+  server->callbacks->onAuthenticationComplete(info);
+  assert(connection_updates.size() == 1 && !interface.isConnected());
+  NimBLEConnInfo next = info;
+  next.handle = 31;
+  server->connected = true;
+  server->callbacks->onConnect(server, next);
+  server->callbacks->onAuthenticationComplete(info); // Prior connection's callback.
+  assert(connection_updates.size() == 1 && !interface.isConnected());
+  server->callbacks->onAuthenticationComplete(next);
+  assert(connection_updates.size() == 2 && connection_updates[1].handle == 31);
+  server->callbacks->onAuthenticationComplete(next);
+  assert(connection_updates.size() == 2);
+  NimBLEDevice::deinit(true);
+}
+
 static void test_real_transport_frames() {
   reset(); SerialBLEInterface interface;
   assert(interface.begin("MeshCore-", "trial", 123456));
@@ -297,7 +340,7 @@ static void test_exact_notify_interval_failure_retention_and_rollover() {
 #if defined(MESH_HIL_BLE_INTERVAL_MS)
   constexpr uint32_t interval = MESH_HIL_BLE_INTERVAL_MS;
 #else
-  constexpr uint32_t interval = 60;
+  constexpr uint32_t interval = 7;
 #endif
   reset(); SerialBLEInterface interface;
   assert(interface.begin("MeshCore-", "interval", 123456));
@@ -332,7 +375,7 @@ static void test_exact_notify_interval_failure_retention_and_rollover() {
   assert(notification_attempts == 3 && transmitted.size() == 2);
   assert(transmitted[1] == std::vector<uint8_t>(second, second + sizeof(second)));
   // Real adapter arithmetic must remain correct when millis wraps at UINT32.
-  test_millis = UINT32_MAX - 7;
+  test_millis = UINT32_MAX - (interval / 2);
   assert(interface.writeFrame(first, sizeof(first)) == sizeof(first));
   interface.checkRecvFrame(received);
   assert(notification_attempts == 4);
@@ -356,14 +399,106 @@ static void test_exact_notify_interval_failure_retention_and_rollover() {
   NimBLEDevice::deinit(true);
 }
 
+static void test_128_contact_stream_under_queue_and_controller_pressure() {
+#if defined(MESH_HIL_BLE_INTERVAL_MS)
+  constexpr uint32_t interval = MESH_HIL_BLE_INTERVAL_MS;
+#else
+  constexpr uint32_t interval = 7;
+#endif
+  reset(); SerialBLEInterface interface;
+  assert(interface.begin("MeshCore-", "contacts", 123456));
+  interface.enable();
+  NimBLEConnInfo info; authenticate(interface, info);
+  auto* server = NimBLEDevice::server.get();
+  auto* tx = server->service->characteristics[0].get();
+  std::vector<std::vector<uint8_t>> expected;
+  for (unsigned index = 0; index < 128; ++index) {
+    std::vector<uint8_t> frame(148);
+    frame[0] = 3; // Contact response: required delivery, never evicted.
+    for (size_t byte = 1; byte < frame.size(); ++byte)
+      frame[byte] = static_cast<uint8_t>(index * 17 + byte * 3);
+    frame[1] = index;
+    expected.push_back(frame);
+  }
+  unsigned queued = 0, full_queue_rejections = 0;
+  auto fill = [&]() {
+    while (queued < expected.size()) {
+      const auto& frame = expected[queued];
+      const size_t accepted = interface.writeFrame(frame.data(), frame.size());
+      if (accepted == 0) { ++full_queue_rejections; break; }
+      assert(accepted == frame.size());
+      ++queued;
+    }
+  };
+  fill();
+  assert(queued == 4 && full_queue_rejections == 1 && interface.isWriteBusy());
+  uint8_t received[MAX_FRAME_SIZE] = {};
+  auto poll = [&]() {
+    test_millis += interval;
+    assert(interface.checkRecvFrame(received) == 0);
+    assert(transmitted.size() <= queued);
+    for (size_t index = 0; index < transmitted.size(); ++index)
+      assert(transmitted[index] == expected[index]);
+  };
+  poll();
+  assert(notification_attempts == 0 && interface.hasPendingIO()); // No CCCD yet.
+  tx->callbacks->onSubscribe(tx, info, 1);
+  mtu = 23;
+  server->callbacks->onMTUChange(247, info); // Stale event cannot override actual MTU.
+  poll();
+  assert(notification_attempts == 0 && interface.hasPendingIO());
+  mtu = 247;
+  server->callbacks->onMTUChange(23, info); // The live peer MTU permits full records.
+  bool injected_failures[128] = {};
+  bool cccd_paused = false, mtu_paused = false;
+  for (unsigned step = 0; step < 512 && transmitted.size() < expected.size(); ++step) {
+    const auto delivered = transmitted.size();
+    if (delivered == 32 && !cccd_paused) {
+      cccd_paused = true;
+      tx->callbacks->onSubscribe(tx, info, 0);
+      const auto attempts = notification_attempts;
+      poll(); poll();
+      assert(notification_attempts == attempts && transmitted.size() == delivered);
+      tx->callbacks->onSubscribe(tx, info, 1);
+    }
+    if (delivered == 64 && !mtu_paused) {
+      mtu_paused = true;
+      mtu = 23;
+      server->callbacks->onMTUChange(247, info);
+      const auto attempts = notification_attempts;
+      poll(); poll();
+      assert(notification_attempts == attempts && transmitted.size() == delivered);
+      mtu = 247;
+      server->callbacks->onMTUChange(23, info);
+    }
+    const bool fail_this_attempt = (delivered == 0 || delivered == 17
+        || delivered == 63 || delivered == 127) && !injected_failures[delivered];
+    accept_notification = !fail_this_attempt;
+    if (fail_this_attempt) injected_failures[delivered] = true;
+    const auto attempts = notification_attempts;
+    poll();
+    assert(notification_attempts == attempts + 1);
+    assert(transmitted.size() == delivered + (fail_this_attempt ? 0 : 1));
+    fill(); // Repeatedly hit all four slots without losing the next record.
+  }
+  assert(queued == 128 && transmitted == expected && cccd_paused && mtu_paused);
+  assert(notification_attempts == 132 && full_queue_rejections > 100);
+  assert(disconnects == 0 && !interface.hasPendingIO());
+  poll(); poll();
+  assert(transmitted == expected && notification_attempts == 132); // No duplicates.
+  NimBLEDevice::deinit(true);
+}
+
 int main() {
   test_identity_modes();
   test_pairing_wake_for_random_and_saved_pins();
   test_startup_failures();
   test_authentication_and_bond_lifecycle();
+  test_connection_parameters_only_on_current_authenticated_transition();
   test_real_transport_frames();
   test_required_reply_backpressure_services_tx_before_rx();
   test_full_queue_best_effort_eviction_still_admits_rx();
   test_exact_notify_interval_failure_retention_and_rollover();
+  test_128_contact_stream_under_queue_and_controller_pressure();
   puts("NimBLE adapter: identity, faults, bonding, stealth and full frames PASS");
 }

test/test_nimble_companion.py

@@ -19,7 +19,13 @@ class NimbleCompanionTests(unittest.TestCase):
     def test_lab15_interval_real_adapter_under_sanitizers(self):
         self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=15")
 
-    def test_fast_interval_is_lab_only_and_bounded(self):
+    def test_lab7_interval_real_adapter_under_sanitizers(self):
+        self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=7")
+
+    def test_lab60_interval_real_adapter_under_sanitizers(self):
+        self.run_adapter("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=60")
+
+    def test_interval_override_requires_lab_and_is_bounded(self):
         invalid = (
             ("-DMESH_HIL_BLE_INTERVAL_MS=15",),
             ("-DMESH_HIL_BLE_BENCH=0", "-DMESH_HIL_BLE_INTERVAL_MS=15"),
@@ -38,6 +44,35 @@ class NimbleCompanionTests(unittest.TestCase):
                 self.assertNotEqual(result.returncode, 0)
                 self.assertIn("MESH_HIL_BLE_INTERVAL_MS", result.stderr)
 
+    def test_actual_preprocessor_policy_modern_legacy_bluedroid_and_overrides(self):
+        source = (ROOT / "src/helpers/esp32/SerialBLEInterface.cpp").read_text()
+        start = source.index("#if defined(MESH_HIL_BLE_INTERVAL_MS)")
+        section = source[start:source.index("\nbool SerialBLEInterface::isReadBusy()", start)]
+        transports = (("-DMESH_USE_NIMBLE_ARDUINO=1", 7),
+                      ("-DCONFIG_NIMBLE_ENABLED=1", 60),
+                      ("-DCONFIG_BLUEDROID_ENABLED=1", 60))
+        for transport, default in transports:
+            cases = [((), default)] + [(("-DMESH_HIL_BLE_BENCH=1",
+                f"-DMESH_HIL_BLE_INTERVAL_MS={interval}"), interval) for interval in (7, 15, 60)]
+            for flags, expected in cases:
+                with self.subTest(transport=transport, flags=flags):
+                    result = subprocess.run(["c++", "-std=c++17", "-x", "c++", "-fsyntax-only",
+                        transport, *flags, "-"], input=section +
+                        f'\nstatic_assert(BLE_WRITE_MIN_INTERVAL == {expected}, "wrong transport pacing");\n',
+                        text=True, capture_output=True, timeout=30)
+                    self.assertEqual(result.returncode, 0, result.stderr)
+            for flags in (("-DMESH_HIL_BLE_INTERVAL_MS=15",),
+                          ("-DMESH_HIL_BLE_BENCH=0", "-DMESH_HIL_BLE_INTERVAL_MS=15"),
+                          ("-DMESH_HIL_BLE_BENCH=2", "-DMESH_HIL_BLE_INTERVAL_MS=15"),
+                          ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=6"),
+                          ("-DMESH_HIL_BLE_BENCH=1", "-DMESH_HIL_BLE_INTERVAL_MS=61")):
+                with self.subTest(transport=transport, invalid=flags):
+                    result = subprocess.run(["c++", "-std=c++17", "-x", "c++", "-fsyntax-only",
+                        transport, *flags, "-"], input=section, text=True,
+                        capture_output=True, timeout=30)
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertIn("MESH_HIL_BLE_INTERVAL_MS", result.stderr)
+
     def run_adapter(self, *flags):
         with tempfile.TemporaryDirectory(prefix="meshcore-nimble-") as temp:
             binary = Path(temp) / "nimble-companion"
@@ -50,9 +85,9 @@ class NimbleCompanionTests(unittest.TestCase):
                 str(ROOT / "src/helpers/esp32/SerialBLEInterface.cpp"),
                 str(ROOT / "test/fixtures/nimble_companion/test_nimble_companion.cpp"),
                 "-o", str(binary),
-            ], text=True, capture_output=True)
+            ], text=True, capture_output=True, timeout=60)
             self.assertEqual(result.returncode, 0, result.stderr)
-            result = subprocess.run([str(binary)], text=True, capture_output=True)
+            result = subprocess.run([str(binary)], text=True, capture_output=True, timeout=30)
             self.assertEqual(result.returncode, 0, result.stderr)