diff --git a/esphome/framework_helpers.py b/esphome/framework_helpers.py index 843ce8b1cb..621d6d8546 100644 --- a/esphome/framework_helpers.py +++ b/esphome/framework_helpers.py @@ -788,6 +788,10 @@ def _stream_response_to_file( # hammering the host or the mirrors. BATCH_DOWNLOAD_WORKERS = 4 +# Concurrent archive extractions per batch; unpacking stops scaling well +# before high core counts since the workers share one disk. +BATCH_EXTRACT_WORKERS = 10 + def run_batch_downloads( header: str, @@ -972,12 +976,19 @@ def resume_fetch_job( 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-prefetch job; the caller's installer retries them.""" + """Warn per failed batch job, keeping the traceback of unexpected errors.""" + from esphome.core import EsphomeError # local import avoids circular dependency + 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)) - _LOGGER.debug("Prefetch failure detail", exc_info=err) + _LOGGER.warning( + message, name, failure_reason(err), exc_info=None if expected else err + ) + _LOGGER.debug(detail, exc_info=err) def download_with_resume( diff --git a/esphome/platformio/registry.py b/esphome/platformio/registry.py index 7cf639514c..aaf374bd67 100644 --- a/esphome/platformio/registry.py +++ b/esphome/platformio/registry.py @@ -14,6 +14,7 @@ from typing import NamedTuple from esphome.core import EsphomeError from esphome.framework_helpers import ( + BATCH_EXTRACT_WORKERS, archive_extract_all, download_from_mirrors, download_with_resume, @@ -303,8 +304,14 @@ def install_package( # Persistent location so an interrupted download resumes across runs. downloads_dir.mkdir(parents=True, exist_ok=True) archive = _archive_path(downloads_dir, name, version) - # The batch header already names each package - log = _LOGGER.debug if extract_progress is not None else _LOGGER.info + # The batch header already names each package; a batched archive that + # unexpectedly needs a real download keeps the INFO line, since the + # shared bar shows no progress for it + log = ( + _LOGGER.debug + if extract_progress is not None and archive.is_file() + else _LOGGER.info + ) log("Downloading %s %s ...", name, version) if mirrors: _LOGGER.warning( @@ -347,18 +354,22 @@ def install_packages(specs: Collection[PackageSpec], downloads_dir: Path) -> Non """ pending: list[tuple[PackageSpec, int]] = [] rest: list[PackageSpec] = [] + seen: set[str] = set() for spec in specs: name, version, dest, mirrors, _expect = spec - if _already_installed(dest) or mirrors: + # 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: 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 - except OSError: + except FileNotFoundError: rest.append(spec) continue + seen.add(f"{name}-{version}") pending.append((spec, size)) if len(pending) < 2: rest = list(specs) @@ -367,7 +378,7 @@ def install_packages(specs: Collection[PackageSpec], downloads_dir: Path) -> Non install_package(name, version, dest, mirrors, downloads_dir, expect=expect) if not pending: return - workers = min(get_usable_cpu_count(), len(pending)) + workers = min(get_usable_cpu_count(), len(pending), BATCH_EXTRACT_WORKERS) _LOGGER.info( "Extracting %d package archive(s) with %d worker(s): %s", len(pending), @@ -393,5 +404,7 @@ 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") + warn_prefetch_failures( + failures[1:], "Could not install %s: %s", detail="Install failure detail" + ) raise failures[0][1] diff --git a/tests/unit_tests/test_framework_helpers.py b/tests/unit_tests/test_framework_helpers.py index 81c1948605..012bd2f608 100644 --- a/tests/unit_tests/test_framework_helpers.py +++ b/tests/unit_tests/test_framework_helpers.py @@ -2357,6 +2357,25 @@ def test_warn_prefetch_failures_names_each_failure( assert "Prefetch of lib failed: gone" in caplog.text +def test_warn_prefetch_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 + + with caplog.at_level(logging.DEBUG): + warn_prefetch_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 + + @pytest.mark.parametrize( ("platform", "input_path", "expected"), [ diff --git a/tests/unit_tests/test_platformio_registry.py b/tests/unit_tests/test_platformio_registry.py index 081ecf6b1f..7a42dceb19 100644 --- a/tests/unit_tests/test_platformio_registry.py +++ b/tests/unit_tests/test_platformio_registry.py @@ -824,6 +824,8 @@ def test_install_package_extract_progress_suppresses_bars( """A batched install routes extraction fractions to the caller and keeps both private bars and per-package INFO lines off the shared bar.""" dest = tmp_path / "pkg" + (tmp_path / "dl").mkdir() + (tmp_path / "dl" / "pkg-1.0.0").write_bytes(b"x") fractions: list[float] = [] with ( caplog.at_level(logging.INFO), @@ -855,3 +857,73 @@ def test_install_package_extract_progress_suppresses_bars( assert fractions == [0.0] assert "Downloading pkg" not in caplog.text assert "Extracting pkg" not in caplog.text + + +def test_install_package_batched_missing_archive_keeps_info_log( + tmp_path: Path, caplog: pytest.LogCaptureFixture +) -> 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), + ), + ): + 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, + ) + assert "Downloading pkg 1.0.0" in caplog.text + + +def test_install_packages_dedupes_duplicate_specs(tmp_path: Path) -> None: + """Duplicate (name, version) entries share one archive and would race + each other; the duplicate takes the sequential path.""" + dl = tmp_path / "dl" + dl.mkdir() + (dl / "a-1.0").write_bytes(b"x") + (dl / "b-2.0").write_bytes(b"y") + specs = [ + _spec("a", "1.0", tmp_path / "a"), + _spec("a", "1.0", tmp_path / "a2"), + _spec("b", "2.0", tmp_path / "b"), + ] + with patch.object(registry, "install_package") as mock_install: + registry.install_packages(specs, dl) + sequential = [ + c for c in mock_install.call_args_list if "extract_progress" not in c[1] + ] + 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"] + + +def test_install_packages_caps_workers(tmp_path: Path) -> None: + """A high core count is capped; the workers share one disk.""" + dl = tmp_path / "dl" + dl.mkdir() + specs = [] + for i in range(12): + (dl / f"p{i}-1.0").write_bytes(b"x") + specs.append(_spec(f"p{i}", "1.0", tmp_path / f"p{i}")) + with ( + patch.object(registry, "get_usable_cpu_count", return_value=64), + patch.object(registry, "run_batch_downloads", return_value=[]) as batch, + patch.object(registry, "install_package"), + ): + registry.install_packages(specs, dl) + assert batch.call_args.kwargs["max_workers"] == 10