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

Add typed Python snapshot fanout coverage

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

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

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

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

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

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

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