packages/cultcache-py/CHANGELOG.md

@@ -2,6 +2,11 @@
 
 ## Unreleased
 
+### Changed
+
+- `msgpack` is a hard dependency, as `pyproject.toml` declared it. The optional-import fallbacks and the `cultcache-py[msgpack]` extra they named (which never existed) are gone: a missing msgpack is an `ImportError` at import.
+- The single-file store's `push`, `push_all` and `delete` take the cross-process lock on `<path>.lock` (flock; LockFileEx over the whole file on Windows) across the read and the replace, waiting up to `lock_timeout` seconds (30 by default, `TimeoutError` after), and exclude the C#, Rust and TypeScript runtimes' writers. The staged file is `<path>.tmp-<pid>-<uuid>`, fsynced before the replace, with a directory fsync after it off Windows. A fixed `<path>.tmp` is no longer used.
+
 ### Breaking
 
 - A single-file store written as `cultcache.store.v3` is read and written: a record that holds an element id (an owned list element with an `[CultElementId]` member, carried through every write) marks the store v3, and a v3 store is carried through a rewrite with its ids. A runtime from before this change refuses a v3 store by name.

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

@@ -6,6 +6,9 @@ from collections.abc import Mapping, Sequence, Set
 from itertools import chain
 from typing import Any, Callable, Generic, Iterable, TypeVar
 
+import msgpack
+from msgpack.exceptions import StackError as _MsgpackStackError
+
 from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember
 from .leaves import Float32, SlotCodec
 from .leaves import any as _any_leaf
@@ -24,12 +27,7 @@ or skipped, the outermost one is depth 1; depth 256 is accepted and 257 refused,
 MAX_SLOT_KEY = 65535
 """The largest slot key a record may define: the largest MessagePack array16 length."""
 
-try:
-    from msgpack.exceptions import StackError as _MsgpackStackError
-
-    _STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
-except ModuleNotFoundError:
-    _STACK_FAULTS = ()
+_STACK_FAULTS: tuple[type[BaseException], ...] = (_MsgpackStackError,)
 
 
 _CONTAINERS = (Mapping, Sequence, Set)
@@ -85,22 +83,10 @@ def _json_payload_decoder(payload: bytes) -> Any:
     return json.loads(payload.decode("utf-8"))
 
 
-def _msgpack() -> Any:
-    try:
-        import msgpack  # type: ignore
-    except ModuleNotFoundError as exc:
-        raise RuntimeError(
-            "MessagePack payload encoding requires the optional 'msgpack' dependency. "
-            "Install with: python -m pip install cultcache-py[msgpack]"
-        ) from exc
-    return msgpack
-
-
 def _pack_tree(tree: Any) -> bytes:
     """The bytes of the tree the leaves produced. `packb` writes everything but a `Float32`, which it refuses with a
     TypeError; only then does an iterative walk pack each array and map header and each value, a `Float32` through a
     single-float packer."""
-    msgpack = _msgpack()
     try:
         return msgpack.packb(tree, use_bin_type=True)  # type: ignore[no-any-return]
     except TypeError:
@@ -125,13 +111,6 @@ def _pack_tree(tree: Any) -> bytes:
 
 
 def _messagepack_payload_decoder(payload: bytes) -> Any:
-    try:
-        import msgpack  # type: ignore
-    except ModuleNotFoundError as exc:
-        raise RuntimeError(
-            "MessagePack payload decoding requires the optional 'msgpack' dependency. "
-            "Install with: python -m pip install cultcache-py[msgpack]"
-        ) from exc
     return msgpack.unpackb(payload, raw=False)
 
 

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

@@ -16,10 +16,7 @@ from collections.abc import Mapping
 from dataclasses import dataclass
 from typing import Any, Callable, NoReturn
 
-try:
-    from msgpack import ExtType as _ExtType
-except ModuleNotFoundError:
-    _ExtType = ()  # type: ignore[assignment,misc]
+from msgpack import ExtType as _ExtType
 
 _bool, _bytes, _list, _int, _float, _str = bool, bytes, list, int, float, str
 _BYTE_LIKE = (_bytes, bytearray, memoryview)

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

@@ -3,9 +3,15 @@
 import base64
 import json
 import os
+import sys
 import threading
