From c0f494450df1eb26ad2dd77cf1df61746feec265 Mon Sep 17 00:00:00 2001 From: "J. Nick Koston" Date: Thu, 30 Jul 2026 11:44:12 -1000 Subject: [PATCH] [git] Fix device adoption failing on first attempt: lock the clone cache against concurrent resolutions (#17923) --- esphome/components/packages/__init__.py | 22 +- esphome/git.py | 420 ++++++++++-- requirements.txt | 2 +- .../component_tests/packages/test_packages.py | 69 ++ tests/unit_tests/test_git.py | 603 +++++++++++++++++- 5 files changed, 1068 insertions(+), 48 deletions(-) diff --git a/esphome/components/packages/__init__.py b/esphome/components/packages/__init__.py index 44a1ebf36e..6cb9d5f03a 100644 --- a/esphome/components/packages/__init__.py +++ b/esphome/components/packages/__init__.py @@ -1,6 +1,7 @@ from collections import UserDict from collections.abc import Callable from functools import reduce +import logging from pathlib import Path from typing import Any @@ -35,6 +36,8 @@ from esphome.const import ( ) from esphome.core import EsphomeError +_LOGGER = logging.getLogger(__name__) + DOMAIN = CONF_PACKAGES # Guard against infinite include chains (e.g. A includes B includes A). MAX_INCLUDE_DEPTH = 20 @@ -267,8 +270,23 @@ def _process_remote_package(config: dict[str, Any]) -> dict[str, Any]: # If loading fails, the cached checkout may be stale — revert and retry once. try: return {CONF_PACKAGES: get_packages(files)} - except cv.Invalid: - revert() + except cv.Invalid as err: + if not revert(): + # The pre-update content is out of reach (lock timeout, the + # checkout moved, or the reset failed; see the log), so a + # retry could not see it. + raise cv.Invalid( + f"Failed to load packages and could not revert the cached " + f"checkout to retry. {err}", + path=err.path, + ) from err + # If the retry succeeds this is the only trace that the + # refreshed upstream content was broken. + _LOGGER.warning( + "Loading packages failed (%s), reverted the cached checkout " + "and retrying", + err, + ) try: return {CONF_PACKAGES: get_packages(files)} except cv.Invalid as err: diff --git a/esphome/git.py b/esphome/git.py index 46cce50d9d..b5abf39a24 100644 --- a/esphome/git.py +++ b/esphome/git.py @@ -1,5 +1,8 @@ -from collections.abc import Callable +from collections.abc import Callable, Iterator +from contextlib import contextmanager from dataclasses import dataclass +from enum import Enum, auto +import errno import hashlib import logging import os @@ -8,23 +11,42 @@ import re import subprocess import sys import time +from typing import TYPE_CHECKING import urllib.parse import esphome.config_validation as cv from esphome.core import CORE, EsphomeError, TimePeriodSeconds from esphome.helpers import add_git_ceiling_directory, rmtree, write_file +if TYPE_CHECKING: + from filelock import FileLock + _LOGGER = logging.getLogger(__name__) # Special value to indicate never refresh NEVER_REFRESH = TimePeriodSeconds(seconds=-1) -# Written inside .git only after every clone step (clone, ref fetch, reset, -# submodule init) has completed. A directory without it is an interrupted -# clone (e.g. the process was killed mid-clone) and must be re-cloned; without -# this check such a directory would be trusted forever when the caller uses -# NEVER_REFRESH. Lives in .git so stash/reset/checkout can never touch it and -# it does not pollute the worktree. +# revert() runs on an already-failing path; bound its wait for the cache +# entry lock so that recovery cannot hang forever behind another process. +_REVERT_LOCK_TIMEOUT_SECONDS = 60 + +# When a complete cache entry already exists, a caller does not wait forever +# behind another process's stalled clone or update (git sets no network +# timeouts): after this bound it uses the existing clone without refreshing +# it. With no complete entry there is nothing to fall back to, so the wait +# is unbounded. +_COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS = 60 + +# Written inside .git only while the entry is a complete, quiescent +# checkout: after every clone step (clone, ref fetch, reset, submodule init) +# has finished, and removed for the duration of a refresh's rewrite +# (stash/fetch/reset). A directory without it is an interrupted clone or +# update (e.g. the process was killed mid-clone) and must be re-cloned; +# without this check such a directory would be trusted forever when the +# caller uses NEVER_REFRESH, and the bounded-wait fallback would hand a +# mid-rewrite tree to a timed-out peer. Lives in .git so +# stash/reset/checkout can never touch it and it does not pollute the +# worktree. _CLONE_COMPLETE_MARKER = "esphome_clone_complete" # Environment variables that scope git to a specific repository. Git hooks and @@ -149,6 +171,16 @@ def run_git_command( return ret.stdout.decode("utf-8").strip() +def _cache_key(url: str, ref: str | None) -> str: + """Cache key identifying one repository checkout. + + The lock path and the entry directory both hash this, keeping them in + agreement. (micro_wake_word still rebuilds the format by hand to locate + manifests; fold it in here if the format ever changes.) + """ + return f"{url}@{ref}" + + def _compute_destination_path(key: str, domain: str) -> Path: base_dir = Path(CORE.data_dir) / domain h = hashlib.new("sha256") @@ -156,22 +188,195 @@ def _compute_destination_path(key: str, domain: str) -> Path: return base_dir / h.hexdigest()[:8] +def _repo_entry_dir(key: str, domain: str, subpath: Path | None) -> Path: + """Worktree directory of one cache entry: the hash dir plus optional subpath.""" + repo_dir = _compute_destination_path(key, domain) + if subpath: + repo_dir = repo_dir / subpath + return repo_dir + + +def _repo_lock_path(key: str, domain: str) -> Path: + """Path of the lock file serializing all work on one cache entry. + + Lives next to the hash directory, never inside it, so the removal of a + broken or incomplete clone can never delete a lock another process holds. + """ + repo_dir = _compute_destination_path(key, domain) + return repo_dir.parent / f"{repo_dir.name}.lock" + + +class _LockStatus(Enum): + ACQUIRED = auto() + # A bounded wait expired while another process held the lock. + TIMEOUT = auto() + # The lock could not be taken at all; callers proceed unlocked, + # matching the behavior before the lock existed. + UNAVAILABLE = auto() + + +# Errnos that mean the filesystem genuinely cannot take file locks (NFS +# without a lock daemon, some FUSE mounts). Any other OSError (permissions, +# read-only volume, full disk) is a cache directory problem, which the git +# commands themselves report clearly when it actually matters. EPERM is +# deliberately absent: it usually means a permissions problem, so it takes +# the generic message that names no cause. On Linux ENOTSUP and EOPNOTSUPP +# are the same value; the set folds them. +_NO_LOCK_SUPPORT_ERRNOS = frozenset( + {errno.ENOLCK, errno.ENOSYS, errno.EOPNOTSUPP, errno.ENOTSUP} +) + + +def _acquire_repo_lock( + lock: "FileLock", + safe_key: str, + timeout: float, + wait_message: str = "Waiting for another process to finish updating %s", +) -> _LockStatus: + """Acquire ``lock``, logging ``wait_message`` when a wait actually begins. + + ``timeout`` of -1 waits forever; a positive value bounds the wait and + can yield ``TIMEOUT``. + """ + from filelock import Timeout + + try: + try: + lock.acquire(blocking=False) + except Timeout: + # Waiting on another process's clone or update can take + # minutes; say so instead of appearing hung. + _LOGGER.info(wait_message, safe_key) + lock.acquire(timeout=timeout) + except Timeout: + return _LockStatus.TIMEOUT + except OSError as err: + if err.errno in _NO_LOCK_SUPPORT_ERRNOS: + _LOGGER.warning( + "The filesystem does not support locking the cache entry for " + "%s (%s), continuing without a lock", + safe_key, + err, + ) + else: + # Not a locking problem (permissions, read-only volume, full + # disk). Still continue unlocked: a pre-seeded read-only cache + # with refresh disabled only reads and must keep working, and + # in every other case the git commands fail with the real error. + _LOGGER.warning( + "Could not take the cache entry lock for %s (%s), " + "continuing without a lock", + safe_key, + err, + ) + return _LockStatus.UNAVAILABLE + return _LockStatus.ACQUIRED + + +@contextmanager +def _repo_cache_lock( + key: str, domain: str, repo_dir: Path +) -> Iterator[tuple[bool, "FileLock | None"]]: + """Hold the cache entry lock for ``key`` over the with block. + + Yields ``(use_existing, lock)``. ``use_existing`` is True when the lock + could not be acquired within the bounded wait but ``repo_dir`` is a + complete cache entry; the caller should use it as-is and do nothing + else. Otherwise ``lock`` is the held lock, released when the block + exits, or ``None`` when the lock could not be taken at all and the + caller proceeds unlocked. + """ + # Lazy import: keeps filelock off the CLI startup import path. + from filelock import FileLock + + safe_key = _redact_url_credentials(key) + # acquire() creates the lock file's directory itself; git clone later + # creates the hash directory next to it. fallback_to_soft would silently + # downgrade ENOSYS to a SoftFileLock, whose stale existence marker from + # another host on a shared cache could hang the unbounded wait forever; + # routing it through the OSError handler runs unlocked instead. + lock: FileLock | None = FileLock( + str(_repo_lock_path(key, domain)), fallback_to_soft=False + ) + status = _acquire_repo_lock(lock, safe_key, _COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS) + if status is _LockStatus.TIMEOUT: + if _clone_complete_marker_path(repo_dir).is_file(): + # Mutual exclusion matters most while no complete entry exists + # (initial clone, recovery re-clone); with one on disk, reading + # it beats hanging behind a stalled holder. + _LOGGER.warning( + "Timed out waiting for another process updating %s, proceeding " + "with the existing clone, which that process may still be " + "changing", + safe_key, + ) + yield True, None + return + # Nothing to fall back to; the holder is producing the clone this + # caller needs. + status = _acquire_repo_lock( + lock, + safe_key, + timeout=-1, + wait_message="Still waiting for the clone of %s, " + "there is no existing clone to fall back on", + ) + if status is not _LockStatus.ACQUIRED: + lock = None + try: + yield False, lock + finally: + if lock is not None: + lock.release() + + def _clone_complete_marker_path(repo_dir: Path) -> Path: return repo_dir / ".git" / _CLONE_COMPLETE_MARKER +def _clear_clone_complete_marker(repo_dir: Path) -> None: + """Best-effort removal of the completion marker. + + If the unlink fails (e.g. a file lock on Windows), the marker stays and + the entry keeps its previous trust level; every consumer of the marker + tolerates that. + """ + try: + _clone_complete_marker_path(repo_dir).unlink(missing_ok=True) + except OSError as err: + _LOGGER.debug("Could not delete clone completion marker: %s", err) + + +def _write_clone_complete_marker( + repo_dir: Path, key: str, hash_dir_name: str, safe_key: str +) -> None: + """Mark the entry as a complete, quiescent checkout. + + The key and hash dir name are recorded purely to make cache debugging + easier. The marker is only a validity signal, so a failed write must not + fail an otherwise complete clone or update: the only cost is a re-clone + on the next run. + """ + try: + write_file( + _clone_complete_marker_path(repo_dir), + f"key={key}\nhash={hash_dir_name}\n", + ) + except EsphomeError as err: + _LOGGER.warning( + "Could not write clone completion marker for %s: %s", safe_key, err + ) + + def _remove_repo_dir(repo_dir: Path) -> None: """Remove a repo directory, deleting the completion marker first. Marker-first ordering guarantees an interrupted removal can never leave a marker behind next to a partially deleted worktree. The unlink is best - effort: if it fails (e.g. a file lock on Windows), rmtree below still - gets the chance to remove the directory, marker included. + effort: if it fails, rmtree below still gets the chance to remove the + directory, marker included. """ - try: - _clone_complete_marker_path(repo_dir).unlink(missing_ok=True) - except OSError as err: - _LOGGER.debug("Could not delete clone completion marker first: %s", err) + _clear_clone_complete_marker(repo_dir) if repo_dir.is_dir(): rmtree(repo_dir) @@ -286,16 +491,79 @@ def resolve_symlink_stub(repo_dir: Path, file_path: Path) -> Path | None: def clone_or_update( *, url: str, - ref: str = None, + ref: str | None = None, refresh: TimePeriodSeconds | None, domain: str, - username: str = None, - password: str = None, + username: str | None = None, + password: str | None = None, init_submodules: bool = False, subpath: Path | None = None, +) -> tuple[Path, Callable[[], bool] | None]: + """Clone a repository into the cache, or refresh an existing clone. + + All work runs under a per-cache-entry inter-process file lock, so + concurrent resolutions of the same repository (two esphome processes, or + a subprocess plus an in-process load) serialize instead of interleaving. + Without the lock, ``repo_dir.is_dir()`` is true from the instant + ``git clone`` creates the directory: a second caller could read a half + populated worktree, or see the missing completion marker and delete the + clone in progress out from under the first caller. + + The lock guards mutation of the cache entry only; it is released when + this function returns, so a caller still reading the worktree can + overlap a later refresh by another process. That residual window is + narrow (the refresh interval is re-checked under the lock) and predates + the lock. + + Locking is best effort: on a filesystem that cannot take file locks a + warning is logged and the work proceeds unlocked, matching the behavior + before the lock existed. A complete cache entry also caps the wait: if + the holder is still busy after a bounded time (e.g. stalled on the + network), the existing clone is used without refreshing it, so a stuck + process cannot hang every peer that already has a good entry. + """ + key = _cache_key(url, ref) + repo_dir = _repo_entry_dir(key, domain, subpath) + with _repo_cache_lock(key, domain, repo_dir) as (use_existing, lock): + if use_existing: + return repo_dir, None + return _clone_or_update_locked( + url=url, + ref=ref, + refresh=refresh, + domain=domain, + username=username, + password=password, + init_submodules=init_submodules, + subpath=subpath, + lock=lock, + ) + + +def _clone_or_update_locked( + *, + url: str, + ref: str | None, + refresh: TimePeriodSeconds | None, + domain: str, + username: str | None, + password: str | None, + init_submodules: bool, + subpath: Path | None, + lock: "FileLock | None", _recover_broken: bool = True, -) -> tuple[Path, Callable[[], None] | None]: - key = f"{url}@{ref}" +) -> tuple[Path, Callable[[], bool] | None]: + """Body of ``clone_or_update``; the caller holds ``lock``. + + Split out because the broken-repository recovery below re-enters this + function: re-acquiring the already-held lock would deadlock, since OS + file locks taken on separate file descriptors conflict even within one + process. ``lock`` is only re-acquired by the returned ``revert`` + callback, which runs after the wrapper's ``finally`` has released it. + ``lock`` is ``None`` when the filesystem cannot take file locks and the + wrapper fell back to running unlocked. + """ + key = _cache_key(url, ref) # The user may have embedded credentials in the URL itself; log this # instead of key. safe_key = _redact_url_credentials(key) @@ -309,10 +577,8 @@ def clone_or_update( "://", f"://{urllib.parse.quote(username)}:{urllib.parse.quote(password)}@" ) - repo_dir = _compute_destination_path(key, domain) - hash_dir_name = repo_dir.name - if subpath: - repo_dir = repo_dir / subpath + hash_dir_name = _compute_destination_path(key, domain).name + repo_dir = _repo_entry_dir(key, domain, subpath) if repo_dir.is_dir() and not _clone_complete_marker_path(repo_dir).is_file(): # The last clone never finished (killed process, container stop) or @@ -353,19 +619,8 @@ def clone_or_update( _remove_repo_dir(repo_dir) raise - # Every git step succeeded; the key and hash dir name are recorded - # purely to make cache debugging easier. The marker is only a - # validity signal, so a failed write must not fail an otherwise - # complete clone: the only cost is a re-clone on the next run. - try: - write_file( - _clone_complete_marker_path(repo_dir), - f"key={key}\nhash={hash_dir_name}\n", - ) - except EsphomeError as err: - _LOGGER.warning( - "Could not write clone completion marker for %s: %s", safe_key, err - ) + # Every git step succeeded. + _write_clone_complete_marker(repo_dir, key, hash_dir_name, safe_key) else: if refresh == NEVER_REFRESH or CORE.skip_external_update: @@ -396,6 +651,13 @@ def clone_or_update( _LOGGER.info("Updating %s", safe_key) _LOGGER.debug("Location: %s", repo_dir) + # The entry is about to be rewritten; drop the marker so a + # timed-out peer's fallback and the incomplete-entry check + # can tell a quiescent complete entry from one mid-rewrite, + # and so an update interrupted by a crash re-clones instead + # of being trusted. + _clear_clone_complete_marker(repo_dir) + # Stash local changes (if any) # Use git_dir to ensure this only affects the specific repo run_git_command( @@ -425,6 +687,15 @@ def clone_or_update( # refresh window would silently accept on the next run. if init_submodules: update_submodules(repo_dir, key) + + # Recorded so revert() can tell whether the checkout is + # still the one this update produced. + new_sha = run_git_command( + ["git", "rev-parse", "HEAD"], git_dir=repo_dir + ) + + # The rewrite finished; the entry is trustworthy again. + _write_clone_complete_marker(repo_dir, key, hash_dir_name, safe_key) except GitException as err: # Repository is in a broken state or update failed # Only attempt recovery once to prevent infinite recursion @@ -444,9 +715,10 @@ def clone_or_update( _remove_repo_dir(repo_dir) _LOGGER.info("Successfully removed broken repository, re-cloning...") - # Recursively call clone_or_update to re-clone - # Set _recover_broken=False to prevent infinite recursion - result = clone_or_update( + # Re-clone while still holding the lock; going through the + # public wrapper would try to re-acquire it and deadlock. + # Set _recover_broken=False to prevent infinite recursion. + result = _clone_or_update_locked( url=original_url, ref=ref, refresh=refresh, @@ -455,14 +727,80 @@ def clone_or_update( password=password, init_submodules=init_submodules, subpath=subpath, + lock=lock, _recover_broken=False, ) _LOGGER.info("Repository %s successfully recovered", safe_key) return result - def revert(): - _LOGGER.info("Reverting changes to %s -> %s", safe_key, old_sha) - run_git_command(["git", "reset", "--hard", old_sha], git_dir=repo_dir) + def revert() -> bool: + """Reset the checkout to the pre-update SHA. + + Returns False when the revert did not happen: the cache + entry lock could not be acquired in time, the checkout + moved since this update (another process refreshed it), or + the reset itself failed. A retry cannot reach the + pre-update content then. + """ + if lock is None: + # The wrapper already warned about the unlockable + # filesystem; revert unlocked like everything else. + status = _LockStatus.UNAVAILABLE + else: + status = _acquire_repo_lock( + lock, safe_key, _REVERT_LOCK_TIMEOUT_SECONDS + ) + if status is _LockStatus.TIMEOUT: + # revert() only runs on an already-failing path; skip + # rather than hang so the original error can surface. + _LOGGER.warning( + "Could not lock %s to revert to %s, skipping revert; " + "the cached checkout keeps the un-reverted content " + "until its next refresh", + safe_key, + old_sha, + ) + return False + try: + # Anything can happen between the wrapper releasing the + # lock and revert() re-acquiring it; only undo this + # process's own update, never a peer's newer refresh. + head = run_git_command( + ["git", "rev-parse", "HEAD"], git_dir=repo_dir + ) + if head != new_sha: + _LOGGER.warning( + "Not reverting %s: the checkout moved since this " + "update (another process refreshed it)", + safe_key, + ) + return False + # Announced only once every skip check has passed, so + # the log says exactly one thing per outcome. + _LOGGER.info("Reverting changes to %s -> %s", safe_key, old_sha) + run_git_command( + ["git", "reset", "--hard", old_sha], git_dir=repo_dir + ) + except GitException as err: + # GitException is a cv.Invalid; letting it escape would + # replace the caller's original error with a bare git + # message. Report the failed reset like the skip above, + # and drop the marker: an entry whose reset fails cannot + # be trusted, so the next use re-clones it instead of + # the refresh window silently accepting it. + _LOGGER.warning( + "Could not revert %s to %s (%s), the entry will be " + "re-cloned on next use", + safe_key, + old_sha, + err, + ) + _clear_clone_complete_marker(repo_dir) + return False + finally: + if status is _LockStatus.ACQUIRED: + lock.release() + return True return repo_dir, revert diff --git a/requirements.txt b/requirements.txt index 541b374b15..8ba908d8b4 100644 --- a/requirements.txt +++ b/requirements.txt @@ -27,7 +27,7 @@ smpclient==7.2.0 requests==2.34.2 py7zr==1.1.3 platformdirs==4.11.0 # native esp-idf toolchain global cache dir -filelock==3.32.0 # lock guarding the PlatformIO python-version cache heal +filelock==3.32.0 # inter-process locks (PlatformIO cache heal, git clone cache); >=3.32 for FileLock(fallback_to_soft=...), older versions silently drop the kwarg # esp-idf >= 5.0 requires this pyparsing >= 3.3.2 diff --git a/tests/component_tests/packages/test_packages.py b/tests/component_tests/packages/test_packages.py index 6990c1c051..39bffd31b7 100644 --- a/tests/component_tests/packages/test_packages.py +++ b/tests/component_tests/packages/test_packages.py @@ -1264,6 +1264,75 @@ def test_remote_packages_no_revert( ] +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +@patch("esphome.git.clone_or_update") +def test_remote_packages_skipped_revert_does_not_retry( + mock_clone_or_update, mock_is_file, mock_load_yaml +) -> None: + """When revert() reports the rollback was skipped, the load is not + retried (the checkout is unchanged) and the error says so.""" + mock_revert = MagicMock(return_value=False) + mock_clone_or_update.return_value = (Path("/tmp/noexists"), mock_revert) + mock_is_file.return_value = True + mock_load_yaml.side_effect = cv.Invalid("bad yaml") + + config = { + CONF_PACKAGES: { + "pkg": { + CONF_URL: "https://github.com/esphome/repo", + CONF_REF: "main", + CONF_FILES: [{CONF_PATH: "file.yaml"}], + CONF_REFRESH: "1d", + } + } + } + with pytest.raises(cv.Invalid, match="could not revert the cached checkout"): + packages_pass(config) + + assert mock_revert.call_count == 1 + assert mock_load_yaml.call_count == 1 + + +@patch("esphome.yaml_util.load_yaml") +@patch("pathlib.Path.is_file") +@patch("esphome.git.clone_or_update") +def test_remote_packages_successful_revert_retries( + mock_clone_or_update, mock_is_file, mock_load_yaml, caplog: pytest.LogCaptureFixture +) -> None: + """A successful revert retries the load against the reverted checkout and + logs the original error, the only trace that upstream was broken.""" + mock_revert = MagicMock(return_value=True) + mock_clone_or_update.return_value = (Path("/tmp/noexists"), mock_revert) + mock_is_file.return_value = True + mock_load_yaml.side_effect = [ + cv.Invalid("bad yaml"), + OrderedDict( + {CONF_SENSOR: [{CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"}]} + ), + ] + + config = { + CONF_PACKAGES: { + "pkg": { + CONF_URL: "https://github.com/esphome/repo", + CONF_REF: "main", + CONF_FILES: [{CONF_PATH: "file.yaml"}], + CONF_REFRESH: "1d", + } + } + } + with caplog.at_level(logging.WARNING): + actual = packages_pass(config) + + assert actual[CONF_SENSOR] == [ + {CONF_PLATFORM: TEST_SENSOR_PLATFORM_1, CONF_NAME: "test"} + ] + assert mock_revert.call_count == 1 + assert mock_load_yaml.call_count == 2 + assert any("reverted the cached checkout" in r.getMessage() for r in caplog.records) + + def test_raw_config_contains_merged_esphome_from_package(tmp_path) -> None: """Test that CORE.raw_config contains esphome section from merged package. diff --git a/tests/unit_tests/test_git.py b/tests/unit_tests/test_git.py index 858eee5e9f..13283fc067 100644 --- a/tests/unit_tests/test_git.py +++ b/tests/unit_tests/test_git.py @@ -1,14 +1,17 @@ """Tests for git.py module.""" from collections.abc import Callable +import errno import logging import os from pathlib import Path import subprocess +import threading import time from typing import Any from unittest.mock import Mock, patch +from filelock import FileLock import pytest from esphome import git @@ -74,17 +77,23 @@ def _simulate_cloned_repo(repo_dir: Path) -> None: def _make_clone_side_effect( - repo_dir: Path, gitmodules: bool = False + repo_dir: Path, + gitmodules: bool = False, + on_clone: Callable[[], None] | None = None, ) -> Callable[..., str]: """Return a run_git_command side effect whose clone creates the repo dir. - With ``gitmodules`` the cloned repo also declares submodules. + With ``gitmodules`` the cloned repo also declares submodules. ``on_clone`` + runs at clone time before the repo dir appears, so a test can probe or + block mid-clone. """ def git_command_side_effect( cmd: list[str], cwd: str | None = None, **kwargs: Any ) -> str: if _get_git_command_type(cmd) == "clone": + if on_clone is not None: + on_clone() _simulate_cloned_repo(repo_dir) if gitmodules: (repo_dir / ".gitmodules").write_text("test") @@ -1491,6 +1500,591 @@ def test_refresh_submodule_failure_recovers_then_raises( ) +def _lock_path(url: str, ref: str | None, domain: str) -> Path: + """The lock file the implementation uses for one cache entry.""" + return git._repo_lock_path(git._cache_key(url, ref), domain) + + +class _SetEventOnWaitLog(logging.Handler): + """Set an event when the 'Waiting for another process' record is emitted, + so tests can react to a caller observably blocking on the lock instead of + racing a wall-clock timer.""" + + def __init__(self, event: threading.Event) -> None: + super().__init__() + self._event = event + + def emit(self, record: logging.LogRecord) -> None: + if "Waiting for another process" in record.getMessage(): + self._event.set() + + +def test_clone_or_update_serializes_concurrent_clones( + tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """Two concurrent callers for the same uncached repo must not both clone. + + Without the per-entry lock both callers pass the is_dir() check before + either clone finishes, so the second one either clones on top of the + first or reads a half populated worktree (device-builder issue 2425). + """ + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + + start_together = threading.Barrier(2) + other_caller_waiting = threading.Event() + + def on_clone() -> None: + # Hold the lock until the other caller is observably blocked on it, + # so the interleaving is guaranteed rather than raced on a timer. + assert other_caller_waiting.wait(timeout=30) + + mock_run_git_command.side_effect = _make_clone_side_effect( + repo_dir, on_clone=on_clone + ) + + results: list[Path] = [] + errors: list[BaseException] = [] + + def call() -> None: + try: + start_together.wait() + result_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain + ) + results.append(result_dir) + except BaseException as err: # noqa: BLE001 - re-raised via errors below + errors.append(err) + + handler = _SetEventOnWaitLog(other_caller_waiting) + git_logger = logging.getLogger("esphome.git") + git_logger.addHandler(handler) + try: + with caplog.at_level(logging.INFO): + threads = [threading.Thread(target=call) for _ in range(2)] + for t in threads: + t.start() + for t in threads: + t.join() + finally: + git_logger.removeHandler(handler) + + assert not errors + assert results == [repo_dir, repo_dir] + clone_calls = [ + c + for c in mock_run_git_command.call_args_list + if _get_git_command_type(c[0][0]) == "clone" + ] + # The second caller waited for the lock, then saw the completed clone. + assert len(clone_calls) == 1 + assert _marker_path(repo_dir).is_file() + + +def test_clone_or_update_creates_lock_file_next_to_hash_dir( + tmp_path: Path, mock_run_git_command: Mock +) -> None: + """The lock file lives beside the hash dir, never inside it. + + Checked while the clone runs (the lock is held): filelock's Windows + backend deletes the lock file on release, so probing after the call + would only work on Unix. + """ + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + lock_path = _lock_path(url, None, domain) + lock_held_during_clone: list[bool] = [] + + mock_run_git_command.side_effect = _make_clone_side_effect( + repo_dir, on_clone=lambda: lock_held_during_clone.append(lock_path.is_file()) + ) + + result_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain + ) + + assert result_dir == repo_dir + assert lock_path.parent == repo_dir.parent + assert lock_held_during_clone == [True] + + +def test_clone_or_update_subpath_locks_at_hash_dir_level( + tmp_path: Path, mock_run_git_command: Mock +) -> None: + """With a subpath the lock still guards the whole hash dir cache entry.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + hash_dir = _compute_repo_dir(url, None, domain) + repo_dir = hash_dir / "lib" + lock_path = _lock_path(url, None, domain) + lock_held_during_clone: list[bool] = [] + + mock_run_git_command.side_effect = _make_clone_side_effect( + repo_dir, on_clone=lambda: lock_held_during_clone.append(lock_path.is_file()) + ) + + result_dir, _ = git.clone_or_update( + url=url, + ref=None, + refresh=git.NEVER_REFRESH, + domain=domain, + subpath=Path("lib"), + ) + + assert result_dir == repo_dir + assert lock_path == hash_dir.parent / f"{hash_dir.name}.lock" + assert lock_held_during_clone == [True] + + +def test_clone_or_update_recovery_holds_lock_without_deadlock( + tmp_path: Path, mock_run_git_command: Mock +) -> None: + """Recovery re-clones while holding the lock and must not re-acquire it. + + A naive re-acquisition would deadlock here, since OS file locks taken on + separate descriptors conflict even within one process. The lock file + itself must survive the recovery rmtree of the broken repo dir: the + re-clone runs after the rmtree, so probing the lock file there proves + it. Probing after the call would only work on Unix, since filelock's + Windows backend deletes the lock file on release. + """ + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + lock_path = _lock_path(url, None, domain) + lock_held_during_reclone: list[bool] = [] + + _setup_old_repo(repo_dir) + + def git_command_side_effect( + cmd: list[str], cwd: str | None = None, **kwargs: Any + ) -> str: + if _get_git_command_type(cmd) == "stash": + raise git.GitCommandError("broken repository") + if _get_git_command_type(cmd) == "clone": + lock_held_during_reclone.append(lock_path.is_file()) + _simulate_cloned_repo(repo_dir) + return "" + + mock_run_git_command.side_effect = git_command_side_effect + + recovered_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + + assert recovered_dir == repo_dir + assert lock_held_during_reclone == [True] + + +def _hold_lock_in_thread(lock_path: Path) -> tuple[threading.Thread, threading.Event]: + """Hold the lock from another thread; returns the thread and its release event. + + OS file locks taken on separate descriptors conflict even within one + process, so a second FileLock instance in a thread contends the same + way another process would. + """ + held = threading.Event() + release = threading.Event() + + def hold() -> None: + with FileLock(str(lock_path)): + held.set() + release.wait(timeout=30) + + holder = threading.Thread(target=hold) + holder.start() + assert held.wait(timeout=30) + return holder, release + + +def test_clone_or_update_logs_wait_on_contended_lock( + tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """A contended acquire logs a redacted waiting message instead of + silently blocking.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://user:hunter2@github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir) + + holder, release = _hold_lock_in_thread(_lock_path(url, None, domain)) + + # Free the lock only once the waiting message has been emitted, so the + # release is caused by the thing being asserted instead of racing it. + handler = _SetEventOnWaitLog(release) + git_logger = logging.getLogger("esphome.git") + git_logger.addHandler(handler) + try: + with caplog.at_level(logging.INFO): + result_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain + ) + finally: + git_logger.removeHandler(handler) + release.set() + holder.join() + + assert result_dir == repo_dir + waiting = [ + r.getMessage() + for r in caplog.records + if "Waiting for another process" in r.getMessage() + ] + assert len(waiting) == 1 + assert "hunter2" not in waiting[0] + assert "://***@" in waiting[0] + + +def test_revert_skips_on_contended_lock( + tmp_path: Path, + mock_run_git_command: Mock, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """revert() only runs on an already-failing path; when it cannot get the + lock within the bounded timeout it warns and skips instead of hanging.""" + CORE.config_path = tmp_path / "test.yaml" + monkeypatch.setattr(git, "_REVERT_LOCK_TIMEOUT_SECONDS", 0.05) + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + _setup_old_repo(repo_dir) + + # A bare return value satisfies the rev-parse; the other commands' + # outputs are unused. + mock_run_git_command.return_value = "old_sha" + + _, revert = git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + assert revert is not None + + holder, release = _hold_lock_in_thread(_lock_path(url, None, domain)) + calls_before = len(mock_run_git_command.call_args_list) + with caplog.at_level(logging.INFO): + assert revert() is False + release.set() + holder.join() + + # No git reset was issued; the wait and the skip were both logged. + assert len(mock_run_git_command.call_args_list) == calls_before + assert any("Waiting for another process" in r.getMessage() for r in caplog.records) + assert any("skipping revert" in r.getMessage() for r in caplog.records) + + +def test_update_clears_marker_while_rewriting( + tmp_path: Path, mock_run_git_command: Mock +) -> None: + """The completion marker is absent while a refresh rewrites the entry + and restored once the rewrite finishes, so a timed-out peer's fallback + never trusts a mid-rewrite tree and a crashed update re-clones.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + _setup_old_repo(repo_dir) + + marker_during_rewrite: list[bool] = [] + + def git_command_side_effect( + cmd: list[str], cwd: str | None = None, **kwargs: Any + ) -> str: + if _get_git_command_type(cmd) in ("stash", "fetch", "reset"): + marker_during_rewrite.append(_marker_path(repo_dir).is_file()) + return "" + + mock_run_git_command.side_effect = git_command_side_effect + + git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + + assert marker_during_rewrite == [False, False, False] + assert _marker_path(repo_dir).is_file() + + +def test_revert_skips_when_checkout_moved( + tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """revert() only undoes this process's own update; when another process + refreshed the entry in the meantime it skips instead of rolling the + peer's newer checkout backwards.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + _setup_old_repo(repo_dir) + + # Pre-update SHA, post-update SHA, then a peer's SHA at revert time. + shas = iter(["old_sha", "new_sha", "peer_sha"]) + + def git_command_side_effect( + cmd: list[str], cwd: str | None = None, **kwargs: Any + ) -> str: + if _get_git_command_type(cmd) == "rev-parse": + return next(shas) + return "" + + mock_run_git_command.side_effect = git_command_side_effect + + _, revert = git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + assert revert is not None + + with caplog.at_level(logging.WARNING): + assert revert() is False + + resets = [ + c[0][0] + for c in mock_run_git_command.call_args_list + if _get_git_command_type(c[0][0]) == "reset" and c[0][0][-1] == "old_sha" + ] + assert resets == [] + assert any("checkout moved" in r.getMessage() for r in caplog.records) + + +def test_revert_returns_false_when_reset_fails( + tmp_path: Path, mock_run_git_command: Mock, caplog: pytest.LogCaptureFixture +) -> None: + """A failed git reset inside revert() is reported through the bool + contract instead of raising a cv.Invalid that would replace the + caller's original error.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + _setup_old_repo(repo_dir) + + def git_command_side_effect( + cmd: list[str], cwd: str | None = None, **kwargs: Any + ) -> str: + if _get_git_command_type(cmd) == "rev-parse": + return "old_sha" + # Only revert's reset targets the recorded SHA; the update path's + # reset targets FETCH_HEAD and must succeed. + if cmd[-1] == "old_sha": + raise git.GitCommandError("object not found") + return "" + + mock_run_git_command.side_effect = git_command_side_effect + + _, revert = git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + assert revert is not None + + with caplog.at_level(logging.WARNING): + assert revert() is False + + assert any("Could not revert" in r.getMessage() for r in caplog.records) + # The entry cannot be trusted after a failed reset; the dropped marker + # forces a re-clone on the next use. + assert not _marker_path(repo_dir).is_file() + + +def _raise_oserror_on_acquire( + monkeypatch: pytest.MonkeyPatch, code: int = errno.ENOLCK +) -> None: + """Make every FileLock acquire fail with the given errno.""" + + def broken_acquire(self: FileLock, *args: Any, **kwargs: Any) -> None: + raise OSError(code, os.strerror(code)) + + monkeypatch.setattr(FileLock, "acquire", broken_acquire) + + +@pytest.mark.parametrize( + ("code", "expected_fragment"), + [ + # Genuinely missing lock support is reported as such. + (errno.ENOLCK, "does not support locking"), + # A cache directory problem is not blamed on lock support; git + # reports the real error when it actually matters. + (errno.EROFS, "Could not take the cache entry lock"), + ], +) +def test_clone_or_update_continues_unlocked_when_filesystem_cannot_lock( + tmp_path: Path, + mock_run_git_command: Mock, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + code: int, + expected_fragment: str, +) -> None: + """A filesystem where taking the lock fails (e.g. NFS without a lock + daemon) degrades to the old unlocked behavior instead of failing.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir) + _raise_oserror_on_acquire(monkeypatch, code) + + with caplog.at_level(logging.WARNING): + result_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain + ) + + assert result_dir == repo_dir + assert _marker_path(repo_dir).is_file() + warnings = [ + r.getMessage() + for r in caplog.records + if "continuing without a lock" in r.getMessage() + ] + assert warnings + assert expected_fragment in warnings[0] + + +@pytest.mark.parametrize("broken_from_start", [True, False]) +def test_revert_continues_unlocked_when_filesystem_cannot_lock( + tmp_path: Path, + mock_run_git_command: Mock, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + broken_from_start: bool, +) -> None: + """A revert still resets when locking is unavailable, whether the wrapper + already fell back to unlocked (revert sees no lock at all) or the + filesystem stops locking between the update and the revert.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + _setup_old_repo(repo_dir) + + # A bare return value satisfies the rev-parse; the other commands' + # outputs are unused. + mock_run_git_command.return_value = "old_sha" + + if broken_from_start: + _raise_oserror_on_acquire(monkeypatch) + + with caplog.at_level(logging.WARNING): + _, revert = git.clone_or_update( + url=url, ref=None, refresh=TimePeriodSeconds(days=1), domain=domain + ) + assert revert is not None + if not broken_from_start: + _raise_oserror_on_acquire(monkeypatch) + # The reset ran (unlocked), so the revert reports success. + assert revert() is True + + assert mock_run_git_command.call_args_list[-1][0][0] == [ + "git", + "reset", + "--hard", + "old_sha", + ] + assert any("continuing without a lock" in r.getMessage() for r in caplog.records) + + +def _script_acquire_statuses( + monkeypatch: pytest.MonkeyPatch, statuses: list["git._LockStatus"] +) -> list[float]: + """Replace _acquire_repo_lock with a scripted sequence; returns the + timeouts it was called with.""" + timeouts: list[float] = [] + status_iter = iter(statuses) + + def fake_acquire( + lock: FileLock, safe_key: str, timeout: float, **kwargs: Any + ) -> "git._LockStatus": + timeouts.append(timeout) + return next(status_iter) + + monkeypatch.setattr(git, "_acquire_repo_lock", fake_acquire) + return timeouts + + +@pytest.mark.parametrize("subpath", [None, Path("lib")]) +def test_clone_or_update_uses_complete_entry_when_lock_wait_times_out( + tmp_path: Path, + mock_run_git_command: Mock, + caplog: pytest.LogCaptureFixture, + monkeypatch: pytest.MonkeyPatch, + subpath: Path | None, +) -> None: + """A bounded wait behind a stalled holder falls back to an existing + complete cache entry instead of hanging every peer.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + if subpath is not None: + repo_dir = repo_dir / subpath + _simulate_cloned_repo(repo_dir) + _mark_clone_complete(repo_dir) + + timeouts = _script_acquire_statuses(monkeypatch, [git._LockStatus.TIMEOUT]) + + with caplog.at_level(logging.WARNING): + result_dir, revert = git.clone_or_update( + url=url, + ref=None, + refresh=TimePeriodSeconds(days=1), + domain=domain, + subpath=subpath, + ) + + assert result_dir == repo_dir + assert revert is None + # Nothing was cloned or refreshed; the existing entry was used as-is. + assert mock_run_git_command.call_args_list == [] + assert timeouts == [git._COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS] + assert any( + "proceeding with the existing clone" in r.getMessage() for r in caplog.records + ) + + +def test_clone_or_update_waits_unbounded_without_complete_entry( + tmp_path: Path, + mock_run_git_command: Mock, + monkeypatch: pytest.MonkeyPatch, +) -> None: + """With no complete entry there is nothing to fall back to, so after the + bounded wait expires the caller keeps waiting for the holder's clone.""" + CORE.config_path = tmp_path / "test.yaml" + + url = "https://github.com/test/repo" + domain = "test" + repo_dir = _compute_repo_dir(url, None, domain) + mock_run_git_command.side_effect = _make_clone_side_effect(repo_dir) + + timeouts = _script_acquire_statuses( + monkeypatch, [git._LockStatus.TIMEOUT, git._LockStatus.ACQUIRED] + ) + + result_dir, _ = git.clone_or_update( + url=url, ref=None, refresh=git.NEVER_REFRESH, domain=domain + ) + + assert result_dir == repo_dir + assert timeouts == [git._COMPLETE_ENTRY_LOCK_TIMEOUT_SECONDS, -1] + # The clone proceeded normally once the lock was finally acquired. + assert _marker_path(repo_dir).is_file() + + def _real_git(*args: str, cwd: Path) -> None: """Run real git to build a test fixture repository.""" subprocess.run( @@ -1674,7 +2268,8 @@ def test_refresh_picks_up_new_remote_commits( # Verify the refresh sequence: rev-parse -> stash -> fetch (depth=1) -> reset call_list = mock_run_git_command.call_args_list cmd_sequence = [_get_git_command_type(c[0][0]) for c in call_list] - assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset"] + # The trailing rev-parse records the post-update SHA for revert(). + assert cmd_sequence == ["rev-parse", "stash", "fetch", "reset", "rev-parse"] fetch_cmd = call_list[2][0][0] assert "--depth=1" in fetch_cmd @@ -1685,7 +2280,7 @@ def test_refresh_picks_up_new_remote_commits( # revert callback should reset back to the recorded pre-update SHA. assert revert is not None - revert() + assert revert() is True assert mock_run_git_command.call_args_list[-1][0][0] == [ "git", "reset",