Merge branch 'dev' into platformio-prefetch-git-clones

# Conflicts:
#	esphome/platformio/prefetch.py

esphome/platformio/prefetch.py

@@ -16,7 +16,7 @@ name and promote with an atomic rename.
 
 from __future__ import annotations
 
-from collections.abc import Iterator
+from collections.abc import Iterable, Iterator
 from concurrent.futures import ThreadPoolExecutor
 from contextlib import contextmanager, suppress
 import hashlib
@@ -55,8 +55,8 @@ def _preserved_sys_path() -> Iterator[None]:
         sys.path[:] = saved
 
 
-# Concurrent registry resolutions / HEAD probes (each is network-bound)
-_RESOLVE_WORKERS = 8
+# Cap for network-bound work: resolutions, HEAD probes, clone floor
+_NETWORK_WORKERS = 8
 
 # A hung child must not block the build; downloads resume on the next run
 _PREFETCH_TIMEOUT = 20 * 60
@@ -341,7 +341,7 @@ def _registry_jobs(
     if not pending:
         return [], 0, []
     # Serial resolutions (registry GET + mirror HEAD each) dominate
-    with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(pending))) as ex:
+    with ThreadPoolExecutor(max_workers=min(_NETWORK_WORKERS, len(pending))) as ex:
         results = list(ex.map(_resolve, pending))
     jobs: list[tuple[str, int, Any]] = []
     installable: list[tuple[str, Any]] = []
