From c1ceb860a0eb8dd511f78ef93d1005330f06d1dd Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 23:09:16 +0100 Subject: [PATCH] Add typed Python snapshot fanout coverage --- packages/cultcache-py/README.md | 11 ++++ .../cultcache-py/src/cultmesh_py/client.py | 17 ++++++ packages/cultcache-py/tests/test_cultcache.py | 61 +++++++++++++++++++ 3 files changed, 89 insertions(+) diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index 45826c4e..b4dcbf98 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -267,6 +267,7 @@ from cultmesh_py import ( CultMeshGameSessionOptions, CultMeshHmacAuthorityLeaseVerifier, CultMeshNodeOptions, + CultMeshSnapshotFanout, peer_exchange_request, ) @@ -388,6 +389,16 @@ replica_notes = client.sync_documents( roles=["read-replica"], documents=[(note_doc, "note:remote")], ) +snapshot_fanout = CultMeshSnapshotFanout( + client, + node.database, + peers, + verse_id="python-interop", + roles=["read-replica"], + documents=[(note_doc, "note:remote")], + on_document=lambda note: print(note), +) +snapshot_fanout.sync_once() 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 c549354e..e4e9ad6d 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/packages/cultcache-py/src/cultmesh_py/client.py @@ -473,17 +473,34 @@ class CultMeshSnapshotFanout: peer_catalog: CultMeshPeerCatalog verse_id: str roles: list[str] | None = None + documents: list[tuple[DocumentDefinition[Any], str]] | None = None schema_ids: list[str] | None = None record_keys: list[str] | None = None shard_id: str | None = None shard_epoch: int | None = None interval_seconds: float = 1.0 on_applied: Callable[[Any], None] | None = None + on_document: Callable[[Any], None] | None = None on_error: Callable[[str, Exception], None] | None = None _stop: threading.Event = field(default_factory=threading.Event, init=False, repr=False) _thread: threading.Thread | None = field(default=None, init=False, repr=False) def sync_once(self) -> list[Any]: + if self.documents is not None: + values = self.client.sync_documents( + self.database, + self.peer_catalog, + verse_id=self.verse_id, + roles=self.roles, + documents=self.documents, + shard_id=self.shard_id, + shard_epoch=self.shard_epoch, + on_error=self.on_error, + ) + for value in values: + if self.on_document is not None: + self.on_document(value) + return values applied = self.client.sync_snapshots( self.database, self.peer_catalog, diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index d163d0c2..1630d05f 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -4515,6 +4515,67 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(seen, ["note:sync-once"]) self.assertEqual(target.database.get_required(document, "note:sync-once")["body"], "from-source") + def test_cultmesh_snapshot_fanout_syncs_typed_documents_and_reports_alias_values(self) -> None: + @dataclass + class CanonicalNote: + body: str + + @dataclass + class UiNote: + body: str + + document = define_database_entry_type( + "mesh.snapshot_loop_alias_note", + [("body", 0)], + cls=CanonicalNote, + schema_id="mesh.snapshot_loop_alias_note.v1", + schema_name="mesh.snapshot_loop_alias_note", + schema_version="mesh.snapshot_loop_alias_note.v1", + ) + alias = define_database_entry_type( + "mesh.snapshot_loop_alias_note.ui", + [("body", 0)], + cls=UiNote, + schema_id="mesh.snapshot_loop_alias_note.v1", + schema_name="mesh.snapshot_loop_alias_note", + schema_version="mesh.snapshot_loop_alias_note.v1", + ) + source = CultMesh.create_node(runtime_id="snapshot-loop-alias-source") + source.database.register_document(document) + source.database.put_raw_message(document, "note:typed-loop", CanonicalNote("from-source")) + target = CultMesh.create_node(runtime_id="snapshot-loop-alias-target") + target.database.register_document(document) + server = CultMesh.serve_node(source) + try: + peers = CultMeshPeerCatalog() + peers.upsert(CultMeshPeerCard( + peer_id="snapshot-loop-alias-source", + verse_id="mesh", + endpoints=(f"cultnet://127.0.0.1:{server.port}",), + roles=("read-replica",), + )) + seen: list[UiNote] = [] + fanout = CultMeshSnapshotFanout( + CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0), + target.database, + peers, + verse_id="mesh", + roles=["read-replica"], + documents=[(alias, "note:typed-loop")], + on_document=seen.append, + ) + + values = fanout.sync_once() + finally: + server.stop() + + self.assertEqual(len(values), 1) + self.assertIsInstance(values[0], UiNote) + self.assertEqual(values[0].body, "from-source") + self.assertEqual(seen, [values[0]]) + self.assertEqual(target.database.get_required(document, "note:typed-loop").body, "from-source") + self.assertIsInstance(target.database.get_required(alias, "note:typed-loop"), UiNote) + def test_cultmesh_snapshot_fanout_runs_background_loop(self) -> None: document = define_database_entry_type( "mesh.snapshot_background_note",