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

Report Python shard log resync pressure

Browse source
This commit is contained in:
Meta Vens 2026-06-14 05:38:25 +01:00
commit 7d6d7fbd3b
3 changed files with 35 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -48,8 +48,9 @@ CultCache, CultNet, and CultMesh. It is evidence, not confetti.
`CultMeshGameSession` can consume typed shard-log responses directly for
database sync and prediction reconciliation.
`CultMeshDatabase.build_shard_log_response(...)` creates the typed response
from the package-local mutation log; `create_shard_log_response(...)` remains
the raw wire-dict compatibility projection.
from the package-local mutation log and reports resync-required responses for
unknown shards or stale epoch cursors; `create_shard_log_response(...)`
remains the raw wire-dict compatibility projection.
- Python CultNet database subscription changes:
Provides typed `CultNetDatabaseChange` parsing plus
`CultNetDatabaseSubscription.read_next_change(...)` / `iter_changes(...)` for

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

@ -596,6 +596,25 @@ class CultMeshDatabase:
raise ValueError("shard_id must be non-empty")
if after_sequence < 0:
raise ValueError("after_sequence must be non-negative")
latest_epoch = self._latest_shard_epoch(shard_id)
if latest_epoch is None:
return CultNetShardLogResponse(
message_id=message_id,
shard_id=shard_id,
shard_epoch=0,
entries=(),
resync_required=True,
reason="unknown_shard",
)
if shard_epoch is not None and shard_epoch != latest_epoch:
return CultNetShardLogResponse(
message_id=message_id,
shard_id=shard_id,
shard_epoch=latest_epoch,
entries=(),
resync_required=True,
reason="stale_epoch",
)
entries = [
CultNetShardLogEntry.from_wire(entry)
for entry in self._shard_logs.get(shard_id, [])
@ -605,14 +624,10 @@ class CultMeshDatabase:
if limit < 0:
raise ValueError("limit must be non-negative")
entries = entries[:limit]
if shard_epoch is not None:
response_shard_epoch = shard_epoch
else:
response_shard_epoch = self._latest_shard_epoch(shard_id) or 0
return CultNetShardLogResponse(
message_id=message_id,
shard_id=shard_id,
shard_epoch=response_shard_epoch,
shard_epoch=latest_epoch,
entries=tuple(entries),
)

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

@ -1415,6 +1415,13 @@ class CultCacheTests(unittest.TestCase):
empty = source.database.create_shard_log_response(shard_id="missing")
self.assertEqual(empty["shardEpoch"], 0)
self.assertEqual(empty["entries"], [])
self.assertTrue(empty["resyncRequired"])
self.assertEqual(empty["reason"], "unknown_shard")
stale = source.database.create_shard_log_response(shard_id="notes", shard_epoch=2)
self.assertEqual(stale["shardEpoch"], 3)
self.assertEqual(stale["entries"], [])
self.assertTrue(stale["resyncRequired"])
self.assertEqual(stale["reason"], "stale_epoch")
with self.assertRaises(ValueError):
source.database.create_shard_log_response(shard_id="notes", after_sequence=-1)
with self.assertRaises(ValueError):
@ -2060,6 +2067,7 @@ class CultCacheTests(unittest.TestCase):
synced_shards = raw_client.sync_shard_catalog(synced_shard_catalog, schema_ids=["mesh.server_note.v1"])
shard_log = raw_client.fetch_shard_log(shard_id="notes", shard_epoch=1)
typed_shard_log = raw_client.fetch_shard_log_response(shard_id="notes", shard_epoch=1)
stale_shard_log = raw_client.fetch_shard_log(shard_id="notes", shard_epoch=0)
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"])
@ -2108,6 +2116,10 @@ class CultCacheTests(unittest.TestCase):
self.assertEqual(synced_shards[0].epoch, 1)
self.assertEqual(synced_shard_catalog.get("notes"), synced_shards[0])
self.assertEqual(shard_log["entries"][0]["changeKind"], "added")
self.assertTrue(stale_shard_log["resyncRequired"])
self.assertEqual(stale_shard_log["reason"], "stale_epoch")
self.assertEqual(stale_shard_log["shardEpoch"], 1)
self.assertEqual(stale_shard_log["entries"], [])
self.assertEqual(typed_shard_log.last_sequence, 1)
self.assertEqual(typed_shard_log.entries[0].put["document"]["recordKey"], "note:1")
self.assertIsInstance(typed_shard_log.entries[0].raw_document, CultNetRawDocumentRecord)