From d306c0a5ec0c33ce8168c2c0db6afc8062ebba58 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 18:07:44 -0400 Subject: [PATCH] feat: harden keyed issue lock store with flock serialization (#438) Rebase onto master (#443 keyed store) and port #438 hardening: per-issue fcntl.flock around bind_session_lock, assess_lock_freshness with dead-PID detection, verify_lock_for_mutation before create_pr, and live lock inventory for claim visibility. Co-Authored-By: Claude Opus 4.8 (1M context) --- gitea_mcp_server.py | 33 +++++ issue_lock_store.py | 255 +++++++++++++++++++++++++++++++-- tests/test_issue_lock_store.py | 77 +++++++++- 3 files changed, 350 insertions(+), 15 deletions(-) diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index c332bba..37a8e9b 100644 --- a/gitea_mcp_server.py +++ b/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 diff --git a/issue_lock_store.py b/issue_lock_store.py index 9ede4bb..6a7b4d7 100644 --- a/issue_lock_store.py +++ b/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 False - lease = lock.get("work_lease") - if not isinstance(lease, dict): - return True - expires = _parse_lease_timestamp(lease.get("expires_at")) - if expires is None: - return True - return expires > _lease_now(now) + return assess_lock_freshness(lock, now=now)["live"] + + +def assess_lock_freshness( + lock_data: dict[str, Any] | None, + *, + now: datetime | None = None, +) -> dict[str, Any]: + """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: @@ -382,4 +496,117 @@ def has_active_issue_lock( continue if is_lease_live(lock): return True - return False \ No newline at end of file + 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) \ No newline at end of file diff --git a/tests/test_issue_lock_store.py b/tests/test_issue_lock_store.py index fca8607..d87fc73 100644 --- a/tests/test_issue_lock_store.py +++ b/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() \ No newline at end of file