From d8d973c4185bc7dfbc8a8c1743bce3f5e9d51cd6 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 13:21:39 +0200 Subject: [PATCH] Pin the single-owner publication cases Soul falsified Observer rewrite of the door key, variant log order, replica log sequences with and without a local write, 4x5000 concurrent writers, a write after a database delete, shard-snapshot resync, and a disposed database. Co-Authored-By: Claude Sonnet 5 --- .../CultNetPublishesCommitsTests.cs | 199 ++++++++++++++++++ 1 file changed, 199 insertions(+) diff --git a/tests/GameCult.Networking.Tests/CultNetPublishesCommitsTests.cs b/tests/GameCult.Networking.Tests/CultNetPublishesCommitsTests.cs index 08ef57cb..fbd1df2a 100644 --- a/tests/GameCult.Networking.Tests/CultNetPublishesCommitsTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetPublishesCommitsTests.cs @@ -10,6 +10,7 @@ using System.Threading.Tasks; using GameCult.Caching; using GameCult.Caching.MessagePack; using NUnit.Framework; +using R3; using static GameCult.Networking.Tests.NetworkingTests; using static GameCult.Networking.Tests.CultNetDocumentRegistryVariantTests; @@ -226,5 +227,203 @@ namespace GameCult.Networking.Tests Assert.That(Of(wire, baseKey.Value), Has.Length.EqualTo(1)); Assert.That(Of(wire, variantKey.Value), Is.Empty, "a variant is never carried as a record"); } + + // An observer that rewrites the very key a door just wrote is a second write: it publishes and logs on its own. + [Test] + public async Task AnObserverRewriteOfTheDoorsOwnKeyIsPublishedAndLoggedToo() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + using var wire = new Wire(database); + await wire.SubscribeAsync(false, typeof(NetworkSchemaNote)); + var key = new CultRecordKey("tests:publish:normalized"); + using var normalizer = cache.Watch().Subscribe(change => + { + if (change.Document?.Text == "raw") + cache.UpsertAsync(Note("normalized"), new CultRecordHandle(change.Key)).GetAwaiter().GetResult(); + }); + + await database.PutAsync(key, Note("raw")); + await Until(() => Of(wire, key.Value).Length >= 2); + await SettleAsync(database, wire); + + Assert.That(cache.Get(key)!.Text, Is.EqualTo("normalized")); + Assert.That(Of(wire, key.Value).Select(change => change.ChangeKind), Is.EqualTo(new[] { "added", "updated" })); + Assert.That(database.GetMutationLog(database.Shards[0].ShardId).Count(entry => entry.Key.Value == key.Value), Is.EqualTo(2)); + } + + private static (CultCache Cache, CultNetDocumentRegistry Documents) VariantCache() + { + var registry = CultDocumentRegistry.Shared; + var cache = new CultCache(registry, CultCacheMessagePack.CreateCodec(registry)); + var documents = new CultNetDocumentRegistry(registry) + .Register(CultNetDocumentBinding.ForDocument(registry)); + return (cache, documents); + } + + private static System.Collections.Generic.List Record(CultNetDatabase database, Func describe) + { + var seen = new System.Collections.Generic.List(); + database.WatchAllChanges().Subscribe(change => seen.Add(describe(change))); + return seen; + } + + private static string KeyOfChange(object change) => ((dynamic)change).Key.Value; + + [Test] + public async Task ABaseWriteAndItsReResolvedVariantAreLoggedInCacheOrder() + { + var (cache, documents) = VariantCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { DocumentRegistry = documents }); + var baseKey = new CultRecordKey("publish-variant-base"); + var variantKey = new CultRecordKey("publish-variant-big"); + await cache.UpsertAsync(new VariantWireFixture { Name = "base", Power = 1 }, new CultRecordHandle(baseKey)); + await cache.UpsertVariantAsync(variantKey, baseKey, new[] + { + cache.Override(nameof(VariantWireFixture.Name), "big") + }); + var published = Record(database, KeyOfChange); + var logged = database.GetMutationLog(database.Shards[0].ShardId).Count; + + await database.PutAsync(baseKey, new VariantWireFixture { Name = "base edited", Power = 2 }); + + Assert.That(published, Is.EqualTo(new[] { baseKey.Value, variantKey.Value })); + Assert.That(database.GetMutationLog(database.Shards[0].ShardId, logged).Select(entry => entry.Key.Value), + Is.EqualTo(new[] { baseKey.Value, variantKey.Value })); + } + + private static CultNetDatabase Replica(CultCache cache, CultNetDocumentRegistry? documents, string schemaId) => + new(cache, new CultNetDatabaseOptions + { + DocumentRegistry = documents, + Shards = + [ + new CultNetShardDescriptor("replica-shard", "runtime-a", epoch: 3, isPrimary: false, schemaIds: [schemaId]) + ] + }); + + [Test] + public async Task AReplicaLogHoldsOnlyThePrimarysSequencesWhateverItResolvesLocally() + { + var (sourceCache, sourceDocuments) = VariantCache(); + var (replicaCache, replicaDocuments) = VariantCache(); + var schemaId = sourceCache.Registry.GetRequired().SchemaId; + var source = new CultNetDatabase(sourceCache, new CultNetDatabaseOptions + { + DocumentRegistry = sourceDocuments, + Shards = [new CultNetShardDescriptor("replica-shard", "runtime-a", epoch: 3, isPrimary: true, schemaIds: [schemaId])] + }); + var replica = Replica(replicaCache, replicaDocuments, schemaId); + var baseKey = new CultRecordKey("publish-variant-base"); + var variantKey = new CultRecordKey("publish-variant-big"); + await replicaCache.UpsertAsync(new VariantWireFixture { Name = "base", Power = 1 }, new CultRecordHandle(baseKey)); + await replicaCache.UpsertVariantAsync(variantKey, baseKey, new[] + { + replicaCache.Override(nameof(VariantWireFixture.Name), "big") + }); + await source.PutAsync(baseKey, new VariantWireFixture { Name = "base edited", Power = 2 }); + + await replica.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + ShardId = "replica-shard", + ShardEpoch = 3, + Entries = source.GetMutationLogMessages("replica-shard").ToArray() + }); + + Assert.That(replicaCache.Get(variantKey)!.Power, Is.EqualTo(2), "the dependent re-resolved"); + Assert.That(replica.GetMutationLog("replica-shard").Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L }), + "the replica log holds the primary's entry and mints none of its own"); + } + + [Test] + public async Task ALocalCacheWriteOnAReplicaIsPublishedOnceAndNeverLogged() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var replica = Replica(cache, null, schemaId); + var published = Record(replica, KeyOfChange); + + await cache.UpsertAsync(Note("local"), new CultRecordHandle(new CultRecordKey("tests:publish:local"))); + + Assert.That(published, Is.EqualTo(new[] { "tests:publish:local" })); + Assert.That(replica.GetMutationLog("replica-shard"), Is.Empty); + } + + [Test] + public void ConcurrentWritersLoseNoLogEntryAndReuseNoSequence() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + const int writers = 4, perWriter = 5000; + + Parallel.For(0, writers, writer => + { + for (var i = 0; i < perWriter; i++) + cache.UpsertAsync(Note("n"), new CultRecordHandle(new CultRecordKey($"tests:publish:c:{writer}:{i}"))) + .GetAwaiter().GetResult(); + }); + + var sequences = database.GetMutationLog(database.Shards[0].ShardId).Select(entry => entry.Sequence).ToArray(); + Assert.That(sequences, Has.Length.EqualTo(writers * perWriter)); + Assert.That(sequences.Distinct().Count(), Is.EqualTo(writers * perWriter)); + } + + [Test] + public async Task AWriteAfterADatabaseDeleteIsStillPublishedAndLogged() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + var key = new CultRecordKey("tests:publish:after-delete"); + await database.PutAsync(key, Note("one")); + await database.DeleteAsync(key); + var published = Record(database, KeyOfChange); + + await cache.UpsertAsync(Note("two"), new CultRecordHandle(key)); + + Assert.That(published, Is.EqualTo(new[] { key.Value })); + Assert.That(database.GetMutationLog(database.Shards[0].ShardId).Select(entry => entry.Kind), + Is.EqualTo(new[] + { + CultNetDatabaseChangeKind.Added, CultNetDatabaseChangeKind.Removed, CultNetDatabaseChangeKind.Added + })); + } + + [Test] + public async Task AShardSnapshotResyncPublishesItsRemovalsAndPutsOnceAndLogsNothing() + { + var cache = new CultCache(); + var schemaId = cache.Registry.GetRequired().SchemaId; + var replica = Replica(cache, null, schemaId); + var stale = new CultRecordKey("tests:publish:stale"); + var fresh = new CultRecordKey("tests:publish:fresh"); + await cache.UpsertAsync(Note("stale"), new CultRecordHandle(stale)); + var published = Record(replica, change => ((dynamic)change).Kind.ToString() + ":" + ((dynamic)change).Key.Value); + var put = replica.Documents.CreateRawDocumentPutMessage( + "snapshot-fresh", new CultRecordHandle(fresh), Note("fresh")); + + await replica.ApplyShardSnapshotResponseAsync(replica.Shards[0], new CultNetSnapshotResponseRawMessage + { + ShardId = "replica-shard", + ShardEpoch = 3, + Documents = [put.Document!], + ShardLogSequence = 5 + }); + + Assert.That(published, Is.EqualTo(new[] { "Removed:" + stale.Value, "Added:" + fresh.Value })); + Assert.That(replica.GetMutationLog("replica-shard"), Is.Empty); + } + + [Test] + public async Task ADisposedDatabaseNoLongerObservesTheCache() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + var published = Record(database, KeyOfChange); + database.Dispose(); + + await cache.UpsertAsync(Note("after"), new CultRecordHandle(new CultRecordKey("tests:publish:disposed"))); + + Assert.That(published, Is.Empty); + } } }