+import time
+import uuid
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any, Callable, TypeVar
+from typing import Any, Callable, Iterator, TypeVar
+
+import msgpack
 
 from .backing_store import (
     CultCacheEnvelope,
@@ -21,6 +27,8 @@ STORE_FORMAT_ELEMENT_IDS = "cultcache.store.v3"
 _READABLE_STORE_FORMATS = (STORE_FORMAT_VERSION, STORE_FORMAT_ELEMENT_IDS)
 _STORE_FORMAT_PREFIX = "cultcache.store."
 _PERSISTED_RECORD_SLOTS = 4
+_LOCK_POLL_SECONDS = 0.01
+LOCK_TIMEOUT_SECONDS = 30.0
 _Contents = TypeVar("_Contents")
 
 
@@ -114,14 +122,14 @@ class JsonLinesBackingStore:
 
 
 class SingleFileMessagePackBackingStore:
-    def __init__(self, path: str | os.PathLike[str]) -> None:
+    def __init__(self, path: str | os.PathLike[str], lock_timeout: float = LOCK_TIMEOUT_SECONDS) -> None:
         self.path = Path(path)
+        self.lock_timeout = lock_timeout
         self._lock = threading.RLock()
         # The header this store read; a rewrite keeps it, so a store that holds element ids never sheds its marker.
         self._format = STORE_FORMAT_VERSION
 
     def pull_all(self) -> list[CultCacheEnvelope]:
-        msgpack = self._msgpack()
         with self._lock:
             # An empty file is not a store (no writer leaves one), so it is decoded and refused.
             # The disk decides the header: a file that is gone is not marked.
@@ -135,39 +143,34 @@ class SingleFileMessagePackBackingStore:
                 if snapshot is not None:
                     self._format, envelopes = snapshot
                     return envelopes
-                return _decode_legacy_envelopes(decoded, msgpack)
+                return _decode_legacy_envelopes(decoded)
             except Exception as exc:
                 raise StoreUnreadableError(self.path, exc) from exc
 
     def push(self, envelope: CultCacheEnvelope) -> None:
-        with self._lock:
+        with self._write_transaction():
             self._replace_all([envelope], self.pull_all())
 
     def push_all(self, envelopes: list[CultCacheEnvelope]) -> None:
-        with self._lock:
+        with self._write_transaction():
             self._replace_all(list(envelopes), self.pull_all())
 
     def delete(self, type: str, key: str) -> None:
-        with self._lock:
+        with self._write_transaction():
             self._replace_all([], [item for item in self.pull_all() if not (item.type == type and item.key == key)])
 
-    def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> None:
-        msgpack = self._msgpack()
-        self.path.parent.mkdir(parents=True, exist_ok=True)
-        temp = self.path.with_suffix(self.path.suffix + ".tmp")
-        temp.write_bytes(msgpack.packb(_encode_snapshot(supplied, arrived, self._format), use_bin_type=True))
-        temp.replace(self.path)
+    @contextmanager
+    def _write_transaction(self) -> Iterator[None]:
+        """The thread lock, then the cross-process lock on `<path>.lock`, held across the read and the replace so two
+        processes never rewrite from the same stale read."""
+        with self._lock:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            with store_write_lock(self.path, self.lock_timeout):
+                yield
+
+    def _replace_all(self, supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope]) -> None:
+        commit_store_file(self.path, msgpack.packb(_encode_snapshot(supplied, arrived, self._format), use_bin_type=True))
 
-    @staticmethod
-    def _msgpack() -> Any:
-        try:
-            import msgpack  # type: ignore
-        except ModuleNotFoundError as exc:
-            raise RuntimeError(
-                "SingleFileMessagePackBackingStore requires the optional 'msgpack' dependency. "
-                "Install with: python -m pip install cultcache-py[msgpack]"
-            ) from exc
-        return msgpack
 
 
 def canonical(candidates: Any) -> list[CultCacheSchemaCatalogEntry]:
@@ -261,6 +264,85 @@ def _encode_snapshot(
     ]
     return [format_version, catalog, records]
 
