lock.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/regenerate/lock.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/regenerate/test_lock.py"
"""
The single-worktree ``regenerate`` file lock ([OP-7], §5.3, Plan 142i R5).
 
Purpose
-------
Serialize concurrent ``regenerate`` runs in one worktree with an atomic
(``O_CREAT|O_EXCL``) lock file carrying the holder's **PID + process-start-time +
acquire-timestamp**. A second live holder fails closed with ``REGEN_LOCK_HELD``.
A lock is reclaimed **only on provable holder death** — a **dead PID** or a
**process-start-time mismatch** (PID reuse) — never on elapsed time ([OP-7]: a
renewal-less file lock cannot tell a slow-but-alive holder from a dead one, so a
staleness auto-reclaim would let a second run steal a live lock and corrupt the
swap). A **live-or-inconclusive** holder is never auto-reclaimed; operator
``--force`` is the sole override and every reclamation is surfaced as a typed
``LockReclaimEvent`` (never an unobserved log line).
 
Responsibilities
----------------
- ``acquire(lock_path, *, force, clock, probe)`` — atomically take the lock or
  fail closed / reclaim-on-provable-death; return a ``LockHandle``.
- ``release(handle)`` — drop our own lock.
- ``LockReclaimEvent`` — the typed, assertable reclamation record.
 
Testability
-----------
``acquire`` takes injectable **clock** + **process-probe** seams so the tests
cover live-PID / dead-PID / start-time-mismatch / live-but-slow / malformed-lock
/ ``--force`` paths without real sleeps, real PID reuse, or live-PID races.
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - acquire/release lifecycle
    L3 - reclaim classification
    L4 - liveness-probe details
 
Contracts
---------
- Atomic create — ``O_CREAT|O_EXCL`` so the lock file has no create race.
- Reclaim ⇔ provable death (dead PID or start-time mismatch) OR ``force``.
- ``force`` reclaims but NEVER skips validation/canonicalisation/derive-verify
  (that is the pipeline's job); it only reclaims the lock and logs the event.
- Every probe FAILURE ⇒ inconclusive ⇒ fail-closed (never mapped to "dead").
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
- oc_chat.registry.regenerate.envelope (REGEN_LOCK_HELD / RegenerateError)
- oc_chat.registry.clock (RealClock wall time for the default acquire timestamp)
"""
from __future__ import annotations
 
import json
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import Callable, Optional
 
from oc_diagnostics import _dbg
 
from oc_chat.registry.clock import RealClock
from oc_chat.registry.regenerate.envelope import REGEN_LOCK_HELD, RegenerateError
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
# Liveness-probe outcomes (only DEAD is a death proof; ALIVE + start-time
# mismatch is the other; every failure maps to INCONCLUSIVE, never DEAD — H8).
PROBE_DEAD = "dead"
PROBE_ALIVE = "alive"
PROBE_INCONCLUSIVE = "inconclusive"
 
# Reclaim reasons (the LockReclaimEvent.reason closed set).
REASON_DEAD_PID = "dead_pid"
REASON_START_TIME_MISMATCH = "start_time_mismatch"
REASON_FORCED = "forced_live_or_inconclusive"
 
#: A clock seam — returns an ISO-8601 wall-clock string (not hash-contributing).
Clock = Callable[[], str]
#: A process-probe seam — ``pid -> (outcome, current_process_start_time|None)``.
Probe = Callable[[int], "tuple[str, Optional[str]]"]
 
 
@dataclass(frozen=True)
class LockReclaimEvent:
    """A typed record of one lock reclamation — the contract surface tests assert
    (surfaced canonically as ``RegenerateResult.lock_events``)."""
 
    lock_path: str
    pid: Optional[int]
    process_start_time: Optional[str]
    acquired_at: Optional[str]
    reason: str
    force: bool
    timestamp: str
 
    def to_dict(self) -> dict[str, object]:
        return {
            "lock_path": self.lock_path,
            "pid": self.pid,
            "process_start_time": self.process_start_time,
            "acquired_at": self.acquired_at,
            "reason": self.reason,
            "force": self.force,
            "timestamp": self.timestamp,
        }
 
 
@dataclass(frozen=True)
class LockHandle:
    """A held lock — pass to ``release``; carries the reclaim event if we stole it."""
 
    lock_path: Path
    pid: int
    reclaim_event: Optional[LockReclaimEvent] = None
 
 
