From d9ffdd77a40236f7366e80029b0b54dfb15dec5e Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 22:13:28 +0100 Subject: [PATCH] Resolve Python raw deletes by schema alias --- packages/cultcache-py/src/cultmesh_py/node.py | 15 +++- .../cultcache-py/src/cultmesh_py/server.py | 11 ++- .../src/cultnet_py/replication.py | 4 +- packages/cultcache-py/tests/test_cultcache.py | 70 +++++++++++++++++++ 4 files changed, 94 insertions(+), 6 deletions(-) diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index e9150d2d..81a50cf1 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -848,8 +848,7 @@ class CultMeshDatabase: def _document_for_schema(self, schema_id: str) -> DocumentDefinition[Any] | None: for document in self.documents: - entry = document.catalog_entry() - if schema_id == entry.schema_id or schema_id in entry.compatible_schema_ids: + if _document_matches_schema_id(document, schema_id): return document return None @@ -931,7 +930,8 @@ class CultMeshDatabase: if document is None: continue record_key = str(delete.get("recordKey")) - previous[(schema_id, record_key)] = self.cache.get(document, record_key) + resolved_schema_id = document.catalog_entry().schema_id + previous[(resolved_schema_id, record_key)] = self.cache.get(document, record_key) return previous def _previous_values_for_raw_records(self, records: Any) -> dict[tuple[str, str], Any]: @@ -979,6 +979,15 @@ def _infer_schema_name(schema_id: str) -> str | None: return schema_id[:marker] if version.isdigit() else None +def _document_matches_schema_id(document: DocumentDefinition[Any], schema_id: str) -> bool: + entry = document.catalog_entry() + if schema_id in (entry.schema_id, entry.schema_name, entry.schema_version): + return True + if schema_id in entry.compatible_schema_ids: + return True + return _infer_schema_name(schema_id) == entry.schema_name + + def create_node( cache_path: str | Path | None = None, *, diff --git a/packages/cultcache-py/src/cultmesh_py/server.py b/packages/cultcache-py/src/cultmesh_py/server.py index ca134c8d..d560cc83 100644 --- a/packages/cultcache-py/src/cultmesh_py/server.py +++ b/packages/cultcache-py/src/cultmesh_py/server.py @@ -382,7 +382,7 @@ class CultMeshLocalServer: message_id=str(message.get("messageId") or ""), code="unregistered_document_delete", )] - return self._database_delete_notifications(message, subscriptions) + return self._database_delete_notifications(message, subscriptions, resolved_schema_id=change.schema_id) def _database_change_notifications( self, @@ -415,13 +415,19 @@ class CultMeshLocalServer: self, message: dict[str, Any], subscriptions: dict[str, _DatabaseSubscription], + *, + resolved_schema_id: str | None = None, ) -> list[dict[str, Any]]: schema_id = str(message.get("schemaId") or "") record_key = str(message.get("recordKey") or "") notifications = [] document = {"schemaId": schema_id, "recordKey": record_key} for subscription in subscriptions.values(): - if not self._subscription_matches_document(subscription, document): + if not self._subscription_matches_document( + subscription, + document, + resolved_schema_id=resolved_schema_id, + ): continue message_id = message.get("messageId") or record_key or "change" notifications.append({ @@ -678,6 +684,7 @@ def _schema_document_map(documents: list[Any]) -> dict[str, Any]: entry = document.catalog_entry() mapped[entry.schema_id] = document mapped[entry.schema_name] = document + mapped[entry.schema_version] = document for compatible_schema_id in entry.compatible_schema_ids: mapped[compatible_schema_id] = document return mapped diff --git a/packages/cultcache-py/src/cultnet_py/replication.py b/packages/cultcache-py/src/cultnet_py/replication.py index ca3f2752..6450d4f9 100644 --- a/packages/cultcache-py/src/cultnet_py/replication.py +++ b/packages/cultcache-py/src/cultnet_py/replication.py @@ -24,6 +24,7 @@ def schema_document_map(documents: Iterable[DocumentDefinition[Any]]) -> dict[st catalog_entry = document.catalog_entry() mapped[catalog_entry.schema_id] = document mapped[catalog_entry.schema_name] = document + mapped[catalog_entry.schema_version] = document for compatible_schema_id in catalog_entry.compatible_schema_ids: mapped[compatible_schema_id] = document return mapped @@ -140,8 +141,9 @@ def apply_shard_log_response( if schema_id not in documents_by_schema_id: continue document = documents_by_schema_id[schema_id] + resolved_schema_id = document.catalog_entry().schema_id cache.delete(document, record_key) - applied.append(CultNetAppliedRecord(schema_id, record_key, "removed", None)) + applied.append(CultNetAppliedRecord(resolved_schema_id, record_key, "removed", None)) return applied diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 9138268a..8442dfbb 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -1881,6 +1881,32 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(cache.get_required(document, "policy:1")["value"], "updated through log") self.assertEqual(cache.get_required_envelope(document, "policy:1").schema_id, local_schema_id) + delete_log = { + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": "log-delete-alias", + "shardId": "interop", + "shardEpoch": 1, + "resyncRequired": False, + "entries": [ + { + "sequence": 1, + "changeKind": "removed", + "delete": { + "schemaVersion": "cultnet.document_delete.v0", + "messageId": "delete-version-alias", + "schemaId": "replica.runtime_policy.v1", + "recordKey": "policy:1", + "shardId": "interop", + "shardEpoch": 1, + }, + } + ], + } + applied_delete = apply_shard_log_response(cache, [document], delete_log) + self.assertEqual(applied_delete[0].schema_id, local_schema_id) + self.assertEqual(applied_delete[0].change_kind, "removed") + self.assertIsNone(cache.get(document, "policy:1")) + def test_cultmesh_node_applies_foreign_schema_snapshot_as_local_update(self) -> None: document = define_database_entry_type( "mesh.runtime-policy", @@ -5314,6 +5340,50 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(node.database.get_required(document, "note:alias-sub")["body"], "notified") self.assertEqual(node.cache.get_required_envelope(document, "note:alias-sub").schema_id, local_schema_id) + def test_cultmesh_local_server_notifies_schema_alias_subscription_for_delete(self) -> None: + document = define_database_entry_type( + "mesh.alias_delete_note", + [("body", 0)], + schema_id="sha256:mesh-alias-delete-note", + schema_name="mesh.alias_delete_note", + schema_version="mesh.alias_delete_note.v1", + ) + node = CultMesh.create_node(runtime_id="mesh-alias-delete-server") + node.database.register_document(document) + node.database.put_raw_message( + document, + "note:delete-alias", + {"body": "remove me"}, + shard_id="notes", + shard_epoch=1, + ) + + server = CultMesh.serve_node(node) + try: + raw_client = CultMesh.create_client("127.0.0.1", server.port, timeout_seconds=2.0) + with raw_client.subscribe_database( + subscription_id="sub-delete-alias", + schema_ids=[document.catalog_entry().schema_id], + include_snapshot=False, + ) as subscription: + subscription.send(document_delete( + message_id="delete-foreign-alias", + schema_id="mesh.alias_delete_note.v1", + record_key="note:delete-alias", + shard_id="notes", + shard_epoch=1, + )) + change = subscription.read_next_change() + finally: + server.stop() + + self.assertEqual(change.subscription_id, "sub-delete-alias") + self.assertEqual(change.change_kind, "removed") + self.assertEqual(change.schema_id, "mesh.alias_delete_note.v1") + self.assertEqual(change.record_key, "note:delete-alias") + self.assertIsNone(change.raw_document) + self.assertIsNone(node.database.get(document, "note:delete-alias")) + def test_cultmesh_local_server_rejects_oversized_snapshot_responses(self) -> None: document = define_database_entry_type( "mesh.snapshot_limit_note",