Fix SoftReadWriteLock state lock timeout

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 <path>.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

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()