Keep BLE transmit queues and recovery bound to their connection session

.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

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

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

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<bool> _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);

test/fixtures/nrf52_ble_rx_queue.cpp

@@ -133,6 +133,9 @@ struct SerialBLEInterface {
   using RxQueue = mesh_nrf52::BleRxFrameQueue<Frame, FRAME_QUEUE_SIZE, RxQueueGuard>;
   RxQueue recv_queue;
   uint8_t send_queue_len = 0;
+  std::atomic<bool> _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);
   }
 

test/fixtures/nrf52_ble_tx_session.cpp

@@ -0,0 +1,263 @@
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <functional>
+#include <vector>
+#include <helpers/BleTxStallWatchdog.h>
+#include <helpers/nrf52/BleRxFrameQueue.h>
+#include <helpers/nrf52/SecuritySessionTimer.h>
+
+static unsigned critical_depth;
+struct Guard {
+  Guard() { ++critical_depth; }
+  ~Guard() { --critical_depth; }
+};
+static std::function<void()> 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 <helpers/CompanionFrameQueue.h>
+#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<void(uint32_t)> 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<uint16_t> 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<Frame, FRAME_QUEUE_SIZE, Guard>;
+  RxQueue recv_queue;
+  uint8_t send_queue_len = 0;
+  Frame send_queue[FRAME_QUEUE_SIZE];
+  std::atomic<bool> _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<uint8_t> bytes; };
+    std::vector<Chunk> chunks;
+    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); }
+    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<uint8_t>(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<uint8_t>(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);
+}

test/test_nrf52_ble_backpressure.py

@@ -19,6 +19,7 @@ HARNESS = r'''
 #include <cassert>
 #include <cstdint>
 #include <cstring>
+#include <atomic>
 #include <vector>
 #include <helpers/CompanionFrameQueue.h>
 #include <helpers/BleTxStallWatchdog.h>
@@ -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<Frame, FRAME_QUEUE_SIZE, RxQueueGuard>;
   uint8_t send_queue_len = 0;
   Frame send_queue[FRAME_QUEUE_SIZE];
+  uint16_t _conn_handle = 1;
+  std::atomic<bool> _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"]

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*) {

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(",

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