CultNet client: the pending-commit buffer is bounded by the changes the incomplete commits announce, not by a per-commit cap; stragglers are not buffered

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs

@@ -31,10 +31,12 @@ namespace GameCult.Networking
             new(StringComparer.Ordinal);
         // Changes of a commit that has not fully arrived, by subscription and commit id.
         // Guarded by its own lock: the queue fills it, Unsubscribe and Dispose evict. A commit that does not complete is bounded
-        // by MaxPendingCommits and MaxCommitCount, and a violation faults the client.
+        // by MaxPendingCommits and by MaxBufferedChanges, and a violation faults the client.
         private const int MaxPendingCommits = 64;
-        private const int MaxCommitCount = 4096;
-        private readonly Dictionary<(string SubscriptionId, string CommitId), List<CultNetDatabaseChangeRawMessage>> _commits = new();
+        // Policy: the most changes the incomplete commits may announce between them, the bound on what the buffer can hold. It
+        // sits far above any commit a source makes; a commit that announces more than this could not be held at all.
+        private const int MaxBufferedChanges = 1_000_000;
+        private readonly Dictionary<(string SubscriptionId, string CommitId), PendingCommit> _commits = new();
         private readonly object _queueLock = new();
         private Task _queue = Task.CompletedTask;
         private bool _disposed;
@@ -243,22 +245,25 @@ 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);
-                List<CultNetDatabaseChangeRawMessage>? changes;
+                PendingCommit? pending;
                 lock (_commits)
                 {
                     if (Fault != null)
                         return Task.CompletedTask;
                     var count = message.CommitCount!.Value;
-                    var refusal = count > MaxCommitCount
-                        ? $"commit '{message.CommitId}' announces {count} changes, more than the {MaxCommitCount} a replica holds"
-                        : !_commits.TryGetValue(held, out changes)
+                    var refusal = !_commits.TryGetValue(held, out pending)
                             ? (_commits.Count >= MaxPendingCommits
                                 ? $"more than {MaxPendingCommits} commits are incomplete at once"
-                                : null)
-                            : changes[0].CommitCount != count
-                                ? $"commit '{message.CommitId}' announces {changes[0].CommitCount} and {count} changes"
-                                : changes.Any(other => other.MessageId == message.MessageId)
+                                : (long)count + _commits.Sum(entry => (long)entry.Value.Announced) > MaxBufferedChanges
+                                    ? $"commit '{message.CommitId}' announces {count} changes; the incomplete commits may announce {MaxBufferedChanges} between them"
+                                    : null)
+                            : pending.Announced != count
+                                ? $"commit '{message.CommitId}' announces {pending.Announced} and {count} changes"
+                                : pending.Ids.Contains(message.MessageId)
                                     ? $"commit '{message.CommitId}' delivered message '{message.MessageId}' twice"
                                     : null;
                     if (refusal != null)
@@ -267,18 +272,27 @@ namespace GameCult.Networking
                         return Task.CompletedTask;
                     }
 
-                    if (!_commits.TryGetValue(held, out changes))
-                        _commits[held] = changes = new List<CultNetDatabaseChangeRawMessage>();
-                    changes.Add(message);
-                    if (changes.Count < count)
+                    if (!_commits.TryGetValue(held, out pending))
+                        _commits[held] = pending = new PendingCommit(count);
+                    pending.Ids.Add(message.MessageId);
+                    pending.Messages.Add(message);
+                    if (pending.Messages.Count < count)
                         return Task.CompletedTask;
                     _commits.Remove(held);
                 }
 
-                return Apply(deliveryMode, changes);
+                return Apply(deliveryMode, pending.Messages);
             });
         }
 
+        private sealed class PendingCommit
+        {
+            public PendingCommit(int announced) => Announced = announced;
+            public int Announced { get; }
+            public List<CultNetDatabaseChangeRawMessage> Messages { get; } = new();
+            public HashSet<string> Ids { get; } = new(StringComparer.Ordinal);
+        }
+
         // The replica no longer matches its source: say so once, drop what was waiting, and apply nothing more.
         private void Break(Exception error)
         {

tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs

@@ -3,6 +3,7 @@ using System;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
+using System.Threading;
 using System.Threading.Tasks;
 using GameCult.Caching;
 using R3;
@@ -139,9 +140,9 @@ namespace GameCult.Networking.Tests
                     batch.Upsert(typeof(ReplicaItem), Item(sku), new CultRecordKey(key));
             }), Is.True);
 
