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

Add Python fanout document sync helper

Browse source
This commit is contained in:
Meta Vens 2026-06-27 23:03:30 +01:00
commit bd98bfe979
3 changed files with 99 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

7
packages/cultcache-py/README.md
View file

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

36
packages/cultcache-py/src/cultmesh_py/client.py
View file

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

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

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