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() {}