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

Accept typed Python CultNet sync responses in CultMesh

Browse source
This commit is contained in:
Meta Vens 2026-06-14 04:26:45 +01:00
commit 27499e766c
4 changed files with 44 additions and 12 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -18,6 +18,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
Provides typed `CultNetRawSnapshotResponse` and `CultNetRawDocumentRecord`
helpers plus `CultNetRawClient.fetch_snapshot_response(...)` for inspecting
peer snapshots before handing them to the existing replication helpers.
`CultMeshDatabase.apply_snapshot_response(...)` and sync paths accept the
typed response while preserving the existing replication owner.
- Python CultNet schema catalog:
Exposes shared wire-message schema descriptors from the package surface so
both the interop peer and CultMesh local server describe the CultNet/CultMesh
@ -34,6 +36,8 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
Provides typed `CultNetShardLogResponse` and `CultNetShardLogEntry` helpers
for inspecting mutation-log entries, resync pressure, and sequence cursors,
while cache mutation remains owned by the existing replication helpers.
`CultMeshDatabase` and `CultMeshGameSession` can consume typed shard-log
responses directly for database sync and prediction reconciliation.
- Python CultNet database subscription changes:
Provides typed `CultNetDatabaseChange` parsing plus
`CultNetDatabaseSubscription.read_next_change(...)` / `iter_changes(...)` for

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

@ -11,6 +11,8 @@ from cultnet_py import (
CultNetAppliedRecord,
CultNetMessage,
CultNetRawClient,
CultNetRawSnapshotResponse,
CultNetShardLogResponse,
apply_raw_snapshot as apply_cultnet_raw_snapshot,
apply_shard_log_response as apply_cultnet_shard_log_response,
document_delete,
@ -459,7 +461,7 @@ class CultMeshDatabase:
shard_id: str | None = None,
shard_epoch: int | None = None,
) -> list[CultNetAppliedRecord]:
response = client.fetch_snapshot(
response = client.fetch_snapshot_response(
schema_ids=schema_ids,
record_keys=record_keys,
shard_id=shard_id,
@ -476,7 +478,7 @@ class CultMeshDatabase:
after_sequence: int = 0,
limit: int | None = None,
) -> list[CultNetAppliedRecord]:
response = client.fetch_shard_log(
response = client.fetch_shard_log_response(
shard_id=shard_id,
shard_epoch=shard_epoch,
after_sequence=after_sequence,
@ -519,15 +521,23 @@ class CultMeshDatabase:
response["shardEpoch"] = self._latest_shard_epoch(shard_id) or 0
return response
def apply_snapshot_response(self, response: dict[str, Any]) -> list[CultNetAppliedRecord]:
previous = self._previous_values_for_snapshot_response(response)
applied = apply_cultnet_raw_snapshot(self.cache, self.documents, response)
def apply_snapshot_response(
self,
response: dict[str, Any] | CultNetRawSnapshotResponse,
) -> list[CultNetAppliedRecord]:
wire = response.to_wire() if isinstance(response, CultNetRawSnapshotResponse) else response
previous = self._previous_values_for_snapshot_response(wire)
applied = apply_cultnet_raw_snapshot(self.cache, self.documents, wire)
self._publish_applied_records(applied, previous)
return applied
def apply_shard_log_response(self, response: dict[str, Any]) -> list[CultNetAppliedRecord]:
previous = self._previous_values_for_shard_log_response(response)
applied = apply_cultnet_shard_log_response(self.cache, self.documents, response)
def apply_shard_log_response(
self,
response: dict[str, Any] | CultNetShardLogResponse,
) -> list[CultNetAppliedRecord]:
wire = response.to_wire() if isinstance(response, CultNetShardLogResponse) else response
previous = self._previous_values_for_shard_log_response(wire)
applied = apply_cultnet_shard_log_response(self.cache, self.documents, wire)
self._publish_applied_records(applied, previous)
return applied

8
packages/cultcache-py/src/cultmesh_py/session.py
View file

@ -8,6 +8,7 @@ from cultnet_py import (
CultNetAppliedRecord,
CultNetClientAuthorityScope,
CultNetRawClient,
CultNetShardLogResponse,
CultNetSimulationConsensusOptions,
CultNetSimulationObservationHub,
)
@ -76,7 +77,10 @@ class CultMeshGameSession:
self._predicted_keys.add((schema_id, key))
return CultMeshPrediction(key=key, schema_id=schema_id, document=document, value=value)
def apply_shard_log_response(self, response: dict[str, Any]) -> list[CultMeshSessionChange]:
def apply_shard_log_response(
self,
response: dict[str, Any] | CultNetShardLogResponse,
) -> list[CultMeshSessionChange]:
applied = self.node.database.apply_shard_log_response(response)
return self._reconcile_applied(applied)
@ -89,7 +93,7 @@ class CultMeshGameSession:
after_sequence: int = 0,
limit: int | None = None,
) -> list[CultMeshSessionChange]:
response = client.fetch_shard_log(
response = client.fetch_shard_log_response(
shard_id=shard_id,
shard_epoch=shard_epoch,
after_sequence=after_sequence,

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

@ -1219,6 +1219,12 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(target.database.get_required(document, "note:1"), {"body": "include"})
self.assertIsNone(target.database.get(document, "note:2"))
typed_target = CultMesh.create_node(runtime_id="mesh-snapshot-typed-target")
typed_target.database.register_document(document)
typed_applied = typed_target.database.apply_snapshot_response(CultNetRawSnapshotResponse.from_wire(response))
self.assertEqual([(record.schema_id, record.record_key) for record in typed_applied], [("mesh.snapshot_note.v1", "note:1")])
self.assertEqual(typed_target.database.get_required(document, "note:1"), {"body": "include"})
def test_cultmesh_database_creates_shard_log_response_from_raw_mutations(self) -> None:
document = define_database_entry_type(
"mesh.log_note",
@ -1265,6 +1271,13 @@ class CultCacheTests(unittest.TestCase):
self.assertIsNone(target.database.get(document, "note:1"))
self.assertIsNone(target.database.get(document, "note:2"))
typed_target = CultMesh.create_node(runtime_id="mesh-log-typed-target")
typed_target.database.register_document(document)
typed_target.database.put(document, "note:1", {"body": "first"})
typed_applied = typed_target.database.apply_shard_log_response(CultNetShardLogResponse.from_wire(response))
self.assertEqual([record.change_kind for record in typed_applied], ["updated", "removed"])
self.assertIsNone(typed_target.database.get(document, "note:1"))
def test_cultmesh_database_watchers_observe_name_and_index_changes(self) -> None:
document = define_database_entry_type(
"mesh.named_watch",
@ -1529,7 +1542,7 @@ class CultCacheTests(unittest.TestCase):
shard_id="inputs",
shard_epoch=1,
)
changes = session.apply_shard_log_response({
authoritative_response = {
"schemaVersion": "cultnet.shard_log_response.v0",
"messageId": "inputs-log",
"shardId": "inputs",
@ -1543,7 +1556,8 @@ class CultCacheTests(unittest.TestCase):
}
],
"resyncRequired": False,
})
}
changes = session.apply_shard_log_response(CultNetShardLogResponse.from_wire(authoritative_response))
self.assertEqual(changes[0].change_kind, "reconciled")
self.assertEqual(node.get_required(note_doc, prediction.key)["body"], "authoritative")