-        private static async Task Until(Func<bool> condition)
+        private static async Task Until(Func<bool> condition, int seconds = 5)
         {
-            var deadline = DateTime.UtcNow.AddSeconds(5);
+            var deadline = DateTime.UtcNow.AddSeconds(seconds);
             while (!condition())
             {
                 if (DateTime.UtcNow > deadline) Assert.Fail("The replica did not reach the expected state.");
@@ -345,6 +346,7 @@ namespace GameCult.Networking.Tests
         public async Task TwoSubscriptionsMayUseOneCommitIdWithoutJoiningTheirCommits()
         {
             using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two")));
+            run.Client.SubscribeAsync("other", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult();
             var emit = run.Peer.Client!;
 
             emit.Emit(Put(run, "items", "a", "two", "9", 2));
@@ -430,7 +432,7 @@ namespace GameCult.Networking.Tests
         {
             using var run = new Replication();
 
-            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 1_000_000));
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 1_000_001));
             await Until(() => run.Client.Fault != null);
         }
 
@@ -439,9 +441,15 @@ namespace GameCult.Networking.Tests
         {
             using var run = new Replication();
 
-            for (var commit = 0; commit < 70; commit++)
+            for (var commit = 0; commit < 64; commit++)
                 run.Peer.Client!.Emit(Put(run, "items", "k" + commit, "v" + commit, "c" + commit, 2));
+            await Task.Delay(200);
+            Assert.That(run.Client.Fault, Is.Null, "64 incomplete commits are within the bound");
+            Assert.That(Pending(run.Client), Is.EqualTo(64));
+
+            run.Peer.Client!.Emit(Put(run, "items", "k64", "v64", "c64", 2));
             await Until(() => run.Client.Fault != null);
+            Assert.That(Pending(run.Client), Is.Zero, "a fault drops what was waiting");
         }
 
         [Test]
@@ -470,5 +478,144 @@ namespace GameCult.Networking.Tests
             var refusal = Assert.Throws<InvalidOperationException>(() => run.Client.SubscribeAsync("again"));
             Assert.That(refusal!.InnerException, Is.SameAs(run.Client.Fault));
         }
