diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index c15ac32e..5b9ee44e 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/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 `.lock`, the lock the C#, Rust and TypeScript runtimes take: flock on Unix, diff --git a/packages/cultcache-py/tests/test_store_replace.py b/packages/cultcache-py/tests/test_store_replace.py new file mode 100644 index 00000000..120fdc60 --- /dev/null +++ b/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()