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

cultcache-py: a write at a store path that is a link is refused before lock and replace (StoreLinkedError)

Browse source 
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-12 21:58:30 +00:00
commit dcf70ea5b8
3 changed files with 58 additions and 9 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/cultcache-py/src/cultcache_py/__init__.py
View file

@ -19,7 +19,7 @@ from .documents import (
read_payload_tree,
)
from .leaves import SlotCodec
from .stores import JsonLinesBackingStore, SchemaConflictError, SingleFileMessagePackBackingStore, StoreUnreadableError
from .stores import JsonLinesBackingStore, SchemaConflictError, SingleFileMessagePackBackingStore, StoreLinkedError, StoreUnreadableError
__all__ = [
"BackingStore",
@ -38,6 +38,7 @@ __all__ = [
"SchemaConflictError",
"SingleFileMessagePackBackingStore",
"SlotCodec",
"StoreLinkedError",
"StoreUnreadableError",
"UnionArm",
"UnreadSlots",

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

@ -440,11 +440,33 @@ class WindowsReplaceKernel(ReplaceKernel):
return 0 if self._k.MoveFileExW(staged, destination, flags) else self._ctypes.get_last_error()
class StoreLinkedError(OSError):
"""A write was asked of a store path that is a link: a replace there would swap the link for a regular file."""
def __init__(self, path: Path) -> None:
super().__init__(f"CultCache store {path} is a link, and a write at it would replace the link; configure the link's target as the store path.")
self.path = path
def _refuse_linked(path: Path) -> None:
"""Refuses a link at the store path: a file or directory link, and on Windows a junction or any reparse point (islink misses
junctions before Python 3.12). Absent is not a link."""
linked = os.path.islink(path)
if not linked and sys.platform == "win32":
try:
linked = bool(os.lstat(path).st_file_attributes & 0x400) # FILE_ATTRIBUTE_REPARSE_POINT
except FileNotFoundError:
pass
if linked:
raise StoreLinkedError(path)
@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."""
_refuse_linked(path)
lock_path = path.with_name(path.name + ".lock")
deadline = time.monotonic() + timeout
while True:

42
packages/cultcache-py/tests/test_cultcache.py
View file

@ -12,6 +12,7 @@ from cultcache_py import (
CultCache,
JsonLinesBackingStore,
SingleFileMessagePackBackingStore,
StoreLinkedError,
StoreUnreadableError,
define_database_entry_type,
define_document_type,
@ -1431,10 +1432,10 @@ class CultCacheTests(unittest.TestCase):
self.assertNotIn("SECRET", str(caught.exception))
self.assertIn("not a valid store", str(caught.exception))
# What a write does through a symbolic link today. R3 decides whether a write resolves
# the link or refuses it; this pins the current behaviour so that change is deliberate.
# A write at a store path that is a link is refused: a replace there would swap the link for a regular file and
# detach the store from its target. Reads through a live link keep working.
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
def test_r3_decides_resolve_or_refuse_a_push_through_a_live_link_replaces_the_link(self) -> None:
def test_a_write_at_a_live_link_is_refused_and_nothing_changes_while_a_read_through_it_works(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
volume = Path(tmp) / "volume"
volume.mkdir()
@ -1442,12 +1443,36 @@ class CultCacheTests(unittest.TestCase):
envelope = CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z")
SingleFileMessagePackBackingStore(target).push(envelope)
before = target.read_bytes()
link = Path(tmp) / "store.cc"
volume_before = sorted(p.name for p in volume.iterdir())
link = Path(tmp) / "link.cc"
link.symlink_to(target)
SingleFileMessagePackBackingStore(link).push(replace(envelope, key="b"))
self.assertFalse(link.is_symlink())
self.assertTrue(link.is_file())
store = SingleFileMessagePackBackingStore(link)
for write in (
lambda: store.push(replace(envelope, key="b")),
lambda: store.push_all([replace(envelope, key="b")]),
lambda: store.delete("t", "a"),
):
with self.assertRaises(StoreLinkedError) as caught:
write()
self.assertIn(str(link), str(caught.exception))
self.assertTrue(link.is_symlink())
self.assertEqual(target.read_bytes(), before)
self.assertEqual(sorted(p.name for p in Path(tmp).iterdir()), ["link.cc", "volume"])
self.assertEqual(sorted(p.name for p in volume.iterdir()), volume_before)
self.assertEqual([item.key for item in store.pull_all()], ["a"])
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
def test_a_write_at_a_live_directory_link_is_refused_as_a_link(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
real = Path(tmp) / "real"
real.mkdir()
link = Path(tmp) / "dirlink"
link.symlink_to(real, target_is_directory=True)
with self.assertRaises(StoreLinkedError):
SingleFileMessagePackBackingStore(link).push(
CultCacheEnvelope(key="a", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
self.assertTrue(link.is_symlink())
self.assertEqual(list(real.iterdir()), [])
@unittest.skipIf(os.name != "posix", "symbolic links need POSIX here")
def test_a_dangling_symlink_at_a_store_path_is_an_os_error_and_nothing_is_written(self) -> None:
@ -1460,7 +1485,8 @@ class CultCacheTests(unittest.TestCase):
store = store_type(path)
with self.assertRaises(FileNotFoundError):
store.pull_all()
with self.assertRaises(FileNotFoundError):
# The single-file store refuses the link itself; the unlocked JSON lines store meets the missing target.
with self.assertRaises(StoreLinkedError if store_type is SingleFileMessagePackBackingStore else FileNotFoundError):
store.push(CultCacheEnvelope(key="k", type="t", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
self.assertTrue(path.is_symlink())
self.assertEqual(list(volume.iterdir()), [])