Cap extraction workers and keep batch failures traceable

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(

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]

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"),
     [

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