From 2cdb730b7b93c3a2933ddcac72f0b115b6adff89 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 13:34:50 +0100 Subject: [PATCH] Schedule Python snapshot fanout --- packages/cultcache-py/PARITY.md | 8 +- .../cultcache-py/src/cultmesh_py/__init__.py | 2 + .../cultcache-py/src/cultmesh_py/client.py | 59 +++++++++++++ packages/cultcache-py/tests/test_cultcache.py | 82 +++++++++++++++++++ .../Contracts/cultnet-distributed-database.md | 7 +- 5 files changed, 151 insertions(+), 7 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index a1bd9a12..f861bea3 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -80,10 +80,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. from endpoint lists, de-duplicate discovery targets, fan out one-shot discovery from endpoints already advertised in local Verse and peer catalogs, fan out raw snapshot fetches and simulation observations to peer-card - endpoints, apply peer snapshots through a caller-owned database, run a - lightweight scheduled simulation-observation fanout loop, and upsert results - into local catalogs with the same count-returning ergonomics as the C# - discovery clients. + endpoints, apply peer snapshots through a caller-owned database, run + lightweight scheduled snapshot and simulation-observation fanout loops, and + upsert results into local catalogs with the same count-returning ergonomics + as the C# discovery clients. - Python CultMesh facade: Exposes C#-matching client entrypoints for Verse discovery, peer exchange, raw CultNet client construction, local nodes, catalogs, sessions, streams, diff --git a/packages/cultcache-py/src/cultmesh_py/__init__.py b/packages/cultcache-py/src/cultmesh_py/__init__.py index 53c38c9e..12b776c3 100644 --- a/packages/cultcache-py/src/cultmesh_py/__init__.py +++ b/packages/cultcache-py/src/cultmesh_py/__init__.py @@ -2,6 +2,7 @@ from .client import ( CultMeshDiscoveryClient, CultMeshPeerExchangeClient, CultMeshSimulationObservationFanout, + CultMeshSnapshotFanout, CultMeshVerseDiscoveryClient, ) from .facade import CultMesh @@ -69,6 +70,7 @@ __all__ = [ "CultMeshDiscoveryClient", "CultMeshPeerExchangeClient", "CultMeshSimulationObservationFanout", + "CultMeshSnapshotFanout", "CultMeshVerseDiscoveryClient", "CultMeshStreamCatalog", "CultMeshStreamConsumerProfile", diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index fe79ddb4..8aa9cef9 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/packages/cultcache-py/src/cultmesh_py/client.py @@ -425,6 +425,65 @@ class CultMeshSimulationObservationFanout: self.flush() +@dataclass +class CultMeshSnapshotFanout: + client: CultMeshDiscoveryClient + database: Any + peer_catalog: CultMeshPeerCatalog + verse_id: str + roles: list[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_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]: + applied = self.client.sync_snapshots( + self.database, + self.peer_catalog, + verse_id=self.verse_id, + roles=self.roles, + schema_ids=self.schema_ids, + record_keys=self.record_keys, + shard_id=self.shard_id, + shard_epoch=self.shard_epoch, + on_error=self.on_error, + ) + for record in applied: + if self.on_applied is not None: + self.on_applied(record) + return applied + + def start(self) -> "CultMeshSnapshotFanout": + if self._thread is not None: + return self + self._stop.clear() + self._thread = threading.Thread(target=self._run, daemon=True) + self._thread.start() + return self + + def stop(self) -> None: + self._stop.set() + if self._thread is not None: + self._thread.join(timeout=max(2.0, self.interval_seconds * 2.0)) + self._thread = None + + def __enter__(self) -> "CultMeshSnapshotFanout": + return self.start() + + def __exit__(self, exc_type: Any, exc: Any, traceback: Any) -> None: + self.stop() + + def _run(self) -> None: + while not self._stop.wait(self.interval_seconds): + self.sync_once() + + def _parse_endpoint(endpoint: str) -> tuple[str, int]: if not endpoint or not endpoint.strip(): raise ValueError("endpoint must be non-empty") diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index ac3a3cf8..06b55496 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -91,6 +91,7 @@ from cultmesh_py import ( CultMeshDiscoveryClient, CultMeshPeerExchangeClient, CultMeshSimulationObservationFanout, + CultMeshSnapshotFanout, CultMeshStreamCatalog, CultMeshStreamConsumerProfile, CultMeshStreamDescriptor, @@ -2703,6 +2704,87 @@ 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_snapshot_fanout_syncs_once_and_reports_applied_records(self) -> None: + document = define_database_entry_type( + "mesh.snapshot_loop_note", + [("body", 0)], + schema_id="mesh.snapshot_loop_note.v1", + ) + source = CultMesh.create_node(runtime_id="snapshot-loop-source") + source.database.register_document(document) + source.database.put_raw_message(document, "note:sync-once", {"body": "from-source"}) + target = CultMesh.create_node(runtime_id="snapshot-loop-target") + target.database.register_document(document) + server = CultMesh.serve_node(source) + try: + peers = CultMeshPeerCatalog() + peers.upsert(CultMeshPeerCard( + peer_id="snapshot-loop-source", + verse_id="mesh", + endpoints=(f"cultnet://127.0.0.1:{server.port}",), + roles=("read-replica",), + )) + seen: list[str] = [] + fanout = CultMeshSnapshotFanout( + CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0), + target.database, + peers, + verse_id="mesh", + roles=["read-replica"], + schema_ids=["mesh.snapshot_loop_note.v1"], + on_applied=lambda record: seen.append(record.record_key), + ) + + applied = fanout.sync_once() + finally: + server.stop() + + self.assertEqual([record.record_key for record in applied], ["note:sync-once"]) + self.assertEqual(seen, ["note:sync-once"]) + self.assertEqual(target.database.get_required(document, "note:sync-once")["body"], "from-source") + + def test_cultmesh_snapshot_fanout_runs_background_loop(self) -> None: + document = define_database_entry_type( + "mesh.snapshot_background_note", + [("body", 0)], + schema_id="mesh.snapshot_background_note.v1", + ) + source = CultMesh.create_node(runtime_id="snapshot-background-source") + source.database.register_document(document) + source.database.put_raw_message(document, "note:background", {"body": "from-background"}) + target = CultMesh.create_node(runtime_id="snapshot-background-target") + target.database.register_document(document) + server = CultMesh.serve_node(source) + try: + peers = CultMeshPeerCatalog() + peers.upsert(CultMeshPeerCard( + peer_id="snapshot-background-source", + verse_id="mesh", + endpoints=(f"cultnet://127.0.0.1:{server.port}",), + roles=("read-replica",), + )) + seen: list[str] = [] + fanout = CultMeshSnapshotFanout( + CultMeshDiscoveryClient("127.0.0.1", server.port, timeout_seconds=2.0), + target.database, + peers, + verse_id="mesh", + roles=["read-replica"], + schema_ids=["mesh.snapshot_background_note.v1"], + interval_seconds=0.05, + on_applied=lambda record: seen.append(record.record_key), + ) + fanout.start() + deadline = time.monotonic() + 2.0 + while not seen and time.monotonic() < deadline: + time.sleep(0.01) + fanout.stop() + finally: + server.stop() + + self.assertTrue(seen) + self.assertEqual(target.database.get_required(document, "note:background")["body"], "from-background") + def test_cultmesh_local_server_serves_node_and_catalogs_over_clients(self) -> None: document = define_database_entry_type( "mesh.server_note", diff --git a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md index 746f1fbb..62f0e76a 100644 --- a/src/GameCult.Networking/Contracts/cultnet-distributed-database.md +++ b/src/GameCult.Networking/Contracts/cultnet-distributed-database.md @@ -94,7 +94,8 @@ Not implemented yet: - membership/failure detection - leader election or automatic shard failover -- scheduled peer-to-peer snapshot fanout beyond Python's one-shot client helper +- cross-runtime scheduled peer-to-peer snapshot fanout beyond Python's local + fanout loop - cross-runtime scheduled peer-to-peer fanout for observation gossip and candidate propagation beyond Python's local fanout loop - cross-runtime rollback/resimulation helpers for simulation frames @@ -149,8 +150,8 @@ Add discovery fanout and operational polish: package-local Python HMAC verifier - standardize cross-runtime simulation-frame rollback/resimulation helpers beyond Python's local pending-prediction rollback surface -- add scheduled peer-to-peer snapshot fanout beyond Python's one-shot client - helper +- standardize cross-runtime scheduled peer-to-peer snapshot fanout beyond + Python's local fanout loop - standardize cross-runtime scheduled peer-to-peer fanout for observation gossip and candidate propagation beyond Python's local fanout loop