Merge pull request 'feat: harden issue lock store against concurrent session clobbering (Closes #438)' (#464) from feat/issue-438-lock-hardening into master

gitea_mcp_server.py

@@ -1382,6 +1382,19 @@ def gitea_lock_issue(
     }
 
     lock_file_path = _save_issue_lock(data)
+    lock_record = issue_lock_store.read_lock_file(lock_file_path) or data
+    freshness = issue_lock_store.assess_lock_freshness(lock_record)
+    competing = [
+        entry
+        for entry in issue_lock_store.list_live_locks()
+        if entry.get("issue_number") != issue_number
+    ]
+    lock_proof = issue_lock_store.format_lock_proof(
+        lock_record,
+        freshness=freshness,
+        competing_live_locks=competing,
+        released=False,
+    )
 
     agent_artifacts = agent_temp_artifacts.find_agent_temp_artifacts_from_porcelain(
         git_state.get("porcelain_status") or ""
@@ -1397,6 +1410,8 @@ def gitea_lock_issue(
         "worktree_path": resolved_worktree,
         "work_lease": work_lease,
         "lock_file_path": lock_file_path,
+        "lock_freshness": freshness,
+        "lock_proof": lock_proof,
     }
     if adoption["adopt"]:
         result["adoption"] = issue_lock_adoption.build_adoption_proof(
@@ -1529,6 +1544,15 @@ def gitea_create_pr(
             f"PR head branch '{head}' does not match locked branch '{locked_branch}' (fail closed)"
         )
 
+    ownership = issue_lock_store.verify_lock_for_mutation(
+        lock_data,
+        issue_number=locked_issue,
+        branch_name=head,
+        worktree_path=worktree_path,
+    )
+    if ownership["block"]:
+        raise ValueError(ownership["reasons"][0])
+
     # Check for forbidden terms anywhere in title/body
     forbidden_terms = ["equivalent", "related", "same as"]
     text_to_check = f"{title} {body}".lower()
@@ -6691,6 +6715,15 @@ def gitea_reconcile_issue_claims(
         heartbeat_lease_minutes=heartbeat_lease_minutes,
         reclaim_after_minutes=reclaim_after_minutes,
     )
+    live_locks = issue_lock_store.list_live_locks()
+    inventory["live_issue_locks"] = live_locks
+    inventory["live_issue_lock_numbers"] = sorted(
+        {
+            int(entry["issue_number"])
+            for entry in live_locks
+            if entry.get("issue_number") is not None
+        }
+    )
     inventory["cleanup_plan"] = issue_claim_heartbeat.build_cleanup_plan(inventory)
     inventory["success"] = True
     inventory["performed"] = False

issue_lock_store.py

@@ -1,18 +1,21 @@
-"""Keyed, persistent issue-lock storage (#443).
+"""Keyed, persistent issue-lock storage (#443) with flock hardening (#438).
 
 Replaces the single global ``/tmp/gitea_issue_lock.json`` slot with per-issue
 lock files under ``GITEA_ISSUE_LOCK_DIR`` (default
 ``~/.cache/gitea-tools/issue-locks``). Each MCP session binds its active lock
 via a per-process pointer file so concurrent repos/issues never clobber each
-other.
+other. Acquisition is serialized per issue with ``fcntl.flock``.
 """
 
 from __future__ import annotations
 
+import errno
+import fcntl
 import json
 import os
 import re
 import tempfile
+from contextlib import contextmanager
 from datetime import datetime, timedelta, timezone
 from typing import Any
 
@@ -24,6 +27,10 @@ AUTHOR_ISSUE_WORK_LEASE = "author_issue_work"
 _SAFE_SEGMENT_RE = re.compile(r"[^A-Za-z0-9._+-]+")
 
 
+class LockContentionError(RuntimeError):
+    """Raised when an exclusive per-issue lock cannot be acquired."""
+
+
 def default_lock_dir() -> str:
     raw = (os.environ.get(LOCK_DIR_ENV) or DEFAULT_LOCK_DIR).strip()
     return raw or DEFAULT_LOCK_DIR
@@ -72,6 +79,41 @@ def _ensure_lock_dir(lock_dir: str | None = None) -> str:
     return root
 
 
+def flock_path(json_path: str) -> str:
+    return f"{json_path}.lock"
+
+
+def is_process_alive(pid: int | None) -> bool:
+    if not pid or pid <= 0:
+        return False
+    try:
+        os.kill(int(pid), 0)
+        return True
+    except OSError as exc:
+        return exc.errno != errno.ESRCH
+    except (TypeError, ValueError):
+        return False
+
+
+@contextmanager
+def _exclusive_file_lock(lock_path: str):
+    os.makedirs(os.path.dirname(lock_path) or ".", exist_ok=True)
+    fd = os.open(lock_path, os.O_CREAT | os.O_RDWR, 0o600)
+    try:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+        except BlockingIOError as exc:
+            raise LockContentionError(
+                f"could not acquire exclusive lock on '{lock_path}'"
+            ) from exc
+        yield fd
+    finally:
+        try:
+            fcntl.flock(fd, fcntl.LOCK_UN)
+        finally:
+            os.close(fd)
+
+
 def read_lock_file(path: str) -> dict[str, Any] | None:
     lock_path = (path or "").strip()
     if not lock_path or not os.path.exists(lock_path):
@@ -126,7 +168,7 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) ->
     record = dict(lock_data)
     record["lock_file_path"] = path
     record["session_pid"] = os.getpid()
-    save_lock_file(path, record)
+    record.setdefault("pid", os.getpid())
 
     pointer = {
         "pid": os.getpid(),
@@ -137,7 +179,32 @@ def bind_session_lock(lock_data: dict[str, Any], lock_dir: str | None = None) ->
         "org": org,
         "repo": repo,
     }
-    save_lock_file(session_pointer_path(root), pointer)
+    sentinel = flock_path(path)
+    try:
+        with _exclusive_file_lock(sentinel):
+            existing = read_lock_file(path)
+            overwrite_block = assess_foreign_lock_overwrite(existing, record)
+            if overwrite_block:
+                raise RuntimeError(overwrite_block)
+            lease_block = assess_same_issue_lease_conflict(
+                existing,
+                issue_number=issue_number,
+                branch_name=str(record.get("branch_name") or ""),
+                worktree_path=str(record.get("worktree_path") or ""),
+            )
+            if lease_block:
+                raise RuntimeError(lease_block)
+            save_lock_file(path, record)
+            save_lock_file(session_pointer_path(root), pointer)
+    except LockContentionError as exc:
+        competing = read_lock_file(path)
+        if competing:
+            owner_pid = competing.get("session_pid") or competing.get("pid")
+            raise RuntimeError(
+                f"Issue #{issue_number} lock contention: {exc}; competing owner "
+                f"pid={owner_pid} (fail closed)"
+            ) from exc
+        raise RuntimeError(f"Issue #{issue_number} lock contention: {exc} (fail closed)") from exc
     return path
 
 
@@ -241,15 +308,62 @@ def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None
 
 
 def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
-    if not lock:
+    return assess_lock_freshness(lock, now=now)["live"]
-        return False
+
-    lease = lock.get("work_lease")
+
-    if not isinstance(lease, dict):
+def assess_lock_freshness(
-        return True
+    lock_data: dict[str, Any] | None,
-    expires = _parse_lease_timestamp(lease.get("expires_at"))
+    *,
-    if expires is None:
+    now: datetime | None = None,
-        return True
+) -> dict[str, Any]:
-    return expires > _lease_now(now)
+    """Classify a lock as live, expired, stale, or absent."""
+    current = _lease_now(now)
+    if not lock_data:
+        return {
+            "status": "absent",
+            "live": False,
+            "stale": False,
+            "reason": "no lock record",
+        }
+
+    expires_at = lease_expires_at(lock_data)
+    lease = lock_data.get("work_lease")
+    heartbeat_at = _parse_lease_timestamp(lock_data.get("last_heartbeat_at"))
+    if heartbeat_at is None and isinstance(lease, dict):
+        heartbeat_at = _parse_lease_timestamp(lease.get("last_heartbeat_at"))
+
+    pid = lock_data.get("session_pid")
+    if pid is None:
+        pid = lock_data.get("pid")
+    pid_alive = is_process_alive(pid) if pid is not None else False
+
+    if expires_at and expires_at <= current:
+        return {
+            "status": "expired",
+            "live": False,
+            "stale": True,
+            "reason": f"lease expired at {expires_at.isoformat()}",
+            "pid_alive": pid_alive,
+        }
+
+    if pid is not None and not pid_alive:
+        return {
+            "status": "stale",
+            "live": False,
+            "stale": True,
+            "reason": f"owner pid {pid} is not alive",
+            "pid_alive": False,
+        }
+
+    return {
+        "status": "live",
+        "live": True,
+        "stale": False,
+        "reason": "lock heartbeat and lease are fresh",
+        "pid_alive": pid_alive,
+        "heartbeat_at": heartbeat_at.isoformat() if heartbeat_at else None,
+        "expires_at": expires_at.isoformat() if expires_at else None,
+    }
 
 
 def _same_realpath(left: str | None, right: str | None) -> bool:
@@ -383,3 +497,116 @@ def has_active_issue_lock(
         if is_lease_live(lock):
             return True
     return False
+
+
+def verify_lock_for_mutation(
+    lock_data: dict[str, Any] | None,
+    *,
+    issue_number: int | None = None,
+    branch_name: str | None = None,
+    worktree_path: str | None = None,
+) -> dict[str, Any]:
+    """Re-check lock ownership immediately before a mutation (#438)."""
+    reasons: list[str] = []
+    if not lock_data:
+        return {"proven": False, "block": True, "reasons": ["issue lock is missing (fail closed)"]}
+
+    freshness = assess_lock_freshness(lock_data)
+    if not freshness["live"]:
+        reasons.append(f"issue lock is not live: {freshness['reason']} (fail closed)")
+
+    if issue_number is not None and lock_data.get("issue_number") != issue_number:
+        reasons.append(
+            f"issue lock targets #{lock_data.get('issue_number')}, expected #{issue_number} (fail closed)"
+        )
+
+    if branch_name is not None and lock_data.get("branch_name") != branch_name:
+        reasons.append(
+            f"issue lock branch '{lock_data.get('branch_name')}' does not match "
+            f"'{branch_name}' (fail closed)"
+        )
+
+    if worktree_path is not None:
+        locked = os.path.realpath(str(lock_data.get("worktree_path") or ""))
+        declared = os.path.realpath(worktree_path)
+        if locked != declared:
+            reasons.append(
+                f"issue lock worktree '{locked}' does not match declared '{declared}' (fail closed)"
+            )
+
+    return {
+        "proven": not reasons,
+        "block": bool(reasons),
+        "reasons": reasons,
+        "freshness": freshness,
+        "lock_proof": format_lock_proof(lock_data, freshness=freshness),
+    }
+
+
+def list_live_locks(
+    *,
+    lock_dir: str | None = None,
+    now: datetime | None = None,
+) -> list[dict[str, Any]]:
+    """Return live per-issue locks for queue visibility."""
+    live: list[dict[str, Any]] = []
+    for path in iter_lock_files(lock_dir):
+        record = read_lock_file(path)
+        if not record:
+            continue
+        freshness = assess_lock_freshness(record, now=now)
+        if not freshness["live"]:
+            continue
+        live.append(
+            {
+                "issue_number": record.get("issue_number"),
+                "branch_name": record.get("branch_name"),
+                "remote": record.get("remote"),
+                "org": record.get("org"),
+                "repo": record.get("repo"),
+                "worktree_path": record.get("worktree_path"),
+                "pid": record.get("session_pid") or record.get("pid"),
+                "claimant": (
+                    record.get("claimant")
+                    or (record.get("work_lease") or {}).get("claimant")
+                ),
+                "freshness": freshness,
+                "lock_path": record.get("lock_file_path") or path,
+            }
+        )
+    return live
+
+
+def format_lock_proof(
+    lock_data: dict[str, Any] | None,
+    *,
+    freshness: dict[str, Any] | None = None,
+    competing_live_locks: list[dict[str, Any]] | None = None,
+    released: bool | None = None,
+) -> str:
+    """Canonical issue-lock proof string for final reports."""
+    if not lock_data:
+        return "issue lock proof: not acquired"
+    fresh = freshness or assess_lock_freshness(lock_data)
+    owner = lock_data.get("claimant") or {}
+    if not owner and isinstance(lock_data.get("work_lease"), dict):
+        owner = lock_data["work_lease"].get("claimant") or {}
+    parts = [
+        "issue lock proof:",
+        f"acquired issue #{lock_data.get('issue_number')}",
+        f"branch {lock_data.get('branch_name')}",
+        f"owner {owner.get('profile') or 'unknown'}",
+        f"pid {lock_data.get('session_pid') or lock_data.get('pid')}",
+        f"freshness {fresh.get('status')}",
+    ]
+    if competing_live_locks is not None:
+        parts.append(
+            "no competing live lock"
+            if not competing_live_locks
+            else f"competing live locks {len(competing_live_locks)}"
+        )
+    if released is True:
+        parts.append("lock released")
+    elif released is False:
+        parts.append("lock retained")
+    return "; ".join(parts)

tests/test_issue_lock_store.py

@@ -1,8 +1,9 @@
-"""Unit tests for keyed issue-lock storage (#443)."""
+"""Unit tests for keyed issue-lock storage (#443) and flock hardening (#438)."""
 import json
 import os
 import sys
 import tempfile
+import threading
 import unittest
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
@@ -176,6 +177,80 @@ class TestIssueLockStore(unittest.TestCase):
         self.assertTrue(os.path.exists(path_b))
         self.assertEqual(ils.read_lock_file(path_a)["issue_number"], 108)
 
+    def test_concurrent_bind_same_issue_only_one_wins(self):
+        barrier = threading.Barrier(2)
+        results: list[str | Exception] = []
+
+        def worker():
+            barrier.wait()
+            try:
+                ils.bind_session_lock(
+                    _lock_record(worktree_path=f"/tmp/wt-{threading.get_ident()}")
+                )
+                results.append("ok")
+            except Exception as exc:  # noqa: BLE001
+                results.append(exc)
+
+        threads = [threading.Thread(target=worker) for _ in range(2)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+
+        successes = [item for item in results if item == "ok"]
+        failures = [item for item in results if isinstance(item, Exception)]
+        self.assertEqual(len(successes), 1)
+        self.assertEqual(len(failures), 1)
+        failure_text = str(failures[0]).lower()
+        self.assertTrue(
+            "active" in failure_text or "lock contention" in failure_text,
+            failures[0],
+        )
+
+    def test_verify_lock_for_mutation_blocks_stale_lock(self):
+        record = _lock_record(
+            work_lease=_lease("2000-01-01T00:00:00Z"),
+        )
+        record["pid"] = 999999
+        record["session_pid"] = 999999
+        result = ils.verify_lock_for_mutation(
+            record,
+            issue_number=420,
+            branch_name="feat/issue-420-server-code-parity",
+            worktree_path="/tmp/wt-420",
+        )
+        self.assertTrue(result["block"])
+        self.assertIn("not live", result["reasons"][0])
+
+    def test_list_live_locks_excludes_stale_records(self):
+        live_path = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=420,
+        )
+        ils.save_lock_file(
+            live_path,
+            _lock_record(worktree_path="/tmp/wt-420"),
+        )
+        stale_path = ils.lock_file_path(
+            remote="prgs",
+            org="Scaled-Tech-Consulting",
+            repo="Gitea-Tools",
+            issue_number=440,
+        )
+        ils.save_lock_file(
+            stale_path,
+            _lock_record(
+                issue_number=440,
+                branch_name="feat/issue-440-recovery",
+                work_lease=_lease("2000-01-01T00:00:00Z"),
+                worktree_path="/tmp/wt-440",
+            ),
+        )
+        live = ils.list_live_locks(lock_dir=self.lock_dir)
+        self.assertEqual([entry["issue_number"] for entry in live], [420])
+
 
 if __name__ == "__main__":
     unittest.main()