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

Add Python CultMesh local server subscriptions

Browse source
This commit is contained in:
Meta Vens 2026-06-14 03:26:07 +01:00
commit f7066836e7
3 changed files with 190 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

3
packages/cultcache-py/PARITY.md
View file

@ -24,7 +24,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
- Python CultMesh local server:
Serves a `CultMeshNode` plus Verse and peer catalogs over the same framed
MessagePack request/response lane used by `CultNetRawClient` and
`CultMeshDiscoveryClient`.
`CultMeshDiscoveryClient`. It also supports live raw database subscriptions
for snapshot, put, and delete notifications on a connected client stream.
- Python CultMesh reactive catalogs:
Verse and peer catalogs expose sorted local views, `get(...)`, response
application, and `watch(...)` callbacks for local discovery updates. Verse

165
packages/cultcache-py/src/cultmesh_py/server.py
View file

@ -7,7 +7,8 @@ from typing import Any
import msgpack
from cultnet_py import read_frame, write_frame
from cultcache_py import CultCacheEnvelope
from cultnet_py import CultNetMessage, read_frame, write_frame
from .node import CultMeshNode
from .wire import (
@ -18,6 +19,21 @@ from .wire import (
)
@dataclass(frozen=True)
class _DatabaseSubscription:
subscription_id: str
schema_ids: set[str]
record_keys: set[str]
def matches(self, document: dict[str, Any]) -> bool:
schema_id = str(document.get("schemaId") or "")
record_key = str(document.get("recordKey") or "")
return (
(not self.schema_ids or schema_id in self.schema_ids)
and (not self.record_keys or record_key in self.record_keys)
)
@dataclass
class CultMeshLocalServer:
node: CultMeshNode
@ -110,6 +126,7 @@ class CultMeshLocalServer:
def _handle_connection(self, client: socket.socket) -> None:
with client:
stream = client.makefile("rwb")
subscriptions: dict[str, _DatabaseSubscription] = {}
while not self._stop.is_set():
try:
message = msgpack.unpackb(read_frame(stream), raw=False)
@ -117,11 +134,126 @@ class CultMeshLocalServer:
return
if not isinstance(message, dict):
continue
response = self.handle_message(message)
if response is None:
continue
write_frame(stream, msgpack.packb(response, use_bin_type=True))
stream.flush()
responses = self._handle_connection_message(message, subscriptions)
for response in responses:
write_frame(stream, msgpack.packb(response, use_bin_type=True))
stream.flush()
def _handle_connection_message(
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
) -> list[dict[str, Any]]:
schema_version = message.get("schemaVersion")
if schema_version == "cultnet.database_subscribe.v0":
subscription_id = str(message.get("subscriptionId") or message.get("messageId") or "")
if not subscription_id:
return []
subscriptions[subscription_id] = _DatabaseSubscription(
subscription_id=subscription_id,
schema_ids={str(value) for value in message.get("schemaIds") or []},
record_keys={str(value) for value in message.get("recordKeys") or []},
)
if message.get("includeSnapshot", True) is False:
return []
return [
self.node.database.create_snapshot_response(
message_id=str(message.get("messageId") or ""),
schema_ids=list(subscriptions[subscription_id].schema_ids),
record_keys=list(subscriptions[subscription_id].record_keys),
)
]
if schema_version == "cultnet.database_unsubscribe.v0":
subscriptions.pop(str(message.get("subscriptionId") or ""), None)
return []
if schema_version == "cultnet.document_put_raw.v0":
return self._handle_raw_put(message, subscriptions)
if schema_version == "cultnet.document_delete.v0":
return self._handle_raw_delete(message, subscriptions)
response = self.handle_message(message)
return [] if response is None else [response]
def _handle_raw_put(
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
) -> list[dict[str, Any]]:
document_record = message.get("document")
if not isinstance(document_record, dict):
return []
schema_id = str(document_record.get("schemaId") or "")
document = self._document_for_schema(schema_id)
if document is None:
return []
record_key = str(document_record["recordKey"])
previous = self.node.cache.get(document, record_key)
value = self.node.cache.put_envelope(document, self._envelope_from_raw_record(document, document_record))
change_kind = "added" if previous is None else "updated"
self.node.database._publish_local_change(document, record_key, change_kind, value, previous)
self.node.database._append_shard_log_put(CultNetMessage("cultnet.document_put_raw.v0", {
key: value for key, value in message.items() if key != "schemaVersion"
}), change_kind)
return self._database_change_notifications(message, document_record, change_kind, subscriptions)
def _handle_raw_delete(
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
) -> list[dict[str, Any]]:
schema_id = str(message.get("schemaId") or "")
record_key = str(message.get("recordKey") or "")
document = self._document_for_schema(schema_id)
if document is None or not record_key:
return []
self.node.database.delete(document, record_key)
self.node.database._append_shard_log_delete(CultNetMessage("cultnet.document_delete.v0", {
key: value for key, value in message.items() if key != "schemaVersion"
}))
return self._database_delete_notifications(message, subscriptions)
def _database_change_notifications(
self,
message: dict[str, Any],
document: dict[str, Any],
change_kind: str,
subscriptions: dict[str, _DatabaseSubscription],
) -> list[dict[str, Any]]:
notifications = []
for subscription in subscriptions.values():
if not subscription.matches(document):
continue
message_id = message.get("messageId") or document.get("recordKey") or "change"
notifications.append({
"schemaVersion": "cultnet.database_change_raw.v0",
"messageId": f"{message_id}:{subscription.subscription_id}",
"subscriptionId": subscription.subscription_id,
"changeKind": change_kind,
"document": document,
})
return notifications
def _database_delete_notifications(
self,
message: dict[str, Any],
subscriptions: dict[str, _DatabaseSubscription],
) -> list[dict[str, Any]]:
schema_id = str(message.get("schemaId") or "")
record_key = str(message.get("recordKey") or "")
notifications = []
document = {"schemaId": schema_id, "recordKey": record_key}
for subscription in subscriptions.values():
if not subscription.matches(document):
continue
message_id = message.get("messageId") or record_key or "change"
notifications.append({
"schemaVersion": "cultnet.database_change_raw.v0",
"messageId": f"{message_id}:{subscription.subscription_id}",
"subscriptionId": subscription.subscription_id,
"changeKind": "removed",
"schemaId": schema_id,
"recordKey": record_key,
})
return notifications
def _hello_response(self) -> dict[str, Any]:
return {
@ -134,6 +266,10 @@ class CultMeshLocalServer:
"cultnet.hello.v0",
"cultnet.schema_catalog_request.v0",
"cultnet.snapshot_request.v0",
"cultnet.database_subscribe.v0",
"cultnet.database_unsubscribe.v0",
"cultnet.document_put_raw.v0",
"cultnet.document_delete.v0",
"cultnet.shard_catalog_request.v0",
"cultnet.shard_log_request.v0",
VERSE_CATALOG_REQUEST,
@ -195,3 +331,20 @@ class CultMeshLocalServer:
"messageId": request.get("messageId", ""),
"shards": shards,
}
def _document_for_schema(self, schema_id: str) -> Any:
for document in self.node.documents:
entry = document.catalog_entry()
if schema_id == entry.schema_id or schema_id in entry.compatible_schema_ids:
return document
return None
def _envelope_from_raw_record(self, document: Any, record: dict[str, Any]) -> Any:
return CultCacheEnvelope(
key=str(record["recordKey"]),
type=document.type,
schema_id=str(record["schemaId"]),
payload=bytes(record["payload"]),
stored_at=str(record.get("storedAt") or ""),
catalog_entry=document.catalog_entry(),
)

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

@ -1663,11 +1663,33 @@ class CultCacheTests(unittest.TestCase):
discovery_client = CultMesh.create_verse_discovery_client("127.0.0.1", server.port, timeout_seconds=2.0)
fetched_verses = discovery_client.fetch_verses(transport_version="cultmesh.v0")
fetched_peers = discovery_client.fetch_peers(verse_id="server-verse", roles=["read-replica"])
with raw_client.subscribe_database(subscription_id="sub-server", schema_ids=["mesh.server_note.v1"]) as subscription:
subscription_snapshot = subscription.read_next()
subscription.send(document_put_raw(
message_id="server-put",
key="note:2",
schema_id="mesh.server_note.v1",
stored_at="2026-06-14T00:00:00Z",
payload=document.encode_payload({"body": "subscribed"}),
shard_id="primary",
shard_epoch=1,
))
subscription_change = subscription.read_next()
subscription.send(document_delete(
message_id="server-delete",
schema_id="mesh.server_note.v1",
record_key="note:2",
shard_id="primary",
shard_epoch=1,
))
subscription_delete = subscription.read_next()
finally:
server.stop()
self.assertEqual(hello_response["runtimeId"], "mesh-server")
self.assertEqual(hello_response["displayName"], "Mesh Server")
self.assertIn("cultnet.database_subscribe.v0", hello_response["supportedMessageVersions"])
self.assertIn("cultnet.document_put_raw.v0", hello_response["supportedMessageVersions"])
self.assertEqual(schema_response["schemas"][0]["schemaId"], "mesh.server_note.v1")
self.assertIn("schemaJson", schema_response["schemas"][0])
self.assertEqual(snapshot_response["documents"][0]["recordKey"], "note:1")
@ -1675,6 +1697,13 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(shard_log["entries"][0]["changeKind"], "added")
self.assertEqual(fetched_verses[0].verse_id, "server-verse")
self.assertEqual(fetched_peers[0].peer_id, "mesh-server")
self.assertEqual(subscription_snapshot["schemaVersion"], "cultnet.snapshot_response_raw.v0")
self.assertEqual(subscription_change["schemaVersion"], "cultnet.database_change_raw.v0")
self.assertEqual(subscription_change["changeKind"], "added")
self.assertEqual(subscription_change["document"]["recordKey"], "note:2")
self.assertEqual(subscription_delete["changeKind"], "removed")
self.assertEqual(subscription_delete["recordKey"], "note:2")
self.assertIsNone(node.database.get(document, "note:2"))
def test_cultmesh_authority_lease_requires_live_matching_lease(self) -> None:
peer = CultMeshPeerCard(