fix(locks): retry atomic_write_bytes on transient PermissionError (WinError 5)

openkb/locks.py

@@ -14,11 +14,14 @@
 import os
 import tempfile
 import threading
+import time
 from pathlib import Path
 from typing import IO, Iterator
 
 import portalocker
 
+logger = logging.getLogger(__name__)
+
 
 def flock(fh: IO, *, exclusive: bool) -> None:
     """Acquire an advisory lock on an open file handle (cross-platform).
@@ -220,6 +223,40 @@ def _target_mode(path: Path) -> int:
         return _default_file_mode()
 
 
+_REPLACE_RETRY_ATTEMPTS = 5
+_REPLACE_RETRY_BASE_DELAY = 0.05  # seconds, doubles each attempt
+
+
+def _replace_with_retry(tmp_path: Path, path: Path) -> None:
+    """Rename *tmp_path* onto *path*, retrying a transient Windows file lock.
+
+    A freshly written temp file can be briefly opened by another process
+    (real-time antivirus scanning, the search indexer, backup/sync agents)
+    right before the rename, which makes ``os.replace()`` raise
+    ``PermissionError`` (Windows ``WinError 5``) even though nothing in this
+    process holds the file open and the lock typically clears within
+    milliseconds. Only ``PermissionError`` is retried; any other ``OSError``
+    (e.g. a real permissions problem) is raised immediately.
+    """
+    delay = _REPLACE_RETRY_BASE_DELAY
+    for attempt in range(_REPLACE_RETRY_ATTEMPTS):
+        try:
+            os.replace(tmp_path, path)
+            return
+        except PermissionError:
+            if attempt == _REPLACE_RETRY_ATTEMPTS - 1:
+                raise
+            logger.debug(
+                "os.replace(%s, %s) hit a transient PermissionError, retrying (attempt %d/%d)",
+                tmp_path,
+                path,
+                attempt + 1,
+                _REPLACE_RETRY_ATTEMPTS,
+            )
+            time.sleep(delay)
+            delay *= 2
+
+
 def atomic_write_bytes(path: Path, content: bytes) -> None:
     """Atomically replace *path* with binary *content*."""
     path.parent.mkdir(parents=True, exist_ok=True)
@@ -232,7 +269,7 @@ def atomic_write_bytes(path: Path, content: bytes) -> None:
             fh.write(content)
             fh.flush()
             os.fsync(fh.fileno())
-        os.replace(tmp_path, path)
+        _replace_with_retry(tmp_path, path)
         _fsync_directory(path.parent)
     finally:
         tmp_path.unlink(missing_ok=True)

tests/test_locks.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import json
+import os
 import stat
 import threading
 
@@ -127,3 +128,59 @@ def test_atomic_write_json_replaces_file(tmp_path):
     atomic_write_json(target, {"a": {"name": "doc.pdf"}}, ensure_ascii=False)
 
     assert json.loads(target.read_text(encoding="utf-8")) == {"a": {"name": "doc.pdf"}}
+
+
+def test_atomic_write_bytes_retries_transient_permission_error(tmp_path, monkeypatch):
+    target = tmp_path / "file.txt"
+    real_replace = os.replace
+    calls = []
+
+    def flaky_replace(src, dst):
+        calls.append((src, dst))
+        if len(calls) < 3:
+            raise PermissionError("simulated transient lock")
+        real_replace(src, dst)
+
+    monkeypatch.setattr("openkb.locks.os.replace", flaky_replace)
+    monkeypatch.setattr("openkb.locks.time.sleep", lambda _seconds: None)
+
+    atomic_write_text(target, "content")
+
+    assert len(calls) == 3
+    assert target.read_text(encoding="utf-8") == "content"
+    assert list(target.parent.glob("*.tmp")) == []
+
+
+def test_atomic_write_bytes_reraises_after_exhausting_retries(tmp_path, monkeypatch):
+    target = tmp_path / "file.txt"
+    calls = []
+
+    def always_fails(src, dst):
+        calls.append((src, dst))
+        raise PermissionError("simulated persistent lock")
+
+    monkeypatch.setattr("openkb.locks.os.replace", always_fails)
+    monkeypatch.setattr("openkb.locks.time.sleep", lambda _seconds: None)
+
+    with pytest.raises(PermissionError):
+        atomic_write_text(target, "content")
+
+    assert len(calls) == 5
+    assert not target.exists()
+    assert list(target.parent.glob("*.tmp")) == []
+
+
+def test_atomic_write_bytes_does_not_retry_other_os_error(tmp_path, monkeypatch):
+    target = tmp_path / "file.txt"
+    calls = []
+
+    def not_a_lock(src, dst):
+        calls.append((src, dst))
+        raise OSError("simulated unrelated failure")
+
+    monkeypatch.setattr("openkb.locks.os.replace", not_a_lock)
+
+    with pytest.raises(OSError, match="simulated unrelated failure"):
+        atomic_write_text(target, "content")
+
+    assert len(calls) == 1