Merge remote-tracking branch 'upstream/api-logs-decode-noraise' into integration

esphome/components/api/client.py

@@ -18,7 +18,7 @@ with warnings.catch_warnings():
 import contextlib
 
 from esphome.const import CONF_KEY, CONF_PORT, __version__
-from esphome.core import CORE
+from esphome.core import CORE, EsphomeError
 from esphome.platformio_api import process_stacktrace
 
 from . import CONF_ENCRYPTION
@@ -32,6 +32,49 @@ if TYPE_CHECKING:
 _LOGGER = logging.getLogger(__name__)
 
 
+class _LogLineProcessor:
+    """Feeds incoming log lines to the stack-trace decoder.
+
+    Two responsibilities beyond just calling the decoder:
+    1. Catch EsphomeError. on_log runs inside an asyncio protocol
+       callback; if an exception escapes, the loop tears the transport
+       down with "Fatal error: protocol.data_received() call failed."
+       and ReconnectLogic immediately reconnects, the device replays
+       the same crash trace, and we loop forever.
+    2. Disable decoding after the first failure. _decode_pc shells out
+       to PlatformIO via _run_idedata, which is expensive; a single
+       crash dump can contain many PC/BT lines and we don't want to
+       retry the failing subprocess for each one.
+    """
+
+    def __init__(self, config: dict[str, Any], platform_handler: Any | None) -> None:
+        self._config = config
+        self._platform_handler = platform_handler
+        self._decode_enabled = True
+        self.backtrace_state = False
+
+    def process_line(self, raw_line: str) -> None:
+        if not self._decode_enabled:
+            return
+        try:
+            if self._platform_handler is not None:
+                self.backtrace_state = self._platform_handler(
+                    self._config, raw_line, self.backtrace_state
+                )
+            else:
+                self.backtrace_state = process_stacktrace(
+                    self._config, raw_line, backtrace_state=self.backtrace_state
+                )
+        except EsphomeError as exc:
+            self._decode_enabled = False
+            self.backtrace_state = False
+            _LOGGER.warning(
+                "Crash trace decoding unavailable (%s). Run "
+                "'esphome compile' for this device to enable PC decoding.",
+                exc,
+            )
+
+
 async def async_run_logs(
     config: dict[str, Any],
     addresses: list[str],
@@ -61,7 +104,6 @@ async def async_run_logs(
         addresses=addresses,  # Pass all addresses for automatic retry
     )
     dashboard = CORE.dashboard
-    backtrace_state = False
 
     # Try platform-specific stacktrace handler first, fall back to generic
     platform_process_stacktrace = None
@@ -71,9 +113,10 @@ async def async_run_logs(
     except (AttributeError, ImportError):
         pass
 
+    processor = _LogLineProcessor(config, platform_process_stacktrace)
+
     def on_log(msg: SubscribeLogsResponse) -> None:
         """Handle a new log message."""
-        nonlocal backtrace_state
         time_ = datetime.now()
         message: bytes = msg.message
         text = message.decode("utf8", "backslashreplace")
@@ -84,14 +127,7 @@ async def async_run_logs(
         for parsed_msg in parse_log_message(text, timestamp):
             print(parsed_msg.replace("\033", "\\033") if dashboard else parsed_msg)
         for raw_line in text.splitlines():
-            if platform_process_stacktrace:
-                backtrace_state = platform_process_stacktrace(
-                    config, raw_line, backtrace_state
-                )
-            else:
-                backtrace_state = process_stacktrace(
-                    config, raw_line, backtrace_state=backtrace_state
-                )
+            processor.process_line(raw_line)
 
     # Safe to fall back to plaintext here only for this diagnostics use
     # case: the stream is one-way from device to client, and this code

esphome/components/usb_host/__init__.py

@@ -10,6 +10,7 @@ from esphome.components.esp32 import (
 )
 import esphome.config_validation as cv
 from esphome.const import CONF_DEVICES, CONF_ID
+from esphome.core import CORE
 from esphome.cpp_types import Component
 from esphome.types import ConfigType
 
@@ -19,14 +20,15 @@ DEPENDENCIES = ["esp32"]
 usb_host_ns = cg.esphome_ns.namespace("usb_host")
 USBHost = usb_host_ns.class_("USBHost", Component)
 USBClient = usb_host_ns.class_("USBClient", Component)
-
+DOMAIN = "usb_host"
 CONF_VID = "vid"
 CONF_PID = "pid"
 CONF_ENABLE_HUBS = "enable_hubs"
 CONF_MAX_TRANSFER_REQUESTS = "max_transfer_requests"
+CONF_MAX_PACKET_SIZE = "max_packet_size"
 
 
-def usb_device_schema(cls=USBClient, vid: int = None, pid: [int] = None) -> cv.Schema:
+def usb_device_schema(cls=USBClient, vid: int = None, pid: int = None) -> cv.Schema:
     schema = cv.COMPONENT_SCHEMA.extend(
         {
             cv.GenerateID(): cv.declare_id(cls),
@@ -43,6 +45,17 @@ def usb_device_schema(cls=USBClient, vid: int = None, pid: [int] = None) -> cv.S
     return schema
 
 
+def _set_max_packet_size(config: dict) -> dict:
+    CORE.data.setdefault(DOMAIN, {})[CONF_MAX_PACKET_SIZE] = config[
+        CONF_MAX_PACKET_SIZE
+    ]
+    return config
+
+
+def get_max_packet_size() -> int:
+    return CORE.data.get(DOMAIN, {}).get(CONF_MAX_PACKET_SIZE, 64)
+
+
 CONFIG_SCHEMA = cv.All(
     cv.COMPONENT_SCHEMA.extend(
         {
@@ -51,10 +64,14 @@ CONFIG_SCHEMA = cv.All(
             cv.Optional(CONF_MAX_TRANSFER_REQUESTS, default=16): cv.int_range(
                 min=1, max=32
             ),
+            cv.Optional(CONF_MAX_PACKET_SIZE, default=64): cv.one_of(
+                64, 128, 256, 512, 1024, int=True
+            ),
             cv.Optional(CONF_DEVICES): cv.ensure_list(usb_device_schema()),
         }
     ),
     only_on_variant(supported=[VARIANT_ESP32P4, VARIANT_ESP32S2, VARIANT_ESP32S3]),
+    _set_max_packet_size,
 )
 
 
@@ -72,8 +89,8 @@ async def to_code(config: ConfigType) -> None:
     if config.get(CONF_ENABLE_HUBS):
         add_idf_sdkconfig_option("CONFIG_USB_HOST_HUBS_SUPPORTED", True)
 
-    max_requests = config[CONF_MAX_TRANSFER_REQUESTS]
-    cg.add_define("USB_HOST_MAX_REQUESTS", max_requests)
+    cg.add_define("USB_HOST_MAX_REQUESTS", config[CONF_MAX_TRANSFER_REQUESTS])
+    cg.add_define("USB_HOST_MAX_PACKET_SIZE", config[CONF_MAX_PACKET_SIZE])
 
     var = cg.new_Pvariable(config[CONF_ID])
     await cg.register_component(var, config)

esphome/components/usb_host/usb_host.h

@@ -66,6 +66,8 @@ static_assert(MAX_REQUESTS >= 1 && MAX_REQUESTS <= 32, "MAX_REQUESTS must be bet
 using trq_bitmask_t = std::conditional<(MAX_REQUESTS <= 16), uint16_t, uint32_t>::type;
 static constexpr trq_bitmask_t ALL_REQUESTS_IN_USE = MAX_REQUESTS == 32 ? ~0 : (1 << MAX_REQUESTS) - 1;
 
+static constexpr size_t USB_MAX_PACKET_SIZE =
+    USB_HOST_MAX_PACKET_SIZE;                        // Max USB packet size (64 for FS, 512 for P4 HS)
 static constexpr size_t USB_EVENT_QUEUE_SIZE = 32;   // Size of event queue between USB task and main loop
 static constexpr size_t USB_TASK_STACK_SIZE = 4096;  // Stack size for USB task (same as ESP-IDF USB examples)
 static constexpr UBaseType_t USB_TASK_PRIORITY = 5;  // Higher priority than main loop (tskIDLE_PRIORITY + 5)

esphome/components/usb_host/usb_host_client.cpp

@@ -217,7 +217,7 @@ void USBClient::setup() {
   // Pre-allocate USB transfer buffers for all slots at startup
   // This avoids any dynamic allocation during runtime
   for (auto &request : this->requests_) {
-    usb_host_transfer_alloc(64, 0, &request.transfer);
+    usb_host_transfer_alloc(USB_MAX_PACKET_SIZE, 0, &request.transfer);
     request.client = this;  // Set once, never changes
   }
 

esphome/components/usb_uart/__init__.py

@@ -1,7 +1,11 @@
 import esphome.codegen as cg
 from esphome.components.const import CONF_DATA_BITS, CONF_PARITY, CONF_STOP_BITS
 from esphome.components.uart import CONF_DEBUG_PREFIX, CONF_FLUSH_TIMEOUT, UARTComponent
-from esphome.components.usb_host import register_usb_client, usb_device_schema
+from esphome.components.usb_host import (
+    get_max_packet_size,
+    register_usb_client,
+    usb_device_schema,
+)
 import esphome.config_validation as cv
 from esphome.const import (
     CONF_BAUD_RATE,
@@ -118,14 +122,14 @@ CONFIG_SCHEMA = cv.ensure_list(
 async def to_code(config):
     # The output chunk pool/queue are compile-time-sized templates shared by all
     # USBUartChannel instances, so use the largest buffer_size across every channel
-    # of every device.  Each chunk is 64 bytes (USB FS MPS); add one extra slot
-    # because LockFreeQueue<T,N> is a ring buffer that wastes one entry.
+    # of every device. Add one extra slot because LockFreeQueue<T,N> is a ring
+    # buffer that wastes one entry.
     max_buffer_size = max(
         channel[CONF_BUFFER_SIZE]
         for device in config
         for channel in device[CONF_CHANNELS]
     )
-    output_chunk_count = max_buffer_size // 64 + 1
+    output_chunk_count = max(max_buffer_size // get_max_packet_size(), 2) + 1
     cg.add_define("USB_UART_OUTPUT_CHUNK_COUNT", output_chunk_count)
 
     for device in config:

esphome/components/usb_uart/usb_uart.cpp

@@ -157,7 +157,7 @@ void USBUartChannel::write_array(const uint8_t *data, size_t len) {
       ESP_LOGE(TAG, "Output pool full - lost %zu bytes", len);
       break;
     }
-    size_t chunk_len = std::min(len, UsbOutputChunk::MAX_CHUNK_SIZE);
+    uint16_t chunk_len = std::min(len, UsbOutputChunk::MAX_CHUNK_SIZE);
     memcpy(chunk->data, data, chunk_len);
     chunk->length = static_cast<uint8_t>(chunk_len);
     // Push always succeeds: pool is sized to queue capacity (SIZE-1), so if
@@ -222,7 +222,7 @@ void USBUartComponent::loop() {
 
 #ifdef USE_UART_DEBUGGER
     if (channel->debug_) {
-      char buf[4 + format_hex_pretty_size(UsbDataChunk::MAX_CHUNK_SIZE)];  // "<<< " + hex
+      char buf[4 + format_hex_pretty_size(usb_host::USB_MAX_PACKET_SIZE)];  // "<<< " + hex
       memcpy(buf, "<<< ", 4);
       format_hex_pretty_to(buf + 4, sizeof(buf) - 4, chunk->data, chunk->length, ',');
       ESP_LOGD(TAG, "%s%s", channel->debug_prefix_.c_str(), buf);
@@ -377,7 +377,7 @@ void USBUartComponent::start_output(USBUartChannel *channel) {
     this->start_output(channel);
   };
 
-  const uint8_t len = chunk->length;
+  const auto len = chunk->length;
   if (!this->transfer_out(ep->bEndpointAddress, callback, chunk->data, len)) {
     // Transfer submission failed — return chunk and release flag so callers can retry.
     channel->output_pool_.release(chunk);
@@ -394,10 +394,10 @@ void USBUartComponent::start_output(USBUartChannel *channel) {
 static void fix_mps(const usb_ep_desc_t *ep) {
   if (ep != nullptr) {
     auto *ep_mutable = const_cast<usb_ep_desc_t *>(ep);
-    if (ep->wMaxPacketSize > 64) {
-      ESP_LOGW(TAG, "Corrected MPS of EP 0x%02X from %u to 64", static_cast<uint8_t>(ep->bEndpointAddress & 0xFF),
-               ep->wMaxPacketSize);
-      ep_mutable->wMaxPacketSize = 64;
+    if (ep->wMaxPacketSize > usb_host::USB_MAX_PACKET_SIZE) {
+      ESP_LOGW(TAG, "Corrected MPS of EP 0x%02X from %u to %u", static_cast<uint8_t>(ep->bEndpointAddress & 0xFF),
+               ep->wMaxPacketSize, usb_host::USB_MAX_PACKET_SIZE);
+      ep_mutable->wMaxPacketSize = usb_host::USB_MAX_PACKET_SIZE;
     }
   }
 }

esphome/components/usb_uart/usb_uart.h

@@ -106,20 +106,19 @@ class RingBuffer {
 
 // Structure for queuing received USB data chunks
 struct UsbDataChunk {
-  static constexpr size_t MAX_CHUNK_SIZE = 64;  // USB packet size
-  uint8_t data[MAX_CHUNK_SIZE];
-  uint8_t length;  // Max 64 bytes, so uint8_t is sufficient
+  uint8_t data[usb_host::USB_MAX_PACKET_SIZE];
+  uint16_t length;
   USBUartChannel *channel;
 
   // Required for EventPool - no cleanup needed for POD types
   void release() {}
 };
 
-// Structure for queuing outgoing USB data chunks (one per USB FS packet)
+// Structure for queuing outgoing USB data chunks (one per USB packet)
 struct UsbOutputChunk {
-  static constexpr size_t MAX_CHUNK_SIZE = 64;  // USB FS MPS
+  static constexpr size_t MAX_CHUNK_SIZE = usb_host::USB_MAX_PACKET_SIZE;
   uint8_t data[MAX_CHUNK_SIZE];
-  uint8_t length;
+  uint16_t length;
 
   // Required for EventPool - no cleanup needed for POD types
   void release() {}

esphome/core/ring_buffer.cpp

@@ -1,11 +1,9 @@
 #include "ring_buffer.h"
 
-#include "esphome/core/helpers.h"
-#include "esphome/core/log.h"
-
 #ifdef USE_ESP32
 
-#include "helpers.h"
+#include "esphome/core/helpers.h"
+#include "esphome/core/log.h"
 
 namespace esphome {
 
@@ -19,12 +17,15 @@ RingBuffer::~RingBuffer() {
   }
 }
 
-std::unique_ptr<RingBuffer> RingBuffer::create(size_t len) {
+std::unique_ptr<RingBuffer> RingBuffer::create(size_t len, MemoryPreference preference) {
   std::unique_ptr<RingBuffer> rb = make_unique<RingBuffer>();
 
   rb->size_ = len;
 
-  RAMAllocator<uint8_t> allocator;
+  const uint8_t type = (preference == MemoryPreference::INTERNAL_FIRST) ? RAMAllocator<uint8_t>::PREFER_INTERNAL
+                                                                        : RAMAllocator<uint8_t>::NONE;
+
+  RAMAllocator<uint8_t> allocator(type);
   rb->storage_ = allocator.allocate(rb->size_);
   if (rb->storage_ == nullptr) {
     return nullptr;

esphome/core/ring_buffer.h

@@ -80,7 +80,12 @@ class RingBuffer {
    */
   BaseType_t reset();
 
-  static std::unique_ptr<RingBuffer> create(size_t len);
+  enum class MemoryPreference {
+    EXTERNAL_FIRST,  // External RAM preferred, fall back to internal (default)
+    INTERNAL_FIRST,  // Internal RAM preferred, fall back to external
+  };
+
+  static std::unique_ptr<RingBuffer> create(size_t len, MemoryPreference preference = MemoryPreference::EXTERNAL_FIRST);
 
  protected:
   /// @brief Discards data from the ring buffer.

tests/unit_tests/components/api/test_client.py

@@ -0,0 +1,98 @@
+"""Tests for esphome.components.api.client."""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from esphome.components.api import client as api_client
+from esphome.core import EsphomeError
+
+
+def test_decoder_swallows_esphome_error() -> None:
+    """A failing stack-trace decode must not propagate.
+
+    on_log runs inside an asyncio protocol callback; if EsphomeError
+    escapes, the loop reports "Fatal error: protocol.data_received()
+    call failed.", tears the connection down, and ReconnectLogic loops
+    forever as the device replays the same crash trace on every
+    reconnect.
+    """
+    config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
+
+    with patch.object(
+        api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
+    ) as mock_process:
+        processor.process_line("PC: 0x4010496e")
+
+    assert mock_process.called
+    assert processor.backtrace_state is False
+
+
+def test_decoder_swallows_platform_handler_error() -> None:
+    """The same protection must apply to the platform-specific handler."""
+    config = {"esphome": {"name": "test"}}
+
+    def platform_handler(_config, _line, _state):
+        raise EsphomeError("no idedata")
+
+    processor = api_client._LogLineProcessor(config, platform_handler)
+    processor.process_line("PC: 0x4010496e")
+
+    assert processor.backtrace_state is False
+
+
+def test_decoder_short_circuits_after_failure() -> None:
+    """After one failure, subsequent lines must not retry the decoder.
+
+    _decode_pc shells out to PlatformIO; a crash dump can contain many
+    PC/BT lines and retrying the failing subprocess for each one would
+    stall log streaming.
+    """
+    config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
+
+    with patch.object(
+        api_client, "process_stacktrace", side_effect=EsphomeError("no idedata")
+    ) as mock_process:
+        processor.process_line("PC: 0x4010496e")
+        processor.process_line("BT0: 0x4010496e")
+        processor.process_line("BT1: 0x401049aa")
+
+    assert mock_process.call_count == 1
+
+
+def test_decoder_threads_backtrace_state() -> None:
+    """When decoding succeeds, backtrace_state is threaded across calls."""
+    config = {"esphome": {"name": "test"}}
+    processor = api_client._LogLineProcessor(config, None)
+
+    with patch.object(
+        api_client, "process_stacktrace", side_effect=[True, False]
+    ) as mock_process:
+        processor.process_line(">>>stack>>>")
+        assert processor.backtrace_state is True
+        processor.process_line("<<<stack<<<")
+        assert processor.backtrace_state is False
+
+    assert mock_process.call_args_list[0].kwargs == {"backtrace_state": False}
+    assert mock_process.call_args_list[1].kwargs == {"backtrace_state": True}
+
+
+def test_decoder_uses_platform_handler_when_provided() -> None:
+    """The platform handler is preferred over the generic one."""
+    config = {"esphome": {"name": "test"}}
+    calls: list[tuple[object, str, bool]] = []
+
+    def platform_handler(cfg, line, state):
+        calls.append((cfg, line, state))
+        return True
+
+    processor = api_client._LogLineProcessor(config, platform_handler)
+
+    with patch.object(api_client, "process_stacktrace") as mock_generic:
+        processor.process_line("BT0: 0x4010496e")
+
+    assert calls == [(config, "BT0: 0x4010496e", False)]
+    assert mock_generic.called is False
+    assert processor.backtrace_state is True