From 5be87c76960211795d3259c0b10bee08ff709255 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Mon, 14 Sep 2026 12:27:55 +0200 Subject: [PATCH] Refuse re-entrant Python cache mutations and adopt legacy-key globals on load Attach, pull, registration, writes and deletes raise when started on a thread already inside a mutation of the same cache (a store call, decoder, encoder, extractor or updater), instead of running inside it and corrupting the state being built. Reads stay reentrant. Internal delegation (put_global, update, delete_global, register_registry, add_backing_store) goes through private helpers so it is not re-entry. A pull adopts a global persisted under a key other than __global__ as that type's global without writing. The first write or delete of it removes the legacy record in the same call; a failed removal takes the new record back out. Two globals of one type on disk are still refused. The lock and unlocked-read comments now state the per-call guarantee. Co-Authored-By: Claude Opus 5 --- .../cultcache-py/src/cultcache_py/cache.py | 129 ++++++++++++++---- packages/cultcache-py/tests/test_cultcache.py | 102 ++++++++++++++ 2 files changed, 201 insertions(+), 30 deletions(-) diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 47edc782..0da2c04a 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -2,7 +2,7 @@ import functools import threading -from dataclasses import dataclass, field +from dataclasses import dataclass, field, replace from typing import Any, Callable, Generic, TypeVar from .backing_store import BackingStore, CultCacheEnvelope @@ -14,8 +14,7 @@ GLOBAL_KEY = "__global__" def _locked(method: F) -> F: - """Runs the method holding the cache's lock, so its validation, store I/O and apply cannot - interleave with another thread's attach, pull, registration, write or delete.""" + """Runs a read holding the cache's lock. Reads are reentrant and may run inside a mutation.""" @functools.wraps(method) def wrapper(self: "CultCache", *args: Any, **kwargs: Any) -> Any: @@ -25,6 +24,29 @@ def _locked(method: F) -> F: return wrapper # type: ignore[return-value] +def _mutating(method: F) -> F: + """Runs an attach, pull, registration, write or delete holding the cache's lock, so its validation, + store I/O and apply cannot interleave with another thread's. One started on a thread that is already + inside a mutation of this cache (from a store call, decoder, encoder, extractor or updater) raises + before touching anything.""" + + @functools.wraps(method) + def wrapper(self: "CultCache", *args: Any, **kwargs: Any) -> Any: + if self._mutating_thread == threading.get_ident(): + raise CultCacheError( + f"CultCache refuses a re-entrant {method.__name__}: a mutation cannot start from inside a store call, " + "decoder, encoder, extractor or updater of another mutation on the same cache" + ) + with self._lock: + self._mutating_thread = threading.get_ident() + try: + return method(self, *args, **kwargs) + finally: + self._mutating_thread = None + + return wrapper # type: ignore[return-value] + + class CultCacheError(RuntimeError): pass @@ -44,6 +66,11 @@ class _State: # The name and index values each held record produced when it was admitted. Removing a # record reads these and runs no extractor, so removal has no fallible step. lookup_keys: dict[str, dict[str, "_LookupKeys"]] = field(default_factory=dict) + # Load compatibility shim: a global persisted under a key other than GLOBAL_KEY is held as that + # type's global under GLOBAL_KEY, and its persisted key is kept here so the first write or delete + # of that global removes it from the home store. Remove once no store holds a global under a key + # other than GLOBAL_KEY. + legacy_global_keys: dict[str, str] = field(default_factory=dict) @dataclass(frozen=True) @@ -98,14 +125,19 @@ class CultCache: def __init__(self) -> None: self._state = _State() - # Reentrant: extractors, decoders and updaters may read or write this cache from inside a - # held call. Single-structure reads (get, get_all, get_envelope, key lookups) stay unlocked: - # each reads one dict under the GIL, and every write installs a record's value, envelope and - # lookups before releasing the lock, so they never observe a store/cache split. + # Mutations hold this lock and refuse to start on the thread already inside one (see _mutating). + # It is reentrant so a mutation's extractors, decoders and updaters may still read the cache. + # Single-structure reads (get, get_all, get_envelope, key lookups) stay unlocked. The guarantee is + # per call: each reads one dict under the GIL and sees it wholly before or after a write's install. + # Two unlocked calls can straddle a write or pull, and see different states. self._lock = threading.RLock() + self._mutating_thread: int | None = None - @_locked + @_mutating def register_document_type(self, document: DocumentDefinition[Any]) -> None: + self._register_document_type(document) + + def _register_document_type(self, document: DocumentDefinition[Any]) -> None: if document.type in self._state.documents: raise CultCacheError(f"Document type already registered: {document.type}") schema_name = document.catalog_entry().schema_name @@ -121,19 +153,19 @@ class CultCache: self._state.documents[document.type] = document self._state.documents_by_schema_name[schema_name] = document - @_locked + @_mutating def register_registry(self, documents: list[DocumentDefinition[Any]] | tuple[DocumentDefinition[Any], ...]) -> None: for document in documents: - self.register_document_type(document) + self._register_document_type(document) - @_locked + @_mutating def register_name_lookup(self, document: DocumentDefinition[Any], extractor: str | Any) -> None: self._assert_registered(document) name_extractors = dict(self._state.name_extractors) name_extractors[document.type] = extractor self._install_lookups(name_extractors, self._state.index_extractors) - @_locked + @_mutating def register_index(self, document: DocumentDefinition[Any], index: str, extractor: str | Any) -> None: self._assert_registered(document) index_extractors = {type: dict(indexes) for type, indexes in self._state.index_extractors.items()} @@ -154,7 +186,7 @@ class CultCache: self._state.lookup_keys, ) = (name_extractors, index_extractors, names, indexes, lookup_keys) - @_locked + @_mutating def add_backing_store(self, store: BackingStore, types: list[str] | tuple[str, ...] | set[str]) -> None: """Makes the store home to the given types, or the generic store when there are none. @@ -162,7 +194,7 @@ class CultCache: """ types = list(types) if not types: - self.add_generic_store(store) + self._add_generic_store(store) return for type in types: if type in self._state.stores_by_type: @@ -174,9 +206,12 @@ class CultCache: self._refuse_home_moves(routes, self._state.generic_store) self._state.stores_by_type = routes - @_locked + @_mutating def add_generic_store(self, store: BackingStore) -> None: """Makes the store home to every type no other store claims. A cache has at most one.""" + self._add_generic_store(store) + + def _add_generic_store(self, store: BackingStore) -> None: if self._state.generic_store is not None: raise CultCacheError( "Backing store would be a second generic store; name the types it is home to" @@ -204,13 +239,14 @@ class CultCache: return "the generic store" return "the store routed to " + ", ".join(sorted(type for type, routed in routes.items() if routed is store)) - @_locked + @_mutating def pull_all_backing_stores(self) -> None: """Builds the complete next state (values, envelopes, name and index lookups) in local structures, running every check and every user extractor; the cache's state is replaced only once that build finishes, so a refused load admits nothing.""" loaded_values: dict[str, dict[str, Any]] = {} loaded_envelopes: dict[str, dict[str, CultCacheEnvelope]] = {} + loaded_legacy_global_keys: dict[str, str] = {} routes, generic = self._state.stores_by_type, self._state.generic_store seen_globals: set[str] = set() stores: list[BackingStore] = [] @@ -241,6 +277,9 @@ class CultCache: if envelope.type in seen_globals: raise CultCacheError(f"Duplicate global document for type: {envelope.type}") seen_globals.add(envelope.type) + if envelope.key != GLOBAL_KEY: + loaded_legacy_global_keys[envelope.type] = envelope.key + envelope = replace(envelope, key=GLOBAL_KEY) value = document.decode_payload(envelope.payload) loaded_values.setdefault(envelope.type, {})[envelope.key] = value loaded_envelopes.setdefault(envelope.type, {})[envelope.key] = envelope @@ -254,7 +293,8 @@ class CultCache: self._state.names, self._state.indexes, self._state.lookup_keys, - ) = (loaded_values, loaded_envelopes, loaded_names, loaded_indexes, loaded_lookup_keys) + self._state.legacy_global_keys, + ) = (loaded_values, loaded_envelopes, loaded_names, loaded_indexes, loaded_lookup_keys, loaded_legacy_global_keys) def get(self, document: DocumentDefinition[T], key: str) -> T | None: self._assert_registered(document) @@ -310,8 +350,11 @@ class CultCache: key = self.get_key_by_index(document, index, value) return None if key is None else self.get(document, key) - @_locked + @_mutating def put(self, document: DocumentDefinition[T], key: str, value: T) -> None: + self._put(document, key, value) + + def _put(self, document: DocumentDefinition[T], key: str, value: T) -> None: self._assert_registered(document) if document.global_document and key != GLOBAL_KEY: raise CultCacheError(f"Global document {document.type} must use key {GLOBAL_KEY}") @@ -325,7 +368,7 @@ class CultCache: ) self._write(document, [(envelope, value)], batch=False) - @_locked + @_mutating def put_envelope(self, document: DocumentDefinition[T], envelope: CultCacheEnvelope) -> T: self._assert_registered(document) self._check_envelope(document, envelope) @@ -333,7 +376,7 @@ class CultCache: self._write(document, [(envelope, value)], batch=False) return value - @_locked + @_mutating def put_envelopes(self, document: DocumentDefinition[T], envelopes: list[CultCacheEnvelope]) -> list[T]: self._assert_registered(document) values: list[T] = [] @@ -367,11 +410,15 @@ class CultCache: for envelope, value in records ] store = self._store_for_type(type) + legacy_key = self._state.legacy_global_keys.get(type) if store is not None: if batch: store.push_all([envelope for envelope, _ in records]) else: store.push(records[0][0]) + if legacy_key is not None: + self._retire_legacy_global(store, type, legacy_key) + self._state.legacy_global_keys.pop(type, None) values = self._state.values.setdefault(type, {}) envelopes = self._state.envelopes.setdefault(type, {}) lookup_keys = self._state.lookup_keys.setdefault(type, {}) @@ -385,29 +432,51 @@ class CultCache: for index, index_value in keys.indexes.items(): self._state.indexes[(type, index, index_value)] = envelope.key - @_locked + @staticmethod + def _retire_legacy_global(store: BackingStore, type: str, legacy_key: str) -> None: + """The first write of a global loaded from a legacy key removes that record in the same write. If + the removal fails, the new record is taken back out, so the store keeps only the legacy record.""" + try: + store.delete(type, legacy_key) + except BaseException: + try: + store.delete(type, GLOBAL_KEY) + except Exception: # noqa: BLE001 - the removal failure is the error reported + pass + raise + + @_mutating def put_global(self, document: DocumentDefinition[T], value: T) -> None: self._assert_global(document) - self.put(document, GLOBAL_KEY, value) + self._put(document, GLOBAL_KEY, value) - @_locked + @_mutating def update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T: + return self._update(document, key, updater) + + def _update(self, document: DocumentDefinition[T], key: str, updater: Any) -> T: current = self.get_required(document, key) updated = updater(current) - self.put(document, key, updated) + self._put(document, key, updated) return updated - @_locked + @_mutating def update_global(self, document: DocumentDefinition[T], updater: Any) -> T: self._assert_global(document) - return self.update(document, GLOBAL_KEY, updater) + return self._update(document, GLOBAL_KEY, updater) - @_locked + @_mutating def delete(self, document: DocumentDefinition[Any], key: str) -> None: + self._delete(document, key) + + def _delete(self, document: DocumentDefinition[Any], key: str) -> None: self._assert_registered(document) store = self._store_for_type(document.type) + legacy_key = self._state.legacy_global_keys.get(document.type) if key == GLOBAL_KEY else None if store is not None: - store.delete(document.type, key) + store.delete(document.type, key if legacy_key is None else legacy_key) + if legacy_key is not None: + del self._state.legacy_global_keys[document.type] self._remove_lookups(document.type, key) for by_type in (self._state.values, self._state.envelopes, self._state.lookup_keys): held = by_type.get(document.type) @@ -416,10 +485,10 @@ class CultCache: if not held: by_type.pop(document.type, None) - @_locked + @_mutating def delete_global(self, document: DocumentDefinition[Any]) -> None: self._assert_global(document) - self.delete(document, GLOBAL_KEY) + self._delete(document, GLOBAL_KEY) # Snapshots iterate dicts a concurrent write would resize mid-iteration, so they hold the lock. @_locked diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 5a5352c0..ac81cc6d 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -5,6 +5,7 @@ import threading import unittest from dataclasses import asdict, dataclass, replace from pathlib import Path +from cultcache_py.backing_store import CultCacheEnvelope from cultcache_py.cache import CultCacheError from cultcache_py import ( CultCache, @@ -628,6 +629,107 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(on_disk[0].payload, cache.get_required_envelope(settings, CultCache.GLOBAL_KEY).payload) self.assertEqual(cache.get_required_global(settings), {"theme": "second"}) + @staticmethod + def _envelope(document, key: str, value) -> CultCacheEnvelope: + catalog_entry = document.catalog_entry() + return CultCacheEnvelope.create( + key=key, + type=document.type, + payload=document.encode_payload(value), + schema_id=catalog_entry.schema_id, + catalog_entry=catalog_entry, + ) + + def test_legacy_key_global_loads_without_writing_and_first_write_leaves_one_global(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + settings = define_document_type("settings", global_document=True) + path = Path(tmp) / "generic.cc" + + def build() -> CultCache: + return CultCache.builder().register_document_type(settings).add_generic_store( + SingleFileMessagePackBackingStore(path) + ).build() + + SingleFileMessagePackBackingStore(path).push(self._envelope(settings, "legacy", {"theme": "old"})) + stored_bytes = path.read_bytes() + + cache = build() + cache.pull_all_backing_stores() + self.assertEqual(cache.get_global(settings), {"theme": "old"}) + self.assertEqual(cache.get(settings, CultCache.GLOBAL_KEY), {"theme": "old"}) + self.assertEqual(path.read_bytes(), stored_bytes) + + cache.put_global(settings, {"theme": "new"}) + self.assertEqual([e.key for e in SingleFileMessagePackBackingStore(path).pull_all()], [CultCache.GLOBAL_KEY]) + reloaded = build() + reloaded.pull_all_backing_stores() + self.assertEqual(reloaded.get_required_global(settings), {"theme": "new"}) + + def test_deleting_a_legacy_key_global_removes_the_legacy_record(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + settings = define_document_type("settings", global_document=True) + path = Path(tmp) / "generic.cc" + SingleFileMessagePackBackingStore(path).push(self._envelope(settings, "legacy", {"theme": "old"})) + cache = CultCache.builder().register_document_type(settings).add_generic_store( + SingleFileMessagePackBackingStore(path) + ).build() + cache.pull_all_backing_stores() + + cache.delete_global(settings) + self.assertEqual(SingleFileMessagePackBackingStore(path).pull_all(), []) + self.assertIsNone(cache.get_global(settings)) + + def test_two_globals_of_one_type_on_disk_are_refused_under_any_keys(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + settings = define_document_type("settings", global_document=True) + for keys in (["one", "two"], ["legacy", CultCache.GLOBAL_KEY]): + path = Path(tmp) / f"{keys[0]}.cc" + store = SingleFileMessagePackBackingStore(path) + store.push_all([self._envelope(settings, key, {"theme": key}) for key in keys]) + cache = CultCache.builder().register_document_type(settings).add_generic_store(store).build() + with self.assertRaisesRegex(CultCacheError, "Duplicate global document for type: settings"): + cache.pull_all_backing_stores() + self.assertIsNone(cache.get_global(settings)) + + def test_decoder_writing_during_pull_raises_and_changes_neither_store_nor_cache(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "generic.cc" + holder: dict[str, CultCache] = {} + + def decode(raw): + holder["cache"].put(item, "side", {"name": "side"}) + return raw + + item = define_document_type("item", decode=decode, encode=lambda value: value) + store = SingleFileMessagePackBackingStore(path) + store.push(self._envelope(item, "a", {"name": "a"})) + stored_bytes = path.read_bytes() + cache = holder["cache"] = CultCache.builder().register_document_type(item).add_generic_store(store).build() + + with self.assertRaisesRegex(CultCacheError, "re-entrant put"): + cache.pull_all_backing_stores() + self.assertEqual(path.read_bytes(), stored_bytes) + self.assertEqual(cache.snapshot_envelopes(), []) + self.assertIsNone(cache.get(item, "side")) + # The refusal released the cache: an ordinary write still works. + cache.put(item, "b", {"name": "b"}) + self.assertEqual(cache.get(item, "b"), {"name": "b"}) + + def test_index_extractor_writing_during_register_index_raises_and_installs_nothing(self) -> None: + thing = define_document_type("thing") + cache = CultCache.builder().register_document_type(thing).build() + cache.put(thing, "k1", {"n": "one", "cat": "x"}) + + def category(value: dict) -> str: + cache.put(thing, "k2", {"n": "two", "cat": "y"}) + return value["cat"] + + with self.assertRaisesRegex(CultCacheError, "re-entrant put"): + cache.register_index(thing, "cat", category) + self.assertIsNone(cache.get(thing, "k2")) + self.assertNotIn("thing", cache._state.index_extractors) + self.assertIsNone(cache.get_key_by_index(thing, "cat", "x")) + def test_put_envelope_refuses_a_global_under_another_key(self) -> None: with tempfile.TemporaryDirectory() as tmp: settings = define_document_type("settings", global_document=True)