diff --git a/esphome/espidf/framework.py b/esphome/espidf/framework.py index f6cb15762e..e71a333f68 100644 --- a/esphome/espidf/framework.py +++ b/esphome/espidf/framework.py @@ -34,7 +34,7 @@ from esphome.framework_helpers import ( run_command_ok, str_to_lst_of_str, tool_version_runs, - warn_prefetch_failures, + warn_batch_failures, ) from esphome.helpers import write_file_if_changed @@ -774,7 +774,7 @@ def _prefetch_idf_tool_archives( for entry in entries ], ) - warn_prefetch_failures(failures) + warn_batch_failures(failures) if len(failures) == len(entries): # A systematic fault, not one flaky mirror: the resume # workaround (#17703) is off for this whole install diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py index 037cdf7a8c..af7ab8a8c0 100644 --- a/esphome/framework_helpers.py +++ b/esphome/framework_helpers.py @@ -969,22 +969,26 @@ def resume_fetch_job( return fetch -def warn_prefetch_failures( - failures: list[tuple[str, BaseException]], - message: str = "Could not prefetch %s: %s", - detail: str = "Prefetch failure detail", -) -> None: - """Warn per failed batch job, keeping the traceback of unexpected errors.""" +def is_expected_fetch_error(err: BaseException) -> bool: + """Download failures the callers degrade on, vs programming errors.""" from esphome.core import EsphomeError # local import avoids circular dependency + return isinstance(err, (EsphomeError, OSError)) + + +def warn_batch_failures( + failures: list[tuple[str, BaseException]], + message: str = "Could not prefetch %s: %s", +) -> None: + """Warn per failed batch job, keeping the traceback of unexpected errors.""" for name, err in failures: - # A programming error must not be reduced to a bare message - expected = isinstance(err, (EsphomeError, OSError)) # failure_reason: a message-less exception must not log blank - _LOGGER.warning( - message, name, failure_reason(err), exc_info=None if expected else err - ) - _LOGGER.debug(detail, exc_info=err) + if is_expected_fetch_error(err): + _LOGGER.warning(message, name, failure_reason(err)) + _LOGGER.debug("Failure detail", exc_info=err) + else: + # A programming error must not be reduced to a bare message + _LOGGER.warning(message, name, failure_reason(err), exc_info=err) def download_with_resume( diff --git a/esphome/platformio/library.py b/esphome/platformio/library.py index 0402311a9a..453d416106 100644 --- a/esphome/platformio/library.py +++ b/esphome/platformio/library.py @@ -35,7 +35,7 @@ from esphome.framework_helpers import ( failure_reason, rmdir, run_batch_downloads, - warn_prefetch_failures, + warn_batch_failures, ) _LOGGER = logging.getLogger(__name__) @@ -1040,7 +1040,7 @@ def _prefetch_wave( ], ) # The sequential call below retries and raises the real error - warn_prefetch_failures( + warn_batch_failures( failures, "Prefetch of %s failed (retrying sequentially): %s" ) except Exception as err: # noqa: BLE001 # pylint: disable=broad-exception-caught diff --git a/esphome/platformio/prefetch.py b/esphome/platformio/prefetch.py index ef8c27c9aa..e2b767e2cc 100644 --- a/esphome/platformio/prefetch.py +++ b/esphome/platformio/prefetch.py @@ -32,12 +32,13 @@ import time from typing import Any, NamedTuple from esphome.framework_helpers import ( + BATCH_EXTRACT_WORKERS, content_length, discard_partial_download, failure_reason, resume_fetch_job, run_batch_downloads, - warn_prefetch_failures, + warn_batch_failures, ) from esphome.helpers import get_bool_env, get_usable_cpu_count, rmtree @@ -681,7 +682,7 @@ def _preinstall( would hang, not fail). Waves skip dependencies; the installed manifests feed the next wave. Any failure falls back to pio run. """ - workers = min(get_usable_cpu_count(), len(entries)) + workers = min(get_usable_cpu_count(), len(entries), BATCH_EXTRACT_WORKERS) # One manager per worker (_install mutates instance state); built # serially because construction rewires the shared manager logger managers: SimpleQueue = SimpleQueue() @@ -876,7 +877,7 @@ def _prefetch(build_dir: Path, env: str) -> None: ) # PlatformIO retries failed packages itself, without resume failures = run_batch_downloads("Downloading PlatformIO packages", jobs) - warn_prefetch_failures(failures) + warn_batch_failures(failures) failed_names = {name for name, _ in failures} elif not groups and not unresolved: # Record the no-work run so the parent skips the next spawn. diff --git a/esphome/platformio/registry.py b/esphome/platformio/registry.py index 11f7c49c44..e6092e8b9c 100644 --- a/esphome/platformio/registry.py +++ b/esphome/platformio/registry.py @@ -18,9 +18,10 @@ from esphome.framework_helpers import ( archive_extract_all, download_from_mirrors, download_with_resume, + is_expected_fetch_error, rmdir, run_batch_downloads, - warn_prefetch_failures, + warn_batch_failures, ) from esphome.helpers import get_usable_cpu_count from esphome.net_retry import fetch_with_retry, http_request @@ -185,6 +186,26 @@ def _already_installed(dest: Path) -> bool: return (dest / ".esphome_extracted").is_file() +def _batched_download_progress( + name: str, version: str, size: int | None, extract_progress: Callable[[float], None] +) -> Callable[[int], None]: + """Download tracker for a batched install: no private bar, no bytes (the + shared bar must never run backwards); the zero tick keeps cancellation + observable and a real refetch is announced once.""" + announced = False + + def progress(done: int) -> None: + nonlocal announced + # size-less registry entries still announce: streaming starts at + # done=0, while a verify no-op credits the full file in one tick + if not announced and done < (size or 1): + _LOGGER.info("Re-downloading %s %s ...", name, version) + announced = True + extract_progress(0.0) + + return progress + + def prefetch_packages( packages: list[tuple[str, str, Path, list[str]]], downloads_dir: Path ) -> None: @@ -253,7 +274,7 @@ def prefetch_packages( [(entry.name, entry.size, partial(_fetch, entry)) for entry in pending], ) for name, err in failures: - if isinstance(err, (EsphomeError, OSError)): + if is_expected_fetch_error(err): # Expected download failures: install_package retries this one # itself, with a visible bar _LOGGER.debug("Prefetch of %s failed: %s", name, err) @@ -323,15 +344,14 @@ def install_package( ) else: url, sha256, size = registry_download(name, version) - # Batched: no private bar, no bytes (the shared bar must never - # run backwards), but the zero tick keeps cancellation observable - download_progress = ( - None - if extract_progress is None - else lambda _done: extract_progress(0.0) - ) download_with_resume( - url, archive, sha256=sha256, size=size, progress=download_progress + url, + archive, + sha256=sha256, + size=size, + progress=None + if extract_progress is None + else _batched_download_progress(name, version, size, extract_progress), ) log("Extracting %s ...", name) archive_extract_all( @@ -353,26 +373,24 @@ def install_packages(specs: Collection[PackageSpec], downloads_dir: Path) -> Non seen: set[str] = set() for spec in specs: name, version, dest, mirrors, _expect = spec + archive = _archive_path(downloads_dir, name, version) # Duplicate entries share one archive and would race each other # between two workers; mirror prefetch_packages' dedupe - if _already_installed(dest) or mirrors or f"{name}-{version}" in seen: + if _already_installed(dest) or mirrors or archive.name in seen: rest.append(spec) continue try: # An archive at its final name already passed sha256/size # verification - size = _archive_path(downloads_dir, name, version).stat().st_size + size = archive.stat().st_size except FileNotFoundError: rest.append(spec) continue - seen.add(f"{name}-{version}") + seen.add(archive.name) pending.append((spec, size)) if len(pending) < 2: - rest = list(specs) - pending = [] - for name, version, dest, mirrors, expect in rest: - install_package(name, version, dest, mirrors, downloads_dir, expect=expect) - if not pending: + for name, version, dest, mirrors, expect in specs: + install_package(name, version, dest, mirrors, downloads_dir, expect=expect) return workers = min(get_usable_cpu_count(), len(pending), BATCH_EXTRACT_WORKERS) _LOGGER.info( @@ -400,7 +418,11 @@ def install_packages(specs: Collection[PackageSpec], downloads_dir: Path) -> Non max_workers=workers, ) if failures: - warn_prefetch_failures( - failures[1:], "Could not install %s: %s", detail="Install failure detail" - ) + # Warn on the first failure too: the raised exception's message may + # not name which package failed + warn_batch_failures(failures, "Could not install %s: %s") raise failures[0][1] + # Sequential remainder after the batch, so a duplicate spec cannot + # unlink the archive its batched twin was sized from + for name, version, dest, mirrors, expect in rest: + install_package(name, version, dest, mirrors, downloads_dir, expect=expect) diff --git a/tests/unit_tests/test_framework_helpers.py b/tests/unit_tests/test_framework_helpers.py index 012bd2f608..dfa9c66992 100644 --- a/tests/unit_tests/test_framework_helpers.py +++ b/tests/unit_tests/test_framework_helpers.py @@ -2345,35 +2345,34 @@ def test_resume_fetch_job_threads_tracker(tmp_path: Path) -> None: ) -def test_warn_prefetch_failures_names_each_failure( +def test_warn_batch_failures_names_each_failure( caplog: pytest.LogCaptureFixture, ) -> None: """The shared failure loop warns per job with the failure reason.""" - from esphome.framework_helpers import warn_prefetch_failures + from esphome.framework_helpers import warn_batch_failures - warn_prefetch_failures([("toolchain-x@1", OSError("down"))]) + warn_batch_failures([("toolchain-x@1", OSError("down"))]) assert "Could not prefetch toolchain-x@1: down" in caplog.text - warn_prefetch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s") + warn_batch_failures([("lib", OSError("gone"))], "Prefetch of %s failed: %s") assert "Prefetch of lib failed: gone" in caplog.text -def test_warn_prefetch_failures_unexpected_error_keeps_traceback( +def test_warn_batch_failures_unexpected_error_keeps_traceback( caplog: pytest.LogCaptureFixture, ) -> None: """An unexpected error type is not reduced to a bare message; expected download failures stay message-only at WARNING.""" - from esphome.framework_helpers import warn_prefetch_failures + from esphome.framework_helpers import warn_batch_failures with caplog.at_level(logging.DEBUG): - warn_prefetch_failures( + warn_batch_failures( [("pkg", TypeError("bad call")), ("lib", OSError("down"))], "Could not install %s: %s", - detail="Install failure detail", ) warnings = {r.getMessage(): r for r in caplog.records if r.levelname == "WARNING"} assert warnings["Could not install pkg: bad call"].exc_info is not None assert warnings["Could not install lib: down"].exc_info is None - assert "Install failure detail" in caplog.text + assert "Failure detail" in caplog.text @pytest.mark.parametrize( diff --git a/tests/unit_tests/test_platformio_prefetch.py b/tests/unit_tests/test_platformio_prefetch.py index 91fb78c6af..9a4a8e9e5c 100644 --- a/tests/unit_tests/test_platformio_prefetch.py +++ b/tests/unit_tests/test_platformio_prefetch.py @@ -1729,3 +1729,20 @@ def test_platformio_private_api_contract() -> None: derived = PackageSpec("https://x/y/archive/master.zip") assert derived.name and not derived.has_custom_name() assert PackageSpec("Foo=https://x/y/archive/master.zip").has_custom_name() + + +def test_preinstall_caps_workers( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A high core count is capped; the workers share one disk.""" + with ( + caplog.at_level(logging.INFO), + patch.object(pf, "get_usable_cpu_count", return_value=64), + ): + pf._preinstall( + _fake_manager(tmp_path), + [(f"p{i}@1", _FakeSpec(name=f"p{i}")) for i in range(11)], + ) + assert "Installing 11 PlatformIO package(s) with 10 extraction worker(s)" in ( + caplog.text + ) diff --git a/tests/unit_tests/test_platformio_registry.py b/tests/unit_tests/test_platformio_registry.py index 7a42dceb19..be7d31c0b7 100644 --- a/tests/unit_tests/test_platformio_registry.py +++ b/tests/unit_tests/test_platformio_registry.py @@ -815,20 +815,20 @@ def test_install_packages_first_failure_reraised( registry.install_packages( [_spec("a", "1.0", tmp_path / "a"), _spec("b", "2.0", tmp_path / "b")], dl ) - assert "Could not install" in caplog.text + # Every failure is named, including the re-raised one: its exception + # message may not identify the package + assert "Could not install a" in caplog.text + assert "Could not install b" in caplog.text -def test_install_package_extract_progress_suppresses_bars( - tmp_path: Path, caplog: pytest.LogCaptureFixture -) -> None: - """A batched install routes extraction fractions to the caller and keeps - both private bars and per-package INFO lines off the shared bar.""" +@contextmanager +def _batched_install(tmp_path: Path, extract_progress, prefill_archive: bool = True): + """Run a batched install_package of pkg@1.0.0; yields the download mock.""" dest = tmp_path / "pkg" - (tmp_path / "dl").mkdir() - (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x") - fractions: list[float] = [] + if prefill_archive: + (tmp_path / "dl").mkdir() + (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x") with ( - caplog.at_level(logging.INFO), patch.object(registry, "download_with_resume") as mock_download, patch.object(registry, "archive_extract_all") as mock_extract, patch.object( @@ -847,8 +847,22 @@ def test_install_package_extract_progress_suppresses_bars( [], tmp_path / "dl", expect=("payload",), - extract_progress=fractions.append, + extract_progress=extract_progress, ) + yield mock_download, mock_extract + + +def test_install_package_extract_progress_suppresses_bars( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A batched install routes extraction fractions to the caller and keeps + both private bars and per-package INFO lines off the shared bar.""" + fractions: list[float] = [] + with ( + caplog.at_level(logging.INFO), + _batched_install(tmp_path, fractions.append) as (mock_download, mock_extract), + ): + pass assert mock_extract.call_args[1]["progress"] == fractions.append # The download tracker reports zero bytes, keeping the shared bar honest download_progress = mock_download.call_args[1]["progress"] @@ -864,29 +878,11 @@ def test_install_package_batched_missing_archive_keeps_info_log( ) -> None: """A batched archive that unexpectedly needs a real download keeps the INFO line; the shared bar shows no progress for it.""" - dest = tmp_path / "pkg" with ( caplog.at_level(logging.INFO), - patch.object(registry, "download_with_resume"), - patch.object(registry, "archive_extract_all") as mock_extract, - patch.object( - registry, - "registry_download", - return_value=("http://x/pkg.tar.gz", "abc123", 42), - ), + _batched_install(tmp_path, lambda _frac: None, prefill_archive=False), ): - mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir( - parents=True - ) - registry.install_package( - "pkg", - "1.0.0", - dest, - [], - tmp_path / "dl", - expect=("payload",), - extract_progress=lambda _frac: None, - ) + pass assert "Downloading pkg 1.0.0" in caplog.text @@ -910,6 +906,8 @@ def test_install_packages_dedupes_duplicate_specs(tmp_path: Path) -> None: batched = [c for c in mock_install.call_args_list if "extract_progress" in c[1]] assert [(c[0][0], c[0][2]) for c in sequential] == [("a", tmp_path / "a2")] assert sorted(c[0][0] for c in batched) == ["a", "b"] + # The duplicate runs after the batch, which unlinks their shared archive + assert mock_install.call_args_list[-1] == sequential[0] def test_install_packages_caps_workers(tmp_path: Path) -> None: @@ -927,3 +925,58 @@ def test_install_packages_caps_workers(tmp_path: Path) -> None: ): registry.install_packages(specs, dl) assert batch.call_args.kwargs["max_workers"] == 10 + + +def test_install_package_batched_refetch_announced_once( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A batched archive that fails verification and refetches is announced; + a verify no-op (full size credited immediately) stays silent.""" + with ( + caplog.at_level(logging.INFO), + _batched_install(tmp_path, lambda _frac: None) as (mock_download, _), + ): + progress = mock_download.call_args[1]["progress"] + progress(42) + assert "Re-downloading pkg 1.0.0" not in caplog.text + progress(10) + progress(20) + assert caplog.text.count("Re-downloading pkg 1.0.0") == 1 + + +def test_install_package_batched_refetch_announced_without_size( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> None: + """A size-less registry entry still announces its refetch on the first + streaming tick.""" + with ( + caplog.at_level(logging.INFO), + patch.object(registry, "download_with_resume") as mock_download, + patch.object(registry, "archive_extract_all") as mock_extract, + patch.object( + registry, + "registry_download", + return_value=("http://x/pkg.tar.gz", "abc123", None), + ), + ): + dest = tmp_path / "pkg" + (tmp_path / "dl").mkdir() + (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x") + mock_extract.side_effect = lambda *_a, **_kw: (dest / "payload").mkdir( + parents=True + ) + registry.install_package( + "pkg", + "1.0.0", + dest, + [], + tmp_path / "dl", + expect=("payload",), + extract_progress=lambda _frac: None, + ) + progress = mock_download.call_args[1]["progress"] + # A verify no-op credits the whole (nonempty) file in one tick + progress(1) + assert "Re-downloading pkg 1.0.0" not in caplog.text + progress(0) + assert "Re-downloading pkg 1.0.0" in caplog.text