[core] Clone git libraries in parallel in the library prefetch (#18836)

esphome/git.py

@@ -457,6 +457,15 @@ def _clone_complete_marker_path(repo_dir: Path) -> Path:
     return repo_dir / ".git" / _CLONE_COMPLETE_MARKER
 
 
+def has_complete_clone(
+    url: str, ref: str | None, domain: str, subpath: Path | None = None
+) -> bool:
+    """Lock-free probe for a complete clone; can go stale immediately, so
+    best-effort decisions only, never a substitute for ``clone_or_update``."""
+    repo_dir = _repo_entry_dir(_cache_key(url, ref), domain, subpath)
+    return _clone_complete_marker_path(repo_dir).is_file()
+
+
 def _clear_clone_complete_marker(repo_dir: Path) -> None:
     """Best-effort removal of the completion marker.
 

esphome/platformio/library.py

@@ -13,7 +13,7 @@ regardless of which toolchain consumes the result.
 """
 
 from collections import deque
-from collections.abc import Callable, Iterable
+from collections.abc import Callable, Hashable, Iterable
 from dataclasses import dataclass, field
 from functools import partial
 import glob
@@ -99,6 +99,17 @@ class Source:
     ) -> Path:
         raise NotImplementedError
 
+    def prefetch_key(self, dir_suffix: str) -> Hashable | None:
+        """Prefetch dedup identity; None = not prefetchable. Sources that
+        could write one cache dir must return equal keys (workers must never
+        share a dir); a coarser key only skips a prefetch."""
+        return None
+
+    def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool:
+        """Whether a completed fetch exists; only consulted when
+        ``prefetch_key()`` is not None, True is the safe default."""
+        return True
+
     def source_root(self, build_path: Path) -> Path:
         """Directory holding the library's own files (manifest + sources).
 
@@ -127,6 +138,9 @@ class URLSource(Source):
             h.update(salt.encode())
         return base_dir / h.hexdigest()[:8] / dir_suffix
 
+    def prefetch_key(self, dir_suffix: str) -> Hashable | None:
+        return self.url if self.size else None
+
     def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool:
         """Whether a completed extraction already exists for this source."""
         return (
@@ -177,14 +191,29 @@ class GitSource(Source):
         self.url = url
         self.ref = ref
 
-    def download(
-        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
-    ) -> Path:
+    @staticmethod
+    def _domain(salt: str, namespace: str) -> str:
         domain = DOMAIN
         if namespace:
             domain = f"{domain}/{namespace}"
         if salt:
             domain = f"{domain}/{salt}"
+        return domain
+
+    def prefetch_key(self, dir_suffix: str) -> Hashable | None:
+        # The clone target dir is hash(url@ref)/<dir_suffix>
+        return (self.url, self.ref, dir_suffix)
+
+    def is_cached(self, dir_suffix: str, salt: str = "", namespace: str = "") -> bool:
+        """Whether a completed clone already exists for this source."""
+        return git.has_complete_clone(
+            self.url, self.ref, self._domain(salt, namespace), Path(dir_suffix)
+        )
+
+    def download(
+        self, dir_suffix: str, force: bool = False, salt: str = "", namespace: str = ""
+    ) -> Path:
+        domain = self._domain(salt, namespace)
         path, _ = git.clone_or_update(
             url=self.url,
             ref=self.ref,
@@ -988,56 +1017,78 @@ def _fetch_source(
     )
 
 
+def _clone_source(
+    component: ConvertedLibrary,
+    salt: str,
+    namespace: str,
+    tracker: Callable[[int], None],
+) -> None:
+    # No byte progress from git; one tick so a cancelled batch stops here
+    tracker(0)
+    component.source.download(
+        component.get_sanitized_name(), salt=salt, namespace=namespace
+    )
+
+
 def _prefetch_wave(
     wave: list[tuple[str, ConvertedLibrary]], salt: str, namespace: str
 ) -> None:
-    """Best-effort parallel download of a wave's registry archives.
+    """Best-effort parallel fetch of a wave's registry archives and git clones.
 
-    The walk's own ``download()`` stays authoritative; duplicate URLs
+    The walk's own ``download()`` stays authoritative; duplicate sources
     prefetch once so two threads never share a cache directory. Archives
     whose size the registry did not report are left to the sequential
     loop, whose per-file bars don't interleave. A node a sibling in the
-    same wave supersedes has its archive fetched in vain (knowing better
+    same wave supersedes has its source fetched in vain (knowing better
     would need the manifests being downloaded).
     """
     try:
