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

Python: a document listing an id registers beside the one owning it; registration refuses typed

Browse source 
Owner beats lister, as in C#. A document owns its schema id; others may list it
as compatible beside its owner, in either registration order (batch 9 refused
any overlap, which forbade keeping v1 beside a v2 that declares v1's id; Soul
F8). A record resolves to the document owning its id, else to the one document
listing it. Several listers and no owner name no single document, and the load
is refused with SchemaConflictError naming the id, the record key and the
listers: refusing at registration would depend on whether the owner had
registered yet.

Every registration refusal is now SchemaConflictError (Soul F10): a second owner
of one schema id, a second document with one schema name, and a second
definition of one document type. Each names the new document's id and both
schema names and document types. Before, the last two raised CultCacheError.

Tests: an owner and a lister register in both orders and each reads its own
records, and a lister alone reads the owner's id; two listers register in both
orders, a record under their id is refused typed without its owner and is the
owner's with it, registered before or after them; two owners, one schema name
and one document type twice are each refused as a schema conflict.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-30 15:31:25 +02:00
commit 84246b685a
2 changed files with 90 additions and 32 deletions
Download patch file Download diff file Expand all files Collapse all files

58
packages/cultcache-py/src/cultcache_py/cache.py
View file

@ -139,25 +139,28 @@ class CultCache:
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}")
# Every registration refusal is a SchemaConflictError naming the new document's schema id and both schema names.
entry = document.catalog_entry()
schema_name = entry.schema_name
def refuse(registered: DocumentDefinition[Any], what: str) -> SchemaConflictError:
return SchemaConflictError(
entry.schema_id,
[registered.catalog_entry().schema_name, schema_name],
"",
message=f"{what} is already registered to document type {registered.type!r} and cannot also be claimed by "
f"{document.type!r}.",
)
if document.type in self._state.documents:
raise refuse(self._state.documents[document.type], f"Document type {document.type!r}")
if schema_name in self._state.documents_by_schema_name:
raise CultCacheError(f"Document schema name already registered: {schema_name}")
# One document carries each schema id, its own or declared compatible, so which document a record resolves to never
# depends on the order documents were registered in.
raise refuse(self._state.documents_by_schema_name[schema_name], f"Schema name {schema_name!r}")
# One document owns each schema id. Others may list it as compatible beside its owner: a record under the id resolves to
# its owner, and to a lister only when nothing owns it. Neither depends on the order documents were registered in.
for registered in self._state.documents.values():
other = registered.catalog_entry()
shared = sorted(({entry.schema_id, *entry.compatible_schema_ids}) & ({other.schema_id, *other.compatible_schema_ids}))
if shared:
raise SchemaConflictError(
shared[0],
[other.schema_name, schema_name],
"",
message=f"Schema id {shared[0]!r} is already carried by document type {registered.type!r} and cannot also be "
f"claimed by {document.type!r}. A schema id, owned or declared compatible, belongs to one registered document.",
)
if registered.catalog_entry().schema_id == entry.schema_id:
raise refuse(registered, f"Schema id {entry.schema_id!r}")
name_extractors = dict(self._state.name_extractors)
index_extractors = {type: dict(indexes) for type, indexes in self._state.index_extractors.items()}
if document.name is not None:
@ -607,14 +610,27 @@ class CultCache:
raise CultCacheError(f"Document type is not global: {document.type}")
def _resolve_document_for_envelope(self, envelope: CultCacheEnvelope) -> DocumentDefinition[Any] | None:
"""A record resolves by its schema id: the one document that owns the id or lists it as compatible. The name its
catalog carries is metadata; it names a document only when the id names none (a store from a runtime whose schema
ids this one cannot know)."""
"""A record resolves by its schema id: the document that owns the id, else the one document that lists it as compatible.
Several listers and no owner name no single document, and the record is refused. The name its catalog carries is
metadata; it names a document only when the id names none (a store from a runtime whose schema ids this one cannot
know)."""
if envelope.schema_id:
for document in self._state.documents.values():
entry = document.catalog_entry()
if entry.schema_id == envelope.schema_id or envelope.schema_id in entry.compatible_schema_ids:
entries = [(document, document.catalog_entry()) for document in self._state.documents.values()]
for document, entry in entries:
if entry.schema_id == envelope.schema_id:
return document
listers = [document for document, entry in entries if envelope.schema_id in entry.compatible_schema_ids]
if len(listers) > 1:
raise SchemaConflictError(
envelope.schema_id,
sorted(document.catalog_entry().schema_name for document in listers),
envelope.key,
message=f"Schema id {envelope.schema_id!r} of record {envelope.key!r} is owned by no registered document and "
f"declared compatible by several ({', '.join(sorted(document.type for document in listers))}), so it resolves "
"to none. Register the document that owns the id, or declare it on one document.",
)
if listers:
return listers[0]
document = self._state.documents.get(envelope.type)
if document is not None:
return document

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

