Keep hardware CDC packet log lines whole under congestion

src/helpers/SerialPacketLog.h

@@ -32,6 +32,17 @@ inline bool& serialLogPortSeen() {
 
 template <class T>
 bool serialLogEmit(T& out, const char* data, size_t len) {
+#if defined(ESP32_PLATFORM) && defined(ARDUINO_USB_CDC_ON_BOOT) && \
+    (ARDUINO_USB_CDC_ON_BOOT == 1) && defined(ARDUINO_USB_MODE) && \
+    (ARDUINO_USB_MODE == 1)
+  // The HWCDC logging view checks this complete record again under the shared
+  // producer guard. Never leave a line prefix behind when its host stalls.
+  const int space = out.availableForWrite();
+  if (space <= 0 || static_cast<size_t>(space) < len) return false;
+  if (out.write(reinterpret_cast<const uint8_t*>(data), len) != len) return false;
+  serialLogPortSeen() = true;
+  return true;
+#else
   if (out.availableForWrite() <= 0) return false;
 
   const uint32_t start = millis();
@@ -55,6 +66,7 @@ bool serialLogEmit(T& out, const char* data, size_t len) {
   }
   serialLogPortSeen() = true;
   return true;
+#endif
 }
 
 template <size_t CAP = SERIAL_LOG_LINE_MAX>

src/helpers/UsbLogging.cpp

@@ -513,6 +513,11 @@ static SingleAttemptNonBlockingStream guarded_esp32_hwcdc_port(
 static AtomicWholeRecordNonBlockingStream<11>
     guarded_esp32_hwcdc_mota_port(guarded_esp32_hwcdc_port);
 
+// Admit each complete packet log line under the same guard as CLI replies.
+// Short records can still use a smaller ring if the preferred allocation fails.
+static AtomicWholeRecordNonBlockingStream<640> guarded_esp32_hwcdc_logging_port(
+    guarded_esp32_hwcdc_port);
+
 static void serviceEsp32HwcdcTxKickExclusive(void*) {
   if (!esp32_hwcdc_tx_kick_pending.load(std::memory_order_acquire)
       || !canAccessEsp32Hwcdc(nullptr)) {
@@ -1393,7 +1398,7 @@ Stream& usbLoggingPort() {
   #if MESH_ESP32_TINYUSB_NONBLOCKING
     return buffered_esp32_tinyusb_logging_port;
   #elif MESH_ESP32_HWCDC_SESSION_GUARD
-    return guarded_esp32_hwcdc_port;
+    return guarded_esp32_hwcdc_logging_port;
   #elif defined(NRF52_PLATFORM)
     return nonblocking_primary_usb_logging_port;
   #else

test/fixtures/hwcdc_write_capacity.cpp

@@ -88,6 +88,8 @@ static size_t once(void*, const uint8_t* bytes, size_t size) {
 }
 static bool canAccess(void*) { return access_allowed; }
 static SingleAttemptNonBlockingStream guarded(session, once, nullptr, canAccess);
+@LOGGING_VIEW@
+static LoggingView logging(guarded);
 }
 static const uint8_t bytes[] = "abcdefghijklmnopqrstuvwx";
 
@@ -179,6 +181,91 @@ int main(int argc, char** argv) {
     ring_free = 5;
     assert(Serial.driverWrite(bytes, 24) == 5);
     assert(g_mock_millis == 5 && zero_sends == 5 && !connected);
+  } else if (name == "logging_low") {
+    const std::vector<uint8_t> line(640, 'L');
+    ring_free = 639;
+    assert(!mesh::serialLogEmit(mesh::logging,
+                               reinterpret_cast<const char*>(line.data()), line.size()));
+    assert(admitted.empty() && sdk_calls == 0 && ring_sends == 0);
+    assert(g_mock_millis == 0 && zero_sends == 0);
+  } else if (name == "logging_full") {
+    const std::vector<uint8_t> line(640, 'L');
+    ring_free = 640;
+    assert(mesh::logging.availableForWrite() == 640);
+    assert(mesh::serialLogEmit(mesh::logging,
+                              reinterpret_cast<const char*>(line.data()), line.size()));
+    assert(admitted == line && sdk_calls == 1 && ring_sends == 1);
+    assert(g_mock_millis == 0 && zero_sends == 0);
+  } else if (name == "logging_fallback_small") {
+    const std::vector<uint8_t> line(100, 'L');
+    ring_free = 256;
+    assert(mesh::logging.availableForWrite() == 256);
+    assert(mesh::serialLogEmit(mesh::logging,
+                              reinterpret_cast<const char*>(line.data()), line.size()));
+    assert(admitted == line && sdk_calls == 1 && ring_sends == 1);
+    assert(g_mock_millis == 0 && zero_sends == 0);
+  } else if (name == "logging_fallback_large") {
+    const std::vector<uint8_t> line(257, 'L');
+    ring_free = 256;
+    assert(!mesh::serialLogEmit(mesh::logging,
+                               reinterpret_cast<const char*>(line.data()), line.size()));
+    assert(admitted.empty() && sdk_calls == 0 && ring_sends == 0);
+    assert(g_mock_millis == 0 && zero_sends == 0);
+  } else if (name == "logging_stale") {
+    struct StaleView : Stream {
+      int availableForWrite() override { return mesh::logging.availableForWrite(); }
+      size_t write(const uint8_t* data, size_t size) override {
+        // A different facade uses capacity after serialLogEmit's outer sample.
+        assert(mesh::guarded.write(bytes, 24) == 24);
+        return mesh::logging.write(data, size);
+      }
+    } stale;
+    const std::vector<uint8_t> line(640, 'L');
+    ring_free = 640;
+    assert(!mesh::serialLogEmit(stale,
+                               reinterpret_cast<const char*>(line.data()), line.size()));
+    assert(admitted == std::vector<uint8_t>(bytes, bytes + 24));
+    assert(sdk_calls == 1 && ring_sends == 1 && g_mock_millis == 0);
+  } else if (name == "logging_line") {
+    ring_free = 640;
+    mesh::SerialLogLine<> line;
+    line.printf("STREAM_RX: ");
+    line.hex(bytes, 24);
+    assert(line.flush(mesh::logging, false));
+    const std::string expected =
+        "STREAM_RX: 6162636465666768696A6B6C6D6E6F707172737475767778\n";
+    assert(admitted == std::vector<uint8_t>(expected.begin(), expected.end()));
+    assert(sdk_calls == 1 && ring_sends == 1 && g_mock_millis == 0);
+  } else if (name == "logging_contention") {
+    ring_free = 640;
+    const std::vector<uint8_t> first(640, 'A'), second(640, 'B');
+    std::mutex mutex;
+    std::condition_variable event;
+    bool entered = false, proceed = false;
+    before_sdk_write = [&] {
+      std::unique_lock<std::mutex> lock(mutex);
+      entered = true;
+      event.notify_all();
+      assert(event.wait_for(lock, std::chrono::seconds(2), [&] { return proceed; }));
+    };
+    std::thread writer([&] {
+      assert(mesh::serialLogEmit(mesh::logging,
+                                reinterpret_cast<const char*>(first.data()), first.size()));
+    });
+    {
+      std::unique_lock<std::mutex> lock(mutex);
+      assert(event.wait_for(lock, std::chrono::seconds(2), [&] { return entered; }));
+    }
+    assert(!mesh::serialLogEmit(mesh::logging,
+                               reinterpret_cast<const char*>(second.data()), second.size()));
+    {
+      std::lock_guard<std::mutex> lock(mutex);
+      proceed = true;
+    }
+    event.notify_all();
+    writer.join();
+    assert(admitted == first && sdk_calls == 1 && ring_sends == 1);
+    assert(g_mock_millis == 0 && zero_sends == 0);
   } else {
     assert(false && "unknown fixture case");
   }

test/test_esp32_usb_serial_hygiene.py

@@ -354,7 +354,16 @@ int main() {
         self.assertIn(
             "esp32_hwcdc_tx_kick_pending.store(true", guarded_write
         )
-        self.assertIn("return Serial.write(data, size);", guarded_write)
+        self.assertIn("return Serial.write(data, attempt);", guarded_write)
+        capacity = guarded_write.index("const int available = Serial.availableForWrite();")
+        full = guarded_write.index("if (available <= 0) return 0;")
+        clamp = guarded_write.index("const size_t attempt = size <")
+        kick_pending = guarded_write.index("esp32_hwcdc_tx_kick_pending.store(true")
+        native_write = guarded_write.index("return Serial.write(data, attempt);")
+        self.assertLess(capacity, full)
+        self.assertLess(full, clamp)
+        self.assertLess(clamp, kick_pending)
+        self.assertLess(kick_pending, native_write)
 
         # Extract complete functions: their pinned-SDK/fallback branches now
         # contain nested #endif directives inside the outer platform guard.

test/test_hwcdc_write_capacity.py

@@ -2,6 +2,7 @@
 """Execute the HWCDC facade and pinned SDK write against a stalled host."""
 from pathlib import Path
 import os
+import re
 import shutil
 import subprocess
 import tempfile
@@ -26,12 +27,17 @@ class HwcdcWriteCapacityTest(unittest.TestCase):
         sdk_write = body(PATCHED, 'size_t HWCDC::write(const uint8_t *buffer, size_t size)')
         sdk_write = sdk_write.replace('HWCDC::write(', 'HWCDC::driverWrite(', 1)
         sdk_available = body(PATCHED, 'int HWCDC::availableForWrite(void)')
+        logging = re.search(r'static\s+(AtomicWholeRecordNonBlockingStream<640>)\s+'
+                            r'guarded_esp32_hwcdc_logging_port\(', source)
+        self.assertIsNotNone(logging, 'use the actual HWCDC logging view type')
         return (FIXTURE.read_text()
                 .replace('@SDK_WRITE@', sdk_write)
                 .replace('@SDK_AVAILABLE@', sdk_available)
-                .replace('@FACADE@', facade))
+                .replace('@FACADE@', facade)
+                .replace('@LOGGING_VIEW@', 'using LoggingView = ' + logging[1] + ';'))
 
-    def compile(self, directory, uncapped=False, zero_timeout=False):
+    def compile(self, directory, uncapped=False, zero_timeout=False,
+                partial_logging=False):
         compiler = os.environ.get('CXX') or shutil.which('g++') or shutil.which('clang++')
         if compiler is None:
             self.skipTest('a host C++17 compiler is required')
@@ -40,14 +46,26 @@ class HwcdcWriteCapacityTest(unittest.TestCase):
         flags = ['-fsanitize=address,undefined', '-fno-sanitize-recover=all',
                  '-fno-pie', '-no-pie'] if os.name != 'nt' else []
         includes = ['-I' + str(ROOT / 'test/mocks'), '-I' + str(ROOT / 'src')]
-        if zero_timeout:
+        if zero_timeout or partial_logging:
             copied = Path(directory) / 'helpers'
             copied.mkdir()
             packet_log = (ROOT / 'src/helpers/SerialPacketLog.h').read_text()
-            self.assertIn('Serial.setTxTimeoutMs(5);', packet_log)
+            if zero_timeout:
+                self.assertIn('Serial.setTxTimeoutMs(5);', packet_log)
+                packet_log = packet_log.replace('Serial.setTxTimeoutMs(5);',
+                                                'Serial.setTxTimeoutMs(0);')
+            if partial_logging:
+                # Restore the actual old non-HWCDC chunked emitter, preserving
+                # the rest of this copied production header byte-for-byte.
+                function = body(packet_log, 'bool serialLogEmit(')
+                start = function.index('#if defined(ESP32_PLATFORM)')
+                old_branch = function.index('\n#else\n')
+                end = function.rindex('\n#endif')
+                restored = (function[:start]
+                            + function[old_branch + len('\n#else\n'):end] + '\n}')
+                packet_log = packet_log.replace(function, restored)
             (copied / 'SerialPacketLog.h').write_text(
-                packet_log.replace('Serial.setTxTimeoutMs(5);',
-                                   'Serial.setTxTimeoutMs(0);'), encoding='ascii')
+                packet_log, encoding='ascii')
             includes.insert(0, '-I' + str(directory))
         binary = Path(directory) / 'test.exe'
         built = subprocess.run([
@@ -62,7 +80,10 @@ class HwcdcWriteCapacityTest(unittest.TestCase):
         with tempfile.TemporaryDirectory(prefix='hwcdc-capacity-') as directory:
             binary = self.compile(directory)
             for case in ('full', 'oversize', 'stale', 'contention', 'mutex',
-                         'isr_drain', 'short', 'mota', 'timeout', 'sdk_fallback'):
+                         'isr_drain', 'short', 'mota', 'timeout', 'sdk_fallback',
+                         'logging_low', 'logging_full', 'logging_line', 'logging_stale',
+                         'logging_contention', 'logging_fallback_small',
+                         'logging_fallback_large'):
                 with self.subTest(case=case):
                     result = subprocess.run([str(binary), case], capture_output=True,
                                             text=True, timeout=10)
@@ -82,6 +103,13 @@ class HwcdcWriteCapacityTest(unittest.TestCase):
                                     text=True, timeout=10)
             self.assertNotEqual(result.returncode, 0)
 
+    def test_partial_logging_capacity_negative_control(self):
+        with tempfile.TemporaryDirectory(prefix='hwcdc-partial-log-') as directory:
+            binary = self.compile(directory, partial_logging=True)
+            result = subprocess.run([str(binary), 'logging_low'], capture_output=True,
+                                    text=True, timeout=10)
+            self.assertNotEqual(result.returncode, 0)
+
 
 if __name__ == '__main__':
     unittest.main(verbosity=2)