diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index b23cbb2e..45826c4e 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -381,6 +381,13 @@ verses = client.fetch_verses(transport_version="cultmesh.v0") mesh_peers = client.fetch_peers(verse_id="python-interop", roles=["read-replica"]) client.sync_peer_catalog(peers, verse_id="python-interop", roles=["read-replica"]) client.fanout_peer_catalog(peers, verse_id="python-interop", roles=["read-replica"]) +replica_notes = client.sync_documents( + node.database, + peers, + verse_id="python-interop", + roles=["read-replica"], + documents=[(note_doc, "note:remote")], +) raw_client = CultNetRawClient("127.0.0.1", 3075) synced_note = node.database.sync_document(raw_client, note_doc, "note:remote") diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index 89a9b442..c549354e 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/packages/cultcache-py/src/cultmesh_py/client.py @@ -6,6 +6,7 @@ from datetime import UTC, datetime from typing import Any, Callable from urllib.parse import urlparse +from cultcache_py.documents import DocumentDefinition from cultnet_py import CultNetRawClient, CultNetRawSnapshotResponse, hello from .wire import ( @@ -279,6 +280,41 @@ class CultMeshDiscoveryClient: applied.extend(database.apply_snapshot_response(response)) return applied + def sync_documents( + self, + database: Any, + catalog: CultMeshPeerCatalog, + *, + verse_id: str, + documents: list[tuple[DocumentDefinition[Any], str]], + roles: list[str] | None = None, + shard_id: str | None = None, + shard_epoch: int | None = None, + on_error: Callable[[str, Exception], None] | None = None, + ) -> list[Any]: + if not documents: + return [] + schema_ids = _distinct_non_empty([ + document.catalog_entry().schema_id + for document, _key in documents + ]) + record_keys = _distinct_non_empty([key for _document, key in documents]) + self.sync_snapshots( + database, + catalog, + verse_id=verse_id, + roles=roles, + schema_ids=schema_ids, + record_keys=record_keys, + shard_id=shard_id, + shard_epoch=shard_epoch, + on_error=on_error, + ) + return [ + database.get_required(document, key) + for document, key in documents + ] + def _fanout_endpoints( self, catalog: CultMeshPeerCatalog, diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index d1c9c568..d163d0c2 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -4420,6 +4420,62 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(target.database.get_required(document, "note:a")["body"], "from-a") self.assertEqual(target.database.get_required(document, "note:b")["body"], "from-b") + def test_cultmesh_discovery_client_sync_documents_returns_requested_aliases(self) -> None: + @dataclass + class CanonicalNote: + body: str + + @dataclass + class UiNote: + body: str + + document = define_database_entry_type( + "mesh.snapshot_fanout_alias_note", + [("body", 0)], + cls=CanonicalNote, + schema_id="mesh.snapshot_fanout_alias_note.v1", + schema_name="mesh.snapshot_fanout_alias_note", + schema_version="mesh.snapshot_fanout_alias_note.v1", + ) + alias = define_database_entry_type( + "mesh.snapshot_fanout_alias_note.ui", + [("body", 0)], + cls=UiNote, + schema_id="mesh.snapshot_fanout_alias_note.v1", + schema_name="mesh.snapshot_fanout_alias_note", + schema_version="mesh.snapshot_fanout_alias_note.v1", + ) + source = CultMesh.create_node(runtime_id="snapshot-alias-source") + source.database.register_document(document) + source.database.put_raw_message(document, "note:alias", CanonicalNote("from-source")) + target = CultMesh.create_node(runtime_id="snapshot-alias-target") + target.database.register_document(document) + server = CultMesh.serve_node(source) + try: + peers = CultMeshPeerCatalog() + peers.upsert(CultMeshPeerCard( + peer_id="snapshot-alias-source", + verse_id="mesh", + endpoints=(f"cultnet://127.0.0.1:{server.port}",), + roles=("read-replica",), + )) + + values = CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0).sync_documents( + target.database, + peers, + verse_id="mesh", + roles=["read-replica"], + documents=[(alias, "note:alias")], + ) + finally: + server.stop() + + self.assertEqual(len(values), 1) + self.assertIsInstance(values[0], UiNote) + self.assertEqual(values[0].body, "from-source") + self.assertEqual(target.database.get_required(document, "note:alias").body, "from-source") + self.assertIsInstance(target.database.get_required(alias, "note:alias"), UiNote) + def test_cultmesh_snapshot_fanout_syncs_once_and_reports_applied_records(self) -> None: document = define_database_entry_type( "mesh.snapshot_loop_note",