From e213cb90315c6d3f187b49c72fd4478d05a2585f Mon Sep 17 00:00:00 2001 From: Javid Khan Date: Sat, 18 Jul 2026 21:34:45 +0530 Subject: [PATCH 1/3] scope a lease's claim to the context that acquired it --- src/filelock/_lease.py | 121 ++++++++++++++++++++++++++++----------- tests/test_soft_lease.py | 23 ++++++++ 2 files changed, 112 insertions(+), 32 deletions(-) diff --git a/src/filelock/_lease.py b/src/filelock/_lease.py index 67248708..86a1d457 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,31 @@ class LeaseCompromise: error: OSError | None = None +@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: Thread | None = None + heartbeat_stop: Event | None = None + + +class _LeaseClaimHolder: + """Holds the claim the owning context currently has, so the claim itself stays readable from any thread.""" + + # A separate class so _ThreadLocalLeaseClaimHolder can make the holder thread-local, mirroring FileLockContext. + # Only the holder is thread-local: 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 +137,16 @@ 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 + # A claim belongs to whichever context acquired it, so it is thread-local exactly when that context is. + # Sharing one claim across a thread-local lock lets a second thread's acquisition attempt stop the heartbeat + # of the thread that actually holds 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 +161,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 +171,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 +231,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, not whatever claim.heartbeat_stop names later: a heartbeat that + # outlives its join timeout would otherwise adopt the next acquisition's event and never stop. + claim.heartbeat_stop = stop = Event() + claim.heartbeat = 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.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 + if claim.heartbeat_stop is not None: + claim.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) - 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 +302,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) From 6e5c3f8d050f6604d8ddb45d802bef573c029f8a Mon Sep 17 00:00:00 2001 From: Javid Khan Date: Sat, 18 Jul 2026 21:35:13 +0530 Subject: [PATCH 2/3] add changelog entry --- docs/changelog/680.bugfix.rst | 2 ++ 1 file changed, 2 insertions(+) create mode 100644 docs/changelog/680.bugfix.rst 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. From 0fc1465142fcf6d0107adb8feef7bc7422786432 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Bern=C3=A1t=20G=C3=A1bor?= Date: Sun, 19 Jul 2026 16:45:38 -0700 Subject: [PATCH 3/3] =?UTF-8?q?=E2=99=BB=EF=B8=8F=20refactor(lease):=20pai?= =?UTF-8?q?r=20a=20heartbeat=20with=20the=20event=20that=20stops=20it?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The thread and its stop event were separate optionals though start sets both and stop clears both, so stopping had to guard an event that cannot be missing while the thread is running. That arc never runs and the matrix requires every branch. Hold the two together instead, which states the invariant in the type and drops the guard. --- src/filelock/_lease.py | 39 ++++++++++++++++++++++----------------- 1 file changed, 22 insertions(+), 17 deletions(-) diff --git a/src/filelock/_lease.py b/src/filelock/_lease.py index 86a1d457..51ba6276 100644 --- a/src/filelock/_lease.py +++ b/src/filelock/_lease.py @@ -40,22 +40,28 @@ 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: Thread | None = None - heartbeat_stop: Event | None = None + heartbeat: _Heartbeat | None = None class _LeaseClaimHolder: - """Holds the claim the owning context currently has, so the claim itself stays readable from any thread.""" + """Holds the claim its owning context acquired.""" - # A separate class so _ThreadLocalLeaseClaimHolder can make the holder thread-local, mirroring FileLockContext. - # Only the holder is thread-local: the claim stays an ordinary object, so the heartbeat thread records a - # compromise where the thread that acquired the lease reads it. + # 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() @@ -137,9 +143,8 @@ def __init__( self._lease_duration = lease_duration self._heartbeat_interval = heartbeat_interval self._on_compromise = on_compromise - # A claim belongs to whichever context acquired it, so it is thread-local exactly when that context is. - # Sharing one claim across a thread-local lock lets a second thread's acquisition attempt stop the heartbeat - # of the thread that actually holds the lease, leaving its marker unrefreshed until a peer reclaims it. + # 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 )() @@ -232,27 +237,27 @@ def _read_peer(self) -> tuple[OwnerRecord, float, int] | None: 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, not whatever claim.heartbeat_stop names later: a heartbeat that + # 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. - claim.heartbeat_stop = stop = Event() - claim.heartbeat = Thread( + stop = Event() + thread = Thread( target=self._refresh_until_stopped, args=(claim, fd, identity, token, stop), name=f"filelock-lease-{os.getpid()}", daemon=True, ) - claim.heartbeat.start() + claim.heartbeat = _Heartbeat(thread, stop) + thread.start() def _stop_heartbeat(self) -> None: claim = self._claim if (heartbeat := claim.heartbeat) is None: return - if claim.heartbeat_stop is not None: - claim.heartbeat_stop.set() + 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,