From 28248ebedee098188d11a519ee2b55296f6a14e1 Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Tue, 29 Sep 2026 14:06:05 +0200 Subject: [PATCH] Adopt Soul pass 2 probes and pin log-before-publish and in-place re-upsert Co-Authored-By: Claude Sonnet 5 --- .../CultNetPublicationOrderTests.cs | 347 ++++++++++++++++++ 1 file changed, 347 insertions(+) create mode 100644 tests/GameCult.Networking.Tests/CultNetPublicationOrderTests.cs diff --git a/tests/GameCult.Networking.Tests/CultNetPublicationOrderTests.cs b/tests/GameCult.Networking.Tests/CultNetPublicationOrderTests.cs new file mode 100644 index 00000000..65e290a4 --- /dev/null +++ b/tests/GameCult.Networking.Tests/CultNetPublicationOrderTests.cs @@ -0,0 +1,347 @@ +#nullable enable +using System; +using System.Collections.Concurrent; +using System.Collections.Generic; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; +using GameCult.Caching; +using NUnit.Framework; +using R3; +using static GameCult.Networking.Tests.NetworkingTests; + +namespace GameCult.Networking.Tests +{ + public sealed class CultNetPublicationOrderTests + { + private static readonly TimeSpan Wait = TimeSpan.FromSeconds(5); + private static NetworkSchemaNote Note(string text) => new() { Schema = "tests.networking_note.v1", Text = text }; + + private static string? TextOf(object? document) => (document as NetworkSchemaNote)?.Text; + + // A1: two writers to one key. A is admitted first, B second; the cache ends at B. A's publication is delayed + // (the window between leaving the cache gate and reaching the handler), B runs to completion meanwhile. + [Test] + public async Task S1_TheLogOrdersOneKeysWritesInAdmissionOrder() + { + var cache = new CultCache(); + var entered = new ManualResetEventSlim(); + var release = new ManualResetEventSlim(); + using var widen = cache.Watch().Subscribe(change => + { + if (change.Document?.Text == "A") + { + entered.Set(); + release.Wait(Wait); + } + }); + var database = new CultNetDatabase(cache); + var seen = new List(); + using var subscriber = database.Watch().Subscribe(change => { lock (seen) seen.Add(change.Document?.Text); }); + var key = new CultRecordKey("soul:order"); + + var a = Task.Run(() => cache.UpsertAsync(Note("A"), new CultRecordHandle(key)).GetAwaiter().GetResult()); + Assert.That(entered.Wait(Wait), Is.True); + await database.PutAsync(key, Note("B")); + release.Set(); + await a; + + var shard = database.Shards[0].ShardId; + var last = database.GetMutationLog(shard).Where(entry => entry.Key.Value == key.Value).OrderBy(entry => entry.Sequence).Last(); + + // A replica built from the log. + var replicaCache = new CultCache(); + var replica = new CultNetDatabase(replicaCache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(shard, "other", epoch: database.Shards[0].Epoch, isPrimary: false)] + }); + await replica.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + ShardId = shard, + ShardEpoch = database.Shards[0].Epoch, + Entries = database.GetMutationLogMessages(shard).ToArray() + }); + + Assert.Multiple(() => + { + Assert.That(cache.Get(key)!.Text, Is.EqualTo("B"), "the cache admitted A then B"); + Assert.That(TextOf(last.Document), Is.EqualTo("B"), "the log's last write for the key must be the cache's last write"); + Assert.That(replicaCache.Get(key)?.Text, Is.EqualTo("B"), "a replica built from the log must match the primary"); + Assert.That(seen.Last(), Is.EqualTo("B"), "a subscriber's final state for the key"); + }); + } + + // Same property, no instrumentation: two writers hammer one key, then the log's last entry is compared with the cache. + [Test] + public void S1b_UninstrumentedOneKeyRaceLeavesLogAndCacheAgreeing() + { + var divergent = 0; + const int rounds = 300; + for (var round = 0; round < rounds; round++) + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + var key = new CultRecordKey("soul:hammer"); + using var start = new Barrier(2); + Parallel.For(0, 2, writer => + { + start.SignalAndWait(); + for (var i = 0; i < 40; i++) + cache.UpsertAsync(Note($"{writer}:{i}"), new CultRecordHandle(key)).GetAwaiter().GetResult(); + }); + var last = database.GetMutationLog(database.Shards[0].ShardId).OrderBy(entry => entry.Sequence).Last(); + if (!ReferenceEquals(last.Document, cache.Get(key))) + divergent++; + } + + TestContext.Progress.WriteLine($"SOUL S1b divergent rounds: {divergent}/{rounds}"); + Assert.That(divergent, Is.EqualTo(0), $"{divergent}/{rounds} rounds ended with the log's last write differing from the cache"); + } + + // A2: a door's context is keyed by instance. A plain removal of the instance the door just admitted, landing in + // the window before the door's own change is handled, consumes the door's context. + [Test] + public async Task S2_ARemovalNeverWearsAReconcilingPutsContext() + { + var cache = new CultCache(); + var entered = new ManualResetEventSlim(); + var release = new ManualResetEventSlim(); + using var widen = cache.Watch().Subscribe(change => + { + if (change.Document?.Text == "authoritative") + { + entered.Set(); + release.Wait(Wait); + } + }); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + RuntimeId = "local", + ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")] + }); + var key = new CultRecordKey("soul:reconcile"); + await database.PutPredictedAsync(key, Note("predicted")); + var published = new ConcurrentQueue<(CultNetDatabaseChangeKind Kind, string? Text)>(); + using var subscriber = database.Watch().Subscribe(change => published.Enqueue((change.Kind, change.Document?.Text))); + var logged = database.GetMutationLog(database.Shards[0].ShardId).Count; + var message = database.Documents.CreateRawDocumentPutMessage("m", new CultRecordHandle(key), Note("authoritative")); + message.ShardId = database.Shards[0].ShardId; + message.ShardEpoch = database.Shards[0].Epoch; + + var a = Task.Run(() => database.ApplyPutAsync(message)); + Assert.That(entered.Wait(Wait), Is.True); + cache.Remove(key); + release.Set(); + await a; + + var log = database.GetMutationLog(database.Shards[0].ShardId, logged).ToArray(); + Assert.Multiple(() => + { + Assert.That(published.Count(change => change.Kind == CultNetDatabaseChangeKind.Removed), Is.EqualTo(1), + "the removal is published as a removal: " + string.Join(", ", published)); + Assert.That(published.Where(change => change.Kind == CultNetDatabaseChangeKind.Reconciled).Select(change => change.Text), + Is.EqualTo(new[] { "authoritative" }), "the reconcile carries the authoritative document"); + Assert.That(log.Where(entry => entry.Kind != CultNetDatabaseChangeKind.Removed).All(entry => entry.Document != null), Is.True, + "no put entry without a document: " + string.Join(", ", log.Select(entry => entry.Kind + ":" + TextOf(entry.Document)))); + }); + } + + // A4: chained replication. B is a replica of P, C a replica of B. A replica-applied delete of a key B does not + // hold (B removed it locally, a supported write) must still keep B's log contiguous for C. + [Test] + public async Task S3_AChainedReplicaSeesNoGapWhenItsSourceAppliesADeleteOfAnAbsentKey() + { + const string shardId = "chain"; + var schemaId = new CultCache().Registry.GetRequired().SchemaId; + CultNetDatabase Make(CultCache cache, bool primary) => new(cache, new CultNetDatabaseOptions + { + Shards = [new CultNetShardDescriptor(shardId, "p", epoch: 2, isPrimary: primary, schemaIds: [schemaId])] + }); + var p = Make(new CultCache(), true); + var bCache = new CultCache(); + var b = Make(bCache, false); + var c = Make(new CultCache(), false); + async Task Pull(CultNetDatabase from, CultNetDatabase to) + { + await to.ApplyShardLogResponseAsync(new CultNetShardLogResponseMessage + { + ShardId = shardId, + ShardEpoch = 2, + Entries = from.GetMutationLogMessages(shardId, to.GetAppliedShardSequence(shardId)).ToArray() + }); + } + + var k = new CultRecordKey("soul:chain:k"); + await p.PutAsync(k, Note("x")); + await Pull(p, b); + await Pull(b, c); + bCache.Remove(k); + await p.DeleteAsync(k); + await p.PutAsync(new CultRecordKey("soul:chain:k2"), Note("y")); + await Pull(p, b); + + Assert.Multiple(() => + { + Assert.That(b.GetMutationLog(shardId).Select(entry => entry.Sequence), Is.EqualTo(new[] { 1L, 2L, 3L }), + "B's log holds every primary entry it applied"); + Assert.DoesNotThrowAsync(() => Pull(b, c), "C pulls B's log without a gap"); + }); + } + + // A5: a subscriber that answers a Reconciled change by predicting the same key again keeps its prediction. + [Test] + public async Task S4_APredictionMadeInReactionToAReconcileSurvives() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + RuntimeId = "local", + ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")] + }); + var key = new CultRecordKey("soul:repredict"); + var kinds = new ConcurrentQueue(); + var repredicted = false; + using var subscriber = database.Watch().Subscribe(change => + { + kinds.Enqueue(change.Kind); + if (change.Kind == CultNetDatabaseChangeKind.Reconciled && !repredicted) + { + repredicted = true; + database.PutPredictedAsync(key, Note("predicted again")).GetAwaiter().GetResult(); + } + }); + CultNetDocumentPutRawMessage Put(string text) + { + var message = database.Documents.CreateRawDocumentPutMessage(text, new CultRecordHandle(key), Note(text)); + message.ShardId = database.Shards[0].ShardId; + message.ShardEpoch = database.Shards[0].Epoch; + return message; + } + + await database.PutPredictedAsync(key, Note("predicted")); + await database.ApplyPutAsync(Put("auth1")); + await database.ApplyPutAsync(Put("auth2")); + + Assert.That(kinds, Is.EqualTo(new[] + { + CultNetDatabaseChangeKind.Predicted, + CultNetDatabaseChangeKind.Reconciled, + CultNetDatabaseChangeKind.Predicted, + CultNetDatabaseChangeKind.Reconciled + })); + } + // Target of AWriteAfterADatabaseDeleteIsStillPublishedAndLogged: a door context must not outlive its door. A + // refused prediction leaves nothing behind that a later write of the same instance would wear. + [Test] + public async Task S6_ARefusedDoorLeavesNoContextForALaterWriteOfTheSameInstance() + { + var registry = CultDocumentRegistry.Shared; + var cache = new CultCache(registry, GameCult.Caching.MessagePack.CultCacheMessagePack.CreateCodec(registry)); + var documents = new CultNetDocumentRegistry(registry) + .Register(CultNetDocumentBinding.ForDocument(registry)); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + DocumentRegistry = documents, + RuntimeId = "local", + ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")] + }); + var baseKey = new CultRecordKey("soul-refuse-base"); + var variantKey = new CultRecordKey("soul-refuse-variant"); + await cache.UpsertAsync(new CultNetDocumentRegistryVariantTests.VariantWireFixture { Name = "base", Power = 1 }, + new CultRecordHandle(baseKey)); + await cache.UpsertVariantAsync(variantKey, baseKey, new[] + { + cache.Override(nameof(CultNetDocumentRegistryVariantTests.VariantWireFixture.Name), "v") + }); + var instance = new CultNetDocumentRegistryVariantTests.VariantWireFixture { Name = "reused", Power = 3 }; + Assert.That(async () => await database.PutPredictedAsync(variantKey, instance), Throws.InvalidOperationException); + var kinds = new ConcurrentQueue(); + using var subscriber = database.Watch().Subscribe(change => kinds.Enqueue(change.Kind)); + var logged = database.GetMutationLog(database.Shards[0].ShardId).Count; + + await cache.UpsertAsync(instance, new CultRecordHandle(new CultRecordKey("soul-refuse-other"))); + + Assert.Multiple(() => + { + Assert.That(kinds, Is.EqualTo(new[] { CultNetDatabaseChangeKind.Added })); + Assert.That(database.GetMutationLog(database.Shards[0].ShardId, logged), Has.Count.EqualTo(1)); + }); + } + + // Target of ADisposedDatabaseNoLongerObservesTheCache: after Dispose the database must stop logging, which a + // durable log store can observe even though the disposed Subject hides publication. + [Test] + public async Task S7_ADisposedDatabaseAppendsNothingToItsDurableLog() + { + var root = System.IO.Path.Combine(System.IO.Path.GetTempPath(), "soul-log-" + Guid.NewGuid().ToString("N")); + var store = new CultNetFileShardMutationLogStore(root); + var cache = new CultCache(); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions { MutationLogStore = store }); + var shard = database.Shards[0].ShardId; + await cache.UpsertAsync(Note("before"), new CultRecordHandle(new CultRecordKey("soul:dispose:before"))); + Assert.That(store.Read(shard), Has.Count.EqualTo(1)); + database.Dispose(); + + try { await cache.UpsertAsync(Note("after"), new CultRecordHandle(new CultRecordKey("soul:dispose:after"))); } + catch { /* a throwing handler would also be a failure below */ } + + Assert.That(store.Read(shard), Has.Count.EqualTo(1)); + System.IO.Directory.Delete(root, true); + } + + // The log entry exists before its change is published: a subscriber that reads the log from its handler finds it. + [Test] + public async Task TheLogEntryIsWrittenBeforeTheChangeIsPublished() + { + var cache = new CultCache(); + var database = new CultNetDatabase(cache); + var key = new CultRecordKey("soul:log-first"); + var loggedWhenPublished = new ConcurrentQueue(); + using var subscriber = database.Watch().Subscribe(change => + loggedWhenPublished.Enqueue(database.GetMutationLog(database.Shards[0].ShardId).Any(entry => entry.Key.Value == key.Value))); + + await database.PutAsync(key, Note("one")); + cache.Remove(key); + + Assert.That(loggedWhenPublished, Is.EqualTo(new[] { true, true })); + } + + // An observer that mutates the admitted instance in place and upserts it again writes a second change of the same + // instance. It is published as its own update and never wears the context of the door's change, even when it + // reaches the database first. + [Test] + public async Task AnInPlaceReUpsertOfTheDoorsInstanceDoesNotWearTheDoorsContext() + { + var cache = new CultCache(); + using var normalizer = cache.Watch().Subscribe(change => + { + if (change.Document?.Text == "authoritative") + { + change.Document.Text = "normalized"; + cache.UpsertAsync(change.Document, new CultRecordHandle(change.Key)).GetAwaiter().GetResult(); + } + }); + var database = new CultNetDatabase(cache, new CultNetDatabaseOptions + { + RuntimeId = "local", + ClientAuthorityScopes = [new CultNetClientAuthorityScope("local")] + }); + var key = new CultRecordKey("soul:in-place"); + var seen = new ConcurrentQueue<(CultNetDatabaseChangeKind Kind, string? Previous)>(); + using var subscriber = database.Watch().Subscribe(change => seen.Enqueue((change.Kind, change.PreviousDocument?.Text))); + var message = database.Documents.CreateRawDocumentPutMessage("m", new CultRecordHandle(key), Note("authoritative")); + message.ShardId = database.Shards[0].ShardId; + message.ShardEpoch = database.Shards[0].Epoch; + + await database.PutPredictedAsync(key, Note("predicted")); + await database.ApplyPutAsync(message); + + Assert.That(seen, Is.EqualTo(new (CultNetDatabaseChangeKind, string?)[] + { + (CultNetDatabaseChangeKind.Predicted, null), + (CultNetDatabaseChangeKind.Reconciled, "predicted"), + (CultNetDatabaseChangeKind.Updated, "normalized") + })); + } + } +}