diff --git a/packages/cultcache-py/README.md b/packages/cultcache-py/README.md index e9b076fa..113e77d8 100644 --- a/packages/cultcache-py/README.md +++ b/packages/cultcache-py/README.md @@ -212,12 +212,14 @@ checks, and stream transport negotiation: ```python from cultcache_py import define_database_entry_type +from cultnet_py import CultNetRawClient from cultmesh_py import CultMeshDiscoveryClient, CultMeshPeerCatalog, CultMeshStreamCatalog, create_node, peer_exchange_request note_doc = define_database_entry_type("mesh.note", [("body", 0)]) node = create_node("mesh.cc", runtime_id="python-runtime") node.register_document(note_doc) node.put(note_doc, "note:1", {"body": "hello"}) +live_note = node.get_required(note_doc, "note:1") peers = CultMeshPeerCatalog() response = peers.create_response(peer_exchange_request("pex-1", verse_id="local")) @@ -227,6 +229,10 @@ 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"]) +raw_client = CultNetRawClient("127.0.0.1", 3075) +node.sync_snapshot(raw_client, schema_ids=[note_doc.catalog_entry().schema_id]) +node.sync_shard_log(raw_client, shard_id="interop", shard_epoch=1) + streams = CultMeshStreamCatalog() ``` diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 40fc1c5b..a1e0026b 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -6,15 +6,18 @@ from typing import Any from cultcache_py import CultCache, SingleFileMessagePackBackingStore from cultcache_py.documents import DocumentDefinition +from cultnet_py import CultNetAppliedRecord, CultNetRawClient, apply_raw_snapshot, apply_shard_log_response @dataclass class CultMeshNode: cache: CultCache = field(default_factory=CultCache) runtime_id: str = "python-runtime" + documents: list[DocumentDefinition[Any]] = field(default_factory=list) def register_document(self, document: DocumentDefinition[Any]) -> None: self.cache.register_document_type(document) + self.documents.append(document) def put(self, document: DocumentDefinition[Any], key: str, value: Any) -> None: self.cache.put(document, key, value) @@ -22,9 +25,52 @@ class CultMeshNode: def get(self, document: DocumentDefinition[Any], key: str) -> Any: return self.cache.get(document, key) + def get_required(self, document: DocumentDefinition[Any], key: str) -> Any: + return self.cache.get_required(document, key) + + def delete(self, document: DocumentDefinition[Any], key: str) -> None: + self.cache.delete(document, key) + + def snapshot(self) -> dict[str, dict[str, Any]]: + return self.cache.snapshot() + def pull(self) -> None: self.cache.pull_all_backing_stores() + def sync_snapshot( + self, + client: CultNetRawClient, + *, + schema_ids: list[str] | None = None, + record_keys: list[str] | None = None, + shard_id: str | None = None, + shard_epoch: int | None = None, + ) -> list[CultNetAppliedRecord]: + response = client.fetch_snapshot( + schema_ids=schema_ids, + record_keys=record_keys, + shard_id=shard_id, + shard_epoch=shard_epoch, + ) + return apply_raw_snapshot(self.cache, self.documents, response) + + def sync_shard_log( + self, + client: CultNetRawClient, + *, + shard_id: str, + shard_epoch: int | None = None, + after_sequence: int = 0, + limit: int | None = None, + ) -> list[CultNetAppliedRecord]: + response = client.fetch_shard_log( + shard_id=shard_id, + shard_epoch=shard_epoch, + after_sequence=after_sequence, + limit=limit, + ) + return apply_shard_log_response(self.cache, self.documents, response) + def create_node(cache_path: str | Path | None = None, *, runtime_id: str = "python-runtime") -> CultMeshNode: cache = CultCache() diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index c862d060..c0ece8cf 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -420,6 +420,103 @@ class CultCacheTests(unittest.TestCase): self.assertEqual([change.change_kind for change in applied_log], ["updated", "removed"]) self.assertIsNone(cache.get(document, "item:1")) + def test_cultmesh_node_syncs_snapshot_and_shard_log_through_cultnet_client(self) -> None: + import msgpack # type: ignore + from cultnet_py import read_frame, write_frame + + document = define_database_entry_type( + "mesh.sync_item", + [ + ("name", 0), + ("category", 1), + ("value", 2, 0), + ], + cls=Item, + ) + node = create_node(runtime_id="python-node") + node.register_document(document) + schema_id = document.catalog_entry().schema_id + ready = threading.Event() + server_error: list[BaseException] = [] + port_holder: list[int] = [] + + def serve_requests() -> None: + try: + with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as server: + server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + server.bind(("127.0.0.1", 0)) + port_holder.append(server.getsockname()[1]) + server.listen(2) + ready.set() + for _ in range(2): + connection, _ = server.accept() + with connection: + stream = connection.makefile("rwb") + request = msgpack.unpackb(read_frame(stream), raw=False) + if request["schemaVersion"] == "cultnet.snapshot_request.v0": + response = { + "schemaVersion": "cultnet.snapshot_response_raw.v0", + "messageId": request["messageId"], + "documents": [ + { + "schemaId": schema_id, + "recordKey": "item:node", + "storedAt": "2026-06-13T00:00:00Z", + "payloadEncoding": "messagepack", + "payload": document.encode_payload(Item("node", "mesh", 1)), + } + ], + } + elif request["schemaVersion"] == "cultnet.shard_log_request.v0": + response = { + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": request["messageId"], + "shardId": request["shardId"], + "shardEpoch": request["shardEpoch"], + "resyncRequired": False, + "entries": [ + { + "sequence": 1, + "changeKind": "updated", + "put": { + "schemaVersion": "cultnet.document_put_raw.v0", + "messageId": "put-node", + "document": { + "schemaId": schema_id, + "recordKey": "item:node", + "storedAt": "2026-06-13T00:00:01Z", + "payloadEncoding": "messagepack", + "payload": document.encode_payload(Item("node", "mesh", 9)), + }, + "shardId": request["shardId"], + "shardEpoch": request["shardEpoch"], + }, + } + ], + } + else: + raise AssertionError(f"unexpected request {request['schemaVersion']}") + write_frame(stream, msgpack.packb(response, use_bin_type=True)) + stream.flush() + except BaseException as error: + server_error.append(error) + ready.set() + + thread = threading.Thread(target=serve_requests, daemon=True) + thread.start() + self.assertTrue(ready.wait(2.0)) + self.assertFalse(server_error) + + client = CultNetRawClient("127.0.0.1", port_holder[0], timeout_seconds=2.0) + snapshot_changes = node.sync_snapshot(client, schema_ids=[schema_id]) + log_changes = node.sync_shard_log(client, shard_id="mesh", shard_epoch=1) + + thread.join(2.0) + self.assertFalse(server_error) + self.assertEqual(snapshot_changes[0].record_key, "item:node") + self.assertEqual(log_changes[0].change_kind, "updated") + self.assertEqual(node.get_required(document, "item:node").value, 9) + def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None: message = document_delete( message_id="delete-1", diff --git a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts index b2d3e21d..228d8223 100644 --- a/packages/cultnet-ts/test/interop/cultnet-interop.test.ts +++ b/packages/cultnet-ts/test/interop/cultnet-interop.test.ts @@ -302,6 +302,10 @@ test("CultNet TS/Rust/C#/Python peers discover each other and exchange raw state assert.ok(pythonCultNetClient.snapshotRecordKeys.includes("note:python-peer")); assert.equal(pythonCultNetClient.shards[0].shardId, "interop"); + const pythonCultMeshNodeSync = await runPythonCultMeshNodeSync(pythonPort); + assert.equal(pythonCultMeshNodeSync.synced[0].recordKey, "note:python-peer"); + assert.equal(pythonCultMeshNodeSync.note.authorRuntimeId, "python-peer"); + const pythonChange = await requestPythonDatabaseChange(pythonPort, { schemaVersion: "cultnet.database_subscribe.v0", messageId: "python-subscribe", @@ -761,6 +765,43 @@ async function runPythonCultNetRawClient(port: number): Promise { return JSON.parse(stdout); } +async function runPythonCultMeshNodeSync(port: number): Promise { + const schemaPathJson = JSON.stringify(interopSchemaPath); + const script = [ + "import json", + "from pathlib import Path", + "from cultcache_py import define_database_entry_type", + "from cultmesh_py import create_node", + "from cultnet_py import CultNetRawClient", + `schema_json = Path(${schemaPathJson}).read_text(encoding='utf-8')`, + "note_doc = define_database_entry_type(", + " 'cultnet.interop-note',", + " [('schemaVersion', 0), ('documentId', 1), ('authorRuntimeId', 2), ('title', 3), ('body', 4), ('tags', 5, [])],", + ` schema_id='${interopNoteSchemaId}',`, + " schema_version='cultnet.interop_note.v0',", + " canonical_schema_json=schema_json,", + ")", + "node = create_node(runtime_id='python-node-sync')", + "node.register_document(note_doc)", + `client = CultNetRawClient('127.0.0.1', ${port}, timeout_seconds=4.0)`, + `changes = node.sync_snapshot(client, schema_ids=['${interopNoteSchemaId}'], record_keys=['note:python-peer'])`, + "note = node.get_required(note_doc, 'note:python-peer')", + "print(json.dumps({", + " 'synced': [{'schemaId': change.schema_id, 'recordKey': change.record_key} for change in changes],", + " 'note': note,", + "}))", + ].join("\n"); + const { stdout } = await execFileAsync(pythonCommand, ["-c", script], { + cwd: cultcachePyRoot, + env: { + ...process.env, + PYTHONPATH: cultcachePySrc, + }, + timeout: 8000, + }); + return JSON.parse(stdout); +} + async function requestPythonDatabaseChange(port: number, subscribe: Record, put: Record): Promise { const socket = connectTcp(port, "127.0.0.1"); const framer = new LengthPrefixedMessageFramer();