Prevent pipelined clients from starving Companion transports

src/helpers/MultiSerialInterface.h

@@ -28,6 +28,7 @@ private:
 
   bool _enabled = false;
   bool _bluetooth_auto_enable = true;
+  uint8_t _nextRxIndex = 0;
   RegisteredInterface _interfaces[MAX_INTERFACES] = {};
   BaseSerialInterface* _lastRxInterface = nullptr;
   BaseSerialInterface* _lockedReplyInterface = nullptr;
@@ -144,6 +145,7 @@ public:
   // enable all interfaces
   void enable() override {
     _enabled = true;
+    _nextRxIndex = 0;
     _lastRxInterface = nullptr;
     _lockedReplyInterface = nullptr;
     for(auto iface : _interfaces){
@@ -157,6 +159,7 @@ public:
   // disable all interfaces
   void disable() override {
     _enabled = false;
+    _nextRxIndex = 0;
     _lastRxInterface = nullptr;
     _lockedReplyInterface = nullptr;
     for(auto iface : _interfaces){
@@ -364,11 +367,15 @@ public:
       _lastRxInterface = nullptr;
     }
 
-    // try to read a frame from any enabled interface
-    for(auto iface : _interfaces){
+    // Start after the last command so a pipelined client cannot starve the
+    // other transports. Locked response streams retain their exclusive poll.
+    for (size_t offset = 0; offset < MAX_INTERFACES; ++offset) {
+      const size_t index = (_nextRxIndex + offset) % MAX_INTERFACES;
+      auto iface = _interfaces[index];
       if(iface.instance && iface.instance->isEnabled()){
         size_t frameSize = iface.instance->checkRecvFrame(dest);
         if(frameSize > 0){
+          _nextRxIndex = (index + 1) % MAX_INTERFACES;
           _lastRxInterface = iface.instance;
           return frameSize; 
         }

test/fixtures/companion_wifi_listener/test.cpp

@@ -4,6 +4,8 @@
 #include <cassert>
 #include <cstdio>
 #include <cstdlib>
+#include <cstring>
+#include <deque>
 #include <limits>
 #include <vector>
 
@@ -24,6 +26,31 @@ static Bytes framed(std::initializer_list<Bytes> frames) {
   return result;
 }
 
+struct RoutingPeer : BaseSerialInterface {
+  bool enabled = false, connected = true;
+  unsigned polls = 0;
+  std::deque<Bytes> input;
+  std::vector<Bytes> output;
+  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 false; }
+  size_t writeFrame(const uint8_t* src, size_t length) override {
+    if (!enabled || !connected) return 0;
+    output.emplace_back(src, src + length);
+    return length;
+  }
+  size_t checkRecvFrame(uint8_t* dest) override {
+    ++polls;
+    if (!connected || input.empty()) return 0;
+    const Bytes frame = input.front(); input.pop_front();
+    memcpy(dest, frame.data(), frame.size());
+    return frame.size();
+  }
+};
+
 struct Fixture {
   SerialWifiInterface transport;
   uint8_t input[MAX_FRAME_SIZE] = {};
@@ -601,6 +628,152 @@ int main(int argc, char** argv) {
     f.transport.loop();
     assert(!socket->connected && f.cancellations == 1);
     assert(f.transport.getStatus().lastDisconnectReason == SerialWifiInterface::DisconnectReason::FrameTimeout);
+  } else if (scenario == 52) {
+    assert(f.start());
+    auto socket = f.connect(A);
+    socket->write_limit = std::numeric_limits<size_t>::max();
+    RoutingPeer usb;
+    MultiSerialInterface manager;
+    assert(manager.addInterface(InterfaceType::WiFi, &f.transport));
+    assert(manager.addInterface(InterfaceType::USB, &usb)); manager.enable();
+    const Bytes wifi_reply{0x0d, 0xa1}, usb_reply{0x0d, 0xb1};
+    unsigned wifi_commands = 0;
+    for (unsigned poll = 0; poll < 10000; ++poll) {
+      const uint8_t query[] = {'<', 2, 0, 0x16, 3};
+      for (uint8_t byte : query) socket->received.push_back(byte);
+      if (poll % 100 == 0) usb.input.push_back(Bytes{0x16, 4});
+      ++mock_millis; manager.loop();
+      assert(manager.checkRecvFrame(f.input) == 2 && f.input[0] == 0x16);
+      auto* route = manager.captureReplyRoute();
+      assert(route == (f.input[1] == 3 ? static_cast<BaseSerialInterface*>(&f.transport) : &usb));
+      const Bytes& reply = route == &usb ? usb_reply : wifi_reply;
+      if (route == &f.transport) ++wifi_commands;
+      assert(manager.writeFrame(reply.data(), reply.size()) == reply.size());
+      assert((poll % 100 < 1 || usb.input.empty())
+          && "an unlocked pipelined WiFi client must not starve a ready USB command");
+    }
+    // Finish the finite WiFi pipeline and the last accepted response.
+    for (unsigned poll = 0; poll < 102; ++poll) {
+      const size_t length = manager.checkRecvFrame(f.input);
+      if (length == 0) break;
+      assert(length == 2 && manager.captureReplyRoute() == &f.transport);
+      ++wifi_commands;
+      assert(manager.writeFrame(wifi_reply.data(), wifi_reply.size()) == wifi_reply.size());
+    }
+    assert(socket->received.empty() && wifi_commands == 10000);
+    assert(f.transport.getStatus().txFrames == wifi_commands);
+    Bytes expected;
+    const Bytes wire_reply = framed({wifi_reply});
+    for (unsigned i = 0; i < wifi_commands; ++i)
+      expected.insert(expected.end(), wire_reply.begin(), wire_reply.end());
+    assert(socket->sent == expected && usb.output == std::vector<Bytes>(100, usb_reply));
+  } else if (scenario == 53) {
+    assert(f.start());
+    auto socket = f.connect(A);
+    socket->write_limit = std::numeric_limits<size_t>::max();
+    RoutingPeer usb, ble, ethernet;
+    MultiSerialInterface manager;
+    BaseSerialInterface* routes[] = {&f.transport, &usb, &ble, &ethernet};
+    RoutingPeer* peers[] = {&usb, &ble, &ethernet};
+    const InterfaceType types[] = {InterfaceType::WiFi, InterfaceType::USB,
+                                  InterfaceType::Bluetooth, InterfaceType::Ethernet};
+    for (unsigned i = 0; i < 4; ++i) assert(manager.addInterface(types[i], routes[i]));
+    manager.enable();
+    socket->received = {'<', 2, 0, 0x16, 0};
+    for (unsigned i = 0; i < 3; ++i) peers[i]->input.push_back(Bytes{0x16, uint8_t(i + 1)});
+    unsigned counts[4] = {};
+    BaseSerialInterface* captured_wifi = nullptr;
+    for (unsigned poll = 0; poll < 16000; ++poll) {
+      ++mock_millis; manager.loop();
+      assert(manager.checkRecvFrame(f.input) == 2);
+      const unsigned index = poll % 4;
+      assert(manager.captureReplyRoute() == routes[index]
+          && "each unlocked ready transport must be served within one registration cycle");
+      assert(f.input[0] == 0x16 && f.input[1] == index);
+      ++counts[index];
+      const uint8_t reply[] = {0x0d, uint8_t(index)};
+      assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+      if (index == 0) {
+        captured_wifi = manager.captureReplyRoute();
+        if (counts[index] < 4000) socket->received = {'<', 2, 0, 0x16, 0};
+      } else if (counts[index] < 4000) {
+        peers[index - 1]->input.push_back(Bytes{0x16, uint8_t(index)});
+      }
+    }
+    assert(manager.checkRecvFrame(f.input) == 0); // Drain WiFi's final response.
+    for (unsigned count : counts) assert(count == 4000);
+    for (unsigned i = 0; i < 3; ++i)
+      assert(peers[i]->output == std::vector<Bytes>(4000, Bytes{0x0d, uint8_t(i + 1)}));
+    Bytes expected;
+    for (unsigned i = 0; i < 4000; ++i) {
+      const Bytes frame = framed({Bytes{0x0d, 0}});
+      expected.insert(expected.end(), frame.begin(), frame.end());
+    }
+    assert(socket->sent == expected && f.transport.getStatus().txFrames == 4000);
+    // A captured delayed route is independent of the most recent requester.
+    assert(manager.captureReplyRoute() == &ethernet && captured_wifi == &f.transport);
+    const uint8_t delayed[] = {0x85, 0x41};
+    assert(manager.writeFrameToRoute(captured_wifi, delayed, sizeof(delayed)) == sizeof(delayed));
+    assert(manager.checkRecvFrame(f.input) == 0);
+    const Bytes last = framed({Bytes{0x85, 0x41}});
+    expected.insert(expected.end(), last.begin(), last.end());
+    assert(socket->sent == expected && ethernet.output.size() == 4000);
+  } else if (scenario == 54) {
+    assert(f.start());
+    auto socket = f.connect(A, Bytes{'<', 2, 0, 0x16, 3});
+    RoutingPeer usb, ble;
+    MultiSerialInterface manager;
+    assert(manager.addInterface(InterfaceType::WiFi, &f.transport));
+    assert(manager.addInterface(InterfaceType::USB, &usb));
+    assert(manager.addInterface(InterfaceType::Bluetooth, &ble)); manager.enable();
+    socket->received = {'<', 2, 0, 0x16, 3};
+    assert(manager.checkRecvFrame(f.input) == 2); manager.lockReplyRoute();
+    usb.input.push_back(Bytes{0x16, 4}); ble.input.push_back(Bytes{0x16, 5});
+    const uint8_t reply[] = {0x03, 0x41};
+    for (unsigned poll = 0; poll < 8; ++poll) {
+      socket->received = {'<', 2, 0, 0x16, 3};
+      assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &f.transport);
+      assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+      socket->write_limit = std::numeric_limits<size_t>::max();
+    }
+    assert(manager.checkRecvFrame(f.input) == 0 && manager.captureReplyRoute() == &f.transport);
+    assert(usb.polls == 0 && ble.polls == 0 && usb.output.empty() && ble.output.empty());
+    socket->connected = false; manager.loop();
+    assert(!manager.isConnected() && manager.checkRecvFrame(f.input) == 0);
+    assert(manager.writeFrame(reply, sizeof(reply)) == 0);
+    assert(usb.polls == 0 && ble.polls == 0);
+    // The response producer owns cancellation; forgetting its disconnected
+    // route releases the lock without leaking the abandoned response.
+    manager.forgetReplyRouteForDisconnected(&f.transport);
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &usb);
+    assert(manager.writeFrame(reply, sizeof(reply)) == sizeof(reply));
+    assert((usb.output == std::vector<Bytes>{Bytes{0x03, 0x41}} && ble.output.empty()));
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &ble);
+  } else if (scenario == 55) {
+    RoutingPeer first, removed, third, fourth, replacement;
+    RoutingPeer* peers[] = {&first, &removed, &third, &fourth};
+    MultiSerialInterface manager;
+    for (auto* peer : peers) assert(manager.addInterface(InterfaceType::USB, peer));
+    manager.enable();
+    for (auto* peer : peers) peer->input.push_back(Bytes{0x16, 3});
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first);
+    assert(manager.removeInterface(&removed));
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &third);
+    assert(manager.addInterface(InterfaceType::Ethernet, &replacement)); replacement.enable();
+    replacement.input.push_back(Bytes{0x16, 3}); first.input.push_back(Bytes{0x16, 3});
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &fourth);
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first);
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &replacement);
+    third.input.push_back(Bytes{0x16, 3}); third.disable();
+    fourth.input.push_back(Bytes{0x16, 3});
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &fourth);
+    assert(third.input.size() == 1 && removed.input.size() == 1);
+    first.input.push_back(Bytes{0x16, 3}); manager.disable();
+    assert(manager.checkRecvFrame(f.input) == 0 && manager.captureReplyRoute() == nullptr);
+    assert(first.input.size() == 1);
+    manager.enable();
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &first);
+    assert(manager.checkRecvFrame(f.input) == 2 && manager.captureReplyRoute() == &third);
   } else {
     assert(false && "unknown scenario");
   }

