From bb9b4c2b6037f0dea9bb8c4d908031e943242fba Mon Sep 17 00:00:00 2001 From: Gavin Mak Date: Tue, 10 Nov 2026 00:32:51 +0000 Subject: [PATCH] git_refs, project: read files-backend refs in Python GitRefs._LoadAll spawns `git for-each-ref` to load HEAD and refs/*, and Project._InitAnyMRef leaves warm snapshots stale after direct UpdateRef writes. On virtualized checkouts with thousands of repositories, these spawns add avoidable process overhead. Read HEAD, packed-refs, and loose refs directly in GitRefs when the repository uses a plain files ref backend, recording (mode, mtime, size, inode) stat tuples during the single directory pass and rechecking packed-refs afterward to guard against concurrent pack-refs updates. The tuples leave out ctime, which backup agents and file system crawlers move without changing a ref. Fall back to `git for-each-ref` when GIT_DIR, GIT_COMMON_DIR, reftable, commondir, custom refStorage config, symlinks, or lockfiles are present. Also add GitRefs.TryResolveRef for single-ref lookups without scanning all of refs/, GitRefs.Invalidate, and GitRefs.UpdateRef to keep cached snapshots warm after _InitAnyMRef writes a direct ref. Bug: 543101585 Change-Id: I804ac45712208e9f9e6437012661bc1de58b9825 Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/655963 Tested-by: Gavin Mak Reviewed-by: Brian Gan Commit-Queue: Gavin Mak --- git_refs.py | 459 +++++++++++++++++++++++++++++++++++++---- project.py | 4 + tests/test_git_refs.py | 334 ++++++++++++++++++++++++++++++ 3 files changed, 758 insertions(+), 39 deletions(-) diff --git a/git_refs.py b/git_refs.py index 70b20139d..4ce6713c1 100644 --- a/git_refs.py +++ b/git_refs.py @@ -13,6 +13,9 @@ # limitations under the License. import os +import re +import stat +from typing import Dict, Optional, Set, Tuple import git_command import platform_utils @@ -29,6 +32,26 @@ R_WORKTREE_M = R_WORKTREE + "m/" R_M = "refs/remotes/m/" # Maximum symbolic-ref indirection depth (matches Git SYMREF_MAXDEPTH). _SYMREF_MAXDEPTH = 5 +_HEX_OID_RE = re.compile(r"^(?:[0-9a-f]{40}|[0-9a-f]{64})$") +_BAD_REF_CHARS = frozenset(" ~^:?*[\\") +_StatKey = Tuple[int, int, int, int] + + +def _IsValidRefName(name: str) -> bool: + """Validate a Git refname per git-check-ref-format rules.""" + if not name or name.endswith(".") or ".." in name or "@{" in name: + return False + if any( + ch in _BAD_REF_CHARS or ord(ch) < 0x20 or ord(ch) == 0x7F for ch in name + ): + return False + parts = name.split("/") + if len(parts) < 2: + return False + for part in parts: + if not part or part.startswith(".") or part.endswith(".lock"): + return False + return True class GitRefs: @@ -36,7 +59,8 @@ class GitRefs: self._gitdir = gitdir self._phyref = None self._symref = None - self._mtime = {} + self._mtime: Dict[str, Optional[_StatKey]] = {} + self._files_authoritative = False @property def all(self): @@ -54,6 +78,18 @@ class GitRefs: """Whether a ref snapshot has already been loaded.""" return self._phyref is not None + @property + def is_files_authoritative(self) -> bool: + """Whether the loaded snapshot came from a complete files scan.""" + return self._phyref is not None and self._files_authoritative + + def Invalidate(self) -> None: + """Discard any cached ref snapshot so the next access reloads.""" + self._phyref = None + self._symref = None + self._mtime = {} + self._files_authoritative = False + def get(self, name): try: return self.all[name] @@ -70,6 +106,9 @@ class GitRefs: if name in self._mtime: del self._mtime[name] + norm_os = os.path.join(*name.strip("/").split("/")) + if norm_os in self._mtime: + del self._mtime[norm_os] def symref(self, name): try: @@ -78,6 +117,80 @@ class GitRefs: except KeyError: return "" + def TryResolveRef(self, name: str) -> Optional[str]: + """Resolve a single ref directly on the files backend without git. + + Args: + name: Reference name ("HEAD" or "refs/..."). + + Returns: + The resolved hex object ID if present, "" if the repository uses a + clean files backend and the ref is authoritatively absent or + unborn, or None if git must be invoked as a fallback. + """ + if name != HEAD and ( + not name.startswith("refs/") or not _IsValidRefName(name) + ): + return None + + if self._phyref is not None and not self._NeedUpdate(): + if name in self._phyref: + return self._phyref[name] + if self._files_authoritative: + return "" + return None + + if self._CheckFilesBackendConfig() is None: + return None + refs_dir = os.path.join(self._gitdir, "refs") + if platform_utils.islink(refs_dir) or not os.path.isdir(refs_dir): + return None + head_file = os.path.join(self._gitdir, HEAD) + if platform_utils.islink(head_file) or not os.path.isfile(head_file): + return None + + cur = name + for _ in range(_SYMREF_MAXDEPTH + 1): + parts = cur.split("/") + for idx in range(1, len(parts) + 1): + if platform_utils.islink( + os.path.join(self._gitdir, *parts[:idx]) + ): + return None + ref_path = os.path.join(self._gitdir, *parts) + if os.path.lexists(ref_path + ".lock"): + return None + + if os.path.lexists(ref_path): + if os.path.isdir(ref_path): + return None + parsed = self._ParseLooseRefFile(ref_path) + if parsed is None: + return None + is_sym, val, _ = parsed + if not is_sym: + return val + cur = val + continue + + if cur == HEAD: + return None + + packed_path = os.path.join(self._gitdir, "packed-refs") + if os.path.lexists(packed_path): + packed_res = self._ReadPackedRefs(packed_path) + if packed_res is None: + return None + packed_map, _ = packed_res + else: + packed_map = {} + + if os.path.lexists(ref_path): + return None + return packed_map.get(cur, "") + + return "" + def UpdateSymref(self, name: str, target: str) -> None: """Update a symbolic reference in an already-loaded ref snapshot. @@ -88,7 +201,7 @@ class GitRefs: if self._phyref is None: return if self._HasExternalRefChanges(name): - self._phyref = None + self.Invalidate() return self._symref[name] = target @@ -100,38 +213,70 @@ class GitRefs: self._TrackMtime(anc) self._TrackMtime(os.path.join(*parts)) - def _HasExternalRefChanges(self, updated_symref: str) -> bool: - """Check whether refs outside |updated_symref| changed on disk. + def UpdateRef(self, name: str) -> None: + """Refresh a direct reference in an already-loaded ref snapshot. Args: - updated_symref: Symbolic ref path just updated on disk. + name: Full reference name just updated on disk (e.g. + "refs/remotes/m/main"). + """ + if self._phyref is None: + return + if self._HasExternalRefChanges(name): + self.Invalidate() + return + + parts = name.strip("/").split("/") + ref_rel = os.path.join(*parts) + parsed = self._ParseLooseRefFile(os.path.join(self._gitdir, ref_rel)) + if parsed is None or parsed[0]: + self.Invalidate() + return + + _, oid, ref_stat = parsed + self._symref.pop(name, None) + self._phyref[name] = oid + self._ResolveSymrefs() + + for idx in range(1, len(parts)): + self._TrackMtime(os.path.join(*parts[:idx])) + self._mtime[ref_rel] = ref_stat + + def _HasExternalRefChanges(self, updated_ref: str) -> bool: + """Check whether refs outside |updated_ref| changed on disk. + + Args: + updated_ref: Reference path just updated on disk. Returns: - True if reftable is in use or any unrelated ref path changed. + True if files backend is unsafe or any unrelated ref path changed. """ - if os.path.exists(os.path.join(self._gitdir, "reftable")): + if self._CheckFilesBackendConfig() is None: return True - if "packed-refs" not in self._mtime and os.path.exists( + if "packed-refs" not in self._mtime and os.path.lexists( os.path.join(self._gitdir, "packed-refs") ): return True - parts = updated_symref.strip("/").split("/") + parts = updated_ref.strip("/").split("/") expected_child = { "/".join(parts[:idx]): parts[idx] for idx in range(1, len(parts)) } - for tracked, mtime in self._mtime.items(): + for tracked, prev_stat in self._mtime.items(): norm = tracked.replace(os.sep, "/") - if norm == updated_symref: + if norm == updated_ref: continue full_path = os.path.join(self._gitdir, tracked) - try: - curr_mtime = os.path.getmtime(full_path) - except OSError: + curr_stat = self._StatPath(full_path) + if prev_stat is None: + if curr_stat is not None: + return True + continue + if curr_stat is None: return True if norm in expected_child: - if curr_mtime != mtime: + if curr_stat != prev_stat: try: children = platform_utils.listdir(full_path) except OSError: @@ -140,14 +285,15 @@ class GitRefs: for child in children: if child == next_comp: continue - if os.path.join(tracked, child) not in self._mtime: + child_key = os.path.join(tracked, child) + if self._mtime.get(child_key) is None: return True - elif curr_mtime != mtime: + elif curr_stat != prev_stat: return True for anc, next_comp in expected_child.items(): anc_os = anc.replace("/", os.sep) - if anc_os not in self._mtime: + if self._mtime.get(anc_os) is None: anc_full = os.path.join(self._gitdir, anc_os) try: if platform_utils.listdir(anc_full) != [next_comp]: @@ -177,20 +323,23 @@ class GitRefs: if self._phyref is None or self._NeedUpdate(): self._LoadAll() - def _NeedUpdate(self): + def _NeedUpdate(self) -> bool: with repo_trace.Trace(": scan refs %s", self._gitdir): - for name, mtime in self._mtime.items(): - try: - if mtime != os.path.getmtime( - os.path.join(self._gitdir, name) - ): - return True - except OSError: + for name, prev_stat in self._mtime.items(): + if ( + self._StatPath(os.path.join(self._gitdir, name)) + != prev_stat + ): return True return False def _LoadAll(self): with repo_trace.Trace(": load refs %s", self._gitdir): + if self._TryReadFilesBackend(): + self._ResolveSymrefs() + return + + self._files_authoritative = False self._phyref = {} self._symref = {} self._mtime = {} @@ -205,19 +354,238 @@ class GitRefs: self._ResolveSymrefs() self._RefreshMtimes() + def _CheckFilesBackendConfig(self) -> Optional[_StatKey]: + """Verify standard files ref backend preconditions and stat config.""" + if "GIT_DIR" in os.environ or "GIT_COMMON_DIR" in os.environ: + return None + if not os.path.isdir(self._gitdir): + return None + for special in ("reftable", "commondir"): + if os.path.lexists(os.path.join(self._gitdir, special)): + return None + + cfg_path = os.path.join(self._gitdir, "config") + if platform_utils.islink(cfg_path): + return None + try: + with open(cfg_path, "rb") as cfg_fd: + if b"refstorage" in cfg_fd.read().lower(): + return None + return self._FormatStat(os.fstat(cfg_fd.fileno())) + except OSError: + return None + + @classmethod + def _ParseLooseRefFile( + cls, path: str + ) -> Optional[Tuple[bool, str, _StatKey]]: + """Parse a loose ref or HEAD file into (is_symref, val, stat).""" + if platform_utils.islink(path): + return None + try: + with open(path, "r", encoding="utf-8") as fd: + raw = fd.read(1024) + st = cls._FormatStat(os.fstat(fd.fileno())) + except (OSError, UnicodeDecodeError): + return None + + line = raw.rstrip("\r\n") + if not line or "\n" in line or "\r" in line: + return None + if line.startswith("ref:"): + target = line[4:].strip() + if not target or target == R_HEADS + ".invalid": + return None + if target != HEAD and ( + not target.startswith("refs/") or not _IsValidRefName(target) + ): + return None + return (True, target, st) + + oid = line.strip() + if not _HEX_OID_RE.match(oid) or cls._IsNullRef(oid): + return None + return (False, oid, st) + + @classmethod + def _ReadPackedRefs( + cls, packed_path: str + ) -> Optional[Tuple[Dict[str, str], _StatKey]]: + """Parse a packed-refs file into (ref_map, stat).""" + if platform_utils.islink(packed_path): + return None + packed: Dict[str, str] = {} + try: + with open(packed_path, "r", encoding="utf-8") as packed_fd: + for raw_line in packed_fd: + if raw_line.startswith("#"): + continue + line = raw_line.rstrip("\r\n") + if line.startswith("^"): + if not _HEX_OID_RE.match(line[1:]): + return None + continue + parts = line.split(" ", 1) + if len(parts) != 2: + return None + oid, ref_name = parts + if ( + not _HEX_OID_RE.match(oid) + or cls._IsNullRef(oid) + or not ref_name.startswith("refs/") + or not _IsValidRefName(ref_name) + ): + return None + packed[ref_name] = oid + st = cls._FormatStat(os.fstat(packed_fd.fileno())) + except (OSError, UnicodeDecodeError): + return None + return (packed, st) + + def _TryReadFilesBackend(self) -> bool: + """Load HEAD, packed-refs, and loose refs directly when safe. + + Returns: + True if the repository uses a standard files ref backend and all + refs and stat fingerprints were read cleanly without invoking git. + """ + cfg_stat = self._CheckFilesBackendConfig() + if cfg_stat is None: + return False + + mtime: Dict[str, Optional[_StatKey]] = { + "config": cfg_stat, + "reftable": None, + } + phyref: Dict[str, str] = {} + symref: Dict[str, str] = {} + + parsed_head = self._ParseLooseRefFile(os.path.join(self._gitdir, HEAD)) + if parsed_head is None: + return False + head_is_sym, head_val, head_stat = parsed_head + mtime[HEAD] = head_stat + if head_is_sym: + symref[HEAD] = head_val + else: + phyref[HEAD] = head_val + + packed_path = os.path.join(self._gitdir, "packed-refs") + packed_stat: Optional[_StatKey] = None + if os.path.lexists(packed_path): + packed_res = self._ReadPackedRefs(packed_path) + if packed_res is None: + return False + packed_map, packed_stat = packed_res + phyref.update(packed_map) + mtime["packed-refs"] = packed_stat + + refs_dir = os.path.join(self._gitdir, "refs") + if platform_utils.islink(refs_dir): + return False + refs_stat = self._StatPath(refs_dir) + if refs_stat is None or not stat.S_ISDIR(refs_stat[0]): + return False + mtime["refs"] = refs_stat + + try: + to_scan = [("refs", "refs", refs_dir)] + while to_scan: + ref_prefix, rel_os_prefix, abs_dir = to_scan.pop() + with os.scandir(abs_dir) as entries: + for entry in entries: + if entry.is_symlink(): + return False + ref_name = f"{ref_prefix}/{entry.name}" + if not _IsValidRefName(ref_name): + return False + rel_os = os.path.join(rel_os_prefix, entry.name) + if entry.is_dir(follow_symlinks=False): + if ref_name in phyref: + return False + # DirEntry.stat() leaves st_ino == 0 on Windows, + # whereas _StatPath() uses os.lstat() and populates + # st_ino. Stat via _StatPath() so _NeedUpdate() and + # _HasExternalRefChanges() compare matching keys. + d_stat = self._StatPath(entry.path) + if d_stat is None or not stat.S_ISDIR(d_stat[0]): + return False + mtime[rel_os] = d_stat + to_scan.append((ref_name, rel_os, entry.path)) + elif entry.is_file(follow_symlinks=False): + parsed = self._ParseLooseRefFile(entry.path) + if parsed is None: + return False + is_sym, val, f_stat = parsed + mtime[rel_os] = f_stat + if is_sym: + phyref.pop(ref_name, None) + symref[ref_name] = val + else: + symref.pop(ref_name, None) + phyref[ref_name] = val + else: + return False + except OSError: + return False + + # Guard against a concurrent `git pack-refs --prune` writing + # packed-refs and unlinking loose refs mid-scan. + if self._StatPath(packed_path) != packed_stat: + return False + + self._phyref = phyref + self._symref = symref + self._mtime = mtime + self._files_authoritative = True + return True + def _RefreshMtimes(self) -> None: """Record current modification times for ref storage paths.""" self._mtime = {} - self._TrackMtime(HEAD) - self._TrackMtime("config") - self._TrackMtime("packed-refs") + self._TrackMtime(HEAD, track_missing=True) + self._TrackMtime("config", track_missing=True) + self._TrackMtime("packed-refs", track_missing=True) + self._TrackMtime("refs", track_missing=True) + self._TrackMtime("reftable", track_missing=True) self._TrackTreeMtimes("refs") self._TrackTreeMtimes("reftable") + @staticmethod + def _FormatStat(st: os.stat_result) -> _StatKey: + # ctime is left out: tools that only touch metadata, such as backup + # agents and file system crawlers, move it without changing a ref. + return (st.st_mode, st.st_mtime_ns, st.st_size, st.st_ino) + + @classmethod + def _StatPath(cls, path: str) -> Optional[_StatKey]: + try: + st = os.lstat(path) + except OSError: + return None + mount_tag = getattr(stat, "IO_REPARSE_TAG_MOUNT_POINT", -1) + is_reparse_dir = getattr(st, "st_reparse_tag", 0) == mount_tag + if stat.S_ISLNK(st.st_mode) or is_reparse_dir: + # Track the symlink target so mutations to symlinked refs or + # packed-refs invalidate the fallback snapshot, while keeping + # S_IFLNK in mode and negative size when dangling. + link_mode = stat.S_IFLNK | stat.S_IMODE(st.st_mode) + try: + target_st = os.stat(path) + except OSError: + return (link_mode, st.st_mtime_ns, -1 - st.st_size, st.st_ino) + return ( + link_mode, + target_st.st_mtime_ns, + target_st.st_size, + target_st.st_ino, + ) + return cls._FormatStat(st) + @staticmethod def _IsNullRef(ref_id: str) -> bool: """Check if a ref_id is a null object ID.""" - return ref_id and all(ch == "0" for ch in ref_id) + return bool(ref_id) and all(ch == "0" for ch in ref_id) def _ReadRefs(self) -> bool: """Read all references using git for-each-ref. @@ -299,32 +667,45 @@ class GitRefs: if ref_id: self._phyref[name] = ref_id - def _TrackMtime(self, name: str) -> None: - """Track the modification time of a specific gitdir path.""" - path = os.path.join(self._gitdir, name) - try: - self._mtime[name] = os.path.getmtime(path) - except OSError: - return + def _TrackMtime(self, name: str, track_missing: bool = False) -> None: + """Track the stat fingerprint of a specific gitdir path.""" + st = self._StatPath(os.path.join(self._gitdir, name)) + if st is not None or track_missing: + self._mtime[name] = st def _TrackTreeMtimes(self, root: str) -> None: """Recursively track modification times for a directory tree.""" root_path = os.path.join(self._gitdir, root) try: if not platform_utils.isdir(root_path): + self._TrackMtime(root) return except OSError: return + visited: Set[Tuple[int, int]] = set() to_scan = [root] while to_scan: name = to_scan.pop() self._TrackMtime(name) path = os.path.join(self._gitdir, name) - if not platform_utils.isdir(path): + try: + dir_st = os.stat(path) + except OSError: continue + if not stat.S_ISDIR(dir_st.st_mode): + continue + if dir_st.st_ino: + dir_key = (dir_st.st_dev, dir_st.st_ino) + if dir_key in visited: + continue + visited.add(dir_key) - for child in platform_utils.listdir(path): + try: + children = platform_utils.listdir(path) + except OSError: + continue + for child in children: child_name = os.path.join(name, child) child_path = os.path.join(self._gitdir, child_name) if platform_utils.isdir(child_path): diff --git a/project.py b/project.py index a9033adff..53aa5c39d 100644 --- a/project.py +++ b/project.py @@ -4664,6 +4664,8 @@ class Project: msg = "manifest set to %s" % self.revisionId dst = self.revisionId + "^0" active_git.UpdateRef(ref, dst, message=msg, detach=True) + if active_git is self.bare_git: + self.bare_ref.UpdateRef(ref) else: remote = self.GetRemote() dst = remote.ToLocal(self.revisionExpr) @@ -4671,6 +4673,8 @@ class Project: msg = "manifest set to %s" % self.revisionExpr if detach: active_git.UpdateRef(ref, dst, message=msg, detach=True) + if active_git is self.bare_git: + self.bare_ref.UpdateRef(ref) else: active_git.symbolic_ref("-m", msg, ref, dst) if active_git is self.bare_git: diff --git a/tests/test_git_refs.py b/tests/test_git_refs.py index 965556bd0..345b17025 100644 --- a/tests/test_git_refs.py +++ b/tests/test_git_refs.py @@ -16,7 +16,9 @@ import os from pathlib import Path +import stat import subprocess +import time from typing import Any, List from unittest import mock @@ -25,6 +27,7 @@ import utils_for_test import git_command import git_refs +import platform_utils def _run(repo: str, *args: str) -> str: @@ -406,3 +409,334 @@ def test_update_symref_propagates_chained_symrefs(tmp_path: Path) -> None: assert refs.get("refs/remotes/m/main") == c2 assert refs.get("refs/remotes/m/alias") == c2 assert refs.get("HEAD") == c2 + + +def test_files_backend_reads_loose_and_packed_without_git( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Files backend reads packed-refs, loose refs, and symrefs in Python.""" + repo = _init_repo(tmp_path, reftable=False) + branch = _run(repo, "symbolic-ref", "--short", "HEAD") + c1 = _run(repo, "rev-parse", "HEAD") + _run(repo, "tag", "-a", "-m", "v1", "v1.0") + tag_oid = _run(repo, "rev-parse", "refs/tags/v1.0") + _run(repo, "branch", "packed-only", c1) + _run(repo, "branch", "overridden", c1) + _run(repo, "pack-refs", "--all") + + Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8") + _run(repo, "add", "a") + _run(repo, "commit", "-q", "-m", "second") + c2 = _run(repo, "rev-parse", "HEAD") + _run(repo, "update-ref", "refs/heads/overridden", c2) + _run(repo, "symbolic-ref", "refs/remotes/m/main", "refs/heads/overridden") + + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + seen_cmds: List[List[str]] = [] + + class SpyGitCommand(git_command.GitCommand): + def __init__( + self, + proj: Any, + cmdv: List[str], + *args: Any, + **kwargs: Any, + ) -> None: + seen_cmds.append(list(cmdv)) + super().__init__(proj, cmdv, *args, **kwargs) + + monkeypatch.setattr(git_command, "GitCommand", SpyGitCommand) + refs = git_refs.GitRefs(os.path.join(repo, ".git")) + assert refs.get("HEAD") == c2 + assert refs.is_files_authoritative + with mock.patch.object( + refs, + "_LoadAll", + side_effect=AssertionError("expected warm GitRefs snapshot"), + ): + assert refs.get(f"refs/heads/{branch}") == c2 + assert refs.get("refs/heads/packed-only") == c1 + assert refs.get("refs/heads/overridden") == c2 + assert refs.get("refs/remotes/m/main") == c2 + assert refs.get("refs/tags/v1.0") == tag_oid + assert seen_cmds == [] + + # A concurrent .lock file triggers clean fallback to git for-each-ref. + lock_file = Path(repo, ".git", "refs", "heads", "packed-only.lock") + lock_file.write_text("", encoding="utf-8") + refs2 = git_refs.GitRefs(os.path.join(repo, ".git")) + assert refs2.get("refs/heads/packed-only") == c1 + assert not refs2.is_files_authoritative + assert any(cmd and cmd[0] == "for-each-ref" for cmd in seen_cmds) + + +def test_try_resolve_ref_loose_packed_symref_and_missing( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """TryResolveRef resolves loose/packed/symref refs without full scan.""" + repo = _init_repo(tmp_path, reftable=False) + c1 = _run(repo, "rev-parse", "HEAD") + _run(repo, "branch", "packed-branch", c1) + _run(repo, "pack-refs", "--all") + + Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8") + _run(repo, "add", "a") + _run(repo, "commit", "-q", "-m", "second") + c2 = _run(repo, "rev-parse", "HEAD") + _run(repo, "update-ref", "refs/remotes/origin/main", c2) + _run( + repo, + "symbolic-ref", + "refs/remotes/m/main", + "refs/remotes/origin/main", + ) + + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + gitdir = os.path.join(repo, ".git") + refs = git_refs.GitRefs(gitdir) + + assert refs.TryResolveRef("refs/remotes/origin/main") == c2 + assert refs.TryResolveRef("refs/heads/packed-branch") == c1 + assert refs.TryResolveRef("refs/remotes/m/main") == c2 + assert refs.TryResolveRef("refs/remotes/origin/missing") == "" + assert not refs.is_loaded + + # A .lock file forces None so callers fall back to git. + lock_file = Path(gitdir, "refs", "remotes", "origin", "main.lock") + lock_file.write_text("", encoding="utf-8") + assert refs.TryResolveRef("refs/remotes/origin/main") is None + + +def test_detects_new_packed_refs_and_same_mtime_replace( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """GitRefs reloads on newly created packed-refs and same-mtime replaces.""" + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + + repo = os.path.join(tmp_path, "fresh") + utils_for_test.init_git_tree(repo, ref_format="files") + gitdir = os.path.join(repo, ".git") + refs = git_refs.GitRefs(gitdir) + assert refs.all == {} + + c1 = "1" * 40 + c2 = "2" * 40 + Path(gitdir, "packed-refs").write_text( + f"# pack-refs with: peeled fully-peeled sorted \n" + f"{c1} refs/remotes/origin/main\n", + encoding="utf-8", + ) + assert refs.get("refs/remotes/origin/main") == c1 + + loose = Path(gitdir, "refs", "remotes", "origin", "topic") + loose.parent.mkdir(parents=True, exist_ok=True) + loose.write_text(c1 + "\n", encoding="utf-8") + assert refs.get("refs/remotes/origin/topic") == c1 + + # Replace the ref the way git does, by renaming a new file over it, then + # put the mtimes back so that only the new inode shows. + file_st = os.stat(loose) + dir_st = os.stat(loose.parent) + replacement = Path(gitdir, "topic.new") + replacement.write_text(c2 + "\n", encoding="utf-8") + os.replace(replacement, loose) + os.utime(loose, ns=(file_st.st_atime_ns, file_st.st_mtime_ns)) + os.utime(loose.parent, ns=(dir_st.st_atime_ns, dir_st.st_mtime_ns)) + assert refs.get("refs/remotes/origin/topic") == c2 + + +@pytest.mark.skipif( + platform_utils.isWindows(), + reason="st_ctime is the creation time on Windows", +) +def test_ctime_only_change_keeps_snapshot_warm( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """A ref file whose ctime alone moves does not invalidate the snapshot.""" + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + repo = _init_repo(tmp_path, reftable=False) + branch = _run(repo, "symbolic-ref", "--short", "HEAD") + head = _run(repo, "rev-parse", "HEAD") + gitdir = os.path.join(repo, ".git") + refs = git_refs.GitRefs(gitdir) + assert refs.get(f"refs/heads/{branch}") == head + + # Backup agents and file system crawlers tag files with xattrs, which moves + # ctime and nothing else. Flipping a mode bit and putting it back does the + # same; repeat it until the timestamp granularity lets the change show. + loose = os.path.join(gitdir, "refs", "heads", branch) + before = os.lstat(loose) + mode = stat.S_IMODE(before.st_mode) + for _ in range(1000): + os.chmod(loose, mode ^ stat.S_IXUSR) + os.chmod(loose, mode) + if os.lstat(loose).st_ctime_ns != before.st_ctime_ns: + break + time.sleep(0.001) + else: + pytest.fail("chmod did not move ctime") + + with mock.patch.object( + refs, + "_LoadAll", + side_effect=AssertionError("expected warm GitRefs snapshot"), + ): + assert refs.get(f"refs/heads/{branch}") == head + + +@pytest.mark.parametrize("reftable", [False, True]) +def test_update_ref_keeps_snapshot_warm( + tmp_path: Path, reftable: bool, monkeypatch: pytest.MonkeyPatch +) -> None: + """UpdateRef stays warm on files backend and reloads on reftable.""" + if reftable and not utils_for_test.supports_reftable(): + pytest.skip("reftable not supported") + + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + repo = _init_repo(tmp_path, reftable=reftable) + branch = _run(repo, "symbolic-ref", "--short", "HEAD") + c1 = _run(repo, "rev-parse", "HEAD") + _run(repo, "symbolic-ref", "refs/remotes/m/main", f"refs/heads/{branch}") + + Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8") + _run(repo, "add", "a") + _run(repo, "commit", "-q", "-m", "second") + c2 = _run(repo, "rev-parse", "HEAD") + + refs = git_refs.GitRefs(os.path.join(repo, ".git")) + assert refs.symref("refs/remotes/m/main") == f"refs/heads/{branch}" + assert refs.get("refs/remotes/m/main") == c2 + + _run(repo, "update-ref", "--no-deref", "refs/remotes/m/main", c1) + refs.UpdateRef("refs/remotes/m/main") + + if reftable: + assert not refs.is_loaded + assert refs.symref("refs/remotes/m/main") == "" + assert refs.get("refs/remotes/m/main") == c1 + else: + with mock.patch.object( + refs, + "_LoadAll", + side_effect=AssertionError("expected warm GitRefs snapshot"), + ): + assert refs.symref("refs/remotes/m/main") == "" + assert refs.get("refs/remotes/m/main") == c1 + + # Setting GIT_COMMON_DIR invalidates instead of reading self._gitdir. + monkeypatch.setenv("GIT_COMMON_DIR", os.path.join(repo, ".git")) + refs.UpdateRef("refs/remotes/m/main") + assert not refs.is_loaded + + +def test_symlinked_ref_fallback_reloads_and_breaks_cycles( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + """Symlink fallback reloads on target changes and breaks dir cycles.""" + monkeypatch.delenv("GIT_DIR", raising=False) + monkeypatch.delenv("GIT_COMMON_DIR", raising=False) + repo = _init_repo(tmp_path, reftable=False) + c1 = _run(repo, "rev-parse", "HEAD") + + Path(os.path.join(repo, "a")).write_text("2", encoding="utf-8") + _run(repo, "add", "a") + _run(repo, "commit", "-q", "-m", "second") + c2 = _run(repo, "rev-parse", "HEAD") + + gitdir = os.path.join(repo, ".git") + ext_dir = tmp_path / "ext_refs" + ext_dir.mkdir() + target_ref = ext_dir / "linked" + target_ref.write_text(c1 + "\n", encoding="utf-8") + + link_ref = os.path.join(gitdir, "refs", "heads", "linked") + loop_dir = os.path.join(gitdir, "refs", "heads", "loop") + refs_root = os.path.join(gitdir, "refs") + used_junction = False + try: + os.symlink(str(target_ref), link_ref) + os.symlink(refs_root, loop_dir, target_is_directory=True) + tracked_ref = "refs/heads/linked" + except OSError: + if not platform_utils.isWindows(): + raise + used_junction = True + ext_link_dir = os.path.join(gitdir, "refs", "remotes", "ext") + os.makedirs(os.path.dirname(ext_link_dir), exist_ok=True) + subprocess.run( + ["cmd", "/c", "mklink", "/J", ext_link_dir, str(ext_dir)], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + subprocess.run( + ["cmd", "/c", "mklink", "/J", loop_dir, refs_root], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + tracked_ref = "refs/remotes/ext/linked" + + try: + # Remove loop_dir before running git for-each-ref, then verify cycle + # handling directly on _TrackTreeMtimes so git itself does not walk + # deep cyclic paths. + if used_junction: + os.rmdir(loop_dir) + else: + os.unlink(loop_dir) + + refs = git_refs.GitRefs(gitdir) + assert refs.get(tracked_ref) == c1 + assert not refs.is_files_authoritative + + replacement = ext_dir / "linked.new" + replacement.write_text(c2 + "\n", encoding="utf-8") + st = os.stat(target_ref) + os.replace(replacement, target_ref) + os.utime(target_ref, ns=(st.st_atime_ns, st.st_mtime_ns)) + assert refs.get(tracked_ref) == c2 + + if used_junction: + subprocess.run( + ["cmd", "/c", "mklink", "/J", loop_dir, refs_root], + check=True, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + ) + else: + os.symlink(refs_root, loop_dir, target_is_directory=True) + + refs._TrackTreeMtimes("refs") + assert os.path.join("refs", "heads", "loop") in refs._mtime + finally: + if used_junction: + for junc in ( + loop_dir, + os.path.join(gitdir, "refs", "remotes", "ext"), + ): + if os.path.lexists(junc): + os.rmdir(junc) + + # Also exercise S_ISLNK target and dangling-link tracking directly so + # unprivileged Windows runs cover POSIX symlink stat transitions too. + lnk = stat.S_IFLNK | 0o777 + reg = stat.S_IFREG | 0o644 + fake_link = os.stat_result((lnk, 10, 1, 1, 0, 0, 8, 1, 1, 1)) + fake_t1 = os.stat_result((reg, 20, 1, 1, 0, 0, 41, 1, 2, 1)) + fake_t2 = os.stat_result((reg, 21, 1, 1, 0, 0, 41, 1, 2, 1)) + with mock.patch("os.lstat", return_value=fake_link): + with mock.patch("os.stat", return_value=fake_t1): + s_t1 = git_refs.GitRefs._StatPath("sym") + with mock.patch("os.stat", return_value=fake_t2): + s_t2 = git_refs.GitRefs._StatPath("sym") + with mock.patch("os.stat", side_effect=FileNotFoundError): + s_dangling = git_refs.GitRefs._StatPath("sym") + assert s_t1 != s_t2 + assert s_t1 != s_dangling + assert s_dangling is not None and not stat.S_ISDIR(s_dangling[0])