Serialize nRF52 BLE receive frames across link resets

.github/workflows/run-unit-tests.yml

@@ -708,6 +708,7 @@ jobs:
           python3 -B test/test_t096_full_memory.py
           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_trace.py
           python3 -B test/test_nrf52_ble_debug_tools.py
           python3 -B test/test_nrf52_usb_power.py

src/helpers/nrf52/BleRxFrameQueue.h

@@ -0,0 +1,120 @@
+#pragma once
+
+#include <stddef.h>
+#include <stdint.h>
+#include <string.h>
+
+namespace mesh_nrf52 {
+
+// Guard serializes task access on the single-core nRF52. Keep every guarded
+// operation bounded to metadata or one frame copy; UART FIFO operations must
+// run outside the guard because Bluefruit takes a FreeRTOS mutex there.
+template <typename Frame, size_t Capacity, typename Guard>
+class BleRxFrameQueue {
+  static_assert(Capacity > 0 && Capacity <= 255, "invalid BLE RX capacity");
+  static constexpr uint16_t INVALID_HANDLE = 0xffff;
+
+  Frame _frames[Capacity];
+  uint32_t _generation = 0;
+  uint16_t _handle = INVALID_HANDLE;
+  uint16_t _resets_in_progress = 0;
+  uint8_t _head = 0;
+  uint8_t _count = 0;
+  bool _enabled = false;
+  bool _secured = false;
+
+  void invalidate() {
+    ++_generation;
+    _head = _count = 0;
+  }
+
+  bool accepts(uint16_t handle, uint32_t generation) const {
+    return _enabled && _secured && !_resets_in_progress
+        && handle == _handle && generation == _generation;
+  }
+
+public:
+  enum class PushResult { Accepted, Full, Stale };
+
+  void setEnabled(bool enabled) {
+    Guard guard;
+    _enabled = enabled;
+    _handle = INVALID_HANDLE;
+    _secured = false;
+    invalidate();
+  }
+
+  void connect(uint16_t handle) {
+    Guard guard;
+    _handle = _enabled ? handle : INVALID_HANDLE;
+    _secured = false;
+    invalidate();
+  }
+
+  void secure(uint16_t handle) {
+    Guard guard;
+    if (_enabled && _handle == handle && handle != INVALID_HANDLE) {
+      _secured = true;
+    }
+  }
+
+  void disconnect(uint16_t handle) {
+    Guard guard;
+    if (_handle != handle) return;
+    _handle = INVALID_HANDLE;
+    _secured = false;
+    invalidate();
+  }
+
+  template <typename Flush>
+  void reset(Flush flush) {
+    {
+      Guard guard;
+      ++_resets_in_progress;
+      invalidate();
+    }
+    flush();
+    {
+      Guard guard;
+      --_resets_in_progress;
+    }
+  }
+
+  bool snapshot(uint16_t handle, uint32_t& generation) const {
+    Guard guard;
+    generation = _generation;
+    return accepts(handle, generation);
+  }
+
+  bool current(uint16_t handle, uint32_t generation) const {
+    Guard guard;
+    return accepts(handle, generation);
+  }
+
+  PushResult push(uint16_t handle, uint32_t generation, const Frame& frame) {
+    Guard guard;
+    if (!accepts(handle, generation)) return PushResult::Stale;
+    if (_count == Capacity) return PushResult::Full;
+    _frames[(_head + _count) % Capacity] = frame;
+    ++_count;
+    return PushResult::Accepted;
+  }
+
+  size_t pop(uint8_t* destination) {
+    Guard guard;
+    if (!_count) return 0;
+    const Frame& frame = _frames[_head];
+    const size_t length = frame.len;
+    memcpy(destination, frame.buf, length);
+    _head = (_head + 1) % Capacity;
+    --_count;
+    return length;
+  }
+
+  uint8_t size() const {
+    Guard guard;
+    return _count;
+  }
+};
+
+}  // namespace mesh_nrf52

src/helpers/nrf52/SerialBLEInterface.cpp

@@ -32,9 +32,6 @@ __attribute__((used, externally_visible)) MeshBleTraceBuffer mesh_ble_trace;
 #define BLE_ADV_INTERVAL_MAX       244    // 152.5ms (units: 0.625ms)
 #define BLE_ADV_FAST_TIMEOUT       30     // seconds
 
-// RX drain buffer size for overflow protection
-#define BLE_RX_DRAIN_BUF_SIZE      32
-
 static SerialBLEInterface* instance = nullptr;
 
 #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE
@@ -45,7 +42,7 @@ void SerialBLEInterface::formatDebugState(char* reply, size_t capacity) {
       (unsigned)_isDeviceConnected, (unsigned)Bluefruit.Advertising.isRunning(),
       (unsigned)Bluefruit.connected(),
       (unsigned)_companionDataSeen.load(std::memory_order_acquire),
-      (unsigned)send_queue_len, (unsigned)recv_queue_len,
+      (unsigned)send_queue_len, (unsigned)recv_queue.size(),
       (unsigned)_tx_disconnect_recovery.pending(),
       (unsigned)(_conn_handle != BLE_CONN_HANDLE_INVALID && bleuart.notifyEnabled(_conn_handle)),
       (unsigned)_debug_mtu.load());
@@ -273,6 +270,7 @@ void SerialBLEInterface::onSecured(uint16_t connection_handle) {
       }
 
       instance->_isDeviceConnected = true;
