From ca833d944ca35273ee870b579f1edde17653390b Mon Sep 17 00:00:00 2001 From: p0ns Date: Sat, 11 Jul 2026 08:27:50 -0300 Subject: [PATCH] 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. --- src/SerialConsole.cpp | 92 +++++- src/SerialConsole.h | 27 +- src/mesh/StreamAPI.cpp | 40 ++- src/mesh/StreamAPI.h | 17 +- src/mesh/StreamFrameWriter.cpp | 65 +++++ src/mesh/StreamFrameWriter.h | 26 ++ test/native-suite-count | 2 +- test/test_stream_api/test_main.cpp | 436 +++++++++++++++++++++++++++++ 8 files changed, 675 insertions(+), 30 deletions(-) create mode 100644 src/mesh/StreamFrameWriter.cpp create mode 100644 src/mesh/StreamFrameWriter.h create mode 100644 test/test_stream_api/test_main.cpp diff --git a/src/SerialConsole.cpp b/src/SerialConsole.cpp index 2e55e31bb..d6810e123 100644 --- a/src/SerialConsole.cpp +++ b/src/SerialConsole.cpp @@ -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); } diff --git a/src/SerialConsole.h b/src/SerialConsole.h index e81ded22f..29614e0bc 100644 --- a/src/SerialConsole.h +++ b/src/SerialConsole.h @@ -2,6 +2,7 @@ #include "RedirectablePrint.h" #include "StreamAPI.h" +#include "mesh/StreamFrameWriter.h" /** * Provides both debug printing and, if the client starts sending protobufs to us, switches to send/receive protobufs * (and starts dropping debug printing - FIXME, eventually those prints should be encapsulated in protobufs). @@ -14,6 +15,7 @@ class SerialConsole : public StreamAPI, public RedirectablePrint, private concur bool usingProtobufs = false; public: + /// Initialize the shared serial stream for console and protobuf traffic. SerialConsole(); /** @@ -22,35 +24,46 @@ class SerialConsole : public StreamAPI, public RedirectablePrint, private concur */ virtual bool handleToRadio(const uint8_t *buf, size_t len) override; - virtual size_t write(uint8_t c) override - { - if (c == '\n') // prefix any newlines with carriage return - RedirectablePrint::write('\r'); - return RedirectablePrint::write(c); - } + /// Write a raw console byte unless the stream is in protobuf mode. + virtual size_t write(uint8_t c) override; + /// Service serial input, pending output, and connection state. virtual int32_t runOnce() override; + /// Flush raw console output when explicitly requested. void flush(); + /// Wake the serial thread after receive activity. void rxInt(); protected: /// Check the current underlying physical link to see if the client is currently connected virtual bool checkIsConnected() override; + /// Wake the serial thread when PhoneAPI queues output. virtual void onNowHasData(uint32_t fromRadioNum) override; /// Track serial API connect/disconnect so we can make console writes /// non-blocking while no host is listening (see setHostDraining()). virtual void onConnectionChanged(bool connected) override; - /// Possibly switch to protobufs if we see a valid protobuf message + /// Emit a framed API log when enabled, or raw output before protobuf mode. virtual void log_to_serial(const char *logLevel, const char *format, va_list arg); + /// Continue retained USB CDC output before PhoneAPI advances. + virtual bool finishPendingFrame() override; + /// Return whether the dedicated log buffer can be safely overwritten. + virtual bool canEncodeLogRecord() override; + /// Write or retain one framed USB CDC message. + virtual bool writeFrame(uint8_t *buf, size_t len, bool bestEffort) override; + private: /// On USB CDC targets, keep console TX non-blocking unless a host is draining the /// port, so a dead host can't stall the main loop and trip the task watchdog. void setHostDraining(bool draining); + +#if defined(ARDUINO_USB_CDC_ON_BOOT) && ARDUINO_USB_CDC_ON_BOOT + StreamFrameWriter frameWriter; +#endif }; // A simple wrapper to allow non class aware code write to the console diff --git a/src/mesh/StreamAPI.cpp b/src/mesh/StreamAPI.cpp index 8c28ee6af..12649e031 100644 --- a/src/mesh/StreamAPI.cpp +++ b/src/mesh/StreamAPI.cpp @@ -9,6 +9,7 @@ #define START2 0xc3 #define HEADER_LEN 4 +/// Poll the underlying stream, drain output, and update connection state. int32_t StreamAPI::runOncePart() { auto result = readStream(); @@ -17,6 +18,7 @@ int32_t StreamAPI::runOncePart() return result; } +/// Consume supplied input bytes, drain output, and update connection state. int32_t StreamAPI::runOncePart(char *buf, uint16_t bufLen) { auto result = readStream(buf, bufLen); @@ -48,6 +50,11 @@ int32_t StreamAPI::readStream(const char *buf, uint16_t bufLen) void StreamAPI::writeStream() { if (canWrite) { + // A transport that retained a short frame must complete it before + // getFromRadio() advances the PhoneAPI state to the next packet. + if (!finishPendingFrame()) + return; + uint32_t len; do { // Send every packet we can @@ -58,6 +65,7 @@ void StreamAPI::writeStream() } } +/// Parse supplied bytes through the framed ToRadio receive state machine. int32_t StreamAPI::handleRecStream(const char *buf, uint16_t bufLen) { uint16_t index = 0; @@ -167,20 +175,27 @@ int32_t StreamAPI::readStream() } } +/// Encode the stream marker and big-endian payload length. +size_t StreamAPI::buildFrameHeader(uint8_t *buf, size_t payloadLen) +{ + buf[0] = START1; + buf[1] = START2; + buf[2] = (payloadLen >> 8) & 0xff; + buf[3] = payloadLen & 0xff; + return payloadLen + HEADER_LEN; +} + /** * Send the current txBuffer over our stream */ -bool StreamAPI::writeFrame(uint8_t *buf, size_t len) +/// Write one framed payload using the transport's failure semantics. +bool StreamAPI::writeFrame(uint8_t *buf, size_t len, bool bestEffort) { + (void)bestEffort; if (len == 0 || !canWrite) return false; - buf[0] = START1; - buf[1] = START2; - buf[2] = (len >> 8) & 0xff; - buf[3] = len & 0xff; - - auto totalLen = len + HEADER_LEN; + const size_t totalLen = buildFrameHeader(buf, len); // Serialize write-readiness checks, writes and write-failure handling // against concurrent stream writes/close. concurrency::LockGuard guard(&streamLock); @@ -197,11 +212,13 @@ bool StreamAPI::writeFrame(uint8_t *buf, size_t len) return false; } +/// Emit the prepared main PhoneAPI payload as required output. bool StreamAPI::emitTxBuffer(size_t len) { - return writeFrame(txBuf, len); + return writeFrame(txBuf, len, false); } +/// Emit the initial reboot notification as a framed FromRadio payload. void StreamAPI::emitRebooted() { // In case we send a FromRadio packet @@ -213,8 +230,13 @@ void StreamAPI::emitRebooted() emitTxBuffer(pb_encode_to_bytes(txBuf + HEADER_LEN, meshtastic_FromRadio_size, &meshtastic_FromRadio_msg, &fromRadioScratch)); } +/// Encode and emit one protobuf LogRecord using the dedicated log buffers. void StreamAPI::emitLogRecord(meshtastic_LogRecord_Level level, const char *src, const char *format, va_list arg) { + // A retained short log frame still points into txBufLog, so do not overwrite it. + if (!canEncodeLogRecord()) + return; + // IMPORTANT: do NOT touch `fromRadioScratch` or `txBuf` here - those // belong to the main packet-emission path and a LOG_ firing during // `writeStream()` would corrupt an in-flight encode. We keep a @@ -237,7 +259,7 @@ void StreamAPI::emitLogRecord(meshtastic_LogRecord_Level level, const char *src, size_t len = pb_encode_to_bytes(txBufLog + HEADER_LEN, meshtastic_FromRadio_size, &meshtastic_FromRadio_msg, &fromRadioScratchLog); - writeFrame(txBufLog, len); + writeFrame(txBufLog, len, true); } /// Hookable to find out when connection changes diff --git a/src/mesh/StreamAPI.h b/src/mesh/StreamAPI.h index e23907b2b..705460532 100644 --- a/src/mesh/StreamAPI.h +++ b/src/mesh/StreamAPI.h @@ -91,12 +91,24 @@ class StreamAPI : public PhoneAPI /// Low level function to emit a protobuf encapsulated log record void emitLogRecord(meshtastic_LogRecord_Level level, const char *src, const char *format, va_list arg); + /// Return whether the transport can accept a frame of the requested size. virtual bool canWriteFrame(size_t frameLen) { return true; } + /// Let transports recover from or close after an incomplete write. virtual void onFrameWriteFailed(size_t frameLen, size_t writtenLen) {} - private: - bool writeFrame(uint8_t *buf, size_t len); + /// Fill in the 4-byte 0x94C3 length header; returns the total frame length. + static size_t buildFrameHeader(uint8_t *buf, size_t payloadLen); + /// Complete retained transport output before dequeuing another PhoneAPI packet. + virtual bool finishPendingFrame() { return true; } + /// Return whether the dedicated log buffer is available for encoding. + virtual bool canEncodeLogRecord() { return true; } + /// Frame and write a payload, optionally using best-effort admission. + virtual bool writeFrame(uint8_t *buf, size_t len, bool bestEffort); + + concurrency::Lock streamLock; + + private: /// Dedicated scratch + tx buffer for LogRecord emission. /// /// The main packet emission path (`writeStream` -> `getFromRadio` -> @@ -117,5 +129,4 @@ class StreamAPI : public PhoneAPI /// interleave on the wire. meshtastic_FromRadio fromRadioScratchLog = {}; uint8_t txBufLog[MAX_STREAM_BUF_SIZE] = {0}; - concurrency::Lock streamLock; }; diff --git a/src/mesh/StreamFrameWriter.cpp b/src/mesh/StreamFrameWriter.cpp new file mode 100644 index 000000000..a805030e8 --- /dev/null +++ b/src/mesh/StreamFrameWriter.cpp @@ -0,0 +1,65 @@ +#include "StreamFrameWriter.h" + +#include + +/// Start a frame or retain required output behind an incomplete frame. +bool StreamFrameWriter::writeFrame(Stream &stream, uint8_t *frame, size_t frameLen, bool bestEffort) +{ + if (!finishPendingFrame(stream)) { + // Single required-frame producer invariant: writeStream() is gated on + // finishPendingFrame(), so a second required frame can never arrive here. + assert(bestEffort || !deferredFrame); + + // Preserve required output that was encoded while another frame tail + // was pending. Best-effort output is dropped instead. + if (!bestEffort && !deferredFrame) { + deferredFrame = frame; + deferredFrameLen = frameLen; + } + return false; + } + + // Never start best-effort output unless its complete frame fits. + if (bestEffort && stream.availableForWrite() < (int)frameLen) + return false; + + size_t written = stream.write(frame, frameLen); + if (written == frameLen) + return true; + + pendingFrame = frame; + pendingFrameLen = frameLen; + pendingFrameOffset = written; + return false; +} + +/// Continue the pending tail with at most one Stream::write() call. +bool StreamFrameWriter::finishPendingFrame(Stream &stream) +{ + if (!pendingFrame) + return true; + + size_t remaining = pendingFrameLen - pendingFrameOffset; + size_t written = stream.write(pendingFrame + pendingFrameOffset, remaining); + if (written > remaining) + written = remaining; + pendingFrameOffset += written; + + if (pendingFrameOffset < pendingFrameLen) + return false; + + pendingFrame = nullptr; + pendingFrameLen = 0; + pendingFrameOffset = 0; + + // Promote required output without starting another write in this call. + if (deferredFrame) { + pendingFrame = deferredFrame; + pendingFrameLen = deferredFrameLen; + deferredFrame = nullptr; + deferredFrameLen = 0; + return false; + } + + return true; +} diff --git a/src/mesh/StreamFrameWriter.h b/src/mesh/StreamFrameWriter.h new file mode 100644 index 000000000..cc9c5bbcc --- /dev/null +++ b/src/mesh/StreamFrameWriter.h @@ -0,0 +1,26 @@ +#pragma once + +#include "Stream.h" +#include +#include + +/** Caller-owned frame storage must remain valid and unchanged until isIdle(). */ +class StreamFrameWriter +{ + public: + /// Start a complete frame, or retain a required frame behind pending output. + bool writeFrame(Stream &stream, uint8_t *frame, size_t frameLen, bool bestEffort); + + /// Make one bounded attempt to finish pending output. + bool finishPendingFrame(Stream &stream); + + /// Return true when no frame buffer is retained. + bool isIdle() const { return pendingFrame == nullptr && deferredFrame == nullptr; } + + private: + uint8_t *pendingFrame = nullptr; + size_t pendingFrameLen = 0; + size_t pendingFrameOffset = 0; + uint8_t *deferredFrame = nullptr; + size_t deferredFrameLen = 0; +}; diff --git a/test/native-suite-count b/test/native-suite-count index 64bb6b746..e85087aff 100644 --- a/test/native-suite-count +++ b/test/native-suite-count @@ -1 +1 @@ -30 +31 diff --git a/test/test_stream_api/test_main.cpp b/test/test_stream_api/test_main.cpp new file mode 100644 index 000000000..84613cf32 --- /dev/null +++ b/test/test_stream_api/test_main.cpp @@ -0,0 +1,436 @@ +#include "MeshTypes.h" +#include "SerialConsole.h" +#include "TestUtil.h" +#include "configuration.h" +#include "mesh-pb-constants.h" +#include "mesh/MeshService.h" +#include "mesh/StreamAPI.h" +#include "mesh/StreamFrameWriter.h" +#include +#include +#include +#include +#include +#include +#include + +/// Output-only stream whose write quotas deterministically simulate backpressure. +class ScriptedStream : public Stream +{ + public: + /// Report that no input bytes are queued. + int available() override { return 0; } + /// Return end-of-input for the output-only stream. + int read() override { return -1; } + /// Return end-of-input without consuming data. + int peek() override { return -1; } + /// Report the configured output capacity. + int availableForWrite() override { return availableCapacity; } + + /// Route single-byte writes through the quota-aware buffer writer. + size_t write(uint8_t value) override { return write(&value, 1); } + + /// Accept at most the next scripted quota and capture accepted bytes. + size_t write(const uint8_t *buffer, size_t size) override + { + requestedLengths.push_back(size); + size_t quota = size; + if (!writeQuotas.empty()) { + quota = writeQuotas.front(); + writeQuotas.pop_front(); + } + size_t accepted = std::min(quota, size); + output.insert(output.end(), buffer, buffer + accepted); + return accepted; + } + + /// Record flush calls without changing captured output. + void flush() override { flushCount++; } + + /// Set the maximum bytes accepted by the next write call. + void queueWrite(size_t quota) { writeQuotas.push_back(quota); } + + int availableCapacity = std::numeric_limits::max(); + unsigned flushCount = 0; + std::deque writeQuotas; + std::vector requestedLengths; + std::vector output; +}; + +/// Print sink that records bytes emitted by the real SerialConsole. +class RecordingPrint : public Print +{ + public: + /// Capture one output byte. + size_t write(uint8_t value) override + { + output.push_back(value); + return 1; + } + + std::vector output; +}; + +/// Installs a MeshService for a test and restores the previous global service. +class ScopedMeshService +{ + public: + /// Install the scoped service. + ScopedMeshService() : previous(service) { service = &instance; } + /// Restore the prior service after StreamAPI fixtures are destroyed. + ~ScopedMeshService() { service = previous; } + + private: + MeshService instance; + MeshService *previous; +}; + +/// Exposes generic StreamAPI hooks and records frame-write behavior. +class StreamAPITestShim : public StreamAPI +{ + public: + /// Construct the shim over a scripted stream. + explicit StreamAPITestShim(Stream *stream) : StreamAPI(stream) {} + + /// Keep connection-timeout handling inactive during tests. + bool checkIsConnected() override { return true; } + + /// Invoke the generic transport implementation rather than this shim's capture hook. + bool writeBaseFrame(uint8_t *buf, size_t len, bool bestEffort = false) { return StreamAPI::writeFrame(buf, len, bestEffort); } + + bool finishReady = true; + bool allowWrite = true; + unsigned finishCalls = 0; + unsigned frameWriteCalls = 0; + unsigned failureCalls = 0; + size_t failedFrameLen = 0; + size_t failedWrittenLen = 0; + std::vector capturedPayload; + + protected: + /// Record the pending-frame gate and return its configured state. + bool finishPendingFrame() override + { + finishCalls++; + return finishReady; + } + + /// Apply the configured generic write-readiness result. + bool canWriteFrame(size_t) override { return allowWrite; } + + /// Capture generic short-write failure metadata. + void onFrameWriteFailed(size_t frameLen, size_t writtenLen) override + { + failureCalls++; + failedFrameLen = frameLen; + failedWrittenLen = writtenLen; + } + + /// Capture one encoded PhoneAPI payload without writing it. + bool writeFrame(uint8_t *buf, size_t len, bool bestEffort) override + { + (void)bestEffort; + frameWriteCalls++; + capturedPayload.assign(buf + 4, buf + 4 + len); + return false; + } +}; + +/// Exposes framed-log hooks and records best-effort writes. +class LogHookStreamAPI : public StreamAPI +{ + public: + /// Construct the log shim over a scripted stream. + explicit LogHookStreamAPI(Stream *stream) : StreamAPI(stream) {} + + /// Keep connection-timeout handling inactive during tests. + bool checkIsConnected() override { return true; } + + /// Encode a formatted log through StreamAPI's production log path. + void emitTestLog(const char *format, ...) + { + va_list args; + va_start(args, format); + emitLogRecord(meshtastic_LogRecord_Level_INFO, "test", format, args); + va_end(args); + } + + bool allowLogEncoding = false; + unsigned frameWriteCalls = 0; + bool lastBestEffort = false; + + protected: + /// Apply the configured log-encoding gate. + bool canEncodeLogRecord() override { return allowLogEncoding; } + + /// Record whether the encoded log was marked best-effort. + bool writeFrame(uint8_t *, size_t, bool bestEffort) override + { + frameWriteCalls++; + lastBestEffort = bestEffort; + return true; + } +}; + +/// Assert byte-for-byte equality between expected and captured stream output. +static void assertBytesEqual(const std::vector &expected, const std::vector &actual) +{ + TEST_ASSERT_EQUAL_UINT(expected.size(), actual.size()); + TEST_ASSERT_EQUAL_UINT8_ARRAY(expected.data(), actual.data(), expected.size()); +} + +/// Verify retries append only the unwritten tail and reproduce the frame exactly once. +void test_frame_writer_continues_only_unwritten_tail() +{ + ScriptedStream stream; + StreamFrameWriter writer; + std::vector frame = {0x94, 0xc3, 0x00, 0x06, 1, 2, 3, 4, 5, 6}; + stream.queueWrite(3); + stream.queueWrite(2); + stream.queueWrite(frame.size()); + + TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false)); + TEST_ASSERT_FALSE(writer.isIdle()); + TEST_ASSERT_FALSE(writer.finishPendingFrame(stream)); + TEST_ASSERT_FALSE(writer.isIdle()); + TEST_ASSERT_TRUE(writer.finishPendingFrame(stream)); + TEST_ASSERT_TRUE(writer.isIdle()); + + std::vector expectedRequests = {10, 7, 5}; + TEST_ASSERT_EQUAL_UINT(expectedRequests.size(), stream.requestedLengths.size()); + TEST_ASSERT_EQUAL_UINT64_ARRAY(expectedRequests.data(), stream.requestedLengths.data(), expectedRequests.size()); + assertBytesEqual(frame, stream.output); + TEST_ASSERT_EQUAL_UINT(0, stream.flushCount); +} + +/// Verify a replacement session receives a complete old frame before its new frame. +void test_frame_writer_completes_retained_tail_before_new_session_frame() +{ + ScriptedStream stream; + StreamFrameWriter writer; + std::vector oldFrame = {0x94, 0xc3, 0x00, 0x03, 0xa1, 0xa2, 0xa3}; + std::vector newFrame = {0x94, 0xc3, 0x00, 0x02, 0xb1, 0xb2}; + stream.queueWrite(3); + stream.queueWrite(oldFrame.size()); + stream.queueWrite(newFrame.size()); + + TEST_ASSERT_FALSE(writer.writeFrame(stream, oldFrame.data(), oldFrame.size(), false)); + TEST_ASSERT_FALSE(writer.isIdle()); + + // A replacement client starts without discarding the accepted old prefix. + TEST_ASSERT_TRUE(writer.writeFrame(stream, newFrame.data(), newFrame.size(), false)); + TEST_ASSERT_TRUE(writer.isIdle()); + + std::vector expected = oldFrame; + expected.insert(expected.end(), newFrame.begin(), newFrame.end()); + assertBytesEqual(expected, stream.output); +} + +/// Verify a required main frame remains ordered behind a partial log frame. +void test_frame_writer_defers_main_behind_partial_log() +{ + ScriptedStream stream; + StreamFrameWriter writer; + std::vector logFrame = {0x94, 0xc3, 0x00, 0x02, 0xa1, 0xa2}; + std::vector mainFrame = {0x94, 0xc3, 0x00, 0x03, 0xb1, 0xb2, 0xb3}; + stream.queueWrite(2); + stream.queueWrite(0); + stream.queueWrite(logFrame.size()); + stream.queueWrite(mainFrame.size()); + + TEST_ASSERT_FALSE(writer.writeFrame(stream, logFrame.data(), logFrame.size(), true)); + TEST_ASSERT_FALSE(writer.isIdle()); + TEST_ASSERT_FALSE(writer.writeFrame(stream, mainFrame.data(), mainFrame.size(), false)); + TEST_ASSERT_FALSE(writer.isIdle()); + TEST_ASSERT_FALSE(writer.finishPendingFrame(stream)); + TEST_ASSERT_FALSE(writer.isIdle()); + TEST_ASSERT_TRUE(writer.finishPendingFrame(stream)); + TEST_ASSERT_TRUE(writer.isIdle()); + + std::vector expected = logFrame; + expected.insert(expected.end(), mainFrame.begin(), mainFrame.end()); + assertBytesEqual(expected, stream.output); + TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size()); + TEST_ASSERT_EQUAL_UINT(0, stream.flushCount); +} + +/// Verify best-effort output starts only when the complete frame fits. +void test_frame_writer_rejects_best_effort_without_full_capacity() +{ + ScriptedStream stream; + StreamFrameWriter writer; + std::vector frame = {0x94, 0xc3, 0x00, 0x02, 1, 2}; + stream.availableCapacity = frame.size() - 1; + + TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), true)); + TEST_ASSERT_TRUE(writer.isIdle()); + TEST_ASSERT_EQUAL_UINT(0, stream.requestedLengths.size()); + + stream.availableCapacity = frame.size(); + TEST_ASSERT_TRUE(writer.writeFrame(stream, frame.data(), frame.size(), true)); + TEST_ASSERT_TRUE(writer.isIdle()); + TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size()); + assertBytesEqual(frame, stream.output); +} + +/// Verify each zero-progress continuation makes one bounded write attempt. +void test_frame_writer_zero_progress_is_one_bounded_attempt() +{ + ScriptedStream stream; + StreamFrameWriter writer; + std::vector frame = {0x94, 0xc3, 0x00, 0x02, 1, 2}; + stream.queueWrite(1); + stream.queueWrite(0); + stream.queueWrite(0); + stream.queueWrite(frame.size()); + + TEST_ASSERT_FALSE(writer.writeFrame(stream, frame.data(), frame.size(), false)); + TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size()); + TEST_ASSERT_FALSE(writer.finishPendingFrame(stream)); + TEST_ASSERT_EQUAL_UINT(2, stream.requestedLengths.size()); + TEST_ASSERT_FALSE(writer.finishPendingFrame(stream)); + TEST_ASSERT_EQUAL_UINT(3, stream.requestedLengths.size()); + TEST_ASSERT_TRUE(writer.finishPendingFrame(stream)); + TEST_ASSERT_EQUAL_UINT(4, stream.requestedLengths.size()); + assertBytesEqual(frame, stream.output); +} + +/// Verify generic StreamAPI framing and successful-write flush behavior. +void test_stream_api_full_write_frames_and_flushes() +{ + ScopedMeshService scopedService; + ScriptedStream stream; + StreamAPITestShim api(&stream); + uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33}; + + TEST_ASSERT_TRUE(api.writeBaseFrame(frame, 3)); + + std::vector expected = {0x94, 0xc3, 0x00, 0x03, 0x11, 0x22, 0x33}; + assertBytesEqual(expected, stream.output); + TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size()); + TEST_ASSERT_EQUAL_UINT(1, stream.flushCount); + TEST_ASSERT_EQUAL_UINT(0, api.failureCalls); +} + +/// Verify generic transports report short writes without flushing or retrying. +void test_stream_api_short_write_reports_failure_without_flush() +{ + ScopedMeshService scopedService; + ScriptedStream stream; + StreamAPITestShim api(&stream); + uint8_t frame[7] = {0, 0, 0, 0, 0x11, 0x22, 0x33}; + stream.queueWrite(5); + + TEST_ASSERT_FALSE(api.writeBaseFrame(frame, 3)); + + TEST_ASSERT_EQUAL_UINT(1, stream.requestedLengths.size()); + TEST_ASSERT_EQUAL_UINT(0, stream.flushCount); + TEST_ASSERT_EQUAL_UINT(1, api.failureCalls); + TEST_ASSERT_EQUAL_UINT(7, api.failedFrameLen); + TEST_ASSERT_EQUAL_UINT(5, api.failedWrittenLen); +} + +/// Verify retained output blocks PhoneAPI from dequeuing the next payload. +void test_stream_api_finishes_pending_before_advancing_phone_api() +{ + ScopedMeshService scopedService; + ScriptedStream stream; + StreamAPITestShim api(&stream); + api.sendConfigComplete(); + api.sendNotification(meshtastic_LogRecord_Level_WARNING, 42, "still queued"); + api.finishReady = false; + + api.runOncePart(nullptr, 0); + TEST_ASSERT_EQUAL_UINT(1, api.finishCalls); + TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls); + + api.finishReady = true; + api.runOncePart(nullptr, 0); + TEST_ASSERT_EQUAL_UINT(2, api.finishCalls); + TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls); + + meshtastic_FromRadio decoded = meshtastic_FromRadio_init_zero; + TEST_ASSERT_TRUE( + pb_decode_from_bytes(api.capturedPayload.data(), api.capturedPayload.size(), &meshtastic_FromRadio_msg, &decoded)); + TEST_ASSERT_EQUAL_UINT(meshtastic_FromRadio_clientNotification_tag, decoded.which_payload_variant); + TEST_ASSERT_EQUAL_UINT32(42, decoded.clientNotification.reply_id); +} + +/// Verify framed logs honor the encoding gate and use best-effort writes. +void test_stream_api_gates_logs_and_marks_them_best_effort() +{ + ScopedMeshService scopedService; + ScriptedStream stream; + LogHookStreamAPI api(&stream); + + api.emitTestLog("blocked %u", 1U); + TEST_ASSERT_EQUAL_UINT(0, api.frameWriteCalls); + + api.allowLogEncoding = true; + api.emitTestLog("allowed %u", 2U); + TEST_ASSERT_EQUAL_UINT(1, api.frameWriteCalls); + TEST_ASSERT_TRUE(api.lastBestEffort); +} + +/// Verify the real SerialConsole emits no unframed bytes in protobuf mode. +void test_serial_console_suppresses_raw_output_in_protobuf_mode() +{ + RecordingPrint sink; + const bool oldHasLora = config.has_lora; + const bool oldHasSecurity = config.has_security; + const bool oldSerialEnabled = config.security.serial_enabled; + const bool oldDebugLogApiEnabled = config.security.debug_log_api_enabled; + + config.has_lora = true; + config.has_security = true; + config.security.serial_enabled = true; + config.security.debug_log_api_enabled = false; + console->setDestination(&sink); + + console->write('A'); + const bool rawBeforeProtobuf = sink.output.size() == 1 && sink.output[0] == 'A'; + sink.output.clear(); + + const uint8_t emptyToRadio = 0; + console->handleToRadio(&emptyToRadio, 0); + console->write('B'); + console->write('\n'); + console->log(MESHTASTIC_LOG_LEVEL_ERROR, "must stay framed"); + const bool emptyAfterProtobuf = sink.output.empty(); + + console->setDestination(&Serial); + config.has_lora = oldHasLora; + config.has_security = oldHasSecurity; + config.security.serial_enabled = oldSerialEnabled; + config.security.debug_log_api_enabled = oldDebugLogApiEnabled; + + TEST_ASSERT_TRUE(rawBeforeProtobuf); + TEST_ASSERT_TRUE(emptyAfterProtobuf); +} + +/// Unity per-test setup; fixtures are local to each test. +void setUp(void) {} +/// Unity per-test teardown; fixtures clean themselves up. +void tearDown(void) {} + +/// Initialize the native environment and run the stream regression suite. +void setup() +{ + initializeTestEnvironment(); + UNITY_BEGIN(); + RUN_TEST(test_frame_writer_continues_only_unwritten_tail); + RUN_TEST(test_frame_writer_completes_retained_tail_before_new_session_frame); + RUN_TEST(test_frame_writer_defers_main_behind_partial_log); + RUN_TEST(test_frame_writer_rejects_best_effort_without_full_capacity); + RUN_TEST(test_frame_writer_zero_progress_is_one_bounded_attempt); + RUN_TEST(test_stream_api_full_write_frames_and_flushes); + RUN_TEST(test_stream_api_short_write_reports_failure_without_flush); + RUN_TEST(test_stream_api_finishes_pending_before_advancing_phone_api); + RUN_TEST(test_stream_api_gates_logs_and_marks_them_best_effort); + // usingProtobufs intentionally has no reset path, so this must run last. + RUN_TEST(test_serial_console_suppresses_raw_output_in_protobuf_mode); + exit(UNITY_END()); +} + +/// Unused Arduino loop required by the native Unity runner. +void loop() {}