Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions docs/changelog/661.bugfix.rst
Original file line number Diff line number Diff line change
@@ -0,0 +1,2 @@
``SoftReadWriteLock`` and ``SoftFileLease`` ride out a transient filesystem error during a heartbeat instead of dropping the lock or reporting a false compromise, and report a lost claim only once refresh cannot recover before the lease would lapse.
``SoftFileLease`` also reclaims a malformed or partial marker that used to block every contender, and a Linux marker distinguishes a PID reused across a reboot.
29 changes: 21 additions & 8 deletions src/filelock/_identity.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,6 @@
import os
import socket
import sys
from contextlib import suppress
from errno import EPERM, ESRCH
from pathlib import Path
from typing import Final
Expand Down Expand Up @@ -103,21 +102,35 @@ def process_alive(pid: int) -> bool:
# after the final ')'. starttime is field 22 overall, the twentieth of those trailing fields (index 19).
_STARTTIME_INDEX: Final[int] = 19

def _read_boot_id() -> int:
# starttime is measured in clock ticks since boot, so on its own it repeats across a reboot. Folding the
# boot id into the high bits makes the Linux token reboot-safe like the absolute clocks macOS and Windows
# expose, while staying a single integer so a 3.29 reader still parses the third marker line. 0 when the
# kernel does not expose a boot id degrades to bare starttime, which stays safe (a reboot collision fails
# closed rather than reclaiming a live marker).
try:
boot_id = Path("/proc/sys/kernel/random/boot_id").read_text(encoding="ascii")
return int(boot_id.strip().replace("-", ""), 16)
except (OSError, ValueError): # pragma: no cover # the kernel always exposes boot_id as a UUID on Linux
return 0

_BOOT_ID: Final[int] = _read_boot_id()

def process_start_token(pid: int) -> int | None:
"""The ``starttime`` from ``/proc/<pid>/stat`` in clock ticks, or ``None`` when the process is gone."""
"""The ``/proc/<pid>/stat`` ``starttime`` folded with the boot id, or ``None`` when the process is gone."""
try:
data = Path(f"/proc/{pid}/stat").read_bytes()
except OSError:
return None
# starttime is measured in clock ticks since boot, so it changes when a PID is reused within a boot; a
# reused PID that happens to match survives as the recorded owner, which fails closed. psutil identifies a
# process by the same (pid, starttime) pair.
# psutil identifies a process by the same (pid, starttime) pair; the boot id extends that across reboots.
fields = data[data.rfind(b")") + 1 :].split()
if len(fields) <= _STARTTIME_INDEX: # pragma: no cover # a truncated /proc read, never seen in practice
return None
with suppress(ValueError):
return int(fields[_STARTTIME_INDEX])
return None # pragma: no cover # /proc always renders starttime as an integer
try:
starttime = int(fields[_STARTTIME_INDEX])
except ValueError: # pragma: no cover # /proc always renders starttime as an integer
return None
return (_BOOT_ID << 64) | starttime

elif sys.platform == "darwin": # pragma: darwin cover
import ctypes
Expand Down
37 changes: 28 additions & 9 deletions src/filelock/_lease.py
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,8 @@

CompromiseReason = Literal["marker-missing", "owner-changed", "refresh-failed"]

_RefreshOutcome = Literal["ok", "lost", "transient"]


@dataclass(frozen=True)
class LeaseCompromise:
Expand Down Expand Up @@ -162,6 +164,11 @@ def _published_record(self) -> OwnerRecord:

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
# A malformed, legacy or strict record is never reclaimed by age: only a peer that published a lease agreed to
Expand Down Expand Up @@ -212,27 +219,39 @@ def _stop_heartbeat(self) -> None:
heartbeat.join(timeout=self._heartbeat_interval)

def _refresh_until_stopped(self, fd: int, identity: tuple[int, int], token: str) -> None:
# The loop ends at the first loss of the claim, so the holder hears about it once.
# 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):
if not self._refresh_claim(fd, identity, token):
outcome, error = self._refresh_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)
return

