project: don't rerun git status after a failed snapshot

When a status snapshot fails, fall back directly to _IsDirtyLegacy()
instead of IsDirty() to avoid running git status twice.

Expose Project.GetDirtyAndHead() so sync bloat checks can inspect
worktree state without invoking private Project methods.

Bug: 565047698
Change-Id: I6432ed82da88778eec9d151fd5b70bedb1efc62b
Reviewed-on: https://gerrit-review.googlesource.com/c/git-repo/+/634424
Reviewed-by: Brian Gan <brgan@google.com>
Commit-Queue: Gavin Mak <gavinmak@google.com>
Tested-by: Gavin Mak <gavinmak@google.com>

project.py

@@ -943,7 +943,26 @@ class Project:
             if has_status_stash:
                 return bool(status.stash_count)
             return self.HasStash()
-        return self.IsDirty(consider_untracked=True) or self.HasStash()
+        # The snapshot already failed; don't run git status again.
+        return self._IsDirtyLegacy(consider_untracked=True) or self.HasStash()
+
+    def GetDirtyAndHead(self) -> Tuple[bool, Optional[str]]:
+        """Return whether the worktree is dirty, and the commit at HEAD.
+
+        Untracked files count as dirty. HEAD is None when it can't be
+        resolved, e.g. on an unborn branch. Both come from one status
+        snapshot when possible.
+        """
+        status = self._GetStatusSnapshot(untracked_files="normal", branch=True)
+        if status is not None:
+            return status.is_dirty(consider_untracked=True), status.branch_oid
+        # The snapshot already failed; don't run git status again.
+        is_dirty = self._IsDirtyLegacy(consider_untracked=True)
+        try:
+            head = self.work_git.rev_parse(HEAD)
+        except GitError:
+            head = None
+        return is_dirty, head
 
     _userident_name = None
     _userident_email = None

subcmds/sync.py

@@ -1614,20 +1614,9 @@ later is required to fix a server side protocol bug.
         # Only check dirty or locally modified projects. These can't be
         # freshly cloned and will accumulate garbage.
         try:
-            status = project._GetStatusSnapshot(
-                untracked_files="normal", branch=True
-            )
-            if status is not None:
-                is_dirty = status.is_dirty(consider_untracked=True)
-                head_rev = status.branch_oid
-            else:
-                is_dirty = project.IsDirty(consider_untracked=True)
-                head_rev = project.work_git.rev_parse(HEAD)
-
+            is_dirty, head_rev = project.GetDirtyAndHead()
             if head_rev is None:
-                # Porcelain v2 reports an unborn branch as "(initial)".  The
-                # legacy rev-parse path failed here and skipped the bloat
-                # calculation, so preserve that behavior.
+                # An unborn branch has no HEAD to compare, so skip it.
                 return None
 
             manifest_rev = project.GetRevisionId(project.bare_ref.all)

tests/test_project.py

@@ -756,6 +756,64 @@ class ProjectTests(unittest.TestCase):
             )
             proj.HasStash.assert_called_once_with()
 