+      instance->recv_queue.secure(connection_handle);
 #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE
       mesh_ble_trace.secured.fetch_add(1);
       meshBleTrace(0xf006, connection_handle, conn->bonded(),
@@ -431,6 +429,10 @@ void SerialBLEInterface::onBLEEvent(ble_evt_t* evt) {
   if (evt->header.evt_id == BLE_GAP_EVT_CONNECTED) {
     ble_gap_evt_connected_t const* connected = &evt->evt.gap_evt.params.connected;
     if (connected->role == BLE_GAP_ROLE_PERIPH) {
+      // Observe the physical session in the BLE worker before its deferred
+      // onConnect callback. A disable/re-enable between the event and that
+      // callback must not allow the old link to reopen the RX queue.
+      instance->recv_queue.connect(evt->evt.gap_evt.conn_handle);
       instance->_peer_address = connected->peer_addr;
       instance->_peer_address_valid = true;
       instance->_bond_removed_for_connection = false;
@@ -480,6 +482,7 @@ void SerialBLEInterface::onBLEEvent(ble_evt_t* evt) {
   } else if (evt->header.evt_id == BLE_GATTS_EVT_SYS_ATTR_MISSING) {
     BLE_DEBUG_PRINTLN("CCCD RAM system attributes missing");
   } else if (evt->header.evt_id == BLE_GAP_EVT_DISCONNECTED) {
+    instance->recv_queue.disconnect(evt->evt.gap_evt.conn_handle);
     ble_gap_evt_disconnected_t const* disconnected = &evt->evt.gap_evt.params.disconnected;
     if (isBondAuthenticationFailure(disconnected->reason)) {
       instance->removeStoredBondForPeer("authentication disconnect");
@@ -986,11 +989,10 @@ void SerialBLEInterface::cancelStealthPairingTransition() {
 
 void SerialBLEInterface::clearBuffers() {
   send_queue_len = 0;
-  recv_queue_len = 0;
   _last_retry_attempt = 0;
   _tx_stall_watchdog.reset();
   _tx_disconnect_recovery.complete();
-  bleuart.flush();
+  recv_queue.reset([this]() { bleuart.flush(); });
 }
 
 void SerialBLEInterface::shiftSendQueueLeft() {
@@ -1002,15 +1004,6 @@ void SerialBLEInterface::shiftSendQueueLeft() {
   }
 }
 
-void SerialBLEInterface::shiftRecvQueueLeft() {
-  if (recv_queue_len > 0) {
-    recv_queue_len--;
-    for (uint8_t i = 0; i < recv_queue_len; i++) {
-      recv_queue[i] = recv_queue[i + 1];
-    }
-  }
-}
-
 size_t SerialBLEInterface::writeBleUartFrame(const Frame& frame) {
   BLEConnection* conn = Bluefruit.Connection(_conn_handle);
   if (conn == nullptr || !conn->connected() ||
@@ -1071,7 +1064,7 @@ void SerialBLEInterface::serviceTxRecovery(uint32_t now) {
 }
 
 void SerialBLEInterface::recoverStalledTx(const char* cause) {
-  meshBleTrace(0xf020, _conn_handle, send_queue_len, recv_queue_len);
+  meshBleTrace(0xf020, _conn_handle, send_queue_len, recv_queue.size());
   if (_tx_disconnect_recovery.pending()) return;
 
   BLE_DEBUG_PRINTLN("SerialBLEInterface: %s; forcing reconnect", cause);
@@ -1080,11 +1073,11 @@ void SerialBLEInterface::recoverStalledTx(const char* cause) {
   // disconnection, but make isConnected() false through the recovery state so
   // no more companion frames enter this damaged stream.
   send_queue_len = 0;
-  recv_queue_len = 0;
+  recv_queue.disconnect(_conn_handle);
   _last_retry_attempt = 0;
   _tx_stall_watchdog.reset();
   _companion_start_timer.cancel();
-  bleuart.flush();
+  recv_queue.reset([this]() { bleuart.flush(); });
 #if COMPANION_FEATURE_BLE_MOTA_SOURCE
   setMotaStreamActive(false);
 #endif
@@ -1113,6 +1106,7 @@ bool SerialBLEInterface::isAdvertising() const {
 void SerialBLEInterface::enable() {
   if (_isEnabled) return;
 
+  recv_queue.setEnabled(true);
   _pairingRequestPending.store(false, std::memory_order_release);
   _isEnabled = true;
   clearBuffers();
@@ -1126,6 +1120,7 @@ void SerialBLEInterface::enable() {
 
 void SerialBLEInterface::disconnect() {
   if (_conn_handle != BLE_CONN_HANDLE_INVALID) {
+    recv_queue.disconnect(_conn_handle);
     const uint32_t result = sd_ble_gap_disconnect(_conn_handle, BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION);
     meshBleTrace(0xf013, _conn_handle, result, 0);
     (void)result;
@@ -1133,6 +1128,7 @@ void SerialBLEInterface::disconnect() {
 }
 
 void SerialBLEInterface::disable() {
+  recv_queue.setEnabled(false);
   _isEnabled = false;
   _pairingRequestPending.store(false, std::memory_order_release);
   BLE_DEBUG_PRINTLN("SerialBLEInterface: disable");
@@ -1298,13 +1294,11 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) {
   
   // TX/recovery above must run even when full. Retain RX until a required
   // command reply can be admitted instead of silently losing that reply.
-  if (recv_queue_len > 0 && hasRequiredReplyCapacity()) {
-    size_t len = recv_queue[0].len;
-    memcpy(dest, recv_queue[0].buf, len);
-    
-    BLE_DEBUG_PRINTLN("readBytes: sz=%u, hdr=%u", (unsigned)len, (unsigned)dest[0]);
-    
-    shiftRecvQueueLeft();
+  if (hasRequiredReplyCapacity()) {
+    const size_t len = recv_queue.pop(dest);
+    if (len) {
+      BLE_DEBUG_PRINTLN("readBytes: sz=%u, hdr=%u", (unsigned)len, (unsigned)dest[0]);
+    }
     return len;
   }
   
@@ -1312,50 +1306,49 @@ size_t SerialBLEInterface::checkRecvFrame(uint8_t dest[]) {
 }
 
 void SerialBLEInterface::onBleUartRX(uint16_t conn_handle) {
-  if (!instance) {
-    return;
-  }
-  
-  meshBleTrace(0xf030, conn_handle, instance->bleuart.available(),
-               instance->isConnected());
-  if (instance->_conn_handle != conn_handle || !instance->isConnected()) {
-    while (instance->bleuart.available() > 0) {
-      instance->bleuart.read();
-    }
-    return;
-  }
-  
-  while (instance->bleuart.available() > 0) {
-    if (instance->recv_queue_len >= FRAME_QUEUE_SIZE) {
-      while (instance->bleuart.available() > 0) {
-        instance->bleuart.read();
-      }
-      meshBleTrace(0xf034, conn_handle, instance->recv_queue_len);
-      BLE_DEBUG_PRINTLN("onBleUartRX: recv queue full, dropping data");
-      break;
-    }
-    
-    int avail = instance->bleuart.available();
-    
+  SerialBLEInterface* self = instance;
+  if (!self) return;
+
+  uint32_t generation;
+  if (!self->recv_queue.snapshot(conn_handle, generation)
+      || !self->isConnected()) return;
+  meshBleTrace(0xf030, conn_handle, self->bleuart.available(), true);
+
+  while (self->recv_queue.current(conn_handle, generation)) {
+    const int avail = self->bleuart.available();
+    if (avail <= 0) return;
     if (avail > MAX_FRAME_SIZE) {
       BLE_DEBUG_PRINTLN("onBleUartRX: WARN: BLE RX overflow, avail=%d, draining all", avail);
-      uint8_t drain_buf[BLE_RX_DRAIN_BUF_SIZE];
-      while (instance->bleuart.available() > 0) {
-        int chunk = instance->bleuart.available() > BLE_RX_DRAIN_BUF_SIZE ? BLE_RX_DRAIN_BUF_SIZE : instance->bleuart.available();
-        instance->bleuart.readBytes(drain_buf, chunk);
-      }
-      continue;
+      self->bleuart.flush();
+      return;
+    }
+
+    // BLEUart::read() is an immediate FIFO read. Stream::readBytes() may wait
+    // for bytes flushed by a simultaneous reset. Stage outside the critical
+    // section, then reject the frame if its connection generation changed.
+    Frame frame = {};
+    const int read_len = self->bleuart.read(frame.buf, avail);
+    if (read_len != avail) return;
+    frame.len = read_len;
+    RxQueue::PushResult result;
+    {
+      RxQueueGuard guard;
+      result = self->recv_queue.push(conn_handle, generation, frame);
+      if (result == RxQueue::PushResult::Accepted) {
+        self->_companionDataSeen.store(true, std::memory_order_release);
+        self->_companion_start_timer.cancel();
+      }
+    }
+    if (result == RxQueue::PushResult::Stale) return;
+    if (result == RxQueue::PushResult::Full) {
+      self->bleuart.flush();
+      meshBleTrace(0xf034, conn_handle, FRAME_QUEUE_SIZE);
+      BLE_DEBUG_PRINTLN("onBleUartRX: recv queue full, dropping data");
+      return;
     }
-    
-    int read_len = avail;
-    instance->recv_queue[instance->recv_queue_len].len = read_len;
-    instance->bleuart.readBytes(instance->recv_queue[instance->recv_queue_len].buf, read_len);
-    instance->recv_queue_len++;
 #if defined(MESH_NRF52_BLE_TRACE) && MESH_NRF52_BLE_TRACE
     mesh_ble_trace.rx.fetch_add(1);
 #endif
-    instance->_companionDataSeen.store(true, std::memory_order_release);
-    instance->_companion_start_timer.cancel();
   }
 }
 
@@ -1393,7 +1386,7 @@ uint16_t SerialBLEInterface::motaResponseCapacity() {
 #endif
 
 bool SerialBLEInterface::isReadBusy() const {
-  return (recv_queue_len > 0);
+  return recv_queue.size() > 0;
 }
 
 bool SerialBLEInterface::isWriteBusy() const {
@@ -1401,5 +1394,5 @@ bool SerialBLEInterface::isWriteBusy() const {
 }
 
 bool SerialBLEInterface::hasPendingIO() const {
-  return recv_queue_len > 0 || send_queue_len > 0;
+  return recv_queue.size() > 0 || send_queue_len > 0;
 }

src/helpers/nrf52/SerialBLEInterface.h

@@ -5,6 +5,7 @@
 #include "../BleTxStallWatchdog.h"
 #include "../UsbLogging.h"
 #include "SecuritySessionTimer.h"
+#include "BleRxFrameQueue.h"
 #include <bluefruit.h>
 #include <atomic>
 #if COMPANION_FEATURE_BLE_MOTA_SOURCE
@@ -105,13 +106,17 @@ class SerialBLEInterface : public BaseSerialInterface {
   uint8_t send_queue_len;
   Frame send_queue[FRAME_QUEUE_SIZE];
   
-  uint8_t recv_queue_len;
-  Frame recv_queue[FRAME_QUEUE_SIZE];
+  struct RxQueueGuard {
+    RxQueueGuard() { taskENTER_CRITICAL(); }
+    ~RxQueueGuard() { taskEXIT_CRITICAL(); }
+  };
+  using RxQueue = mesh_nrf52::BleRxFrameQueue<Frame, FRAME_QUEUE_SIZE,
+                                             RxQueueGuard>;
+  RxQueue recv_queue;
 
   void clearBuffers();
   bool hasRequiredReplyCapacity() const;
   void shiftSendQueueLeft();
-  void shiftRecvQueueLeft();
   size_t writeBleUartFrame(const Frame& frame);
   void recoverStalledTx(const char* cause);
   void serviceTxRecovery(uint32_t now);
@@ -156,7 +161,6 @@ public:
     _bonded_only = false;
     _bonded_only_configure_pending = false;
     send_queue_len = 0;
-    recv_queue_len = 0;
   }
 
   /**

test/fixtures/nrf52_ble_rx_queue.cpp

@@ -0,0 +1,477 @@
+#include <algorithm>
+#include <atomic>
+#include <cassert>
+#include <cstdint>
+#include <cstring>
+#include <deque>
+#include <functional>
+#include <utility>
+#include <vector>
+#include <helpers/BleTxStallWatchdog.h>
+#include <helpers/nrf52/SecuritySessionTimer.h>
+
+static unsigned critical_depth = 0;
+static std::deque<std::function<void()>> deferred;
+static std::function<void()> during_copy;
+
+// Model a single-core RTOS: a preemption requested during a critical section
+// runs when the outermost guard exits. All UART/logging operations assert that
+// interrupts and task switching are enabled.
+static void preempt(std::function<void()> operation) {
+  if (critical_depth) deferred.push_back(std::move(operation));
+  else operation();
+}
+
+struct TestGuard {
+  TestGuard() { ++critical_depth; }
+  ~TestGuard() {
+    assert(critical_depth);
+    if (--critical_depth) return;
+    while (!deferred.empty()) {
+      auto operation = std::move(deferred.front());
+      deferred.pop_front();
+      operation();
+    }
+  }
+};
+
+static void* guarded_copy(void* dest, const void* source, size_t length) {
+  assert(critical_depth);
+  const size_t first = length / 2;
+  std::memcpy(dest, source, first);
+  if (during_copy) {
+    auto operation = std::move(during_copy);
+    during_copy = nullptr;
+    preempt(std::move(operation));
+  }
+  std::memcpy(static_cast<uint8_t*>(dest) + first,
+              static_cast<const uint8_t*>(source) + first, length - first);
+  return dest;
+}
+
+#define memcpy guarded_copy
+#include <helpers/nrf52/BleRxFrameQueue.h>
+#undef memcpy
+
+#define MAX_FRAME_SIZE 176
+#define FRAME_QUEUE_SIZE 12
+#define COMPANION_RADIO_FULL 1
+#define COMPANION_FEATURE_BLE_MOTA_SOURCE 0
+#define BLE_CONN_HANDLE_INVALID 0xffff
+#define BLE_DEBUG_PRINTLN(...) do { assert(!critical_depth); } while (0)
+#define meshBleTrace(...) do { assert(!critical_depth); } while (0)
+
+enum {
+  BLE_GAP_EVT_CONNECTED = 1, BLE_GAP_EVT_DISCONNECTED,
+  BLE_GAP_EVT_CONN_SEC_UPDATE, BLE_GATTS_EVT_WRITE,
+  BLE_GATTS_EVT_SYS_ATTR_MISSING, BLE_GAP_EVT_CONN_PARAM_UPDATE_REQUEST,
+  BLE_GAP_ROLE_PERIPH = 1, BLE_UUID_TYPE_BLE = 1,
+  BLE_UUID_DESCRIPTOR_CLIENT_CHAR_CONFIG = 0x2902,
+  BLE_HCI_AUTHENTICATION_FAILURE = 5,
+  BLE_HCI_STATUS_CODE_PIN_OR_KEY_MISSING = 6,
+  BLE_HCI_CONN_TERMINATED_DUE_TO_MIC_FAILURE = 0x3d,
+  BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION = 0x13,
+  NRF_SUCCESS = 0
+};
+struct ble_gap_addr_t { uint8_t addr[6] = {}; };
+struct ble_gap_evt_connected_t {
+  uint8_t role = BLE_GAP_ROLE_PERIPH;
+  ble_gap_addr_t peer_addr;
+};
+struct ble_gap_evt_disconnected_t { uint8_t reason = 0; };
+struct ble_gatts_evt_write_t {
+  struct { uint8_t type = 0; uint16_t uuid = 0; } uuid;
+  uint16_t len = 0;
+  uint8_t data[2] = {};
+};
+struct ble_evt_t {
+  struct { uint16_t evt_id = 0; } header;
+  struct {
+    struct { uint16_t conn_handle = 7; } common_evt;
+    struct {
+      uint16_t conn_handle = 7;
+      struct {
+        ble_gap_evt_connected_t connected;
+        ble_gap_evt_disconnected_t disconnected;
+      } params;
+    } gap_evt;
+    struct {
+      uint16_t conn_handle = 7;
+      struct { ble_gatts_evt_write_t write; } params;
+    } gatts_evt;
+  } evt;
+};
+struct BLEConnection {
+  bool connected() const { return true; }
+  bool secured() const { return true; }
+  bool bonded() const { return true; }
+};
+struct {
+  BLEConnection connection;
+  BLEConnection* Connection(uint16_t) { return &connection; }
+} Bluefruit;
+static bool cccd_written_this_connection = false;
+static struct { std::atomic<bool> valid{false}; } cccd_ram_cache;
+static void captureCccdDeferred(uint16_t, ble_gap_addr_t*) {}
+template <typename... Args> static bool ada_callback(Args...) { return true; }
+static uint32_t sd_ble_gap_disconnect(uint16_t, uint8_t) { return NRF_SUCCESS; }
+static uint32_t sd_ble_gap_conn_param_update(uint16_t, const void*) { return NRF_SUCCESS; }
+static uint32_t millis() { return 100; }
+
+struct SerialBLEInterface {
+  struct Frame {
+    uint8_t len = 0;
+    uint8_t buf[MAX_FRAME_SIZE] = {};
+    Frame& operator=(const Frame& source) {
+      assert(critical_depth);
+      len = source.len;
+      guarded_copy(buf, source.buf, sizeof(buf));
+      return *this;
+    }
+  };
+  using RxQueueGuard = TestGuard;
+  using RxQueue = mesh_nrf52::BleRxFrameQueue<Frame, FRAME_QUEUE_SIZE, RxQueueGuard>;
+  RxQueue recv_queue;
+  uint8_t send_queue_len = 0;
+  unsigned long _last_retry_attempt = 0;
+  mesh::BleTxStallWatchdog _tx_stall_watchdog;
+  mesh::BleDisconnectRecovery _tx_disconnect_recovery;
+  std::atomic<bool> _companionDataSeen{false};
+  std::atomic<bool> _pairingRequestPending{false};
+  SecuritySessionTimer _security_timer;
+  SecuritySessionTimer _companion_start_timer;
+  ble_gap_addr_t _peer_address;
+  bool _peer_address_valid = false, _bond_removed_for_connection = false;
+  bool _bonded_only = false, _isDeviceConnected = false;
+  uint16_t _conn_handle = BLE_CONN_HANDLE_INVALID;
+  unsigned removed_bonds = 0, bonded_recoveries = 0;
+  bool connected = true;
+  struct Uart {
+    std::vector<uint8_t> bytes;
+    std::function<void()> before_read, after_read, during_flush;
+    unsigned reads = 0, flushes = 0;
+    int available() const {
+      assert(!critical_depth);
+      return bytes.size();
+    }
+    int read(uint8_t* destination, size_t length) {
+      assert(!critical_depth);
+      ++reads;
+      if (before_read) {
+        auto operation = std::move(before_read);
+        before_read = nullptr;
+        operation();
+      }
+      length = std::min(length, bytes.size());
+      if (length) std::memcpy(destination, bytes.data(), length);
+      bytes.erase(bytes.begin(), bytes.begin() + length);
+      if (after_read) {
+        auto operation = std::move(after_read);
+        after_read = nullptr;
+        operation();
+      }
+      return length;
+    }
+    void flush() {
+      assert(!critical_depth);
+      ++flushes;
+      bytes.clear();
+      if (during_flush) {
+        auto operation = std::move(during_flush);
+        during_flush = nullptr;
+        operation();
+      }
+    }
+  } bleuart;
+  void open(uint16_t handle = 7) {
+    recv_queue.setEnabled(true);
+    recv_queue.connect(handle);
+    recv_queue.secure(handle);
+  }
+  bool isConnected() const { assert(!critical_depth); return connected; }
+  bool isValidConnection(uint16_t handle) const { return handle == _conn_handle; }
+  bool removeStoredBondForPeer(const char*) {
+    assert(!critical_depth && _peer_address_valid);
+    ++removed_bonds;
+    _bond_removed_for_connection = true;
+    return true;
+  }
+  void requestBondedOnlyRecovery(const char*) {
+    assert(!critical_depth);
+    ++bonded_recoveries;
+  }
+  void clearBuffers();
+  static void onBLEEvent(ble_evt_t*);
+  static void onConnect(uint16_t);
+  static void onDisconnect(uint16_t, uint8_t);
+  static void onBleUartRX(uint16_t);
+  bool isReadBusy() const;
+  bool hasPendingIO() const;
+};
+
+static SerialBLEInterface* instance;
+@METHODS@
+
+using Queue = SerialBLEInterface::RxQueue;
+using Result = Queue::PushResult;
+
+static SerialBLEInterface::Frame frame(uint32_t id, uint8_t length = MAX_FRAME_SIZE) {
+  SerialBLEInterface::Frame value;
+  value.len = length;
+  std::memset(value.buf, id % 251, sizeof(value.buf));
+  std::memcpy(value.buf, &id, sizeof(id));
+  return value;
+}
+
+static void expect_pop(Queue& queue, uint32_t id, uint8_t length = MAX_FRAME_SIZE) {
+  uint8_t destination[MAX_FRAME_SIZE] = {};
+  assert(queue.pop(destination) == length);
+  uint32_t actual;
+  std::memcpy(&actual, destination, sizeof(actual));
+  assert(actual == id);
+  for (unsigned i = sizeof(id); i < length; ++i) assert(destination[i] == id % 251);
+}
+
+static void ring_order_capacity_and_interleaved_copy() {
+  SerialBLEInterface link;
+  link.open();
+  Queue& queue = link.recv_queue;
+  uint32_t generation;
+  assert(queue.snapshot(7, generation));
+  for (unsigned i = 0; i < 12; ++i)
+    assert(queue.push(7, generation, frame(i)) == Result::Accepted);
+  assert(queue.push(7, generation, frame(12)) == Result::Full);
+  assert(queue.size() == 12 && link.isReadBusy() && link.hasPendingIO());
+
+  // Request producer execution halfway through the actual pop memcpy. Its
+  // append must run after the copy/dequeue completes, preserving the head.
+  during_copy = [&]() {
+    assert(queue.size() == 11);
+    assert(queue.push(7, generation, frame(12)) == Result::Accepted);
+  };
+  expect_pop(queue, 0);
+  for (unsigned i = 1; i <= 12; ++i) expect_pop(queue, i);
+  assert(!link.isReadBusy() && !link.hasPendingIO());
+  link.send_queue_len = 1;
+  assert(link.hasPendingIO());
+
+  // Exercise both indices wrapping through many complete rings.
+  for (uint32_t base = 100; base < 1300; base += 12) {
+    for (unsigned offset = 0; offset < 12; ++offset)
+      assert(queue.push(7, generation, frame(base + offset)) == Result::Accepted);
+    for (unsigned offset = 0; offset < 12; ++offset) expect_pop(queue, base + offset);
+  }
+
+  assert(queue.push(7, generation, frame(99)) == Result::Accepted);
+  during_copy = [&]() { link.clearBuffers(); };
+  expect_pop(queue, 99);
+  assert(queue.size() == 0 && !queue.current(7, generation));
+
+  assert(queue.snapshot(7, generation));
+  // Dequeue attempted during a producer copy cannot inspect an unpublished
+  // slot. The custom Frame assignment checks that push guards its copy too.
+  during_copy = [&]() { expect_pop(queue, 88); };
+  assert(queue.push(7, generation, frame(88)) == Result::Accepted);
+  assert(queue.size() == 0);
+}
+
+static void session_reset_and_stale_handles() {
+  SerialBLEInterface link;
+  Queue& queue = link.recv_queue;
+  uint32_t generation;
+  assert(!queue.snapshot(7, generation));
+  queue.setEnabled(true);
+  queue.connect(7);
+  assert(!queue.snapshot(7, generation));
+  queue.secure(8);
+  assert(!queue.snapshot(7, generation));
+  queue.secure(7);
+  assert(queue.snapshot(7, generation));
+  const uint32_t old = generation;
+  assert(queue.push(7, old, frame(1)) == Result::Accepted);
+  queue.reset([&]() {
+    assert(!critical_depth && queue.size() == 0);
+    assert(!queue.snapshot(7, generation));
+    assert(queue.push(7, old, frame(2)) == Result::Stale);
+    // Overlapping resets must keep RX suspended until BOTH flushes finish.
+    queue.reset([&]() { assert(!queue.snapshot(7, generation)); });
+    assert(!queue.snapshot(7, generation));
+  });
+  assert(queue.snapshot(7, generation) && generation != old);
+  assert(queue.push(7, old, frame(2)) == Result::Stale);
+  assert(queue.push(7, generation, frame(3)) == Result::Accepted);
+  queue.disconnect(8); // A stale disconnect cannot reset the active handle.
+  expect_pop(queue, 3);
+  queue.disconnect(7);
+  queue.connect(7); // The SoftDevice can reuse a connection handle.
+  queue.secure(7);
+  assert(queue.push(7, generation, frame(4)) == Result::Stale);
+  assert(queue.snapshot(7, generation));
+  queue.setEnabled(false);
+  queue.setEnabled(true);
+  queue.secure(7); // A delayed security callback must not revive that link.
+  assert(!queue.snapshot(7, generation));
+  queue.connect(7);
+  queue.secure(7);
+  assert(queue.snapshot(7, generation));
+  assert(queue.push(7, generation, frame(5, 4)) == Result::Accepted);
+  expect_pop(queue, 5, 4);
+
+  // The BLE worker records connect before enqueuing/delegating the higher
+  // level callbacks. A later enable cannot revive that earlier event.
+  queue.setEnabled(false);
+  queue.connect(7);
+  queue.setEnabled(true);
+  link.clearBuffers(); // Delayed onConnect clears buffers, not session ID.
+  queue.secure(7);
+  assert(!queue.snapshot(7, generation));
+  queue.connect(7);
+  queue.setEnabled(false);
+  queue.setEnabled(true);
+  link.clearBuffers();
+  queue.secure(7);
+  assert(!queue.snapshot(7, generation));
+  queue.connect(7);
+  queue.secure(7);
+  assert(queue.snapshot(7, generation));
+}
+
+static void callback_read_reset_and_reconnect() {
+  SerialBLEInterface link;
+  instance = &link;
+  link.open();
+  link._companion_start_timer.start(100);
+  link.bleuart.bytes = {1, 2, 3, 4};
+  SerialBLEInterface::onBleUartRX(7);
+  uint8_t destination[MAX_FRAME_SIZE] = {};
+  assert(link.recv_queue.pop(destination) == 4 && destination[3] == 4);
+  assert(link._companionDataSeen && !link._companion_start_timer.pending());
+
+  // A reset after FIFO read but before queue commit invalidates the staged
+  // frame. New-session bytes remain for their own callback.
+  link.bleuart.bytes = {10, 11, 12};
+  link.bleuart.after_read = [&]() {
+    link.recv_queue.setEnabled(false);
+    link.clearBuffers();
+    link.open();
+    link._companionDataSeen = false;
+    link._companion_start_timer.start(200);
+    link.bleuart.bytes = {20, 21};
+  };
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.size() == 0 && !link._companionDataSeen);
+  assert(link._companion_start_timer.pending() && link.bleuart.available() == 2);
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.pop(destination) == 2 && destination[0] == 20);
+
+  link.bleuart.bytes = {30, 31};
+  link.bleuart.before_read = [&]() { link.clearBuffers(); };
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.size() == 0); // Short FIFO reads never publish bytes.
+
+  link.bleuart.bytes = {40, 41};
+  const unsigned reads = link.bleuart.reads;
+  SerialBLEInterface::onBleUartRX(8);
+  assert(link.bleuart.reads == reads && link.bleuart.available() == 2);
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.pop(destination) == 2 && destination[0] == 40);
+
+  link.bleuart.bytes.assign(MAX_FRAME_SIZE + 1, 0x55);
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.size() == 0 && !link.bleuart.available());
+
+  for (unsigned i = 0; i < FRAME_QUEUE_SIZE; ++i) {
+    link.bleuart.bytes = {static_cast<uint8_t>(i)};
+    SerialBLEInterface::onBleUartRX(7);
+  }
+  link.bleuart.bytes = {99};
+  SerialBLEInterface::onBleUartRX(7);
+  assert(link.recv_queue.size() == FRAME_QUEUE_SIZE && !link.bleuart.available());
+  for (unsigned i = 0; i < FRAME_QUEUE_SIZE; ++i)
+    assert(link.recv_queue.pop(destination) == 1 && destination[0] == i);
+
+  // FIFO flush can block/yield; none of its work may retain a task guard.
+  link.bleuart.during_flush = [&]() {
+    assert(!critical_depth);
+    uint32_t token;
+    assert(!link.recv_queue.snapshot(7, token));
+    link.clearBuffers();
+    assert(!link.recv_queue.snapshot(7, token));
+  };
+  link.clearBuffers();
+  uint32_t token;
+  assert(link.recv_queue.snapshot(7, token));
+  assert(!critical_depth && deferred.empty());
+}
+
+static void production_connection_events_preserve_disconnect_cleanup() {
+  for (uint8_t reason : {uint8_t(BLE_HCI_AUTHENTICATION_FAILURE),
+                         uint8_t(BLE_HCI_STATUS_CODE_PIN_OR_KEY_MISSING),
+                         uint8_t(BLE_HCI_CONN_TERMINATED_DUE_TO_MIC_FAILURE),
+                         uint8_t(BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION)}) {
+    SerialBLEInterface link;
+    instance = &link;
+    link.recv_queue.setEnabled(true);
+    link._bonded_only = true;
+    link._bond_removed_for_connection = true;
+    cccd_written_this_connection = true;
+    ble_evt_t event;
+    event.header.evt_id = BLE_GAP_EVT_CONNECTED;
+    event.evt.gap_evt.params.connected.peer_addr.addr[0] = 42;
+    SerialBLEInterface::onBLEEvent(&event);
+    assert(link._peer_address_valid && link._peer_address.addr[0] == 42);
+    assert(!link._bond_removed_for_connection && !cccd_written_this_connection);
+    SerialBLEInterface::onConnect(7);
+    assert(link._conn_handle == 7 && link._security_timer.pending());
+    link.recv_queue.secure(7);
+    uint32_t generation;
+    assert(link.recv_queue.snapshot(7, generation));
+    assert(link.recv_queue.push(7, generation, frame(42)) == Result::Accepted);
+    link._isDeviceConnected = true;
+    link._companion_start_timer.start(100);
+    link.send_queue_len = 3;
+    cccd_written_this_connection = true;
+
+    event.header.evt_id = BLE_GAP_EVT_DISCONNECTED;
+    event.evt.gap_evt.params.disconnected.reason = reason;
+    SerialBLEInterface::onBLEEvent(&event);
+    assert(link.recv_queue.size() == 0);
+    assert(!link.recv_queue.current(7, generation));
+    assert(link.recv_queue.push(7, generation, frame(43)) == Result::Stale);
+    assert(!link._peer_address_valid && !cccd_written_this_connection);
+    const unsigned authentication_failure =
+        reason != BLE_HCI_REMOTE_USER_TERMINATED_CONNECTION;
+    assert(link.removed_bonds == authentication_failure);
+    assert(link.bonded_recoveries == authentication_failure);
+    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.send_queue_len == 0 && link.recv_queue.size() == 0);
+  }
+
+  // Run the actual event and deferred connect methods across an enable
+  // transition, so the test also verifies where session identity is assigned.
+  SerialBLEInterface link;
+  instance = &link;
+  link.recv_queue.setEnabled(true);
+  ble_evt_t event;
+  event.header.evt_id = BLE_GAP_EVT_CONNECTED;
+  SerialBLEInterface::onBLEEvent(&event);
+  link.recv_queue.setEnabled(false);
+  link.recv_queue.setEnabled(true);
+  SerialBLEInterface::onConnect(7);
+  link.recv_queue.secure(7);
+  uint32_t generation;
+  assert(!link.recv_queue.snapshot(7, generation));
+  SerialBLEInterface::onBLEEvent(&event); // A new physical session can open.
+  SerialBLEInterface::onConnect(7);
+  link.recv_queue.secure(7);
+  assert(link.recv_queue.snapshot(7, generation));
+}
+
+int main() {
+  ring_order_capacity_and_interleaved_copy();
+  session_reset_and_stale_handles();
+  callback_read_reset_and_reconnect();
+  production_connection_events_preserve_disconnect_cleanup();
+}

test/test_nrf52_ble_backpressure.py

@@ -22,6 +22,7 @@ HARNESS = r'''
 #include <vector>
 #include <helpers/CompanionFrameQueue.h>
 #include <helpers/BleTxStallWatchdog.h>
+#include <helpers/nrf52/BleRxFrameQueue.h>
 
 #define MAX_FRAME_SIZE 176
 #define FRAME_QUEUE_SIZE 12
@@ -33,8 +34,16 @@ uint32_t millis() { return current_millis; }
 
 struct SerialBLEInterface {
   struct Frame { uint8_t len = 0; uint8_t buf[MAX_FRAME_SIZE] = {}; };
-  uint8_t send_queue_len = 0, recv_queue_len = 0;
-  Frame send_queue[FRAME_QUEUE_SIZE], recv_queue[FRAME_QUEUE_SIZE];
+  struct RxQueueGuard {};
+  using RxQueue = mesh_nrf52::BleRxFrameQueue<Frame, FRAME_QUEUE_SIZE, RxQueueGuard>;
+  uint8_t send_queue_len = 0;
+  Frame send_queue[FRAME_QUEUE_SIZE];
+  RxQueue recv_queue;
+  SerialBLEInterface() {
+    recv_queue.setEnabled(true);
+    recv_queue.connect(1);
+    recv_queue.secure(1);
+  }
   unsigned long _last_retry_attempt = 0;
   mesh::BleTxStallWatchdog _tx_stall_watchdog;
   mesh::BleDisconnectRecovery _tx_disconnect_recovery;
@@ -53,12 +62,15 @@ struct SerialBLEInterface {
     return frame.len;
   }
   void receive(const uint8_t* bytes, size_t size) {
-    assert(recv_queue_len < FRAME_QUEUE_SIZE && size <= MAX_FRAME_SIZE);
-    recv_queue[recv_queue_len].len = size;
-    memcpy(recv_queue[recv_queue_len++].buf, bytes, size);
+    assert(size <= MAX_FRAME_SIZE);
+    uint32_t generation;
+    assert(recv_queue.snapshot(1, generation));
+    Frame frame;
+    frame.len = size;
+    memcpy(frame.buf, bytes, size);
+    assert(recv_queue.push(1, generation, frame) == RxQueue::PushResult::Accepted);
   }
   void shiftSendQueueLeft();
-  void shiftRecvQueueLeft();
   bool hasRequiredReplyCapacity() const;
   bool isWriteBusy() const;
   size_t writeFrame(const uint8_t*, size_t);
@@ -78,19 +90,19 @@ static void full_required_queue() {
   link.receive(command, sizeof(command));
   uint8_t received[MAX_FRAME_SIZE] = {};
   assert(link.checkRecvFrame(received) == 0);
-  assert(link.recv_queue_len == 1 && link.send_queue_len == FRAME_QUEUE_SIZE);
+  assert(link.recv_queue.size() == 1 && link.send_queue_len == FRAME_QUEUE_SIZE);
   assert(link.write_attempts == 1 && link.transmitted.empty());
   ++current_millis;
   assert(link.checkRecvFrame(received) == 0);
   assert(link.write_attempts == 1); // Existing 250 ms retry throttle is retained.
-  assert(link.recv_queue_len == 1 && link.isWriteBusy());
+  assert(link.recv_queue.size() == 1 && link.isWriteBusy());
 
   current_millis = 351;
   link.allow_write = true;
   // A completed TX frees admission on this SAME call, before RX is examined.
   assert(link.checkRecvFrame(received) == sizeof(command));
   assert(memcmp(received, command, sizeof(command)) == 0);
-  assert(link.recv_queue_len == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1);
+  assert(link.recv_queue.size() == 0 && link.send_queue_len == FRAME_QUEUE_SIZE - 1);
   const uint8_t reply[] = {29, 'A', '7', '|', 'O', 'K'};
   assert(link.writeFrame(reply, sizeof(reply)) == sizeof(reply));
   for (unsigned i = 0; i < FRAME_QUEUE_SIZE + 2; ++i) {
@@ -101,7 +113,7 @@ static void full_required_queue() {
   for (uint8_t i = 0; i < FRAME_QUEUE_SIZE; ++i)
     assert((link.transmitted[i] == std::vector<uint8_t>{0x0d, i}));
   assert(link.transmitted.back() == std::vector<uint8_t>(reply, reply + sizeof(reply)));
-  assert(link.send_queue_len == 0 && link.recv_queue_len == 0);
+  assert(link.send_queue_len == 0 && link.recv_queue.size() == 0);
 }
 
 static void best_effort_eviction() {
@@ -149,7 +161,7 @@ static void producer_threshold_and_recovery_ordering() {
   link._tx_disconnect_recovery.begin();
   uint8_t received[MAX_FRAME_SIZE] = {};
   assert(link.checkRecvFrame(received) == 0);
-  assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue_len == 1);
+  assert(link.recovery_services == 1 && link.write_attempts == 0 && link.recv_queue.size() == 1);
 
   SerialBLEInterface partial;
   partial.partial_write = true;
@@ -157,7 +169,7 @@ static void producer_threshold_and_recovery_ordering() {
   partial.receive(command, sizeof(command));
   assert(partial.checkRecvFrame(received) == 0);
   assert(partial.recoveries == 1 && partial.write_attempts == 1);
-  assert(partial.recv_queue_len == 1); // Recovery is invoked instead of dispatch.
+  assert(partial.recv_queue.size() == 1); // Recovery is invoked instead of dispatch.
 }
 
 int main() {
@@ -172,7 +184,6 @@ 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::shiftRecvQueueLeft()",
                       "size_t SerialBLEInterface::writeFrame(",
                       "size_t SerialBLEInterface::checkRecvFrame(",
                       "bool SerialBLEInterface::isWriteBusy() const"]

test/test_nrf52_ble_rx_queue.py

@@ -0,0 +1,56 @@
+#!/usr/bin/env python3
+"""Exercise the production RX ring and callback with task preemption hooks."""
+
+from pathlib import Path
+import os
+import subprocess
+import tempfile
+import unittest
+
+from test_t096_full_memory import method
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+class Nrf52BleRxQueueTests(unittest.TestCase):
+    def test_session_identity_is_observed_before_deferred_lifecycle_callbacks(self):
+        source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text()
+        events = method(source, "void SerialBLEInterface::onBLEEvent(")
+        self.assertIn("recv_queue.connect(evt->evt.gap_evt.conn_handle)", events)
+        self.assertIn("recv_queue.disconnect(evt->evt.gap_evt.conn_handle)", events)
+        for signature in ("void SerialBLEInterface::onConnect(",
+                          "void SerialBLEInterface::onDisconnect("):
+            lifecycle = method(source, signature)
+            self.assertNotIn("recv_queue.connect(", lifecycle)
+            self.assertNotIn("recv_queue.disconnect(", lifecycle)
+
+    def test_production_queue_and_callback_interleavings_under_sanitizers(self):
+        source = (ROOT / "src/helpers/nrf52/SerialBLEInterface.cpp").read_text()
+        signatures = ["void SerialBLEInterface::clearBuffers()",
+                      "static bool isBondAuthenticationFailure(",
+                      "void SerialBLEInterface::onBLEEvent(",
+                      "void SerialBLEInterface::onConnect(",
+                      "void SerialBLEInterface::onDisconnect(",
+                      "void SerialBLEInterface::onBleUartRX(",
+                      "bool SerialBLEInterface::isReadBusy() const",
+                      "bool SerialBLEInterface::hasPendingIO() const"]
+        methods = "\n".join(method(source, signature) for signature in signatures)
+        harness = (ROOT / "test/fixtures/nrf52_ble_rx_queue.cpp").read_text()
+        sanitizers = [] if os.name == "nt" else [
+            "-fsanitize=address,undefined", "-fno-sanitize-recover=all",
+            "-fno-pie", "-no-pie"]
+        with tempfile.TemporaryDirectory(prefix="meshcore-nrf-ble-rx-") as temp:
+            binary = Path(temp) / "rx_queue"
+            built = subprocess.run([
+                "c++", "-std=c++17", "-x", "c++", "-", "-I" + str(ROOT / "src"),
+                *sanitizers, "-o", str(binary)],
+                input=harness.replace("@METHODS@", methods),
+                text=True, capture_output=True, timeout=30)
+            self.assertEqual(built.returncode, 0, built.stderr)
+            result = subprocess.run([str(binary)], text=True,
+                                    capture_output=True, timeout=10)
+            self.assertEqual(result.returncode, 0, result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()