-        components: list[ConvertedLibrary] = []
-        seen: set[str] = set()
+        archives: list[ConvertedLibrary] = []
+        clones: list[ConvertedLibrary] = []
+        seen: set[Hashable] = set()
         for _key, component in wave:
             source = component.source
-            if not isinstance(source, URLSource) or not source.size:
+            name = component.get_sanitized_name()
+            dedup_key = source.prefetch_key(name)
+            if dedup_key is None or dedup_key in seen:
                 continue
-            if source.url in seen:
-                continue
-            seen.add(source.url)
+            seen.add(dedup_key)
             try:
-                cached = source.is_cached(
-                    component.get_sanitized_name(), salt=salt, namespace=namespace
-                )
+                cached = source.is_cached(name, salt=salt, namespace=namespace)
             except OSError as err:
                 # Best-effort, but visibly: a systematic probe failure makes
-                # every warm build re-download every archive
+                # every warm build re-fetch every source
                 _LOGGER.warning("Cache probe for %s failed: %s", component.name, err)
                 cached = False
             if cached:
                 # A warm build must stay silent
                 continue
-            components.append(component)
-        if not components:
+            (archives if isinstance(source, URLSource) else clones).append(component)
+        if not archives and not clones:
             return
         # Single-item waves (a dependency chain discovers one archive per
         # wave) go through the same runner: one download method, one bar
