From af2f89be70f1bbb178485ba08c8d70c9a78a4dcc Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 22:35:12 +0000 Subject: [PATCH] cultcache-py: the cache holds one record per key and carries a store's unregistered records as foreign One key-identity judge in _write, an unheld-removal refusal at foreign keys, a load that carries unregistered records instead of refusing, and a load that refuses one key in two stores. The single-file store's commit guard stays the disk judge. Co-Authored-By: Claude Sonnet 5.5 --- docs/write-ownership-map.md | 4 +- packages/cultcache-py/CHANGELOG.md | 3 + packages/cultcache-py/README.md | 6 +- .../cultcache-py/src/cultcache_py/__init__.py | 3 +- .../cultcache-py/src/cultcache_py/cache.py | 102 ++++++++++++++++-- .../cultcache-py/src/cultcache_py/stores.py | 8 +- .../tests/test_key_identity_vector.py | 83 ++++++++++++++ 7 files changed, 194 insertions(+), 15 deletions(-) diff --git a/docs/write-ownership-map.md b/docs/write-ownership-map.md index e1f046cf..14d4bec1 100644 --- a/docs/write-ownership-map.md +++ b/docs/write-ownership-map.md @@ -33,8 +33,8 @@ Probes: scratch tests in this worktree, removed. C# filtered runs on Starfire (o | Probe | C# | TypeScript | Rust | Python | |---|---|---|---|---| -| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | C, and **both** records live at k | C, both live at k | C, both live at k | -| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R at load: "No schema is registered for persisted schema id" | R at load, same | R at load: "Unknown persisted document type" | +| P1 cache {A,B}: B at k, then A at k | **C**: B's record replaced, k now A's id | C, and **both** records live at k | C, both live at k | R `SchemaConflictError`, A's write refused, B kept (was C, both live) | +| P2 cache {A} only, store holds B at k, write A at k | R `CultSchemaConflictException` (B is foreign, carried) | R at load: "No schema is registered for persisted schema id" | R at load, same | R `SchemaConflictError` (B is foreign, carried; the load admits it; was R at load) | | P3 cacheless `WriteDocument` over B, registry null / {A} / {A,B} | R / R / R | no cacheless write | no cacheless write | no cacheless write | | P4 cache {A,B}: remove k and upsert A at k in one batch | C (k is A's) | n/a | n/a | n/a | | P5 cache {A,B}: remove, then add in a second commit | C | n/a | n/a | n/a | diff --git a/packages/cultcache-py/CHANGELOG.md b/packages/cultcache-py/CHANGELOG.md index e0026b8e..e8357a33 100644 --- a/packages/cultcache-py/CHANGELOG.md +++ b/packages/cultcache-py/CHANGELOG.md @@ -9,6 +9,9 @@ ### Breaking +- A cache holds one record per key whatever its type (`variants:ruling:cache-refuses-retype`). A `put`, `put_envelope`, `put_envelopes` or `put_global` of a second type under a key the cache holds, or under a key a store carries foreign, is refused with `SchemaConflictError` naming the key, even with no store, and changes nothing; retype a key by deleting it, then putting the new type. A write may replace another registered type's record only when the written type owns its schema id (declares it or lists it compatible) and both types share one home store. A load that reads one key from two stores, whatever the types, is refused with `CultCacheError` naming the key and both types, and admits nothing. A store that holds a record no registered type claims now loads and is carried: the cache does not serve it, the store keeps its bytes through every write, `delete` or `put` at its key is refused with `SchemaConflictError`, and registering a type that declares its schema id claims it at the next load. It raised `CultCacheError` "Unknown persisted document type". Two global types share the key `__global__`, so a second global type in one cache is now refused. +- Added `foreign_records()` and the frozen `ForeignRecord(key, schema_id, schema_name, stored_at)`, sorted by key in UTF-16 code-unit order. `stores.retyped` and `stores.written_owns` lose their underscore: the cache judge and the single-file store's commit guard ask one predicate. + - A single-file store holding two records under one key is unreadable: the v1 reader and the legacy envelope-array reader refuse it with `StoreUnreadableError` (a key is compared by its exact code points, no case folding or normalisation). A catalog slot whose canonical schema is not a string is refused the same way. A store that loaded before can now be refused: repair it with the runtime that wrote the duplicate. - A `put` or `push_all` that would land a second type under a key the store holds is refused with `SchemaConflictError`, and the file is left as it was. Retype a key by deleting it, then putting the new type. A write replaces the record at a key only when the written type owns it: its catalog entry declares the stored schema id (its own, or one it lists compatible), so a renamed type that lists the old id compatible may replace the record, or it has the stored record's schema name. This runtime does not compare shapes, so a same-name type is never refused for dropping persisted slots as C# refuses it. - Writes of the same type replace by key: the last record wins, in a batch as in the file. diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 236009ed..43e6c90e 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -19,7 +19,11 @@ discovery, sessions, local server) build on it, mirroring `cultcache-rs` / `[format, schemaCatalog, records]` - records store raw MessagePack payload bytes under schema-catalog identity - payloads decode only through the registered document definition for that type -- unknown persisted types fail closed +- a cache holds one record per key whatever its type: a write of a second type at a held key is refused with + `SchemaConflictError` (in memory too), and two stores holding one key refuse the load +- a store record no registered type claims loads and is carried as foreign: `cache.foreign_records()` lists it, the store + keeps its bytes through every write, a write or removal at its key is refused, and registering a type that declares its + schema id claims it at the next load - global documents are singleton-style per type - one home store per document type: a store added with `add_backing_store` owns its types, otherwise the generic store is the home; a type claimed by a diff --git a/packages/cultcache-py/src/cultcache_py/__init__.py b/packages/cultcache-py/src/cultcache_py/__init__.py index aabd21bf..dc73ab1b 100644 --- a/packages/cultcache-py/src/cultcache_py/__init__.py +++ b/packages/cultcache-py/src/cultcache_py/__init__.py @@ -1,7 +1,7 @@ from .backing_store import BackingStore, CultCacheEnvelope from .backing_store import CultCacheSchemaCatalogEntry, CultCacheSchemaCatalogMember from . import leaves -from .cache import CultCache, CultCacheBuilder +from .cache import CultCache, CultCacheBuilder, ForeignRecord from .documents import ( MAX_PAYLOAD_DEPTH, MAX_SLOT_KEY, @@ -29,6 +29,7 @@ __all__ = [ "CultCacheSchemaCatalogMember", "DatabaseEntryField", "DocumentDefinition", + "ForeignRecord", "JsonLinesBackingStore", "MAX_PAYLOAD_DEPTH", "MAX_SLOT_KEY", diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 9c402968..315e64ac 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -9,7 +9,7 @@ from typing import Any, Callable, Generic, Iterable, TypeVar from .backing_store import BackingStore, CultCacheEnvelope from .documents import DocumentDefinition, extract_value -from .stores import SchemaConflictError +from .stores import SchemaConflictError, retyped, written_owns T = TypeVar("T") F = TypeVar("F", bound=Callable[..., Any]) @@ -120,6 +120,20 @@ class _State: # 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) + # The records the last pull read that no registered type claims, sorted by key. Derived from the pull and swapped with the + # rest of the state, written nowhere else; it decides only refusals at its keys. A store keeps their bytes. + foreign: list["ForeignRecord"] = field(default_factory=list) + + +@dataclass(frozen=True) +class ForeignRecord: + """A record a store holds that no registered type claims: the cache does not serve it, refuses a write or a removal at its + key, and a store keeps its bytes. Registering a type that declares its schema id claims it at the next load.""" + + key: str + schema_id: str + schema_name: str + stored_at: str @dataclass(frozen=True) @@ -322,15 +336,23 @@ class CultCache: loaded_legacy_global_keys: dict[str, str] = {} routes, generic = self._state.stores_by_type, self._state.generic_store seen_globals: set[str] = set() + foreign: list[ForeignRecord] = [] + type_at_key: dict[str, tuple[str, int]] = {} # one record per key across every store: its type (or schema name) and store stores: list[BackingStore] = [] for candidate in [*self._state.stores_by_type.values(), self._state.generic_store]: if candidate is not None and all(candidate is not store for store in stores): stores.append(candidate) - for store in stores: + for position, store in enumerate(stores): for envelope in store.pull_all(): document = self._resolve_document_for_envelope(envelope) if document is None: - raise CultCacheError(f"Unknown persisted document type: {envelope.type}") + entry = envelope.catalog_entry + held_name = entry.schema_name if entry is not None else envelope.type + self._claim_key(type_at_key, envelope.key, held_name, position) + foreign.append( + ForeignRecord(envelope.key, envelope.schema_id or envelope.type, held_name, envelope.stored_at) + ) + continue if envelope.type != document.type: envelope = CultCacheEnvelope( key=envelope.key, @@ -353,6 +375,7 @@ class CultCache: if envelope.key != GLOBAL_KEY: loaded_legacy_global_keys[envelope.type] = envelope.key envelope = replace(envelope, key=GLOBAL_KEY) + self._claim_key(type_at_key, envelope.key, envelope.type, position) envelope = self._stamped(document, envelope) value = document.decode_payload(envelope.payload) loaded_values.setdefault(envelope.type, {})[envelope.key] = value @@ -369,7 +392,28 @@ class CultCache: self._state.indexes, self._state.lookup_keys, self._state.legacy_global_keys, - ) = (loaded_values, loaded_envelopes, loaded_names, loaded_indexes, loaded_lookup_keys, loaded_legacy_global_keys) + self._state.foreign, + ) = ( + loaded_values, + loaded_envelopes, + loaded_names, + loaded_indexes, + loaded_lookup_keys, + loaded_legacy_global_keys, + sorted(foreign, key=lambda record: _utf16_order(record.key)), + ) + + @staticmethod + def _claim_key(type_at_key: dict[str, tuple[str, int]], key: str, type: str, store: int) -> None: + """A key holds one record whatever its type, across every store a load reads. The refusal names the key and both types, + never a payload.""" + held, held_store = type_at_key.setdefault(key, (type, store)) + if held != type or held_store != store: + raise CultCacheError(f"Record {key!r} is held by both {held} and {type}: a key holds one record; the load admitted nothing.") + + def foreign_records(self) -> list[ForeignRecord]: + """The records the last load read that no registered type claims, sorted by key (UTF-16 code-unit order).""" + return list(self._state.foreign) def get(self, document: DocumentDefinition[T], key: str) -> T | None: self._assert_registered(document) @@ -497,6 +541,7 @@ class CultCache: (envelope, value, self._lookup_keys_for(value, name_extractor, index_extractors)) for envelope, value in records ] + displaced = self._refuse_retype(document, [envelope for envelope, _ in records]) self._refuse_index_sharing( [(type, envelope.key, keys) for envelope, _, keys in admitted], self._state.indexes, @@ -515,6 +560,8 @@ class CultCache: values = self._state.values.setdefault(type, {}) envelopes = self._state.envelopes.setdefault(type, {}) lookup_keys = self._state.lookup_keys.setdefault(type, {}) + for other, key in displaced: + self._drop(other, key) for envelope, value, keys in admitted: self._remove_lookups(type, envelope.key) values[envelope.key] = value @@ -525,6 +572,39 @@ class CultCache: for index, index_value in keys.indexes.items(): self._state.indexes[(type, index, index_value)] = envelope.key + def _refuse_retype(self, document: DocumentDefinition[Any], written: list[CultCacheEnvelope]) -> list[tuple[str, str]]: + """The one key-identity judge (variants:ruling:cache-refuses-retype): a key holds one record whatever its type, so a write + may not land at a key this cache holds another type's record at, unless the written type owns that record's schema id + (the single-file store's commit guard asks the same question of the file), and may not land at a key a store carries + foreign. Returns the (type, key) records an owning write replaces; refuses, changing nothing, otherwise.""" + replaced: list[tuple[str, str]] = [] + foreign = {record.key: record for record in self._state.foreign} + for envelope in written: + carried = foreign.get(envelope.key) + if carried is not None: + raise self._foreign_refusal(carried) + for other, held_envelopes in self._state.envelopes.items(): + held = held_envelopes.get(envelope.key) + if other == document.type or held is None: + continue + if not written_owns(held, envelope) or self._store_for_type(other) is not self._store_for_type(document.type): + raise retyped(held, envelope) + replaced.append((other, envelope.key)) + return replaced + + @staticmethod + def _foreign_refusal(record: ForeignRecord) -> SchemaConflictError: + return SchemaConflictError( + record.schema_id, + [record.schema_name], + record.key, + message=( + f"Record {record.key!r} is held by the store as a {record.schema_name} (schema id {record.schema_id!r}) that no " + "registered type claims; a write or removal there would replace what this cache cannot read. Register the type " + "that owns the schema id, or remove the record with the runtime that wrote it." + ), + ) + @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 @@ -564,19 +644,27 @@ class CultCache: def _delete(self, document: DocumentDefinition[Any], key: str) -> None: self._assert_registered(document) + if key not in self._state.envelopes.get(document.type, {}): + carried = next((record for record in self._state.foreign if record.key == key), None) + if carried is not None: + raise self._foreign_refusal(carried) 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 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) + self._drop(document.type, key) + + def _drop(self, type: str, key: str) -> None: + """Removes the held record and its lookups from the cache's maps; the store is the caller's.""" + self._remove_lookups(type, key) for by_type in (self._state.values, self._state.envelopes, self._state.lookup_keys): - held = by_type.get(document.type) + held = by_type.get(type) if held is not None: held.pop(key, None) if not held: - by_type.pop(document.type, None) + by_type.pop(type, None) @_mutating def delete_global(self, document: DocumentDefinition[Any]) -> None: diff --git a/packages/cultcache-py/src/cultcache_py/stores.py b/packages/cultcache-py/src/cultcache_py/stores.py index e0cfdd7a..61841956 100644 --- a/packages/cultcache-py/src/cultcache_py/stores.py +++ b/packages/cultcache-py/src/cultcache_py/stores.py @@ -245,7 +245,7 @@ def _derive_catalog(records: list[tuple[CultCacheEnvelope, bool]]) -> list[CultC return [entry for entry, _ in chosen.values()] -def _retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConflictError: +def retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConflictError: """The refusal of a write that would put a second type under a key the store already holds (C# CacheBackingStore.Retyped).""" return SchemaConflictError( _schema_id_for(held), @@ -258,7 +258,7 @@ def _retyped(held: CultCacheEnvelope, written: CultCacheEnvelope) -> SchemaConfl ) -def _written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool: +def written_owns(held: CultCacheEnvelope, written: CultCacheEnvelope) -> bool: """Whether a write of `written` may replace `held` at its key (C# WrittenOwns): the written type's entry declares the held record's schema id (its own id or one it lists compatible), or it is the type of the same schema name. The same-name half stands in for C#'s "reads the published entry whole": this runtime does not compare shapes, so it never refuses a @@ -277,8 +277,8 @@ def _encode_snapshot( landing: dict[str, CultCacheEnvelope] = {} for envelope in supplied: for held in (landing.get(envelope.key), held_by_key.get(envelope.key)): - if held is not None and not _written_owns(held, envelope): - raise _retyped(held, envelope) + if held is not None and not written_owns(held, envelope): + raise retyped(held, envelope) landing[envelope.key] = envelope supplied = list(landing.values()) replaced = set(landing) diff --git a/packages/cultcache-py/tests/test_key_identity_vector.py b/packages/cultcache-py/tests/test_key_identity_vector.py index affd5278..16d6364a 100644 --- a/packages/cultcache-py/tests/test_key_identity_vector.py +++ b/packages/cultcache-py/tests/test_key_identity_vector.py @@ -137,3 +137,86 @@ class KeyIdentityVectorTests(unittest.TestCase): for key in kept: self.assertIsNotNone(stored(vector, key)) self.assertEqual(stored(path, key), stored(vector, key)) + + +def alpha_beta(*names: str, store=None) -> CultCache: + made = CultCache.builder() + for name in names: + made.register_document_type(DOCUMENTS[name]) + if store is not None: + made.add_generic_store(store) + return made.build() + + +class KeyIdentityTests(unittest.TestCase): + def test_every_write_path_refuses_a_second_type_and_never_echoes_the_payload(self) -> None: + alpha, beta = DOCUMENTS["alpha"], DOCUMENTS["beta"] + for name in ("put", "put_envelope", "put_envelopes"): + with self.subTest(path=name): + cache = alpha_beta("alpha", "beta") + cache.put(alpha, "k", {"text": "held"}) + envelope = cache.get_envelope(alpha, "k") + other = type(envelope).create( + key="k", type=beta.type, payload=beta.encode_payload({"text": "CANARY"}), schema_id=IDS["beta"] + ) + writes = { + "put": lambda: cache.put(beta, "k", {"text": "CANARY"}), + "put_envelope": lambda: cache.put_envelope(beta, other), + "put_envelopes": lambda: cache.put_envelopes(beta, [other]), + } + with self.assertRaises(SchemaConflictError) as caught: + writes[name]() + self.assertEqual(caught.exception.record_key, "k") + self.assertNotIn("CANARY", str(caught.exception)) + self.assertEqual(cache.get(alpha, "k"), {"text": "held"}) + self.assertIsNone(cache.get(beta, "k")) + + def test_two_routed_stores_holding_one_key_refuse_the_pull_and_keep_the_earlier_state(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + one, two = Path(tmp) / "one.cc", Path(tmp) / "two.cc" + shutil.copyfile(VECTORS / "catalog-store.cc", one) + shutil.copyfile(VECTORS / "run-store.cc", two) + cache = alpha_beta("alpha", "beta") + cache.add_backing_store(SingleFileMessagePackBackingStore(one), [DOCUMENTS["alpha"].type]) + cache.pull_all_backing_stores() + cache.add_backing_store(SingleFileMessagePackBackingStore(two), [DOCUMENTS["beta"].type]) + with self.assertRaises(Exception) as caught: + cache.pull_all_backing_stores() + self.assertIn("'k'", str(caught.exception)) + self.assertEqual(cache.get(DOCUMENTS["alpha"], "k"), {"text": "a"}) + self.assertIsNone(cache.get(DOCUMENTS["beta"], "k")) + + def test_a_foreign_record_is_listed_and_its_bytes_survive_writes_and_removals_elsewhere(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "store.cc" + shutil.copyfile(VECTORS / "foreign-store.cc", path) + before = stored(path, "k") + cache = alpha_beta("alpha", store=SingleFileMessagePackBackingStore(path)) + cache.pull_all_backing_stores() + [record] = cache.foreign_records() + self.assertEqual((record.key, record.schema_id, record.schema_name), ("k", IDS["gamma"], "tests.ki_gamma")) + self.assertEqual(record.stored_at, before[1]) + cache.put(DOCUMENTS["alpha"], "m", {"text": "m"}) + cache.delete(DOCUMENTS["alpha"], "j") + self.assertEqual(stored(path, "k"), before) + self.assertEqual(file_records(path), [["k", "tests.ki_gamma"], ["m", "tests.ki_alpha"]]) + claimed = alpha_beta("alpha", "gamma", store=SingleFileMessagePackBackingStore(path)) + claimed.pull_all_backing_stores() + self.assertEqual(claimed.foreign_records(), []) + self.assertEqual(claimed.get(DOCUMENTS["gamma"], "k"), {"text": "g"}) + + def test_the_foreign_list_is_sorted_by_utf16_code_unit_and_replaced_by_the_next_pull(self) -> None: + with tempfile.TemporaryDirectory() as tmp: + path = Path(tmp) / "store.cc" + store = SingleFileMessagePackBackingStore(path) + gamma = DOCUMENTS["gamma"] + writer = alpha_beta("gamma", store=store) + for key in ("\U00010400", "Ａ", "b", "B"): + writer.put(gamma, key, {"text": "x"}) + cache = alpha_beta("alpha", store=SingleFileMessagePackBackingStore(path)) + cache.pull_all_backing_stores() + self.assertEqual([record.key for record in cache.foreign_records()], ["B", "b", "\U00010400", "Ａ"]) + for key in ("B", "b", "\U00010400", "Ａ"): + writer.delete(gamma, key) + cache.pull_all_backing_stores() + self.assertEqual(cache.foreign_records(), [])