diff --git a/docs/full_companion_usb_switcher.md b/docs/full_companion_usb_switcher.md index 0a9b4e5b..6cd84460 100644 --- a/docs/full_companion_usb_switcher.md +++ b/docs/full_companion_usb_switcher.md @@ -82,8 +82,9 @@ switched; they remain binary. 1. Companion initializes the normal USB Binary Companion interface, then gives its primary stream to the ASCII terminal before normal loop service begins. -2. While the prompt has no buffered input, the terminal peeks at the next byte. - It does not remove that byte. +2. While the prompt has no buffered input, the terminal skips blank CR/LF + delimiters and peeks at the next byte without removing it. A queued binary + frame after a blank line is detected before any ASCII banner is emitted. 3. If the byte is `<`, the terminal temporarily releases the stream and enables the existing `ArduinoSerialInterface` frame parser. 4. The parser consumes the original `<`, the two-byte length, and the payload. @@ -94,11 +95,21 @@ switched; they remain binary. 6. If no complete frame arrives within one second, the parser state is reset and the ASCII terminal prints a new banner and prompt. -The switcher checks framing, not client identity. Any syntactically complete -Binary Companion frame confirms binary mode; it does not require the first +The switcher checks framing, not client identity. A complete Binary Companion +frame with 1-176 payload bytes confirms binary mode; it does not require the first command to be `CMD_APP_START` or `CMD_DEVICE_QUERY`. Normal command validation still occurs after the frame parser returns the payload. +Oversized frames are consumed and discarded, never truncated into executable +commands or counted as proof of a Binary client. ASCII, framed Binary, and mOTA +input handlers process at most the initial queued-byte count, so a continuously +sending host cannot keep the radio loop trapped in one input call. + +ASCII and mOTA reject an entire line containing an embedded NUL or exceeding its +line buffer, discarding through the next CR/LF. A valid-looking suffix is not +executed. Binary payloads may still contain NUL bytes. The immediate manual +terminal-stop token remains an exception to ordinary line-oriented handling. + The empty-prompt requirement prevents a literal `<` in the middle of a command from silently changing modes. A literal `<` typed as the first character does start a probe, but the terminal returns after the one-second timeout if no diff --git a/docs/ota_protocol.md b/docs/ota_protocol.md index 5956b48b..5a1991fa 100644 --- a/docs/ota_protocol.md +++ b/docs/ota_protocol.md @@ -831,6 +831,35 @@ CDC/UART receive rings even though the host successfully wrote the complete reply. Chunking is internal to the transport and does not change `OP_READ` or the `MotaSource` random-access contract. +Each response uses one deadline, starting after request transmission, for its +magic, header, payload and checksum together. The deadline does not restart +for each received byte. USB folder sessions use 600 ms; BLE/TCP folder sessions +use 3000 ms. Each chunk of a larger logical read has its own deadline. A slow +trickle can therefore fail sooner than before; a maximum 197-byte response +takes about 17.1 ms on a 115200-baud UART, excluding host processing latency. + +A shared USB text/control session explicitly opts into retaining a timed-out +response's sync/header/payload/checksum position. No subsequent USB request is +sent until that old response reaches its known checksum boundary, or an actual +host-session reset clears it. The request supplies the +payload length: the response itself carries neither a length nor a request ID. +Late binary bytes therefore cannot become USB `ota folder off` control text. +A complete reply followed by that ordinary shutdown line still works. If a +software handoff or failed attach leaves a response outstanding, USB input is +quarantined until its boundary; commands typed during that wait are discarded. +Leaving a log-only USB/network handoff preserves that response for the tracked +consumer rather than discarding its tail as raw input. A deliberate USB mOTA +entry cancels startup probe and deferred ASCII timers so they cannot reclaim +the stream or emit terminal text into an active transfer. +If the sender abandons a partial reply, reconnect the USB host session to +recover. Bare shutdown text inside a partial payload is inherently ambiguous +and is not interpreted as control. This assumes one ordered reply per request; +unsolicited duplicate or reordered binary replies cannot be unambiguously +separated by this legacy wire format. + +Dedicated UART, BLE and TCP sources retain their legacy retry/stale-drain +behavior; their absolute response deadline still applies to every request. + `OP_DEFLATE_BLOCK` lets a host-folder source perform the optional transport compression without linking an encoder into the embedded seeder. `len=0, off=0` queries the exact encoded length; subsequent chunks are at most 190 bytes, keeping the total response payload at 192 bytes. The host independently encodes each manifest diff --git a/examples/companion_radio/main.cpp b/examples/companion_radio/main.cpp index d8bc9f39..770ff645 100644 --- a/examples/companion_radio/main.cpp +++ b/examples/companion_radio/main.cpp @@ -9,6 +9,10 @@ #include "CompanionBluetooth.h" #include "CompanionWireless.h" #include "CompanionWiFi.h" +#if COMPANION_FEATURE_USB_MOTA_SOURCE || COMPANION_FEATURE_BLE_MOTA_SOURCE +#include +#include +#endif #if defined(COMPANION_PAIRING_UI_HIL) void companionPairingUiHilProbe(); #endif @@ -321,9 +325,6 @@ void serviceWioE5HeadlessControls() { #endif #if COMPANION_FEATURE_BLE_MOTA_SOURCE -#include -#include - class Nrf52BleMotaSourceControl : public mesh::companion::MotaSourceControl { public: Nrf52BleMotaSourceControl() @@ -352,6 +353,7 @@ public: _packets_sent_at_start = context.manager.packetsSent(); _last_packets_sent = 0; bluetooth_interface.setMotaStreamActive(true); + _source.resetSessionState(); // setActive(true) cleared the old ring generation. if (!context.attach_folder_source( &_source, mesh::ota::OtaContext::FOLDER_LINK_BLE, "ble", reply, reply_size)) { @@ -636,12 +638,22 @@ static bool usb_logging_reply_state = false; #endif static mesh::UsbBinaryStartupProbe usb_binary_startup_probe; static mesh::UsbAsciiSessionDefault usb_ascii_session_default; +static bool acceptUsbBinaryStartupFrame(uint32_t completed_at) { + // Radio/mesh work and the parser itself can cross the deadline after the + // main-loop precheck. Reject before command dispatch or Binary-client proof. + if (usb_binary_startup_probe.hasTimedOut(completed_at)) return false; + // This first valid frame establishes Binary ownership immediately, not at + // the later terminal-service poll. Later frames have no startup deadline. + usb_binary_startup_probe.cancel(); + return true; +} #if COMPANION_FEATURE_USB_MOTA_SOURCE static bool usb_mota_mode = false; static mesh::UsbMotaEntryOrigin usb_mota_entry_origin = mesh::UsbMotaEntryOrigin::BINARY; static char usb_mota_line[32]; static size_t usb_mota_line_len = 0; +static bool usb_mota_discard_line = false; static bool usb_mota_disconnect_armed = false; #endif @@ -801,8 +813,12 @@ static void resetUsbMotaMode() { usb_mota_mode = false; usb_mota_line_len = 0; usb_mota_line[0] = 0; + usb_mota_discard_line = false; usb_mota_disconnect_armed = false; - usb_serial_interface.setPassthroughMode(false); + // A software owner change is not a transport reset. Retain any late binary + // reply and quarantine input until its known boundary is consumed. + usb_serial_interface.setPassthroughMode( + mesh::ota::OtaContext::serialFolderSource().hasPendingResponse()); } static void leaveUsbMotaMode(bool acknowledge) { @@ -826,13 +842,27 @@ static void leaveUsbMotaMode(bool acknowledge) { } static bool enterUsbMotaMode(mesh::UsbMotaEntryOrigin origin) { + // An explicit seeder command selects its own protocol owner, even if it + // arrived after a rejected startup frame. Neither a stale Binary deadline + // nor deferred default-ASCII restoration may reclaim a live mOTA stream. + usb_binary_startup_probe.cancel(); + usb_ascii_session_default.cancel(); + auto& source = mesh::ota::OtaContext::serialFolderSource(); + if (source.hasPendingResponse()) { + queueUsbTerminalControlReply("\r\nERR waiting for previous mOTA response\r\n"); + if (mesh::shouldRestoreAsciiAfterMotaFailure(origin)) enterUsbTerminalMode(); + return false; + } cancelUsbSerialOperations(); mesh::discardUsbTerminalOutput(); + source.enableSharedTextControl(); + source.resetSessionState(); usb_serial_interface.setPassthroughMode(true); usb_mota_mode = true; usb_mota_entry_origin = origin; usb_mota_line_len = 0; usb_mota_line[0] = 0; + usb_mota_discard_line = false; usb_mota_disconnect_armed = isUsbTerminalDataConnected(); #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \ && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT @@ -876,14 +906,20 @@ static void serviceUsbMota() { } #endif - // SerialMotaSource consumes framed responses synchronously while serving a - // block. Bytes left here are host control text, notably motatool's automatic - // `ota folder off` on a clean shutdown. - Stream& usb_input = mesh::usbCompanionPort(); - while (usb_input.available()) { - int value = usb_input.read(); + // A failed transaction can leave a late binary response, not just host text. + // Retain its known body boundary before recognizing clean shutdown controls. + // The static source accessor never allocates/acquires an OTA context here. + auto& source = mesh::ota::OtaContext::serialFolderSource(); + int pending = source.availableControlBytes(); + while (pending-- > 0) { + int value = source.readControlByte(); + if (value == mesh::ota::SerialMotaSource::RESPONSE_BYTE) continue; if (value < 0) break; char c = (char)value; + if (usb_mota_discard_line) { + if (c == '\r' || c == '\n') usb_mota_discard_line = false; + continue; + } if (c == '\r' || c == '\n') { if (usb_mota_line_len == 0) continue; usb_mota_line[usb_mota_line_len] = 0; @@ -896,15 +932,41 @@ static void serviceUsbMota() { } continue; } - if (usb_mota_line_len < sizeof(usb_mota_line) - 1) { + if (c != 0 && usb_mota_line_len < sizeof(usb_mota_line) - 1) { usb_mota_line[usb_mota_line_len++] = c; usb_mota_line[usb_mota_line_len] = 0; } else { usb_mota_line_len = 0; usb_mota_line[0] = 0; + usb_mota_discard_line = true; } } } + +static bool serviceUsbMotaResponseBoundary() { + if (usb_mota_mode) return false; + auto& source = mesh::ota::OtaContext::serialFolderSource(); + if (!source.hasPendingResponse()) return false; + // Failed attach or software handoff must not turn late binary data into + // terminal commands or Companion frames. Input typed while waiting is + // discarded; only an actual response boundary or host reset ends the wait. + if (!usb_serial_interface.isPassthroughMode()) { + usb_serial_interface.setPassthroughMode(true); + } + int pending = source.availableControlBytes(); + while (pending-- > 0) { + if (source.readControlByte() == -1) break; + if (!source.hasPendingResponse()) { + source.resetSessionState(); + usb_serial_interface.setPassthroughMode( + the_mesh.isTerminalMode() || usb_logging_network_parked + || (!mesh::hasDedicatedUsbLoggingPort() + && mesh::isUsbLoggingEnabled())); + break; + } + } + return true; +} #endif // Consume an exact TinyUSB owner-task edge, or HWCDC's framework-debounced SOF @@ -923,6 +985,7 @@ static void resetUsbTerminalHostSession() { if (usb_mota_mode) { leaveUsbMotaMode(false); } + mesh::ota::OtaContext::serialFolderSource().resetSessionState(); #endif if (the_mesh.isTerminalMode()) leaveUsbTerminalMode(false); mesh::discardUsbTerminalOutput(); @@ -1034,6 +1097,7 @@ static void serviceUsbTerminal() { serviceUsbLoggingOwnership(mesh::isUsbLoggingEnabled()); if (usb_logging_network_parked) return; #if COMPANION_FEATURE_USB_MOTA_SOURCE + if (serviceUsbMotaResponseBoundary()) return; if (usb_mota_mode) { serviceUsbMota(); return; @@ -1059,24 +1123,31 @@ static void serviceUsbTerminal() { return; } - // Companion boots as a useful ASCII terminal. MeshCLI's first framed - // command begins with '<'; hand that byte over untouched at an empty prompt. - // A malformed or accidental probe times out and restores the terminal. - if (!usb_logging_terminal_mode - && usb_binary_startup_probe.shouldStart( - usb_terminal_line_len == 0, usb_terminal_discard_line, - mesh::usbCompanionPort().peek())) { - const uint32_t frame_count = usb_serial_interface.getCompletedFrameCount(); - leaveUsbTerminalMode(false); - usb_binary_startup_probe.start(millis(), frame_count); - return; - } - Stream& usb_input = mesh::usbCompanionPort(); - if (the_mesh.isTerminalWaitingForInput() && usb_input.available() > 0) { - enterUsbTerminalMode(); - } - while (usb_input.available()) { + bool blank_input = false; + int pending = usb_input.available(); + while (pending-- > 0) { + // A leftover CR/LF (for example after mOTA shutdown) must not hide a + // following frame or cause a banner to poison its first binary reply. + if (usb_terminal_line_len == 0 && !usb_terminal_discard_line) { + const int next_byte = usb_input.peek(); + if (next_byte == '\r' || next_byte == '\n') { + usb_input.read(); + blank_input = true; + continue; + } + // Re-check at each empty prompt, before consuming the '<' byte. A + // malformed or accidental probe still returns to ASCII after one second. + if (!usb_logging_terminal_mode + && usb_binary_startup_probe.shouldStart(true, false, next_byte)) { + const uint32_t frame_count = usb_serial_interface.getCompletedFrameCount(); + leaveUsbTerminalMode(false); + usb_binary_startup_probe.start(millis(), frame_count); + return; + } + } + if (the_mesh.isTerminalWaitingForInput()) enterUsbTerminalMode(); + int value = usb_input.read(); if (value < 0) break; char c = (char)value; @@ -1089,6 +1160,15 @@ static void serviceUsbTerminal() { continue; } + if (c == 0) { + // The handlers consume C strings. Reject the whole malformed line, not + // an apparently valid mutating command before an embedded NUL. + clearUsbTerminalLine(); + usb_terminal_discard_line = true; + usbTerminalOutput().print("\r\n ERROR: invalid command\r\n"); + continue; + } + if (c == '\b' || c == 0x7F) { if (usb_terminal_line_len > 0) { usb_terminal_line_len = mesh::cli::eraseLastTerminalInput( @@ -1122,9 +1202,9 @@ static void serviceUsbTerminal() { return; } #endif - // On dual CDC hardware only CDC1 is the logging-reader endpoint. - if (!mesh::hasDedicatedUsbLoggingPort() - && strlen(usb_terminal_line) == usb_terminal_line_len) + // NUL-containing lines were discarded before reaching this point. On + // dual CDC hardware only CDC1 is the logging-reader endpoint. + if (!mesh::hasDedicatedUsbLoggingPort()) mesh::noteUsbLoggingStatsCommand(usb_terminal_line); the_mesh.handleTerminalCommand(usb_terminal_line); clearUsbTerminalLine(); @@ -1166,6 +1246,9 @@ static void serviceUsbTerminal() { return; } } + // Enter alone should still reveal the otherwise silent ASCII session. + if (blank_input && usb_input.available() == 0 + && the_mesh.isTerminalWaitingForInput()) enterUsbTerminalMode(); } static void serviceUsbLoggingOwnership(bool logging_enabled) { @@ -1197,11 +1280,21 @@ static void serviceUsbLoggingOwnership(bool logging_enabled) { && logging_action != mesh::UsbLoggingTerminalAction::PARK_USB_FOR_NETWORK) { // Bytes typed while USB was log-only belong to neither the new ASCII CLI // nor a new Binary client. Bound the discard to the existing RX snapshot. +#if COMPANION_FEATURE_USB_MOTA_SOURCE + const bool response_pending = + mesh::ota::OtaContext::serialFolderSource().hasPendingResponse(); +#else + const bool response_pending = false; +#endif + // A timed-out mOTA transaction still owns its exact response boundary. + // Raw reads would consume its tail without advancing that tracker, leaving + // the stream quarantined forever. Let its bounded consumer drain instead. Stream& input = mesh::usbCompanionPort(); - int pending = input.available(); + int pending = response_pending ? 0 : input.available(); while (pending-- > 0) input.read(); usb_logging_network_parked = false; - usb_serial_interface.setPassthroughMode(the_mesh.isTerminalMode()); + usb_serial_interface.setPassthroughMode( + the_mesh.isTerminalMode() || response_pending); } switch (logging_action) { case mesh::UsbLoggingTerminalAction::PARK_USB_FOR_NETWORK: @@ -3347,6 +3440,7 @@ void setup() { #endif // keep frames intact and pace the contact stream when the host is slow usb_serial_interface.enableFlowControl(true); + usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame); #if defined(ESP32) && defined(ARDUINO_USB_MODE) && ARDUINO_USB_MODE == 1 \ && defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT // The ESP32 USB-Serial-JTAG peripheral (HWCDC) has no DTR concept at all. @@ -3485,6 +3579,9 @@ void loop() { #endif #if defined(ENABLE_USB_INTERFACE) expireUsbBinaryStartupProbeBeforeDispatch(); +#if COMPANION_FEATURE_USB_MOTA_SOURCE + serviceUsbMotaResponseBoundary(); +#endif #endif the_mesh.loop(); #if defined(NRF52_PLATFORM) && defined(EXTRAFS) && !defined(QSPIFLASH) diff --git a/src/helpers/esp32/WiFiOtaSeeder.cpp b/src/helpers/esp32/WiFiOtaSeeder.cpp index bbb10aa8..1924e1b0 100644 --- a/src/helpers/esp32/WiFiOtaSeeder.cpp +++ b/src/helpers/esp32/WiFiOtaSeeder.cpp @@ -115,6 +115,7 @@ void WiFiOtaSeeder::loop() { } seeder_client = incoming; + seeder_source.resetSessionState(); // The old socket cannot supply this session's reply. if (!context.attach_folder_source(&seeder_source, OtaContext::FOLDER_LINK_TCP, "tcp", attach_reply, sizeof(attach_reply))) { seeder_client.stop(); diff --git a/src/helpers/ota/MotaSourceSerial.cpp b/src/helpers/ota/MotaSourceSerial.cpp index fba687d9..6f51f097 100644 --- a/src/helpers/ota/MotaSourceSerial.cpp +++ b/src/helpers/ota/MotaSourceSerial.cpp @@ -1,23 +1,122 @@ #include "MotaSourceSerial.h" #include "MotaSeederProto.h" #include "OtaByteIO.h" +#include #include namespace mesh { namespace ota { -bool SerialMotaSource::readByteT(uint8_t& b) { - uint32_t t0 = millis(); - while ((millis() - t0) < _to) { - int c = _io.read(); - if (c >= 0) { b = (uint8_t)c; return true; } +void SerialMotaSource::resetSessionState() { + _response_pending = false; + _response_stage = SYNC_FIRST; + _response_payload_len = _response_remaining = 0; + _control_offset = _control_len = 0; + _control_discard_line = false; +} + +void SerialMotaSource::queueControlByte(uint8_t b) { + const bool newline = b == '\r' || b == '\n'; + if (_control_discard_line) { + if (!newline) return; + _control_discard_line = false; + } + if (_control_len == sizeof(_control_queue)) { + // Invalidate any prefix already handed to the ASCII owner, then discard + // through the real line ending. Never expose an overlong line's suffix. + _control_offset = 0; + _control_len = 1; + _control_queue[0] = 0; + _control_discard_line = !newline; + if (!newline) return; + } + _control_queue[(_control_offset + _control_len) % sizeof(_control_queue)] = b; + ++_control_len; +} + +int SerialMotaSource::takeControlByte() { + if (_control_len == 0) return -1; + const uint8_t value = _control_queue[_control_offset]; + _control_offset = (_control_offset + 1) % sizeof(_control_queue); + --_control_len; + return value; +} + +int SerialMotaSource::readTrackedByte() { + const int value = _io.read(); + if (value < 0) return value; + if (!_shared_text_control) return value; + const uint8_t b = (uint8_t)value; + if (!_response_pending) { + queueControlByte(b); + return value; + } + switch (_response_stage) { + case SYNC_FIRST: + if (b == MOTA_SEEDER_RSP_MAGIC0) _response_stage = SYNC_SECOND; + else queueControlByte(b); + break; + case SYNC_SECOND: + if (b == MOTA_SEEDER_RSP_MAGIC1) { + _response_stage = RESPONSE_OP; + } else { + queueControlByte(MOTA_SEEDER_RSP_MAGIC0); + if (b != MOTA_SEEDER_RSP_MAGIC0) { + queueControlByte(b); + _response_stage = SYNC_FIRST; + } + } + break; + case RESPONSE_OP: + _response_stage = RESPONSE_STATUS; + break; + case RESPONSE_STATUS: + _response_remaining = b == MS_STATUS_OK ? _response_payload_len : 0; + _response_stage = _response_remaining ? RESPONSE_PAYLOAD : RESPONSE_CHECKSUM; + break; + case RESPONSE_PAYLOAD: + if (--_response_remaining == 0) _response_stage = RESPONSE_CHECKSUM; + break; + case RESPONSE_CHECKSUM: + // The request determines payload length, even when checksum/op fails. + // Only this full wire boundary, or an explicit owner reset, permits a + // new request; a late body tail can never become ASCII control text. + _response_pending = false; + _response_stage = SYNC_FIRST; + break; + } + return value; +} + +int SerialMotaSource::availableControlBytes() { + const int available = _io.available(); + if (available <= 0) return _control_len; + return available > INT_MAX - _control_len ? INT_MAX : available + _control_len; +} + +int SerialMotaSource::readControlByte() { + if (!_shared_text_control) return _io.read(); + if (_control_len != 0) return takeControlByte(); + if (readTrackedByte() < 0) return -1; + return _control_len ? takeControlByte() : RESPONSE_BYTE; +} + +bool SerialMotaSource::readByteT(uint8_t& b, uint32_t started) { + while ((uint32_t)(millis() - started) < _to) { + int c = readTrackedByte(); + if (c >= 0) { + // A read or scheduler tick can cross the deadline after the precheck. + if ((uint32_t)(millis() - started) >= _to) return false; + b = (uint8_t)c; + return true; + } delay(1); // let BLE/WiFi callbacks deliver the response } return false; } -bool SerialMotaSource::readExact(uint8_t* b, uint32_t n) { - for (uint32_t i = 0; i < n; i++) if (!readByteT(b[i])) return false; +bool SerialMotaSource::readExact(uint8_t* b, uint32_t n, uint32_t started) { + for (uint32_t i = 0; i < n; i++) if (!readByteT(b[i], started)) return false; return true; } @@ -25,7 +124,13 @@ bool SerialMotaSource::readExact(uint8_t* b, uint32_t n) { // checksum, then scans for the response magic and validates op+status+checksum before delivering payload. bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len) { - while (_io.read() >= 0) {} // drop any stale/partial bytes before a fresh request + // Drop the currently queued stale/partial input only. A continuously writing + // host cannot keep the primary mesh loop trapped before the request. + int pending = _io.available(); + while (pending-- > 0) if (readTrackedByte() < 0) break; + // Shared text ownership must never overwrite an incomplete old response's + // body length. The wire has no request ID or response payload length. + if (_response_pending) return false; if (arglen > 7) return false; // largest source request is READ(idx,off,len) uint8_t xs = op; for (uint8_t i = 0; i < arglen; i++) xs ^= args[i]; @@ -37,13 +142,19 @@ bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen, if (arglen) { memcpy(frame + frame_len, args, arglen); frame_len += arglen; } frame[frame_len++] = xs; if (_io.write(frame, frame_len) != frame_len) return false; + if (_shared_text_control) { + _response_pending = true; + _response_stage = SYNC_FIRST; + _response_payload_len = payload_len; + _response_remaining = 0; + } if (_write_policy == MotaStreamWritePolicy::FlushTransmit) _io.flush(); // scan for response magic 'm' 's' (tolerate leading noise) - uint32_t t0 = millis(); bool got = false; + const uint32_t t0 = millis(); bool got = false; uint8_t prev = 0; while ((millis() - t0) < _to) { - int c = _io.read(); + int c = readTrackedByte(); if (c < 0) { delay(1); continue; } if (prev == MOTA_SEEDER_RSP_MAGIC0 && (uint8_t)c == MOTA_SEEDER_RSP_MAGIC1) { got = true; break; } prev = (uint8_t)c; @@ -51,16 +162,16 @@ bool SerialMotaSource::txn(uint8_t op, const uint8_t* args, uint8_t arglen, if (!got) return false; uint8_t hdr[2]; - if (!readExact(hdr, 2)) return false; // op, status + if (!readExact(hdr, 2, t0)) return false; // op, status if (hdr[0] != op) return false; uint8_t rxs = (uint8_t)(MOTA_SEEDER_RSP_MAGIC0 ^ MOTA_SEEDER_RSP_MAGIC1) ^ hdr[0] ^ hdr[1]; bool ok = (hdr[1] == MS_STATUS_OK); if (ok && payload_len) { - if (!readExact(payload, payload_len)) return false; + if (!readExact(payload, payload_len, t0)) return false; for (uint32_t i = 0; i < payload_len; i++) rxs ^= payload[i]; } uint8_t xsum; - if (!readByteT(xsum)) return false; + if (!readByteT(xsum, t0)) return false; if (xsum != rxs) return false; // corrupt frame -> caller retries return ok; } diff --git a/src/helpers/ota/MotaSourceSerial.h b/src/helpers/ota/MotaSourceSerial.h index 3b8daf80..f4888df9 100644 --- a/src/helpers/ota/MotaSourceSerial.h +++ b/src/helpers/ota/MotaSourceSerial.h @@ -7,8 +7,10 @@ // A MotaSource backed by a host "mota-seeder" daemon over a dedicated framed Stream, such as a spare // UART, USB CDC, or TCP client. The device pulls catalog + bytes on demand (MotaSeederProto.h); the folder // image is never held on the device - it streams through. Do not share the Stream with a text CLI because -// command/log text would collide with the binary framing. Reads block up to `timeout_ms`; keep the host -// endpoint responsive so a round-trip does not stall the primary transfer. +// command/log text would collide with the binary framing. Each request's entire +// response (sync, header, payload, checksum) has one `timeout_ms` deadline after +// request transmission. Larger reads use multiple bounded requests. Keep the +// host responsive so a round-trip does not stall the primary transfer. namespace mesh { namespace ota { @@ -25,12 +27,40 @@ public: bool read_deflated_block(uint8_t idx, uint16_t block, uint8_t* buf, uint16_t cap, uint16_t* len) override; + // Only shared USB text/control ownership opts into retained boundaries. + // Dedicated UART/BLE/TCP keep their legacy retry and stale-drain behavior. + void enableSharedTextControl() { _shared_text_control = true; } + // A shared USB owner consumes only out-of-response control bytes. One call + // reads at most one underlying byte; RESPONSE_BYTE means binary was consumed. + static constexpr int RESPONSE_BYTE = -2; + int availableControlBytes(); + int readControlByte(); + bool hasPendingResponse() const { return _response_pending; } + void resetSessionState(); + private: // Send a request (op+args) and read its response header; on OK, `payload` (if non-null) receives // `payload_len` bytes. Returns true iff a well-formed OK response for `op` arrived in time. bool txn(uint8_t op, const uint8_t* args, uint8_t arglen, uint8_t* payload, uint32_t payload_len); - bool readByteT(uint8_t& b); // one byte within the timeout - bool readExact(uint8_t* b, uint32_t n); + bool readByteT(uint8_t& b, uint32_t started); // within the shared response deadline + bool readExact(uint8_t* b, uint32_t n, uint32_t started); + int readTrackedByte(); + void queueControlByte(uint8_t b); + int takeControlByte(); + + enum ResponseStage : uint8_t { + SYNC_FIRST, SYNC_SECOND, RESPONSE_OP, RESPONSE_STATUS, + RESPONSE_PAYLOAD, RESPONSE_CHECKSUM, + }; + bool _shared_text_control = false; + bool _response_pending = false; + ResponseStage _response_stage = SYNC_FIRST; + uint32_t _response_payload_len = 0; + uint32_t _response_remaining = 0; + uint8_t _control_queue[64] = {}; + uint8_t _control_offset = 0; + uint8_t _control_len = 0; + bool _control_discard_line = false; Stream& _io; uint32_t _to; diff --git a/src/helpers/ota/OtaContext.h b/src/helpers/ota/OtaContext.h index f45a6900..f2c9fc96 100644 --- a/src/helpers/ota/OtaContext.h +++ b/src/helpers/ota/OtaContext.h @@ -502,13 +502,16 @@ struct OtaContext { // Attach/detach an external folder of `.mota` served by a host daemon over the seeder UART (the node // then advertises + relays them alongside its own fw). Only built when OTA_FOLDER_SERIAL is configured. #if defined(OTA_FOLDER_SERIAL) - bool attach_folder(char* msg, size_t cap) { + static SerialMotaSource& serialFolderSource() { static SerialMotaSource src(OTA_FOLDER_SERIAL_STREAM, OTA_FOLDER_SERIAL_WRITE_POLICY, 600); + return src; + } + bool attach_folder(char* msg, size_t cap) { #ifdef OTA_FOLDER_SERIAL_BEGIN OTA_FOLDER_SERIAL_STREAM.begin(OTA_FOLDER_SERIAL_BAUD); // dedicated UART; console is already up #endif - return attach_folder_source(&src, FOLDER_LINK_SERIAL, "serial", msg, cap); + return attach_folder_source(&serialFolderSource(), FOLDER_LINK_SERIAL, "serial", msg, cap); } #endif void detach_folder(bool preserve_capture = false) { diff --git a/test/fixtures/shared_mota_queue/test_shared_mota_queue.cpp b/test/fixtures/shared_mota_queue/test_shared_mota_queue.cpp index 75cbb603..f36246f3 100644 --- a/test/fixtures/shared_mota_queue/test_shared_mota_queue.cpp +++ b/test/fixtures/shared_mota_queue/test_shared_mota_queue.cpp @@ -85,7 +85,12 @@ Log& usbLoggingPort() { static Log log; return log; } namespace ota { class SerialMotaSource : public ::Folder { public: + inline static unsigned session_resets = 0; SerialMotaSource(int, MotaStreamWritePolicy, uint32_t) {} + void resetSessionState() { + assert(::bluetooth_interface.active); + ++session_resets; + } }; } } #include "ble_control_under_test.h" @@ -225,10 +230,12 @@ int main() { #else Nrf52BleMotaSourceControl ble; assert(!ble.start(reply, sizeof reply)); // no subscription, no storage loan + assert(SerialMotaSource::session_resets == 0); assert(!ota_context_if_active() && q.buffer.capacity() == 256); for (int disconnect = 0; disconnect < 2; ++disconnect) { bluetooth_interface.ready = true; assert(ble.start(reply, sizeof reply)); + assert(SerialMotaSource::session_resets == static_cast(disconnect + 1)); assert(ble.status().attached && q.buffer.capacity() == 128); assert(ble.status().offered == 1 && ble.status().advertised == 1); if (disconnect) { @@ -255,6 +262,7 @@ int main() { "USB", reply, sizeof reply)); bluetooth_interface.ready = true; assert(!ble.start(reply, sizeof reply)); + assert(SerialMotaSource::session_resets == 2); assert(ble.status().another_link_active); assert(ble.stop(reply, sizeof reply)); ota_release_context_if_idle(false); diff --git a/test/fixtures/shared_mota_queue/test_wifi_shared_queue.h b/test/fixtures/shared_mota_queue/test_wifi_shared_queue.h index ed213335..e130ba85 100644 --- a/test/fixtures/shared_mota_queue/test_wifi_shared_queue.h +++ b/test/fixtures/shared_mota_queue/test_wifi_shared_queue.h @@ -115,6 +115,36 @@ static void test_wifi_shared_queue(Queue& q, CheckMessages check_messages) { packets.clear(); } + // TCP keeps legacy retries on the same connection. A genuinely new socket + // starts its next catalog handshake without old transport state. + WiFi.connected = true; + auto stalled = connect_host(); + WiFiOtaSeeder::loop(); + assert(WiFiOtaSeeder::isAttached() && ota_ctx().manager.servedCount() == 1); + stalled->respond = [](const uint8_t* request, size_t) { + assert(request[2] == MS_OP_COUNT); + return std::vector{'m', 's', MS_OP_COUNT, MS_STATUS_OK}; + }; + const unsigned before_timeout = stalled->requests; + ota_ctx().manager.refresh_sources(); + assert(stalled->requests == before_timeout + 1); + assert(ota_ctx().manager.servedCount() == 0); + WiFiOtaSeeder::loop(); + ota_ctx().manager.refresh_sources(); + assert(stalled->requests == before_timeout + 2); + stalled->connected = false; + WiFiOtaSeeder::loop(); + ota_release_context_if_idle(true); + auto replacement = connect_host(); + WiFiOtaSeeder::loop(); + assert(replacement->requests >= 2 && WiFiOtaSeeder::isAttached()); + assert(ota_ctx().manager.servedCount() == 1); + WiFiOtaSeeder::stop(); + ota_release_context_if_idle(true); + assert(!ota_context_if_active() && q.buffer.capacity() == 256); + check_messages(); + packets.clear(); + // An incoming TCP client and listener stop must not detach a USB source. char reply[160]; assert(ota_acquire_context(reply, sizeof reply)); diff --git a/test/test_companion_ota_includes.py b/test/test_companion_ota_includes.py new file mode 100644 index 00000000..b94bcd65 --- /dev/null +++ b/test/test_companion_ota_includes.py @@ -0,0 +1,94 @@ +#!/usr/bin/env python3 +"""Compile the production Companion include block without transitive OTA headers.""" + +from pathlib import Path +import subprocess +import tempfile +import unittest + + +ROOT = Path(__file__).resolve().parents[1] +MAIN = ROOT / "examples/companion_radio/main.cpp" + + +class CompanionOtaIncludeTest(unittest.TestCase): + def compile_include_block(self, flags, usb, ble): + # Preserve the actual production includes and their feature guards. + # Unlike handler-extraction tests, this harness does not inject either + # OTA header. MyMesh intentionally lacks its optional queue-dependent + # OtaContext include, just like the Heltec V4 Full configuration. + prefix = MAIN.read_text().split( + "// Believe it or not, this std C function is busted on some platforms!", + 1, + )[0] + self.assertIn('#include "MyMesh.h"', prefix) + with tempfile.TemporaryDirectory(prefix="meshcore-companion-includes-") as temp: + directory = Path(temp) + for header in ( + "helpers/ui/StartupScreen.h", "helpers/ui/DisplayPowerSettings.h", + "helpers/UsbLoggingWatchdog.h", "helpers/UsbLoggingClientActivity.h", + "Mesh.h", "helpers/BluetoothMac.h", "CompanionBluetooth.h", + "CompanionWireless.h", "CompanionWiFi.h", "esp_bt.h", "esp_pm.h", + "esp_sleep.h", "esp_system.h", "esp_heap_caps.h", + ): + path = directory / header + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("#pragma once\n") + (directory / "MyMesh.h").write_text( + '#pragma once\n#include "CompanionFeatures.h"\n' + ) + if not (usb or ble): + # Prove that small/non-OTA images do not gain these dependencies. + for header in ("MotaSourceSerial.h", "OtaContext.h"): + path = directory / "helpers/ota" / header + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text('#error "OTA headers must remain feature guarded"\n') + checks = ( + "static_assert(COMPANION_FEATURE_USB_MOTA_SOURCE == %d, \"USB profile\");\n" + "static_assert(COMPANION_FEATURE_BLE_MOTA_SOURCE == %d, \"BLE profile\");\n" + ) % (usb, ble) + if usb or ble: + checks += ( + "static_assert(sizeof(mesh::ota::OtaContext) != 0, \"context declared\");\n" + "static_assert(mesh::ota::SerialMotaSource::RESPONSE_BYTE == -2, " + "\"source declared\");\n" + ) + if usb: + checks += ( + "mesh::ota::SerialMotaSource& source_under_test() {\n" + " return mesh::ota::OtaContext::serialFolderSource();\n}\n" + ) + source = directory / "includes.cpp" + source.write_text( + "class Stream;\nStream& include_test_serial_port();\n" + "#define OTA_FOLDER_SERIAL_STREAM include_test_serial_port()\n" + + prefix + checks + ) + result = subprocess.run([ + "c++", "-std=c++17", "-fsyntax-only", *flags, + "-I", temp, "-I", str(ROOT / "src"), + "-I", str(ROOT / "test/mocks"), + "-I", str(ROOT / "examples/companion_radio"), str(source), + ], text=True, capture_output=True) + self.assertEqual(result.returncode, 0, result.stderr) + + def test_usb_only_full_esp32_declares_actual_ota_types(self): + self.compile_include_block([ + "-DESP32_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", "-DBLE_PIN_CODE=123456", + "-DWIFI_SSID=\"\"", "-DENABLE_USB_INTERFACE=1", "-DENABLE_OTA=1", + "-DOTA_FOLDER_SERIAL=1", "-DOTA_SEEDER_ONLY=1", + ], usb=1, ble=0) + + def test_ble_only_full_nrf52_declares_actual_ota_types(self): + self.compile_include_block([ + "-DNRF52_PLATFORM=1", "-DCOMPANION_RADIO_FULL=1", "-DBLE_PIN_CODE=123456", + "-DENABLE_OTA=1", "-DOTA_SEEDER_ONLY=1", + "-DCOMPANION_FEATURE_BLE_MOTA_SOURCE=1", + ], usb=0, ble=1) + + def test_non_ota_image_keeps_ota_headers_out(self): + self.compile_include_block(["-DSTM32_PLATFORM=1"], usb=0, ble=0) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_companion_usb_input.py b/test/test_companion_usb_input.py new file mode 100644 index 00000000..460f2331 --- /dev/null +++ b/test/test_companion_usb_input.py @@ -0,0 +1,559 @@ +#!/usr/bin/env python3 +"""Run actual Companion ASCII/probe and mOTA input handlers on host streams.""" + +from pathlib import Path +import subprocess +import tempfile +import unittest + +from test_replay_reset_integration import extract_braced + + +ROOT = Path(__file__).resolve().parents[1] +MAIN = (ROOT / "examples/companion_radio/main.cpp").read_text() +CONTEXT = (ROOT / "src/helpers/ota/OtaContext.h").read_text() + + +HARNESS = r''' +#include +#include +#include +#include +#include +#include +#include +#include "helpers/ArduinoSerialInterface.h" +#include "helpers/ota/MotaSourceSerial.h" +#include "helpers/UsbAsciiBinarySwitch.h" +#include "helpers/CLICommandUtils.h" +#define COMPANION_FEATURE_USB_MOTA_SOURCE 1 +#define COMPANION_FEATURE_NETWORK_TERMINAL 0 +#define MESH_USB_LOGGING_AVAILABLE 1 +struct Input : Stream { + std::deque bytes; + std::string output; + std::deque> responses; + unsigned source_requests=0; + bool replenish=false; + uint32_t advance_per_read=0; + unsigned reads=0; + int available() override { return static_cast(bytes.size()); } + int read() override { + if(bytes.empty()) return -1; + uint8_t result=bytes.front(); bytes.pop_front(); ++reads; + g_mock_millis+=advance_per_read; + if(replenish) bytes.push_back('x'); + return result; + } + int peek() override { return bytes.empty()?-1:bytes.front(); } + int availableForWrite() override { return 4096; } + size_t write(uint8_t value) override { output+=char(value); return 1; } + size_t write(const uint8_t* p,size_t n) override { + if(n>=4 && p[0]=='M' && p[1]=='S') { + ++source_requests; + if(!responses.empty()) { + for(uint8_t value:responses.front()) bytes.push_back(value); + responses.pop_front(); + } + } + output.append(reinterpret_cast(p),n); return n; + } + void push(const char* p) { while(*p) bytes.push_back(uint8_t(*p++)); } + void frame(uint8_t tail=3) { + const uint8_t frame[]={'<',2,0,0x16,tail}; + for(uint8_t c : frame) bytes.push_back(c); + } +} input; +#define OTA_FOLDER_SERIAL_STREAM input +#define OTA_FOLDER_SERIAL_WRITE_POLICY mesh::ota::MotaStreamWritePolicy::NoFlush +namespace mesh { namespace ota { +struct OtaContext { +@SOURCE_ACCESSOR@ +}; +} } +ArduinoSerialInterface usb_serial_interface; +static Stream& usbTerminalOutput() { return input; } +static bool connected=true, logging=false, dedicated=false; +static std::vector stats_commands; +namespace mesh { +bool isUsbLoggingEnabled() { return logging; } +bool hasDedicatedUsbLoggingPort() { return dedicated; } +void noteUsbLoggingStatsCommand(const char* command) { + stats_commands.emplace_back(command); +} +Stream& usbCompanionPort() { return input; } +void discardUsbTerminalOutput() {} +} +struct Mesh { + bool terminal=false, waiting=false, attach_succeeds=true; + bool attach_with_source=false; + unsigned banners=0, attaches=0, detaches=0; + std::vector commands; + bool isTerminalMode() const { return terminal; } + bool isTerminalWaitingForInput() const { return waiting; } + void enterTerminalMode(bool banner) { + terminal=true; waiting=!banner; + if(banner) { ++banners; input.output+="BANNER"; } + } + void exitTerminalMode() { terminal=false; waiting=false; } + void resetUsbHostSessionInput() {} + void handleTerminalCommand(char* command) { commands.emplace_back(command); } + bool handleLocalControlCommand(const char* command,char* reply,size_t size) { + if(!strcmp(command,"ota folder on")) { + ++attaches; + if(attach_with_source) { + auto& source=mesh::ota::OtaContext::serialFolderSource(); + mesh::ota::MotaDesc desc; + attach_succeeds=source.count()!=0 && source.describe(0,desc); + } + snprintf(reply,size,"%s",attach_succeeds?"OK attached":"ERR attach failed"); + } else { + assert(!strcmp(command,"ota folder off")); ++detaches; + snprintf(reply,size,"OK detached"); + } + return true; + } +} the_mesh; +static char usb_terminal_line[400]={}; +static size_t usb_terminal_line_len=0; +static bool usb_terminal_discard_line=false; +static bool usb_logging_terminal_mode=false,usb_logging_network_parked=false; +static mesh::UsbBinaryStartupProbe usb_binary_startup_probe; +static mesh::UsbAsciiSessionDefault usb_ascii_session_default; +static bool usb_mota_mode=false,usb_mota_disconnect_armed=false; +static mesh::UsbMotaEntryOrigin usb_mota_entry_origin=mesh::UsbMotaEntryOrigin::BINARY; +static char usb_mota_line[32]={}; +static size_t usb_mota_line_len=0; +static bool usb_mota_discard_line=false; +static const char USB_TERMINAL_START_TOKEN[]="+++MESHCORE-TERM-START"; +static const char USB_TERMINAL_STOP_TOKEN[]="+++MESHCORE-TERM-STOP"; +static const char USB_MOTA_START_TOKEN[]="ota folder on"; +static bool isUsbTerminalDataConnected() { return connected; } +static void redrawUsbTerminalInput() {} +static void serviceUsbLoggingOwnership(bool) {} +static void cancelUsbSerialOperations() {} +static void queueUsbTerminalControlReply(const char* reply) { input.output+=reply; } +static void drainUsbTerminalOutputBeforeProtocolSwitch() {} +@FUNCTIONS@ +static void reset() { + input=Input(); the_mesh=Mesh(); connected=true; logging=dedicated=false; + stats_commands.clear(); + mesh::ota::OtaContext::serialFolderSource().resetSessionState(); + usb_serial_interface=ArduinoSerialInterface(); + usb_serial_interface.begin(input,USB_TERMINAL_START_TOKEN,USB_MOTA_START_TOKEN); + usb_serial_interface.enable(); + usb_serial_interface.setReceiveFrameCheck(acceptUsbBinaryStartupFrame); + usb_logging_terminal_mode=usb_logging_network_parked=false; + usb_mota_mode=usb_mota_disconnect_armed=usb_mota_discard_line=false; + usb_mota_line_len=0; usb_mota_line[0]=0; + clearUsbTerminalLine(); usb_terminal_discard_line=false; + g_mock_millis=100; + enterUsbTerminalMode(false); +} +static void finishProbe() { + uint8_t frame[MAX_FRAME_SIZE]={}; + assert(usb_serial_interface.checkRecvFrame(frame)==2); + assert(frame[0]==0x16 && frame[1]==3); + serviceUsbTerminal(); + assert(!the_mesh.terminal && !usb_serial_interface.isPassthroughMode()); + assert(!usb_binary_startup_probe.isActive()); +} +static std::vector response(uint8_t op,const std::vector& payload, + uint8_t status=0) { + std::vector result={'m','s',op,status}; + result.insert(result.end(),payload.begin(),payload.end()); + uint8_t checksum=0; + for(uint8_t value:result) checksum^=value; + result.push_back(checksum); + return result; +} +int main() { + for(const char* leading : {"","\r","\n","\r\n","\n\r\n\r"}) { + reset(); input.push(leading); input.frame(); + serviceUsbTerminal(); + assert(!the_mesh.terminal && usb_binary_startup_probe.isActive()); + assert(input.available()==5 && input.peek()=='<'); + assert(input.output.empty() && the_mesh.banners==0); + finishProbe(); + } + // Empty Enter reveals the silent terminal without fabricating a command. + for(const char* leading : {"\r","\n","\r\n"}) { + reset(); input.push(leading); serviceUsbTerminal(); + assert(the_mesh.terminal && !the_mesh.waiting && the_mesh.banners==1); + assert(!usb_binary_startup_probe.isActive() && input.available()==0); + assert(the_mesh.commands.empty()); + serviceUsbTerminal(); assert(the_mesh.banners==1); + } + reset(); input.push("\r\nget name\r\n"); serviceUsbTerminal(); + assert(the_mesh.banners==1 && the_mesh.commands.size()==1); + assert(the_mesh.commands[0]=="get name" && the_mesh.terminal); + // Reaching an empty prompt through an edit also leaves the frame untouched. + reset(); input.push("x\b"); input.frame(); serviceUsbTerminal(); + assert(usb_binary_startup_probe.isActive() && input.available()==5); + finishProbe(); + // A typed prefix and an overlong-line discard cannot become binary probes. + reset(); input.push("x"); input.frame(); serviceUsbTerminal(); + assert(the_mesh.terminal && !usb_binary_startup_probe.isActive()); + reset(); the_mesh.waiting=false; usb_terminal_discard_line=true; + input.frame(); serviceUsbTerminal(); + assert(the_mesh.terminal && !usb_binary_startup_probe.isActive()); + assert(usb_terminal_discard_line && usb_terminal_line_len==0); + // Single-CDC logging forbids frames; a dedicated logger does not block CDC0. + reset(); logging=true; usb_logging_terminal_mode=true; + input.push("\r\n"); input.frame(); serviceUsbTerminal(); + assert(the_mesh.terminal && !usb_binary_startup_probe.isActive()); + reset(); logging=dedicated=true; input.push("\r\n"); input.frame(); + serviceUsbTerminal(); assert(usb_binary_startup_probe.isActive()); finishProbe(); + reset(); usb_logging_network_parked=true; input.frame(); serviceUsbTerminal(); + assert(input.available()==5 && !usb_binary_startup_probe.isActive()); + // NUL is legal inside a Binary frame, not a plaintext-line terminator. + reset(); input.push("\r\n"); input.frame(0); serviceUsbTerminal(); + uint8_t nul_frame[MAX_FRAME_SIZE]={}; + assert(usb_serial_interface.checkRecvFrame(nul_frame)==2); + assert(nul_frame[0]==0x16 && nul_frame[1]==0); + serviceUsbTerminal(); assert(!the_mesh.terminal && !usb_binary_startup_probe.isActive()); + + // The gate uses the actual completion timestamp, before returning a command + // or granting client proof, including delayed mesh work and millis rollover. + for(uint32_t start : {100u,UINT32_MAX-20}) { + for(uint32_t elapsed : {999u,1000u,1001u}) { + reset(); g_mock_millis=start; input.frame(); serviceUsbTerminal(); + assert(usb_binary_startup_probe.isActive()); + g_mock_millis=start+elapsed; + uint8_t command[MAX_FRAME_SIZE]={}; + const bool accepted=elapsed<1000; + assert(usb_serial_interface.checkRecvFrame(command)==(accepted?2u:0u)); + assert(usb_serial_interface.hasReceivedFrame()==accepted); + assert(usb_serial_interface.getCompletedFrameCount()==(accepted?1u:0u)); + assert(usb_serial_interface.getLastFrameMillis()==(accepted?millis():0)); + assert(!usb_serial_interface.isReadBusy()); + serviceUsbTerminal(); + assert(the_mesh.terminal==!accepted && !usb_binary_startup_probe.isActive()); + } + } + reset(); input.frame(); serviceUsbTerminal(); + g_mock_millis=1099; expireUsbBinaryStartupProbeBeforeDispatch(); + assert(usb_binary_startup_probe.isActive()); + // Time spent inside mesh.loop after its earlier deadline precheck. + g_mock_millis=1100; + uint8_t delayed_command[MAX_FRAME_SIZE]={}; + assert(usb_serial_interface.checkRecvFrame(delayed_command)==0); + assert(!usb_serial_interface.hasReceivedFrame()); + serviceUsbTerminal(); assert(the_mesh.terminal); + reset(); input.frame(); serviceUsbTerminal(); + g_mock_millis=1098; input.advance_per_read=1; + assert(usb_serial_interface.checkRecvFrame(delayed_command)==0); + assert(millis()==1103 && !usb_serial_interface.hasReceivedFrame()); + input.advance_per_read=0; serviceUsbTerminal(); assert(the_mesh.terminal); + reset(); input.frame(); serviceUsbTerminal(); g_mock_millis=1099; + assert(usb_serial_interface.checkRecvFrame(delayed_command)==2); + assert(!usb_binary_startup_probe.isActive()); + // The first timely frame establishes ownership before the service poll. + g_mock_millis=1100; input.frame(); + assert(usb_serial_interface.checkRecvFrame(delayed_command)==2); + assert(usb_serial_interface.getCompletedFrameCount()==2); + serviceUsbTerminal(); assert(!the_mesh.terminal); + // Once explicitly selected, newline-free STOP retains its legacy behavior. + reset(); input.push(USB_TERMINAL_STOP_TOKEN); serviceUsbTerminal(); + assert(!the_mesh.terminal && !usb_binary_startup_probe.isActive()); + g_mock_millis+=1001; input.frame(); + assert(usb_serial_interface.checkRecvFrame(delayed_command)==2); + + // Text-first attach, clean exit, and the remaining CRLF cannot corrupt a + // immediately following framed client command. + reset(); input.push("ota folder on\r\n"); serviceUsbTerminal(); + assert(usb_mota_mode && usb_mota_entry_origin==mesh::UsbMotaEntryOrigin::ASCII); + assert(!the_mesh.terminal && the_mesh.attaches==1); + input.push("ota folder off\r\n"); input.frame(); serviceUsbTerminal(); + assert(!usb_mota_mode && the_mesh.terminal && the_mesh.waiting); + assert(the_mesh.detaches==1); + input.output.clear(); serviceUsbTerminal(); + assert(usb_binary_startup_probe.isActive() && input.available()==5); + assert(input.output.empty()); finishProbe(); + + // Failed ASCII entry stays ASCII; failed Binary entry stays Binary. + reset(); the_mesh.attach_succeeds=false; + input.push("ota folder on\r\n"); serviceUsbTerminal(); + assert(!usb_mota_mode && the_mesh.terminal && the_mesh.banners==2); + assert(usb_serial_interface.isPassthroughMode() && !usb_mota_discard_line); + reset(); leaveUsbTerminalMode(false); the_mesh.attach_succeeds=false; + input.push("ota folder on\r\n"); uint8_t frame[MAX_FRAME_SIZE]={}; + assert(usb_serial_interface.checkRecvFrame(frame)==0); + serviceUsbTerminal(); + assert(!usb_mota_mode && !the_mesh.terminal && !usb_serial_interface.isPassthroughMode()); + assert(!usb_mota_discard_line); + + // Every NUL position rejects the whole ASCII owner-transition/mutating + // line, then the next real command is handled normally. + for(const char* command : {"ota folder on","ota folder off","set usb.logging on","get stats"}) { + for(size_t position=0;position<=strlen(command);++position) { + reset(); + for(size_t i=0;i{"get stats"}); + assert(stats_commands==the_mesh.commands); + reset(); dedicated=true; input.push("get stats\r\n"); serviceUsbTerminal(); + assert(the_mesh.commands==std::vector{"get stats"}); + assert(stats_commands.empty()); + + // Overlong lines never accept their suffix as an independent shutdown. + for(unsigned length : {32u,40u,4096u}) { + reset(); leaveUsbTerminalMode(false); + assert(enterUsbMotaMode(mesh::UsbMotaEntryOrigin::ASCII)); + for(unsigned i=0;i(dangerous,dangerous+sizeof(dangerous)-1)); + for(size_t prefix=0;prefix desc(38,'x'); + const char mutation[]="\nset usb.logging on\n"; + std::copy(mutation,mutation+sizeof(mutation)-1,desc.begin()+2); + const uint8_t embedded_frame[]={'<',2,0,0x16,3}; + std::copy(embedded_frame,embedded_frame+sizeof(embedded_frame),desc.begin()+25); + for(bool ascii : {false,true}) { + for(bool describe : {false,true}) { + const auto delayed=describe?response(2,desc):response(1,{1}); + for(size_t prefix=0;prefix payload(controls,controls+sizeof(controls)-1); + payload.insert(payload.end(),{'<',2,0,0x16,3}); + for (uint8_t status : {uint8_t(0),uint8_t(1)}) { + const auto late=response(3,status==0?payload:std::vector{},status); + for (size_t prefix=0;prefix data(payload.size()); + assert(!source.read(0,0,data.data(),data.size())); + assert(source.hasPendingResponse()); + logging=network=true; serviceUsbLoggingOwnership(true); + assert(!usb_mota_mode && usb_logging_network_parked); + assert(usb_serial_interface.isPassthroughMode()); + assert(the_mesh.attaches==1 && the_mesh.detaches==1); + if (logging_off_before_close) { + logging=false; serviceUsbLoggingOwnership(false); + assert(!usb_logging_network_parked && source.hasPendingResponse()); + assert(usb_serial_interface.isPassthroughMode()); + } + network=false; logging=logging_after_close; + uint8_t command[MAX_FRAME_SIZE]={}; + for (size_t i=prefix;i>8)}; + for (uint32_t i=0;i +#include +#include +#include +#include +#include +#include +#include "helpers/ota/MotaSourceSerial.h" +#include "helpers/BleMotaStream.h" +#include "helpers/ota/MotaSeederProto.h" +#include "helpers/ota/OtaByteIO.h" +#include "helpers/ota/FolderMotaStore.h" +using namespace mesh::ota; +''' + source += native[start:end] + source += r''' +int main(int argc, char** argv) { + ::testing::InitGoogleTest(&argc, argv); + return RUN_ALL_TESTS(); +} +''' + googletest = ROOT / ".pio/libdeps/native/googletest/googletest" + self.assertTrue(googletest.is_dir(), "Native googletest dependency must be present") + with tempfile.TemporaryDirectory() as directory: + cpp = Path(directory) / "serial-deadline.cpp" + cpp.write_text(source) + for sanitizer in (False, True): + with self.subTest(sanitizer=sanitizer): + binary = Path(directory) / ("serial-deadline-" + str(sanitizer)) + flags = (["-fsanitize=address,undefined", "-fno-sanitize-recover=all", + "-fno-pie", "-no-pie"] if sanitizer else []) + build = subprocess.run([ + "g++", "-std=c++17", "-pthread", *flags, + "-I", str(ROOT / "test/mocks"), "-I", str(ROOT / "src"), + "-I", str(googletest / "include"), "-I", str(googletest), + str(cpp), str(ROOT / "src/helpers/ota/MotaSourceSerial.cpp"), + str(ROOT / "src/helpers/ota/FolderMotaStore.cpp"), + str(googletest / "src/gtest-all.cc"), "-o", str(binary)], + capture_output=True, text=True) + self.assertEqual(build.returncode, 0, build.stdout + build.stderr) + run = subprocess.run([str(binary), "--gtest_brief=1"], + capture_output=True, text=True, timeout=30) + self.assertEqual(run.returncode, 0, run.stdout + run.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/test/test_ota/test_ota_core.cpp b/test/test_ota/test_ota_core.cpp index 71e33b2a..32fe7d85 100644 --- a/test/test_ota/test_ota_core.cpp +++ b/test/test_ota/test_ota_core.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include "helpers/ota/MotaContainer.h" #include "helpers/ota/MerkleTree.h" @@ -12,6 +13,7 @@ #include "helpers/ota/FirmwareInfo.h" #include "helpers/ota/MotaSeederProto.h" #include "helpers/ota/MotaSourceSerial.h" +#include "helpers/BleMotaStream.h" #include "helpers/ota/FolderMotaStore.h" #include "helpers/ota/SignerAllowlist.h" #include "helpers/ota/OtaStore.h" @@ -886,6 +888,10 @@ public: return -1; } + int available() override { + return static_cast(response.size() - response_pos); + } + void flush() override { ++flush_calls; if (flush_discards_receive) { @@ -900,7 +906,7 @@ public: bool flush_discards_receive = false; size_t flush_calls = 0; -private: +protected: std::vector response; size_t response_pos = 0; }; @@ -955,6 +961,10 @@ public: return -1; } + int available() override { + return static_cast(response.size() - response_pos); + } + void flush() override { ++flush_calls; } std::vector encoded; @@ -1065,6 +1075,403 @@ TEST(MotaSourceSerial, DeflateExtensionFailureLeavesRawFallbackAvailable) { EXPECT_EQ(stream.lengths, (std::vector{0})); } +class TimedMotaSeederStream : public FakeMotaSeederStream { +public: + std::vector byte_at; + uint32_t started = 0; + uint32_t delay_per_byte = 0; + bool wrong_op = false, corrupt_checksum = false, error_status = false; + unsigned stale_remaining = 0, stale_refills = 0, stale_reads = 0; + unsigned stale_reads_at_request = 0; + + size_t write(const uint8_t* request, size_t len) override { + stale_reads_at_request = stale_reads; + stale_remaining = stale_refills = 0; + started = millis(); + const size_t written = FakeMotaSeederStream::write(request, len); + if (wrong_op) ++response[2]; + if (error_status) { + response.resize(5); + response[3] = MS_STATUS_ERR; + } + uint8_t checksum = 0; + for (size_t i = 0; i + 1 < response.size(); ++i) checksum ^= response[i]; + response.back() = checksum ^ (corrupt_checksum ? 1 : 0); + return written; + } + + uint32_t byteTime(size_t index) const { + return index < byte_at.size() ? byte_at[index] + : uint32_t(index) * delay_per_byte; + } + + int available() override { + if (stale_remaining) return static_cast(stale_remaining); + const uint32_t elapsed = millis() - started; + size_t count = 0; + while (response_pos + count < response.size() + && byteTime(response_pos + count) <= elapsed) ++count; + return static_cast(count); + } + + int read() override { + if (stale_remaining) { + --stale_remaining; + ++stale_reads; + if (stale_refills) { --stale_refills; ++stale_remaining; } + return 'x'; + } + if (response_pos >= response.size() + || uint32_t(millis() - started) < byteTime(response_pos)) return -1; + return FakeMotaSeederStream::read(); + } + + size_t unreadResponse() const { return response.size() - response_pos; } +}; + +TEST(MotaSourceSerial, SharedDeadlineBoundsTricklingPayload) { + resetArduinoMock(); + TimedMotaSeederStream stream; + stream.byte_at.resize(MOTA_SEEDER_READ_MAX + 5); + for (size_t i = 4; i < stream.byte_at.size(); ++i) { + stream.byte_at[i] = uint32_t(i - 3) * 590; + } + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600); + std::array data{}; + EXPECT_FALSE(source.read(0, 0, data.data(), data.size())); + EXPECT_EQ(millis(), 600u); + EXPECT_EQ(data[0], 0u); + EXPECT_EQ(data[1], 0u); + EXPECT_GT(stream.unreadResponse(), 1u); +} + +TEST(MotaSourceSerial, SyncHeaderPayloadAndChecksumShareOneDeadline) { + resetArduinoMock(); + TimedMotaSeederStream stream; + stream.byte_at = {0, 300, 350, 450, 550, 575, 600}; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600); + uint8_t data[2] = {}; + EXPECT_FALSE(source.read(0, 0, data, sizeof(data))); + EXPECT_EQ(millis(), 600u); + EXPECT_EQ(stream.unreadResponse(), 1u); +} + +TEST(MotaSourceSerial, DeadlineIsStrictAndHandlesMillisRollover) { + for (uint32_t started : {0u, UINT32_MAX - 20}) { + for (uint32_t checksum_at : {599u, 600u}) { + SCOPED_TRACE(started); + SCOPED_TRACE(checksum_at); + resetArduinoMock(); + g_mock_millis = started; + TimedMotaSeederStream stream; + stream.byte_at = {0, 0, 0, 0, 0, checksum_at}; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600); + uint8_t data = 0; + EXPECT_EQ(source.read(0, 0, &data, 1), checksum_at < 600); + EXPECT_EQ(uint32_t(millis() - started), checksum_at); + EXPECT_EQ(stream.unreadResponse(), checksum_at < 600 ? 0u : 1u); + } + } +} + +TEST(MotaSourceSerial, ActualReadCompletionMustPrecedeDeadline) { + class AdvancingReadStream : public FakeMotaSeederStream { + public: + int read() override { + const int value = FakeMotaSeederStream::read(); + if (value >= 0) ++g_mock_millis; + return value; + } + }; + for (uint32_t start : {0u, UINT32_MAX - 3}) { + for (uint32_t timeout : {5u, 6u, 7u}) { + for (bool shared : {false, true}) { + SCOPED_TRACE(start); + SCOPED_TRACE(timeout); + SCOPED_TRACE(shared); + resetArduinoMock(); + g_mock_millis = start; + AdvancingReadStream stream; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, timeout); + if (shared) source.enableSharedTextControl(); + uint8_t data = 0; + EXPECT_EQ(source.read(0, 0, &data, 1), timeout > 6); + EXPECT_EQ(uint32_t(millis() - start), std::min(timeout, 6u)); + EXPECT_EQ(source.hasPendingResponse(), shared && timeout < 6); + } + } + } +} + +TEST(MotaSourceSerial, LargestRepliesFitExistingUsbUartAndWirelessDeadlines) { + for (const auto profile : {std::pair{400, 2}, + std::pair{600, 3}, + std::pair{3000, 15}}) { + SCOPED_TRACE(profile.first); + resetArduinoMock(); + TimedMotaSeederStream stream; + stream.delay_per_byte = profile.second; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, profile.first); + std::array data{}; + ASSERT_TRUE(source.read(0, 0x1000, data.data(), data.size())); + EXPECT_EQ(millis(), uint32_t(MOTA_SEEDER_READ_MAX + 4) * profile.second); + EXPECT_LT(millis(), profile.first); + for (size_t i = 0; i < data.size(); ++i) EXPECT_EQ(data[i], uint8_t(i)); + } +} + +TEST(MotaSourceSerial, StaleDrainDoesNotChaseNewlyArrivingInput) { + resetArduinoMock(); + TimedMotaSeederStream stream; + stream.stale_remaining = 4; + stream.stale_refills = 512; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600); + uint8_t data = 0; + ASSERT_TRUE(source.read(0, 0x5A, &data, 1)); + EXPECT_EQ(stream.stale_reads_at_request, 4u); + EXPECT_EQ(data, 0x5A); +} + +TEST(MotaSourceSerial, ErrorChecksumAndWrongOpStayFailClosedAndRetryResynchronizes) { + for (unsigned failure = 0; failure < 3; ++failure) { + SCOPED_TRACE(failure); + resetArduinoMock(); + TimedMotaSeederStream stream; + stream.error_status = failure == 0; + stream.corrupt_checksum = failure == 1; + stream.wrong_op = failure == 2; + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 600); + uint8_t data[8] = {}; + EXPECT_FALSE(source.read(0, 0x20, data, sizeof(data))); + stream.error_status = stream.corrupt_checksum = stream.wrong_op = false; + ASSERT_TRUE(source.read(0, 0x30, data, sizeof(data))); + EXPECT_EQ(stream.unreadResponse(), 0u); + for (size_t i = 0; i < sizeof(data); ++i) EXPECT_EQ(data[i], uint8_t(0x30 + i)); + } +} + +class ManualMotaSeederStream : public Stream { +public: + using Stream::write; + std::vector> replies; + std::vector input; + size_t position = 0, writes = 0; + size_t write(const uint8_t*, size_t len) override { + if (writes < replies.size()) push(replies[writes]); + ++writes; + return len; + } + void push(const std::vector& bytes) { + input.insert(input.end(), bytes.begin(), bytes.end()); + } + int available() override { return static_cast(input.size() - position); } + int read() override { return position < input.size() ? input[position++] : -1; } +}; + +static std::vector serialResponse(uint8_t op, + const std::vector& payload, + uint8_t status = MS_STATUS_OK) { + std::vector response = {'m', 's', op, status}; + response.insert(response.end(), payload.begin(), payload.end()); + uint8_t checksum = 0; + for (uint8_t value : response) checksum ^= value; + response.push_back(checksum); + return response; +} + +static std::vector serialControlBytes(SerialMotaSource& source) { + std::vector result; + int pending = source.availableControlBytes(); + while (pending-- > 0) { + const int value = source.readControlByte(); + if (value == -1) break; + if (value >= 0) result.push_back(static_cast(value)); + } + return result; +} + +TEST(MotaSourceSerial, LateReplyBoundariesCoverAllSourceOperationsAndErrorStatus) { + const char stop[] = "ota folder off\r\n"; + const std::vector controls(stop, stop + sizeof(stop) - 1); + for (uint8_t op : {MS_OP_COUNT, MS_OP_DESCRIBE, MS_OP_READ, MS_OP_DEFLATE_BLOCK}) { + for (bool error : {false, true}) { + std::vector payload; + if (!error) { + if (op == MS_OP_COUNT) payload = {1}; + else if (op == MS_OP_DESCRIBE) payload.assign(MOTA_DESC_WIRE, 'x'); + else if (op == MS_OP_DEFLATE_BLOCK) payload = {17, 0}; + else { + const char body[] = "x\nota folder off\n"; + payload.assign(body, body + sizeof(body) - 1); + } + } + const auto response = serialResponse(op, payload, error ? MS_STATUS_ERR : MS_STATUS_OK); + for (size_t prefix = 0; prefix < response.size(); ++prefix) { + SCOPED_TRACE(op); + SCOPED_TRACE(error); + SCOPED_TRACE(prefix); + resetArduinoMock(); + ManualMotaSeederStream stream; + stream.replies.emplace_back(response.begin(), response.begin() + prefix); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + source.enableSharedTextControl(); + uint8_t data[64] = {}; + uint16_t encoded_len = 123; + MotaDesc desc; + if (op == MS_OP_COUNT) EXPECT_EQ(source.count(), 0); + else if (op == MS_OP_DESCRIBE) EXPECT_FALSE(source.describe(0, desc)); + else if (op == MS_OP_DEFLATE_BLOCK) { + EXPECT_FALSE(source.read_deflated_block(0, 0, data, sizeof(data), &encoded_len)); + EXPECT_EQ(encoded_len, 0u); + } else EXPECT_FALSE(source.read(0, 0, data, payload.size() ? payload.size() : 17)); + EXPECT_TRUE(source.hasPendingResponse()); + for (unsigned retry = 0; retry < 3; ++retry) { + EXPECT_FALSE(source.read(0, 0, data, 7)); + EXPECT_EQ(stream.writes, 1u); + EXPECT_TRUE(source.hasPendingResponse()); + } + stream.push(std::vector(response.begin() + prefix, response.end())); + stream.push(controls); + EXPECT_EQ(serialControlBytes(source), controls); + EXPECT_FALSE(source.hasPendingResponse()); + } + } + } +} + +TEST(MotaSourceSerial, StaleDrainRetainsOldBodyLengthAcrossDifferentSizeRetries) { + resetArduinoMock(); + ManualMotaSeederStream stream; + const auto old_response = serialResponse(MS_OP_READ, {'x', '\n', 'o', 't', 'a', ' ', 'f', 'o', + 'l', 'd', 'e', 'r', ' ', 'o', 'f', 'f', '\n'}); + stream.replies.emplace_back(old_response.begin(), old_response.begin() + 6); + stream.replies.push_back(serialResponse(MS_OP_READ, {42, 43})); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + source.enableSharedTextControl(); + uint8_t old_data[17] = {}; + EXPECT_FALSE(source.read(0, 0, old_data, sizeof(old_data))); + uint8_t new_data[2] = {}; + for (size_t index = 6; index + 1 < old_response.size(); ++index) { + stream.push({old_response[index]}); + EXPECT_FALSE(source.read(0, 99, new_data, sizeof(new_data))); + EXPECT_EQ(stream.writes, 1u); + EXPECT_TRUE(source.hasPendingResponse()); + EXPECT_TRUE(serialControlBytes(source).empty()); + } + stream.push({old_response.back()}); + const char stop[] = "ota folder off\r\n"; + const std::vector controls(stop, stop + sizeof(stop) - 1); + stream.push(controls); + EXPECT_TRUE(source.read(0, 99, new_data, sizeof(new_data))); + EXPECT_EQ(stream.writes, 2u); + EXPECT_EQ(new_data[0], 42u); + EXPECT_EQ(new_data[1], 43u); + EXPECT_EQ(serialControlBytes(source), controls); + EXPECT_FALSE(source.hasPendingResponse()); +} + +TEST(MotaSourceSerial, LateDeflateChunkLengthIncludesItsTwoByteSizePrefix) { + const char body[] = "x\nota folder off\n"; + std::vector payload = {sizeof(body) - 1, 0}; + payload.insert(payload.end(), body, body + sizeof(body) - 1); + const auto delayed = serialResponse(MS_OP_DEFLATE_BLOCK, payload); + const char stop[] = "ota folder off\r\n"; + const std::vector controls(stop, stop + sizeof(stop) - 1); + for (size_t prefix = 0; prefix < delayed.size(); ++prefix) { + SCOPED_TRACE(prefix); + resetArduinoMock(); + ManualMotaSeederStream stream; + stream.replies.push_back(serialResponse(MS_OP_DEFLATE_BLOCK, {sizeof(body) - 1, 0})); + stream.replies.emplace_back(delayed.begin(), delayed.begin() + prefix); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + source.enableSharedTextControl(); + uint8_t data[64] = {}; + uint16_t encoded_len = 123; + EXPECT_FALSE(source.read_deflated_block(0, 0, data, sizeof(data), &encoded_len)); + EXPECT_EQ(encoded_len, 0u); + EXPECT_EQ(stream.writes, 2u); + EXPECT_TRUE(source.hasPendingResponse()); + stream.push(std::vector(delayed.begin() + prefix, delayed.end())); + stream.push(controls); + EXPECT_EQ(serialControlBytes(source), controls); + EXPECT_FALSE(source.hasPendingResponse()); + } +} + +TEST(MotaSourceSerial, ControlQueueOverflowRejectsWholeLineAndResetClearsQuarantine) { + resetArduinoMock(); + ManualMotaSeederStream stream; + std::vector malformed(100, 'x'); + const char suffix[] = "ota folder off\n"; + malformed.insert(malformed.end(), suffix, suffix + sizeof(suffix) - 1); + stream.push(malformed); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + source.enableSharedTextControl(); + uint8_t data = 0; + EXPECT_FALSE(source.read(0, 0, &data, 1)); + EXPECT_TRUE(source.hasPendingResponse()); + EXPECT_EQ(serialControlBytes(source), (std::vector{0, '\n'})); + source.resetSessionState(); + EXPECT_FALSE(source.hasPendingResponse()); + EXPECT_EQ(source.availableControlBytes(), 0); + stream.replies.resize(2); + stream.replies[1] = serialResponse(MS_OP_READ, {42}); + EXPECT_TRUE(source.read(0, 0, &data, 1)); + EXPECT_EQ(data, 42u); +} + +TEST(MotaSourceSerial, DefaultBleRetryAndNewRingSessionPreserveLegacyBehavior) { + resetArduinoMock(); + BleMotaStream stream; + struct Host { + BleMotaStream* stream; + unsigned requests = 0; + std::vector reply; + } host{&stream, 0, {'m', 's', MS_OP_READ, MS_STATUS_OK, 'x'}}; + stream.setSender([](void* context, const uint8_t*, size_t len) { + auto& host = *static_cast(context); + ++host.requests; + if (!host.reply.empty()) EXPECT_TRUE(host.stream->pushRx(host.reply.data(), host.reply.size())); + return len; + }, &host); + stream.setActive(true); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + uint8_t data[17] = {}; + EXPECT_FALSE(source.read(0, 0, data, sizeof(data))); + EXPECT_FALSE(source.hasPendingResponse()); + host.reply = serialResponse(MS_OP_COUNT, {1}); + EXPECT_EQ(source.count(), 1); + EXPECT_EQ(host.requests, 2u); // Default non-text transports retain ordinary retries. + stream.setActive(false); + stream.setActive(true); // New generation drops the old ring contents/writers. + EXPECT_EQ(stream.available(), 0); + EXPECT_FALSE(source.hasPendingResponse()); + source.resetSessionState(); // The production BLE start pairs these two resets. + host.reply = serialResponse(MS_OP_COUNT, {1}); + EXPECT_EQ(source.count(), 1); + EXPECT_EQ(host.requests, 3u); + EXPECT_FALSE(source.hasPendingResponse()); +} + +TEST(MotaSourceSerial, DefaultRetryCanShareTcpWithLegacyFolderStoreReads) { + resetArduinoMock(); + ManualMotaSeederStream stream; + stream.replies.push_back({'m', 's', MS_OP_COUNT, MS_STATUS_OK}); + stream.replies.push_back(serialResponse(MS_OP_STAT, {1, 13, 0, 0, 0})); + stream.replies.push_back(serialResponse(MS_OP_COUNT, {1})); + SerialMotaSource source(stream, MotaStreamWritePolicy::NoFlush, 4); + FolderMotaStore store(stream, MotaStreamWritePolicy::NoFlush, 4); + EXPECT_EQ(source.count(), 0); + EXPECT_FALSE(source.hasPendingResponse()); + const auto old_response = serialResponse(MS_OP_COUNT, {1}); + stream.push(std::vector(old_response.begin() + 4, old_response.end())); + EXPECT_TRUE(store.reopen()); // Its independent legacy drain consumes the old tail. + EXPECT_EQ(store.staged_size(), 13u); + EXPECT_EQ(source.count(), 1); // No USB-only retained state poisons this source. + EXPECT_EQ(stream.writes, 3u); + EXPECT_FALSE(source.hasPendingResponse()); +} + class FakeFolderMotaSeederStream : public Stream { public: using Stream::write;