From 9181e3e0ec55dbb1fe5d2544eebfe12e39193044 Mon Sep 17 00:00:00 2001 From: mikecarper Date: Fri, 16 Oct 2026 00:01:28 -0700 Subject: [PATCH] Keep BLE transmit queues and recovery bound to their connection session --- .github/workflows/run-unit-tests.yml | 1 + src/helpers/nrf52/BleRxFrameQueue.h | 5 + src/helpers/nrf52/SerialBLEInterface.cpp | 141 +++++++++--- src/helpers/nrf52/SerialBLEInterface.h | 12 +- test/fixtures/nrf52_ble_rx_queue.cpp | 7 + test/fixtures/nrf52_ble_tx_session.cpp | 263 ++++++++++++++++++++++ test/test_nrf52_ble_backpressure.py | 15 +- test/test_nrf52_ble_reconnect_watchdog.py | 1 + test/test_nrf52_ble_rx_queue.py | 2 + test/test_nrf52_ble_tx_session.py | 104 +++++++++ 10 files changed, 516 insertions(+), 35 deletions(-) create mode 100644 test/fixtures/nrf52_ble_tx_session.cpp create mode 100644 test/test_nrf52_ble_tx_session.py diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index ea812ec91..f343abed2 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -709,6 +709,7 @@ jobs: python3 -B test/test_nrf52_ble_startup.py python3 -B test/test_nrf52_ble_backpressure.py python3 -B test/test_nrf52_ble_rx_queue.py -v + python3 -B test/test_nrf52_ble_tx_session.py -v python3 -B test/test_nrf52_ble_trace.py python3 -B test/test_nrf52_ble_debug_tools.py python3 -B test/test_nrf52_usb_power.py diff --git a/src/helpers/nrf52/BleRxFrameQueue.h b/src/helpers/nrf52/BleRxFrameQueue.h index 000f7dc97..e5b6a63e2 100644 --- a/src/helpers/nrf52/BleRxFrameQueue.h +++ b/src/helpers/nrf52/BleRxFrameQueue.h @@ -115,6 +115,11 @@ public: Guard guard; return _count; } + + uint32_t generation() const { + Guard guard; + return _generation; + } }; } // namespace mesh_nrf52 diff --git a/src/helpers/nrf52/SerialBLEInterface.cpp b/src/helpers/nrf52/SerialBLEInterface.cpp index 368cd65fb..d67ee38c3 100644 --- a/src/helpers/nrf52/SerialBLEInterface.cpp +++ b/src/helpers/nrf52/SerialBLEInterface.cpp @@ -988,11 +988,23 @@ void SerialBLEInterface::cancelStealthPairingTransition() { } void SerialBLEInterface::clearBuffers() { + _tx_reset_pending.store(true, std::memory_order_release); + recv_queue.reset([this]() { bleuart.flush(); }); +} + +void SerialBLEInterface::resetTxState() { send_queue_len = 0; _last_retry_attempt = 0; _tx_stall_watchdog.reset(); _tx_disconnect_recovery.complete(); - recv_queue.reset([this]() { bleuart.flush(); }); + _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; + _tx_recovery_generation = 0; +} + +bool SerialBLEInterface::servicePendingTxReset() { + if (!_tx_reset_pending.exchange(false, std::memory_order_acq_rel)) return false; + resetTxState(); + return true; } void SerialBLEInterface::shiftSendQueueLeft() { @@ -1005,9 +1017,13 @@ void SerialBLEInterface::shiftSendQueueLeft() { } size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) { - BLEConnection* conn = Bluefruit.Connection(_conn_handle); + const uint16_t handle = _conn_handle; + uint32_t generation; + if (!recv_queue.snapshot(handle, generation) + || generation != _send_queue_generation) return 0; + BLEConnection* conn = Bluefruit.Connection(handle); if (conn == nullptr || !conn->connected() || - !bleuart.notifyEnabled(_conn_handle)) { + !bleuart.notifyEnabled(handle)) { return 0; } @@ -1021,37 +1037,53 @@ size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) { // already on the stream and must never be followed by a whole-frame retry. return mesh::writeBleFrameInChunks( frame.buf, frame.len, mtu - 3, - [this](const uint8_t* data, size_t len) { - return bleuart.write(_conn_handle, data, len); + [this, handle, generation](const uint8_t* data, size_t len) -> size_t { + // Never continue an old frame on a replacement connection, including + // when SoftDevice reuses the same numeric handle after a reconnect. + // FIFO/SDK calls remain outside the RX metadata critical section. + if (!recv_queue.current(handle, generation)) return 0; + const size_t written = bleuart.write(handle, data, len); + return recv_queue.current(handle, generation) ? written : 0; }); } void SerialBLEInterface::serviceTxRecovery(uint32_t now) { if (!_tx_disconnect_recovery.pending()) return; + if (servicePendingTxReset()) return; + const uint16_t handle = _tx_recovery_handle; + const uint32_t generation = _tx_recovery_generation; + if (handle == BLE_CONN_HANDLE_INVALID || _conn_handle != handle + || recv_queue.generation() != generation) { + resetTxState(); + return; + } - BLEConnection* conn = _conn_handle == BLE_CONN_HANDLE_INVALID - ? nullptr - : Bluefruit.Connection(_conn_handle); - if (_conn_handle == BLE_CONN_HANDLE_INVALID || conn == nullptr || - !conn->connected()) { + BLEConnection* conn = Bluefruit.Connection(handle); + if (conn == nullptr || !conn->connected()) { BLE_DEBUG_PRINTLN("SerialBLEInterface: stalled TX link is already closed"); - _conn_handle = BLE_CONN_HANDLE_INVALID; - _isDeviceConnected = false; - _peer_address_valid = false; - _security_timer.cancel(); - _companion_start_timer.cancel(); - clearBuffers(); - if (advertisingAllowed() && !isAdvertising()) { + { + RxQueueGuard guard; + if (_conn_handle != handle || recv_queue.generation() != generation) return; + _conn_handle = BLE_CONN_HANDLE_INVALID; + _isDeviceConnected = false; + _peer_address_valid = false; + _security_timer.cancel(); + _companion_start_timer.cancel(); + resetTxState(); + } + if (_conn_handle == BLE_CONN_HANDLE_INVALID && advertisingAllowed() && !isAdvertising()) { startAdvertising("advertising failed after TX recovery"); } return; } if (!_tx_disconnect_recovery.shouldAttempt(now)) return; + if (servicePendingTxReset() || _conn_handle != handle + || recv_queue.generation() != generation) return; const uint32_t result = sd_ble_gap_disconnect( - _conn_handle, BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION); - meshBleTrace(0xf013, _conn_handle, result, 1); + handle, BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION); + meshBleTrace(0xf013, handle, result, 1); if (result == NRF_SUCCESS) { BLE_DEBUG_PRINTLN("SerialBLEInterface: stalled TX disconnect requested"); } else if (result == NRF_ERROR_INVALID_STATE) { @@ -1063,8 +1095,14 @@ void SerialBLEInterface::serviceTxRecovery(uint32_t now) { } } -void SerialBLEInterface::recoverStalledTx(const char* cause) { - meshBleTrace(0xf020, _conn_handle, send_queue_len, recv_queue.size()); +void SerialBLEInterface::recoverStalledTx(const char* cause, + uint16_t expected_handle, + uint32_t expected_generation) { + if (expected_handle == BLE_CONN_HANDLE_INVALID) { + expected_handle = _conn_handle; + if (!recv_queue.snapshot(expected_handle, expected_generation)) return; + } + meshBleTrace(0xf020, expected_handle, send_queue_len, recv_queue.size()); if (_tx_disconnect_recovery.pending()) return; BLE_DEBUG_PRINTLN("SerialBLEInterface: %s; forcing reconnect", cause); @@ -1072,16 +1110,24 @@ void SerialBLEInterface::recoverStalledTx(const char* cause) { // Keep the physical connection state intact until the SoftDevice confirms // disconnection, but make isConnected() false through the recovery state so // no more companion frames enter this damaged stream. - send_queue_len = 0; - recv_queue.disconnect(_conn_handle); - _last_retry_attempt = 0; - _tx_stall_watchdog.reset(); - _companion_start_timer.cancel(); - recv_queue.reset([this]() { bleuart.flush(); }); + { + RxQueueGuard guard; + if (_tx_reset_pending.load(std::memory_order_acquire) + || !recv_queue.current(expected_handle, expected_generation)) return; + send_queue_len = 0; + _last_retry_attempt = 0; + _tx_stall_watchdog.reset(); + _companion_start_timer.cancel(); + recv_queue.disconnect(expected_handle); + _tx_recovery_handle = expected_handle; + _tx_recovery_generation = recv_queue.generation(); + _tx_disconnect_recovery.begin(); + } + // Invalidate RX immediately, but let normal connection cleanup flush its + // FIFO. An old recovery must not flush a replacement link after yielding. #if COMPANION_FEATURE_BLE_MOTA_SOURCE setMotaStreamActive(false); #endif - _tx_disconnect_recovery.begin(); serviceTxRecovery((uint32_t)millis()); } @@ -1110,6 +1156,7 @@ void SerialBLEInterface::enable() { _pairingRequestPending.store(false, std::memory_order_release); _isEnabled = true; clearBuffers(); + servicePendingTxReset(); _last_health_check = millis(); if (advertisingAllowed()) { @@ -1129,6 +1176,7 @@ void SerialBLEInterface::disconnect() { void SerialBLEInterface::disable() { recv_queue.setEnabled(false); + _tx_reset_pending.store(true, std::memory_order_release); _isEnabled = false; _pairingRequestPending.store(false, std::memory_order_release); BLE_DEBUG_PRINTLN("SerialBLEInterface: disable"); @@ -1152,6 +1200,7 @@ void SerialBLEInterface::loop() { _debug_last_loop_ms = now; #endif serviceBondedOnlyTransition(); + servicePendingTxReset(); if (_tx_disconnect_recovery.pending()) { serviceTxRecovery(now); return; @@ -1191,6 +1240,7 @@ void SerialBLEInterface::loop() { } bool SerialBLEInterface::hasRequiredReplyCapacity() const { + if (_tx_reset_pending.load(std::memory_order_acquire)) return true; if (send_queue_len < FRAME_QUEUE_SIZE) return true; // Required responses may evict an unsent packet log, never another required // reply. Match enqueueCompanionFrame without growing either bounded queue. @@ -1207,6 +1257,12 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { return 0; } + servicePendingTxReset(); + const uint16_t handle = _conn_handle; + uint32_t generation; + if (!recv_queue.snapshot(handle, generation)) return 0; + if (send_queue_len && _send_queue_generation != generation) resetTxState(); + _send_queue_generation = generation; bool connected = isConnected(); if (connected && len > 0) { if (!mesh::enqueueCompanionFrame(send_queue, send_queue_len, FRAME_QUEUE_SIZE, @@ -1215,6 +1271,10 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { BLE_DEBUG_PRINTLN("writeFrame(), send_queue is full!"); return 0; } + if (servicePendingTxReset() || !recv_queue.current(handle, generation)) { + resetTxState(); + return 0; + } return len; } return 0; @@ -1223,6 +1283,7 @@ size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { const uint32_t check_now = (uint32_t)millis(); serviceBondedOnlyTransition(); + servicePendingTxReset(); if (_tx_disconnect_recovery.pending()) { serviceTxRecovery(check_now); return 0; @@ -1239,11 +1300,20 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { bool throttle_active = (_last_retry_attempt > 0 && (now - _last_retry_attempt) < BLE_RETRY_THROTTLE_MS); if (!throttle_active) { + const uint16_t handle = _conn_handle; + const uint32_t generation = _send_queue_generation; Frame frame_to_send = send_queue[0]; const bool delivery_required = mesh::companionFrameRequiresDelivery( frame_to_send.buf, frame_to_send.len); size_t written = writeBleUartFrame(frame_to_send); + // A callback may reset or replace the link while the SDK waits for a + // notification slot. Discard only the old TX state; do not recover or + // dequeue against the replacement session using the old write result. + if (servicePendingTxReset() || !recv_queue.current(handle, generation)) { + resetTxState(); + return 0; + } meshBleTrace(0xf031, _conn_handle, frame_to_send.len, written); #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE if (written) mesh_ble_trace.tx.fetch_add(1); @@ -1261,7 +1331,7 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { // protocol frame. Reconnect instead of following it with another // frame on the same BLE UART stream. meshBleTrace(0xf025, _conn_handle, written, frame_to_send.len); - recoverStalledTx("partial BLE UART frame"); + recoverStalledTx("partial BLE UART frame", handle, generation); return 0; } else { if (!isConnected()) { @@ -1275,7 +1345,7 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) { if (delivery_required) { if (_tx_stall_watchdog.noteBlocked((uint32_t)now)) { meshBleTrace(0xf024, _conn_handle); - recoverStalledTx("command reply blocked for 10 seconds"); + recoverStalledTx("command reply blocked for 10 seconds", handle, generation); return 0; } } else { @@ -1362,7 +1432,10 @@ void SerialBLEInterface::onBleUartOverflow(uint16_t handle, uint16_t leftover) { #endif bool SerialBLEInterface::isConnected() const { - return !_tx_disconnect_recovery.pending() && _isDeviceConnected && + uint32_t generation; + return _isEnabled && (_tx_reset_pending.load(std::memory_order_acquire) + || !_tx_disconnect_recovery.pending()) && _isDeviceConnected && + recv_queue.snapshot(_conn_handle, generation) && Bluefruit.connected() > 0; } @@ -1390,9 +1463,11 @@ bool SerialBLEInterface::isReadBusy() const { } bool SerialBLEInterface::isWriteBusy() const { - return send_queue_len >= (FRAME_QUEUE_SIZE * 2 / 3); + return !_tx_reset_pending.load(std::memory_order_acquire) + && send_queue_len >= (FRAME_QUEUE_SIZE * 2 / 3); } bool SerialBLEInterface::hasPendingIO() const { - return recv_queue.size() > 0 || send_queue_len > 0; + return recv_queue.size() > 0 + || (!_tx_reset_pending.load(std::memory_order_acquire) && send_queue_len > 0); } diff --git a/src/helpers/nrf52/SerialBLEInterface.h b/src/helpers/nrf52/SerialBLEInterface.h index 59010f340..a2db384d8 100644 --- a/src/helpers/nrf52/SerialBLEInterface.h +++ b/src/helpers/nrf52/SerialBLEInterface.h @@ -105,6 +105,12 @@ class SerialBLEInterface : public BaseSerialInterface { uint8_t send_queue_len; Frame send_queue[FRAME_QUEUE_SIZE]; + // Only the application task mutates the TX queue and retry state. Deferred + // Bluefruit callbacks request a reset instead of interrupting queue copies. + std::atomic _tx_reset_pending{false}; + uint32_t _send_queue_generation = 0; + uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; + uint32_t _tx_recovery_generation = 0; struct RxQueueGuard { RxQueueGuard() { taskENTER_CRITICAL(); } @@ -115,10 +121,14 @@ class SerialBLEInterface : public BaseSerialInterface { RxQueue recv_queue; void clearBuffers(); + void resetTxState(); + bool servicePendingTxReset(); bool hasRequiredReplyCapacity() const; void shiftSendQueueLeft(); size_t writeBleUartFrame(const Frame& frame); - void recoverStalledTx(const char* cause); + void recoverStalledTx(const char* cause, + uint16_t expected_handle = BLE_CONN_HANDLE_INVALID, + uint32_t expected_generation = 0); void serviceTxRecovery(uint32_t now); bool removeStoredBondForPeer(const char* cause); void noteSuccessfulConnection(const ble_gap_addr_t& peer_address); diff --git a/test/fixtures/nrf52_ble_rx_queue.cpp b/test/fixtures/nrf52_ble_rx_queue.cpp index db91a1123..9efa87d8d 100644 --- a/test/fixtures/nrf52_ble_rx_queue.cpp +++ b/test/fixtures/nrf52_ble_rx_queue.cpp @@ -133,6 +133,9 @@ struct SerialBLEInterface { using RxQueue = mesh_nrf52::BleRxFrameQueue; RxQueue recv_queue; uint8_t send_queue_len = 0; + std::atomic _tx_reset_pending{false}; + uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; + uint32_t _tx_recovery_generation = 0; unsigned long _last_retry_attempt = 0; mesh::BleTxStallWatchdog _tx_stall_watchdog; mesh::BleDisconnectRecovery _tx_disconnect_recovery; @@ -201,6 +204,8 @@ struct SerialBLEInterface { ++bonded_recoveries; } void clearBuffers(); + void resetTxState(); + bool servicePendingTxReset(); static void onBLEEvent(ble_evt_t*); static void onConnect(uint16_t); static void onDisconnect(uint16_t, uint8_t); @@ -446,6 +451,8 @@ static void production_connection_events_preserve_disconnect_cleanup() { SerialBLEInterface::onDisconnect(7, reason); assert(link._conn_handle == BLE_CONN_HANDLE_INVALID && !link._isDeviceConnected); assert(!link._security_timer.pending() && !link._companion_start_timer.pending()); + assert(link._tx_reset_pending && link.send_queue_len == 3); + assert(link.servicePendingTxReset()); // Application task applies the TX reset. assert(link.send_queue_len == 0 && link.recv_queue.size() == 0); } diff --git a/test/fixtures/nrf52_ble_tx_session.cpp b/test/fixtures/nrf52_ble_tx_session.cpp new file mode 100644 index 000000000..23b9bd43e --- /dev/null +++ b/test/fixtures/nrf52_ble_tx_session.cpp @@ -0,0 +1,263 @@ +#include +#include +#include +#include +#include +#include +#include +#include +#include + +static unsigned critical_depth; +struct Guard { + Guard() { ++critical_depth; } + ~Guard() { --critical_depth; } +}; +static std::function during_copy; +static void* scheduled_copy(void* dest, const void* src, size_t length) { + void* result = std::memcpy(dest, src, length); + if (during_copy) { + auto operation = std::move(during_copy); + during_copy = nullptr; + operation(); + } + return result; +} +#define memcpy scheduled_copy +#include +#undef memcpy + +#define MAX_FRAME_SIZE 176 +#define FRAME_QUEUE_SIZE 12 +#define BLE_CONN_HANDLE_INVALID 0xffff +#define BLE_RETRY_THROTTLE_MS 250 +#define BLE_DEBUG_PRINTLN(...) assert(!critical_depth) +static std::function trace_hook; +static void trace_event(uint32_t event) { + assert(!critical_depth); + if (trace_hook) trace_hook(event); +} +#define meshBleTrace(event, ...) trace_event(event) +#define NRF_SUCCESS 0 +#define NRF_ERROR_INVALID_STATE 8 +#define BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION 0x13 +static std::vector disconnect_requests; +static uint32_t sd_ble_gap_disconnect(uint16_t handle, uint8_t) { + assert(!critical_depth); + disconnect_requests.push_back(handle); + return NRF_SUCCESS; +} +static uint32_t now_ms = 100; +static uint32_t millis() { return now_ms; } +struct BLEConnection { + uint16_t mtu = 23; + bool connected() const { assert(!critical_depth); return true; } + uint16_t getMtu() const { assert(!critical_depth); return mtu; } +}; +static struct { + BLEConnection connection; + BLEConnection* Connection(uint16_t handle) { + assert(!critical_depth); + return handle == BLE_CONN_HANDLE_INVALID ? nullptr : &connection; + } + unsigned connected() const { assert(!critical_depth); return 1; } +} Bluefruit; + +struct SerialBLEInterface { + struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; }; + using RxQueueGuard = Guard; + using RxQueue = mesh_nrf52::BleRxFrameQueue; + RxQueue recv_queue; + uint8_t send_queue_len = 0; + Frame send_queue[FRAME_QUEUE_SIZE]; + std::atomic _tx_reset_pending{false}; + uint32_t _send_queue_generation = 0; + uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; + uint32_t _tx_recovery_generation = 0; + uint16_t _conn_handle = 1; + bool _isEnabled = true, _isDeviceConnected = true; + unsigned long _last_retry_attempt = 0; + mesh::BleTxStallWatchdog _tx_stall_watchdog; + mesh::BleDisconnectRecovery _tx_disconnect_recovery; + SecuritySessionTimer _companion_start_timer; + SecuritySessionTimer _security_timer; + bool _peer_address_valid = true; + struct Uart { + struct Chunk { uint16_t handle; std::vector bytes; }; + std::vector chunks; + 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); } + size_t write(uint16_t handle, const uint8_t* data, size_t length) { + assert(!critical_depth); + if (++attempts == fail_on_attempt) return 0; + chunks.push_back({handle, std::vector(data, data + length)}); + if (after_write) { + auto operation = std::move(after_write); + after_write = nullptr; + operation(); + } + return length; + } + } bleuart; + SerialBLEInterface() { open(1); } + void open(uint16_t handle) { + _conn_handle = handle; + recv_queue.setEnabled(true); + recv_queue.connect(handle); + recv_queue.secure(handle); + _isEnabled = _isDeviceConnected = true; + } + void reconnect(uint16_t handle, bool deferred_callback = true) { + recv_queue.disconnect(_conn_handle); + open(handle); + if (deferred_callback) clearBuffers(); + } + void serviceBondedOnlyTransition() {} + bool advertisingAllowed() const { return false; } + bool isAdvertising() const { return false; } + bool startAdvertising(const char*) { assert(false); return false; } + void serviceTxRecovery(uint32_t); + void clearBuffers(); + void resetTxState(); + bool servicePendingTxReset(); + void shiftSendQueueLeft(); + size_t writeBleUartFrame(const Frame&); + bool hasRequiredReplyCapacity() const; + size_t writeFrame(const uint8_t*, size_t); + size_t checkRecvFrame(uint8_t*); + void recoverStalledTx(const char*, uint16_t = BLE_CONN_HANDLE_INVALID, uint32_t = 0); + bool isConnected() const; + bool isWriteBusy() const; + bool hasPendingIO() const; +}; +@METHODS@ + +static void enqueue_reset() { + SerialBLEInterface port; + const uint8_t old_reply[] = {1, 11}, interrupted[] = {2, 22}, fresh[] = {3, 33}; + assert(port.writeFrame(old_reply, sizeof(old_reply)) == sizeof(old_reply)); + port._last_retry_attempt = 77; + port._tx_stall_watchdog.noteBlocked(100); + during_copy = [&]() { port.clearBuffers(); }; + assert(port.writeFrame(interrupted, sizeof(interrupted)) == 0); + assert(port.send_queue_len == 0 && !port._last_retry_attempt); + assert(!port._tx_stall_watchdog.active() && !port._tx_disconnect_recovery.pending()); + assert(port.writeFrame(fresh, sizeof(fresh)) == sizeof(fresh)); + uint8_t received[MAX_FRAME_SIZE]; + assert(port.checkRecvFrame(received) == 0 && port.send_queue_len == 0); + assert(port.bleuart.chunks.size() == 1 && port.bleuart.chunks[0].bytes[1] == 33); +} +static void callback_ownership() { + SerialBLEInterface port; + const uint8_t reply[] = {1, 11}; + for (unsigned i = 0; i < FRAME_QUEUE_SIZE; ++i) + assert(port.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + assert(port.isWriteBusy() && port.hasPendingIO()); + port.clearBuffers(); + // A callback invalidates acceptance immediately without mutating memory + // that the interrupted main-task enqueue/copy may still be using. + assert(port.send_queue_len == FRAME_QUEUE_SIZE && port._tx_reset_pending); + assert(!port.isWriteBusy() && !port.hasPendingIO() && port.hasRequiredReplyCapacity()); + assert(port.servicePendingTxReset() && !port.servicePendingTxReset()); + assert(port.send_queue_len == 0); +} +static void connection_changes() { + for (uint16_t replacement : {uint16_t(1), uint16_t(2)}) + for (bool deferred_callback : {false, true}) { + SerialBLEInterface port; + const size_t requests = disconnect_requests.size(); + uint8_t frame[40]; + for (unsigned i = 0; i < sizeof(frame); ++i) frame[i] = i; + assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame)); + port.bleuart.after_write = [&]() { port.reconnect(replacement, deferred_callback); }; + uint8_t received[MAX_FRAME_SIZE]; + assert(port.checkRecvFrame(received) == 0); + assert(port.bleuart.chunks.size() == 1); + assert(port.bleuart.chunks[0].handle == 1 && port.bleuart.chunks[0].bytes.size() == 20); + assert(port.send_queue_len == 0 && !port._tx_disconnect_recovery.pending()); + // Old partial completion must not recover/disconnect the new session. + assert(disconnect_requests.size() == requests && port.isConnected()); + const uint8_t fresh[] = {1, 99}; + assert(port.writeFrame(fresh, sizeof(fresh)) == sizeof(fresh)); + assert(port.checkRecvFrame(received) == 0 && port.send_queue_len == 0); + assert(port.bleuart.chunks.size() == 2); + assert(port.bleuart.chunks[1].handle == replacement + && port.bleuart.chunks[1].bytes == std::vector(fresh, fresh + sizeof(fresh))); + } +} +static void old_queue_and_recovery() { + uint8_t received[MAX_FRAME_SIZE]; + const uint8_t reply[] = {1, 11}; + { + SerialBLEInterface port; + assert(port.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + port.reconnect(2, false); // Physical event precedes its deferred callback. + assert(port.checkRecvFrame(received) == 0 && port.bleuart.chunks.empty()); + assert(port.send_queue_len == 0 && !port._tx_disconnect_recovery.pending()); + } + { + SerialBLEInterface port; + const size_t requests = disconnect_requests.size(); + uint8_t frame[40] = {1}; + assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame)); + port.bleuart.fail_on_attempt = 2; + assert(port.checkRecvFrame(received) == 0); + assert(port.bleuart.chunks.size() == 1 && port._tx_disconnect_recovery.pending()); + assert(port.send_queue_len == 0 && disconnect_requests.size() == requests+1); + assert(disconnect_requests.back() == 1); + assert(port.checkRecvFrame(received) == 0 && port._tx_disconnect_recovery.pending()); + assert(disconnect_requests.size() == requests+1); // Own recovery persists and throttles retries. + now_ms += 1001; + assert(port.checkRecvFrame(received) == 0 && port._tx_disconnect_recovery.pending()); + assert(disconnect_requests.size() == requests+2 && disconnect_requests.back() == 1); + port.reconnect(2); + assert(port.isConnected()); // Old recovery cannot poison a secured new link. + assert(port.writeFrame(reply, sizeof(reply)) == sizeof(reply)); + assert(port.checkRecvFrame(received) == 0 && !port._tx_disconnect_recovery.pending()); + } +} +static void recovery_connection_changes() { + for (uint32_t scheduling_point : {0xf031U, 0xf025U, 0xf020U}) + for (uint16_t replacement : {uint16_t(1), uint16_t(2)}) { + SerialBLEInterface port; + const size_t requests = disconnect_requests.size(); + uint8_t frame[40] = {1}, received[MAX_FRAME_SIZE]; + assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame)); + port.bleuart.fail_on_attempt = 2; + bool changed = false; + trace_hook = [&](uint32_t event) { + if (!changed && event == scheduling_point) { changed = true; port.reconnect(replacement); } + }; + assert(port.checkRecvFrame(received) == 0 && changed); + trace_hook = nullptr; + assert(disconnect_requests.size() == requests); + assert(port.checkRecvFrame(received) == 0 && !port._tx_disconnect_recovery.pending()); + assert(port.isConnected() && disconnect_requests.size() == requests); + } + // A physical connect event may precede its deferred callback, even when + // SoftDevice reuses a handle. A retry must still not disconnect that link. + for (uint16_t replacement : {uint16_t(1), uint16_t(2)}) { + SerialBLEInterface port; + uint8_t frame[40] = {1}, received[MAX_FRAME_SIZE]; + assert(port.writeFrame(frame, sizeof(frame)) == sizeof(frame)); + port.bleuart.fail_on_attempt = 2; + assert(port.checkRecvFrame(received) == 0 && port._tx_disconnect_recovery.pending()); + const size_t requests = disconnect_requests.size(); + port.reconnect(replacement, false); + now_ms += 1001; + assert(port.checkRecvFrame(received) == 0 && !port._tx_disconnect_recovery.pending()); + assert(disconnect_requests.size() == requests && port.isConnected()); + } +} +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 { + enqueue_reset(); callback_ownership(); connection_changes(); old_queue_and_recovery(); + recovery_connection_changes(); + } + assert(!critical_depth && !during_copy); +} diff --git a/test/test_nrf52_ble_backpressure.py b/test/test_nrf52_ble_backpressure.py index 9e27fa37e..73eee1d31 100644 --- a/test/test_nrf52_ble_backpressure.py +++ b/test/test_nrf52_ble_backpressure.py @@ -19,6 +19,7 @@ HARNESS = r''' #include #include #include +#include #include #include #include @@ -26,6 +27,7 @@ HARNESS = r''' #define MAX_FRAME_SIZE 176 #define FRAME_QUEUE_SIZE 12 +#define BLE_CONN_HANDLE_INVALID 0xffff #define BLE_RETRY_THROTTLE_MS 250 #define BLE_DEBUG_PRINTLN(...) do {} while (0) #define meshBleTrace(...) do {} while (0) @@ -38,6 +40,11 @@ struct SerialBLEInterface { using RxQueue = mesh_nrf52::BleRxFrameQueue; uint8_t send_queue_len = 0; Frame send_queue[FRAME_QUEUE_SIZE]; + uint16_t _conn_handle = 1; + std::atomic _tx_reset_pending{false}; + uint32_t _send_queue_generation = 0; + uint16_t _tx_recovery_handle = BLE_CONN_HANDLE_INVALID; + uint32_t _tx_recovery_generation = 0; RxQueue recv_queue; SerialBLEInterface() { recv_queue.setEnabled(true); @@ -53,7 +60,9 @@ struct SerialBLEInterface { bool isConnected() const { return connected && !_tx_disconnect_recovery.pending(); } void serviceBondedOnlyTransition() {} void serviceTxRecovery(uint32_t) { ++recovery_services; } - void recoverStalledTx(const char*) { ++recoveries; _tx_disconnect_recovery.begin(); } + void recoverStalledTx(const char*, uint16_t = BLE_CONN_HANDLE_INVALID, uint32_t = 0) { + ++recoveries; _tx_disconnect_recovery.begin(); + } size_t writeBleUartFrame(const Frame& frame) { ++write_attempts; if (partial_write) return 1; @@ -71,6 +80,8 @@ struct SerialBLEInterface { assert(recv_queue.push(1, generation, frame) == RxQueue::PushResult::Accepted); } void shiftSendQueueLeft(); + void resetTxState(); + bool servicePendingTxReset(); bool hasRequiredReplyCapacity() const; bool isWriteBusy() const; size_t writeFrame(const uint8_t*, size_t); @@ -184,6 +195,8 @@ class Nrf52BleBackpressureTests(unittest.TestCase): def test_real_queue_dispatch_throttle_and_recovery_under_sanitizers(self): source = SOURCE.read_text() signatures = ["void SerialBLEInterface::shiftSendQueueLeft()", + "void SerialBLEInterface::resetTxState()", + "bool SerialBLEInterface::servicePendingTxReset()", "size_t SerialBLEInterface::writeFrame(", "size_t SerialBLEInterface::checkRecvFrame(", "bool SerialBLEInterface::isWriteBusy() const"] diff --git a/test/test_nrf52_ble_reconnect_watchdog.py b/test/test_nrf52_ble_reconnect_watchdog.py index 1a6080ef3..89fd20d8e 100644 --- a/test/test_nrf52_ble_reconnect_watchdog.py +++ b/test/test_nrf52_ble_reconnect_watchdog.py @@ -48,6 +48,7 @@ struct SerialBLEInterface { int advertisements = 0; int recovery_services = 0; void serviceBondedOnlyTransition() {} + bool servicePendingTxReset() { return false; } void serviceTxRecovery(uint32_t) { ++recovery_services; } void disconnect() { ++disconnects; } void recoverStalledTx(const char*) { diff --git a/test/test_nrf52_ble_rx_queue.py b/test/test_nrf52_ble_rx_queue.py index 626d116d3..9ee99168b 100644 --- a/test/test_nrf52_ble_rx_queue.py +++ b/test/test_nrf52_ble_rx_queue.py @@ -27,6 +27,8 @@ class Nrf52BleRxQueueTests(unittest.TestCase): def test_production_queue_and_callback_interleavings_under_sanitizers(self): source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text() signatures = ["void SerialBLEInterface::clearBuffers()", + "void SerialBLEInterface::resetTxState()", + "bool SerialBLEInterface::servicePendingTxReset()", "static bool isBondAuthenticationFailure(", "void SerialBLEInterface::onBLEEvent(", "void SerialBLEInterface::onConnect(", diff --git a/test/test_nrf52_ble_tx_session.py b/test/test_nrf52_ble_tx_session.py new file mode 100644 index 000000000..427dc8957 --- /dev/null +++ b/test/test_nrf52_ble_tx_session.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""Run actual BLE TX methods across callback resets and connection changes.""" +from pathlib import Path +import os +import shutil +import subprocess +import tempfile +import unittest + +from cpp_source import body + +ROOT = Path(__file__).resolve().parents[1] +LEGACY_CLEAR = r''' +void SerialBLEInterface::clearBuffers() { + send_queue_len = 0; + _last_retry_attempt = 0; + _tx_stall_watchdog.reset(); + _tx_disconnect_recovery.complete(); + recv_queue.reset([this]() { bleuart.flush(); }); +} +''' +LEGACY_ENQUEUE = r''' +size_t SerialBLEInterface::writeFrame(const uint8_t src[], size_t len) { + if (len > MAX_FRAME_SIZE) return 0; + if (isConnected() && len > 0) { + if (!mesh::enqueueCompanionFrame(send_queue, send_queue_len, FRAME_QUEUE_SIZE, src, len)) return 0; + return len; + } + return 0; +} +''' +LEGACY_WRITE = r''' +size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) { + BLEConnection* conn = Bluefruit.Connection(_conn_handle); + if (!conn || !conn->connected() || !bleuart.notifyEnabled(_conn_handle)) return 0; + const uint16_t mtu = conn->getMtu(); + if (mtu <= 3) return 0; + return mesh::writeBleFrameInChunks(frame.buf, frame.len, mtu - 3, + [this](const uint8_t* data, size_t len) { return bleuart.write(_conn_handle, data, len); }); +} +''' + + +class Nrf52BleTxSessionTests(unittest.TestCase): + @classmethod + def setUpClass(cls): + compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if not compiler: + raise RuntimeError("BLE TX session regression requires a host C++17 compiler") + source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text() + signatures = ( + "void SerialBLEInterface::clearBuffers()", "void SerialBLEInterface::resetTxState()", + "bool SerialBLEInterface::servicePendingTxReset()", "void SerialBLEInterface::shiftSendQueueLeft()", + "size_t SerialBLEInterface::writeBleUartFrame(", "bool SerialBLEInterface::hasRequiredReplyCapacity() const", + "size_t SerialBLEInterface::writeFrame(", "size_t SerialBLEInterface::checkRecvFrame(", + "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) + 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)} + fixture = (ROOT / "test/fixtures/nrf52_ble_tx_session.cpp").read_text(encoding="ascii") + cls.work = tempfile.TemporaryDirectory(prefix="meshcore-ble-tx-session-") + cls.binaries = {} + for name, methods in variants.items(): + directory = Path(cls.work.name) + path, binary = directory / (name + ".cpp"), directory / name + path.write_text(fixture.replace("@METHODS@", methods), encoding="ascii") + result = subprocess.run([ + compiler, "-std=c++17", "-O2", "-g", "-Wall", "-Wextra", "-Werror", "-Wno-unused-parameter", + "-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie", + "-I" + str(ROOT / "src"), str(path), "-o", str(binary)], + capture_output=True, text=True, timeout=30) + if result.returncode: + cls.work.cleanup() + raise AssertionError(result.stdout + result.stderr) + cls.binaries[name] = binary + + @classmethod + def tearDownClass(cls): + cls.work.cleanup() + + def run_case(self, variant, scenario): + return subprocess.run([str(self.binaries[variant]), scenario], capture_output=True, + text=True, timeout=10) + + def test_production_tx_state_is_main_owned_and_fragments_stay_in_session(self): + result = self.run_case("production", "all") + self.assertEqual(result.returncode, 0, result.stdout + result.stderr) + + def test_previous_reset_and_chunk_paths_reproduce_the_failures(self): + for variant, scenario, assertion in ( + ("legacy_enqueue", "enqueue", "writeFrame(interrupted"), + ("legacy_write", "connection", "chunks.size() == 1")): + with self.subTest(variant=variant): + result = self.run_case(variant, scenario) + self.assertNotEqual(result.returncode, 0) + self.assertIn(assertion, result.stderr) + + +if __name__ == "__main__": + unittest.main()