diff --git a/.github/workflows/run-unit-tests.yml b/.github/workflows/run-unit-tests.yml index 79aed4f28..14ab4ddda 100644 --- a/.github/workflows/run-unit-tests.yml +++ b/.github/workflows/run-unit-tests.yml @@ -565,6 +565,7 @@ jobs: run: | python3 -B test/test_companion_usb_default.py python3 -B test/test_companion_usb_input.py + python3 -B test/test_arduino_serial_backpressure.py -v python3 -B test/test_companion_ota_includes.py python3 -B test/test_companion_usb_logging_handoff.py python3 -B test/test_companion_usb_mota_exit.py diff --git a/src/helpers/ArduinoSerialInterface.cpp b/src/helpers/ArduinoSerialInterface.cpp index 92d3d17f9..4e54a42a6 100644 --- a/src/helpers/ArduinoSerialInterface.cpp +++ b/src/helpers/ArduinoSerialInterface.cpp @@ -21,10 +21,14 @@ void ArduinoSerialInterface::resetReceiveState() { _frame_len = 0; rx_len = 0; _last_rx_byte_ms = 0; + _receive_capacity_paused = false; + _receive_capacity_paused_at = 0; } void ArduinoSerialInterface::serviceReceiveTimeout() { + updateReceiveCapacityPause(millis()); if (_state != RECV_STATE_IDLE && _serial != nullptr + && !_receive_capacity_paused && _serial->available() == 0 && (uint32_t)(millis() - _last_rx_byte_ms) >= RX_FRAME_TIMEOUT_MS) { // A truncated length-prefixed frame must not hold the MCU awake forever or @@ -33,6 +37,40 @@ void ArduinoSerialInterface::serviceReceiveTimeout() { } } +bool ArduinoSerialInterface::responseCapacityAvailable() const { + if (!_flow_ctl || _tx_queue_len < TX_QUEUE_SIZE) return true; + // Match enqueueFrame(): a required reply may replace an unsent best-effort + // push, but the prefix of a partly sent frame must remain intact. + for (uint8_t i = _tx_offset == 0 ? 0 : 1; i < _tx_queue_len; ++i) { + const TxFrame& frame = _tx_queue[i]; + if (!mesh::companionFrameRequiresDelivery(&frame.buf[3], frame.len - 3)) { + return true; + } + } + return false; +} + +void ArduinoSerialInterface::updateReceiveCapacityPause(uint32_t now) { + if (_state == RECV_STATE_IDLE) { + _receive_capacity_paused = false; + _receive_capacity_paused_at = 0; + return; + } + const bool pause = !responseCapacityAvailable(); + if (pause && !_receive_capacity_paused) { + // Backpressure must not retroactively extend an already expired frame. + if ((uint32_t)(now - _last_rx_byte_ms) >= RX_FRAME_TIMEOUT_MS) return; + _receive_capacity_paused = true; + _receive_capacity_paused_at = now; + } else if (!pause && _receive_capacity_paused) { + // Bytes retained for reply capacity keep the framing budget they had when + // RX was paused, even if the host takes a long time to resume reading. + _last_rx_byte_ms += (uint32_t)(now - _receive_capacity_paused_at); + _receive_capacity_paused = false; + _receive_capacity_paused_at = 0; + } +} + void ArduinoSerialInterface::resetTransmitState() { _tx_queue_len = 0; _tx_offset = 0; @@ -82,6 +120,7 @@ bool ArduinoSerialInterface::enqueueFrame(const uint8_t src[], size_t len) { frame.buf[1] = (uint8_t)(len & 0xFF); frame.buf[2] = (uint8_t)(len >> 8); memcpy(&frame.buf[3], src, len); + updateReceiveCapacityPause(millis()); return true; } @@ -204,6 +243,7 @@ bool ArduinoSerialInterface::isConnected() const { void ArduinoSerialInterface::loop() { serviceReceiveTimeout(); serviceTransmit(); + updateReceiveCapacityPause(millis()); } bool ArduinoSerialInterface::isReadBusy() const { @@ -240,6 +280,7 @@ size_t ArduinoSerialInterface::writeFrame(const uint8_t src[], size_t len) { } if (!enqueueFrame(src, len)) return 0; serviceTransmit(); + updateReceiveCapacityPause(millis()); return len; } @@ -255,7 +296,12 @@ size_t ArduinoSerialInterface::checkRecvFrame(uint8_t dest[]) { if (_serial == nullptr || dest == nullptr) return 0; serviceReceiveTimeout(); serviceTransmit(); + updateReceiveCapacityPause(millis()); if (_passthroughMode) return 0; + // Immediate command handlers do not retain rejected replies. Leave the + // complete input and any partial parser state untouched until one required + // response can be admitted, rather than executing a command without a reply. + if (!responseCapacityAvailable()) return 0; // Newly arriving bytes belong to the next mesh loop; a continuously writing // host must not keep the radio/watchdog loop trapped in this parser. diff --git a/src/helpers/ArduinoSerialInterface.h b/src/helpers/ArduinoSerialInterface.h index 1492234e1..532956735 100644 --- a/src/helpers/ArduinoSerialInterface.h +++ b/src/helpers/ArduinoSerialInterface.h @@ -38,6 +38,8 @@ private: uint32_t _last_frame_ms; uint32_t _completed_frame_count; uint32_t _last_rx_byte_ms; + uint32_t _receive_capacity_paused_at; + bool _receive_capacity_paused; bool _has_received_frame; Stream* _serial; const char* _controlSequence; @@ -51,6 +53,8 @@ private: bool checkControlLineByte(uint8_t c); void resetReceiveState(); void serviceReceiveTimeout(); + bool responseCapacityAvailable() const; + void updateReceiveCapacityPause(uint32_t now); void resetTransmitState(); bool enqueueFrame(const uint8_t src[], size_t len); void serviceTransmit(); @@ -63,6 +67,7 @@ public: _controlSequenceCursor(nullptr), _secondaryControlSequenceCursor(nullptr), _frame_len(0), rx_len(0), _last_frame_ms(0), _completed_frame_count(0), _last_rx_byte_ms(0), + _receive_capacity_paused_at(0), _receive_capacity_paused(false), _has_received_frame(false), _serial(nullptr), _controlSequence(nullptr), _secondaryControlSequence(nullptr), _conn_check(nullptr), @@ -114,11 +119,13 @@ public: // Optional: queue complete frames and drain one at a time according to the // stream's TX capacity, so short writes cannot discard or interleave bytes. // Report busy while a frame remains queued so bulk streams get paced. + // Defer input, including control lines, while a required reply cannot fit. // Only enable this for streams which really implement availableForWrite() // (USB-CDC does, the Print default returns 0). void enableFlowControl(bool enable) { if (!enable) resetTransmitState(); _flow_ctl = enable; + updateReceiveCapacityPause(millis()); } // BaseSerialInterface methods diff --git a/test/fixtures/arduino_serial_backpressure/test.cpp b/test/fixtures/arduino_serial_backpressure/test.cpp new file mode 100644 index 000000000..082cc12d2 --- /dev/null +++ b/test/fixtures/arduino_serial_backpressure/test.cpp @@ -0,0 +1,364 @@ +#include +#include + +#include +#include +#include +#include +#include +#include + +#define REQUIRE(condition) do { \ + if (!(condition)) { \ + std::fprintf(stderr, "FAIL line %d: %s\n", __LINE__, #condition); \ + std::exit(1); \ + } \ +} while (0) + +using Bytes = std::vector; +static const char START_TOKEN[] = "+++MESHCORE-TERM-START"; +static const char MOTA_TOKEN[] = "ota folder on"; + +class BufferStream : public Stream { +public: + std::deque input; + Bytes output; + int capacity = 0; + size_t max_write = std::numeric_limits::max(); + unsigned reads = 0; + + void push(const Bytes& bytes) { input.insert(input.end(), bytes.begin(), bytes.end()); } + void push(const char* text) { while (*text) input.push_back(uint8_t(*text++)); } + int available() override { return static_cast(input.size()); } + int availableForWrite() override { return capacity; } + int peek() override { return input.empty() ? -1 : input.front(); } + int read() override { + if (input.empty()) return -1; + const uint8_t value = input.front(); + input.pop_front(); + ++reads; + return value; + } + size_t write(uint8_t value) override { return write(&value, 1); } + size_t write(const uint8_t* data, size_t len) override { + const size_t written = std::min(std::min(len, max_write), size_t(std::max(capacity, 0))); + output.insert(output.end(), data, data + written); + capacity -= int(written); + return written; + } +}; + +#define RESP_CODE_OK 0 +#define RESP_CODE_ERR 1 +struct MyMesh { + BaseSerialInterface* _serial; + void writeOKFrame(BaseSerialInterface* route = nullptr); + void writeErrFrame(uint8_t code, BaseSerialInterface* route = nullptr); +}; +// The runner extracts these immediate reply methods from production MyMesh.cpp. +#include "immediate_reply_methods.inc" + +static Bytes framed(const std::vector& frames) { + Bytes result; + for (const Bytes& frame : frames) { + result.push_back('>'); + result.push_back(uint8_t(frame.size())); + result.push_back(uint8_t(frame.size() >> 8)); + result.insert(result.end(), frame.begin(), frame.end()); + } + return result; +} + +struct Fixture { + BufferStream stream; + ArduinoSerialInterface usb; + MultiSerialInterface manager; + MyMesh mesh{&manager}; + uint8_t command[MAX_FRAME_SIZE] = {}; + + Fixture() { + g_mock_millis = 100; + usb.begin(stream, START_TOKEN, MOTA_TOKEN); + usb.enableFlowControl(true); + REQUIRE(manager.addInterface(InterfaceType::USB, &usb)); + manager.enable(); + } + size_t tick() { return manager.checkRecvFrame(command); } + void issue(uint8_t tag) { stream.push(Bytes{'<', 1, 0, tag}); } + void reply() { mesh.writeErrFrame(command[0]); } + void enqueue(const Bytes& frame) { + REQUIRE(usb.writeFrame(frame.data(), frame.size()) == frame.size()); + } + std::vector fillRequired() { + std::vector result; + for (uint8_t i = 0; i < 4; ++i) { + result.push_back(Bytes{1, uint8_t(0xA0 + i)}); + enqueue(result.back()); + } + return result; + } + void flush() { + stream.capacity = 65536; + stream.max_write = std::numeric_limits::max(); + manager.loop(); + REQUIRE(!usb.isWriteBusy()); + } +}; + +static void burst() { + Fixture f; + std::vector expected; + for (uint8_t i = 0; i < 5; ++i) f.issue(uint8_t(0x40 + i)); + for (uint8_t i = 0; i < 4; ++i) { + REQUIRE(f.tick() == 1); + REQUIRE(f.command[0] == 0x40 + i); + f.reply(); + expected.push_back(Bytes{1, uint8_t(0x40 + i)}); + } + const unsigned reads = f.stream.reads; + for (unsigned i = 0; i < 8; ++i) { + REQUIRE(f.tick() == 0); + REQUIRE(f.stream.reads == reads && f.stream.input.size() == 4); + REQUIRE(f.usb.getCompletedFrameCount() == 4); + g_mock_millis += 1000; + } + // Release exactly one complete reply, then accept the waiting command into + // that slot even though every later write is still backpressured. + f.stream.capacity = 5; + REQUIRE(f.tick() == 1 && f.command[0] == 0x44); + f.reply(); + expected.push_back(Bytes{1, 0x44}); + f.flush(); + REQUIRE(f.stream.output == framed(expected)); + REQUIRE(f.stream.input.empty() && !f.manager.hasPendingIO()); +} + +static void partialRxBudget() { + for (uint32_t started : {100U, UINT32_MAX - 500U}) { + for (const Bytes& prefix : {Bytes{'<'}, Bytes{'<', 2}, Bytes{'<', 2, 0, 0x40}}) { + Fixture f; + g_mock_millis = started; + f.stream.push(prefix); + REQUIRE(f.tick() == 0 && f.usb.isReadBusy()); + g_mock_millis += 250; + f.fillRequired(); + g_mock_millis += 5000; + f.manager.loop(); + REQUIRE(f.usb.isReadBusy()); + f.flush(); + g_mock_millis += 749; + f.manager.loop(); + REQUIRE(f.usb.isReadBusy()); + ++g_mock_millis; + f.manager.loop(); + REQUIRE(!f.usb.isReadBusy() && !f.manager.hasPendingIO()); + } + } +} + +static void partialRxResume() { + const Bytes wire{'<', 6, 0, 0x70, '<', '>', '\r', 0, 0xEE}; + for (size_t split = 1; split < wire.size(); ++split) { + Fixture f; + f.stream.push(Bytes(wire.begin(), wire.begin() + split)); + REQUIRE(f.tick() == 0 && f.usb.isReadBusy()); + std::vector expected = f.fillRequired(); + f.stream.push(Bytes(wire.begin() + split, wire.end())); + f.issue(0x72); + const unsigned reads = f.stream.reads; + g_mock_millis += 5000; + REQUIRE(f.tick() == 0 && f.stream.reads == reads); + f.stream.capacity = 5; + REQUIRE(f.tick() == 6); + REQUIRE(Bytes(f.command, f.command + 6) == Bytes(wire.begin() + 3, wire.end())); + f.reply(); + expected.push_back(Bytes{1, 0x70}); + REQUIRE(f.tick() == 0 && f.stream.input.size() == 4); + f.flush(); + REQUIRE(f.tick() == 1 && f.command[0] == 0x72); + f.reply(); + expected.push_back(Bytes{1, 0x72}); + f.flush(); + REQUIRE(f.stream.output == framed(expected)); + } +} + +static void evictableCapacity() { + for (bool pinned : {false, true}) { + Fixture f; + Bytes push(MAX_FRAME_SIZE, 0xA5); + push[0] = 0x88; + f.enqueue(push); + if (pinned) { + f.stream.capacity = 2; + f.manager.loop(); + REQUIRE(f.stream.output.size() == 2); + } + const Bytes required{1, 0xA0}; + for (unsigned i = 0; i < 3; ++i) f.enqueue(required); + f.issue(0x40); + const unsigned reads = f.stream.reads; + if (pinned) { + REQUIRE(f.tick() == 0 && f.stream.reads == reads); + // The partly transmitted best-effort head cannot create reply capacity + // until every byte has left the stream; its suffix must remain intact. + f.stream.capacity = MAX_FRAME_SIZE + 1; + } + REQUIRE(f.tick() == 1 && f.command[0] == 0x40); + f.reply(); + f.flush(); + std::vector expected; + if (pinned) expected.push_back(push); + for (unsigned i = 0; i < 3; ++i) expected.push_back(required); + expected.push_back(Bytes{1, 0x40}); + REQUIRE(f.stream.output == framed(expected)); + } + // An unsent best-effort tail remains evictable behind a partly sent required + // head; admission must not stop merely because all four slots are occupied. + Fixture f; + const Bytes required{1, 0xA0}; + f.enqueue(required); + f.stream.capacity = 2; + f.manager.loop(); + f.enqueue(Bytes{0x88, 0xFF}); + f.enqueue(required); + f.enqueue(required); + f.issue(0x40); + REQUIRE(f.tick() == 1); + f.reply(); + f.flush(); + REQUIRE(f.stream.output == framed({required, required, required, Bytes{1, 0x40}})); +} + +static void controlsAndPassthrough() { + for (bool secondary : {false, true}) { + for (bool partial : {false, true}) { + Fixture f; + const char* token = secondary ? MOTA_TOKEN : START_TOKEN; + if (partial) { + f.stream.push(Bytes(token, token + 5)); + REQUIRE(f.tick() == 0); + token += 5; + } + f.fillRequired(); + f.stream.push(token); + f.stream.push("\rver\r"); + const unsigned reads = f.stream.reads; + REQUIRE(f.tick() == 0 && f.stream.reads == reads); + REQUIRE(!f.usb.takeControlSequence() && !f.usb.takeSecondaryControlSequence()); + f.flush(); + REQUIRE(f.tick() == 0); + REQUIRE(f.usb.takeControlSequence() == !secondary); + REQUIRE(f.usb.takeSecondaryControlSequence() == secondary); + REQUIRE(f.stream.input == std::deque({'v', 'e', 'r', '\r'})); + f.usb.setPassthroughMode(true); + const Bytes before = f.stream.output; + const uint8_t response[] = {0}; + REQUIRE(f.usb.writeFrame(response, sizeof(response)) == sizeof(response)); + REQUIRE(f.tick() == 0 && f.stream.input.size() == 4); + REQUIRE(f.stream.output == before); + } + } +} + +static void resetsAndUncontrolledStreams() { + for (unsigned mode = 0; mode < 4; ++mode) { + Fixture f; + f.stream.push(Bytes{'<', 2, 0, 0x40}); + REQUIRE(f.tick() == 0 && f.usb.isReadBusy()); + f.fillRequired(); + g_mock_millis += 5000; + if (mode == 0) { + f.usb.enableFlowControl(false); + f.stream.push(Bytes{0x41}); + REQUIRE(f.tick() == 2 && f.command[0] == 0x40 && f.command[1] == 0x41); + } else { + if (mode == 1) f.usb.resetSessionState(); + if (mode == 2) { f.usb.setPassthroughMode(true); f.usb.setPassthroughMode(false); } + if (mode == 3) { f.usb.disable(); f.usb.enable(); } + f.issue(0x42); + REQUIRE(f.tick() == 1 && f.command[0] == 0x42); + } + REQUIRE(!f.usb.isReadBusy() && !f.manager.hasPendingIO()); + } + Fixture f; + f.usb.enableFlowControl(false); + f.stream.capacity = 4096; + for (unsigned i = 0; i < 5; ++i) { + f.issue(0x40); + REQUIRE(f.tick() == 1); + f.mesh.writeOKFrame(); + } + REQUIRE(f.stream.output == framed({Bytes{0}, Bytes{0}, Bytes{0}, Bytes{0}, Bytes{0}})); +} + +static void expiredBeforeBackpressure() { + Fixture f; + f.stream.push(Bytes{'<', 2, 0, 0x40}); + REQUIRE(f.tick() == 0 && f.usb.isReadBusy()); + g_mock_millis += 1000; + f.fillRequired(); + f.manager.loop(); + REQUIRE(!f.usb.isReadBusy()); + // The normal incomplete-frame timeout still works when there is no TX + // backpressure and must remain exact across the millis wrap boundary. + Fixture normal; + g_mock_millis = UINT32_MAX - 500U; + normal.stream.push(Bytes{'<', 2}); + REQUIRE(normal.tick() == 0); + g_mock_millis += 999; + normal.manager.loop(); + REQUIRE(normal.usb.isReadBusy()); + ++g_mock_millis; + normal.manager.loop(); + REQUIRE(!normal.usb.isReadBusy()); +} + +class OtherInterface : public BaseSerialInterface { +public: + bool pending = true; + void enable() override {} + void disable() override {} + bool isEnabled() const override { return true; } + bool isConnected() const override { return true; } + bool isReadBusy() const override { return false; } + bool isWriteBusy() const override { return false; } + size_t writeFrame(const uint8_t*, size_t len) override { return len; } + size_t checkRecvFrame(uint8_t* dest) override { + if (!pending) return 0; + pending = false; + dest[0] = 0x79; + return 1; + } +}; + +static void independentRoute() { + Fixture f; + OtherInterface other; + REQUIRE(f.manager.addInterface(InterfaceType::Bluetooth, &other)); + f.fillRequired(); + f.issue(0x40); + REQUIRE(f.tick() == 1 && f.command[0] == 0x79); + REQUIRE(f.manager.captureReplyRoute() == &other); + REQUIRE(f.stream.reads == 0 && f.stream.input.size() == 4); + f.flush(); + REQUIRE(f.tick() == 1 && f.command[0] == 0x40); + REQUIRE(f.manager.captureReplyRoute() == &f.usb); +} + +int main(int argc, char** argv) { + REQUIRE(argc == 2); + const int scenario = std::atoi(argv[1]); + switch (scenario) { + case 0: burst(); break; + case 1: partialRxBudget(); break; + case 2: partialRxResume(); break; + case 3: evictableCapacity(); break; + case 4: controlsAndPassthrough(); break; + case 5: resetsAndUncontrolledStreams(); break; + case 6: expiredBeforeBackpressure(); break; + case 7: independentRoute(); break; + default: REQUIRE(false); + } + std::printf("PASS: Arduino serial backpressure scenario %d\n", scenario); +} diff --git a/test/test_arduino_serial_backpressure.py b/test/test_arduino_serial_backpressure.py new file mode 100644 index 000000000..b6eb3d150 --- /dev/null +++ b/test/test_arduino_serial_backpressure.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Exercise production USB framing and immediate replies under backpressure.""" + +from pathlib import Path +import os +import shutil +import subprocess +import sys +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + + +ROOT = Path(__file__).resolve().parents[1] +FIXTURE = ROOT / "test/fixtures/arduino_serial_backpressure/test.cpp" +TRANSPORT = ROOT / "src/helpers/ArduinoSerialInterface.cpp" +SANITIZERS = ( + ["-fsanitize=address,undefined", "-fno-sanitize-recover=all", "-fno-pie", "-no-pie"] + if sys.platform.startswith("linux") else [] +) +SCENARIOS = ( + "five-command burst retains fifth command until required reply fits", + "partial headers and payloads preserve timeout budget across rollover", + "every RX split resumes intact without consuming the following command", + "only unsent best-effort frames provide replacement reply capacity", + "terminal and mOTA control tokens preserve their following input", + "session, passthrough and flow-control resets clear backpressure state", + "full TX queue cannot revive an already expired inbound frame", + "blocked USB input does not stop commands from another transport", +) + + +class ArduinoSerialBackpressureTests(unittest.TestCase): + def setUp(self): + self.compiler = os.environ.get("CXX") or shutil.which("g++") or shutil.which("clang++") + if self.compiler is None: + self.skipTest("a host C++17 compiler is required") + + def build(self, directory, transport): + directory = Path(directory) + mesh = (ROOT / "examples/companion_radio/MyMesh.cpp").read_text() + methods = "\n".join(extract_braced(mesh, signature) for signature in ( + "void MyMesh::writeOKFrame(", "void MyMesh::writeErrFrame(", + )) + (directory / "immediate_reply_methods.inc").write_text(methods) + candidate = directory / "ArduinoSerialInterface.cpp" + candidate.write_text(transport) + binary = directory / "serial-backpressure" + built = subprocess.run([ + self.compiler, "-std=c++17", *SANITIZERS, + "-I" + str(ROOT / "test/mocks"), "-I" + str(ROOT / "src"), + "-I" + str(ROOT / "src/helpers"), "-I" + str(directory), + str(FIXTURE), str(candidate), "-o", str(binary), + ], capture_output=True, text=True, timeout=60) + self.assertEqual(built.returncode, 0, built.stdout + built.stderr) + return binary + + def run_scenario(self, binary, scenario): + return subprocess.run([str(binary), str(scenario)], capture_output=True, + text=True, timeout=15) + + def test_production_parser_queue_and_immediate_reply_methods(self): + with tempfile.TemporaryDirectory(prefix="meshcore-serial-backpressure-") as directory: + binary = self.build(directory, TRANSPORT.read_text()) + for scenario, description in enumerate(SCENARIOS): + with self.subTest(scenario=description): + checked = self.run_scenario(binary, scenario) + self.assertEqual(checked.returncode, 0, checked.stdout + checked.stderr) + self.assertIn(f"PASS: Arduino serial backpressure scenario {scenario}", + checked.stdout) + + def test_negative_controls_reproduce_lost_reply_and_partial_frame(self): + source = TRANSPORT.read_text() + controls = ( + (" if (!responseCapacityAvailable()) return 0;\n", "", 0), + (" && !_receive_capacity_paused\n", "", 1), + ) + for old, new, scenario in controls: + with self.subTest(removed=old.strip()): + self.assertEqual(source.count(old), 1) + with tempfile.TemporaryDirectory(prefix="meshcore-serial-negative-") as directory: + binary = self.build(directory, source.replace(old, new)) + checked = self.run_scenario(binary, scenario) + self.assertNotEqual(checked.returncode, 0, + "negative control unexpectedly passed") + self.assertIn("FAIL line", checked.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_companion_usb_logging_reply.py b/test/test_companion_usb_logging_reply.py index 859607611..195878feb 100644 --- a/test/test_companion_usb_logging_reply.py +++ b/test/test_companion_usb_logging_reply.py @@ -196,12 +196,22 @@ int main() { reset(); stream.capacity=0; command("set usb.logging on"); cancelUsbSerialOperations(); usb_serial_interface.resetSessionState(); stream.capacity=4096; serviceUsbLoggingReplyBarrier(); assert(!logging); - // A queue unable to accept the acknowledgement cannot start the handoff. + // A full reply queue leaves the logging command unread until its + // acknowledgement can fit; draining then permits the normal handoff. reset(); stream.capacity=0; const uint8_t required[]={29,1}; for(unsigned i=0;i<4;++i) assert(usb_serial_interface.writeFrame(required,2)==2); - command("set usb.logging on"); + stream.receive("set usb.logging on"); + assert(interface_manager.checkRecvFrame(the_mesh.cmd_frame)==0); + assert(stream.cursor==0 && stream.available()>0); assert(!usb_logging_reply_pending && !logging && ble.tx.empty()); + stream.capacity=4096; + size_t resumed=interface_manager.checkRecvFrame(the_mesh.cmd_frame); + assert(resumed>0 && interface_manager.captureReplyRoute()==&usb_serial_interface); + the_mesh.handleCmdFrame(resumed); + assert(usb_logging_reply_pending && !logging && !the_mesh.terminal); + serviceUsbLoggingReplyBarrier(); + assert(logging && the_mesh.terminal && !usb_logging_reply_pending); // A later explicit setting supersedes an earlier pending on request. reset(); stream.capacity=0; command("set usb.logging on"); the_mesh.applyUsbLoggingState(false);