@dataclass(frozen=True)
class _LockRecord:
    pid: int
    process_start_time: str
    acquired_at: str
 
 
def _default_clock() -> str:
    # Wall-clock via the central registry clock (clock-discipline: registry code
    # must not import time/datetime directly). The timestamp is runtime lock
    # metadata — never hash-contributing.
    return RealClock().now_wall()
 
 
# --- liveness probe (no new dependency: ctypes on Windows, /proc on POSIX) ---
def _probe_posix(pid: int) -> "tuple[str, Optional[str]]":
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return (PROBE_DEAD, None)  # ESRCH — provably no such process
    except PermissionError:
        return (PROBE_ALIVE, _proc_start_time_posix(pid))  # EPERM — exists (alive)
    except OSError:
        return (PROBE_INCONCLUSIVE, None)  # any other failure — never "dead"
    return (PROBE_ALIVE, _proc_start_time_posix(pid))
 
 
def _proc_start_time_posix(pid: int) -> Optional[str]:
    try:
        stat = Path(f"/proc/{pid}/stat").read_text(encoding="utf-8")
        tail = stat[stat.rfind(")") + 2:].split()  # skip pid + (comm)
        return tail[19]  # field 22 (starttime) = tail index 19 (after state@0)
    except (OSError, ValueError, IndexError):
        return None
 
 
def _probe_windows(pid: int) -> "tuple[str, Optional[str]]":
    import ctypes
    from ctypes import wintypes
 
    PROCESS_QUERY_LIMITED_INFORMATION = 0x1000
    ERROR_INVALID_PARAMETER = 87  # OpenProcess on a nonexistent PID -> dead
    kernel32 = ctypes.WinDLL("kernel32", use_last_error=True)
    kernel32.OpenProcess.restype = wintypes.HANDLE
    kernel32.OpenProcess.argtypes = [wintypes.DWORD, wintypes.BOOL, wintypes.DWORD]
    kernel32.CloseHandle.argtypes = [wintypes.HANDLE]
    kernel32.GetProcessTimes.argtypes = [
        wintypes.HANDLE,
        ctypes.POINTER(wintypes.FILETIME),
        ctypes.POINTER(wintypes.FILETIME),
        ctypes.POINTER(wintypes.FILETIME),
        ctypes.POINTER(wintypes.FILETIME),
    ]
    handle = kernel32.OpenProcess(PROCESS_QUERY_LIMITED_INFORMATION, False, pid)
    if not handle:
        err = ctypes.get_last_error()
        if err == ERROR_INVALID_PARAMETER:
            return (PROBE_DEAD, None)
        return (PROBE_INCONCLUSIVE, None)  # ACCESS_DENIED etc. — never "dead"
    try:
        creation = wintypes.FILETIME()
        exit_t = wintypes.FILETIME()
        kernel_t = wintypes.FILETIME()
        user_t = wintypes.FILETIME()
        ok = kernel32.GetProcessTimes(
            handle, ctypes.byref(creation), ctypes.byref(exit_t),
            ctypes.byref(kernel_t), ctypes.byref(user_t),
        )
        if not ok:
            return (PROBE_INCONCLUSIVE, None)
        return (PROBE_ALIVE, f"{creation.dwHighDateTime}:{creation.dwLowDateTime}")
    finally:
        kernel32.CloseHandle(handle)
 
 
def _default_probe(pid: int) -> "tuple[str, Optional[str]]":
    if sys.platform == "win32":
        return _probe_windows(pid)
    return _probe_posix(pid)
 
 
def _read_record(lock_path: Path) -> Optional[_LockRecord]:
    """Parse the lock record, or ``None`` if missing/unreadable/malformed (H6)."""
    try:
        data = json.loads(lock_path.read_text(encoding="utf-8"))
        return _LockRecord(
            pid=int(data["pid"]),
            process_start_time=str(data["process_start_time"]),
            acquired_at=str(data["acquired_at"]),
        )
    except (OSError, ValueError, KeyError, TypeError):
        return None
 
 
def _atomic_create(lock_path: Path, pid: int, start_time: str, acquired_at: str) -> None:
    """Create the lock file atomically (``O_CREAT|O_EXCL``) — raises
    ``FileExistsError`` if a lock already exists (the no-race create)."""
    payload = json.dumps(
        {"pid": pid, "process_start_time": start_time, "acquired_at": acquired_at},
        sort_keys=True,
    )
    fd = os.open(str(lock_path), os.O_CREAT | os.O_EXCL | os.O_WRONLY, 0o644)
    try:
        os.write(fd, payload.encode("utf-8"))
    finally:
        os.close(fd)
 
 