-        _LOGGER.info(
-            "Downloading %d library archive(s): %s",
-            len(components),
-            ", ".join(c.name for c in components),
-        )
+        if archives:
+            _LOGGER.info(
+                "Downloading %d library archive(s): %s",
+                len(archives),
+                ", ".join(c.name for c in archives),
+            )
+        if clones:
+            _LOGGER.info(
+                "Cloning %d library repo(s): %s",
+                len(clones),
+                ", ".join(c.name for c in clones),
+            )
         failures = run_batch_downloads(
             "Downloading libraries",
             [
                 (c.name, c.source.size, partial(_fetch_source, c, salt, namespace))
-                for c in components
-            ],
+                for c in archives
+            ]
+            # Size 0: clones share the worker pool without skewing the
+            # byte bar, whose total stays the archive sum
+            + [(c.name, 0, partial(_clone_source, c, salt, namespace)) for c in clones],
         )
         # The sequential call below retries and raises the real error
         warn_prefetch_failures(

tests/unit_tests/test_git.py

@@ -714,6 +714,25 @@ def test_run_git_command_without_git_dir_raises_error(
         git.run_git_command(["git", "clone", "https://invalid.url/repo.git"])
 
 
+def test_has_complete_clone(tmp_path: Path) -> None:
+    """The lock-free probe tracks the completion marker, subpath included."""
+    CORE.config_path = tmp_path / "test.yaml"
+
+    url = "https://github.com/test/repo"
+    subpath = Path("lib")
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    repo_dir = _compute_repo_dir(url, "v1", "test_domain") / subpath
+    (repo_dir / ".git").mkdir(parents=True)
+    # A directory without the marker is an incomplete clone
+    assert not git.has_complete_clone(url, "v1", "test_domain", subpath)
+
+    _mark_clone_complete(repo_dir)
+    assert git.has_complete_clone(url, "v1", "test_domain", subpath)
+    # The ref is part of the cache key
+    assert not git.has_complete_clone(url, "v2", "test_domain", subpath)
+
+
 def test_clone_or_update_with_never_refresh(
     tmp_path: Path, mock_run_git_command: Mock
 ) -> None:

tests/unit_tests/test_platformio_library.py

@@ -638,7 +638,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
 ) -> None:
     """Registry archives in one wave download concurrently, deduped by URL;
-    git/local sources and failures are left to the sequential call."""
+    local sources and failures are left to the sequential call."""
     calls: list[str] = []
 
     def fake_download(
@@ -658,7 +658,7 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
         # into the same cache directory)
         ("b2", ConvertedLibrary("b2", "1.0", URLSource("https://x/b.tar.gz", 1))),
         ("c", ConvertedLibrary("c", "1.0", URLSource("https://x/boom.tar.gz", 1))),
-        ("g", ConvertedLibrary("g", "*", lib.GitSource("https://x/g.git", None))),
+        ("l", ConvertedLibrary("l", "*", LocalSource("/some/lib"))),
     ]
     lib._prefetch_wave(wave, "", "idf")
     assert sorted(calls) == [
@@ -670,6 +670,83 @@ def test_prefetch_wave_downloads_registry_archives_in_parallel(
     assert "Prefetch of c failed (retrying sequentially)" in caplog.text
 
 
+def test_prefetch_wave_clones_git_sources_in_parallel(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """Git sources join the same prefetch batch as the archives, deduped by
+    clone target; a clone failure warns and is left to the sequential call."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+
+    def fake_clone(self, dir_suffix, force=False, salt="", namespace=""):
+        calls.append(f"{self}/{dir_suffix}")
+        if "boom" in self.url:
+            raise RuntimeError("boom")
+
+    monkeypatch.setattr(GitSource, "download", fake_clone)
+    wave = [
+        ("a", ConvertedLibrary("a", "1.0", URLSource("https://x/a.tar.gz", 1))),
+        ("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        # Same url@ref and target dir must clone once
+        ("g2", ConvertedLibrary("g", "*", GitSource("https://x/g.git", "v1"))),
+        ("h", ConvertedLibrary("h", "*", GitSource("https://x/boom.git", None))),
+    ]
+    monkeypatch.setattr(
+        URLSource, "download", lambda self, dir_suffix, progress=None, **kw: None
+    )
+    lib._prefetch_wave(wave, "", "idf")
+    assert sorted(calls) == ["https://x/boom.git/h", "https://x/g.git#v1/g"]
+    assert "Cloning 2 library repo(s): g, h" in caplog.text
+    assert "Prefetch of h failed (retrying sequentially)" in caplog.text
+
+
+def test_source_base_prefetch_defaults() -> None:
+    """The base Source is not prefetchable and reports cached (nothing to do)."""
+    source = Source()
+    assert source.prefetch_key("x") is None
+    assert source.is_cached("x") is True
+
+
+def test_prefetch_wave_single_clone_uses_the_batch(
+    monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """A wave with only git sources still clones through the batch runner."""
+    caplog.set_level("INFO")
+    calls: list[str] = []
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: False)
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, force=False, salt="", namespace="": calls.append(
+            self.url
+        ),
+    )
+    lib._prefetch_wave(
+        [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))],
+        "",
+        "idf",
+    )
+    assert calls == ["https://x/g.git"]
+    assert "Cloning 1 library repo(s): g" in caplog.text
+    assert "Downloading" not in caplog.text
+
+
+def test_prefetch_wave_warm_git_cache_is_silent(
+    setup_core, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
+) -> None:
+    """An already-complete clone is neither re-fetched nor announced."""
+    caplog.set_level("INFO")
+    monkeypatch.setattr(
+        GitSource,
+        "download",
+        lambda self, dir_suffix, **kw: (_ for _ in ()).throw(AssertionError("cloned")),
+    )
+    monkeypatch.setattr(GitSource, "is_cached", lambda self, *a, **kw: True)
+    wave = [("g", ConvertedLibrary("g", "*", GitSource("https://x/g.git", None)))]
+    lib._prefetch_wave(wave, "", "idf")
+    assert "Cloning" not in caplog.text
+
+
 def test_prefetch_wave_unknown_size_left_to_sequential(
     setup_core, monkeypatch: pytest.MonkeyPatch
 ) -> None: