Close #2311 follow-ups: extend cross-process flock + unify state-store

gateway/checkpoint_handler.py

@@ -49,13 +49,15 @@
 - Uses checkpoint_loader for atomic writes to checkpoint branch
 """
 
+import contextlib
 import os
 import re
 import subprocess
 import sys
 import tempfile
 import threading
 import time
+from collections.abc import Generator
 from datetime import UTC, datetime
 from pathlib import Path
 
@@ -95,6 +97,7 @@
     UsageSaveError,
     update_usage_from_checkpoint,
 )
+from egg_git.cross_process_lock import bare_repo_lock
 from egg_logging import get_logger
 
 try:
@@ -879,11 +882,10 @@ def store_checkpoint_v2(
             return False
 
         target = _resolve_checkpoint_target(checkpoint_repo, remote, repo_path)
-        repo_lock = _get_repo_lock(repo_path)
 
         try:
             with (
-                repo_lock,
+                _get_repo_lock(repo_path),
                 tempfile.TemporaryDirectory(
                     prefix="checkpoint_", ignore_cleanup_errors=True
                 ) as temp_dir,
@@ -1337,13 +1339,27 @@ def _run_git(
 _repo_locks_guard = threading.Lock()
 
 
-def _get_repo_lock(repo_path: str) -> threading.Lock:
+@contextlib.contextmanager
+def _get_repo_lock(repo_path: str) -> Generator[None]:
+    """Hold the per-repo lock for serializing checkpoint git operations.
+
+    Combines two layers, mirroring ``WorktreeManager._get_repo_lock``:
+
+    * A per-repo ``threading.Lock`` for in-process serialization (#2069).
+    * ``bare_repo_lock`` for cross-process serialization against the
+      orchestrator's state-store, which runs git from a different pod
+      but shares the same hostPath-mounted bare repo (#2311).  Without
+      this, a checkpoint store's ``git worktree add`` can race a
+      state-store commit on ``.git/config.lock`` and fail with
+      ``could not lock config file .git/config: File exists``.
+    """
     with _repo_locks_guard:
-        lock = _repo_locks.get(repo_path)
-        if lock is None:
-            lock = threading.Lock()
-            _repo_locks[repo_path] = lock
-        return lock
+        thread_lock = _repo_locks.get(repo_path)
+        if thread_lock is None:
+            thread_lock = threading.Lock()
+            _repo_locks[repo_path] = thread_lock
+    with thread_lock, bare_repo_lock(repo_path):
+        yield
 
 
 def get_checkpoint_handler(github_token: str | None = None) -> CheckpointHandler:

gateway/tests/test_checkpoint_handler.py

@@ -1,9 +1,20 @@
-"""Tests for checkpoint_handler module - checkpoint creation and session-end."""
+"""Tests for checkpoint_handler module - checkpoint creation and session-end.
 
+Note: the module-level ``_stub_bare_repo_lock`` autouse fixture below replaces
+the cross-process flock primitive with a no-op for every test in this file.
+Tests added here that need to exercise the real ``bare_repo_lock`` path
+(rather than just ``_get_repo_lock``'s in-process serialization) must opt out
+explicitly or live in ``shared/tests/test_cross_process_lock.py`` /
+``orchestrator/tests/test_state_store.py`` instead.
+"""
+
+import contextlib
 import subprocess
 from datetime import UTC, datetime, timedelta
 from unittest.mock import MagicMock, patch
 
+import pytest
+
 # Import from conftest-loaded modules
 from checkpoint_handler import (
     _extract_repo_from_remote,
@@ -14,6 +25,27 @@
 from session_manager import Session, _hash_token
 
 
+@pytest.fixture(autouse=True)
+def _stub_bare_repo_lock(monkeypatch):
+    """Replace ``bare_repo_lock`` with a no-op for these unit tests.
+
+    The cross-process flock primitive (#2311) requires a real
+    ``<repo>/.git/`` to exist so it can ``mkdir`` the sentinel and
+    ``os.open`` an fd.  These tests pass sentinel paths like
+    ``/fake/repo`` which would fail the mkdir.  Cross-process behaviour
+    is covered by ``shared/tests/test_cross_process_lock.py`` and the
+    worktree integration test — here we only need ``_get_repo_lock``'s
+    in-process serialization to work.
+    """
+    import checkpoint_handler
+
+    @contextlib.contextmanager
+    def _noop(repo_path):
+        yield
+
+    monkeypatch.setattr(checkpoint_handler, "bare_repo_lock", _noop)
+
+
 class TestCaptureAndStoreCheckpointsForPush:
     """Tests for capture_and_store_checkpoints_for_push function."""
 

orchestrator/state_store.py

@@ -13,7 +13,6 @@
 pattern for cross-host recovery.
 """
 
-import fcntl
 import json
 import logging
 import os
@@ -29,7 +28,7 @@
 from pathlib import Path
 from typing import Any, ClassVar, Literal
 
-from egg_git.cross_process_lock import lock_path_for_repo
+from egg_git.cross_process_lock import bare_repo_lock, lock_path_for_repo
 from models import Pipeline, PipelineMode, PipelinePhase, PipelineStatus
 from pydantic import ValidationError
 
@@ -120,15 +119,6 @@ class StateStore:
 
     PIPELINES_DIR = ".egg-state/pipelines"
 
-    # -- cross-process git serialization ------------------------------------
-    # RLock allows compound operations (_commit_state, delete_pipeline) to
-    # hold the lock while inner _run_git calls re-enter without deadlocking.
-    # fcntl.flock provides cross-process serialization via the shared
-    # filesystem — threading locks only protect within a single process.
-    _thread_lock: ClassVar[threading.RLock] = threading.RLock()
-    _flock_fds: ClassVar[dict[str, int]] = {}
-    _flock_depth: ClassVar[int] = 0  # nesting depth, protected by _thread_lock
-
     # -- remote sync state -------------------------------------------------
     _push_in_flight: ClassVar[bool] = False
     _push_pending: ClassVar[bool] = False
@@ -166,41 +156,27 @@ def _lock_path(self) -> Path:
         """
         return lock_path_for_repo(self.repo_path)
 
-    @classmethod
-    def _get_flock_fd(cls, lock_path: Path) -> int:
-        """Get or create a file descriptor for cross-process flock."""
-        key = str(lock_path)
-        if key not in cls._flock_fds:
-            lock_path.parent.mkdir(parents=True, exist_ok=True)
-            cls._flock_fds[key] = os.open(str(lock_path), os.O_CREAT | os.O_RDWR)
-        return cls._flock_fds[key]
-
     @contextmanager
     def _git_op(self) -> Generator[None]:
-        """Acquire thread + process locks for git operations.
-
-        Combines a reentrant threading lock (for in-process thread
-        serialization) with an ``fcntl.flock`` file lock (for cross-process
-        serialization via shared filesystem).
-
-        Reentrant: safe to nest.  Compound operations (e.g. ``_commit_state``)
-        hold the lock for their entire duration while inner ``_run_git`` calls
-        re-enter without releasing.
+        """Acquire the cross-process lock for git operations.
+
+        Delegates to ``egg_git.cross_process_lock.bare_repo_lock``, which
+        provides a reentrant in-process thread lock plus an ``fcntl.flock``
+        file lock keyed on the same inode the gateway uses.  Nested calls
+        re-enter via the shared depth counter inside ``bare_repo_lock``,
+        so compound operations (e.g. ``_commit_state``) can hold the lock
+        across several inner ``_run_git`` calls without self-deadlocking.
+
+        Until this delegation, ``StateStore`` maintained a parallel
+        implementation of the same flock protocol.  Both kept the same
+        invariants and ``flock`` keys on the inode regardless of fd, so
+        cross-pod serialisation worked — but two implementations was a
+        drift trap, and would self-deadlock if ever co-located in one
+        process (each owned its own fd, and ``flock(2)`` treats fds on the
+        same file as independent for the calling process).
         """
-        self._thread_lock.acquire()
-        try:
-            fd = self._get_flock_fd(self._lock_path)
-            if StateStore._flock_depth == 0:
-                fcntl.flock(fd, fcntl.LOCK_EX)
-            StateStore._flock_depth += 1
-            try:
-                yield
-            finally:
-                StateStore._flock_depth -= 1
-                if StateStore._flock_depth == 0:
-                    fcntl.flock(fd, fcntl.LOCK_UN)
-        finally:
-            self._thread_lock.release()
+        with bare_repo_lock(self.repo_path):
+            yield
 
     # -- worktree lifecycle ------------------------------------------------
 
@@ -243,11 +219,12 @@ def _ensure_worktree(self) -> Path:
         # retry inside ``_add_worktree_with_branch_recovery`` and surface
         # misleading one-shot 500s on whichever arrived second (#2177).
         # The same root cause produced #2234's ENOENT race.  ``_git_op``
-        # is reentrant via ``_flock_depth``, so nested ``_run_git`` calls
-        # compose without deadlock.  Cost: lock window grows from "one
-        # git command" to "the worktree-bring-up sequence" — tens of ms
-        # in steady state; the cold-start ``_restore_from_remote`` fetch
-        # is the longest case, runs at most once per repo per process.
+        # delegates to ``bare_repo_lock``, which is reentrant via a depth
+        # counter, so nested ``_run_git`` calls compose without deadlock.
+        # Cost: lock window grows from "one git command" to "the
+        # worktree-bring-up sequence" — tens of ms in steady state; the
+        # cold-start ``_restore_from_remote`` fetch is the longest case,
+        # runs at most once per repo per process.
         with self._git_op():
             # Clean up stale admin dir for THIS worktree only (e.g., from crashes).
             # IMPORTANT: Do NOT use `git worktree prune` — the orchestrator cannot

orchestrator/tests/test_state_store.py

@@ -7,10 +7,14 @@
 import os
 import shutil
 import subprocess
+import sys
 import threading
+import time
+from pathlib import Path
 from unittest.mock import MagicMock, patch
 
 import pytest
+from egg_git import cross_process_lock
 from gateway_client import PushResult
 from models import Pipeline, PipelinePhase, PipelineStatus
 from state_store import (
@@ -27,6 +31,21 @@
 )
 
 
+def _flock_depth_for(repo_path) -> int:
+    """Return the cross-process lock nesting depth for ``repo_path``.
+
+    Reaches into the ``bare_repo_lock`` per-repo state (keyed on the
+    resolved path) so tests can assert reentrant nesting.  Returns 0 if
+    the repo has no active state.
+    """
+    try:
+        key = str(repo_path.resolve())
+    except OSError:
+        key = str(repo_path)
+    state = cross_process_lock._per_repo_state.get(key)
+    return state.depth if state is not None else 0
+
+
 @pytest.fixture
 def mock_git():
     """Mock git operations."""
@@ -1015,10 +1034,7 @@ class TestRunGitLocking:
     @pytest.fixture(autouse=True)
     def reset_flock_state(self):
         yield
-        StateStore._flock_depth = 0
-        for fd in StateStore._flock_fds.values():
-            os.close(fd)
-        StateStore._flock_fds.clear()
+        cross_process_lock.reset_for_tests()
 
     def test_retry_succeeds_after_index_lock_error(self, tmp_path):
         """Test that _run_git retries on index.lock contention and succeeds."""
@@ -1155,7 +1171,7 @@ def tracking_run(*args, **kwargs):
             # Record the flock nesting depth during each subprocess call.
             # If compound locking works, depth should be >= 2 (outer _commit_state
             # + inner _run_git).
-            depth_during_calls.append(StateStore._flock_depth)
+            depth_during_calls.append(_flock_depth_for(tmp_path))
             return MagicMock(stdout="abc1234\n", returncode=0)
 
         with patch("subprocess.run", side_effect=tracking_run):
@@ -1165,6 +1181,78 @@ def tracking_run(*args, **kwargs):
         assert len(depth_during_calls) >= 2  # at least add + diff (or add + diff + commit)
         assert all(d >= 2 for d in depth_during_calls)
 
+    def test_git_op_serialises_against_gateway_bare_repo_lock(self, tmp_path):
+        """``StateStore._git_op`` and ``bare_repo_lock`` lock the same inode.
+
+        Regression for #2311: in production the gateway holds
+        ``bare_repo_lock`` and the orchestrator holds ``_git_op`` from
+        different pods.  Before #2312 they used independent flock
+        implementations; after the unification in this PR, ``_git_op``
+        delegates to ``bare_repo_lock`` directly.  This test exercises
+        both wrappers across a process boundary against the same
+        ``<repo>/.git/.egg-cross-process.lock`` inode and asserts that
+        the parent's ``bare_repo_lock`` blocks while a child holds
+        ``StateStore._git_op``.
+        """
+        import textwrap
+
+        from egg_git.cross_process_lock import bare_repo_lock
+
+        (tmp_path / ".git").mkdir()
+        sentinel = tmp_path / "held"
+
+        project_root = Path(__file__).resolve().parent.parent.parent
+        holder_script = textwrap.dedent(
+            f"""
+            import sys, time
+            sys.path.insert(0, {str(project_root / "orchestrator")!r})
+            sys.path.insert(0, {str(project_root / "shared")!r})
+            from pathlib import Path
+            from state_store import StateStore
+
+            store = StateStore(Path({str(tmp_path)!r}))
+            with store._git_op():
+                open({str(sentinel)!r}, "w").close()
+                time.sleep(1.0)
+            """
+        )
+
+        proc = subprocess.Popen(
+            [sys.executable, "-c", holder_script],
+            env={**os.environ, "PYTHONUNBUFFERED": "1"},
+            stderr=subprocess.PIPE,
+        )
+
+        try:
+            deadline = time.monotonic() + 5
+            while not sentinel.exists() and time.monotonic() < deadline:
+                time.sleep(0.01)
+            if not sentinel.exists():
+                stderr_text = ""
+                if proc.poll() is None:
+                    proc.kill()
+                    proc.wait(timeout=2)
+                if proc.stderr is not None:
+                    stderr_text = proc.stderr.read().decode("utf-8", errors="replace")
+                raise AssertionError(
+                    "child never signalled lock acquisition"
+                    + (f"; child stderr:\n{stderr_text}" if stderr_text else "")
+                )
+
+            start = time.monotonic()
+            with bare_repo_lock(tmp_path):
+                wait = time.monotonic() - start
+            proc.wait(timeout=5)
+        finally:
+            if proc.poll() is None:
+                proc.kill()
+                proc.wait(timeout=2)
+
+        assert wait > 0.5, (
+            f"parent acquired lock too quickly ({wait:.3f}s) — "
+            "_git_op and bare_repo_lock did not serialise on the same inode"
+        )
+
 
 class TestRemoteSync:
     """Tests for remote sync (push/restore) of the state branch."""
@@ -1924,13 +2012,13 @@ def fake_run(*args, check=True, cwd=None):
         original_remove = StateStore._remove_admin_dir_for_path
 
         def tracked_remove(self, target_path):
-            # _flock_depth >= 2 proves *both* wraps are in place: the
-            # outer one in _ensure_worktree (depth 1) and the inner one
-            # in _add_worktree_with_branch_recovery (depth 2).  A weaker
+            # depth >= 2 proves *both* wraps are in place: the outer one
+            # in _ensure_worktree (depth 1) and the inner one in
+            # _add_worktree_with_branch_recovery (depth 2).  A weaker
             # ``> 0`` assertion would still pass if a future refactor
             # dropped the outer wrap and left only the inner one — and
             # that would re-open the #2177 race.
-            depth_when_removing_admin.append(StateStore._flock_depth)
+            depth_when_removing_admin.append(_flock_depth_for(store_seed.repo_path))
             return original_remove(self, target_path)
 
         def caller():