Merge remote-tracking branch 'origin/esp8266-native-parallel-extract' into esp8266-native-parallel-extract

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

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(

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

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.

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)

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(

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
+    )

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