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

Apply Python CultNet raw replication records

Browse source
This commit is contained in:
Meta Vens 2026-06-13 23:28:44 +01:00
commit 25afb9915c
4 changed files with 190 additions and 2 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -147,6 +147,8 @@ member order.
```python
from cultnet_py import (
CultNetRawClient,
apply_raw_snapshot,
apply_shard_log_response,
compute_simulation_claim_hash,
database_subscribe,
decode_frame,
@ -165,9 +167,13 @@ client = CultNetRawClient("127.0.0.1", 3075)
catalog = client.fetch_schema_catalog(kinds=["wire_message"])
snapshot = client.fetch_snapshot(schema_ids=["cultnet.interop-note"])
shard_catalog = client.fetch_shard_catalog(schema_ids=["cultnet.interop-note"])
# With a local CultCache and the matching registered document definitions:
# applied = apply_raw_snapshot(cache, [note_doc], snapshot)
subscription = database_subscribe(subscription_id="ui", schema_ids=["cultnet.interop-note"])
shard_request = shard_catalog_request(message_id="shards", schema_ids=["cultnet.interop-note"])
shard_log = client.fetch_shard_log(shard_id="interop", shard_epoch=1, after_sequence=0)
# log_changes = apply_shard_log_response(cache, [note_doc], shard_log)
claim_hash = compute_simulation_claim_hash("frame:42", "subject:player-1", "hit")
observation = simulation_observation(
message_id="obs-1",
@ -249,8 +255,8 @@ Current receipts:
snapshot, and shard catalog.
- `packages/cultcache-py/tests/test_cultcache.py` covers Python CultMesh
Verse catalog, peer exchange, authority lease, stream negotiation, CultNet
helper shapes, simulation claim hashing, and witness artifact bundle payload
slots.
helper shapes, raw snapshot/shard-log application, simulation claim hashing,
and witness artifact bundle payload slots.
- `cultcache_py`, `cultnet_py`, and `cultmesh_py` ship `py.typed` markers so
downstream type checkers can inspect the package surface instead of treating
the runtime as an untyped xenos swamp.

12
packages/cultcache-py/src/cultnet_py/__init__.py
View file

@ -18,10 +18,21 @@ from .messages import (
snapshot_request,
witness_artifact_bundle,
)
from .replication import (
CultNetAppliedRecord,
apply_raw_document_record,
apply_raw_snapshot,
apply_shard_log_response,
schema_document_map,
)
__all__ = [
"CultNetMessage",
"CultNetRawClient",
"CultNetAppliedRecord",
"apply_raw_document_record",
"apply_raw_snapshot",
"apply_shard_log_response",
"compute_simulation_claim_hash",
"database_subscribe",
"database_unsubscribe",
@ -35,6 +46,7 @@ __all__ = [
"parse_message",
"read_frame",
"schema_catalog_request",
"schema_document_map",
"shard_catalog_request",
"shard_log_request",
"simulation_observation",

95
packages/cultcache-py/src/cultnet_py/replication.py Normal file
View file

@ -0,0 +1,95 @@
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Iterable
from cultcache_py import CultCache, CultCacheEnvelope
from cultcache_py.documents import DocumentDefinition
@dataclass(frozen=True)
class CultNetAppliedRecord:
schema_id: str
record_key: str
change_kind: str
value: Any | None = None
def schema_document_map(documents: Iterable[DocumentDefinition[Any]]) -> dict[str, DocumentDefinition[Any]]:
mapped: dict[str, DocumentDefinition[Any]] = {}
for document in documents:
catalog_entry = document.catalog_entry()
mapped[catalog_entry.schema_id] = document
for compatible_schema_id in catalog_entry.compatible_schema_ids:
mapped[compatible_schema_id] = document
return mapped
def apply_raw_document_record(
cache: CultCache,
documents_by_schema_id: dict[str, DocumentDefinition[Any]],
record: dict[str, Any],
) -> CultNetAppliedRecord:
schema_id = str(record["schemaId"])
document = documents_by_schema_id[schema_id]
envelope = CultCacheEnvelope(
key=str(record["recordKey"]),
type=document.type,
schema_id=schema_id,
payload=bytes(record["payload"]),
stored_at=str(record.get("storedAt") or ""),
catalog_entry=document.catalog_entry(),
)
value = cache.put_envelope(document, envelope)
return CultNetAppliedRecord(
schema_id=schema_id,
record_key=envelope.key,
change_kind="added",
value=value,
)
def apply_raw_snapshot(
cache: CultCache,
documents: Iterable[DocumentDefinition[Any]],
response: dict[str, Any],
) -> list[CultNetAppliedRecord]:
if response.get("schemaVersion") != "cultnet.snapshot_response_raw.v0":
raise ValueError(f"Expected cultnet.snapshot_response_raw.v0, received {response.get('schemaVersion')!r}")
documents_by_schema_id = schema_document_map(documents)
return [
apply_raw_document_record(cache, documents_by_schema_id, record)
for record in response.get("documents", [])
if isinstance(record, dict)
]
def apply_shard_log_response(
cache: CultCache,
documents: Iterable[DocumentDefinition[Any]],
response: dict[str, Any],
) -> list[CultNetAppliedRecord]:
if response.get("schemaVersion") != "cultnet.shard_log_response.v0":
raise ValueError(f"Expected cultnet.shard_log_response.v0, received {response.get('schemaVersion')!r}")
if response.get("resyncRequired") is True:
reason = response.get("reason") or "unspecified"
raise ValueError(f"Shard log response requires resync: {reason}")
documents_by_schema_id = schema_document_map(documents)
applied: list[CultNetAppliedRecord] = []
for entry in response.get("entries", []):
if not isinstance(entry, dict):
continue
change_kind = entry.get("changeKind")
if change_kind in ("added", "updated") and isinstance(entry.get("put"), dict):
put = entry["put"]
result = apply_raw_document_record(cache, documents_by_schema_id, put["document"])
applied.append(CultNetAppliedRecord(result.schema_id, result.record_key, str(change_kind), result.value))
elif change_kind == "removed" and isinstance(entry.get("delete"), dict):
delete = entry["delete"]
schema_id = str(delete["schemaId"])
record_key = str(delete["recordKey"])
document = documents_by_schema_id[schema_id]
cache.delete(document, record_key)
applied.append(CultNetAppliedRecord(schema_id, record_key, "removed", None))
return applied

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

@ -20,6 +20,8 @@ from cultcache_py.interop import read_note, write_note
from cultnet_py import (
compute_simulation_claim_hash,
CultNetRawClient,
apply_raw_snapshot,
apply_shard_log_response,
database_subscribe,
database_unsubscribe,
decode_frame,
@ -331,6 +333,79 @@ class CultCacheTests(unittest.TestCase):
"cultnet.shard_log_request.v0",
])
def test_cultnet_replication_helpers_apply_raw_snapshot_and_shard_log(self) -> None:
document = define_database_entry_type(
"replica.item",
[
("name", 0),
("category", 1),
("value", 2, 0),
],
cls=Item,
)
cache = CultCache()
cache.register_document_type(document)
schema_id = document.catalog_entry().schema_id
snapshot = {
"schemaVersion": "cultnet.snapshot_response_raw.v0",
"messageId": "snapshot-1",
"documents": [
{
"schemaId": schema_id,
"recordKey": "item:1",
"storedAt": "2026-06-13T00:00:00Z",
"payloadEncoding": "messagepack",
"payload": document.encode_payload(Item("sword", "gear", 3)),
}
],
}
applied_snapshot = apply_raw_snapshot(cache, [document], snapshot)
self.assertEqual(applied_snapshot[0].record_key, "item:1")
self.assertEqual(cache.get_required(document, "item:1").value, 3)
shard_log = {
"schemaVersion": "cultnet.shard_log_response.v0",
"messageId": "log-1",
"shardId": "interop",
"shardEpoch": 1,
"resyncRequired": False,
"entries": [
{
"sequence": 1,
"changeKind": "updated",
"put": {
"schemaVersion": "cultnet.document_put_raw.v0",
"messageId": "put-1",
"document": {
"schemaId": schema_id,
"recordKey": "item:1",
"storedAt": "2026-06-13T00:00:01Z",
"payloadEncoding": "messagepack",
"payload": document.encode_payload(Item("sword", "gear", 12)),
},
"shardId": "interop",
"shardEpoch": 1,
},
},
{
"sequence": 2,
"changeKind": "removed",
"delete": {
"schemaVersion": "cultnet.document_delete.v0",
"messageId": "delete-1",
"schemaId": schema_id,
"recordKey": "item:1",
"shardId": "interop",
"shardEpoch": 1,
},
},
],
}
applied_log = apply_shard_log_response(cache, [document], shard_log)
self.assertEqual([change.change_kind for change in applied_log], ["updated", "removed"])
self.assertIsNone(cache.get(document, "item:1"))
def test_cultnet_document_delete_helper_matches_schema_v0_shape(self) -> None:
message = document_delete(
message_id="delete-1",