diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 2206322c..971a53c2 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/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> _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? 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(); - 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 Messages { get; } = new(); + public HashSet 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) { diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs index 38faa79d..0e2f2031 100644 --- a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs +++ b/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 condition) + private static async Task Until(Func 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(() => 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().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))); + } } }