From abd1a92eb24e1af3fc7cdd1bf98a88dbdebc2938 Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 04:57:59 +0000 Subject: [PATCH] Replica commit hardening: Reconcile sends its diff as one tagged commit; straggler guard inside the lock; real-server and budget pins (#91) --- .../CultNetDatabaseSubscriptionClient.cs | 7 +- .../CultNetDatabaseSubscriptionServer.cs | 15 +- .../CultNetReplicaCommitTests.cs | 199 +++++++++++++++++- 3 files changed, 214 insertions(+), 7 deletions(-) diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 971a53c2..bf3772e6 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -245,14 +245,13 @@ namespace GameCult.Networking return Apply(deliveryMode, new[] { message }); // The commit's other changes follow on this connection; the replica applies them as one commit. - // A straggler of a subscription that was ended, or never made, is not buffered. - if (!_deliveryModes.ContainsKey(message.SubscriptionId)) - return Task.CompletedTask; var held = (message.SubscriptionId, message.CommitId); PendingCommit? pending; lock (_commits) { - if (Fault != null) + // A straggler of a subscription that was ended, or never made, is not buffered. Judged under the lock that + // Unsubscribe evicts under: a check before it could pass, lose the race to the eviction, and then buffer. + if (Fault != null || !_deliveryModes.ContainsKey(message.SubscriptionId)) return Task.CompletedTask; var count = message.CommitCount!.Value; var refusal = !_commits.TryGetValue(held, out pending) diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs index 1325f85e..3e23cf10 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionServer.cs @@ -77,7 +77,20 @@ namespace GameCult.Networking { if (_disposed) return; foreach (var entry in _requests.ToArray()) - Reconcile(entry.Key, entry.Value); + { + // Reconcile can run after the cache has admitted a commit and before Watch has seen it, so its diff can carry + // changed values; it is held and sent as one commit of its own, as Watch's is, for the replica to judge whole. + var outbox = _outbox = new CommitOutbox(Guid.NewGuid().ToString("N")); + try + { + Reconcile(entry.Key, entry.Value); + } + finally + { + _outbox = null; + outbox.Send(entry.Key.Peer); + } + } } } diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs index 0e2f2031..6059d7ba 100644 --- a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -6,6 +6,7 @@ using System.Linq; using System.Threading; using System.Threading.Tasks; using GameCult.Caching; +using LiteNetLib; using R3; using MessagePack; using NUnit.Framework; @@ -105,14 +106,18 @@ namespace GameCult.Networking.Tests public List Faults { get; } = new(); public CultNetDatabase Database { get; } private readonly CultNetDatabaseSubscriptionServer _server; + public void Reconcile() => _server.Reconcile(); + public object Gate => typeof(CultNetDatabaseSubscriptionServer) + .GetField("_lifecycleGate", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(_server)!; - public Replication(Action? seedSource = null, Action? seedReplica = null) + public Replication(Action? seedSource = null, Action? seedReplica = null, + Func? authorizeRecord = null) { seedSource?.Invoke(Source); seedReplica?.Invoke(Replica); Database = new CultNetDatabase(Source); var wire = new LoopbackServer(); - _server = new CultNetDatabaseSubscriptionServer(wire, Database); + _server = new CultNetDatabaseSubscriptionServer(wire, Database, authorizeRecord: authorizeRecord); var documents = new CultNetDocumentRegistry(Replica.Registry) .Register(CultNetDocumentBinding.ForDocument(Replica.Registry)); Client = new CultNetDatabaseSubscriptionClient(new Client(wire, Peer), Replica, documents); @@ -602,6 +607,196 @@ namespace GameCult.Networking.Tests Assert.That(run.Client.Fault!.Message, Is.EqualTo("first")); } + private static object CommitsOf(CultNetDatabaseSubscriptionClient client) => + typeof(CultNetDatabaseSubscriptionClient) + .GetField("_commits", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(client)!; + + // The check that a subscription is live belongs to the same lock Unsubscribe evicts under. The test holds that lock (it is + // re-entrant), waits until the straggler's handler is contending for it, ends the subscription, and releases: a handler + // that had already passed its check by then would buffer a commit nobody will ever finish. + [Test] + public void AStragglerThatArrivedBeforeUnsubscribeBuffersNothingAfterIt() + { + using var run = new Replication(); + Task tail; + + lock (CommitsOf(run.Client)) + { + var contended = Monitor.LockContentionCount; + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2)); + Assert.That(SpinWait.SpinUntil(() => Monitor.LockContentionCount > contended, 10_000), "the handler never reached the lock"); + run.Client.Unsubscribe("items"); + tail = (Task)typeof(CultNetDatabaseSubscriptionClient) + .GetField("_queue", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(run.Client)!; + } + tail.GetAwaiter().GetResult(); + + Assert.That(Pending(run.Client), Is.Zero); + Assert.That(run.Client.Fault, Is.Null); + } + + // The 1,000,000 budget is the client's, over every subscription's incomplete commits, not each subscription's own. + [Test] + public async Task TheBudgetIsSharedByAllSubscriptions() + { + using var run = new Replication(); + run.Client.SubscribeAsync("items2", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "1", 600_000)); + run.Peer.Client!.Emit(Put(run, "items2", "b", "two", "2", 400_000)); + await Task.Delay(200); + Assert.That(run.Client.Fault, Is.Null, "600k and 400k fit in one million"); + Assert.That(Pending(run.Client), Is.EqualTo(2)); + + run.Peer.Client!.Emit(Put(run, "items2", "c", "three", "3", 2)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task TwoSubscriptionsAnnouncingMoreThanTheBudgetBetweenThemFault() + { + using var run = new Replication(); + run.Client.SubscribeAsync("items2", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "1", 600_000)); + run.Peer.Client!.Emit(Put(run, "items2", "b", "two", "2", 600_000)); + await Until(() => run.Client.Fault != null); + Assert.That(Pending(run.Client), Is.Zero); + } + + [Test] + public async Task AStragglerFromAnUnknownIdIsDroppedWhileAnotherSubscriptionIsLive() + { + using var run = new Replication(); + run.Client.SubscribeAsync("items2", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult(); + run.Peer.Client!.Emit(Put(run, "items2", "b", "two", "4", 2)); + await Until(() => Pending(run.Client) == 1); + + run.Peer.Client!.Emit(Put(run, "items-ghost", "a", "one", "4", 2)); + run.Peer.Client!.Emit(Put(run, "items-ghost", "c", "three", "5", 3)); + run.Peer.Client!.Emit(Put(run, "item", "e", "five", "6", 2)); + await Task.Delay(200); + + Assert.That(Pending(run.Client), Is.EqualTo(1), "only the live subscription's commit is held"); + Assert.That(run.Client.Fault, Is.Null); + run.Peer.Client!.Emit(Put(run, "items2", "d", "four", "4", 2)); + await Until(() => run.ReplicaItem("b") != null); + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + // Reconcile() can run after the cache admitted a commit and before the subscription's Watch has seen it (the observer + // runs on the writer's thread after the cache gate opens). Its diff then carries changed values, and a swap sent record by + // record faults the replica. The test holds the subscription server's gate to stand in that window. + [Test] + public async Task AReconcileThatRacesACommitSendsItsSwapAsOneCommit() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + Task write; + + lock (run.Gate) + { + write = Task.Run(() => Stage(run.Source, ("a", "two"), ("b", "one"))); + Assert.That(SpinWait.SpinUntil(() => run.Source.Get(new CultRecordKey("a"))!.Sku == "two", 10_000), + "the source did not take the swap"); + run.Reconcile(); + } + await write; + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.Client.Fault, Is.Null); + Assert.That(run.Peer.Changes.Select(change => change.CommitCount), Is.All.EqualTo(2)); + } + + [Test] + public async Task ReconcilingAfterAPolicyChangeThatFollowsASwapLeavesTheReplicaIntact() + { + var hidden = new HashSet(); + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two")), + authorizeRecord: (_, _, key, _) => !hidden.Contains(key)); + hidden.Add("b"); + + Stage(run.Source, ("a", "two"), ("b", "one")); + await Until(() => run.ReplicaItem("b") == null && run.ReplicaItem("a")!.Sku == "two"); + + hidden.Remove("b"); + hidden.Add("a"); + run.Reconcile(); + await Until(() => run.ReplicaItem("b") != null && run.ReplicaItem("a") == null); + + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.Client.Fault, Is.Null); + } + + // Source, wire and client all real: a CultNetDatabaseServer behind a Server on a loopback socket, a LiteNetLib client + // standing in for the transport of a CultNetDatabaseSubscriptionClient. No handler is called directly. + private sealed class SocketClient : ICultNetSchemaClient + { + private readonly NetManager _manager; + private readonly NetPeer _peer; + private readonly List> _snapshots = new(); + private readonly List> _changes = new(); + + public SocketClient(int port, string key) + { + var listener = new EventBasedNetListener(); + var connected = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var decoder = new LiteNetLibTransportConnection(); + listener.PeerConnectedEvent += _ => connected.TrySetResult(true); + listener.NetworkReceiveEvent += (_, reader, _, _) => + { + switch (LiteNetLibTransportConnection.DecodeSchema(decoder.Receive(reader.GetRemainingBytes()))) + { + case CultNetSnapshotResponseRawMessage snapshot: foreach (var handler in _snapshots.ToArray()) handler(snapshot); break; + case CultNetDatabaseChangeRawMessage change: foreach (var handler in _changes.ToArray()) handler(change); break; + } + }; + _manager = new NetManager(listener) { UnsyncedEvents = true }; + _manager.Start(0); + _peer = _manager.Connect("127.0.0.1", port, key); + if (!connected.Task.Wait(TimeSpan.FromSeconds(10))) Assert.Fail("the client did not connect"); + } + + public bool Connected => _peer.ConnectionState == ConnectionState.Connected; + public void Connect(string host, int port) { } + public void SendCultNet(T message) where T : ICultNetSchemaMessage => + _peer.Send(CultNetSchemaMessageSerialization.Serialize(message), DeliveryMethod.ReliableOrdered); + public void OnCultNet(Action callback) where T : ICultNetSchemaMessage + { + if (callback is Action snapshot) _snapshots.Add(snapshot); + else if (callback is Action change) _changes.Add(change); + } + public void Dispose() => _manager.Stop(); + } + + [Test] + public async Task ASwapThroughTheRealDatabaseServerReachesTheClientIntact() + { + var source = new CultCache(); + Stage(source, ("a", "one"), ("b", "two")); + using var database = new CultNetDatabase(source); + using var server = new Server(source, ServerSecurityOptions.Development()) { Port = 0 }; + using var databaseServer = new CultNetDatabaseServer(server, database); + server.Start(); + var replica = new CultCache(); + var documents = new CultNetDocumentRegistry(replica.Registry) + .Register(CultNetDocumentBinding.ForDocument(replica.Registry)); + using var client = new CultNetDatabaseSubscriptionClient( + new SocketClient(server.Port, ServerSecurityOptions.Development().ConnectionKey), replica, documents); + var faults = new List(); + client.Faulted += faults.Add; + await client.SubscribeAsync("items", schemaIds: new[] { source.Registry.GetRequired().SchemaId }); + Assert.That(replica.Get(new CultRecordKey("a"))!.Sku, Is.EqualTo("one")); + + Stage(source, ("a", "two"), ("b", "one")); + await Until(() => replica.Get(new CultRecordKey("a"))!.Sku == "two"); + + Assert.That(replica.Get(new CultRecordKey("b"))!.Sku, Is.EqualTo("one")); + Assert.That(replica.GetByIndex("sku", "one"), Is.SameAs(replica.Get(new CultRecordKey("b")))); + Assert.That(client.Fault, Is.Null); + Assert.That(faults, Is.Empty); + } + [TestCase(5000)] [TestCase(100_000)] public async Task ACommitOfManyChangesThroughTheRealSubscriptionServerAppliesIntact(int size)