diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index f0264c47..90c2fe96 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -77,9 +77,10 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti. - Python CultMesh discovery client: Fetches Verse catalogs and peer exchange responses from live endpoints over CultNet frames. It can also parse `cultnet://host:port` endpoints, discover - from endpoint lists, de-duplicate discovery targets, and upsert results into - local Verse and peer catalogs with the same count-returning ergonomics as the - C# discovery clients. + from endpoint lists, de-duplicate discovery targets, fan out one-shot + discovery from endpoints already advertised in local Verse and peer catalogs, + 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/README.md b/packages/cultcache-py/README.md index f1ec04ed..7a0e3d09 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -281,6 +281,7 @@ client = CultMesh.create_verse_discovery_client("127.0.0.1", 3075) 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"]) raw_client = CultNetRawClient("127.0.0.1", 3075) node.database.sync_snapshot(raw_client, schema_ids=[note_doc.catalog_entry().schema_id]) diff --git a/packages/cultcache-py/src/cultmesh_py/client.py b/packages/cultcache-py/src/cultmesh_py/client.py index 914b7ce4..884d724c 100644 --- a/packages/cultcache-py/src/cultmesh_py/client.py +++ b/packages/cultcache-py/src/cultmesh_py/client.py @@ -114,6 +114,23 @@ class CultMeshDiscoveryClient: count += client.discover_verse_catalog(catalog, transport_version=transport_version) return count + def fanout_verse_catalog( + self, + catalog: CultMeshVerseCatalog, + *, + transport_version: str | None = None, + ) -> int: + endpoints = [ + endpoint + for verse in catalog.verses + for endpoint in verse.discovery_endpoints + ] + return self.discover_verse_catalog( + catalog, + endpoints=endpoints, + transport_version=transport_version, + ) + def sync_peer_catalog( self, catalog: CultMeshPeerCatalog, @@ -155,6 +172,27 @@ class CultMeshDiscoveryClient: count += client.discover_peer_catalog(catalog, verse_id=verse_id, roles=roles, limit=limit) return count + def fanout_peer_catalog( + self, + catalog: CultMeshPeerCatalog, + *, + verse_id: str, + roles: list[str] | None = None, + limit: int | None = None, + ) -> int: + endpoints = [ + endpoint + for peer in catalog.find(verse_id) + for endpoint in peer.endpoints + ] + return self.discover_peer_catalog( + catalog, + verse_id=verse_id, + endpoints=endpoints, + roles=roles, + limit=limit, + ) + @classmethod def from_endpoint(cls, endpoint: str, *, timeout_seconds: float = 4.0) -> "CultMeshDiscoveryClient": host, port = _parse_endpoint(endpoint) @@ -175,6 +213,14 @@ class CultMeshVerseDiscoveryClient(CultMeshDiscoveryClient): transport_version=transport_version, ) + def fanout( + self, + catalog: CultMeshVerseCatalog, + *, + transport_version: str | None = None, + ) -> int: + return self.fanout_verse_catalog(catalog, transport_version=transport_version) + class CultMeshPeerExchangeClient(CultMeshDiscoveryClient): def discover( @@ -194,6 +240,16 @@ class CultMeshPeerExchangeClient(CultMeshDiscoveryClient): limit=limit, ) + def fanout( + self, + catalog: CultMeshPeerCatalog, + *, + verse_id: str, + roles: list[str] | None = None, + limit: int | None = None, + ) -> int: + return self.fanout_peer_catalog(catalog, verse_id=verse_id, roles=roles, limit=limit) + def _parse_endpoint(endpoint: str) -> tuple[str, int]: if not endpoint or not endpoint.strip(): diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 3de41730..c316fdda 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -2299,9 +2299,9 @@ class CultCacheTests(unittest.TestCase): server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind(("127.0.0.1", 0)) port_holder.append(server.getsockname()[1]) - server.listen(6) + server.listen(8) ready.set() - for _ in range(6): + for _ in range(8): connection, _ = server.accept() with connection: stream = connection.makefile("rwb") @@ -2347,6 +2347,31 @@ class CultCacheTests(unittest.TestCase): endpoints=[endpoint, endpoint, ""], roles=["read-replica"], ) + fanout_verses = CultMeshVerseCatalog() + fanout_verses.upsert( + CultMeshVerseDescriptor( + verse_id="aetheria-main", + display_name="Known Aetheria", + authority_model="federated", + compatibility=CultMeshVerseCompatibility("cultmesh.v0", "rules-1"), + discovery_endpoints=(endpoint, endpoint, ""), + ) + ) + fanout_peers = CultMeshPeerCatalog() + fanout_peers.upsert( + CultMeshPeerCard( + peer_id="known-peer", + verse_id="aetheria-main", + endpoints=(endpoint, endpoint, ""), + roles=("read-replica",), + ) + ) + fanout_verse_count = verse_discovery.fanout(fanout_verses, transport_version="cultmesh.v0") + fanout_peer_count = peer_discovery.fanout( + fanout_peers, + verse_id="aetheria-main", + roles=["read-replica"], + ) thread.join(2.0) self.assertFalse(server_error) @@ -2361,6 +2386,13 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(discovered_peer_count, 1) self.assertEqual(discovered_verses.get("aetheria-main").display_name, "Aetheria") self.assertEqual(discovered_peers.find("aetheria-main", role="read-replica")[0].peer_id, "peer-a") + self.assertEqual(fanout_verse_count, 1) + self.assertEqual(fanout_peer_count, 1) + self.assertEqual(fanout_verses.get("aetheria-main").display_name, "Aetheria") + self.assertIn( + "peer-a", + {peer.peer_id for peer in fanout_peers.find("aetheria-main", role="read-replica")}, + ) def test_cultmesh_local_server_serves_node_and_catalogs_over_clients(self) -> None: document = define_database_entry_type(