From dbe3171d1ff9eaed582f23ef4d3a3c91d133215c Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 04:06:45 +0000 Subject: [PATCH 1/3] CultNet client: a straggler's subscription is judged under the lock Unsubscribe evicts under; tests pin the shared budget, ghost ids, a real database server, and Reconcile after a policy change Co-Authored-By: Claude Sonnet 5.5 --- .../CultNetDatabaseSubscriptionClient.cs | 7 +- .../CultNetReplicaCommitTests.cs | 173 +++++++++++++++++- 2 files changed, 174 insertions(+), 6 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/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs index 0e2f2031..a883eeac 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,16 @@ namespace GameCult.Networking.Tests public List Faults { get; } = new(); public CultNetDatabase Database { get; } private readonly CultNetDatabaseSubscriptionServer _server; + public void Reconcile() => _server.Reconcile(); - 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 +605,172 @@ 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), lets a straggler's handler arrive at 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(); + + lock (CommitsOf(run.Client)) + { + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2)); + Thread.Sleep(250); + run.Client.Unsubscribe("items"); + } + Thread.Sleep(250); + + 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, "ghost", "a", "one", "4", 2)); + run.Peer.Client!.Emit(Put(run, "ghost", "c", "three", "5", 3)); + 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); + } + + // A public Reconcile() diffs the projection against the source outside any commit, so its messages carry no commit. That is + // sound because a projection only ever differs from the source by records the policy hid or showed, never by a changed + // value (a changed record is re-judged the moment its commit arrives): both sides of the diff are subsets of one source + // state, whose unique values are distinct, so no order of its removals, updates and additions can meet a held value. + [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); + Assert.That(run.Peer.Changes.Last().CommitId, Is.Null, "the reconcile diff is not a commit"); + } + + // 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) -- 2.47.3 From 8d1319f65566d1c2abc96a2abb3a7947111ffdea Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 04:06:58 +0000 Subject: [PATCH 2/3] CultNet tests: the ghost id shares a prefix with a live subscription Co-Authored-By: Claude Sonnet 5.5 --- tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs index a883eeac..4d872341 100644 --- a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -666,8 +666,8 @@ namespace GameCult.Networking.Tests run.Peer.Client!.Emit(Put(run, "items2", "b", "two", "4", 2)); await Until(() => Pending(run.Client) == 1); - run.Peer.Client!.Emit(Put(run, "ghost", "a", "one", "4", 2)); - run.Peer.Client!.Emit(Put(run, "ghost", "c", "three", "5", 3)); + run.Peer.Client!.Emit(Put(run, "items-ghost", "a", "one", "4", 2)); + run.Peer.Client!.Emit(Put(run, "items-ghost", "c", "three", "5", 3)); await Task.Delay(200); Assert.That(Pending(run.Client), Is.EqualTo(1), "only the live subscription's commit is held"); -- 2.47.3 From e3c31b0882c134303377e88f96e1ec4db37817a8 Mon Sep 17 00:00:00 2001 From: Libby Date: Wed, 11 Nov 2026 04:39:46 +0000 Subject: [PATCH 3/3] CultNet: a public Reconcile sends its diff as one commit of its own; the race with a commit is reproduced, the straggler test no longer sleeps Co-Authored-By: Claude Sonnet 5.5 --- .../CultNetDatabaseSubscriptionServer.cs | 15 ++++++- .../CultNetReplicaCommitTests.cs | 44 +++++++++++++++---- 2 files changed, 49 insertions(+), 10 deletions(-) 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 4d872341..6059d7ba 100644 --- a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -107,6 +107,8 @@ namespace GameCult.Networking.Tests 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, Func? authorizeRecord = null) @@ -610,20 +612,24 @@ namespace GameCult.Networking.Tests .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), lets a straggler's handler arrive at it, ends the subscription, and releases: a handler that had already - // passed its check by then would buffer a commit nobody will ever finish. + // 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)); - Thread.Sleep(250); + 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)!; } - Thread.Sleep(250); + tail.GetAwaiter().GetResult(); Assert.That(Pending(run.Client), Is.Zero); Assert.That(run.Client.Fault, Is.Null); @@ -668,6 +674,7 @@ namespace GameCult.Networking.Tests 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"); @@ -677,10 +684,30 @@ namespace GameCult.Networking.Tests Assert.That(run.ReplicaItem("a"), Is.Null); } - // A public Reconcile() diffs the projection against the source outside any commit, so its messages carry no commit. That is - // sound because a projection only ever differs from the source by records the policy hid or showed, never by a changed - // value (a changed record is re-judged the moment its commit arrives): both sides of the diff are subsets of one source - // state, whose unique values are distinct, so no order of its removals, updates and additions can meet a held value. + // 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() { @@ -699,7 +726,6 @@ namespace GameCult.Networking.Tests Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); Assert.That(run.Client.Fault, Is.Null); - Assert.That(run.Peer.Changes.Last().CommitId, Is.Null, "the reconcile diff is not a commit"); } // Source, wire and client all real: a CultNetDatabaseServer behind a Server on a loopback socket, a LiteNetLib client -- 2.47.3