From abdfff6f4e82562d0c302896de0163822ec3d883 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Sat, 27 Jun 2026 23:37:15 +0100 Subject: [PATCH] Add publication sync and reconciliation coverage --- packages/cultcache-py/tests/test_cultcache.py | 56 +++++++++++++ src/GameCult.Mesh/CultMesh.cs | 83 +++++++++++++++++++ .../CultMeshStreamingTests.cs | 56 +++++++++++++ 3 files changed, 195 insertions(+) diff --git a/packages/cultcache-py/tests/test_cultcache.py b/packages/cultcache-py/tests/test_cultcache.py index fb8b2b2b..bc27892f 100644 --- a/packages/cultcache-py/tests/test_cultcache.py +++ b/packages/cultcache-py/tests/test_cultcache.py @@ -3742,6 +3742,62 @@ class CultCacheTests(unittest.TestCase): self.assertEqual(session.pending_predictions(), ()) self.assertEqual(session.resimulation_inputs(), ()) + def test_cultmesh_game_session_partial_reconciliation_keeps_unmatched_predictions(self) -> None: + note_doc = define_database_entry_type( + "mesh.partial_input", + [("body", 0)], + schema_id="mesh.partial_input.v1", + ) + node = CultMesh.create_node(runtime_id="client-partial") + node.register_document(note_doc) + session = CultMesh.create_game_session( + node, + CultMeshGameSessionOptions( + client_authority_scopes=( + CultNetClientAuthorityScope( + "client-partial", + schema_ids=("mesh.partial_input.v1",), + key_prefix="input:client-partial", + ), + ) + ), + ) + pending = session.predict(note_doc, "input:client-partial:pending", {"body": "pending-local"}) + accepted = session.predict(note_doc, "input:client-partial:accepted", {"body": "accepted-local"}) + put = document_put_raw( + message_id="authoritative-partial", + key=accepted.key, + schema_id=accepted.schema_id, + stored_at="2026-06-13T00:00:00Z", + payload=note_doc.encode_payload({"body": "accepted-authority"}), + shard_id="inputs", + shard_epoch=1, + ) + + changes = session.apply_shard_log_response({ + "schemaVersion": "cultnet.shard_log_response.v0", + "messageId": "inputs-partial-log", + "shardId": "inputs", + "shardEpoch": 1, + "entries": [ + { + "sequence": 1, + "committedAt": "2026-06-13T00:00:00Z", + "changeKind": "updated", + "put": put.to_wire(), + } + ], + "resyncRequired": False, + }) + + self.assertEqual([(change.change_kind, change.record_key) for change in changes], [ + ("reconciled", accepted.key), + ]) + self.assertEqual(node.get_required(note_doc, accepted.key)["body"], "accepted-authority") + self.assertEqual(node.get_required(note_doc, pending.key)["body"], "pending-local") + self.assertEqual([prediction.key for prediction in session.pending_predictions()], [pending.key]) + self.assertEqual([prediction.key for prediction in session.resimulation_inputs()], [pending.key]) + def test_cultmesh_game_session_rolls_back_pending_predictions(self) -> None: note_doc = define_database_entry_type( "mesh.rollback_input", diff --git a/src/GameCult.Mesh/CultMesh.cs b/src/GameCult.Mesh/CultMesh.cs index 29b47fb6..e06ced87 100644 --- a/src/GameCult.Mesh/CultMesh.cs +++ b/src/GameCult.Mesh/CultMesh.cs @@ -1544,6 +1544,89 @@ namespace GameCult.Mesh return DocumentFromPublication(source, key, verse.Context, storeOptions, peerOptions); } + /// + /// Reads one typed document from a configured publication source and hydrates it into a local node. + /// + public static async Task SyncDocumentFromPublicationAsync( + CultMeshNode node, + CultMeshDocumentPublicationSource source, + CultRecordKey key, + CultMeshVerseContext context, + CultMeshStoreDocumentOptions? storeOptions = null, + CultMeshPeerSnapshotDocumentOptions? peerOptions = null, + bool flush = false) + where TDocument : class + { + if (node == null) throw new ArgumentNullException(nameof(node)); + if (source == null) throw new ArgumentNullException(nameof(source)); + if (context == null) throw new ArgumentNullException(nameof(context)); + + var document = await DocumentFromPublication( + source, + key, + context, + storeOptions, + peerOptions) + .LatestAsync() + .ConfigureAwait(false); + + await Document(node, key, context) + .ReplaceAsync(document) + .ConfigureAwait(false); + + if (flush) + await node.FlushAsync().ConfigureAwait(false); + + return document; + } + + /// + /// Reads one typed document from a configured publication source and hydrates it into a local node. + /// + public static Task SyncDocumentFromPublicationAsync( + CultMeshNode node, + CultMeshDocumentPublicationSource source, + CultRecordKey key, + CultMeshVerse verse, + CultMeshStoreDocumentOptions? storeOptions = null, + CultMeshPeerSnapshotDocumentOptions? peerOptions = null, + bool flush = false) + where TDocument : class + { + if (verse == null) throw new ArgumentNullException(nameof(verse)); + return SyncDocumentFromPublicationAsync( + node, + source, + key, + verse.Context, + storeOptions, + peerOptions, + flush); + } + + /// + /// Reads one typed document from a configured publication source and hydrates it into a local node. + /// + public static Task SyncDocumentFromPublicationAsync( + CultMeshNode node, + CultMeshDocumentPublicationSource source, + string recordKey, + CultMeshVerse verse, + CultMeshStoreDocumentOptions? storeOptions = null, + CultMeshPeerSnapshotDocumentOptions? peerOptions = null, + bool flush = false) + where TDocument : class + { + if (string.IsNullOrWhiteSpace(recordKey)) throw new ArgumentException("Value must be non-empty.", nameof(recordKey)); + return SyncDocumentFromPublicationAsync( + node, + source, + new CultRecordKey(recordKey), + verse, + storeOptions, + peerOptions, + flush); + } /// /// Creates a typed document handle directly over one distributed CultNet database record. /// diff --git a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs index 4f6114ea..45fbf6d3 100644 --- a/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs +++ b/tests/GameCult.Mesh.Tests/CultMeshStreamingTests.cs @@ -2305,6 +2305,62 @@ public sealed class CultMeshStreamingTests (await mixedAliasSchemaHandle.LatestAsync()).Text.Should().Be("compatible-record-key"); } + [Test] + public async Task PublicationHelpers_SyncConfiguredPeerSourceIntoLocalNodeAlias() + { + var sourceCache = new CultCache(); + var registry = new CultNetDocumentRegistry(CultDocumentRegistry.Shared); + registry.Register(CultNetDocumentBinding.ForDocument(sourceCache.Registry)); + var key = new CultRecordKey("mesh-note:publication-sync-alias"); + await sourceCache.UpsertAsync(new MeshNoteDocument + { + Schema = "tests.mesh_note.v1", + Text = "publication-synced", + Revision = 51 + }, new CultRecordHandle(key)); + var requests = new List(); + var verse = CultMesh.Verse("starbridge", "unity-support"); + using var node = await CultMesh.CreateNodeAsync( + Path.Combine(Path.GetTempPath(), $"cultmesh-publication-sync-alias-{Guid.NewGuid():N}.ccmp"), + new CultMeshNodeOptions + { + StartServer = false, + CacheOptions = new CultCacheOpenOptions + { + Registry = sourceCache.Registry, + PullOnOpen = false + }, + DatabaseOptions = new CultNetDatabaseOptions + { + DocumentRegistry = registry + } + }); + + var synced = await CultMesh.SyncDocumentFromPublicationAsync( + node, + CultMeshDocumentPublicationSource.PeerSnapshot( + () => new MeshSnapshotSchemaClient(request => + { + requests.Add(request); + return registry.CreateRawSnapshotResponse(sourceCache, request.MessageId, request); + }), + "cultnet://publication-sync.test:3075"), + key, + verse, + peerOptions: new CultMeshPeerSnapshotDocumentOptions + { + MessageIdPrefix = "mesh-test-publication-sync", + PollInterval = TimeSpan.FromMilliseconds(10) + }); + var localAlias = await node.Document(key, verse).LatestAsync(); + + synced.Text.Should().Be("publication-synced"); + synced.Revision.Should().Be(51); + localAlias.Text.Should().Be("publication-synced"); + localAlias.Revision.Should().Be(51); + requests.Should().ContainSingle().Which.RecordKeys.Should().Equal(key.Value); + } + [Test] public async Task SnapshotHelpers_FetchApplyAndDecodeScopedSnapshots() {