sched: add held cross-process ownership with real-process tests (T11)

src/wf_scheduling/ownership.py

@@ -0,0 +1,123 @@
+"""Exclusive file-store ownership via a held cross-process lock (T11).
+
+The scheduler process owns the store exclusively before recovery. Ownership
+is a held lock on a dedicated lock file (not a stale PID file or an
+expiring lease): the handle stays open for the owner's lifetime and the OS
+releases it on process death. A second owner attempting acquisition while
+the lock is held is rejected. Where locking is unsupported, scheduler
+startup is rejected instead of running unprotected.
+
+Locking design (Windows-tested, per the store transaction boundary): one
+new module holding ``msvcrt.locking`` (Windows) / ``fcntl.flock`` (POSIX)
+on ``<store root>/scheduler.lock``. No ad-hoc per-run lock files.
+"""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import BinaryIO
+
+
+class SecondOwnerError(Exception):
+    """Another process already owns the schedule store."""
+
+
+class StartupRejected(Exception):
+    """Scheduler startup rejected where locking is unsupported."""
+
+
+class SchedulerOwnership:
+    """Held exclusive ownership of a schedule store root."""
+
+    def __init__(self, root: Path, *, owner: str) -> None:
+        self.root = root
+        self.owner = owner
+        self._handle: BinaryIO | None = None
+        self._locked = False
+
+    @property
+    def lock_path(self) -> Path:
+        return self.root / "scheduler.lock"
+
+    def acquire(self) -> SchedulerOwnership:
+        """Acquire the held lock non-blockingly or raise SecondOwnerError."""
+        self.lock_path.parent.mkdir(parents=True, exist_ok=True)
+        try:
+            handle = open(self.lock_path, "a+b")
+        except OSError as exc:
+            raise StartupRejected(f"cannot open scheduler lock: {exc}") from exc
+        try:
+            _lock_nonblocking(handle)
+        except SecondOwnerError:
+            handle.close()
+            raise
+        except StartupRejected:
+            handle.close()
+            raise
+        except OSError as exc:
+            handle.close()
+            raise StartupRejected(f"unsupported scheduler locking: {exc}") from exc
+        self._handle = handle
+        self._locked = True
+        return self
+
+    def release(self) -> None:
+        """Release the held lock (idempotent)."""
+        if not self._locked or self._handle is None:
+            return
+        try:
+            _unlock(self._handle)
+        finally:
+            try:
+                self._handle.close()
+            finally:
+                self._handle = None
+                self._locked = False
+
+    def __enter__(self) -> SchedulerOwnership:
+        return self.acquire()
+
+    def __exit__(self, *exc: object) -> None:
+        self.release()
+
+
+def _lock_nonblocking(handle: BinaryIO) -> None:
+    import os
+
+    if os.name == "nt":
+        import msvcrt
+
+        handle.seek(0)
+        try:
+            msvcrt.locking(handle.fileno(), msvcrt.LK_NBLCK, 1)
+        except OSError as exc:
+            raise SecondOwnerError("schedule store owned by another process") from exc
+        return
+    try:
+        import fcntl
+    except ImportError as exc:
+        raise StartupRejected("file locking unsupported on this platform") from exc
+    try:
+        fcntl.flock(handle.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError as exc:
+        raise SecondOwnerError("schedule store owned by another process") from exc
+
+
+def _unlock(handle: BinaryIO) -> None:
+    import os
+
+    if os.name == "nt":
+        import msvcrt
+
+        try:
+            handle.seek(0)
+            msvcrt.locking(handle.fileno(), msvcrt.LK_UNLCK, 1)
+        except OSError:
+            pass
+        return
+    try:
+        import fcntl
+
+        fcntl.flock(handle.fileno(), fcntl.LOCK_UN)
+    except OSError, ImportError:
+        pass