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)

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);
+                    }
+                }
             }
         }
 

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<Exception> 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<CultCache>? seedSource = null, Action<CultCache>? seedReplica = null)
+            public Replication(Action<CultCache>? seedSource = null, Action<CultCache>? seedReplica = null,
+                Func<CultNetDatabaseSubscribeMessage, ICultNetSchemaServerPeer, string, string, bool>? 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<ReplicaItem>(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<ReplicaItem>(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<string>();
+            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<Action<CultNetSnapshotResponseRawMessage>> _snapshots = new();
+            private readonly List<Action<CultNetDatabaseChangeRawMessage>> _changes = new();
+
+            public SocketClient(int port, string key)
+            {
+                var listener = new EventBasedNetListener();
+                var connected = new TaskCompletionSource<bool>(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>(T message) where T : ICultNetSchemaMessage =>
+                _peer.Send(CultNetSchemaMessageSerialization.Serialize(message), DeliveryMethod.ReliableOrdered);
+            public void OnCultNet<T>(Action<T> callback) where T : ICultNetSchemaMessage
+            {
+                if (callback is Action<CultNetSnapshotResponseRawMessage> snapshot) _snapshots.Add(snapshot);
+                else if (callback is Action<CultNetDatabaseChangeRawMessage> 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<ReplicaItem>(replica.Registry));
+            using var client = new CultNetDatabaseSubscriptionClient(
+                new SocketClient(server.Port, ServerSecurityOptions.Development().ConnectionKey), replica, documents);
+            var faults = new List<Exception>();
+            client.Faulted += faults.Add;
+            await client.SubscribeAsync("items", schemaIds: new[] { source.Registry.GetRequired<ReplicaItem>().SchemaId });
+            Assert.That(replica.Get<ReplicaItem>(new CultRecordKey("a"))!.Sku, Is.EqualTo("one"));
+
+            Stage(source, ("a", "two"), ("b", "one"));
+            await Until(() => replica.Get<ReplicaItem>(new CultRecordKey("a"))!.Sku == "two");
+
+            Assert.That(replica.Get<ReplicaItem>(new CultRecordKey("b"))!.Sku, Is.EqualTo("one"));
+            Assert.That(replica.GetByIndex<ReplicaItem>("sku", "one"), Is.SameAs(replica.Get<ReplicaItem>(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)