Keep BLE contact sync from blocking other companion transports

examples/companion_radio/MyMesh.cpp

@@ -369,7 +369,8 @@ void MyMesh::writeDisabledFrame() {
   _serial->writeFrame(buf, 1);
 }
 
-bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact) {
+bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact,
+                                  BaseSerialInterface* route) {
   int i = 0;
   out_frame[i++] = code;
   memcpy(&out_frame[i], contact.id.pub_key, PUB_KEY_SIZE);
@@ -389,7 +390,8 @@ bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact) {
   i += 4;
   memcpy(&out_frame[i], &contact.lastmod, 4);
   i += 4;
-  return _serial->writeFrame(out_frame, i) == (size_t)i;
+  return (route != NULL ? _serial->writeFrameToRoute(route, out_frame, i)
+                        : _serial->writeFrame(out_frame, i)) == (size_t)i;
 }
 
 void MyMesh::stopContactsIterator() {
@@ -399,7 +401,7 @@ void MyMesh::stopContactsIterator() {
   _iter_start_pending = false;
   _iter_contact_pending = false;
   _iter_next_frame_at = 0;
-  if (_serial != NULL) _serial->unlockReplyRoute();
+  _iter_reply_route = NULL;
 }
 
 static constexpr int CONTACT_UPDATE_FRAME_MIN_LEN =
@@ -2271,6 +2273,7 @@ MyMesh::MyMesh(mesh::Radio &radio, mesh::RNG &rng, mesh::RTCClock &rtc, SimpleMe
       _serial(NULL), _mota_source_control(NULL),
       telemetry(MAX_PACKET_PAYLOAD - 4), _store(&store), _ui(ui), _iter(0) {
   _iter_started = false;
+  _iter_reply_route = NULL;
   _iter_start_pending = false;
   _iter_contact_pending = false;
   _iter_next_frame_at = 0;
@@ -5181,8 +5184,8 @@ void MyMesh::serviceSigningSession() {
   }
 }
 
-void MyMesh::cancelSerialResponseStream() {
-  stopContactsIterator();
+void MyMesh::cancelSerialResponseStream(BaseSerialInterface* route) {
+  if (route == NULL || _iter_reply_route == route) stopContactsIterator();
 }
 
 void MyMesh::cancelSerialOperationsForRoute(BaseSerialInterface* route) {
@@ -5268,7 +5271,7 @@ void MyMesh::handleCmdFrame(size_t len) {
     MESH_DEBUG_PRINTLN("App %s connected", app_name);
 
     BaseSerialInterface* app_route = _serial->captureReplyRoute();
-    stopContactsIterator(); // stop any left-over ContactsIterator
+    cancelSerialResponseStream(app_route);
     cancelSerialOperationsForRoute(app_route);
     int i = 0;
     out_frame[i++] = RESP_CODE_SELF_INFO;
@@ -5535,9 +5538,10 @@ void MyMesh::handleCmdFrame(size_t len) {
         _iter_filter_since = 0;
       }
 
-      // CONTACTS_START, every CONTACT, and END_OF_CONTACTS are one response
-      // transaction. Keep them on the transport which requested the list.
-      _serial->lockReplyRoute();
+      // Retain this stream's destination independently of later commands.
+      // BLE's paced contact sync must not hold USB/WiFi command polling until
+      // the entire list finishes, or redirect their replies to the BLE app.
+      _iter_reply_route = _serial->captureReplyRoute();
       _iter = startContactsIterator();
       _iter_started = true;
       _iter_start_pending = true;
@@ -5550,7 +5554,7 @@ void MyMesh::handleCmdFrame(size_t len) {
       uint8_t reply[5];
       reply[0] = RESP_CODE_CONTACTS_START;
       memcpy(&reply[1], &_iter_total_count, 4); // total, NOT filtered count
-      if (_serial->writeFrame(reply, 5) == 5) {
+      if (_serial->writeFrameToRoute(_iter_reply_route, reply, 5) == 5) {
         _iter_start_pending = false;
         _iter_next_frame_at = futureMillis(CONTACT_STREAM_FRAME_INTERVAL_MS);
       }
@@ -10123,6 +10127,9 @@ void MyMesh::checkCLIRescueCmd() {
 
 void MyMesh::checkSerialInterface() {
   size_t len = _serial->checkRecvFrame(cmd_frame);
+  if (_iter_started && !_serial->isReplyRouteAvailable(_iter_reply_route)) {
+    stopContactsIterator();
+  }
   if (!_serial->isConnected()) {
     stopContactsIterator();
     return;
@@ -10130,11 +10137,12 @@ void MyMesh::checkSerialInterface() {
 
   if (len > 0) {
     handleCmdFrame(len);
-  } else if (_iter_started              // check if our ContactsIterator is 'running'
+  }
+  if (_iter_started              // check if our ContactsIterator is 'running'
              && (_iter_next_frame_at == 0
                  || _iter_next_frame_at == _ms->getMillis()
                  || millisHasNowPassed(_iter_next_frame_at))
-             && !_serial->isWriteBusy() // don't spam the Serial Interface too quickly!
+             && !_serial->isReplyRouteWriteBusy(_iter_reply_route)
   ) {
     if (_iter_table_revision != getContactTableRevision()) {
       // The paced stream walks the live compacting table. If an advert, app
@@ -10154,7 +10162,8 @@ void MyMesh::checkSerialInterface() {
       uint8_t reply[5];
       reply[0] = RESP_CODE_CONTACTS_START;
       memcpy(&reply[1], &_iter_total_count, 4);
-      if (_serial->writeFrame(reply, sizeof(reply)) == sizeof(reply)) {
+      if (_serial->writeFrameToRoute(_iter_reply_route, reply, sizeof(reply))
+          == sizeof(reply)) {
         _iter_start_pending = false;
         _iter_next_frame_at = futureMillis(CONTACT_STREAM_FRAME_INTERVAL_MS);
       }
@@ -10171,7 +10180,7 @@ void MyMesh::checkSerialInterface() {
         out_frame[0] = RESP_CODE_END_OF_CONTACTS;
         memcpy(&out_frame[1], &_most_recent_lastmod,
                4); // include the most recent lastmod, so app can update their 'since'
-        if (_serial->writeFrame(out_frame, 5) == 5) {
+        if (_serial->writeFrameToRoute(_iter_reply_route, out_frame, 5) == 5) {
           stopContactsIterator();
         }
         return;
@@ -10181,7 +10190,8 @@ void MyMesh::checkSerialInterface() {
     // Do not advance past a contact until its complete frame has been accepted
     // by the selected transport. A transiently full/lost route must not turn a
     // successful CONTACTS_START/END transaction into a silently short list.
-    if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact)) {
+    if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact,
+                              _iter_reply_route)) {
       if (_iter_pending_contact.lastmod > _most_recent_lastmod) {
         _most_recent_lastmod = _iter_pending_contact.lastmod;
       }
@@ -10346,7 +10356,8 @@ bool MyMesh::canRecoverUsbLogging() const {
       && !_terminal_trace_pending
 #endif
       && !(_serial && _serial->hasPendingIO())
-      && !(_iter_started && _serial && _serial->isConnected());
+      && !(_iter_started && _serial
+           && _serial->isReplyRouteAvailable(_iter_reply_route));
 }
 
 bool MyMesh::hasPendingWork() const {
@@ -10370,7 +10381,8 @@ bool MyMesh::hasPendingWork() const {
   const bool contact_write_needs_polling = isContactWriteDue();
 #endif
   return (_serial != NULL && _serial->hasPendingIO())
-      || (_iter_started && _serial != NULL && _serial->isConnected())
+      || (_iter_started && _serial != NULL
+          && _serial->isReplyRouteAvailable(_iter_reply_route))
       || hasPendingOtaApply()
       || command_radio_apply_pending
       || _scheduled_reboot_at != 0

examples/companion_radio/MyMesh.h

@@ -188,7 +188,7 @@ public:
   void activateRadio();
   bool isRadioReady() const { return _radio_available; }
   void startInterface(BaseSerialInterface &serial);
-  void cancelSerialResponseStream();
+  void cancelSerialResponseStream(BaseSerialInterface* route = NULL);
   void cancelSerialOperationsForRoute(BaseSerialInterface* route);
   bool hasFiniteDelayedReplyForRoute(BaseSerialInterface* route) const;
   void resetUsbHostSessionInput();
@@ -481,7 +481,8 @@ private:
                      BaseSerialInterface* route = nullptr);
   size_t writePendingSerialFrame(const uint8_t frame[], size_t len, uint32_t now);
   void writeDisabledFrame();
-  bool writeContactRespFrame(uint8_t code, const ContactInfo &contact);
+  bool writeContactRespFrame(uint8_t code, const ContactInfo &contact,
+                             BaseSerialInterface* route = NULL);
   void stopContactsIterator();
   bool updateContactFromFrame(ContactInfo &contact, uint32_t& last_mod, const uint8_t *frame, int len);
   bool addToOfflineQueue(const uint8_t frame[], int len);
@@ -650,6 +651,7 @@ private:
   AbstractUITask* _ui;
 
   ContactsIterator _iter;
+  BaseSerialInterface* _iter_reply_route;
   ContactInfo _iter_pending_contact;
   uint32_t _iter_filter_since;
   uint32_t _iter_next_frame_at;

examples/companion_radio/main.cpp

@@ -628,9 +628,7 @@ static void serviceCompanionPowerSaving(bool force = false) {
 
 #if defined(ETHERNET_ENABLED)
 static void cancelCompanionEthernetSession(void*) {
-  if (interface_manager.isReplyRouteFor(&ethernet_interface)) {
-    the_mesh.cancelSerialResponseStream();
-  }
+  the_mesh.cancelSerialResponseStream(&ethernet_interface);
   the_mesh.cancelSerialOperationsForRoute(&ethernet_interface);
   interface_manager.forgetReplyRouteForDisconnected(&ethernet_interface);
 }
@@ -777,12 +775,9 @@ static void cancelUsbSerialOperations() {
   // A cancelled reply must not enable diagnostics for the next USB host.
   usb_logging_reply_pending = false;
 #endif
-  // Contact enumeration uses the manager's pinned streaming route; delayed
-  // single replies capture their own route inside MyMesh. Cancel only USB's
-  // ownership so a simultaneous BLE/WiFi operation keeps running.
-  if (interface_manager.isReplyRouteFor(&usb_serial_interface)) {
-    the_mesh.cancelSerialResponseStream();
-  }
+  // Contact streams and delayed replies capture their own destinations.
+  // Cancel only USB's ownership so simultaneous BLE/WiFi work keeps running.
+  the_mesh.cancelSerialResponseStream(&usb_serial_interface);
   the_mesh.cancelSerialOperationsForRoute(&usb_serial_interface);
   interface_manager.forgetReplyRouteForDisconnected(&usb_serial_interface);
 }
@@ -2308,9 +2303,7 @@ void halt() {
   }
 
   static void cancelCompanionWiFiSession(void*) {
-    if (interface_manager.isReplyRouteFor(&wifi_interface)) {
-      the_mesh.cancelSerialResponseStream();
-    }
+    the_mesh.cancelSerialResponseStream(&wifi_interface);
     the_mesh.cancelSerialOperationsForRoute(&wifi_interface);
     interface_manager.forgetReplyRouteForDisconnected(&wifi_interface);
   }
@@ -2753,6 +2746,7 @@ static bool hasCompanionNonBluetoothClient() {
 
 static void disableCompanionBluetoothForCli() {
   companion_bluetooth_off_at = 0;
+  the_mesh.cancelSerialResponseStream(&bluetooth_interface);
   the_mesh.cancelSerialOperationsForRoute(&bluetooth_interface);
   interface_manager.disableBluetooth();
   interface_manager.forgetReplyRouteForDisconnected(&bluetooth_interface);
@@ -3422,9 +3416,7 @@ void setup() {
 // add usb interface
 #if defined(RP2040_PLATFORM) && defined(ENABLE_WIFI_INTERFACE)
   wifi_interface.setSessionChangedCallback([](void*) {
-    if (interface_manager.isReplyRouteFor(&wifi_interface)) {
-      the_mesh.cancelSerialResponseStream();
-    }
+    the_mesh.cancelSerialResponseStream(&wifi_interface);
     the_mesh.cancelSerialOperationsForRoute(&wifi_interface);
     interface_manager.forgetReplyRouteForDisconnected(&wifi_interface);
   }, nullptr);

test/test_companion_bluetooth_control.py

@@ -51,6 +51,7 @@ struct {
    bluetooth_preference=durable_bluetooth_preference=enabled;return true;
  }
  bool isAnyNetworkTerminalMode(){return network_terminal;}
+ void cancelSerialResponseStream(BaseSerialInterface*) {}
  void cancelSerialOperationsForRoute(BaseSerialInterface* route){
    assert(route==&bluetooth_interface);++cancelled;
  }

test/test_companion_contact_stream_concurrency.py

@@ -0,0 +1,280 @@
+#!/usr/bin/env python3
+"""Run production contact streaming with interleaved USB and paced BLE clients."""
+from pathlib import Path
+import os
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+from test_replay_reset_integration import extract_braced
+
+ROOT = Path(__file__).resolve().parents[1]
+
+HARNESS = r'''
+#include <algorithm>
+#include <cassert>
+#include <cstdint>
+#include <cstdio>
+#include <cstring>
+#include <deque>
+#include <limits>
+#include <vector>
+#include "helpers/ArduinoSerialInterface.h"
+#include "helpers/MultiSerialInterface.h"
+#define PUB_KEY_SIZE 32
+#define MAX_PATH_SIZE 64
+template<class... Args> static void ignoreDebug(Args&&...) {}
+#define MESH_DEBUG_PRINTLN(...) ignoreDebug(__VA_ARGS__)
+namespace StrHelper {
+static void strzcpy(void* out,const char* in,size_t size) {
+  memset(out,0,size);memcpy(out,in,std::min(size,strlen(in)));
+}
+}
+@CONSTANTS@
+@BUFFER@
+struct ContactInfo {
+  struct { uint8_t pub_key[32]{}; } id;
+  uint8_t type=1, flags=0, out_path_len=0;
+  char name[32]{};
+  uint32_t last_advert_timestamp=0, gps_lat=0, gps_lon=0, lastmod=1;
+  bool copyPathTo(uint8_t* out) const { memset(out,0,MAX_PATH_SIZE); return true; }
+};
+struct ContactsIterator {
+  unsigned next=0, count=0;
+  template<class T> bool hasNext(T*,ContactInfo& c) {
+    if(next==count) return false;
+    c=ContactInfo(); c.id.pub_key[0]=static_cast<uint8_t>(++next);
+    snprintf(c.name,sizeof(c.name),"HIL-%u",next);
+    c.lastmod=next; return true;
+  }
+};
+struct Ble : BaseSerialInterface {
+  bool enabled=false, connected=true;
+  unsigned last_write=0, capacity=4;
+  std::deque<std::vector<uint8_t>> input, pending;
+  std::vector<std::vector<uint8_t>> delivered;
+  void enable() override { enabled=true; }
+  void disable() override { enabled=false; }
+  bool isEnabled() const override { return enabled; }
+  bool isConnected() const override { return connected; }
+  bool isReadBusy() const override { return !input.empty(); }
+  bool isWriteBusy() const override { return millis()-last_write<60; }
+  size_t writeFrame(const uint8_t* src,size_t n) override {
+    if(!enabled||!connected||pending.size()>=capacity) return 0;
+    pending.emplace_back(src,src+n); return n;
+  }
+  size_t checkRecvFrame(uint8_t* dest) override {
+    if(connected&&!pending.empty()&&!isWriteBusy()) {
+      delivered.push_back(pending.front()); pending.pop_front(); last_write=millis();
+    }
+    if(!connected||input.empty()) return 0;
+    auto frame=input.front(); input.pop_front();
+    memcpy(dest,frame.data(),frame.size()); return frame.size();
+  }
+};
+struct MyMesh {
+  BaseSerialInterface* _serial=nullptr;
+  struct Clock { unsigned getMillis() const {return millis();} } clock, *_ms=&clock;
+  struct Store { bool hasIncompleteContactLoad() const {return false;} } store, *_store=&store;
+  ContactsIterator _iter;
+  BaseSerialInterface* _iter_reply_route=nullptr;
+  ContactInfo _iter_pending_contact;
+  uint32_t _iter_filter_since=0, _iter_next_frame_at=0, _iter_total_count=0;
+  uint32_t _iter_table_revision=0, _most_recent_lastmod=0;
+  bool _iter_started=false, _iter_start_pending=false, _iter_contact_pending=false;
+  unsigned contacts=54, revision=1;
+  uint8_t out_frame[MAX_FRAME_SIZE+4]{},cmd_frame[MAX_FRAME_SIZE+1]{};
+  bool millisHasNowPassed(uint32_t time) const {return static_cast<int32_t>(time-millis())<0;}
+  uint32_t futureMillis(uint32_t delta) const {return millis()+delta;}
+  uint32_t getContactTableRevision() const {return revision;}
+  unsigned getNumContacts() const {return contacts;}
+  ContactsIterator startContactsIterator() {return {0,contacts};}
+  void writeErrFrame(uint8_t error) {
+    uint8_t reply[]={RESP_CODE_ERR,error}; assert(_serial->writeFrame(reply,2)==2);
+  }
+  void cancelSerialOperationsForRoute(BaseSerialInterface*) {}
+  void stopContactsIterator();
+  void cancelSerialResponseStream(BaseSerialInterface* route=nullptr);
+  bool writeContactRespFrame(uint8_t,const ContactInfo&,BaseSerialInterface* route=nullptr);
+  void checkSerialInterface();
+  void handleCmdFrame(size_t len) {
+    @GET_CONTACTS@
+    else if(cmd_frame[0]==CMD_APP_START&&len>=8) {
+      @APP_BEGIN@
+      memset(out_frame,0xA5,70);out_frame[0]=RESP_CODE_SELF_INFO;
+      assert(_serial->writeFrame(out_frame,70)==70);
+    } else {
+      assert(cmd_frame[0]==CMD_GET_BATT_AND_STORAGE);
+      memset(out_frame,0x5A,11);out_frame[0]=RESP_CODE_BATT_AND_STORAGE;
+      assert(_serial->writeFrame(out_frame,11)==11);
+    }
+  }
+};
+@METHODS@
+struct Fixture {
+  BufferStream stream;
+  ArduinoSerialInterface usb;
+  Ble ble;
+  MultiSerialInterface manager;
+  MyMesh mesh;
+  bool drain_usb=true;
+  Fixture() {
+    g_mock_millis=100;
+    usb.begin(stream); usb.enableFlowControl(true);
+    assert(manager.addInterface(InterfaceType::Bluetooth,&ble));
+    assert(manager.addInterface(InterfaceType::USB,&usb)); manager.enable();
+    mesh._serial=&manager;
+  }
+  void step() {
+    if(drain_usb) stream.write_capacity=4096;
+    mesh.checkSerialInterface(); manager.loop(); ++g_mock_millis;
+  }
+  void run(unsigned count) {while(count--) step();}
+  void sendUsb(std::initializer_list<uint8_t> payload) {
+    const uint8_t header[]={'<',static_cast<uint8_t>(payload.size()),0};
+    stream.push(header,3); stream.push(payload.begin(),payload.size());
+  }
+  std::vector<std::vector<uint8_t>> usbFrames() const {
+    std::vector<std::vector<uint8_t>> frames;
+    for(size_t pos=0;pos<stream.output.size();) {
+      assert(stream.output[pos]=='>'&&pos+3<=stream.output.size());
+      size_t n=stream.output[pos+1]|(stream.output[pos+2]<<8);
+      assert(n>0&&pos+n+3<=stream.output.size());
+      frames.emplace_back(stream.output.begin()+pos+3,stream.output.begin()+pos+3+n);
+      pos+=n+3;
+    }
+    return frames;
+  }
+  void startBle() {ble.input.push_back({CMD_GET_CONTACTS});step();assert(mesh._iter_started);}
+  void finish() {run(6000);assert(!mesh._iter_started);}
+};
+static unsigned count(const std::vector<std::vector<uint8_t>>& frames,uint8_t type) {
+  return std::count_if(frames.begin(),frames.end(),[type](const auto& f){return f[0]==type;});
+}
+static void assertBleSnapshot(const Fixture& f,unsigned contacts=54) {
+  assert(count(f.ble.delivered,RESP_CODE_CONTACTS_START)==1);
+  assert(count(f.ble.delivered,RESP_CODE_CONTACT)==contacts);
+  assert(count(f.ble.delivered,RESP_CODE_END_OF_CONTACTS)==1);
+  for(unsigned i=1;i<=contacts;++i) {
+    auto at=std::find_if(f.ble.delivered.begin(),f.ble.delivered.end(),[i](const auto& v){return v[0]==RESP_CODE_CONTACT&&v[1]==i;});
+    assert(at!=f.ble.delivered.end());
+  }
+}
+int main() {
+  // Fifty-four BLE contacts need over three seconds, but each independent USB
+  // request is handled immediately, even when it arrives on every loop pass.
+  {
+    Fixture f;f.startBle();unsigned queries=0,apps=0;
+    for(unsigned now=0;now<4000;++now) {
+      const bool app=now%97==96;
+      if(app) {
+        f.mesh.cancelSerialResponseStream(&f.usb);
+        f.manager.forgetReplyRouteForDisconnected(&f.usb);f.usb.resetSessionState();
+        f.sendUsb({CMD_APP_START,0,0,0,0,0,0,0});++apps;
+      } else {f.sendUsb({CMD_GET_BATT_AND_STORAGE});++queries;}
+      f.step();assert(f.usbFrames().size()==queries+apps);
+    }
+    f.finish();assertBleSnapshot(f);
+    assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0);
+    assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==0);
+    assert(count(f.usbFrames(),RESP_CODE_SELF_INFO)==apps);
+    assert(count(f.usbFrames(),RESP_CODE_BATT_AND_STORAGE)==queries);
+    for(const auto& response:f.usbFrames()) {
+      assert(response.size()==(response[0]==RESP_CODE_SELF_INFO?70u:11u));
+      assert(response.back()==(response[0]==RESP_CODE_SELF_INFO?0xA5:0x5A));
+    }
+  }
+  // A second client cannot replace the sole contact iterator; its ERR remains
+  // on that client's route, while the BLE list keeps its original destination.
+  {
+    Fixture f;f.startBle();f.sendUsb({CMD_GET_CONTACTS,0xFF,0xFF,0xFF,0xFF});f.step();
+    assert(f.usbFrames().size()==1&&f.usbFrames()[0][0]==RESP_CODE_ERR);
+    assert(f.usbFrames()[0][1]==ERR_CODE_BAD_STATE);
+    f.finish();assertBleSnapshot(f);
+  }
+  // Session loss cancels only its contact stream, irrespective of whichever
+  // interface supplied the newest command. No remaining contacts leak to USB.
+  {
+    Fixture f;f.startBle();f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step();
+    f.mesh.cancelSerialResponseStream(&f.usb);assert(f.mesh._iter_started);
+    f.mesh.cancelSerialResponseStream(&f.ble);assert(!f.mesh._iter_started);
+    f.run(300);assert(count(f.usbFrames(),RESP_CODE_CONTACT)==0);
+  }
+  {
+    Fixture f;f.startBle();f.ble.connected=false;f.step();assert(!f.mesh._iter_started);
+    f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step();assert(f.usbFrames()[0][0]==RESP_CODE_BATT_AND_STORAGE);
+  }
+  // The cached contact survives a failed BLE admission and an intervening
+  // command which overwrites both shared command/output scratch buffers.
+  {
+    Fixture f;f.mesh.contacts=3;f.startBle();f.ble.capacity=0;f.run(70);
+    assert(f.mesh._iter_contact_pending&&f.mesh._iter_pending_contact.id.pub_key[0]==1);
+    f.sendUsb({CMD_GET_BATT_AND_STORAGE});f.step();
+    assert(f.usbFrames()[0].back()==0x5A&&f.mesh._iter_contact_pending);
+    f.ble.capacity=4;f.finish();assertBleSnapshot(f,3);
+  }
+  {
+    Fixture f;f.startBle();f.ble.input.push_back({CMD_APP_START,0,0,0,0,0,0,0});
+    f.step();assert(!f.mesh._iter_started);f.run(300);
+    assert(count(f.ble.delivered,RESP_CODE_SELF_INFO)==1);
+    assert(count(f.usbFrames(),RESP_CODE_SELF_INFO)==0);
+  }
+  // A backed-up USB contact stream also leaves BLE queries responsive; once
+  // capacity returns, every START/contact/END is retained on the USB route.
+  {
+    Fixture f;f.drain_usb=false;f.stream.write_capacity=0;
+    f.sendUsb({CMD_GET_CONTACTS});f.step();assert(f.mesh._iter_started);
+    f.mesh.cancelSerialResponseStream(&f.ble);assert(f.mesh._iter_started);
+    f.ble.input.push_back({CMD_GET_CONTACTS});f.step();
+    f.ble.input.push_back({CMD_GET_BATT_AND_STORAGE});f.step();f.run(100);
+    assert(count(f.ble.delivered,RESP_CODE_ERR)==1);
+    assert(count(f.ble.delivered,RESP_CODE_BATT_AND_STORAGE)==1);
+    f.drain_usb=true;f.finish();
+    assert(count(f.usbFrames(),RESP_CODE_CONTACTS_START)==1);
+    assert(count(f.usbFrames(),RESP_CODE_CONTACT)==54);
+    assert(count(f.usbFrames(),RESP_CODE_END_OF_CONTACTS)==1);
+    assert(count(f.ble.delivered,RESP_CODE_CONTACT)==0);
+  }
+}
+'''
+
+
+class ContactStreamConcurrencyTests(unittest.TestCase):
+    def test_paced_stream_keeps_routes_without_blocking_other_clients(self):
+        compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
+        if not compiler:
+            self.skipTest('a host C++17 compiler is required')
+        mesh = (ROOT / 'examples/companion_radio/MyMesh.cpp').read_text()
+        native = (ROOT / 'test/test_serial_mode_switch/test_serial_mode_switch.cpp').read_text()
+        constants = '\n'.join(re.findall(
+            r'^#define (?:CONTACT_STREAM_FRAME_INTERVAL_MS|CMD_GET_CONTACTS|CMD_APP_START|CMD_GET_BATT_AND_STORAGE|RESP_CODE_ERR|RESP_CODE_CONTACTS_START|RESP_CODE_CONTACT|RESP_CODE_END_OF_CONTACTS|RESP_CODE_SELF_INFO|RESP_CODE_BATT_AND_STORAGE|ERR_CODE_BAD_STATE|ERR_CODE_FILE_IO_ERROR)\s+.+$',
+            mesh, re.MULTILINE))
+        contacts = extract_braced(mesh, '} else if (cmd_frame[0] == CMD_GET_CONTACTS').replace('} else if', 'if', 1)
+        app = extract_braced(mesh, '} else if (cmd_frame[0] == CMD_APP_START')
+        app = app[app.index('{') + 1:app.index('    int i = 0;')]
+        methods = '\n'.join(extract_braced(mesh, sig) for sig in (
+            'bool MyMesh::writeContactRespFrame(', 'void MyMesh::stopContactsIterator(',
+            'void MyMesh::cancelSerialResponseStream(', 'void MyMesh::checkSerialInterface()',
+        ))
+        program = HARNESS.replace('@CONSTANTS@', constants).replace('@BUFFER@', extract_braced(native, 'class BufferStream') + ';')
+        program = program.replace('@GET_CONTACTS@', contacts).replace('@APP_BEGIN@', app).replace('@METHODS@', methods)
+        sanitizers = ['-fsanitize=address,undefined', '-fno-sanitize-recover=all', '-fno-pie', '-no-pie'] if sys.platform.startswith('linux') else []
+        with tempfile.TemporaryDirectory(prefix='meshcore-contact-concurrency-') as directory:
+            binary = Path(directory) / 'concurrency'
+            result = subprocess.run([
+                compiler, '-std=c++17', '-Wall', '-Wextra', '-Werror',
+                '-Wno-unused-parameter', '-Wno-sign-compare',
+                *sanitizers,
+                f'-I{ROOT / "test/mocks"}', f'-I{ROOT / "src"}', '-x', 'c++', '-',
+                str(ROOT / 'src/helpers/ArduinoSerialInterface.cpp'), '-o', str(binary),
+            ], input=program, text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            result = subprocess.run([str(binary)], text=True, capture_output=True)
+            self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
+
+
+if __name__ == '__main__':
+    unittest.main()

test/test_companion_contact_stream_contract.py

@@ -30,18 +30,16 @@ class ContactStreamContractTests(unittest.TestCase):
     def test_contact_frame_write_result_is_observable(self):
         header = source("examples/companion_radio/MyMesh.h")
         impl = source("examples/companion_radio/MyMesh.cpp")
-        self.assertIn(
-            "bool writeContactRespFrame(uint8_t code, const ContactInfo &contact);",
-            header,
-        )
+        self.assertRegex(header, r"bool writeContactRespFrame\(uint8_t code,\s*"
+                         r"const ContactInfo &contact,\s*"
+                         r"BaseSerialInterface\* route = NULL\);")
         body = function_body(
             impl,
-            "bool MyMesh::writeContactRespFrame(uint8_t code, const ContactInfo &contact)",
-        )
-        self.assertRegex(
-            body,
-            r"return\s+_serial->writeFrame\(out_frame,\s*i\)\s*==\s*\(size_t\)i\s*;",
+            "bool MyMesh::writeContactRespFrame(",
         )
+        self.assertIn("_serial->writeFrameToRoute(route, out_frame, i)", body)
+        self.assertIn("_serial->writeFrame(out_frame, i)", body)
+        self.assertRegex(body, r"\)\s*==\s*\(size_t\)i\s*;")
 
     def test_iterator_retries_start_contact_and_end_frames(self):
         impl = source("examples/companion_radio/MyMesh.cpp")
@@ -52,17 +50,18 @@ class ContactStreamContractTests(unittest.TestCase):
         self.assertRegex(
             body,
             r"if\s*\(writeContactRespFrame\(RESP_CODE_CONTACT,\s*"
-            r"_iter_pending_contact\)\)",
+            r"_iter_pending_contact,\s*_iter_reply_route\)\)",
         )
         self.assertRegex(
             body,
-            r"if\s*\(_serial->writeFrame\(out_frame,\s*5\)\s*==\s*5\)\s*\{\s*"
+            r"if\s*\(_serial->writeFrameToRoute\(_iter_reply_route,\s*"
+            r"out_frame,\s*5\)\s*==\s*5\)\s*\{\s*"
             r"stopContactsIterator\(\)\s*;",
         )
 
         # The cached contact is cleared only inside the successful-write arm.
         success = body.index(
-            "if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact))"
+            "if (writeContactRespFrame(RESP_CODE_CONTACT, _iter_pending_contact,"
         )
         clear = body.index("_iter_contact_pending = false;", success)
         self.assertGreater(clear, success)

test/test_companion_radio_settings_transaction.py

@@ -238,6 +238,9 @@ static uint32_t now_ms=100;
 struct Serial {
   bool hasPendingIO() const { return false; }
   bool isConnected() const { return false; }
+  bool isReplyRouteAvailable(const Serial* route) const {
+    return route==this&&isConnected();
+  }
 };
 struct {
   bool isWatchdogObserving() const { return false; }
@@ -253,6 +256,7 @@ struct MyMesh {
   const void *sign_data=nullptr;
   const void *emergency_client_repeat_packet=nullptr;
   Serial* _serial=nullptr;
+  Serial* _iter_reply_route=nullptr;
 #if COMPANION_FEATURE_TEMP_RADIO
   bool _temp_radio_applied=false;
   uint32_t _temp_radio_set_at=0, _temp_radio_revert_at=0, _temp_radio_retry_at=0;

test/test_companion_usb_logging_reply.py

@@ -96,7 +96,9 @@ struct MyMesh {
   uint8_t cmd_frame[MAX_FRAME_SIZE+1]{}, out_frame[MAX_FRAME_SIZE+4]{};
   char reply_buf[166]{};
   bool isTerminalMode() const { return terminal; }
-  void cancelSerialResponseStream() { ++cancellations; }
+  void cancelSerialResponseStream(BaseSerialInterface* route) {
+    if (interface_manager.isReplyRouteFor(route)) ++cancellations;
+  }
   void cancelSerialOperationsForRoute(BaseSerialInterface*) {}
   void enterTerminalMode(bool show_banner) {
     terminal=true;

test/test_companion_wifi_session.py

@@ -58,7 +58,8 @@ struct MyMesh {
   uint8_t private_key_backup_sender[6] = {};
   struct Ack { BaseSerialInterface* reply_route; bool radio_retry; };
   Ack expected_ack_table[EXPECTED_ACK_TABLE_SIZE] = {};
-  void cancelSerialResponseStream() {
+  void cancelSerialResponseStream(BaseSerialInterface* route) {
+    if (!interface_manager.isReplyRouteFor(route)) return;
     ++stream_cancels;
     if (streaming) { streaming = false; interface_manager.unlockReplyRoute(); }
   }

test/test_nrf52_usb_logging_contract.py

@@ -520,11 +520,8 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase):
             main.index("static void cancelUsbSerialOperations()"):
             main.index("static void enterUsbTerminalMode(")
         ]
-        route_check = usb_cancel.index(
-            "interface_manager.isReplyRouteFor(&usb_serial_interface)"
-        )
         route_cancel = usb_cancel.index(
-            "the_mesh.cancelSerialResponseStream()"
+            "the_mesh.cancelSerialResponseStream(&usb_serial_interface)"
         )
         delayed_cancel = usb_cancel.index(
             "the_mesh.cancelSerialOperationsForRoute(&usb_serial_interface)"
@@ -532,7 +529,6 @@ class Nrf52UsbLoggingContractTest(unittest.TestCase):
         route_forget = usb_cancel.index(
             "interface_manager.forgetReplyRouteForDisconnected("
         )
-        self.assertLess(route_check, route_cancel)
         self.assertLess(route_cancel, delayed_cancel)
         self.assertLess(delayed_cancel, route_forget)
         self.assertIn("the_mesh.resetUsbHostSessionInput();", reset_helper)

test/test_usb_logging_watchdog_roles.py

@@ -39,6 +39,9 @@ struct SerialInterface {
  bool pending=false, connected=false, passthrough=true;
  bool hasPendingIO() const{return pending;}
  bool isConnected() const{return connected;}
+ bool isReplyRouteAvailable(const SerialInterface* route) const {
+   return route==this&&connected;
+ }
  bool isPassthroughMode() const{return passthrough;}
 };
 struct RoleMesh {
@@ -54,6 +57,7 @@ struct RoleMesh {
  Store* _store=&store;
  SerialInterface serial;
  SerialInterface* _serial=&serial;
+ SerialInterface* _iter_reply_route=&serial;
  bool hasOutbound() const{return outbound;}
  bool isAnyTempRadioActive() const{return temporary;}
  bool hasPendingOtaApply() const{return ota_apply;}
@@ -84,6 +88,7 @@ void reset(){
  // Assignment copies the fixture's dependency pointers; point them back at
  // the live fixture objects, not the temporary used to reset scalar state.
  the_mesh._store=&the_mesh.store;the_mesh._serial=&the_mesh.serial;
+ the_mesh._iter_reply_route=&the_mesh.serial;
  command[0]=0;command_overflow=false;
 #if defined(ENABLE_USB_INTERFACE)
  usb_serial_interface=SerialInterface{};usb_connected=false;