def _make_event(
    lock_path: Path, record: Optional[_LockRecord], reason: str, *, force: bool, clock: Clock
) -> LockReclaimEvent:
    return LockReclaimEvent(
        lock_path=str(lock_path),
        pid=record.pid if record is not None else None,
        process_start_time=record.process_start_time if record is not None else None,
        acquired_at=record.acquired_at if record is not None else None,
        reason=reason,
        force=force,
        timestamp=clock(),
    )
 
 
def _decide_reclaim(
    lock_path: Path, record: Optional[_LockRecord], force: bool, probe: Probe, clock: Clock
) -> LockReclaimEvent:
    """Return a ``LockReclaimEvent`` if reclaimable; else raise ``REGEN_LOCK_HELD``.
 
    Reclaimable ⇔ provable holder death (dead PID or start-time mismatch), OR
    ``force`` on a live/inconclusive holder. A malformed record (``None``) is
    inconclusive — never auto-reclaimed.
    """
    reason: Optional[str] = None
    if record is not None:
        outcome, current_start = probe(record.pid)
        if outcome == PROBE_DEAD:
            reason = REASON_DEAD_PID
        elif (
            outcome == PROBE_ALIVE
            and current_start
            and record.process_start_time
            and current_start != record.process_start_time
        ):
            reason = REASON_START_TIME_MISMATCH  # PID reuse — the holder is gone
    if reason is not None:
        return _make_event(lock_path, record, reason, force=False, clock=clock)
    if force:
        return _make_event(lock_path, record, REASON_FORCED, force=True, clock=clock)
    raise RegenerateError(
        f"regenerate lock held by a live-or-inconclusive process at {lock_path}",
        code=REGEN_LOCK_HELD,
        hint="another regenerate is running; wait for it, or pass --force to reclaim (logged)",
    )
 
 
def acquire(
    lock_path: Path,
    *,
    force: bool = False,
    clock: Optional[Clock] = None,
    probe: Optional[Probe] = None,
) -> LockHandle:
    """Acquire the regenerate lock at ``lock_path``.
 
    Raises ``REGEN_LOCK_HELD`` when the lock is held by a live-or-inconclusive
    process and ``force`` is false. Creates ``lock_path.parent`` first (H5).
    """
    func = "acquire"
    lock_path = Path(lock_path)
    clock = clock or _default_clock
    probe = probe or _default_probe
    lock_path.parent.mkdir(parents=True, exist_ok=True)  # H5
    self_pid = os.getpid()
    self_start = probe(self_pid)[1] or ""
    acquired_at = clock()
    try:
        _atomic_create(lock_path, self_pid, self_start, acquired_at)
        return LockHandle(lock_path, self_pid, None)
    except FileExistsError:
        pass
    # The lock exists — classify the holder and reclaim only if provably dead
    # (or forced). A raise here (REGEN_LOCK_HELD) propagates unchanged.
    record = _read_record(lock_path)
    event = _decide_reclaim(lock_path, record, force, probe, clock)
    lock_path.unlink(missing_ok=True)
    try:
        _atomic_create(lock_path, self_pid, self_start, acquired_at)
    except FileExistsError as exc:
        raise RegenerateError(
            "regenerate lock re-acquired by another process during reclamation",
            code=REGEN_LOCK_HELD,
            hint="a concurrent regenerate won the race; retry",
        ) from exc
    _dbg("OC_CHAT.MAIN", func, f"reclaimed lock ({event.reason})", 2)
    return LockHandle(lock_path, self_pid, event)
 
 
def release(handle: LockHandle) -> None:
    """Release a held lock (best-effort; only removes our own lock record)."""
    try:
        record = _read_record(handle.lock_path)
        if record is None or record.pid == handle.pid:
            handle.lock_path.unlink(missing_ok=True)
    except OSError:
        pass
 
 
__all__ = [
    "PROBE_DEAD",
    "PROBE_ALIVE",
    "PROBE_INCONCLUSIVE",
    "REASON_DEAD_PID",
    "REASON_START_TIME_MISMATCH",
    "REASON_FORCED",
    "LockReclaimEvent",
    "LockHandle",
    "acquire",
    "release",
]