+
+        private static int Pending(CultNetDatabaseSubscriptionClient client)
+        {
+            var commits = (System.Collections.IDictionary)typeof(CultNetDatabaseSubscriptionClient)
+                .GetField("_commits", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!.GetValue(client)!;
+            lock (commits)
+                return commits.Count;
+        }
+
+        [Test]
+        public async Task TheIncompleteCommitsMayAnnounceOneMillionChangesBetweenThem()
+        {
+            using var run = new Replication();
+
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "1", 500_000));
+            run.Peer.Client!.Emit(Put(run, "items", "b", "two", "2", 500_000));
+            await Task.Delay(200);
+            Assert.That(run.Client.Fault, Is.Null, "500k and 500k fit");
+
+            run.Peer.Client!.Emit(Put(run, "items", "c", "three", "3", 2));
+            await Until(() => run.Client.Fault != null);
+        }
+
+        [Test]
+        public async Task ACommitOfOneMillionChangesIsHeldAndOneMillionAndOneIsNot()
+        {
+            using var held = new Replication();
+            held.Peer.Client!.Emit(Put(held, "items", "a", "one", "1", 1_000_000));
+            await Task.Delay(200);
+            Assert.That(held.Client.Fault, Is.Null);
+            Assert.That(Pending(held.Client), Is.EqualTo(1));
+
+            using var over = new Replication();
+            over.Peer.Client!.Emit(Put(over, "items", "a", "one", "1", 1_000_001));
+            await Until(() => over.Client.Fault != null);
+        }
+
+        [Test]
+        public async Task ACommitThatShrinksItsAnnouncedSizeFaults()
+        {
+            using var run = new Replication();
+
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 3));
+            run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 2));
+            await Until(() => run.Client.Fault != null);
+        }
+
+        [Test]
+        public async Task UnsubscribingOneSubscriptionKeepsTheCommitsOfAnother()
+        {
+            using var run = new Replication();
+            run.Client.SubscribeAsync("items2", recordKeys: new[] { "nothing-matches" }).GetAwaiter().GetResult();
+            var emit = run.Peer.Client!;
+
+            emit.Emit(Put(run, "items", "a", "one", "4", 2));
+            emit.Emit(Put(run, "items2", "b", "two", "4", 2));
+            await Task.Delay(100);
+            Assert.That(Pending(run.Client), Is.EqualTo(2));
+
+            run.Client.Unsubscribe("items");
+            Assert.That(Pending(run.Client), Is.EqualTo(1), "only the unsubscribed subscription's commits leave");
+            emit.Emit(Put(run, "items2", "c", "three", "4", 2));
+            await Until(() => run.ReplicaItem("b") != null);
+            Assert.That(run.ReplicaItem("c"), Is.Not.Null);
+            Assert.That(run.ReplicaItem("a"), Is.Null);
+        }
+
+        [Test]
+        public async Task AStragglerAfterUnsubscribeOrFromAnUnknownSubscriptionIsNotBuffered()
+        {
+            using var run = new Replication();
+            run.Client.Unsubscribe("items");
+
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2));
+            run.Peer.Client!.Emit(Put(run, "never-subscribed", "b", "two", "4", 2));
+            await Task.Delay(150);
+
+            Assert.That(Pending(run.Client), Is.Zero);
+            Assert.That(run.Client.Fault, Is.Null);
+        }
+
+        [Test]
+        public void DisposingTheClientDropsTheCommitsStillWaiting()
+        {
+            var run = new Replication();
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "4", 2));
+            SpinWait.SpinUntil(() => Pending(run.Client) == 1, 2000);
+            Assert.That(Pending(run.Client), Is.EqualTo(1));
+
+            run.Dispose();
+
+            Assert.That(Pending(run.Client), Is.Zero);
+        }
+
+        [Test]
+        public async Task AFaultedClientBuffersNothingMore()
+        {
+            using var run = new Replication();
+            run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 3, id: "same"));
+            run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 3, id: "same"));
+            await Until(() => run.Client.Fault != null);
+
+            run.Peer.Client!.Emit(Put(run, "items", "c", "three", "8", 2));
+            run.Peer.Client!.Emit(Put(run, "items", "d", "four", "9", 2));
+            await Task.Delay(150);
+
+            Assert.That(Pending(run.Client), Is.Zero);
+            Assert.That(run.Faults, Has.Count.EqualTo(1), "a fault is reported once");
+        }
+
+        [Test]
+        public void AFaultIsReportedOnceHoweverOftenItIsRaised()
+        {
+            using var run = new Replication();
+            var raise = typeof(CultNetDatabaseSubscriptionClient)
+                .GetMethod("Break", System.Reflection.BindingFlags.Instance | System.Reflection.BindingFlags.NonPublic)!;
+
+            raise.Invoke(run.Client, new object[] { new InvalidOperationException("first") });
+            raise.Invoke(run.Client, new object[] { new InvalidOperationException("second") });
+
+            Assert.That(run.Faults, Has.Count.EqualTo(1));
+            Assert.That(run.Client.Fault!.Message, Is.EqualTo("first"));
+        }
+
+        [TestCase(5000)]
+        [TestCase(100_000)]
+        public async Task ACommitOfManyChangesThroughTheRealSubscriptionServerAppliesIntact(int size)
+        {
+            using var run = new Replication();
+            var keys = Enumerable.Range(0, size).Select(i => ("k" + i, "s" + i)).ToArray();
+
+            Stage(run.Source, keys);
+            await Until(() => run.ReplicaItem("k" + (size - 1)) != null || run.Client.Fault != null, seconds: 180);
+
+            Assert.That(run.Client.Fault, Is.Null);
+            Assert.That(run.Replica.GetAll<ReplicaItem>().Count(), Is.EqualTo(size));
+            Assert.That(run.Peer.Changes.Select(change => change.CommitCount).Distinct().Single(), Is.EqualTo(size));
+            Assert.That(run.ReplicaItem("k" + (size / 2))!.Sku, Is.EqualTo("s" + (size / 2)));
+        }
     }
 }