+def commit_store_file(path: Path, data: bytes) -> None:
+    """Writes `data` to a unique staged file beside the store, fsyncs it, replaces the store with it and, off Windows,
+    fsyncs the directory so the replace is durable. A failure before the replace removes the staged file and leaves the
+    store untouched. The caller holds the store lock."""
+    staged = path.with_name(f"{path.name}.tmp-{os.getpid()}-{uuid.uuid4().hex}")
+    try:
+        fd = os.open(staged, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_BINARY", 0), 0o666)
+        try:
+            view = memoryview(data)
+            while view:
+                view = view[os.write(fd, view):]
+            os.fsync(fd)
+        finally:
+            os.close(fd)
+        os.replace(staged, path)
+    except BaseException:
+        try:
+            os.unlink(staged)
+        except OSError:
+            pass
+        raise
+    if sys.platform != "win32":
+        directory = os.open(path.parent, os.O_RDONLY)
+        try:
+            os.fsync(directory)
+        finally:
+            os.close(directory)
+
+
+@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,
+    LockFileEx over the whole file on Windows, polled every 10 ms with no handle held between polls. The lock file is
+    created if absent and never written or deleted."""
+    lock_path = path.with_name(path.name + ".lock")
+    deadline = time.monotonic() + timeout
+    while True:
+        release = _try_lock(lock_path)
+        if release is not None:
+            break
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"CultCache timed out after {timeout} s waiting for the store lock {lock_path}.")
+        time.sleep(_LOCK_POLL_SECONDS)
+    try:
+        yield
+    finally:
+        release()
+
+
+def _try_lock(lock_path: Path) -> Callable[[], None] | None:
+    if sys.platform == "win32":
+        return _try_lock_windows(lock_path)
+    import fcntl
+
+    fd = os.open(lock_path, os.O_RDWR | os.O_CREAT, 0o666)
+    try:
+        fcntl.flock(fd, fcntl.LOCK_EX | fcntl.LOCK_NB)
+    except OSError:
+        os.close(fd)
+        return None
+    return lambda: os.close(fd)  # closing drops the flock
+
+
+def _try_lock_windows(lock_path: Path) -> Callable[[], None] | None:
+    import ctypes
+    import msvcrt
+    from ctypes import wintypes
+
+    kernel = ctypes.WinDLL("kernel32", use_last_error=True)
+    kernel.LockFileEx.argtypes = [wintypes.HANDLE, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, wintypes.DWORD, ctypes.c_void_p]
+    fd = os.open(lock_path, os.O_RDWR | os.O_CREAT | os.O_BINARY, 0o666)
+    handle = msvcrt.get_osfhandle(fd)
+    overlapped = ctypes.create_string_buffer(32)
+    # LOCKFILE_EXCLUSIVE_LOCK | LOCKFILE_FAIL_IMMEDIATELY over the whole range, as the TypeScript runtime takes it.
+    if not kernel.LockFileEx(handle, 0x3, 0, 0xFFFFFFFF, 0xFFFFFFFF, overlapped):
+        os.close(fd)
+        return None
+    return lambda: os.close(fd)  # closing drops the lock
+
 
 def _read_store(path: Path, read: Callable[[Path], _Contents]) -> _Contents | None:
     """Reads a store file, or returns None when nothing is at its path.
@@ -351,7 +433,7 @@ def _decode_snapshot(decoded: Any) -> tuple[str, list[CultCacheEnvelope]] | None
     return decoded[0], envelopes
 
 
-def _decode_legacy_envelopes(decoded: Any, msgpack: Any) -> list[CultCacheEnvelope]:
+def _decode_legacy_envelopes(decoded: Any) -> list[CultCacheEnvelope]:
     if not isinstance(decoded, list):
         raise ValueError("CultCache MessagePack store is not a recognized snapshot")
     envelopes: list[CultCacheEnvelope] = []

packages/cultcache-py/tests/test_slot_nil.py

@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+
+import msgpack
+
+from cultcache_py import database_entry_field, define_database_entry_type, define_record_codec, leaves as t
+
+# tests/vectors/slot-nil is written only by the C# reference (SlotNilVectorTests): what it reads and rewrites for a nil, a
+# str in a byte[] slot, and a record short of its last slots. This runtime accepts what the reference accepts and rewrites
+# the same bytes.
+VECTORS = Path(__file__).resolve().parents[3] / "tests" / "vectors" / "slot-nil"
+
+
+def _probe():
+    inner = define_record_codec("SlotNilInner", [database_entry_field("x", 0, default="", codec=t.string)])
+    return define_database_entry_type(
+        "tests.slot-nil",
+        [
+            database_entry_field("s", 0, default="", codec=t.nullable(t.string)),
+            database_entry_field("i", 1, default=0, codec=t.int32),
+            database_entry_field("b", 2, default=b"", codec=t.nullable(t.bytes)),
+            database_entry_field("r", 3, default=None, codec=t.nullable(inner)),
+            database_entry_field("l", 4, default=[], codec=t.nullable(t.list(t.string))),
+        ],
+    )
+
+
+class SlotNilVectorTests(unittest.TestCase):
+    def test_slots_read_and_rewrite_exactly_what_the_reference_does_for_nil_str_in_bytes_and_short_records(self) -> None:
+        rows = msgpack.unpackb((VECTORS / "reads.msgpack").read_bytes(), raw=False)
+        probe = _probe()
+        self.assertEqual(17, len(rows))
+        accepted = 0
+        for name, payload, was_accepted, rewritten in rows:
+            with self.subTest(name):
+                if was_accepted:
+                    self.assertEqual(rewritten.hex(), probe.encode_payload(probe.decode_payload(payload)).hex())
+                    accepted += 1
+                else:
+                    with self.assertRaises(Exception):
+                        probe.decode_payload(payload)
+        self.assertEqual(15, accepted)
+
+    def test_a_character_outside_the_bmp_is_four_utf8_bytes(self) -> None:
+        rows = {name: (payload, rewritten) for name, payload, _, rewritten in msgpack.unpackb((VECTORS / "reads.msgpack").read_bytes(), raw=False)}
+        probe = _probe()
+        self.assertEqual("\U0001F600", probe.decode_payload(rows["string-astral"][0])["s"])
+        self.assertEqual(b"\xf0\x9f\x98\x80", probe.decode_payload(rows["bytes-str-astral"][0])["b"])
+
+
+if __name__ == "__main__":
+    unittest.main()

packages/cultcache-py/tests/test_store_lock.py

@@ -0,0 +1,135 @@
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+from unittest import mock
+
+from cultcache_py import SingleFileMessagePackBackingStore, stores
+from cultcache_py.backing_store import CultCacheEnvelope
+
+REPO = Path(__file__).resolve().parents[3]
+TS_STORE_FILE = REPO / "packages" / "cultcache-ts" / "dist" / "store-file.js"
+HOLD_SECONDS = 1.5
+
+# A holder that takes the lock the way the Rust runtime does (an exclusive file lock), not through this package.
+PY_HOLDER = """
+import fcntl, os, sys, time
+fd = os.open(sys.argv[1] + '.lock', os.O_RDWR | os.O_CREAT, 0o666)
+fcntl.flock(fd, fcntl.LOCK_EX)
+print('HELD', flush=True)
+time.sleep(float(sys.argv[2]))
+"""
+# A holder that is the TypeScript runtime's own withStoreWriteLock, built from packages/cultcache-ts.
+TS_HOLDER = """
+const { withStoreWriteLock } = require(process.argv[1]);
+withStoreWriteLock(process.argv[2], async () => {
+  console.log('HELD');
+  await new Promise((r) => setTimeout(r, Number(process.argv[3]) * 1000));
+});
+"""
+
+
+def envelope(key: str) -> CultCacheEnvelope:
+    return CultCacheEnvelope(key=key, type="t", payload=b"\x01", stored_at="2026-11-06T00:00:00Z")
+
+
+def start_holder(argv: list[str]) -> subprocess.Popen[str]:
+    holder = subprocess.Popen(argv, stdout=subprocess.PIPE, text=True)
+    assert holder.stdout is not None and holder.stdout.readline().strip() == "HELD"
+    holder.held_at = time.monotonic()  # type: ignore[attr-defined]
+    return holder
+
+
+@unittest.skipIf(sys.platform == "win32", "the foreign holders here use flock; Windows is covered by the TS holder")
+class StoreLockTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.dir = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True)
+        self.path = self.dir / "state.cc"
+
+    def assert_write_waits_for(self, holder: subprocess.Popen[str]) -> None:
+        self.addCleanup(holder.wait)
+        self.addCleanup(holder.stdout.close)  # type: ignore[union-attr]
+        refused = SingleFileMessagePackBackingStore(self.path, lock_timeout=0.2)
+        with self.assertRaises(TimeoutError):
+            refused.push(envelope("blocked"))
+        self.assertFalse(self.path.exists(), "a write that could not take the lock must not touch the store")
+
+        SingleFileMessagePackBackingStore(self.path, lock_timeout=10).push(envelope("after"))
+        self.assertGreaterEqual(time.monotonic() - holder.held_at, HOLD_SECONDS - 0.1, "the write finished while the holder still held the lock")  # type: ignore[attr-defined]
+        self.assertEqual(["after"], [e.key for e in SingleFileMessagePackBackingStore(self.path).pull_all()])
+
+    def test_write_waits_while_a_foreign_process_holds_the_lock(self) -> None:
+        self.assert_write_waits_for(start_holder([sys.executable, "-c", PY_HOLDER, str(self.path), str(HOLD_SECONDS)]))
+
+    @unittest.skipUnless(shutil.which("node") and TS_STORE_FILE.exists(), "needs node and a built packages/cultcache-ts (npm run build)")
+    def test_write_waits_while_the_typescript_runtime_holds_the_lock(self) -> None:
+        holder = start_holder(["node", "-e", TS_HOLDER, str(TS_STORE_FILE), str(self.path), str(HOLD_SECONDS)])
+        self.assert_write_waits_for(holder)
+
+    def test_the_lock_is_released_after_each_write(self) -> None:
+        store = SingleFileMessagePackBackingStore(self.path, lock_timeout=0.2)
+        for key in "abc":
+            store.push(envelope(key))
+        self.assertEqual(["a", "b", "c"], sorted(e.key for e in store.pull_all()))
+
+    def test_concurrent_processes_lose_no_writes(self) -> None:
+        script = (
+            "import sys\n"
+            "from cultcache_py import SingleFileMessagePackBackingStore as S\n"
+            "from cultcache_py.backing_store import CultCacheEnvelope as E\n"
+            "s = S(sys.argv[1])\n"
+            "for i in range(25):\n"
+            "    s.push(E(key=f'{sys.argv[2]}{i}', type='t', payload=b'\\x01', stored_at='x'))\n"
+        )
+        env = {**os.environ, "PYTHONPATH": str(Path(stores.__file__).parents[1])}
+        procs = [subprocess.Popen([sys.executable, "-c", script, str(self.path), tag], env=env) for tag in "pqr"]
+        for proc in procs:
+            self.assertEqual(0, proc.wait())
+        self.assertEqual(75, len(SingleFileMessagePackBackingStore(self.path).pull_all()))
+
+
+class StagedWriteTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.dir = Path(tempfile.mkdtemp())
+        self.addCleanup(shutil.rmtree, self.dir, ignore_errors=True)
+        self.path = self.dir / "state.cc"
+
+    def test_the_staged_file_is_fsynced_before_the_replace(self) -> None:
+        order: list[str] = []
+        real_fsync, real_replace = os.fsync, os.replace
+        with mock.patch("os.fsync", side_effect=lambda fd: (order.append("fsync"), real_fsync(fd))[1]), mock.patch(
+            "os.replace", side_effect=lambda a, b: (order.append("replace"), real_replace(a, b))[1]
+        ):
+            SingleFileMessagePackBackingStore(self.path).push(envelope("a"))
+        self.assertEqual("fsync", order[0])
+        self.assertLess(order.index("fsync"), order.index("replace"))
+
+    def test_the_staged_name_is_unique_per_write(self) -> None:
+        names: list[str] = []
+        real_replace = os.replace
+        with mock.patch("os.replace", side_effect=lambda a, b: (names.append(os.fspath(a)), real_replace(a, b))[1]):
+            store = SingleFileMessagePackBackingStore(self.path)
+            store.push(envelope("a"))
+            store.push(envelope("b"))
+        self.assertEqual(2, len(set(names)))
+        self.assertNotIn(os.fspath(self.path) + ".tmp", names)
+
+    def test_a_failed_replace_leaves_the_store_and_no_staged_file(self) -> None:
+        store = SingleFileMessagePackBackingStore(self.path)
+        store.push(envelope("a"))
+        before = self.path.read_bytes()
+        with mock.patch("os.replace", side_effect=OSError("boom")), self.assertRaises(OSError):
+            store.push(envelope("b"))
+        self.assertEqual(before, self.path.read_bytes())
+        self.assertEqual([], [p.name for p in self.dir.iterdir() if ".tmp" in p.name])
+
+
+if __name__ == "__main__":
+    unittest.main()