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

Sync Python CultMesh nodes from CultNet

Browse source
This commit is contained in:
Meta Vens 2026-06-13 23:51:00 +01:00
commit 17b15c972a
4 changed files with 190 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -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()
```

46
packages/cultcache-py/src/cultmesh_py/node.py
View file

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

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

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

41
packages/cultnet-ts/test/interop/cultnet-interop.test.ts
View file

@ -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<any> {
return JSON.parse(stdout);
}
async function runPythonCultMeshNodeSync(port: number): Promise<any> {
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<string, unknown>, put: Record<string, unknown>): Promise<any> {
const socket = connectTcp(port, "127.0.0.1");
const framer = new LengthPrefixedMessageFramer();