test/test_companion_wifi_listener.py

@@ -73,10 +73,51 @@ SCENARIOS = (
     "reply-capacity pause preserves the remaining inbound timeout budget",
     "new queue pressure cannot retroactively forgive an expired inbound frame",
     "reply-capacity timeout pause survives millis rollover",
+    "pipelined WiFi commands cannot starve a ready USB command or steal its replies",
+    "four continuously ready transports each receive bounded service and preserve captured replies",
+    "locked WiFi owner remains exclusive through idle and disconnect until route cleanup",
+    "round-robin skips removed and disabled slots and restarts registration order on enable",
 )
 
 
 class CompanionWiFiListenerTests(unittest.TestCase):
+    def test_fixed_priority_negative_control_reproduces_cross_transport_starvation(self):
+        compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
+        if compiler is None:
+            self.skipTest("a host C++17 compiler is required")
+        header = (ROOT / "src/helpers/MultiSerialInterface.h").read_text(encoding="ascii")
+        rotating_index = "const size_t index = (_nextRxIndex + offset) % MAX_INTERFACES;"
+        self.assertEqual(header.count(rotating_index), 1)
+        # Retain all routing, locking, and backend code. Only restore the old
+        # fixed-priority scan, which always started at registration slot zero.
+        legacy = header.replace(rotating_index, "const size_t index = offset;")
+        with tempfile.TemporaryDirectory(prefix="meshcore-wifi-priority-") as directory:
+            work = Path(directory)
+            (work / "helpers").mkdir()
+            (work / "helpers/MultiSerialInterface.h").write_text(legacy, encoding="ascii")
+            binary = work / "fixed-priority.exe"
+            compiled = subprocess.run(
+                [compiler, "-std=c++17", "-Wall", "-Wextra", "-Werror",
+                 *SANITIZERS, "-DESP32=1", f"-I{work}", f"-I{MOCKS}",
+                 f"-I{ROOT / 'src'}", f"-I{ROOT / 'src/helpers'}",
+                 str(FIXTURE), str(ROOT / "src/helpers/wifi/SerialWifiInterface.cpp"),
+                 "-o", str(binary)], capture_output=True, text=True, timeout=60,
+            )
+            self.assertEqual(compiled.returncode, 0, compiled.stdout + compiled.stderr)
+            for scenario, oracle in (
+                    (52, "an unlocked pipelined WiFi client must not starve a ready USB command"),
+                    (53, "each unlocked ready transport must be served within one registration cycle")):
+                with self.subTest(scenario=scenario):
+                    checked = subprocess.run([str(binary), str(scenario)], capture_output=True,
+                                             text=True, timeout=10)
+                    self.assertNotEqual(checked.returncode, 0)
+                    self.assertIn(oracle, checked.stderr)
+            # Exclusive response streams are an intentional policy and were
+            # already correct; the old polling order must still pass this.
+            checked = subprocess.run([str(binary), "54"], capture_output=True, text=True, timeout=10)
+            self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr)
+            self.assertIn("PASS: Wi-Fi listener scenario 54", checked.stdout)
+
     def test_production_listener_faults(self):
         compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++")
         if compiler is None: