From 85a786e796820d27dae8a8af0dda7c2c18df2385 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Mon, 19 Oct 2026 13:12:54 -0700 Subject: [PATCH] Wait for nRF52 BLE client data before sending queued frames --- src/helpers/nrf52/SerialBLEInterface.cpp | 12 ++++ test/fixtures/nrf52_ble_tx_session.cpp | 88 +++++++++++++++++++++++- test/test_nrf52_ble_backpressure.py | 1 + test/test_nrf52_ble_tx_session.py | 21 +++++- 4 files changed, 119 insertions(+), 3 deletions(-) diff --git a/src/helpers/nrf52/SerialBLEInterface.cpp b/src/helpers/nrf52/SerialBLEInterface.cpp index b0a5a4599..244b5a34f 100644 --- a/src/helpers/nrf52/SerialBLEInterface.cpp +++ b/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"); diff --git a/test/fixtures/nrf52_ble_tx_session.cpp b/test/fixtures/nrf52_ble_tx_session.cpp index 23b9bd43e..946e57cd8 100644 --- a/test/fixtures/nrf52_ble_tx_session.cpp +++ b/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 _tx_reset_pending{false}; + std::atomic _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 bytes; }; std::vector chunks; + std::vector rx_bytes; std::function 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(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(); diff --git a/test/test_nrf52_ble_backpressure.py b/test/test_nrf52_ble_backpressure.py index 73eee1d31..f16321d8b 100644 --- a/test/test_nrf52_ble_backpressure.py +++ b/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 _tx_reset_pending{false}; + std::atomic _companionDataSeen{true}; uint32_t _send_queue_generation = 0; uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; uint32_t _tx_recovery_generation = 0; diff --git a/test/test_nrf52_ble_tx_session.py b/test/test_nrf52_ble_tx_session.py index 427dc8957..1a7f7ae6b 100644 --- a/test/test_nrf52_ble_tx_session.py +++ b/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()