diff --git a/docs/changelog/680.bugfix.rst b/docs/changelog/680.bugfix.rst new file mode 100644 index 00000000..17dec578 --- /dev/null +++ b/docs/changelog/680.bugfix.rst @@ -0,0 +1,2 @@ +A ``SoftFileLease`` acquired on one thread keeps its claim when another thread fails to acquire the same lease object, +so its heartbeat carries on refreshing the marker instead of being torn down and letting a peer take the live claim. diff --git a/src/filelock/_lease.py b/src/filelock/_lease.py index 67248708..51ba6276 100644 --- a/src/filelock/_lease.py +++ b/src/filelock/_lease.py @@ -5,7 +5,7 @@ import time from contextlib import suppress from dataclasses import dataclass -from threading import Event, Thread, current_thread +from threading import Event, Thread, current_thread, local from typing import TYPE_CHECKING, Literal from ._error import LeaseSettingsMismatch @@ -40,6 +40,37 @@ class LeaseCompromise: error: OSError | None = None +@dataclass(frozen=True) +class _Heartbeat: + """A running heartbeat and the event that stops it, which only ever exist together.""" + + thread: Thread + stop: Event + + +@dataclass +class _LeaseClaim: + """The state of one claim: its token, its heartbeat, and how that claim was lost.""" + + token: str | None = None + compromise: LeaseCompromise | None = None + heartbeat: _Heartbeat | None = None + + +class _LeaseClaimHolder: + """Holds the claim its owning context acquired.""" + + # Only the holder is thread-local, mirroring FileLockContext: the claim stays an ordinary object, so the heartbeat + # thread records a compromise where the thread that acquired the lease reads it. + + def __init__(self) -> 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. @@ -112,10 +143,15 @@ def __init__( self._lease_duration = lease_duration self._heartbeat_interval = heartbeat_interval self._on_compromise = on_compromise - self._token: str | None = None - self._compromise: LeaseCompromise | None = None - self._heartbeat_stop = Event() - self._heartbeat: Thread | None = None + # 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: @@ -130,7 +166,7 @@ def token(self) -> str | None: :returns: the token while the lease is held, ``None`` otherwise. It identifies a claim; it does not fence one. """ - return self._token + return self._claim.token @property def compromise(self) -> LeaseCompromise | None: @@ -140,27 +176,28 @@ def compromise(self) -> LeaseCompromise | None: :returns: the :class:`LeaseCompromise`, or ``None`` while the claim still holds """ - return self._compromise + 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 - self._token = secrets.token_hex(16) - self._compromise = None + 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. Hand it the fd and the inode it verified instead. + # 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(fd, identity, self._token) + self._start_heartbeat(claim, fd, identity, token) def _release(self) -> None: self._stop_heartbeat() - self._token = None + self._claim.token = None super()._release() def _published_record(self) -> OwnerRecord: - return super()._published_record()._replace(token=self._token, lease_duration=self._lease_duration) + 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: @@ -199,53 +236,70 @@ def _read_peer(self) -> tuple[OwnerRecord, float, int] | None: return owner, mtime, ino return None - def _start_heartbeat(self, fd: int, identity: tuple[int, int], token: str) -> None: - self._heartbeat_stop = Event() - self._heartbeat = Thread( + 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=(fd, identity, token), + args=(claim, fd, identity, token, stop), name=f"filelock-lease-{os.getpid()}", daemon=True, ) - self._heartbeat.start() + claim.heartbeat = _Heartbeat(thread, stop) + thread.start() def _stop_heartbeat(self) -> None: - if (heartbeat := self._heartbeat) is None: + claim = self._claim + if (heartbeat := claim.heartbeat) is None: return - self._heartbeat_stop.set() - self._heartbeat = None + heartbeat.stop.set() + claim.heartbeat = None # on_compromise runs on the heartbeat thread and may release the lease, which lands back here. - if heartbeat is not current_thread(): - heartbeat.join(timeout=self._heartbeat_interval) + if heartbeat.thread is not current_thread(): + heartbeat.thread.join(timeout=self._heartbeat_interval) - def _refresh_until_stopped(self, fd: int, identity: tuple[int, int], token: str) -> None: + 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 self._heartbeat_stop.wait(self._heartbeat_interval): - outcome, error = self._refresh_claim(fd, identity, token) + 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("refresh-failed", error, token) + self._report_compromise(claim, "refresh-failed", error, token) return - def _refresh_claim(self, fd: int, identity: tuple[int, int], token: str) -> tuple[_RefreshOutcome, OSError | None]: + 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("marker-missing", error, token) + 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("owner-changed", None, token) + self._report_compromise(claim, "owner-changed", None, token) return "lost", None try: touch(self.lock_file, fd=fd) @@ -253,11 +307,19 @@ def _refresh_claim(self, fd: int, identity: tuple[int, int], token: str) -> tupl return "transient", error return "ok", None - def _report_compromise(self, reason: CompromiseReason, error: OSError | None, token: str) -> None: - # The token is the one this thread published, not self._token, which a release may already have cleared. - self._compromise = LeaseCompromise(lock_file=self.lock_file, token=token, reason=reason, error=error) + 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(self._compromise) + self._on_compromise(claim.compromise) __all__ = [ diff --git a/tests/test_soft_lease.py b/tests/test_soft_lease.py index 5ae9f9b1..3a82bc3a 100644 --- a/tests/test_soft_lease.py +++ b/tests/test_soft_lease.py @@ -4,7 +4,9 @@ import os import socket import time +from contextlib import suppress from errno import EIO, ENOENT +from threading import Thread from types import SimpleNamespace from typing import TYPE_CHECKING, Final, cast @@ -318,6 +320,27 @@ def release_the_claim(_: LeaseCompromise) -> None: assert not lease.is_locked +def test_lease_keeps_its_claim_when_another_thread_fails_to_acquire(marker: Path) -> None: + # The context is thread-local, so a second thread contending on the same lease object must leave the holder's + # claim alone: a torn-down heartbeat stops refreshing the marker and a peer takes it while the holder still holds. + holder = _lease(marker) + + def contend() -> None: + with suppress(Timeout): + holder.acquire() + + with holder: + token = holder.token + contender = Thread(target=contend) + contender.start() + contender.join() + + assert holder.token == token + time.sleep(_DURATION * 1.5) # only a surviving heartbeat keeps the claim past this + with pytest.raises(Timeout): + _lease(marker).acquire() + + def test_lease_rejects_a_peer_configured_with_another_duration(marker: Path) -> None: holder = _lease(marker)