def _refresh_claim(self, fd: int, identity: tuple[int, int], token: str) -> bool:
def _refresh_claim(self, fd: int, identity: tuple[int, int], token: str) -> tuple[_RefreshOutcome, OSError | None]:
try:
st = os.lstat(self.lock_file)
except OSError as error:
except FileNotFoundError as error:
self._report_compromise("marker-missing", error, token)
return False
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)
return False
return "lost", None
try:
touch(self.lock_file, fd=fd)
except OSError as error:
self._report_compromise("refresh-failed", error, token)
return False
return True
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.
Expand Down
31 changes: 22 additions & 9 deletions src/filelock/_soft_rw/_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -689,16 +689,21 @@ def _refresh_marker(self) -> bool:
# Open once with O_NOFOLLOW and touch that exact descriptor. Refreshing through the verified fd
# (instead of re-opening by name) closes the window where a peer unlinks our marker and drops a symlink
# or a different file at the path between the read and the touch: utime then lands on the inode we
# verified, or nowhere. A failed open means the marker is gone or was swapped out (a peer evicted us),
# so stop the heartbeat now instead of waiting a tick.
fd = _open_marker(marker_name, dir_fd=dir_fd)
if fd is None:
# verified, or nowhere. Only an unambiguous loss stops the heartbeat: the marker gone, or a peer's token
# in its place. A transient filesystem error (ESTALE / EIO on the NFS-style filesystems this lock targets)
# keeps the heartbeat alive to retry next tick, the way the touch below already does, so one blip does not
# silently drop a held lock.
try:
fd = _open_marker_fd(marker_name, dir_fd=dir_fd)
except FileNotFoundError:
return False
except OSError:
return True
try:
try:
data = os.read(fd, _MAX_MARKER_SIZE + 1)
except OSError: # pragma: no cover - e.g. EAGAIN from a hostile FIFO that has a writer attached
return False
data = _read_marker_fd(fd)
except OSError: # a transient read error or EAGAIN from a hostile FIFO; retry rather than drop the lock
return True
info = _parse_marker_bytes(data)
# Token mismatch means another process already evicted our marker and created its own; stop the
# thread so it does not keep a stranger's file alive.
Expand Down Expand Up @@ -764,13 +769,21 @@ def _read_marker(name: str, *, dir_fd: int | None = None) -> tuple[_MarkerInfo |
return _parse_marker_bytes(data), st.st_mtime


def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None:
def _read_marker_fd(fd: int) -> bytes:
return os.read(fd, _MAX_MARKER_SIZE + 1)


def _open_marker_fd(name: str, *, dir_fd: int | None = None) -> int:
# The file is ours; these guard a hostile mid-flight swap. O_NOFOLLOW rejects a symlink; O_NONBLOCK keeps
# a real FIFO from blocking the open forever, so it reads as a malformed marker instead of wedging a peer
# that holds the state lock.
flags = os.O_RDONLY | _O_NOFOLLOW | _O_NONBLOCK
return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags)


def _open_marker(name: str, *, dir_fd: int | None = None) -> int | None:
try:
return os.open(name, flags, dir_fd=dir_fd) if _SUPPORTS_DIR_FD and dir_fd is not None else os.open(name, flags)
return _open_marker_fd(name, dir_fd=dir_fd)
except OSError:
return None

