diff --git a/packages/cultcache-py/src/cultcache_py/cache.py b/packages/cultcache-py/src/cultcache_py/cache.py index 78ecfe2e..8f4ac759 100644 --- a/packages/cultcache-py/src/cultcache_py/cache.py +++ b/packages/cultcache-py/src/cultcache_py/cache.py @@ -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 diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 90a99f7d..a33eeb4b 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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")