diff --git a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs index 7df7349a..2206322c 100644 --- a/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs +++ b/src/GameCult.Networking/CultNetDatabaseSubscriptionClient.cs @@ -29,7 +29,11 @@ namespace GameCult.Networking private readonly ConcurrentDictionary _initialSnapshots = new(); private readonly ConcurrentDictionary _deliveryModes = new(StringComparer.Ordinal); - // Changes of a commit that has not fully arrived, by subscription and commit id. Touched only inside the queue. + // 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. + private const int MaxPendingCommits = 64; + private const int MaxCommitCount = 4096; private readonly Dictionary<(string SubscriptionId, string CommitId), List> _commits = new(); private readonly object _queueLock = new(); private Task _queue = Task.CompletedTask; @@ -120,6 +124,8 @@ namespace GameCult.Networking CultNetDatabaseSubscriptionDeliveryMode deliveryMode = CultNetDatabaseSubscriptionDeliveryMode.ReplicateToCache) { if (_disposed) throw new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient)); + if (Fault != null) + throw new InvalidOperationException("The replica refused a commit and differs from its source; subscribe with a new client.", Fault); if (!_client.Connected) throw new InvalidOperationException("CultNet schema client must be connected before subscribing."); if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); @@ -177,6 +183,9 @@ namespace GameCult.Networking if (_disposed) return; if (string.IsNullOrWhiteSpace(subscriptionId)) throw new ArgumentException("Subscription id is required.", nameof(subscriptionId)); _deliveryModes.TryRemove(subscriptionId, out _); + lock (_commits) + foreach (var held in _commits.Keys.Where(key => key.SubscriptionId == subscriptionId).ToArray()) + _commits.Remove(held); _client.SendCultNet(new CultNetDatabaseUnsubscribeMessage { MessageId = Guid.NewGuid().ToString("N"), @@ -192,6 +201,8 @@ namespace GameCult.Networking pending.Completion.TrySetException(new ObjectDisposedException(nameof(CultNetDatabaseSubscriptionClient))); _initialSnapshots.Clear(); _deliveryModes.Clear(); + lock (_commits) + _commits.Clear(); _client.Dispose(); } @@ -233,16 +244,55 @@ namespace GameCult.Networking // The commit's other changes follow on this connection; the replica applies them as one commit. var held = (message.SubscriptionId, message.CommitId); - if (!_commits.TryGetValue(held, out var changes)) - _commits[held] = changes = new List(); - changes.Add(message); - if (changes.Count < message.CommitCount) - return Task.CompletedTask; - _commits.Remove(held); + List? changes; + 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) + ? (_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) + ? $"commit '{message.CommitId}' delivered message '{message.MessageId}' twice" + : null; + if (refusal != null) + { + Break(new InvalidOperationException($"CultNet subscription '{message.SubscriptionId}': {refusal}.")); + return Task.CompletedTask; + } + + if (!_commits.TryGetValue(held, out changes)) + _commits[held] = changes = new List(); + changes.Add(message); + if (changes.Count < count) + return Task.CompletedTask; + _commits.Remove(held); + } + return Apply(deliveryMode, changes); }); } + // The replica no longer matches its source: say so once, drop what was waiting, and apply nothing more. + private void Break(Exception error) + { + lock (_commits) + { + if (Fault != null) + return; + Fault = error; + _commits.Clear(); + } + + Faulted?.Invoke(error); + } + private Task Apply(CultNetDatabaseSubscriptionDeliveryMode deliveryMode, IReadOnlyList changes) { // A replica that refused a commit no longer matches its source. Applying later commits on top would hide that, so @@ -258,8 +308,7 @@ namespace GameCult.Networking } catch (Exception error) { - Fault = error; - Faulted?.Invoke(error); + Break(error); return Task.CompletedTask; } diff --git a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs index b3d6b962..854742a0 100644 --- a/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs +++ b/tests/GameCult.Networking.Tests/CultNetReplicaCommitTests.cs @@ -63,6 +63,8 @@ namespace GameCult.Networking.Tests { if (message is CultNetDatabaseSubscribeMessage subscribe) _server.Dispatch(subscribe, _peer).GetAwaiter().GetResult(); + else if (message is CultNetDatabaseUnsubscribeMessage unsubscribe) + _server.Dispatch(unsubscribe, _peer).GetAwaiter().GetResult(); } public void OnCultNet(Action callback) where T : ICultNetSchemaMessage { @@ -313,5 +315,158 @@ namespace GameCult.Networking.Tests Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("two")); Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); } + + // Hand-made wire messages: what a peer sends that the real server would not. + private static CultNetDatabaseChangeRawMessage Put(Replication run, string subscription, string key, string sku, string? commit, int count, string? id = null) => + new() + { + MessageId = id ?? Guid.NewGuid().ToString("N"), + SubscriptionId = subscription, + ChangeKind = "updated", + Document = run.Database.Documents.ToRawRecord( + run.Source.Registry.GetRequired(), new CultRecordKey(key), Item(sku), "2026-01-01T00:00:00Z"), + CommitId = commit, + CommitCount = commit == null ? null : count + }; + + private static CultNetDatabaseChangeRawMessage Removal(string subscription, string key, string commit, int count) => + new() + { + MessageId = Guid.NewGuid().ToString("N"), + SubscriptionId = subscription, + ChangeKind = "removed", + RecordKey = key, + SchemaId = "tests.replica_commit_item.v1", + CommitId = commit, + CommitCount = count + }; + + [Test] + public async Task TwoSubscriptionsMayUseOneCommitIdWithoutJoiningTheirCommits() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"))); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "a", "two", "9", 2)); + emit.Emit(Put(run, "other", "x", "ex", "9", 2)); + await Task.Delay(100); + Assert.That(run.ReplicaItem("a")!.Sku, Is.EqualTo("one"), "one change of each of two commits is not a commit"); + Assert.That(run.ReplicaItem("x"), Is.Null); + + emit.Emit(Put(run, "items", "b", "one", "9", 2)); + await Until(() => run.ReplicaItem("a")!.Sku == "two"); + Assert.That(run.ReplicaItem("b")!.Sku, Is.EqualTo("one")); + Assert.That(run.ReplicaItem("x"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + } + + [Test] + public async Task ARemovalAfterTheFirstChangeOfACommitIsApplied() + { + using var run = new Replication(source => Stage(source, ("a", "one"))); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "c", "one", "5", 2)); + emit.Emit(Removal("items", "a", "5", 2)); + await Until(() => run.ReplicaItem("c") != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.Client.Fault, Is.Null); + } + + [Test] + public void ASnapshotOfManyDocumentsAppliesEveryOne() + { + using var run = new Replication(source => Stage(source, ("a", "one"), ("b", "two"), ("c", "three"), ("d", "four"))); + + Assert.That(new[] { "a", "b", "c", "d" }.Select(key => run.ReplicaItem(key)?.Sku), Is.EqualTo(new[] { "one", "two", "three", "four" })); + } + + [Test] + public async Task ManyCommitsInARowAreNotMistakenForAnUnboundedBuffer() + { + using var run = new Replication(source => Stage(source, ("a", "s0"), ("b", "s1"))); + + for (var round = 1; round <= 80; round++) + { + var a = "s" + (round % 2); + var b = "s" + ((round + 1) % 2); + Stage(run.Source, ("a", a), ("b", b)); + } + + await Until(() => run.ReplicaItem("a")!.Sku == "s0"); + Assert.That(run.Client.Fault, Is.Null, "completed commits leave the buffer"); + } + + [Test] + public async Task AMessageDeliveredTwiceInOneCommitFaults() + { + 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); + + Assert.That(run.Faults, Has.Count.EqualTo(1)); + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + [Test] + public async Task ACommitThatChangesItsAnnouncedSizeFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 2)); + run.Peer.Client!.Emit(Put(run, "items", "b", "two", "3", 3)); + await Until(() => run.Client.Fault != null); + + Assert.That(run.ReplicaItem("a"), Is.Null); + } + + [Test] + public async Task ACommitAnnouncingMoreChangesThanAReplicaHoldsFaults() + { + using var run = new Replication(); + + run.Peer.Client!.Emit(Put(run, "items", "a", "one", "3", 1_000_000)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task MoreIncompleteCommitsThanTheBoundFault() + { + using var run = new Replication(); + + for (var commit = 0; commit < 70; commit++) + run.Peer.Client!.Emit(Put(run, "items", "k" + commit, "v" + commit, "c" + commit, 2)); + await Until(() => run.Client.Fault != null); + } + + [Test] + public async Task UnsubscribingDropsTheCommitsStillWaiting() + { + using var run = new Replication(); + var emit = run.Peer.Client!; + + emit.Emit(Put(run, "items", "a", "one", "4", 2)); + await Task.Delay(50); + run.Client.Unsubscribe("items"); + emit.Emit(Put(run, "items", "b", "two", "4", 2)); + await Task.Delay(100); + + Assert.That(run.ReplicaItem("a"), Is.Null); + Assert.That(run.ReplicaItem("b"), Is.Null, "the second half of a dropped commit is not a commit"); + } + + [Test] + public async Task AFaultedClientRefusesANewSubscription() + { + using var run = new Replication(seedReplica: replica => Stage(replica, ("local", "taken"))); + Stage(run.Source, ("a", "taken")); + await Until(() => run.Client.Fault != null); + + var refusal = Assert.Throws(() => run.Client.SubscribeAsync("again")); + Assert.That(refusal!.InnerException, Is.SameAs(run.Client.Fault)); + } } }