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

Resolve Python raw deletes by schema alias

Browse source
This commit is contained in:
Meta Vens 2026-06-27 22:13:28 +01:00
commit d9ffdd77a4
4 changed files with 94 additions and 6 deletions
Download patch file Download diff file Expand all files Collapse all files

15
packages/cultcache-py/src/cultmesh_py/node.py
View file

@ -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,
*,

11
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -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

4
packages/cultcache-py/src/cultnet_py/replication.py
View file

@ -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

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

@ -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",