@ -629,8 +629,8 @@ class CultCacheTests(unittest.TestCase):
self.assertIsNone(cache.get(first, "k"), schema_id)
def _assert_second_claim_refused(self, first, second, schema_id: str) -> None:
"""One document carries each schema id, owned or declared compatible, whichever order the documents register in: a
second claimant is refused, typed, naming both, and the cache keeps what it had."""
"""A registration refusal is typed, names the id and both schema names and document types, and the cache keeps what it
had."""
from cultcache_py import SchemaConflictError
cache = CultCache()
@ -641,27 +641,63 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(refused.exception.schema_names, [first.catalog_entry().schema_name, second.catalog_entry().schema_name])
self.assertIn(first.type, str(refused.exception))
self.assertIn(second.type, str(refused.exception))
with self.assertRaisesRegex(CultCacheError, "not registered"):
cache.put(second, "k", {"name": "k"})
if second is not first:
with self.assertRaisesRegex(CultCacheError, "not registered"):
cache.put(second, "k", {"name": "k"})
def test_a_document_listing_an_id_another_document_owns_is_refused_in_either_order(self) -> None:
def _pull(self, documents: list, catalog: list, records: list) -> CultCache:
tmp = tempfile.mkdtemp()
path = Path(tmp) / "store.msgpack"
self._store_of(path, catalog, records)
builder = CultCache.builder()
for document in documents:
builder = builder.register_document_type(document)
cache = builder.add_generic_store(SingleFileMessagePackBackingStore(path)).build()
cache.pull_all_backing_stores()
return cache
def test_a_document_listing_an_id_another_owns_registers_beside_it_in_either_order_and_the_owner_wins(self) -> None:
owner = define_database_entry_type("tests.owner", [("name", 0)], schema_id="id.owned", schema_name="tests.owner", schema_version="tests.owner.v1")
lister = define_database_entry_type(
"tests.lister", [("name", 0)], schema_id="id.lister", schema_name="tests.lister", schema_version="tests.lister.v1",
compatible_schema_ids=["id.lister", "id.owned"],
)
owner = define_database_entry_type("tests.owner", [("name", 0)], schema_id="id.owned", schema_name="tests.owner", schema_version="tests.owner.v1")
self._assert_second_claim_refused(lister, owner, "id.owned")
self._assert_second_claim_refused(owner, lister, "id.owned")
catalog = [
["id.owned", "tests.owner", "tests.owner.v1", "h1", "", ["id.owned"], []],
["id.lister", "tests.lister", "tests.lister.v1", "h2", "", ["id.lister", "id.owned"], []],
]
records = [
["o", "id.owned", "2026-09-30T00:00:00Z", owner.encode_payload({"name": "o"})],
["l", "id.lister", "2026-09-30T00:00:00Z", lister.encode_payload({"name": "l"})],
]
for order in ([owner, lister], [lister, owner]):
cache = self._pull(order, catalog, records)
self.assertEqual(cache.get_required(owner, "o")["name"], "o")
self.assertIsNone(cache.get(lister, "o"))
self.assertEqual(cache.get_required(lister, "l")["name"], "l")
self.assertEqual(self._pull([lister], catalog, records[:1]).get_required(lister, "o")["name"], "o")
def test_two_documents_listing_one_id_register_and_a_record_under_it_needs_an_owner(self) -> None:
from cultcache_py import SchemaConflictError
def test_two_documents_listing_one_compatible_id_are_refused_in_either_order(self) -> None:
a = define_database_entry_type(
"tests.list-a", [("name", 0)], schema_id="id.a", schema_name="tests.list-a", compatible_schema_ids=["id.a", "id.shared"],
)
b = define_database_entry_type(
"tests.list-b", [("name", 0)], schema_id="id.b", schema_name="tests.list-b", compatible_schema_ids=["id.b", "id.shared"],
)
self._assert_second_claim_refused(a, b, "id.shared")
self._assert_second_claim_refused(b, a, "id.shared")
owner = define_database_entry_type("tests.shared", [("name", 0)], schema_id="id.shared", schema_name="tests.shared")
catalog = [["id.shared", "tests.shared", "tests.shared.v1", "h", "", ["id.shared"], []]]
records = [["k", "id.shared", "2026-09-30T00:00:00Z", owner.encode_payload({"name": "k"})]]
for listers in ([a, b], [b, a]):
with self.assertRaises(SchemaConflictError) as refused:
self._pull(listers, catalog, records)
self.assertEqual(refused.exception.schema_id, "id.shared")
self.assertEqual(refused.exception.record_key, "k")
self.assertEqual(refused.exception.schema_names, ["tests.list-a", "tests.list-b"])
for documents in ([*listers, owner], [owner, *listers]):
cache = self._pull(documents, catalog, records)
self.assertEqual(cache.get_required(owner, "k")["name"], "k")
def test_two_documents_owning_one_schema_id_are_refused_in_either_order(self) -> None:
# Each declares an older id and not its own, so the id they share is only ever their own.
@ -670,6 +706,12 @@ class CultCacheTests(unittest.TestCase):
self._assert_second_claim_refused(a, b, "id.same")
self._assert_second_claim_refused(b, a, "id.same")
def test_every_registration_refusal_is_a_schema_conflict(self) -> None:
first = define_database_entry_type("tests.named", [("name", 0)], schema_id="id.first", schema_name="tests.named")
same_name = define_database_entry_type("tests.renamed", [("name", 0)], schema_id="id.second", schema_name="tests.named")
self._assert_second_claim_refused(first, same_name, "id.second")
self._assert_second_claim_refused(first, first, "id.first")
def test_interop_cli_helpers_round_trip_v1_store(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
store_path = str(Path(tmp) / "cache.msgpack")