🐛 fix(soft): ignore a restore claim with an invalid pid

docs/changelog/+restore-claim-pid.bugfix.rst

@@ -0,0 +1,4 @@
+Fix :class:`~filelock.SoftFileLock`, :class:`~filelock.MarkerSoftFileLock` and :class:`~filelock.SoftFileLease`
+raising ``OverflowError`` out of ``acquire()``, or never reclaiming a dead holder's lock, when a stranded restore claim
+beside the lock names a process number outside ``1..2**31-1`` or spells it in non-ASCII digits. Such a claim is now
+ignored, as no filelock writes one.

src/filelock/_soft.py

@@ -230,7 +230,8 @@ def _restore_stranded_marker(self, breaker_pid: int, marker: os.stat_result) ->
     def _stranded_candidates(self, breaker_pid: int) -> list[tuple[str, bool]] | None:  # pragma: win32 no cover
         directory, name = os.path.split(self._lock_path)
         stranded: Final = re.compile(rf"{re.escape(name)}\.break\.{breaker_pid}\.[0-9a-f]{{32}}")
-        claimed: Final = re.compile(rf"{re.escape(name)}\.restore\.(\d+)\.[0-9a-f]{{32}}")
+        # [0-9], not \d: a str pattern's \d takes any Unicode digit, and int() reads fullwidth ones as another number.
+        claimed: Final = re.compile(rf"{re.escape(name)}\.restore\.([0-9]+)\.[0-9a-f]{{32}}")
         candidates: list[tuple[str, bool]] = []
         with os.scandir(directory or os.curdir) as entries:
             for entry in entries:
@@ -239,9 +240,10 @@ def _stranded_candidates(self, breaker_pid: int) -> list[tuple[str, bool]] | Non
                 path = str(Path(directory) / entry.name)
                 if stranded.fullmatch(entry.name):
                     candidates.append((path, False))
-                elif found := claimed.fullmatch(entry.name):
-                    ours = int(found[1]) == os.getpid()
-                    if not (ours or owner_is_stale(int(found[1]), host_name(), None)):
+                elif (found := claimed.fullmatch(entry.name)) and 1 <= (recoverer := int(found[1])) <= 2**31 - 1:
+                    # No filelock writes a claim outside the pid range; os.kill would raise OverflowError on one.
+                    ours = recoverer == os.getpid()
+                    if not (ours or owner_is_stale(recoverer, host_name(), None)):
                         return None
                     candidates.append((path, ours))
         return sorted(candidates)

tests/test_soft_stale.py

@@ -375,7 +375,7 @@ def test_stale_break_restores_a_marker_of_another_host_with_the_breakers_pid(  #
     assert (lock.read_text(encoding="utf-8"), list(tmp_path.glob("test.lock.break.*"))) == (foreign, [])
 
 
-def _restore_name(lock: Path, pid: int) -> Path:  # pragma: win32 no cover
+def _restore_name(lock: Path, pid: int | str) -> Path:  # pragma: win32 no cover
     return lock.with_name(f"test.lock.restore.{pid}.{'0' * 32}")
 
 
@@ -506,6 +506,30 @@ def test_stale_break_skips_a_record_a_live_recoverer_claimed(  # pragma: win32 n
     assert (lock.read_text(encoding="utf-8"), claimed.read_text(encoding="utf-8")) == (record, foreign)
 
 
+@pytest.mark.skipif(sys.platform == "win32", reason="Windows cannot exchange two names")
+@STALE_BREAKERS
+@NEEDS_STALE_RECLAIM
+@pytest.mark.parametrize(
+    "recoverer",
+    [
+        pytest.param(str(2**64), id="oversized"),
+        pytest.param("0", id="zero"),
+        # This process's pid in fullwidth digits, which int() reads back as the same number.
+        pytest.param("".join(chr(0xFF10 + int(digit)) for digit in str(os.getpid())), id="wide"),
+    ],
+)
+def test_stale_break_ignores_a_claim_naming_no_valid_pid(  # pragma: win32 no cover
+    tmp_path: Path, lock_type: type[SoftFileLock], record: str, recoverer: str
+) -> None:
+    (lock := tmp_path / "test.lock").write_text(record, encoding="utf-8")
+    foreign: Final = record.replace(_HOST, "elsewhere")
+    (planted := _restore_name(lock, recoverer)).write_text(foreign, encoding="utf-8")
+    with held_after_reclaim(lock_type(lock)) as held:
+        assert held.is_locked
+
+    assert planted.read_text(encoding="utf-8") == foreign
+
+
 def _refuse_returning_a_claim(mocker: MockerFixture) -> None:  # pragma: win32 no cover
     rename: Final = Path.rename