@@ -374,14 +374,47 @@ def _registry_jobs(
     return jobs, failed, installable
 
 
+# The schemes pio's VCSClientFactory dispatches on (Git/Hg/SvnClient)
+_VCS_URI_PREFIXES = ("git+", "hg+", "svn+", "git://", "hg://", "svn://")
+
+
+def _is_vcs_spec_uri(url: str | None) -> bool:
+    """Whether pio's ``install_from_uri`` would clone this URI (PackageSpec
+    normalizes git URLs to ``git+``). The .git check runs first so an
+    un-normalized repo URL fails as a clone, not as an archive download."""
+    if not url or url.startswith(("file://", "symlink://")):
+        return False
+    if url.split("#", 1)[0].endswith(".git"):
+        return True
+    if url.startswith(("http://", "https://")):
+        return False
+    return url.startswith(_VCS_URI_PREFIXES)
+
+
+# (name, spec) from wave 1, (name, spec, compatibility) from dep waves
+_Entry = tuple[str, Any] | tuple[str, Any, Any]
+
+
+def _entry_is_vcs(entry: _Entry) -> bool:
+    """Whether this pre-install entry is cloned rather than unpacked."""
+    return _is_vcs_spec_uri(entry[1].uri)
+
+
+def _clones_first(entries: Iterable[_Entry]) -> list[_Entry]:
+    """Clones first: they wait on the network, so they must not queue
+    behind CPU-bound archive extractions in the pre-install pool."""
+    return sorted(entries, key=lambda entry: not _entry_is_vcs(entry))
+
+
 def _uri_jobs(
     manager: Any, specs: list[Any], seen: set[str]
 ) -> tuple[list[tuple[str, int, Any]], int, list[tuple[str, Any]]]:
     """Jobs for direct-URL specs; a HEAD sizes each for the combined bar.
 
     Also returns how many HEAD probes errored (an absent length is not an
-    error) and the ``(name, spec)`` pairs whose archives will be
-    installable.
+    error) and the ``(name, spec)`` pairs to pre-install: downloaded
+    archives, plus VCS specs, which have no archive -- the pre-install
+    itself clones them, in parallel instead of one at a time in pio run.
     """
     from esphome.net_retry import fetch_with_retry, http_request
 
@@ -389,13 +422,25 @@ def _uri_jobs(
     installable: list[tuple[str, Any]] = []
     for spec in specs:
         url = spec.uri
-        if not url or not url.startswith(("http://", "https://")):
-            continue  # git+/file specs are cloned/copied, not downloaded
-        if url.split("#", 1)[0].endswith(".git"):
-            continue  # bare-URL VCS spec; PlatformIO clones it
+        if not url:
+            continue
+        is_vcs = _is_vcs_spec_uri(url)
+        if not is_vcs and not url.startswith(("http://", "https://")):
+            if not url.startswith(("file://", "symlink://")):
+                _LOGGER.debug(
+                    "Unrecognized package URI, leaving it to pio run: %s", url
+                )
+            continue  # file/symlink specs are copied in place by pio run
         if manager.get_package(spec):
             continue
         name = spec.name or url.rsplit("/", 1)[-1]
+        if is_vcs:
+            # The pre-install clones it, gated like the cached-archive
+            # branch below: only a custom name is the destination dir.
+            # Platform tool specs always parse as custom-named
+            if spec.has_custom_name():
+                installable.append((name, spec))
+            continue
         # PlatformIO downloads URL specs with no checksum
         dl_path = Path(manager.compute_download_path(url, ""))
         if dl_path.is_file():
@@ -409,7 +454,7 @@ def _uri_jobs(
         if str(dl_path) in seen:
             continue  # another spec already claimed this .part
         seen.add(str(dl_path))
-        candidates.append((spec.name, url, dl_path, spec))
+        candidates.append((name, url, dl_path, spec))
 
     errors: list[str] = []
 
@@ -435,7 +480,7 @@ def _uri_jobs(
 
     if not candidates:
         return [], 0, installable
-    with ThreadPoolExecutor(max_workers=min(_RESOLVE_WORKERS, len(candidates))) as ex:
+    with ThreadPoolExecutor(max_workers=min(_NETWORK_WORKERS, len(candidates))) as ex:
         sizes = list(ex.map(_head_size, [url for _, url, _, _ in candidates]))
     jobs: list[tuple[str, int, Any]] = []
     failed = 0
@@ -583,10 +628,6 @@ def _uri_fetch_job(manager: Any, url: str, dl_path: Path, size: int) -> Any:
     return run
 
 
-# (name, spec) from wave 1, (name, spec, compatibility) from dep waves
-_Entry = tuple[str, Any] | tuple[str, Any, Any]
-
-
 def _dependency_entries(
     manager: Any, entries: list[_Entry], seen_names: set[str]
 ) -> list[_Entry]:
@@ -701,7 +742,14 @@ 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))
+    entries = _clones_first(entries)
+    clones = sum(1 for entry in entries if _entry_is_vcs(entry))
+    # Network-bound clones run wide even on small-core runners; capped
+    # since each worker builds a sibling manager and may run a
+    # postinstall, and a mixed wave's extractions inherit the width
+    workers = min(
+        max(get_usable_cpu_count(), min(clones, _NETWORK_WORKERS)), len(entries)
+    )
     # One manager per worker (_install mutates instance state); built
     # serially because construction rewires the shared manager logger
     managers: SimpleQueue = SimpleQueue()
@@ -733,7 +781,7 @@ def _preinstall(
             raise
 
     _LOGGER.info(
-        "Installing %d PlatformIO package(s) with %d extraction worker(s): %s",
+        "Installing %d PlatformIO package(s) with %d worker(s): %s",
         len(entries),
         workers,
         ", ".join(name for name, *_ in entries),

tests/unit_tests/test_platformio_prefetch.py

@@ -588,7 +588,8 @@ def test_registry_jobs_one_bad_spec_keeps_the_rest(tmp_path: Path) -> None:
 
 
 def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
-    """HEAD sizes direct-URL specs; git and unreachable URLs are skipped."""
+    """HEAD sizes direct-URL specs; VCS specs skip the download but are
+    still installable (the pre-install clones them in parallel)."""
     m = _fake_manager(tmp_path)
     resp = MagicMock()
     resp.headers = {"content-length": "2222"}
@@ -597,15 +598,14 @@ def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
             m,
             [
                 _FakeSpec(uri="https://x/big.zip", name="big", custom_name=True),
-                _FakeSpec(uri="git+https://x/repo.git", name="repo"),
-                _FakeSpec(uri="https://x/repo.git#v1", name="barevcs"),
+                _FakeSpec(uri="git+https://x/repo.git", name="repo", custom_name=True),
                 _FakeSpec(name="registry"),
             ],
             set(),
         )
     assert failed == 0
     assert [(n, s) for n, s, _ in jobs] == [("big", 2222)]
-    assert [n for n, _ in installable] == ["big"]
+    assert [n for n, _ in installable] == ["repo", "big"]
     # a successful HEAD with no Content-Length is a clean skip
     resp.headers = {}
     with patch("esphome.net_retry.http_request", return_value=resp):
@@ -614,6 +614,67 @@ def test_uri_jobs_head_sizes_the_bar(tmp_path: Path) -> None:
         ) == ([], 0, [])
 
 
+def test_uri_jobs_vcs_specs_installable_without_probe(tmp_path: Path) -> None:
+    """VCS specs never probe the network; custom-named uninstalled ones
+    pre-install, everything else is left to pio run."""
+    m = _fake_manager(tmp_path)
+    with patch("esphome.net_retry.http_request") as mock_head:
+        jobs, failed, installable = pf._uri_jobs(
+            m,
+            [
+                _FakeSpec(
+                    uri="git+https://x/tool.git#1.0", name="tool", custom_name=True
+                ),
+                _FakeSpec(uri="hg+https://x/old", name="mercurial", custom_name=True),
+                _FakeSpec(uri="git+https://x/derived.git", name="derived"),
+                # An un-normalized repo URL classifies as VCS (never as a
+                # downloadable archive), then drops here as derived-name
+                _FakeSpec(uri="https://x/unnorm.git", name="unnorm"),
+                _FakeSpec(uri="file:///local/dir", name="local"),
+                # A local .git path is copied in place, never cloned
+                _FakeSpec(uri="file:///local/repo.git", name="localgit"),
+                _FakeSpec(uri="symlink:///local/dir", name="link"),
+            ],
+            set(),
+        )
+    mock_head.assert_not_called()
+    assert (jobs, failed) == ([], 0)
+    assert [n for n, _ in installable] == ["tool", "mercurial"]
+    # Positive classification: an unknown scheme is left to pio run
+    with patch("esphome.net_retry.http_request") as mock_head:
+        assert pf._uri_jobs(
+            m, [_FakeSpec(uri="weird://x/pkg", name="weird")], set()
+        ) == ([], 0, [])
+    mock_head.assert_not_called()
+
+    m.get_package.return_value = object()  # already installed: warm and silent
+    with patch("esphome.net_retry.http_request"):
+        assert pf._uri_jobs(
+            m,
+            [
+                _FakeSpec(
+                    uri="git+https://x/tool.git#1.0", name="tool", custom_name=True
+                )
+            ],
+            set(),
+        ) == ([], 0, [])
+
+
+def test_clones_first_orders_vcs_before_archives() -> None:
+    """The pre-install pool receives clones first: they wait on the
+    network and must not queue behind CPU-bound archive extractions."""
+    archive = ("zip", _FakeSpec(uri="https://x/a.zip", name="zip"))
+    registry = ("reg", _FakeSpec(uri=None, name="reg"))
+    clone = ("repo", _FakeSpec(uri="git+https://x/repo.git", name="repo"))
+    ordered = pf._clones_first([archive, registry, clone])
+    assert ordered[0] == clone
+    # Stable partition: non-clone relative order is preserved
+    assert ordered[1:] == [archive, registry]
+    assert pf._entry_is_vcs(clone)
+    assert not pf._entry_is_vcs(archive)
+    assert not pf._entry_is_vcs(registry)
+
+
 def test_uri_jobs_head_failure_counts_as_unresolved(
     tmp_path: Path, caplog: pytest.LogCaptureFixture
 ) -> None:
@@ -1600,12 +1661,9 @@ def test_preinstall_dependency_wave_skips_seen_names(tmp_path: Path) -> None:
     assert installed == ["noise-c"]
 
 
-def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None:
-    """Each worker thread gets its own pre-built manager and installs
-    genuinely overlap (the barrier deadlocks a serial pool). The worker
-    count is pinned so a 1-CPU host cannot serialize the pool."""
-    barrier = threading.Barrier(2, timeout=5)
-    used: set = set()
+def _wave_manager(tmp_path, on_install):
+    """A minimal pio-manager stand-in for _preinstall pool tests;
+    ``on_install(manager, spec)`` observes each _install call."""
 
     class _WaveManager:
         package_dir = str(tmp_path)
@@ -1636,20 +1694,59 @@ def test_preinstall_uses_distinct_managers_in_parallel(tmp_path: Path) -> None:
             return None
 
         def _install(self, spec, skip_dependencies, compatibility=None) -> None:
-            used.add(id(self))
-            barrier.wait()
+            on_install(self, spec)
 
-    seed = _WaveManager(str(tmp_path))
-    with patch.object(pf, "get_usable_cpu_count", return_value=2):
+    return _WaveManager
+
+
+@pytest.mark.parametrize(
+    ("cpu_count", "entries"),
+    [
+        (2, [("a@1", _FakeSpec(name="a")), ("b@1", _FakeSpec(name="b"))]),
+        # The clone floor: network-bound clones run wide on a 1-CPU host
+        (
+            1,
+            [
+                (f"r{i}", _FakeSpec(uri=f"git+https://x/r{i}.git", name=f"r{i}"))
+                for i in range(4)
+            ],
+        ),
+    ],
+    ids=("cpu-sized", "clone-floor"),
+)
+def test_preinstall_pool_width(tmp_path: Path, cpu_count: int, entries: list) -> None:
+    """The barrier deadlocks unless every entry gets its own manager
+    and runs concurrently."""
+    barrier = threading.Barrier(len(entries), timeout=5)
+    used: set = set()
+
+    def on_install(mgr, spec) -> None:
+        used.add(id(mgr))
+        barrier.wait()
+
+    cls = _wave_manager(tmp_path, on_install)
+    seed = cls(str(tmp_path))
+    with patch.object(pf, "get_usable_cpu_count", return_value=cpu_count):
+        pf._preinstall(seed, entries)
+    assert len(used) == len(entries)
+    assert id(seed) not in used
+
+
+def test_preinstall_orders_clones_before_extractions(tmp_path: Path) -> None:
+    """With one worker, the clone installs before the archive
+    regardless of caller order."""
+    order: list[str] = []
+    cls = _wave_manager(tmp_path, lambda mgr, spec: order.append(spec.name))
+    seed = cls(str(tmp_path))
+    with patch.object(pf, "get_usable_cpu_count", return_value=1):
         pf._preinstall(
             seed,
             [
-                ("a@1", _FakeSpec(name="a")),
-                ("b@1", _FakeSpec(name="b")),
+                ("zip", _FakeSpec(uri="https://x/a.zip", name="zip")),
+                ("repo", _FakeSpec(uri="git+https://x/repo.git", name="repo")),
             ],
         )
-    assert len(used) == 2
-    assert id(seed) not in used
+    assert order == ["repo", "zip"]
 
 
 def test_sibling_manager_and_sigterm() -> None:
@@ -1803,3 +1900,13 @@ 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()
+    # _is_vcs_spec_uri relies on bare .git URLs normalizing to git+, on
+    # both parse paths (raw string, and requirements= for platform tools)
+    assert PackageSpec("https://github.com/x/y.git#v1").uri.startswith("git+")
+    platform_tool = PackageSpec(
+        owner="o", name="tool-x", requirements="https://github.com/x/y.git"
+    )
+    assert platform_tool.uri.startswith("git+")
+    # A URL requirement re-parses as name=url, marking the name custom;
+    # this is what keeps platform tool clones in the parallel pre-install
+    assert platform_tool.has_custom_name()