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

cultcache-py: refuse a legacy duplicate key; refuse a write that would retype a key

Browse source 
The legacy envelope-array decoder refuses a duplicate key as v1 does. The single-file writer
refuses (SchemaConflictError, C# Retyped wording) a put/push_all that would land a second type
under a key the store holds or the batch already names; file unchanged. Replace-by-key stays
for the same type. Retyping is remove-then-add.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Libby 2026-11-11 14:45:00 +00:00
commit 3379191b00
2 changed files with 103 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -245,11 +245,33 @@ def _derive_catalog(records: list[tuple[CultCacheEnvelope, bool]]) -> list[CultC
return [entry for entry, _ in chosen.values()]
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),
[held.type],
held.key,
message=(
f"Record '{held.key}' is a {held.type} (schema id '{_schema_id_for(held)}'); writing a {written.type} there would "
"retype the key. Remove the record first, in the same batch or an earlier commit, then add the other type's."
),
)
def _encode_snapshot(
supplied: list[CultCacheEnvelope], arrived: list[CultCacheEnvelope], format_version: str
) -> list[Any]:
# A key names one record whatever its type, so a supplied record replaces the arrived one by key alone.
replaced = {envelope.key for envelope in supplied}
# A key names one record whatever its type, so a supplied record replaces the arrived one by key alone, and only of the
# same type: retyping a key is a removal and an add, so a record of another type under the key refuses the write.
held_by_key = {envelope.key: envelope for envelope in arrived}
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 held.type != envelope.type:
raise _retyped(held, envelope)
landing[envelope.key] = envelope
supplied = list(landing.values())
replaced = set(landing)
kept = [envelope for envelope in arrived if envelope.key not in replaced]
envelopes = [*kept, *supplied]
catalog = [
@ -442,9 +464,13 @@ 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] = []
seen: set[Any] = set()
for item in decoded:
if not isinstance(item, dict):
raise ValueError("CultCache legacy envelopes must be MessagePack maps")
if item.get("key") in seen:
raise ValueError(f"CultCache record {item['key']!r} appears more than once")
seen.add(item.get("key"))
try:
payload = item["payload"]
envelopes.append(

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

@ -1604,17 +1604,88 @@ class CultCacheTests(unittest.TestCase):
self.assertIsNotNone(refused.exception.__cause__, f"{vector} {operation}")
self.assertEqual(store_path.read_bytes(), content, f"{vector} {operation} rewrote a file it cannot read")
def test_a_supplied_record_replaces_an_arrived_one_by_key_alone(self) -> None:
@staticmethod
def _rec(type: str, key: str, second: int = 0) -> CultCacheEnvelope:
return CultCacheEnvelope(key=key, type=type, payload=b"\x90", stored_at=f"2026-09-30T00:00:0{second}Z")
def test_a_supplied_record_replaces_an_arrived_one_of_the_same_type_by_key(self) -> None:
import msgpack # type: ignore
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
store = SingleFileMessagePackBackingStore(path)
store.push(CultCacheEnvelope(key="k", type="first", payload=b"\x90", stored_at="2026-09-30T00:00:00Z"))
store.push(CultCacheEnvelope(key="k", type="second", payload=b"\x90", stored_at="2026-09-30T00:00:01Z"))
store.push(self._rec("first", "k"))
store.push(self._rec("first", "k", 1))
store.push_all([self._rec("first", "k", 2), self._rec("first", "k", 3)])
records = msgpack.unpackb(path.read_bytes(), raw=False)[2]
self.assertEqual([record[0] for record in records], ["k"])
self.assertEqual([envelope.type for envelope in store.pull_all()], ["second"])
self.assertEqual([(e.type, e.stored_at) for e in store.pull_all()], [("first", "2026-09-30T00:00:03Z")])
def test_a_write_that_would_retype_a_key_is_refused_and_leaves_the_file(self) -> None:
from cultcache_py import SchemaConflictError
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
store = SingleFileMessagePackBackingStore(path)
store.push(self._rec("first", "k"))
content = path.read_bytes()
for run in (
lambda: store.push(self._rec("second", "k", 1)),
lambda: store.push_all([self._rec("other", "j"), self._rec("second", "k", 1)]),
):
with self.assertRaises(SchemaConflictError) as refused:
run()
self.assertEqual(refused.exception.record_key, "k")
self.assertEqual(refused.exception.schema_names, ["first"])
self.assertIn("would retype the key", str(refused.exception))
self.assertEqual(path.read_bytes(), content)
# Retyping is a removal and an add.
store.delete("first", "k")
store.push(self._rec("second", "k", 1))
self.assertEqual([e.type for e in SingleFileMessagePackBackingStore(path).pull_all()], ["second"])
def test_a_batch_naming_one_key_under_two_types_is_refused_whole(self) -> None:
from cultcache_py import SchemaConflictError
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
store = SingleFileMessagePackBackingStore(path)
with self.assertRaises(SchemaConflictError):
store.push_all([self._rec("first", "k"), self._rec("second", "k", 1)])
self.assertFalse(path.exists())
def test_a_cache_put_of_a_second_type_under_a_key_is_refused_and_persists_nothing_new(self) -> None:
from cultcache_py import SchemaConflictError
other = define_document_type("other", encode=lambda item: asdict(item), decode=lambda raw: Item(**raw))
document = item_doc()
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp) / "store.msgpack"
def build() -> CultCache:
cache = (
CultCache.builder()
.register_document_type(document)
.register_document_type(other)
.add_generic_store(SingleFileMessagePackBackingStore(path))
.build()
)
cache.pull_all_backing_stores()
return cache
cache = build()
cache.put(document, "shared", Item(name="A", category="c", value=1))
content = path.read_bytes()
with self.assertRaises(SchemaConflictError):
cache.put(other, "shared", Item(name="B", category="c", value=2))
self.assertEqual(path.read_bytes(), content)
self.assertIsNone(cache.get(other, "shared"))
reloaded = build()
self.assertEqual(reloaded.get_required(document, "shared").value, 1)
self.assertIsNone(reloaded.get(other, "shared"))
cache.delete(document, "shared")
cache.put(other, "shared", Item(name="B", category="c", value=2))
self.assertEqual(build().get_required(other, "shared").value, 2)
def test_a_catalog_entry_with_a_non_string_canonical_schema_is_unreadable(self) -> None:
# Direct probe of the c0 vector; its manifest row is added by the TS cut.