[usb_uart] Fix FTDI RX data stall / corruption and input restart reliability (#17348)

Co-authored-by: Oliver Kleinecke <kleinecke.oliver@googlemail.com>
Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

esphome/components/usb_uart/ft23xx.cpp

@@ -3,6 +3,7 @@
 #include "usb_uart.h"
 #include "usb/usb_host.h"
 #include "esphome/core/log.h"
+#include "esphome/core/application.h"
 #include "esphome/components/uart/uart_debugger.h"
 
 #include "esphome/components/bytebuffer/bytebuffer.h"
@@ -396,7 +397,14 @@ int USBUartTypeFT23XX::set_dtr_rts_(USBUartChannel *channel) {
 }
 
 void USBUartTypeFT23XX::start_input(USBUartChannel *channel) {
-  if (!channel->initialised_.load() || channel->input_started_.load())
+  if (!channel->initialised_.load())
+    return;
+
+  // Use compare_exchange_strong to avoid a check-then-act race: start_input() is called
+  // from both the USB task (self-restart on success) and the main loop (backpressure
+  // restart), so a plain load()/store() pair can let both threads submit a transfer.
+  auto started = false;
+  if (!channel->input_started_.compare_exchange_strong(started, true))
     return;
 
   const auto *ep = channel->cdc_dev_.in_ep;
@@ -408,39 +416,55 @@ void USBUartTypeFT23XX::start_input(USBUartChannel *channel) {
       return;
     }
 
+    // FTDI prepends a 2-byte modem/line status header to every bulk IN packet.
     size_t uart_data_len = (status.data_len > 2) ? (status.data_len - 2) : 0;
 
     if (uart_data_len > 0) {
       ESP_LOGV(TAG, "RX callback: Received %zu bytes, channel=%d", uart_data_len, channel->index_);
       if (!channel->dummy_receiver_) {
-        // Copy the entire received UART payload into the ring buffer in one
-        // operation to avoid per-byte overhead and reduce the chance of
-        // heap activity in hot paths.
-        channel->input_buffer_.push(status.data + 2, uart_data_len);
+        UsbDataChunk *chunk = this->chunk_pool_.allocate();
+        if (chunk == nullptr) {
+          this->usb_data_queue_.increment_dropped_count();
+          channel->input_started_.store(false);
+          // Queue is full — wake the main loop to drain it, then let read_array()
+          // retrigger start_input() rather than spinning here in the USB task.
+          this->enable_loop_soon_any_context();
+          App.wake_loop_threadsafe();
+          return;
+        }
+        // Strip the 2-byte FTDI header before queuing.
+        memcpy(chunk->data, status.data + 2, uart_data_len);
+        chunk->length = static_cast<uint16_t>(uart_data_len);
+        chunk->channel = channel;
+        this->usb_data_queue_.push(chunk);
 #ifdef USE_UART_DEBUGGER
         if (channel->debug_) {
-          // Debug path creates a temporary vector for logging only; this is
-          // acceptable because debug mode is opt-in and not used in release.
           uart::UARTDebug::log_hex(uart::UART_DIRECTION_RX,
                                    std::vector<uint8_t>(status.data + 2, status.data + 2 + uart_data_len), ',',
                                    channel->debug_prefix_);
         }
 #endif
+        this->enable_loop_soon_any_context();
+        App.wake_loop_threadsafe();
       }
-    } else {
+    } else if (status.data_len >= 2) {
       ESP_LOGVV(TAG, "RX: Status packet, modem=0x%02X line=0x%02X, ch=%d", status.data[0], status.data[1],
                 channel->index_);
     }
 
     channel->input_started_.store(false);
-    if (channel->dummy_receiver_ ||
-        channel->input_buffer_.get_free_space() >= channel->cdc_dev_.in_ep->wMaxPacketSize) {
-      this->start_input(channel);
-    }
+    this->start_input(channel);
   };
 
-  channel->input_started_.store(true);
-  this->transfer_in(ep->bEndpointAddress, callback, ep->wMaxPacketSize);
+  if (!this->transfer_in(ep->bEndpointAddress, callback, ep->wMaxPacketSize)) {
+    ESP_LOGE(TAG, "RX transfer submission failed for ep=0x%02X", ep->bEndpointAddress);
+    channel->input_started_.store(false);
+  }
+}
+
+void USBUartTypeFT23XX::on_rx_overflow(USBUartChannel *channel) {
+  ESP_LOGW(TAG, "RX buffer overflow on channel %d, clearing to resync", channel->index_);
+  channel->input_buffer_.clear();
 }
 
 void USBUartTypeFT23XX::enable_channels() {

esphome/components/usb_uart/usb_uart.cpp

@@ -228,6 +228,11 @@ void USBUartComponent::loop() {
     }
 #endif
 
+    // If there is not enough space for the full chunk, let the device subclass
+    // handle it (e.g. FTDI clears the buffer to prevent mid-telegram corruption).
+    if (channel->input_buffer_.get_free_space() < chunk->length) {
+      this->on_rx_overflow(channel);
+    }
     // Push data to ring buffer (now safe in main loop)
     channel->input_buffer_.push(chunk->data, chunk->length);
 

esphome/components/usb_uart/usb_uart.h

@@ -192,9 +192,13 @@ class USBUartComponent : public usb_host::USBClient {
 
   void add_channel(USBUartChannel *channel) { this->channels_.push_back(channel); }
 
-  void start_input(USBUartChannel *channel);
+  virtual void start_input(USBUartChannel *channel);
   void start_output(USBUartChannel *channel);
 
+  // Called from loop() when input_buffer_ has insufficient space for the incoming chunk.
+  // Default is a no-op; override in device-specific subclasses that need resync on overflow.
+  virtual void on_rx_overflow(USBUartChannel *channel) {}
+
   // Lock-free data transfer from USB task to main loop
   static constexpr int USB_DATA_QUEUE_SIZE = 32;
   LockFreeQueue<UsbDataChunk, USB_DATA_QUEUE_SIZE> usb_data_queue_;
@@ -248,7 +252,8 @@ class USBUartTypeFT23XX : public USBUartTypeCdcAcm {
  public:
   USBUartTypeFT23XX(uint16_t vid, uint16_t pid) : USBUartTypeCdcAcm(vid, pid) {}
 
-  void start_input(USBUartChannel *channel);
+  void start_input(USBUartChannel *channel) override;
+  void on_rx_overflow(USBUartChannel *channel) override;
 
  protected:
   std::vector<CdcEps> parse_descriptors(usb_device_handle_t dev_hdl) override;