From a4682615234ac61ff7825e2a922807b11e6772fa Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 21 Feb 2026 19:41:26 -0600 Subject: [PATCH 1/2] [scheduler] De-template and consolidate scheduler helper functions (#14164) --- esphome/core/scheduler.cpp | 14 +++++++++---- esphome/core/scheduler.h | 41 +++++++------------------------------- 2 files changed, 17 insertions(+), 38 deletions(-) diff --git a/esphome/core/scheduler.cpp b/esphome/core/scheduler.cpp index 36b65f6ff7c..e4e0751e100 100644 --- a/esphome/core/scheduler.cpp +++ b/esphome/core/scheduler.cpp @@ -119,10 +119,16 @@ uint32_t Scheduler::calculate_interval_offset_(uint32_t delay) { // Remove before 2026.8.0 along with all retry code bool Scheduler::is_retry_cancelled_locked_(Component *component, NameType name_type, const char *static_name, uint32_t hash_or_id) { - return has_cancelled_timeout_in_container_locked_(this->items_, component, name_type, static_name, hash_or_id, - /* match_retry= */ true) || - has_cancelled_timeout_in_container_locked_(this->to_add_, component, name_type, static_name, hash_or_id, - /* match_retry= */ true); + for (auto *container : {&this->items_, &this->to_add_}) { + for (auto &item : *container) { + if (item && this->is_item_removed_locked_(item.get()) && + this->matches_item_locked_(item, component, name_type, static_name, hash_or_id, SchedulerItem::TIMEOUT, + /* match_retry= */ true, /* skip_removed= */ false)) { + return true; + } + } + } + return false; } // Common implementation for both timeout and interval diff --git a/esphome/core/scheduler.h b/esphome/core/scheduler.h index 384d76b6b08..16b0ded3126 100644 --- a/esphome/core/scheduler.h +++ b/esphome/core/scheduler.h @@ -308,8 +308,8 @@ class Scheduler { SchedulerItem::Type type, bool match_retry, bool skip_removed = true) const { // THREAD SAFETY: Check for nullptr first to prevent LoadProhibited crashes. On multi-threaded // platforms, items can be moved out of defer_queue_ during processing, leaving nullptr entries. - // PR #11305 added nullptr checks in callers (mark_matching_items_removed_locked_() and - // has_cancelled_timeout_in_container_locked_()), but this check provides defense-in-depth: helper + // PR #11305 added nullptr checks in callers (mark_matching_items_removed_locked_()), but this check + // provides defense-in-depth: helper // functions should be safe regardless of caller behavior. // Fixes: https://github.com/esphome/esphome/issues/11940 if (!item) @@ -403,8 +403,7 @@ class Scheduler { // SAFETY: Moving out the unique_ptr leaves a nullptr in the vector at defer_queue_front_. // This is intentional and safe because: // 1. The vector is only cleaned up by cleanup_defer_queue_locked_() at the end of this function - // 2. Any code iterating defer_queue_ MUST check for nullptr items (see mark_matching_items_removed_locked_ - // and has_cancelled_timeout_in_container_locked_ in scheduler.h) + // 2. Any code iterating defer_queue_ MUST check for nullptr items (see mark_matching_items_removed_locked_) // 3. The lock protects concurrent access, but the nullptr remains until cleanup item = std::move(this->defer_queue_[this->defer_queue_front_]); this->defer_queue_front_++; @@ -497,19 +496,16 @@ class Scheduler { // name_type determines matching: STATIC_STRING uses static_name, others use hash_or_id // Returns the number of items marked for removal // IMPORTANT: Must be called with scheduler lock held - template - size_t mark_matching_items_removed_locked_(Container &container, Component *component, NameType name_type, - const char *static_name, uint32_t hash_or_id, SchedulerItem::Type type, - bool match_retry) { + size_t mark_matching_items_removed_locked_(std::vector> &container, + Component *component, NameType name_type, const char *static_name, + uint32_t hash_or_id, SchedulerItem::Type type, bool match_retry) { size_t count = 0; for (auto &item : container) { // Skip nullptr items (can happen in defer_queue_ when items are being processed) // The defer_queue_ uses index-based processing: items are std::moved out but left in the // vector as nullptr until cleanup. Even though this function is called with lock held, // the vector can still contain nullptr items from the processing loop. This check prevents crashes. - if (!item) - continue; - if (this->matches_item_locked_(item, component, name_type, static_name, hash_or_id, type, match_retry)) { + if (item && this->matches_item_locked_(item, component, name_type, static_name, hash_or_id, type, match_retry)) { this->set_item_removed_(item.get(), true); count++; } @@ -517,29 +513,6 @@ class Scheduler { return count; } - // Template helper to check if any item in a container matches our criteria - // name_type determines matching: STATIC_STRING uses static_name, others use hash_or_id - // IMPORTANT: Must be called with scheduler lock held - template - bool has_cancelled_timeout_in_container_locked_(const Container &container, Component *component, NameType name_type, - const char *static_name, uint32_t hash_or_id, - bool match_retry) const { - for (const auto &item : container) { - // Skip nullptr items (can happen in defer_queue_ when items are being processed) - // The defer_queue_ uses index-based processing: items are std::moved out but left in the - // vector as nullptr until cleanup. If this function is called during defer queue processing, - // it will iterate over these nullptr items. This check prevents crashes. - if (!item) - continue; - if (this->is_item_removed_locked_(item.get()) && - this->matches_item_locked_(item, component, name_type, static_name, hash_or_id, SchedulerItem::TIMEOUT, - match_retry, /* skip_removed= */ false)) { - return true; - } - } - return false; - } - Mutex lock_; std::vector> items_; std::vector> to_add_; From d5c9c56fdfcdd0112e1913f05f84e297908460a7 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Sat, 21 Feb 2026 19:41:43 -0600 Subject: [PATCH 2/2] [platformio] Add exponential backoff and session reset to download retries (#14191) --- esphome/platformio_api.py | 41 ++++- tests/unit_tests/test_platformio_api.py | 196 +++++++++++++++++++++++- 2 files changed, 231 insertions(+), 6 deletions(-) diff --git a/esphome/platformio_api.py b/esphome/platformio_api.py index d42f89d0290..5d4065207f0 100644 --- a/esphome/platformio_api.py +++ b/esphome/platformio_api.py @@ -5,6 +5,7 @@ import os from pathlib import Path import re import subprocess +import time from typing import Any from esphome.const import CONF_COMPILE_PROCESS_LIMIT, CONF_ESPHOME, KEY_CORE @@ -44,31 +45,61 @@ def patch_structhash(): def patch_file_downloader(): - """Patch PlatformIO's FileDownloader to retry on PackageException errors.""" + """Patch PlatformIO's FileDownloader to retry on PackageException errors. + + PlatformIO's FileDownloader uses HTTPSession which lacks built-in retry + for 502/503 errors. We add retries with exponential backoff and close the + session between attempts to force a fresh TCP connection, which may route + to a different CDN edge node. + """ from platformio.package.download import FileDownloader from platformio.package.exception import PackageException + if getattr(FileDownloader.__init__, "_esphome_patched", False): + return + original_init = FileDownloader.__init__ def patched_init(self, *args: Any, **kwargs: Any) -> None: - max_retries = 3 + max_retries = 5 for attempt in range(max_retries): try: - return original_init(self, *args, **kwargs) + original_init(self, *args, **kwargs) + return except PackageException as e: if attempt < max_retries - 1: + # Exponential backoff: 2, 4, 8, 16 seconds + delay = 2 ** (attempt + 1) _LOGGER.warning( - "Package download failed: %s. Retrying... (attempt %d/%d)", + "Package download failed: %s. " + "Retrying in %d seconds... (attempt %d/%d)", str(e), + delay, attempt + 1, max_retries, ) + # Close the response and session to free resources + # and force a new TCP connection on retry, which may + # route to a different CDN edge node + # pylint: disable=protected-access,broad-except + try: + if ( + hasattr(self, "_http_response") + and self._http_response is not None + ): + self._http_response.close() + if hasattr(self, "_http_session"): + self._http_session.close() + except Exception: + pass + # pylint: enable=protected-access,broad-except + time.sleep(delay) else: # Final attempt - re-raise raise - return None + patched_init._esphome_patched = True # type: ignore[attr-defined] # pylint: disable=protected-access FileDownloader.__init__ = patched_init diff --git a/tests/unit_tests/test_platformio_api.py b/tests/unit_tests/test_platformio_api.py index 4d7b635e595..16861442777 100644 --- a/tests/unit_tests/test_platformio_api.py +++ b/tests/unit_tests/test_platformio_api.py @@ -6,7 +6,7 @@ import os from pathlib import Path import shutil from types import SimpleNamespace -from unittest.mock import MagicMock, Mock, patch +from unittest.mock import MagicMock, Mock, call, patch import pytest @@ -673,6 +673,200 @@ def test_process_stacktrace_bad_alloc( assert state is False +def test_patch_file_downloader_succeeds_first_try() -> None: + """Test patch_file_downloader succeeds on first attempt.""" + mock_exception_cls = type("PackageException", (Exception,), {}) + original_init = MagicMock() + + with patch.dict( + "sys.modules", + { + "platformio": MagicMock(), + "platformio.package": MagicMock(), + "platformio.package.download": SimpleNamespace( + FileDownloader=type("FileDownloader", (), {"__init__": original_init}) + ), + "platformio.package.exception": SimpleNamespace( + PackageException=mock_exception_cls + ), + }, + ): + platformio_api.patch_file_downloader() + + from platformio.package.download import FileDownloader + + instance = object.__new__(FileDownloader) + FileDownloader.__init__(instance, "http://example.com/file.zip") + + original_init.assert_called_once() + + +def test_patch_file_downloader_retries_on_failure() -> None: + """Test patch_file_downloader retries with backoff on PackageException.""" + mock_exception_cls = type("PackageException", (Exception,), {}) + call_count = 0 + + def failing_init(self, *args, **kwargs): + nonlocal call_count + call_count += 1 + if call_count < 3: + raise mock_exception_cls(f"502 error attempt {call_count}") + + with ( + patch.dict( + "sys.modules", + { + "platformio": MagicMock(), + "platformio.package": MagicMock(), + "platformio.package.download": SimpleNamespace( + FileDownloader=type( + "FileDownloader", (), {"__init__": failing_init} + ) + ), + "platformio.package.exception": SimpleNamespace( + PackageException=mock_exception_cls + ), + }, + ), + patch("time.sleep") as mock_sleep, + ): + platformio_api.patch_file_downloader() + + from platformio.package.download import FileDownloader + + instance = object.__new__(FileDownloader) + FileDownloader.__init__(instance, "http://example.com/file.zip") + + # Should have been called 3 times (2 failures + 1 success) + assert call_count == 3 + + # Should have slept with exponential backoff: 2s, 4s + assert mock_sleep.call_count == 2 + mock_sleep.assert_any_call(2) + mock_sleep.assert_any_call(4) + + +def test_patch_file_downloader_raises_after_max_retries() -> None: + """Test patch_file_downloader raises after exhausting all retries.""" + mock_exception_cls = type("PackageException", (Exception,), {}) + + def always_failing_init(self, *args, **kwargs): + raise mock_exception_cls("502 error") + + with ( + patch.dict( + "sys.modules", + { + "platformio": MagicMock(), + "platformio.package": MagicMock(), + "platformio.package.download": SimpleNamespace( + FileDownloader=type( + "FileDownloader", (), {"__init__": always_failing_init} + ) + ), + "platformio.package.exception": SimpleNamespace( + PackageException=mock_exception_cls + ), + }, + ), + patch("time.sleep") as mock_sleep, + ): + platformio_api.patch_file_downloader() + + from platformio.package.download import FileDownloader + + instance = object.__new__(FileDownloader) + with pytest.raises(mock_exception_cls, match="502 error"): + FileDownloader.__init__(instance, "http://example.com/file.zip") + + # Should have slept 4 times (before attempts 2-5), not on final attempt + assert mock_sleep.call_count == 4 + mock_sleep.assert_has_calls([call(2), call(4), call(8), call(16)]) + + +def test_patch_file_downloader_closes_session_and_response_between_retries() -> None: + """Test patch_file_downloader closes HTTP session and response between retries.""" + mock_exception_cls = type("PackageException", (Exception,), {}) + mock_session = MagicMock() + mock_response = MagicMock() + call_count = 0 + + def failing_init_with_session(self, *args, **kwargs): + nonlocal call_count + call_count += 1 + self._http_session = mock_session + self._http_response = mock_response + if call_count < 2: + raise mock_exception_cls("502 error") + + with ( + patch.dict( + "sys.modules", + { + "platformio": MagicMock(), + "platformio.package": MagicMock(), + "platformio.package.download": SimpleNamespace( + FileDownloader=type( + "FileDownloader", + (), + {"__init__": failing_init_with_session}, + ) + ), + "platformio.package.exception": SimpleNamespace( + PackageException=mock_exception_cls + ), + }, + ), + patch("time.sleep"), + ): + platformio_api.patch_file_downloader() + + from platformio.package.download import FileDownloader + + instance = object.__new__(FileDownloader) + FileDownloader.__init__(instance, "http://example.com/file.zip") + + # Both response and session should have been closed between retries + mock_response.close.assert_called_once() + mock_session.close.assert_called_once() + + +def test_patch_file_downloader_idempotent() -> None: + """Test patch_file_downloader does not stack wrappers when called multiple times.""" + mock_exception_cls = type("PackageException", (Exception,), {}) + call_count = 0 + + def counting_init(self, *args, **kwargs): + nonlocal call_count + call_count += 1 + + with patch.dict( + "sys.modules", + { + "platformio": MagicMock(), + "platformio.package": MagicMock(), + "platformio.package.download": SimpleNamespace( + FileDownloader=type("FileDownloader", (), {"__init__": counting_init}) + ), + "platformio.package.exception": SimpleNamespace( + PackageException=mock_exception_cls + ), + }, + ): + # Patch multiple times + platformio_api.patch_file_downloader() + platformio_api.patch_file_downloader() + platformio_api.patch_file_downloader() + + from platformio.package.download import FileDownloader + + instance = object.__new__(FileDownloader) + FileDownloader.__init__(instance, "http://example.com/file.zip") + + # Should only be called once, not 3 times from stacked wrappers + assert call_count == 1 + + def test_platformio_log_filter_allows_non_platformio_messages() -> None: """Test that non-platformio logger messages are allowed through.""" log_filter = platformio_api.PlatformioLogFilter()