diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 312d937f01..9b8b42ab31 100644 --- a/esphome/components/api/client.py +++ b/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 diff --git a/esphome/components/usb_host/__init__.py b/esphome/components/usb_host/__init__.py index 338bd8d572..8e591bd80c 100644 --- a/esphome/components/usb_host/__init__.py +++ b/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) diff --git a/esphome/components/usb_host/usb_host.h b/esphome/components/usb_host/usb_host.h index dcb76a3a3b..480fd86750 100644 --- a/esphome/components/usb_host/usb_host.h +++ b/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) diff --git a/esphome/components/usb_host/usb_host_client.cpp b/esphome/components/usb_host/usb_host_client.cpp index c34c7ef67d..4ee8e2ac5e 100644 --- a/esphome/components/usb_host/usb_host_client.cpp +++ b/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 } diff --git a/esphome/components/usb_uart/__init__.py b/esphome/components/usb_uart/__init__.py index d542788fb9..1cf78fdbd5 100644 --- a/esphome/components/usb_uart/__init__.py +++ b/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 is a ring buffer that wastes one entry. + # of every device. Add one extra slot because LockFreeQueue 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: diff --git a/esphome/components/usb_uart/usb_uart.cpp b/esphome/components/usb_uart/usb_uart.cpp index 30ec61fdc4..e3bf5e40bc 100644 --- a/esphome/components/usb_uart/usb_uart.cpp +++ b/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(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(ep); - if (ep->wMaxPacketSize > 64) { - ESP_LOGW(TAG, "Corrected MPS of EP 0x%02X from %u to 64", static_cast(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(ep->bEndpointAddress & 0xFF), + ep->wMaxPacketSize, usb_host::USB_MAX_PACKET_SIZE); + ep_mutable->wMaxPacketSize = usb_host::USB_MAX_PACKET_SIZE; } } } diff --git a/esphome/components/usb_uart/usb_uart.h b/esphome/components/usb_uart/usb_uart.h index f9648b795b..e88c41c0cb 100644 --- a/esphome/components/usb_uart/usb_uart.h +++ b/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() {} diff --git a/esphome/core/ring_buffer.cpp b/esphome/core/ring_buffer.cpp index 6a2232599f..2e0802eceb 100644 --- a/esphome/core/ring_buffer.cpp +++ b/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::create(size_t len) { +std::unique_ptr RingBuffer::create(size_t len, MemoryPreference preference) { std::unique_ptr rb = make_unique(); rb->size_ = len; - RAMAllocator allocator; + const uint8_t type = (preference == MemoryPreference::INTERNAL_FIRST) ? RAMAllocator::PREFER_INTERNAL + : RAMAllocator::NONE; + + RAMAllocator allocator(type); rb->storage_ = allocator.allocate(rb->size_); if (rb->storage_ == nullptr) { return nullptr; diff --git a/esphome/core/ring_buffer.h b/esphome/core/ring_buffer.h index 98a273781f..4acd07d5b0 100644 --- a/esphome/core/ring_buffer.h +++ b/esphome/core/ring_buffer.h @@ -80,7 +80,12 @@ class RingBuffer { */ BaseType_t reset(); - static std::unique_ptr 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 create(size_t len, MemoryPreference preference = MemoryPreference::EXTERNAL_FIRST); protected: /// @brief Discards data from the ring buffer. diff --git a/tests/unit_tests/components/api/__init__.py b/tests/unit_tests/components/api/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/tests/unit_tests/components/api/test_client.py b/tests/unit_tests/components/api/test_client.py new file mode 100644 index 0000000000..1e16c5ebbc --- /dev/null +++ b/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("<< 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