from __future__ import annotations import os import secrets import time from contextlib import suppress from dataclasses import dataclass from threading import Event, Thread, current_thread, local from typing import TYPE_CHECKING, Literal from ._error import LeaseSettingsMismatch from ._identity import owner_is_stale from ._marker import MarkerSoftFileLock, OwnerMode, OwnerRecord, parse_marker from ._soft import _read_lock_file from ._util import break_lock_file, touch if TYPE_CHECKING: import sys from collections.abc import Callable from ._api import LockOptions if sys.version_info >= (3, 11): # pragma: no cover (py311+) from typing import Unpack else: # pragma: no cover ( None: self.claim = _LeaseClaim() class _ThreadLocalLeaseClaimHolder(_LeaseClaimHolder, local): """A thread local version of the ``_LeaseClaimHolder`` class.""" class SoftFileLease(MarkerSoftFileLock): """ Existence lock whose claim expires, so a peer may take it while the previous holder still runs. A lease trades mutual exclusion for progress. The holder publishes a claim and refreshes it every ``heartbeat_interval`` seconds; a contender takes the marker once it is ``lease_duration`` seconds stale. Nothing stops the expired holder: it keeps running, and it keeps using whatever the lock protects. Treat the lease as a hint about who *should* be working, not as a guarantee that only one worker is. To make a protected resource reject a superseded holder, that resource must be linearizable and must fence on a monotonic generation it controls. :attr:`token` names a claim; it does not fence one. Where overlap is unacceptable, use :class:`StrictSoftFileLock ` instead. Every contender for a path must agree on ``lease_duration``. A contender that finds a claim published under a different duration raises :class:`LeaseSettingsMismatch ` rather than apply its own expiry to a peer that never agreed to it. Expiry reclaims less on Windows, which refuses to rename or delete a file another process holds open. A peer there takes an expired claim only once the previous holder's process exits and its handle closes; a holder that lives on but stops refreshing keeps the marker. Unix reclaims the marker either way. ``on_compromise`` fires from the heartbeat thread when a refresh fails, or when the marker vanishes or names another owner. The holder should stop touching the protected resource when it runs. Because it runs on that thread, a ``release()`` inside it only takes effect when the lease was built with ``thread_local=False``; the default thread-local context hides the claim from every thread but the one that acquired it, so the release does nothing. Signal the acquiring thread instead when the context stays thread-local. .. versionadded:: 3.30.0 """ _owner_mode: OwnerMode = "lease" #: lease_duration replaces the legacy age-based lifetime, so accepting both would give one lock two expiry clocks. _lifetime_supported: bool = False _lifetime_unsupported_reason: str = "lease_duration sets when a lease expires" def __init__( self, lock_file: str | os.PathLike[str], *, lease_duration: float = 30.0, heartbeat_interval: float | None = None, on_compromise: Callable[[LeaseCompromise], None] | None = None, **kwargs: Unpack[LockOptions], ) -> None: """ Create a lease. :param lease_duration: seconds of marker staleness after which a contender may take the claim. Every contender for the path must pass the same value. :param heartbeat_interval: seconds between refreshes. Defaults to a third of ``lease_duration``, leaving room for two missed refreshes before a peer may take the claim. Must be shorter than ``lease_duration``. :param on_compromise: called from the heartbeat thread with a :class:`LeaseCompromise` when the claim is lost. :param kwargs: every other :class:`BaseFileLock ` option, ``timeout`` and ``mode`` among them. The metaclass passes them all by keyword, and taking them here lets :class:`AsyncSoftFileLease ` add the async plumbing a fixed signature would hide. """ if lease_duration <= 0: msg = f"lease_duration must be positive, got {lease_duration!r}" raise ValueError(msg) if heartbeat_interval is None: heartbeat_interval = lease_duration / 3 if not 0 < heartbeat_interval < lease_duration: msg = f"heartbeat_interval must be positive and below lease_duration, got {heartbeat_interval!r}" raise ValueError(msg) super().__init__(lock_file, **kwargs) self._lease_duration = lease_duration self._heartbeat_interval = heartbeat_interval self._on_compromise = on_compromise # Sharing one claim across a thread-local lock lets a second thread's failed acquisition stop the heartbeat of # the thread holding the lease, leaving its marker unrefreshed until a peer reclaims it. self._claims: _LeaseClaimHolder = ( _ThreadLocalLeaseClaimHolder if self.is_thread_local() else _LeaseClaimHolder )() @property def _claim(self) -> _LeaseClaim: return self._claims.claim @property def lease_duration(self) -> float: """The staleness in seconds after which a contender may take this claim.""" return self._lease_duration @property def token(self) -> str | None: """ The token naming the claim this process published. :returns: the token while the lease is held, ``None`` otherwise. It identifies a claim; it does not fence one. """ return self._claim.token @property def compromise(self) -> LeaseCompromise | None: """ The loss of claim the heartbeat observed. :returns: the :class:`LeaseCompromise`, or ``None`` while the claim still holds """ return self._claim.compromise def _acquire(self) -> None: claim = self._claim self._stop_heartbeat() # no earlier claim's heartbeat outlives the acquisition of the next one claim.token = token = secrets.token_hex(16) claim.compromise = None super()._acquire() # The context is thread-local by default, so the heartbeat thread cannot read the descriptor this one just # published, nor the claim this one owns. Hand it the fd, the inode it verified and the claim instead. if (fd := self._context.lock_file_fd) is not None and ( identity := self._context.lock_file_fd_identity ) is not None: self._start_heartbeat(claim, fd, identity, token) def _release(self) -> None: self._stop_heartbeat() self._claim.token = None super()._release() def _published_record(self) -> OwnerRecord: return super()._published_record()._replace(token=self._claim.token, lease_duration=self._lease_duration) def _try_break_stale_lock(self) -> None: if (peer := self._read_peer()) is None: # Not a readable protocol 2 lease record: a partial write, a foreign or legacy protocol 1 marker, or the # strict sentinel. The base self-heal evicts a genuinely malformed marker once it ages past the grace # window and leaves a legitimate legacy or strict holder in place, so a corrupt marker no longer wedges # every lease contender until its own timeout. super()._try_break_stale_lock() return owner, mtime, ino = peer # Only a peer that published a lease agreed to be superseded by one, so a record stating any other contract is # never reclaimed by age. Raise the mismatch outside the read so the suppression cannot swallow it. if owner.mode != "lease": return if owner.lease_duration != self._lease_duration: msg = ( f"{self.lock_file} holds a lease of {owner.lease_duration!r}s but this contender configured " f"{self._lease_duration!r}s; every contender for a path must agree on lease_duration" ) raise LeaseSettingsMismatch(msg) # A break can fail for reasons a contender must ride out rather than raise on: a peer broke the marker first, # or Windows refuses to rename a file whose holder still has it open. Poll again instead. with suppress(OSError): # A dead or recycled owner is reclaimed at once; a live owner past its lease duration is superseded on the # schedule every contender agreed to. if owner_is_stale(owner.pid, owner.hostname, owner.start): break_lock_file(self.lock_file, mtime, ino) return if time.time() - mtime >= self._lease_duration: break_lock_file(self.lock_file, mtime, ino) def _read_peer(self) -> tuple[OwnerRecord, float, int] | None: with suppress(OSError, ValueError): content, mtime, ino = _read_lock_file(self.lock_file) if (owner := parse_marker(content)) is not None: return owner, mtime, ino return None def _start_heartbeat(self, claim: _LeaseClaim, fd: int, identity: tuple[int, int], token: str) -> None: # The thread watches the event it was handed rather than whatever the claim names later: a heartbeat that # outlives its join timeout would otherwise adopt the next acquisition's event and never stop. stop = Event() thread = Thread( target=self._refresh_until_stopped, args=(claim, fd, identity, token, stop), name=f"filelock-lease-{os.getpid()}", daemon=True, ) # Record the heartbeat before starting the thread so a release racing this acquire on a shared, # non-thread-local claim always sees it and sets the stop event; the thread then exits at its first wait # instead of outliving the release. A start that raises leaves the unstarted thread for _stop_heartbeat. claim.heartbeat = _Heartbeat(thread, stop) thread.start() def _stop_heartbeat(self) -> None: claim = self._claim if (heartbeat := claim.heartbeat) is None: return heartbeat.stop.set() claim.heartbeat = None # thread.ident is None until start() runs: a heartbeat recorded before its thread started (a start that # raised, or a release racing acquire on a shared claim) has nothing to join, and the stop above makes it # exit at once. on_compromise runs on the heartbeat thread and may release the lease, landing back here. if heartbeat.thread.ident is not None and heartbeat.thread is not current_thread(): heartbeat.thread.join(timeout=self._heartbeat_interval) def _refresh_until_stopped( self, claim: _LeaseClaim, fd: int, identity: tuple[int, int], token: str, stop: Event, ) -> None: # The loop ends at the first loss of the claim, so the holder hears about it once. A transient filesystem # error (ESTALE / EIO on the NFS-style filesystems a lease targets) is not a loss: retry rather than raise a # false compromise. Report the claim unrefreshable only once failures have run long enough that a contender # could take it before the next success would land, a margin before the marker actually ages out, the way # restic declares a lock unrefreshable ahead of its stale time. last_success = time.monotonic() while not stop.wait(self._heartbeat_interval): outcome, error = self._refresh_claim(claim, fd, identity, token) if outcome == "lost": return if outcome == "ok": last_success = time.monotonic() elif time.monotonic() - last_success >= self._lease_duration - self._heartbeat_interval: self._report_compromise(claim, "refresh-failed", error, token) return def _refresh_claim( self, claim: _LeaseClaim, fd: int, identity: tuple[int, int], token: str, ) -> tuple[_RefreshOutcome, OSError | None]: try: st = os.lstat(self.lock_file) except FileNotFoundError as error: self._report_compromise(claim, "marker-missing", error, token) return "lost", None except OSError as error: return "transient", error # A peer that took the expired claim replaced the marker, so the pathname now names its inode, not ours. if (st.st_dev, st.st_ino) != identity: self._report_compromise(claim, "owner-changed", None, token) return "lost", None try: touch(self.lock_file, fd=fd) except OSError as error: return "transient", error return "ok", None def _report_compromise( self, claim: _LeaseClaim, reason: CompromiseReason, error: OSError | None, token: str, ) -> None: # Record it on the claim this heartbeat serves, not on self._claim: a thread-local claim read from the # heartbeat thread is a different, empty one, so the holder would never see the loss it is being told about. # The token is the one this thread published, not claim.token, which a release may already have cleared. claim.compromise = LeaseCompromise(lock_file=self.lock_file, token=token, reason=reason, error=error) if self._on_compromise is not None: self._on_compromise(claim.compromise) __all__ = [ "CompromiseReason", "LeaseCompromise", "SoftFileLease", ]