From f4d3fb1a1810b707558db7990fbc4d25b231690b Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Fri, 1 May 2026 12:01:19 -0500 Subject: [PATCH] [api] Short-circuit log-line decoder after first failure _decode_pc shells out to PlatformIO via _run_idedata; without a populated build dir for the device that subprocess fails for every PC/BT line in a crash dump. Disable decoding after the first EsphomeError (per logs session) and emit a single user-facing warning instead of retrying on every line. Also rename the helper to _LogLineProcessor since it now owns the per-session decode-enabled state, not just the decode call. --- esphome/components/api/client.py | 69 +++++++++++-------- .../unit_tests/components/api/test_client.py | 68 +++++++++++------- 2 files changed, 85 insertions(+), 52 deletions(-) diff --git a/esphome/components/api/client.py b/esphome/components/api/client.py index 31ae8163da7..9b8b42ab315 100644 --- a/esphome/components/api/client.py +++ b/esphome/components/api/client.py @@ -32,30 +32,47 @@ if TYPE_CHECKING: _LOGGER = logging.getLogger(__name__) -def _process_stacktrace_line( - config: dict[str, Any], - raw_line: str, - backtrace_state: bool, - platform_process_stacktrace: Any | None, -) -> bool: - """Run the stack-trace decoder for a single log line. +class _LogLineProcessor: + """Feeds incoming log lines to the stack-trace decoder. - 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. Stack-trace decoding requires a populated - build dir for the device, which may not exist (e.g. flashed from - another machine); log and continue instead of killing the - connection. + 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. """ - try: - if platform_process_stacktrace: - return platform_process_stacktrace(config, raw_line, backtrace_state) - return process_stacktrace(config, raw_line, backtrace_state=backtrace_state) - except EsphomeError as exc: - _LOGGER.debug("Stack-trace decoding failed: %s", exc) - return False + + 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( @@ -87,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 @@ -97,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") @@ -110,9 +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(): - backtrace_state = _process_stacktrace_line( - config, raw_line, backtrace_state, platform_process_stacktrace - ) + 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/tests/unit_tests/components/api/test_client.py b/tests/unit_tests/components/api/test_client.py index 9f038b093d5..1e16c5ebbc1 100644 --- a/tests/unit_tests/components/api/test_client.py +++ b/tests/unit_tests/components/api/test_client.py @@ -8,7 +8,7 @@ from esphome.components.api import client as api_client from esphome.core import EsphomeError -def test_process_stacktrace_line_swallows_esphome_error() -> None: +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 @@ -18,50 +18,68 @@ def test_process_stacktrace_line_swallows_esphome_error() -> None: 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: - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", True, None - ) + processor.process_line("PC: 0x4010496e") assert mock_process.called - assert result is False + assert processor.backtrace_state is False -def test_process_stacktrace_line_swallows_platform_handler_error() -> None: +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") - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", True, platform_handler - ) + processor = api_client._LogLineProcessor(config, platform_handler) + processor.process_line("PC: 0x4010496e") - assert result is False + assert processor.backtrace_state is False -def test_process_stacktrace_line_returns_handler_result() -> None: - """When decoding succeeds, the handler's result is returned unchanged.""" +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", return_value=True + api_client, "process_stacktrace", side_effect=EsphomeError("no idedata") ) as mock_process: - result = api_client._process_stacktrace_line( - config, "PC: 0x4010496e", False, None - ) + processor.process_line("PC: 0x4010496e") + processor.process_line("BT0: 0x4010496e") + processor.process_line("BT1: 0x401049aa") - mock_process.assert_called_once_with( - config, "PC: 0x4010496e", backtrace_state=False - ) - assert result is True + assert mock_process.call_count == 1 -def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None: +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]] = [] @@ -70,11 +88,11 @@ def test_process_stacktrace_line_uses_platform_handler_when_provided() -> None: calls.append((cfg, line, state)) return True + processor = api_client._LogLineProcessor(config, platform_handler) + with patch.object(api_client, "process_stacktrace") as mock_generic: - result = api_client._process_stacktrace_line( - config, "BT0: 0x4010496e", False, platform_handler - ) + processor.process_line("BT0: 0x4010496e") assert calls == [(config, "BT0: 0x4010496e", False)] assert mock_generic.called is False - assert result is True + assert processor.backtrace_state is True