GameCult/CultLib
Watch
9
0
Fork
You've already forked CultLib
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

cultcache-py: single-file store takes the cross-process lock, stages uniquely and fsyncs

Browse source 
push, push_all and delete hold <path>.lock (flock; LockFileEx over the whole file on
Windows, as cultcache-ts store-file.ts) across the read and the replace. The staged file
is <path>.tmp-<pid>-<uuid>, fsynced before os.replace, with a directory fsync after off
Windows; a failed write removes its staged file and leaves the store.

Tests fail when the rule breaks: a write waits while a flock holder (Rust's lock) and the
TypeScript runtime's withStoreWriteLock hold the lock, three processes lose no writes,
fsync precedes replace, staged names differ, a failed replace leaves no staged file.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-06 20:49:32 +00:00
commit b2434fc7fc
2 changed files with 240 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

130
packages/cultcache-py/src/cultcache_py/stores.py
View file

@ -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] = []

135
packages/cultcache-py/tests/test_store_lock.py Normal file
View file

@ -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()