sysadminandClaude Opus 4.8
5d59c57c98
`gitea_update_pr_branch_by_merge` advanced a PR's remote head but never advanced the linked durable issue lock's recorded head. After the owning session died the drifted lock became unrecoverable and no further synchronization was possible (PR #866 / issue #855). Write-side (prevents future drift): - issue_lock_store.assess/apply_durable_lock_head_refresh: on a successful sync, CAS-refresh the durable lock's recorded head from the exact expected PR head to the resulting head, re-verifying repo/issue/branch/worktree/ identity/profile/live-session ownership, with read-after-write verification. - gitea_update_pr_branch_by_merge now refreshes the lock after the remote advance and reports a PARTIAL LIFECYCLE FAILURE (success=False) when the refresh fails, instead of falsely reporting a full synchronization. Read-side (recovers already-drifted locks): - issue_lock_worktree.read_merge_sync_provenance: server-side git observation proving a remote head is a sanctioned base-into-branch merge that preserved the branch mainline back to the recorded head. - issue_lock_recovery: new HEAD_RELATION_REMOTE_MERGE_SYNCED accepts a dead-session lock whose recorded head is a strict merge-sync ancestor of the live PR head — and only that. Rewrites, rebases, force-pushes, non-ancestor heads, dirty worktrees, live/competing owners, and wrong repo/issue/branch/ identity/profile all stay protected. No existing exact-head, branch-protection, parity, workspace, identity, role, or mutation-safety gate is weakened. All provenance is server-derived; nothing is reachable from an MCP caller. Tests: tests/test_issue_871_durable_lock_head_refresh.py (32 cases) covering first/second sync, CAS, ownership, partial-failure, merge-sync recovery happy-path and every fail-closed branch. Full suite: 4789 passed, 13 pre- existing baseline failures unchanged. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]> Claude-Session: https://claude.ai/code/session_01FZPyVh2DGczQrDxtqwGH5p
1044 lines
36 KiB
Python
1044 lines
36 KiB
Python
"""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. 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
|
|
|
|
LOCK_DIR_ENV = "GITEA_ISSUE_LOCK_DIR"
|
|
DEFAULT_LOCK_DIR = os.path.expanduser("~/.cache/gitea-tools/issue-locks")
|
|
WORK_LEASE_TTL_HOURS = 4
|
|
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
|
|
|
|
|
|
def _sanitize_segment(value: str) -> str:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
return "_"
|
|
return _SAFE_SEGMENT_RE.sub("_", text)
|
|
|
|
|
|
def lock_key(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
) -> str:
|
|
return "-".join(
|
|
_sanitize_segment(part)
|
|
for part in (remote, org, repo, str(issue_number))
|
|
)
|
|
|
|
|
|
def lock_file_path(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
lock_dir: str | None = None,
|
|
) -> str:
|
|
root = (lock_dir or default_lock_dir()).strip()
|
|
return os.path.join(root, f"{lock_key(remote=remote, org=org, repo=repo, issue_number=issue_number)}.json")
|
|
|
|
|
|
def session_pointer_path(lock_dir: str | None = None) -> str:
|
|
root = (lock_dir or default_lock_dir()).strip()
|
|
return os.path.join(root, f"session-{os.getpid()}.json")
|
|
|
|
|
|
def _ensure_lock_dir(lock_dir: str | None = None) -> str:
|
|
root = (lock_dir or default_lock_dir()).strip()
|
|
os.makedirs(root, mode=0o700, exist_ok=True)
|
|
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):
|
|
return None
|
|
try:
|
|
with open(lock_path, encoding="utf-8") as handle:
|
|
data = json.load(handle)
|
|
except (OSError, json.JSONDecodeError):
|
|
return None
|
|
return data if isinstance(data, dict) else None
|
|
|
|
|
|
def save_lock_file(path: str, data: dict[str, Any]) -> None:
|
|
lock_path = (path or "").strip()
|
|
if not lock_path:
|
|
raise ValueError("lock path is required (fail closed)")
|
|
parent = os.path.dirname(lock_path) or "."
|
|
os.makedirs(parent, mode=0o700, exist_ok=True)
|
|
payload = json.dumps(data, indent=2, sort_keys=True) + "\n"
|
|
fd, temp_path = tempfile.mkstemp(prefix=".lock-", suffix=".json", dir=parent)
|
|
try:
|
|
with os.fdopen(fd, "w", encoding="utf-8") as handle:
|
|
handle.write(payload)
|
|
handle.flush()
|
|
os.fsync(handle.fileno())
|
|
os.replace(temp_path, lock_path)
|
|
finally:
|
|
if os.path.exists(temp_path):
|
|
try:
|
|
os.remove(temp_path)
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def lock_generation(lock: dict[str, Any] | None) -> int:
|
|
"""Monotonic write counter for a durable lock record (#772 AC5).
|
|
|
|
Absent or unusable values read as ``0`` so a lock written before generations
|
|
existed still participates in compare-and-swap: its first recovery expects
|
|
``0`` and writes ``1``.
|
|
"""
|
|
if not isinstance(lock, dict):
|
|
return 0
|
|
try:
|
|
return int(lock.get("lock_generation") or 0)
|
|
except (TypeError, ValueError):
|
|
return 0
|
|
|
|
|
|
def bind_session_lock(
|
|
lock_data: dict[str, Any],
|
|
lock_dir: str | None = None,
|
|
*,
|
|
expected_generation: int | None = None,
|
|
renewal_sanctioned: bool = False,
|
|
) -> str:
|
|
"""Persist a keyed lock and bind it to the current process session.
|
|
|
|
``expected_generation`` turns the write into a compare-and-swap (#772 AC5).
|
|
Recovery decides it may take over a claim by reading the durable lock, but
|
|
that read and this write are separate steps; without a CAS two replacement
|
|
sessions can both observe the same dead owner, both pass assessment, and
|
|
both write — the second silently clobbering the first. Passing the
|
|
generation observed at assessment time makes exactly one of them win: the
|
|
loser's expectation no longer matches and it fails closed.
|
|
"""
|
|
remote = str(lock_data.get("remote") or "")
|
|
org = str(lock_data.get("org") or "")
|
|
repo = str(lock_data.get("repo") or "")
|
|
issue_number = int(lock_data.get("issue_number") or 0)
|
|
if not remote or not org or not repo or issue_number <= 0:
|
|
raise ValueError("lock record must include remote, org, repo, and issue_number")
|
|
|
|
root = _ensure_lock_dir(lock_dir)
|
|
path = lock_file_path(
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
issue_number=issue_number,
|
|
lock_dir=root,
|
|
)
|
|
record = dict(lock_data)
|
|
record["lock_file_path"] = path
|
|
record["session_pid"] = os.getpid()
|
|
record.setdefault("pid", os.getpid())
|
|
|
|
pointer = {
|
|
"pid": os.getpid(),
|
|
"lock_file_path": path,
|
|
"issue_number": issue_number,
|
|
"branch_name": record.get("branch_name"),
|
|
"remote": remote,
|
|
"org": org,
|
|
"repo": repo,
|
|
}
|
|
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 ""),
|
|
renewal_sanctioned=renewal_sanctioned,
|
|
)
|
|
if lease_block:
|
|
raise RuntimeError(lease_block)
|
|
# #772 AC5: compare-and-swap inside the same critical section that
|
|
# already serializes writers, so the check and the write cannot be
|
|
# separated by another session's successful recovery.
|
|
current_generation = lock_generation(existing)
|
|
if (
|
|
expected_generation is not None
|
|
and current_generation != expected_generation
|
|
):
|
|
raise RuntimeError(
|
|
f"Issue #{issue_number} lock generation changed: expected "
|
|
f"{expected_generation}, found {current_generation}; another "
|
|
"session already recovered or replaced this claim (fail closed)"
|
|
)
|
|
record["lock_generation"] = current_generation + 1
|
|
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
|
|
|
|
|
|
def read_session_issue_lock(lock_dir: str | None = None) -> dict[str, Any] | None:
|
|
root = (lock_dir or default_lock_dir()).strip()
|
|
pointer = read_lock_file(session_pointer_path(root))
|
|
if not pointer:
|
|
return None
|
|
lock_path = str(pointer.get("lock_file_path") or "").strip()
|
|
if not lock_path:
|
|
return None
|
|
return read_lock_file(lock_path)
|
|
|
|
|
|
def load_issue_lock(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
lock_dir: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
return read_lock_file(
|
|
lock_file_path(
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
issue_number=issue_number,
|
|
lock_dir=lock_dir,
|
|
)
|
|
)
|
|
|
|
|
|
def iter_lock_files(lock_dir: str | None = None) -> list[str]:
|
|
root = (lock_dir or default_lock_dir()).strip()
|
|
if not os.path.isdir(root):
|
|
return []
|
|
paths: list[str] = []
|
|
for name in os.listdir(root):
|
|
if not name.endswith(".json") or name.startswith("session-"):
|
|
continue
|
|
paths.append(os.path.join(root, name))
|
|
return sorted(paths)
|
|
|
|
|
|
def find_lock_for_branch(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
branch_name: str,
|
|
lock_dir: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
target = (branch_name or "").strip()
|
|
if not target:
|
|
return None
|
|
for path in iter_lock_files(lock_dir):
|
|
lock = read_lock_file(path)
|
|
if not lock:
|
|
continue
|
|
if (
|
|
str(lock.get("remote") or "") == remote
|
|
and str(lock.get("org") or "") == org
|
|
and str(lock.get("repo") or "") == repo
|
|
and str(lock.get("branch_name") or "").strip() == target
|
|
):
|
|
lock = dict(lock)
|
|
lock.setdefault("lock_file_path", path)
|
|
return lock
|
|
return None
|
|
|
|
|
|
def _lease_now(now: datetime | None = None) -> datetime:
|
|
return now or datetime.now(timezone.utc)
|
|
|
|
|
|
def _parse_lease_timestamp(value: str | None) -> datetime | None:
|
|
text = (value or "").strip()
|
|
if not text:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(text.replace("Z", "+00:00")).astimezone(timezone.utc)
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def lease_expires_at(lock: dict[str, Any] | None) -> datetime | None:
|
|
if not lock:
|
|
return None
|
|
lease = lock.get("work_lease")
|
|
if not isinstance(lease, dict):
|
|
return None
|
|
return _parse_lease_timestamp(lease.get("expires_at"))
|
|
|
|
|
|
def is_lease_expired(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
|
|
expires = lease_expires_at(lock)
|
|
if expires is None:
|
|
return False
|
|
return expires <= _lease_now(now)
|
|
|
|
|
|
def is_lease_live(lock: dict[str, Any] | None, *, now: datetime | None = None) -> bool:
|
|
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:
|
|
if not left or not right:
|
|
return False
|
|
try:
|
|
return os.path.realpath(left) == os.path.realpath(right)
|
|
except OSError:
|
|
return left == right
|
|
|
|
|
|
|
|
def assess_expired_lock_reclaim(
|
|
existing_lock: dict[str, Any] | None,
|
|
*,
|
|
now: datetime | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Decide whether an expired/stale author issue lock may be reclaimed (#601).
|
|
|
|
Required proof (any reclaim of non-live lock):
|
|
* lease not live (expired or dead pid)
|
|
* owner process dead OR worktree missing
|
|
* no force-delete of live foreign ownership
|
|
"""
|
|
if not existing_lock:
|
|
return {
|
|
"reclaim_allowed": True,
|
|
"reasons": ["no existing lock"],
|
|
"freshness": assess_lock_freshness(None, now=now),
|
|
}
|
|
freshness = assess_lock_freshness(existing_lock, now=now)
|
|
if freshness.get("live"):
|
|
return {
|
|
"reclaim_allowed": False,
|
|
"reasons": ["lock is still live; cannot reclaim (fail closed)"],
|
|
"freshness": freshness,
|
|
}
|
|
pid = existing_lock.get("session_pid")
|
|
if pid is None:
|
|
pid = existing_lock.get("pid")
|
|
dead = not is_process_alive(pid) if pid is not None else True
|
|
wt = str(existing_lock.get("worktree_path") or "")
|
|
missing_wt = (not wt) or (not os.path.isdir(os.path.realpath(wt)))
|
|
if not (dead or missing_wt):
|
|
return {
|
|
"reclaim_allowed": False,
|
|
"reasons": [
|
|
"expired/stale lock still has live owner pid and present worktree; "
|
|
"recovery review required (fail closed)"
|
|
],
|
|
"freshness": freshness,
|
|
"owner_pid_dead": dead,
|
|
"worktree_missing": missing_wt,
|
|
}
|
|
return {
|
|
"reclaim_allowed": True,
|
|
"reasons": [
|
|
"non-live lock with dead process and/or missing worktree; "
|
|
"sanctioned reclaim allowed"
|
|
],
|
|
"freshness": freshness,
|
|
"owner_pid_dead": dead,
|
|
"worktree_missing": missing_wt,
|
|
"prior_branch": existing_lock.get("branch_name"),
|
|
"prior_worktree": existing_lock.get("worktree_path"),
|
|
"prior_pid": pid,
|
|
}
|
|
|
|
|
|
def assess_same_issue_lease_conflict(
|
|
existing_lock: dict[str, Any] | None,
|
|
*,
|
|
issue_number: int,
|
|
branch_name: str,
|
|
worktree_path: str,
|
|
operation_type: str = AUTHOR_ISSUE_WORK_LEASE,
|
|
renewal_sanctioned: bool = False,
|
|
now: datetime | None = None,
|
|
) -> str | None:
|
|
"""Return a fail-closed error when a competing live lease blocks acquisition.
|
|
|
|
``renewal_sanctioned`` is set only when
|
|
``issue_lock_renewal.assess_exact_owner_lease_renewal`` has already proven,
|
|
from the durable lock plus live server-side observation, that this session
|
|
is the exact recorded owner of an *expired* lease (#760). It is never a
|
|
caller-supplied parameter of any MCP tool (#760 AC14): the server computes
|
|
it and passes it down. Left False, every pre-existing disposition is
|
|
unchanged.
|
|
"""
|
|
if not existing_lock:
|
|
return None
|
|
|
|
existing_issue = existing_lock.get("issue_number")
|
|
lease = existing_lock.get("work_lease")
|
|
existing_operation = (
|
|
lease.get("operation_type")
|
|
if isinstance(lease, dict)
|
|
else AUTHOR_ISSUE_WORK_LEASE
|
|
)
|
|
if existing_issue != issue_number or existing_operation != operation_type:
|
|
return None
|
|
|
|
existing_branch = existing_lock.get("branch_name")
|
|
existing_worktree = existing_lock.get("worktree_path")
|
|
same_owner = (
|
|
existing_branch == branch_name
|
|
and _same_realpath(str(existing_worktree or ""), worktree_path)
|
|
)
|
|
if is_lease_expired(existing_lock, now=now):
|
|
# #760 AC1/AC2: exact-owner renewal is a different disposition from
|
|
# foreign takeover and is evaluated first. Before this, both branches
|
|
# below returned unconditionally, so the same_owner allowance further
|
|
# down was unreachable for every expired lease — an owner could never
|
|
# renew its own lock once the wall clock passed, no matter how complete
|
|
# its ownership evidence. Requires BOTH the locally recomputed
|
|
# same_owner match and the server-proven renewal waiver; either alone is
|
|
# insufficient.
|
|
if same_owner and renewal_sanctioned:
|
|
return None
|
|
reclaim = assess_expired_lock_reclaim(existing_lock, now=now)
|
|
if reclaim.get("reclaim_allowed"):
|
|
# #601: expired + dead pid / missing worktree may be reclaimed
|
|
# through the normal lock path (sanctioned overwrite).
|
|
return None
|
|
return (
|
|
f"Issue #{issue_number} has an expired {operation_type} lease on "
|
|
f"branch '{existing_branch}' from worktree '{existing_worktree}'. "
|
|
"Recovery review is required before takeover (fail closed)"
|
|
)
|
|
if same_owner:
|
|
return None
|
|
return (
|
|
f"Issue #{issue_number} already has an active {operation_type} lease on "
|
|
f"branch '{existing_branch}' from worktree '{existing_worktree}' "
|
|
"(fail closed)"
|
|
)
|
|
|
|
|
|
def assess_foreign_lock_overwrite(
|
|
existing_lock: dict[str, Any] | None,
|
|
incoming_lock: dict[str, Any],
|
|
*,
|
|
now: datetime | None = None,
|
|
) -> str | None:
|
|
"""Block writes that would clobber an unrelated live lease on the same key."""
|
|
if not existing_lock:
|
|
return None
|
|
|
|
same_issue = existing_lock.get("issue_number") == incoming_lock.get("issue_number")
|
|
same_branch = existing_lock.get("branch_name") == incoming_lock.get("branch_name")
|
|
same_worktree = _same_realpath(
|
|
str(existing_lock.get("worktree_path") or ""),
|
|
str(incoming_lock.get("worktree_path") or ""),
|
|
)
|
|
if same_issue and same_branch and same_worktree:
|
|
return None
|
|
if not is_lease_live(existing_lock, now=now):
|
|
return None
|
|
return (
|
|
"Refusing to overwrite a live foreign issue lock "
|
|
f"(issue #{existing_lock.get('issue_number')}, "
|
|
f"branch '{existing_lock.get('branch_name')}', "
|
|
f"worktree '{existing_lock.get('worktree_path')}') (fail closed)"
|
|
)
|
|
|
|
|
|
def find_live_lock_for_branch(
|
|
branch_name: str,
|
|
lock_dir: str | None = None,
|
|
) -> dict[str, Any] | None:
|
|
target = (branch_name or "").strip()
|
|
if not target:
|
|
return None
|
|
for path in iter_lock_files(lock_dir):
|
|
lock = read_lock_file(path)
|
|
if not lock:
|
|
continue
|
|
if str(lock.get("branch_name") or "").strip() != target:
|
|
continue
|
|
if not is_lease_live(lock):
|
|
continue
|
|
record = dict(lock)
|
|
record.setdefault("lock_file_path", path)
|
|
return record
|
|
return None
|
|
|
|
|
|
def resolve_locked_branch_for_session(
|
|
branch_name: str | None = None,
|
|
lock_dir: str | None = None,
|
|
) -> str:
|
|
if branch_name:
|
|
lock = find_live_lock_for_branch(branch_name, lock_dir)
|
|
if lock:
|
|
return str(lock.get("branch_name") or "")
|
|
lock = read_session_issue_lock(lock_dir)
|
|
return str((lock or {}).get("branch_name") or "")
|
|
|
|
|
|
def has_active_issue_lock(
|
|
branch: str,
|
|
*,
|
|
lock_dir: str | None = None,
|
|
) -> bool:
|
|
target = (branch or "").strip()
|
|
if not target:
|
|
return False
|
|
for path in iter_lock_files(lock_dir):
|
|
lock = read_lock_file(path)
|
|
if not lock:
|
|
continue
|
|
if str(lock.get("branch_name") or "").strip() != target:
|
|
continue
|
|
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)
|
|
|
|
|
|
# ── #871: durable linked-issue lock head refresh after branch synchronization ──
|
|
_FULL_SHA_RE = re.compile(r"^[0-9a-f]{40}$", re.IGNORECASE)
|
|
|
|
# Provenance recorded on the lock when the head is refreshed by a sanctioned
|
|
# merge-based branch synchronization (``gitea_update_pr_branch_by_merge``).
|
|
LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC = "gitea_update_pr_branch_by_merge"
|
|
|
|
|
|
def _norm_sha(value: Any) -> str | None:
|
|
text = str(value or "").strip().lower()
|
|
return text if _FULL_SHA_RE.match(text) else None
|
|
|
|
|
|
def _lock_claimant_view(lock: dict[str, Any] | None) -> dict[str, Any]:
|
|
if not isinstance(lock, dict):
|
|
return {}
|
|
claimant = lock.get("claimant")
|
|
if not isinstance(claimant, dict):
|
|
lease = lock.get("work_lease")
|
|
claimant = lease.get("claimant") if isinstance(lease, dict) else None
|
|
return dict(claimant) if isinstance(claimant, dict) else {}
|
|
|
|
|
|
def assess_durable_lock_head_refresh(
|
|
existing_lock: dict[str, Any] | None,
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
branch_name: str,
|
|
worktree_path: str,
|
|
pr_number: int | None,
|
|
identity: str | None,
|
|
profile: str | None,
|
|
current_pid: int | None,
|
|
expected_old_head: str | None,
|
|
new_head: str | None,
|
|
base_head: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Fail-closed assessment for refreshing a durable lock's recorded head (#871).
|
|
|
|
A successful ``gitea_update_pr_branch_by_merge`` advances the *remote* PR head
|
|
but must also advance the durable linked-issue lock so a later dead-session
|
|
recovery can prove ownership. This decides whether that refresh is permitted;
|
|
it mutates nothing.
|
|
|
|
Every element of durable ownership is re-verified against the persisted lock —
|
|
repository, issue, branch, worktree, claimant identity/profile, and the live
|
|
owning session — and the recorded head is compare-and-swapped: the lock's
|
|
currently recorded synced head (if any) must equal ``expected_old_head``, so a
|
|
lock whose head or provenance changed concurrently is never overwritten.
|
|
"""
|
|
reasons: list[str] = []
|
|
old = _norm_sha(expected_old_head)
|
|
new = _norm_sha(new_head)
|
|
evidence: dict[str, Any] = {
|
|
"issue_number": issue_number,
|
|
"branch_name": branch_name,
|
|
"worktree_path": worktree_path,
|
|
"pr_number": pr_number,
|
|
"expected_old_head": old,
|
|
"new_head": new,
|
|
"base_head": _norm_sha(base_head),
|
|
}
|
|
|
|
if not isinstance(existing_lock, dict) or not existing_lock:
|
|
reasons.append("no durable lock exists for this issue; nothing to refresh")
|
|
return {"allowed": False, "reasons": reasons, "evidence": evidence,
|
|
"expected_generation": 0}
|
|
|
|
lock = dict(existing_lock)
|
|
evidence["current_generation"] = lock_generation(lock)
|
|
|
|
if lock.get("issue_number") != issue_number:
|
|
reasons.append(
|
|
f"durable lock targets issue #{lock.get('issue_number')}, not "
|
|
f"#{issue_number}; refusing head refresh"
|
|
)
|
|
|
|
for field, expected in (("remote", remote), ("org", org), ("repo", repo)):
|
|
actual = str(lock.get(field) or "").strip()
|
|
if actual != str(expected or "").strip():
|
|
reasons.append(
|
|
f"lock {field} '{actual}' does not match requested "
|
|
f"'{str(expected or '').strip()}'"
|
|
)
|
|
|
|
locked_branch = str(lock.get("branch_name") or "").strip()
|
|
if locked_branch != str(branch_name or "").strip():
|
|
reasons.append(
|
|
f"lock branch '{locked_branch}' does not match requested "
|
|
f"'{str(branch_name or '').strip()}'"
|
|
)
|
|
|
|
locked_worktree = str(lock.get("worktree_path") or "").strip()
|
|
try:
|
|
same_wt = bool(locked_worktree) and bool(worktree_path) and (
|
|
os.path.realpath(locked_worktree) == os.path.realpath(worktree_path)
|
|
)
|
|
except OSError:
|
|
same_wt = locked_worktree == (worktree_path or "")
|
|
if not same_wt:
|
|
reasons.append(
|
|
f"lock worktree '{locked_worktree}' does not match declared "
|
|
f"'{str(worktree_path or '').strip()}'"
|
|
)
|
|
|
|
claimant = _lock_claimant_view(lock)
|
|
locked_identity = str(claimant.get("username") or "").strip()
|
|
locked_profile = str(claimant.get("profile") or "").strip()
|
|
if not locked_identity or not locked_profile:
|
|
reasons.append(
|
|
"durable lock does not record a claimant identity/profile; "
|
|
"ownership could not be proven for head refresh"
|
|
)
|
|
if not str(identity or "").strip() or not str(profile or "").strip():
|
|
reasons.append(
|
|
"active session identity/profile is unknown; ownership could not be "
|
|
"proven for head refresh"
|
|
)
|
|
if locked_identity and str(identity or "").strip() and locked_identity != str(identity).strip():
|
|
reasons.append(
|
|
f"lock claimant '{locked_identity}' does not match active identity "
|
|
f"'{str(identity).strip()}'"
|
|
)
|
|
if locked_profile and str(profile or "").strip() and locked_profile != str(profile).strip():
|
|
reasons.append(
|
|
f"lock profile '{locked_profile}' does not match active profile "
|
|
f"'{str(profile).strip()}'"
|
|
)
|
|
|
|
# The refresh is written by the LIVE owning author session. A refresh is not
|
|
# a recovery: the current process must be the recorded owner.
|
|
recorded_pid = lock.get("session_pid")
|
|
if recorded_pid is None:
|
|
recorded_pid = lock.get("pid")
|
|
evidence["recorded_pid"] = recorded_pid
|
|
evidence["current_pid"] = current_pid
|
|
if current_pid is None:
|
|
reasons.append("current session pid is unknown; cannot prove live ownership")
|
|
else:
|
|
try:
|
|
if recorded_pid is None or int(recorded_pid) != int(current_pid):
|
|
reasons.append(
|
|
f"durable lock is owned by pid {recorded_pid}, not the current "
|
|
f"session pid {current_pid}; head refresh requires the live owner"
|
|
)
|
|
except (TypeError, ValueError):
|
|
reasons.append(
|
|
"durable lock owner pid is malformed; cannot prove live ownership"
|
|
)
|
|
|
|
if not old:
|
|
reasons.append("expected_old_head is not a full 40-char hex SHA (fail closed)")
|
|
if not new:
|
|
reasons.append("new_head is not a full 40-char hex SHA (fail closed)")
|
|
if old and new and old == new:
|
|
reasons.append(
|
|
"new head equals the expected old head; a sync must advance the head"
|
|
)
|
|
|
|
# Compare-and-swap on the recorded head: if the lock already records a synced
|
|
# head it must be exactly the expected old head, else another sync moved it.
|
|
recorded_synced = _norm_sha(lock.get("synced_pr_head"))
|
|
evidence["recorded_synced_pr_head"] = recorded_synced
|
|
if recorded_synced is not None and old is not None and recorded_synced != old:
|
|
reasons.append(
|
|
f"durable lock already records synced head {recorded_synced}, not the "
|
|
f"expected old head {old}; a concurrent sync changed it (CAS fail closed)"
|
|
)
|
|
|
|
if reasons:
|
|
return {"allowed": False, "reasons": reasons, "evidence": evidence,
|
|
"expected_generation": lock_generation(lock)}
|
|
|
|
return {
|
|
"allowed": True,
|
|
"reasons": [
|
|
f"durable lock for issue #{issue_number} branch '{locked_branch}' is "
|
|
f"owned by the live session; refresh recorded head {old} -> {new}"
|
|
],
|
|
"evidence": evidence,
|
|
"expected_generation": lock_generation(lock),
|
|
}
|
|
|
|
|
|
def apply_durable_lock_head_refresh(
|
|
*,
|
|
remote: str,
|
|
org: str,
|
|
repo: str,
|
|
issue_number: int,
|
|
branch_name: str,
|
|
worktree_path: str,
|
|
pr_number: int | None,
|
|
identity: str | None,
|
|
profile: str | None,
|
|
current_pid: int | None,
|
|
expected_old_head: str | None,
|
|
new_head: str | None,
|
|
synced_at: str,
|
|
base_head: str | None = None,
|
|
provenance: str = LOCK_HEAD_REFRESH_PROVENANCE_MERGE_SYNC,
|
|
lock_dir: str | None = None,
|
|
) -> dict[str, Any]:
|
|
"""CAS-refresh the durable lock's recorded head after a branch sync (#871).
|
|
|
|
Reads the durable lock from disk, re-asserts ownership via
|
|
``assess_durable_lock_head_refresh``, and — only when permitted — writes the
|
|
new synced head through ``bind_session_lock`` with a generation compare-and-
|
|
swap. Then re-reads the lock and proves it records the complete new head
|
|
(read-after-write). Any failure at any step returns ``refreshed=False`` with
|
|
reasons; the caller must treat that as a partial lifecycle failure and never
|
|
report a fully successful synchronization.
|
|
"""
|
|
existing = load_issue_lock(
|
|
remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
|
|
)
|
|
assessment = assess_durable_lock_head_refresh(
|
|
existing,
|
|
remote=remote,
|
|
org=org,
|
|
repo=repo,
|
|
issue_number=issue_number,
|
|
branch_name=branch_name,
|
|
worktree_path=worktree_path,
|
|
pr_number=pr_number,
|
|
identity=identity,
|
|
profile=profile,
|
|
current_pid=current_pid,
|
|
expected_old_head=expected_old_head,
|
|
new_head=new_head,
|
|
base_head=base_head,
|
|
)
|
|
result: dict[str, Any] = {
|
|
"refreshed": False,
|
|
"read_after_write_ok": False,
|
|
"prior_head": _norm_sha(expected_old_head),
|
|
"new_head": _norm_sha(new_head),
|
|
"reasons": list(assessment.get("reasons") or []),
|
|
"evidence": assessment.get("evidence"),
|
|
}
|
|
if not assessment.get("allowed"):
|
|
return result
|
|
|
|
new = _norm_sha(new_head)
|
|
old = _norm_sha(expected_old_head)
|
|
record = dict(existing or {})
|
|
sync_block = {
|
|
"last_synced_pr_head": new,
|
|
"prior_pr_head": old,
|
|
"base_head": _norm_sha(base_head),
|
|
"pr_number": pr_number,
|
|
"provenance": provenance,
|
|
"synced_at": synced_at,
|
|
"synced_by_pid": current_pid,
|
|
"synced_by": {
|
|
"username": str(identity or "").strip() or None,
|
|
"profile": str(profile or "").strip() or None,
|
|
},
|
|
}
|
|
record["synced_pr_head"] = new
|
|
record["branch_sync"] = sync_block
|
|
history = record.get("branch_sync_history")
|
|
if not isinstance(history, list):
|
|
history = []
|
|
history = list(history)
|
|
history.append(sync_block)
|
|
record["branch_sync_history"] = history
|
|
|
|
try:
|
|
bind_session_lock(
|
|
record,
|
|
lock_dir=lock_dir,
|
|
expected_generation=assessment.get("expected_generation"),
|
|
renewal_sanctioned=True,
|
|
)
|
|
except Exception as exc: # CAS miss or write failure — partial lifecycle failure
|
|
result["reasons"].append(
|
|
f"durable lock head refresh write failed (fail closed): {exc}"
|
|
)
|
|
return result
|
|
|
|
after = load_issue_lock(
|
|
remote=remote, org=org, repo=repo, issue_number=issue_number, lock_dir=lock_dir
|
|
)
|
|
after_head = _norm_sha((after or {}).get("synced_pr_head"))
|
|
result["lock_generation_after"] = lock_generation(after)
|
|
if after_head == new and new is not None:
|
|
result["refreshed"] = True
|
|
result["read_after_write_ok"] = True
|
|
result["reasons"].append(
|
|
f"durable lock recorded head refreshed to {new} and verified by "
|
|
"read-after-write"
|
|
)
|
|
else:
|
|
result["reasons"].append(
|
|
"read-after-write verification failed: durable lock does not record "
|
|
f"the new head {new} (found {after_head}); partial lifecycle failure"
|
|
)
|
|
return result |