cultcache-py: replace the store POSIX-first on Windows through ctypes kernel32, MoveFileExW fallback, bounded retry

Mirrors the TypeScript decision table (variants:ruling:windows-replace-posix-first). Both store writes route through _replace_store; os.replace stays on other OSes.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

packages/cultcache-py/src/cultcache_py/stores.py

@@ -130,7 +130,7 @@ class JsonLinesBackingStore:
                 "stored_at": envelope.stored_at,
             }, ensure_ascii=False, sort_keys=True))
         temp.write_text("\n".join(lines) + ("\n" if lines else ""), encoding="utf-8")
-        temp.replace(self.path)
+        _replace_store(temp, self.path)
 
 
 class SingleFileMessagePackBackingStore:
@@ -323,7 +323,7 @@ def commit_store_file(path: Path, data: bytes) -> None:
             os.fsync(fd)
         finally:
             os.close(fd)
-        os.replace(staged, path)
+        _replace_store(staged, path)
     except BaseException:
         try:
             os.unlink(staged)
@@ -338,6 +338,108 @@ def commit_store_file(path: Path, data: bytes) -> None:
             os.close(directory)
 
 
+REPLACE_ATTEMPTS = 25
+REPLACE_RETRY_SECONDS = 0.02
+_ERROR_ACCESS_DENIED = 5
+_ERROR_SHARING_VIOLATION = 32
+_ERROR_LOCK_VIOLATION = 33
+_POSIX_UNSUPPORTED = (1, 50, 87)  # incorrect function, not supported, invalid parameter: FAT, exFAT, before Windows 10 1607
+MOVEFILE_REPLACE_WRITE_THROUGH = 0x9  # MOVEFILE_REPLACE_EXISTING | MOVEFILE_WRITE_THROUGH
+
+
+class ReplaceKernel:
+    """The kernel32 calls a Windows store replace needs, each answering 0 or a Win32 error code."""
+
+    def rename_posix(self, staged: str, destination: str) -> int:
+        """CreateFileW(staged, DELETE|SYNCHRONIZE), then FileRenameInfoEx with REPLACE_IF_EXISTS|POSIX_SEMANTICS, then CloseHandle."""
+        raise NotImplementedError
+
+    def move_file_ex(self, staged: str, destination: str, flags: int) -> int:
+        raise NotImplementedError
+
+    def sleep(self, seconds: float) -> None:
+        time.sleep(seconds)
+
+
+def _replace_store(staged: str | os.PathLike[str], destination: str | os.PathLike[str]) -> None:
+    """Replaces `destination` with `staged`: POSIX-semantics rename first on Windows (so a reader holding the store open
+    never sees it absent), os.replace elsewhere."""
+    if sys.platform == "win32":
+        replace_store_windows(os.fspath(staged), os.fspath(destination), WindowsReplaceKernel())
+    else:
+        os.replace(staged, destination)
+
+
+def replace_store_windows(staged: str, destination: str, kernel: ReplaceKernel) -> None:
+    """POSIX-semantics rename, falling back for the rest of the call to write-through MoveFileExW where POSIX semantics are
+    unsupported; errors 5, 32 and 33 retry up to 25 times 20 ms apart, any other code raises at once. The TypeScript and C#
+    runtimes take the same decision table."""
+    posix = True
+    last = 0
+    for _ in range(REPLACE_ATTEMPTS):
+        code = kernel.rename_posix(staged, destination) if posix else kernel.move_file_ex(staged, destination, MOVEFILE_REPLACE_WRITE_THROUGH)
+        if posix and code in _POSIX_UNSUPPORTED:
+            posix = False
+            code = kernel.move_file_ex(staged, destination, MOVEFILE_REPLACE_WRITE_THROUGH)
+        if code == 0:
+            return
+        if code not in (_ERROR_ACCESS_DENIED, _ERROR_SHARING_VIOLATION, _ERROR_LOCK_VIOLATION):
+            raise _replace_error(f"CultCache could not replace the store: Win32 error {code}.", code)
+        last = code
+        kernel.sleep(REPLACE_RETRY_SECONDS)
+    raise _replace_error(f"CultCache could not replace the store after {REPLACE_ATTEMPTS} attempts: last Win32 error {last}.", last)
+
+
+def _replace_error(message: str, code: int) -> OSError:
+    error = OSError(message)
+    error.winerror = code  # type: ignore[attr-defined]
+    return error
+
+
+class WindowsReplaceKernel(ReplaceKernel):
+    """kernel32 bound by ctypes, as _try_lock_windows binds it. Windows only."""
+
+    def __init__(self) -> None:
+        import ctypes
+        from ctypes import wintypes
+
+        self._ctypes = ctypes
+        k = self._k = ctypes.WinDLL("kernel32", use_last_error=True)
+        k.CreateFileW.restype = wintypes.HANDLE
+        k.CreateFileW.argtypes = [wintypes.LPCWSTR, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p, wintypes.DWORD, wintypes.DWORD, wintypes.HANDLE]
+        k.SetFileInformationByHandle.restype = wintypes.BOOL
+        k.SetFileInformationByHandle.argtypes = [wintypes.HANDLE, ctypes.c_int, ctypes.c_void_p, wintypes.DWORD]
+        k.CloseHandle.restype = wintypes.BOOL
+        k.CloseHandle.argtypes = [wintypes.HANDLE]
+        k.MoveFileExW.restype = wintypes.BOOL
+        k.MoveFileExW.argtypes = [wintypes.LPCWSTR, wintypes.LPCWSTR, wintypes.DWORD]
+
+    def rename_posix(self, staged: str, destination: str) -> int:
+        ctypes, k = self._ctypes, self._k
+        delete_synchronize, share_all, open_existing, normal = 0x00110000, 0x7, 3, 0x80
+        handle = k.CreateFileW(staged, delete_synchronize, share_all, None, open_existing, normal, None)
+        if handle is None or ctypes.c_void_p(handle).value in (None, ctypes.c_void_p(-1).value):
+            return ctypes.get_last_error()
+        try:
+            # FILE_RENAME_INFO: Flags (DWORD), RootDirectory (pointer-aligned HANDLE), FileNameLength (DWORD), FileName (WCHAR[]).
+            name = os.path.abspath(destination).encode("utf-16-le")
+            pointer = ctypes.sizeof(ctypes.c_void_p)
+            length_at = pointer + pointer
+            info = ctypes.create_string_buffer(length_at + 4 + len(name) + 2)
+            info[0:4] = (0x3).to_bytes(4, "little")  # FILE_RENAME_FLAG_REPLACE_IF_EXISTS | FILE_RENAME_FLAG_POSIX_SEMANTICS
+            info[length_at:length_at + 4] = len(name).to_bytes(4, "little")
+            info[length_at + 4:length_at + 4 + len(name)] = name
+            file_rename_info_ex = 22
+            if k.SetFileInformationByHandle(handle, file_rename_info_ex, info, len(info)):
+                return 0
+            return ctypes.get_last_error()
+        finally:
+            k.CloseHandle(handle)
+
+    def move_file_ex(self, staged: str, destination: str, flags: int) -> int:
+        return 0 if self._k.MoveFileExW(staged, destination, flags) else self._ctypes.get_last_error()
+
+
 @contextmanager
 def store_write_lock(path: Path, timeout: float = LOCK_TIMEOUT_SECONDS) -> Iterator[None]:
     """Holds the exclusive store lock on `<path>.lock`, the lock the C#, Rust and TypeScript runtimes take: flock on Unix,

packages/cultcache-py/tests/test_store_replace.py

@@ -0,0 +1,124 @@
+from __future__ import annotations
+
+import os
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from cultcache_py import SingleFileMessagePackBackingStore, stores
+from cultcache_py.backing_store import CultCacheEnvelope
+from cultcache_py.stores import MOVEFILE_REPLACE_WRITE_THROUGH, ReplaceKernel, replace_store_windows
+
+
+class FakeKernel(ReplaceKernel):
+    def __init__(self, posix: list[int], move_ex: list[int]) -> None:
+        self.posix, self.move_ex = list(posix), list(move_ex)
+        self.log: list[str] = []
+        self.sleeps: list[float] = []
+
+    def rename_posix(self, staged, destination):
+        self.log.append("posix")
+        return self.posix.pop(0) if self.posix else 0
+
+    def move_file_ex(self, staged, destination, flags):
+        self.log.append(f"moveex:{flags}")
+        return self.move_ex.pop(0) if self.move_ex else 0
+
+    def sleep(self, seconds):
+        self.sleeps.append(seconds)
+
+
+def run(posix, move_ex):
+    kernel = FakeKernel(posix, move_ex)
+    return kernel, lambda: replace_store_windows("a", "b", kernel)
+
+
+class WindowsReplaceTable(unittest.TestCase):
+    def test_falls_back_to_write_through_move_where_posix_rename_is_unsupported(self):
+        self.assertEqual(MOVEFILE_REPLACE_WRITE_THROUGH, 0x9)
+        for unsupported in (1, 50, 87):
+            kernel, go = run([unsupported], [0])
+            go()
+            self.assertEqual(kernel.log, ["posix", "moveex:9"])
+
+    def test_once_on_move_the_rest_of_the_call_stays_there(self):
+        kernel, go = run([87], [32, 0])
+        go()
+        self.assertEqual(kernel.log, ["posix", "moveex:9", "moveex:9"])
+
+    def test_transient_codes_retry_the_same_primitive(self):
+        for transient in (5, 32, 33):
+            kernel, go = run([transient, transient, 0], [])
+            go()
+            self.assertEqual(kernel.log, ["posix", "posix", "posix"])
+            self.assertEqual(kernel.sleeps, [0.02, 0.02])
+
+    def test_a_working_posix_rename_is_the_whole_replace(self):
+        kernel, go = run([0], [])
+        go()
+        self.assertEqual(kernel.log, ["posix"])
+        self.assertEqual(kernel.sleeps, [])
+
+    def test_unsupported_after_the_fallback_is_an_error_not_another_fallback(self):
+        kernel, go = run([87], [32, 87])
+        with self.assertRaisesRegex(OSError, "Win32 error 87") as caught:
+            go()
+        self.assertEqual(caught.exception.winerror, 87)
+        self.assertEqual(kernel.log, ["posix", "moveex:9", "moveex:9"])
+        for unsupported in (1, 50, 87):
+            kernel, go = run([87], [unsupported])
+            with self.assertRaisesRegex(OSError, f"Win32 error {unsupported}"):
+                go()
+            self.assertEqual(kernel.log, ["posix", "moveex:9"])
+
+    def test_gives_up_after_25_attempts_naming_the_last_code(self):
+        kernel, go = run([32] * 30, [])
+        with self.assertRaisesRegex(OSError, "after 25 attempts: last Win32 error 32\\."):
+            go()
+        self.assertEqual(len(kernel.log), 25)
+
+    def test_other_codes_fail_at_once(self):
+        kernel, go = run([2], [])
+        with self.assertRaisesRegex(OSError, "Win32 error 2"):
+            go()
+        self.assertEqual(len(kernel.log), 1)
+        kernel, go = run([87], [3])
+        with self.assertRaisesRegex(OSError, "Win32 error 3"):
+            go()
+
+
+class StoreWritesRouteThroughTheReplace(unittest.TestCase):
+    def test_both_stores_replace_through_replace_store(self):
+        seen: list[tuple[Path, Path]] = []
+        real = stores._replace_store
+
+        def spy(staged, destination):
+            seen.append((Path(staged), Path(destination)))
+            real(staged, destination)
+
+        with tempfile.TemporaryDirectory() as directory, mock.patch.object(stores, "_replace_store", spy):
+            envelope = CultCacheEnvelope(key="k", type="t", payload=b"\x01", stored_at="2026-11-06T00:00:00Z")
+            binary = Path(directory) / "s.cc"
+            SingleFileMessagePackBackingStore(binary).push_all([envelope])
+            lines = Path(directory) / "s.jsonl"
+            stores.JsonLinesBackingStore(lines).push_all([envelope])
+            self.assertEqual([destination for _, destination in seen], [binary, lines])
+            self.assertTrue(binary.read_bytes() and lines.read_text())
+
+    @unittest.skipUnless(sys.platform == "win32", "needs a real Windows")
+    def test_real_kernel_replaces_through_both_primitives(self):
+        kernel = stores.WindowsReplaceKernel()
+        with tempfile.TemporaryDirectory() as directory:
+            for name, replace in (("posix", kernel.rename_posix), ("moveex", lambda s, d: kernel.move_file_ex(s, d, MOVEFILE_REPLACE_WRITE_THROUGH))):
+                staged, store = Path(directory, name + ".tmp"), Path(directory, name)
+                store.write_bytes(b"old")
+                staged.write_bytes(b"new")
+                self.assertEqual(replace(str(staged), str(store)), 0)
+                self.assertEqual(store.read_bytes(), b"new")
+            self.assertEqual(kernel.rename_posix(str(Path(directory, "missing")), str(Path(directory, "x"))), 2)
+
+
+if __name__ == "__main__":
+    unittest.main()