From 7d6d7fbd3bf157f9148ee1dc0d90c5d20c077fcd Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sun, 14 Jun 2026 05:38:25 +0100 Subject: [PATCH] Report Python shard log resync pressure --- packages/cultcache-py/PARITY.md | 5 ++-- packages/cultcache-py/src/cultmesh_py/node.py | 25 +++++++++++++++---- packages/cultcache-py/tests/test_cultcache.py | 12 +++++++++ 3 files changed, 35 insertions(+), 7 deletions(-) diff --git a/packages/cultcache-py/PARITY.md b/packages/cultcache-py/PARITY.md index 915d8067..7715be67 100644 --- a/packages/cultcache-py/PARITY.md +++ b/packages/cultcache-py/PARITY.md @@ -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 diff --git a/packages/cultcache-py/src/cultmesh_py/node.py b/packages/cultcache-py/src/cultmesh_py/node.py index 39109aae..223fc4d2 100644 --- a/packages/cultcache-py/src/cultmesh_py/node.py +++ b/packages/cultcache-py/src/cultmesh_py/node.py @@ -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), ) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index 5bc49a11..0af2f5f9 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -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)