+    def test_dirty_and_head_agree_across_status_paths(self) -> None:
+        """Porcelain v2 and legacy plumbing report the same state."""
+        with utils_for_test.TempGitTree() as tempdir:
+            proj = _create_mock_project(tempdir)
+
+            def check(expected: Tuple[bool, Optional[str]]) -> None:
+                for use_status in (True, False):
+                    with self.subTest(expected=expected, status=use_status):
+                        with mock.patch.object(
+                            project, "git_require", return_value=use_status
+                        ):
+                            self.assertEqual(expected, proj.GetDirtyAndHead())
+
+            check((False, None))
+            Path(tempdir, "untracked").write_text("new")
+            check((True, None))
+
+            Path(tempdir, "tracked").write_text("initial")
+            proj.work_git.add("tracked")
+            proj.work_git.commit("-m", "initial")
+            head = proj.work_git.rev_parse("HEAD")
+            check((True, head))
+            os.remove(os.path.join(tempdir, "untracked"))
+            check((False, head))
+
+    def test_dirty_and_head_fallback_skips_second_snapshot(self) -> None:
+        """A failed snapshot goes straight to the legacy plumbing."""
+        with utils_for_test.TempGitTree() as tempdir:
+            proj = _create_mock_project(tempdir)
+            proj._GetStatusSnapshot = mock.MagicMock(return_value=None)
+            proj._IsDirtyLegacy = mock.MagicMock(return_value=False)
+            proj.work_git = mock.MagicMock()
+            proj.work_git.rev_parse.return_value = "head"
+
+            self.assertEqual((False, "head"), proj.GetDirtyAndHead())
+
+            proj._GetStatusSnapshot.assert_called_once_with(
+                untracked_files="normal", branch=True
+            )
+            proj._IsDirtyLegacy.assert_called_once_with(consider_untracked=True)
+            proj.work_git.rev_parse.assert_called_once_with("HEAD")
+
+    def test_dirty_or_stash_fallback_skips_second_snapshot(self) -> None:
+        """A failed snapshot goes straight to the legacy dirty check."""
+        with utils_for_test.TempGitTree() as tempdir:
+            proj = _create_mock_project(tempdir)
+            proj._GetStatusSnapshot = mock.MagicMock(return_value=None)
+            proj._IsDirtyLegacy = mock.MagicMock(return_value=False)
+            proj.HasStash = mock.MagicMock(return_value=False)
+
+            with mock.patch.object(project, "git_require", return_value=True):
+                self.assertFalse(proj._HasDirtyOrStash())
+
+            proj._GetStatusSnapshot.assert_called_once_with(
+                untracked_files="normal", show_stash=True
+            )
+            proj._IsDirtyLegacy.assert_called_once_with(consider_untracked=True)
+
     def test_old_git_dirty_check_uses_legacy_plumbing(self) -> None:
         """Git clients before 2.11 retain the existing dirty-check path."""
         with utils_for_test.TempGitTree() as tempdir:

tests/test_subcmds_sync.py

@@ -30,7 +30,6 @@ import pytest
 import command
 from error import GitError
 from error import RepoExitError
-import git_status
 import manifest_xml
 from project import SyncNetworkHalfResult
 from subcmds import sync
@@ -990,11 +989,9 @@ class CheckForBloatedProjects(unittest.TestCase):
         self.cmd.git_event_log = mock.MagicMock()
         self.cmd._bloated_projects = []
 
-    def test_one_project_reuses_status_head_oid(self) -> None:
-        """The bloat scan gets dirty state and HEAD from one snapshot."""
-        status = git_status.StatusSnapshot()
-        status.branch_oid = "local"
-        self.project._GetStatusSnapshot.return_value = status
+    def test_one_project_uses_dirty_and_head(self) -> None:
+        """A project whose HEAD left the manifest revision is measured."""
+        self.project.GetDirtyAndHead.return_value = (False, "local")
         self.project.GetRevisionId.return_value = "manifest"
         self.project.bare_git.count_objects.return_value = (
             "packs: 0\nsize-pack: 0\nsize-garbage: 0\n"
@@ -1006,15 +1003,28 @@ class CheckForBloatedProjects(unittest.TestCase):
         ):
             self.assertIsNone(self.cmd._CheckOneBloatedProject(0))
 
-        self.project.IsDirty.assert_not_called()
-        self.project.work_git.rev_parse.assert_not_called()
+        self.project.GetDirtyAndHead.assert_called_once_with()
+        self.project.bare_git.count_objects.assert_called_once_with("-v")
+
+    def test_one_dirty_project_is_measured(self) -> None:
+        """A dirty project is measured even if HEAD matches the manifest."""
+        self.project.GetDirtyAndHead.return_value = (True, "local")
+        self.project.GetRevisionId.return_value = "local"
+        self.project.bare_git.count_objects.return_value = (
+            "packs: 0\nsize-pack: 0\nsize-garbage: 0\n"
+        )
+        with mock.patch.object(
+            sync.Sync,
+            "get_parallel_context",
+            return_value={"projects": [self.project]},
+        ):
+            self.assertIsNone(self.cmd._CheckOneBloatedProject(0))
+
         self.project.bare_git.count_objects.assert_called_once_with("-v")
 
     def test_one_unborn_project_skips_bloat_check(self) -> None:
-        """A porcelain initial branch behaves like failed rev-parse HEAD."""
-        status = git_status.StatusSnapshot()
-        status.index_changes["staged"] = git_status.StatusEntry("staged", "M")
-        self.project._GetStatusSnapshot.return_value = status
+        """A project without a resolvable HEAD is skipped."""
+        self.project.GetDirtyAndHead.return_value = (True, None)
 
         with mock.patch.object(
             sync.Sync,