diff --git a/src/filelock/_soft_rw/_sync.py b/src/filelock/_soft_rw/_sync.py index 9d8ba019..4242391f 100644 --- a/src/filelock/_soft_rw/_sync.py +++ b/src/filelock/_soft_rw/_sync.py @@ -394,14 +394,22 @@ def release(self, *, force: bool = False) -> None: else: self._unlink_writer_marker_if_ours(hold.token) - def _unlink_writer_marker_if_ours(self, token: str) -> None: + def _unlink_writer_marker_if_ours(self, token: str, *, deadline: float | None = None) -> None: # Remove the writer marker only while it still carries our token. If this holder was paused long # enough (a stop-the-world GC pause, SIGSTOP, a suspended VM) for a peer to evict the marker as # stale and claim the writer slot itself, the file now at .write is the peer's live marker; # unlinking it by path would let a second writer through and break mutual exclusion. The state lock # serializes this against a concurrent break/claim, and the heartbeat is already stopped, so the # token we read is authoritative. Mirrors the token re-check the stale-break path already does. - with self._locks.state: + # ``deadline`` bounds the wait the same way acquisition does; release() passes ``None`` (its existing, + # unbounded behavior - it takes no timeout of its own to honor). The failed-acquisition cleanup call + # below is already past its own deadline, so it hands in the current instant as a one-shot, + # non-blocking attempt (#725): piling more unbounded waiting onto a call the caller expected to have + # already returned from would just move the "blocks forever" bug here instead of fixing it. + state = self._try_acquire_state(deadline) + if state is None: + return + with state: if (read := _read_marker(self._paths.write)) is None: return info, _ = read @@ -516,6 +524,19 @@ def _validate_reentrant(self, mode: _Mode) -> AcquireReturnProxy: hold.level += 1 return AcquireReturnProxy(lock=self) + def _try_acquire_state(self, deadline: float | None) -> AcquireReturnProxy | None: + # The internal ".state" mutex has no heartbeat of its own (#725): a marker left behind by a peer that + # died mid-transition on another host is never evicted as stale, so waiting for it must still honor the + # caller's own acquisition deadline rather than blocking indefinitely underneath it. ``None`` here means + # "not yet, try again" rather than a hard failure, mirroring every other predicate this feeds into so + # _wait_for's own deadline check is what ultimately raises Timeout(self.lock_file) - not this call, + # which would otherwise leak the ".state" path as an implementation detail. + timeout = -1.0 if deadline is None else max(deadline - time.perf_counter(), 0.0) + try: + return self._locks.state.acquire(timeout=timeout) + except Timeout: + return None + def _acquire_writer_slot( self, token: str, @@ -528,11 +549,17 @@ def _acquire_writer_slot( self._open_readers_dir() def try_claim_writer() -> bool: - with self._locks.state: + state = self._try_acquire_state(deadline) + if state is None: + return False + with state: return self._claim_writer_marker(token) def readers_drained_touching() -> bool: - with self._locks.state: + state = self._try_acquire_state(deadline) + if state is None: + return False + with state: # A peer may replace an expired marker while this process pauses. Refresh only our token; touching a # successor's marker would let this acquisition proceed without owning the writer slot. if not self._touch_writer_marker_if_ours(token) and not self._claim_writer_marker(token): @@ -546,7 +573,9 @@ def readers_drained_touching() -> bool: except Timeout: # Give up our writer claim so readers can make progress again, but only while the marker is # still ours: a peer may have evicted it as stale and claimed the slot while phase 2 waited. - self._unlink_writer_marker_if_ours(token) + # We're already past our own deadline, so this cleanup gets a single non-blocking attempt at + # ".state" rather than more unbounded waiting (#725). + self._unlink_writer_marker_if_ours(token, deadline=time.perf_counter()) raise return self._paths.write, False @@ -597,7 +626,10 @@ def _acquire_reader_slot( full_reader_path = str(Path(self._paths.readers) / reader_name) def try_claim_reader() -> bool: - with self._locks.state: + state = self._try_acquire_state(deadline) + if state is None: + return False + with state: _break_stale_marker(self._paths.write, stale_threshold=self.stale_threshold, now=time.time()) if _file_exists(self._paths.write): return False diff --git a/tests/soft_rw/test_soft_rw_sync.py b/tests/soft_rw/test_soft_rw_sync.py index 6cbfea5c..353d5327 100644 --- a/tests/soft_rw/test_soft_rw_sync.py +++ b/tests/soft_rw/test_soft_rw_sync.py @@ -19,6 +19,7 @@ from filelock import Timeout from filelock import _util as util_mod +from filelock._identity import host_name from filelock._soft_rw import SoftReadWriteLock from filelock._soft_rw import _sync as sync_mod from tests.capability_marks import NEEDS_FILE_MODE, NEEDS_FORK, NEEDS_POSIX_SIGNALS, SKIP_ON_UNRELIABLE_PROCESS_SYNC @@ -1300,3 +1301,73 @@ def test_refresh_marker_stops_once_a_peer_owns_the_marker(lock_file: str) -> Non assert lock._refresh_marker() is False finally: lock.close() + + +def test_acquire_write_respects_timeout_for_unreclaimable_state_marker(lock_file: str) -> None: + # owner_is_stale refuses to probe a PID recorded under a different hostname (it cannot prove a foreign host's + # process is dead), so a ".state" marker from another host is never reclaimed. Writing one directly is the + # actual failure this issue describes, and needs no subprocess or liveness check to exercise (#725). + Path(f"{lock_file}.state").write_text(f"424242\n{host_name()}-other\n", encoding="utf-8") + lock = _make_lock(lock_file) + try: + start = time.perf_counter() + with pytest.raises(Timeout) as exc_info: + lock.acquire_write(timeout=0.3) + elapsed = time.perf_counter() - start + # Pre-fix this blocks forever, since the marker can never be proven stale; 2s is generous slack above the + # 0.3s deadline actually being asserted. + assert elapsed < 2 + # The internal ".state" mutex is what actually ran out of time; the caller only knows about the public + # lock file, so the exception must not leak the companion path as an implementation detail. + assert exc_info.value.lock_file == lock_file + finally: + lock.close() + + +def test_acquire_read_respects_timeout_for_unreclaimable_state_marker(lock_file: str) -> None: + # Same fix, the reader's side of _acquire_reader_slot (#725). + Path(f"{lock_file}.state").write_text(f"424242\n{host_name()}-other\n", encoding="utf-8") + lock = _make_lock(lock_file) + try: + start = time.perf_counter() + with pytest.raises(Timeout) as exc_info: + lock.acquire_read(timeout=0.3) + assert time.perf_counter() - start < 2 + assert exc_info.value.lock_file == lock_file + finally: + lock.close() + + +def test_writer_phase2_respects_timeout_when_state_becomes_unreclaimable( + lock_file: str, + monkeypatch: pytest.MonkeyPatch, +) -> None: + # Phase 1 (claiming the writer marker) can succeed before the ".state" marker is ever contended, then phase 2 + # (waiting for readers to drain) re-acquires ".state" on every poll. A live reader keeps phase 2 looping; on + # the first poll sleep we simulate ".state" becoming an unreclaimable cross-host marker (#725). + reader = _make_lock(lock_file, heartbeat_interval=10, stale_threshold=40) + reader.acquire_read(timeout=2) + writer = _make_lock(lock_file, heartbeat_interval=10, stale_threshold=40) + state_marker = f"{lock_file}.state" + real_sleep = time.sleep + swapped = threading.Event() + + def hook(seconds: float) -> None: + if not swapped.is_set(): + swapped.set() + Path(state_marker).write_text(f"424242\n{host_name()}-other\n", encoding="utf-8") + real_sleep(seconds) + + monkeypatch.setattr(sync_mod.time, "sleep", hook) + try: + start = time.perf_counter() + with pytest.raises(Timeout) as exc_info: + writer.acquire_write(timeout=0.3) + elapsed = time.perf_counter() - start + assert swapped.is_set() + assert elapsed < 2 + assert exc_info.value.lock_file == lock_file + finally: + monkeypatch.setattr(sync_mod.time, "sleep", real_sleep) + writer.close() + reader.close()