Fix serial protobuf corruption on short USB CDC writes (#10976)

* Fix serial protobuf corruption on short writes SerialConsole shared raw debug output and framed protobuf traffic on one HWCDC stream. Raw text could interleave inside an active frame when API logging was disabled. Separately, HWCDC deliberately returns a short write after bounded backpressure; StreamAPI abandoned that frame after PhoneAPI had already advanced, so the next 0x94c3 header landed inside the previous declared payload. Suppress all unframed output after protobuf mode starts and honor the existing config-replay log pause. For HWCDC, retain short frame tails in the persistent tx buffers and finish them on later loop passes before dequeuing another FromRadio packet. Logs remain best-effort and only start when their complete frame fits; main packets are deferred rather than dropped if a synchronous log is pending. Do not call HWCDC flush for framed serial output, since its no-progress path can discard queued bytes. TCP and non-HWCDC transports keep their existing behavior. Validated on Cardputer ADV (200-node DB) and Heltec Tracker V2: 400 initial full-DB sessions plus 140 final sessions across both API-log settings, zero malformed frames/timeouts/incomplete DBs; 2-20s forced reader stalls resume with complete DBs; abrupt stalled-client replacement 5/5 per board; 150s post-close reboot counts stable. Builds pass for Cardputer, Heltec, tbeam, and rak4631. * Add serial frame continuation regression tests Extract the HWCDC pending/deferred frame state machine into a small transport-independent helper so native tests exercise the same production logic used by SerialConsole. Keep framing, persistent buffer ownership and locking in SerialConsole. Cover short-tail continuation, deferred main-frame ordering, best-effort log admission, bounded zero-progress calls, generic StreamAPI failure semantics, PhoneAPI advancement gating, framed-log gating and raw output suppression. The coverage suite passes 31/31 suites and 582/582 tests. * Address review: guard flush, assert deferred invariant, dedup framing - Make SerialConsole::flush() a no-op in protobuf mode: HWCDC::flush()'s no-progress path discards queued TX bytes, which would tear a framed stream when the sleep path flushes with a stalled host. - Assert the single required-frame producer invariant in StreamFrameWriter::writeFrame() instead of silently dropping a second required frame. - Hoist 0x94C3 header construction into StreamAPI::buildFrameHeader() so SerialConsole no longer re-hardcodes the framing constants. * Test retained serial tail across client replacement Model a replacement client arriving while an older required frame has an unwritten tail. Require the old frame to complete before the new frame starts, so a new 0x94C3 header can never land inside the old declared payload. * Document serial frame APIs and regression tests Add concise Doxygen comments for the frame continuation hooks, production helper, native test doubles, and regression scenarios. Document why retained frame tails intentionally survive client disconnects: HWCDC may still hold the accepted prefix, so dropping metadata could insert a new frame header inside the old declared payload.
2026-07-11 06:27:50 -05:00
parent 8e27a8c715
commit ca833d944c
8 changed files with 675 additions and 30 deletions
@@ -3,7 +3,9 @@
 #include "NodeDB.h"
 #include "PowerFSM.h"
 #include "Throttle.h"
+#include "concurrency/LockGuard.h"
 #include "configuration.h"
+#include "main.h"
 #include "time.h"

 #if defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT
@@ -28,6 +30,7 @@

 SerialConsole *console;

+/// Create the shared serial console once and register receive wakeups.
 void consoleInit()
 {
    if (console) {
@@ -44,6 +47,7 @@ void consoleInit()
    DEBUG_PORT.rpInit(); // Simply sets up semaphore
 }

+/// Print and flush an unclassified formatted console message.
 void consolePrintf(const char *format, ...)
 {
    va_list arg;
@@ -53,6 +57,7 @@ void consolePrintf(const char *format, ...)
    console->flush();
 }

+/// Initialize console, protobuf transport, serial port, and worker thread state.
 SerialConsole::SerialConsole() : StreamAPI(&Port), RedirectablePrint(&Port), concurrency::OSThread("SerialConsole")
 {
    api_type = TYPE_SERIAL;
@@ -80,6 +85,7 @@ SerialConsole::SerialConsole() : StreamAPI(&Port), RedirectablePrint(&Port), con
 #endif
 }

+/// Service one serial API iteration and select the next polling interval.
 int32_t SerialConsole::runOnce()
 {
 #ifdef HELTEC_MESH_SOLAR
@@ -100,29 +106,48 @@ int32_t SerialConsole::runOnce()
 #endif
 }

+/// Flush raw output while preserving queued protobuf frames.
 void SerialConsole::flush()
 {
+    // HWCDC::flush()'s no-progress path discards queued TX bytes, which would tear a
+    // framed protobuf stream; framed output is drained by the TX interrupt instead.
+    if (usingProtobufs)
+        return;
+
    Port.flush();
 }

-// trigger tx of serial data
+/// Write raw console data only before protobuf framing becomes active.
+size_t SerialConsole::write(uint8_t c)
+{
+    // Once a protobuf client is active, unframed bytes would corrupt its stream.
+    if (usingProtobufs)
+        return 1;
+
+    if (c == '\n')
+        RedirectablePrint::write('\r');
+    return RedirectablePrint::write(c);
+}
+
+/// Wake the serial worker when PhoneAPI queues output.
 void SerialConsole::onNowHasData(uint32_t fromRadioNum)
 {
    setIntervalFromNow(0);
 }

-// trigger rx of serial data
+/// Wake the serial worker when receive activity is signaled.
 void SerialConsole::rxInt()
 {
    setIntervalFromNow(0);
 }

-// For the serial port we can't really detect if any client is on the other side, so instead just look for recent messages
+/// Infer serial client connectivity from recent API contact.
 bool SerialConsole::checkIsConnected()
 {
    return Throttle::isWithinTimespanMs(lastContactMsec, SERIAL_CONNECTION_TIMEOUT);
 }

+/// Select bounded or non-blocking HWCDC writes based on host liveness.
 void SerialConsole::setHostDraining(bool draining)
 {
 #ifdef IS_USB_SERIAL
@@ -134,17 +159,59 @@ void SerialConsole::setHostDraining(bool draining)
 #endif
 }

+/// Update HWCDC timeout mode around generic connection handling.
 void SerialConsole::onConnectionChanged(bool connected)
 {
    // Order matters on disconnect: make console TX non-blocking *before* the
    // PowerFSM/close handling below emits more log lines to a dead port.
-    if (!connected)
+    if (!connected) {
        setHostDraining(false);
+        // Keep any retained tail: HWCDC may still hold its prefix, and dropping metadata
+        // would let the next frame header land inside that frame's declared payload.
+    }
    StreamAPI::onConnectionChanged(connected);
    if (connected)
        setHostDraining(true);
 }

+/// Continue retained USB CDC output under the shared stream lock.
+bool SerialConsole::finishPendingFrame()
+{
+#ifdef IS_USB_SERIAL
+    concurrency::LockGuard guard(&streamLock);
+    return frameWriter.finishPendingFrame(Port);
+#else
+    return true;
+#endif
+}
+
+/// Protect the retained log buffer from being overwritten.
+bool SerialConsole::canEncodeLogRecord()
+{
+#ifdef IS_USB_SERIAL
+    concurrency::LockGuard guard(&streamLock);
+    return frameWriter.isIdle();
+#else
+    return true;
+#endif
+}
+
+/// Frame USB CDC output and retain any unwritten tail.
+bool SerialConsole::writeFrame(uint8_t *buf, size_t len, bool bestEffort)
+{
+#ifdef IS_USB_SERIAL
+    if (len == 0 || !canWrite)
+        return false;
+
+    const size_t totalLen = buildFrameHeader(buf, len);
+
+    concurrency::LockGuard guard(&streamLock);
+    return frameWriter.writeFrame(Port, buf, totalLen, bestEffort);
+#else
+    return StreamAPI::writeFrame(buf, len, bestEffort);
+#endif
+}
+
 /**
 * we override this to notice when we've received a protobuf over the serial
 * stream.  Then we shut off debug serial output.
@@ -167,12 +234,17 @@ bool SerialConsole::handleToRadio(const uint8_t *buf, size_t len)
    }
 }

+/// Route logs without allowing raw bytes into an active protobuf stream.
 void SerialConsole::log_to_serial(const char *logLevel, const char *format, va_list arg)
 {
-    if (usingProtobufs && config.security.debug_log_api_enabled) {
-        meshtastic_LogRecord_Level ll = RedirectablePrint::getLogLevel(logLevel);
-        auto thread = concurrency::OSThread::currentThread;
-        emitLogRecord(ll, thread ? thread->ThreadName.c_str() : "", format, arg);
-    } else
-        RedirectablePrint::log_to_serial(logLevel, format, arg);
+    if (usingProtobufs) {
+        if (config.security.debug_log_api_enabled && !pauseBluetoothLogging) {
+            meshtastic_LogRecord_Level ll = RedirectablePrint::getLogLevel(logLevel);
+            auto thread = concurrency::OSThread::currentThread;
+            emitLogRecord(ll, thread ? thread->ThreadName.c_str() : "", format, arg);
+        }
+        return;
+    }
+
+    RedirectablePrint::log_to_serial(logLevel, format, arg);
 }