Expand Down
2 changes: 1 addition & 1 deletion src/filelock/_strict.py
Original file line number Diff line number Diff line change
Expand Up @@ -366,7 +366,7 @@ def _parse_claim(
not trailing,
magic == _CLAIM_MAGIC,
token == name_parts[1],
1 <= pid <= 2**32 - 1,
1 <= pid <= 2**31 - 1,
str(pid) == pid_text,
start is None or (start >= 0 and str(start) == start_text),
hostname.encode().hex() == hostname_hex
Expand Down
37 changes: 31 additions & 6 deletions tests/soft_rw/test_soft_rw_sync.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,7 +7,7 @@
import threading
import time
from contextlib import contextmanager, suppress
from errno import EIO
from errno import EIO, ENOENT
from multiprocessing import Event, Process
from pathlib import Path
from typing import TYPE_CHECKING, Final, Literal
Expand Down Expand Up @@ -601,18 +601,43 @@ def boom(name: str, *, fd: int | None = None) -> None: # noqa: ARG001
lock.close()


@pytest.mark.parametrize(
"target", [pytest.param("_open_marker_fd", id="open"), pytest.param("_read_marker_fd", id="read")]
)
def test_heartbeat_survives_a_transient_marker_error(
lock_file: str, monkeypatch: pytest.MonkeyPatch, target: str
) -> None:
# A transient ESTALE/EIO opening or reading the marker is routine on the NFS-style filesystems this lock targets.
# Unlike the marker actually vanishing, it must not stop the heartbeat and drop the lease.
def boom(*_args: object, **_kwargs: object) -> None:
raise OSError(EIO, "Input/output error")

lock = _make_lock(lock_file, heartbeat_interval=0.02, stale_threshold=0.2)
lock.acquire_write(timeout=2)
try:
hold = lock._hold
assert hold is not None
monkeypatch.setattr(sync_mod, target, boom)
time.sleep(0.2) # ~10 ticks, every one failing the open or read
assert hold.heartbeat_thread.is_alive()
assert not hold.heartbeat_stop.is_set()
finally:
lock.release(force=True)
lock.close()


def test_heartbeat_stops_when_marker_evicted(lock_file: str, monkeypatch: pytest.MonkeyPatch) -> None:
# A failed O_NOFOLLOW open means the marker we held was unlinked or swapped for a symlink. A peer
# evicted us; this is not a transient hiccup, so the heartbeat must stop rather than keep retrying.
def gone(name: str, *, dir_fd: int | None = None) -> int | None: # noqa: ARG001
return None
# An O_NOFOLLOW open failing with ENOENT means the marker we held is gone: a peer evicted us. Unlike a transient
# filesystem error, this is an unambiguous loss, so the heartbeat must stop rather than keep retrying.
def gone(name: str, *, dir_fd: int | None = None) -> int: # noqa: ARG001
raise FileNotFoundError(ENOENT, "No such file or directory")

lock = _make_lock(lock_file, heartbeat_interval=0.02, stale_threshold=0.2)
lock.acquire_write(timeout=2)
try:
hold = lock._hold
assert hold is not None
monkeypatch.setattr(sync_mod, "_open_marker", gone)
monkeypatch.setattr(sync_mod, "_open_marker_fd", gone)
assert hold.heartbeat_stop.wait(timeout=2)
hold.heartbeat_thread.join(timeout=2)
assert not hold.heartbeat_thread.is_alive()
Expand Down
37 changes: 36 additions & 1 deletion tests/test_soft_lease.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,11 @@
from __future__ import annotations

import itertools
import os
import socket
import sys
import time
from errno import EIO
from typing import TYPE_CHECKING, Final, cast

import pytest
Expand Down Expand Up @@ -106,6 +108,16 @@ def test_lease_peer_takes_an_expired_claim(marker: Path, mocker: MockerFixture)
holder.release()


def test_lease_self_heals_a_malformed_marker(marker: Path) -> None:
# A partial write or a foreign file leaves a marker the lease parser cannot read. Rather than block every
# contender until timeout, the base self-heal evicts it once it ages past the malformed grace window.
marker.write_text("not a protocol 2 record\n", encoding="utf-8")
os.utime(marker, (0, 0))

with _lease(marker) as lease:
assert lease.is_lock_held_by_us


def test_lease_reclaims_a_dead_same_host_holder(marker: Path) -> None:
marker.write_text(
f"filelock/2\npid=999999\nhost={socket.gethostname()}\nmode=lease\ntoken=abc\nduration={_DURATION!r}\n",
Expand Down Expand Up @@ -149,11 +161,34 @@ def test_lease_reports_compromise_when_a_refresh_fails(marker: Path, mocker: Moc
lease = _lease(marker, on_compromise=seen.append)

with lease:
time.sleep(_HEARTBEAT * 5)
time.sleep(_DURATION + _HEARTBEAT) # past the refresh margin, so a persistent failure is reported once

assert [(c.reason, c.error) for c in seen] == [("refresh-failed", failure)]


@pytest.mark.parametrize("target", [pytest.param("touch", id="touch"), pytest.param("os.lstat", id="lstat")])
def test_lease_tolerates_a_transient_refresh_error(marker: Path, mocker: MockerFixture, target: str) -> None:
# A transient ESTALE/EIO on the refresh path that recovers before the lease could lapse must not raise a
# compromise: the marker was ours last tick, so retry rather than tell the holder to abandon its work.
import filelock._lease as lease_mod

ticks = itertools.count()
real = cast("Callable[..., object]", lease_mod.touch if target == "touch" else os.lstat)

def flaky(path: str, *args: object, **kwargs: object) -> object:
if path.endswith(marker.name) and next(ticks) < 2:
raise OSError(EIO, "Input/output error")
return real(path, *args, **kwargs)

mocker.patch(f"filelock._lease.{target}", side_effect=flaky)
seen: list[LeaseCompromise] = []
lease = _lease(marker, on_compromise=seen.append)
with lease:
time.sleep(_HEARTBEAT * 6) # several ticks: the first two fail, the rest recover
assert seen == []
assert lease.compromise is None


@_REQUIRES_TAKING_A_LIVE_HOLDERS_MARKER
def test_lease_reports_one_compromise_per_claim(marker: Path) -> None:
seen: list[LeaseCompromise] = []
Expand Down
6 changes: 3 additions & 3 deletions tests/test_strict_soft_failures.py
Original file line number Diff line number Diff line change
Expand Up @@ -737,19 +737,19 @@ def test_strict_soft_accepts_maximum_pid(tmp_path: Path) -> None:
claims = Path(f"{lock_path}.filelock") / "claims"
claims.mkdir(parents=True)
(claims / f"held-v1-{token}.claim").write_text(
f"filelock-strict-v1\n{token}\n4294967295\n686f7374\n4242\n",
f"filelock-strict-v1\n{token}\n2147483647\n686f7374\n4242\n",
encoding="ascii",
newline="",
)

assert StrictSoftFileLock(lock_path).claims[0].pid == 2**32 - 1
assert StrictSoftFileLock(lock_path).claims[0].pid == 2**31 - 1


@pytest.mark.parametrize(
("pid", "hostname_hex"),
[
pytest.param("01", "686f7374", id="leading-zero-pid"),
pytest.param("4294967296", "686f7374", id="pid-overflow"),
pytest.param("2147483648", "686f7374", id="pid-above-signed-int"),
pytest.param("1", "686F7374", id="uppercase-hostname-hex"),
pytest.param("1", "20686f737420", id="spaced-hostname"),
pytest.param("1", "686f737400", id="null-hostname"),
Expand Down