diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs new file mode 100644 index 00000000..659e73b0 --- /dev/null +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs @@ -0,0 +1,182 @@ +using System; +using System.Linq; +using System.Threading; +using System.Threading.Tasks; + +using FluentAssertions; +using Xunit; + +namespace DynamicData.Tests.Cache; + +public static partial class SuspendNotificationsFixture +{ + public sealed class IntegrationTests + : IntegrationTestFixtureBase + { + [Fact] + public void ResumeDeliversPendingChangesWithoutHoldingTheLock() + { + // On resume, the changes accumulated while suspended must be delivered to + // subscribers WITHOUT the cache lock held. A subscriber that blocks mid-delivery + // must therefore not stall an unrelated, lock-requiring operation on another + // thread. If the lock were held during delivery, the concurrent operation would + // block for the full duration of the blocked subscriber. + // + // Dedicated threads are used (rather than the thread pool) so the test is immune + // to pool starvation when run alongside other test collections. + using var cache = new SourceCache(static x => x); + using var deliveryStarted = new ManualResetEventSlim(false); + using var releaseDelivery = new ManualResetEventSlim(false); + using var concurrentOpDone = new ManualResetEventSlim(false); + + // An active subscriber (connected before suspension) that blocks on its first delivery. + var blockOnce = true; + using var slowSub = cache.Connect().Subscribe(_ => + { + if (blockOnce) + { + blockOnce = false; + deliveryStarted.Set(); + releaseDelivery.Wait(TimeSpan.FromSeconds(30)); + } + }); + + var suspend = cache.SuspendNotifications(); + cache.AddOrUpdate(Enumerable.Range(0, 10)); + + // Resume on a background thread: the accumulated changes are delivered to the + // slow subscriber, which blocks partway through. + var resumeThread = new Thread(suspend.Dispose) { IsBackground = true }; + resumeThread.Start(); + deliveryStarted.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue("delivery of pending changes should have started"); + + // While delivery is blocked, a concurrent lock-requiring operation must complete + // promptly. It cannot if the delivery is happening under the cache lock. The + // distinguishing gap is large (milliseconds if the lock is free, ~30s if held), + // so a generous threshold stays reliable under cold-start JIT and heavy load. + var concurrentThread = new Thread(() => + { + cache.SuspendNotifications().Dispose(); + concurrentOpDone.Set(); + }) { IsBackground = true }; + concurrentThread.Start(); + + var completedWhileBlocked = concurrentOpDone.Wait(TimeSpan.FromSeconds(10)); + + releaseDelivery.Set(); + resumeThread.Join(TimeSpan.FromSeconds(30)).Should().BeTrue("resume should complete"); + concurrentThread.Join(TimeSpan.FromSeconds(30)).Should().BeTrue("concurrent operation should complete"); + + completedWhileBlocked.Should().BeTrue("a concurrent operation must not block while pending changes are delivered; the lock must not be held during delivery"); + cache.Count.Should().Be(10, "all items should be present after resume"); + } + + [Fact] + public void StaleResumeSignalIsSuppressedByConcurrentReSuspend() + { + // Deterministic reproduction of the suspend/resume state-divergence race (#1131). + // Resume decrements the suspend count in one step and emits its resume signal in a + // later step. If a re-suspend slips in between, the resume signal must NOT fire: + // otherwise the suspended-notification subject would say "resumed" while the count + // says "suspended", and a connection made during the re-suspension would wrongly + // activate and receive data while notifications are suspended. + // + // The interleaving is forced deterministically: an active subscriber blocks the + // resume thread inside delivery, after the count has been decremented to zero but + // before the resume signal, giving the main thread a window to re-suspend and connect. + using var cache = new SourceCache(static x => x); + var dataSet = Enumerable.Range(0, 50).ToList(); + + using var deliveryStarted = new ManualResetEventSlim(false); + using var releaseDelivery = new ManualResetEventSlim(false); + + var blockOnce = true; + using var activeSub = cache.Connect().Subscribe(_ => + { + if (blockOnce) + { + blockOnce = false; + deliveryStarted.Set(); + releaseDelivery.Wait(TimeSpan.FromSeconds(30)); + } + }); + + var suspend1 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet); + + // Resume on a background thread: it decrements the count to zero and, while delivering + // the accumulated changes to the blocking subscriber, parks BEFORE the resume signal. + var resumeThread = new Thread(suspend1.Dispose) { IsBackground = true }; + resumeThread.Start(); + deliveryStarted.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue("delivery should have started"); + + // The resume thread is parked after decrementing the count but before signalling + // resume. Re-suspend and connect a new subscriber while the count is transiently zero. + var suspend2 = cache.SuspendNotifications(); + using var lateResults = cache.Connect().AsAggregator(); + + // Let the resume thread proceed to its now-stale resume signal. + releaseDelivery.Set(); + resumeThread.Join(TimeSpan.FromSeconds(30)).Should().BeTrue("resume should complete"); + + // The late subscriber connected while re-suspended: it must NOT have activated, + // because notifications ARE suspended (suspend2 is active). The stale resume signal + // must be suppressed by re-checking the suspend count. + lateResults.Messages.Count.Should().Be(0, "a connection made during re-suspension must not activate on a stale resume signal"); + lateResults.Data.Count.Should().Be(0, "no data should be delivered while suspended"); + + // Releasing the real suspension delivers the data normally. + suspend2.Dispose(); + lateResults.Data.Count.Should().Be(dataSet.Count, "all data should arrive once truly resumed"); + lateResults.Messages.Count.Should().Be(1, "a single changeset on the real resume"); + } + + [Fact] + public async Task ConcurrentSuspendDuringResumeDoesNotCorrupt() + { + // Stress test: races resume against re-suspend on two threads. + // Both orderings are correct (tested deterministically above). + // This test verifies no corruption, deadlocks, or data loss under contention. + const int iterations = 200; + var dataSet1 = Enumerable.Range(0, 100).ToList(); + var dataSet2 = Enumerable.Range(1000, 100).ToList(); + var allData = dataSet1.Concat(dataSet2).ToList(); + + for (var iter = 0; iter < iterations; iter++) + { + using var cache = new SourceCache(static x => x); + + var suspend1 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet1); + using var results = cache.Connect().AsAggregator(); + + using var barrier = new Barrier(2); + var resumeTask = Task.Run(() => + { + barrier.SignalAndWait(); + suspend1.Dispose(); + }); + + var reSuspendTask = Task.Run(() => + { + barrier.SignalAndWait(); + return cache.SuspendNotifications(); + }); + + await Task.WhenAll(resumeTask, reSuspendTask); + var suspend2 = await reSuspendTask; + + cache.AddOrUpdate(dataSet2); + suspend2.Dispose(); + + results.Summary.Overall.Adds.Should().Be(allData.Count, $"iteration {iter}: exactly {allData.Count} adds"); + results.Summary.Overall.Removes.Should().Be(0, $"iteration {iter}: no removes"); + results.Summary.Overall.Updates.Should().Be(0, $"iteration {iter}: no updates because keys don't overlap"); + results.Data.Count.Should().Be(allData.Count, $"iteration {iter}: {allData.Count} items in final state"); + results.Data.Keys.OrderBy(k => k).Should().Equal(allData, $"iteration {iter}: all keys present in order"); + results.Error.Should().BeNull($"iteration {iter}: no errors"); + results.IsCompleted.Should().BeFalse($"iteration {iter}: not completed"); + } + } + } +} diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs new file mode 100644 index 00000000..b2e8947c --- /dev/null +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -0,0 +1,464 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive.Linq; +using System.Threading.Tasks; + +using FluentAssertions; +using Xunit; + +using DynamicData.Kernel; + +namespace DynamicData.Tests.Cache +{ + public static partial class SuspendNotificationsFixture + { + public sealed partial class UnitTests + : IDisposable + { + private readonly SourceCache _source = new(static x => x); + + private readonly ChangeSetAggregator _results; + + private readonly List _countChangeHistory = []; + + private readonly IDisposable _countChangeSubscription; + + public UnitTests() + { + _results = _source.Connect().AsAggregator(); + _countChangeSubscription = _source.CountChanged.Do(_countChangeHistory.Add).Subscribe(); + } + + [Fact] + public void NotificationsCanBeSuspended() + { + // Arrange + using var suspend = _source.SuspendNotifications(); + + // Act + _source.AddOrUpdate(1); + + // Assert + _results.Messages.Count.Should().Be(0, "Should have no item updates"); + _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void SuspendingNotificationsDoesNotImpactPreview() + { + // Arrange + using var previewResults = _source.Preview().AsAggregator(); + using var suspend = _source.SuspendNotifications(); + + // Act + _source.AddOrUpdate(1); + + // Assert + previewResults.Messages.Count.Should().Be(1, "should have received a message in Preview"); + _results.Messages.Count.Should().Be(0, "should not have gotten any updates"); + _results.Data.Count.Should().Be(0, "should not receive data after suspend"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void SuspendingNotificationsPreventsWatch() + { + // Arrange + var gotData = false; + using var suspend = _source.SuspendNotifications(); + using var sub = _source.Watch(1).Do(_ => gotData = true).Subscribe(); + + // Act + _source.AddOrUpdate(1); + + // Assert + gotData.Should().BeFalse("Should not have received data after suspend"); + _results.Messages.Count.Should().Be(0, "Should have no item updates"); + _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void NotificationsCanBeResumed() + { + // Arrange + { + using var suspend = _source.SuspendNotifications(); + } + + // Act + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + + // Assert + _results.Messages.Count.Should().Be(37, "Should receive updates after resume"); + _results.Data.Count.Should().Be(37, "Should receive data after resume"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void ExistingDataNotEmittedWhileSuspended() + { + // Arrange + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + + // Act + using var results = _source.Connect().AsAggregator(); + + // Assert + results.Messages.Count.Should().Be(0, "Should have no item updates"); + results.Data.Count.Should().Be(0, "Should not receive data after suspend"); + results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void ExistingDataNotEmittedViaWatchUntilResumed() + { + // Arrange + var gotData = false; + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + using var sub = _source.Watch(1).Do(_ => gotData = true).Subscribe(); + + // Act + suspend.Dispose(); + + // Assert + gotData.Should().BeTrue("should have received a notice after the suspend was released"); + } + + [Fact] + public void ExistingDataNotEmittedUntilResumed() + { + // Arrange + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + using var results = _source.Connect().AsAggregator(); + + // Act + suspend.Dispose(); + + // Assert + results.Messages.Count.Should().Be(1, "Should receive updates after resume"); + results.Data.Count.Should().Be(37, "Should receive data after resume"); + results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void ExistingAndNewDataEmittedAsASingleChangesetOnResume() + { + // Arrange + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + using var results = _source.Connect().AsAggregator(); + Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); + + // Act + suspend.Dispose(); + + // Assert + results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); + results.Data.Count.Should().Be(37 * 2, "Should receive data after resume"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void PendingNotificationsEmittedAsSingleChangeSetOnResume() + { + // Arrange + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); + _source.RemoveKey(1); + + // Act + suspend.Dispose(); + + // Assert + _results.Data.Count.Should().Be(36, "Should receive data after resume"); + _results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); + _results.Messages[0].Adds.Should().Be(37, "Should have 37 adds"); + _results.Messages[0].Removes.Should().Be(1, "Should show the remove"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void MultipleSuspendsAreCumulative() + { + // Arrange + var suspend = _source.SuspendNotifications(); + using var suspend2 = _source.SuspendNotifications(); + _source.AddOrUpdate(1); + + // Act + suspend.Dispose(); + + // Assert + _results.Messages.Count.Should().Be(0, "Should have no item updates"); + _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void MultipleSuspendsCanBeResumed() + { + // Arrange + var suspend = _source.SuspendNotifications(); + var suspend2 = _source.SuspendNotifications(); + _source.AddOrUpdate(1); + suspend.Dispose(); + + // Act + suspend2.Dispose(); + + // Assert + _results.Messages.Count.Should().Be(1, "Should receive updates after resume"); + _results.Data.Count.Should().Be(1, "Should receive data after resume"); + _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); + } + + [Fact] + public void OnCompletedFiresIfCacheDisposedWhileSuspended() + { + // Arrange + using var suspend = _source.SuspendNotifications(); + using var results = _source.Connect().AsAggregator(); + Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); + + // Act + _source.Dispose(); + + // Assert + results.IsCompleted.Should().BeTrue("IsCompleted should fire even if Notifications are suspended"); + results.Messages.Count.Should().Be(0, "Shouldn't receive any Changesets"); + results.Data.Count.Should().Be(0, "Shouldn't receive any Data"); + } + + [Fact] + public void CountNotificationsCanBeSuspended() + { + // Arrange + using var suspend = _source.SuspendCount(); + + // Act + _source.AddOrUpdate(1); + + // Assert + _countChangeHistory.Count.Should().Be(1, "Should Not receive count updates"); + _countChangeHistory[0].Should().Be(0, "Should have only received the empty list"); + } + + [Fact] + public void CountNotificationsCanBeResumed() + { + // Arrange + { + using var suspend = _source.SuspendCount(); + } + + // Act + _source.AddOrUpdate(1); + + // Assert + _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); + _countChangeHistory[0].Should().Be(0, "Should have received the empty list"); + _countChangeHistory[1].Should().Be(1, "Should have received the updated count"); + } + + [Fact] + public void CountChangedAlwaysStartsWithInitialEvenWhenSuspended() + { + // Arrange + _source.AddOrUpdate(Enumerable.Range(1, 50)); + var countChangeHistory = new List(); + using var suspend = _source.SuspendCount(); + using var countChangeSubscription = _source.CountChanged.Do(countChangeHistory.Add).Subscribe(); + + // Act + Enumerable.Range(100, 50).ForEach(_source.AddOrUpdate); + + // Assert + countChangeHistory.Count.Should().Be(1, "Should receive initial value"); + countChangeHistory[0].Should().Be(50, "Should have received the correct initial value"); + } + + [Fact] + public void PendingCountNotificationsEmittedOnResume() + { + // Arrange + var suspend = _source.SuspendCount(); + _source.AddOrUpdate(1); + _source.AddOrUpdate(2); + _source.AddOrUpdate(3); + + // Act + suspend.Dispose(); + + // Assert + _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); + _countChangeHistory[0].Should().Be(0, "Should have received the initial 0 count"); + _countChangeHistory[1].Should().Be(3, "Should have received the updated count"); + } + + [Fact] + public void MultipleCountSuspendsAreCumulative() + { + // Arrange + var suspend = _source.SuspendCount(); + using var suspend2 = _source.SuspendCount(); + _source.AddOrUpdate(1); + _source.AddOrUpdate(2); + _source.AddOrUpdate(3); + + // Act + suspend.Dispose(); + + // Assert + _countChangeHistory.Count.Should().Be(1, "Should Not receive count updates"); + _countChangeHistory[0].Should().Be(0, "Should have only received the empty list"); + } + + [Fact] + public void MultipleCountSuspendsCanBeResumed() + { + // Arrange + var suspend = _source.SuspendCount(); + var suspend2 = _source.SuspendCount(); + _source.AddOrUpdate(1); + _source.AddOrUpdate(2); + _source.AddOrUpdate(3); + suspend.Dispose(); + + // Act + suspend2.Dispose(); + + // Assert + _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); + _countChangeHistory[0].Should().Be(0, "Should have received the initial 0 count"); + _countChangeHistory[1].Should().Be(3, "Should have received the updated count"); + } + + [Fact] + public async Task SuspensionsAreThreadSafe() + { + // Arrange + var suspend = _source.SuspendNotifications(); + var tasks = Enumerable.Range(1, 100).Select(x => Task.Run(() => _source.AddOrUpdate(x))).ToArray(); + await Task.WhenAll(tasks); + + // Act + await Task.Run(suspend.Dispose); + + // Assert + _results.Data.Count.Should().Be(100, "Should receive data after resume"); + _results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); + _results.Messages[0].Adds.Should().Be(100, "Should have 100 adds"); + } + + [Fact] + public void ResumeThenReSuspendDeliversFirstBatchOnly() + { + // Forces the ordering: resume completes before re-suspend. + // The deferred subscriber activates with the first batch snapshot, + // then re-suspend holds the second batch until final resume. + using var cache = new SourceCache(static x => x); + var dataSet1 = Enumerable.Range(0, 100).ToList(); + var dataSet2 = Enumerable.Range(1000, 100).ToList(); + var allData = dataSet1.Concat(dataSet2).ToList(); + + var suspend1 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet1); + + using var results = cache.Connect().AsAggregator(); + results.Messages.Count.Should().Be(0, "no messages during suspension"); + + // Resume first — subscriber activates + suspend1.Dispose(); + + results.Messages.Count.Should().Be(1, "exactly one message after resume"); + results.Messages[0].Adds.Should().Be(dataSet1.Count, $"snapshot should have {dataSet1.Count} adds"); + results.Messages[0].Removes.Should().Be(0, "no removes"); + results.Messages[0].Updates.Should().Be(0, "no updates"); + results.Messages[0].Select(x => x.Key).Should().Equal(dataSet1, "snapshot should contain first batch keys"); + + // Re-suspend, write second batch + var suspend2 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet2); + + results.Messages.Count.Should().Be(1, "still one message — second batch held by suspension"); + results.Summary.Overall.Adds.Should().Be(dataSet1.Count, $"still {dataSet1.Count} adds total"); + + // Final resume + suspend2.Dispose(); + + results.Messages.Count.Should().Be(2, "two messages total"); + results.Messages[1].Adds.Should().Be(dataSet2.Count, $"second message has {dataSet2.Count} adds"); + results.Messages[1].Removes.Should().Be(0, "no removes in second message"); + results.Messages[1].Updates.Should().Be(0, "no updates in second message"); + results.Messages[1].Select(x => x.Key).Should().Equal(dataSet2, "second message should contain second batch keys"); + + results.Summary.Overall.Adds.Should().Be(allData.Count, $"exactly {allData.Count} adds total"); + results.Summary.Overall.Removes.Should().Be(0, "no removes"); + results.Data.Count.Should().Be(allData.Count, $"{allData.Count} items in final state"); + results.Error.Should().BeNull(); + results.IsCompleted.Should().BeFalse(); + } + + [Fact] + public void ReSuspendThenResumeDeliversAllInSingleBatch() + { + // Forces the ordering: re-suspend before resume. + // Suspend count goes 1→2→1, no resume signal fires. + // Both batches accumulate and arrive as a single changeset on final resume. + using var cache = new SourceCache(static x => x); + var dataSet1 = Enumerable.Range(0, 100).ToList(); + var dataSet2 = Enumerable.Range(1000, 100).ToList(); + var allData = dataSet1.Concat(dataSet2).ToList(); + + var suspend1 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet1); + + using var results = cache.Connect().AsAggregator(); + results.Messages.Count.Should().Be(0, "no messages during suspension"); + + // Re-suspend first — count goes 1→2 + var suspend2 = cache.SuspendNotifications(); + + // Resume first suspend — count goes 2→1, still suspended + suspend1.Dispose(); + + results.Messages.Count.Should().Be(0, "no messages — still suspended (count=1)"); + results.Summary.Overall.Adds.Should().Be(0, "no adds — still suspended"); + + // Write second batch while still suspended + cache.AddOrUpdate(dataSet2); + + results.Messages.Count.Should().Be(0, "still no messages"); + + // Final resume — count goes 1→0 + suspend2.Dispose(); + + results.Messages.Count.Should().Be(1, "single message with all data"); + results.Messages[0].Adds.Should().Be(allData.Count, $"all {allData.Count} items in one changeset"); + results.Messages[0].Removes.Should().Be(0, "no removes"); + results.Messages[0].Updates.Should().Be(0, "no updates"); + results.Messages[0].Select(c => c.Key).OrderBy(k => k).Should().Equal(allData, "should contain both batches in order"); + + results.Summary.Overall.Adds.Should().Be(allData.Count, $"exactly {allData.Count} adds total"); + results.Summary.Overall.Removes.Should().Be(0, "no removes"); + results.Summary.Overall.Updates.Should().Be(0, "no updates"); + results.Data.Count.Should().Be(allData.Count, $"{allData.Count} items in final state"); + results.Error.Should().BeNull(); + results.IsCompleted.Should().BeFalse(); + } + + public void Dispose() + { + _source.Dispose(); + _results.Dispose(); + _countChangeSubscription.Dispose(); + } + } + } +} diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.cs index db39604a..e490eb96 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.cs @@ -1,590 +1,4 @@ -using System; -using System.Collections.Generic; -using System.Linq; -using System.Reactive.Linq; -using System.Threading; -using System.Threading.Tasks; -using DynamicData.Kernel; -using FluentAssertions; +namespace DynamicData.Tests.Cache; -using Xunit; +public static partial class SuspendNotificationsFixture; -namespace DynamicData.Tests.Cache; - -public sealed class SuspendNotificationsFixture : IDisposable -{ - private readonly SourceCache _source = new(static x => x); - - private readonly ChangeSetAggregator _results; - - private readonly List _countChangeHistory = []; - - private readonly IDisposable _countChangeSubscription; - - public SuspendNotificationsFixture() - { - _results = _source.Connect().AsAggregator(); - _countChangeSubscription = _source.CountChanged.Do(_countChangeHistory.Add).Subscribe(); - } - - [Fact] - public void NotificationsCanBeSuspended() - { - // Arrange - using var suspend = _source.SuspendNotifications(); - - // Act - _source.AddOrUpdate(1); - - // Assert - _results.Messages.Count.Should().Be(0, "Should have no item updates"); - _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void SuspendingNotificationsDoesNotImpactPreview() - { - // Arrange - using var previewResults = _source.Preview().AsAggregator(); - using var suspend = _source.SuspendNotifications(); - - // Act - _source.AddOrUpdate(1); - - // Assert - previewResults.Messages.Count.Should().Be(1, "should have received a message in Preview"); - _results.Messages.Count.Should().Be(0, "should not have gotten any updates"); - _results.Data.Count.Should().Be(0, "should not receive data after suspend"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void SuspendingNotificationsPreventsWatch() - { - // Arrange - var gotData = false; - using var suspend = _source.SuspendNotifications(); - using var sub = _source.Watch(1).Do(_ => gotData = true).Subscribe(); - - // Act - _source.AddOrUpdate(1); - - // Assert - gotData.Should().BeFalse("Should not have received data after suspend"); - _results.Messages.Count.Should().Be(0, "Should have no item updates"); - _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void NotificationsCanBeResumed() - { - // Arrange - { - using var suspend = _source.SuspendNotifications(); - } - - // Act - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - - // Assert - _results.Messages.Count.Should().Be(37, "Should receive updates after resume"); - _results.Data.Count.Should().Be(37, "Should receive data after resume"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void ExistingDataNotEmittedWhileSuspended() - { - // Arrange - var suspend = _source.SuspendNotifications(); - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - - // Act - using var results = _source.Connect().AsAggregator(); - - // Assert - results.Messages.Count.Should().Be(0, "Should have no item updates"); - results.Data.Count.Should().Be(0, "Should not receive data after suspend"); - results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void ExistingDataNotEmittedViaWatchUntilResumed() - { - // Arrange - var gotData = false; - var suspend = _source.SuspendNotifications(); - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - using var sub = _source.Watch(1).Do(_ => gotData = true).Subscribe(); - - // Act - suspend.Dispose(); - - // Assert - gotData.Should().BeTrue("should have received a notice after the suspend was released"); - } - - [Fact] - public void ExistingDataNotEmittedUntilResumed() - { - // Arrange - var suspend = _source.SuspendNotifications(); - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - using var results = _source.Connect().AsAggregator(); - - // Act - suspend.Dispose(); - - // Assert - results.Messages.Count.Should().Be(1, "Should receive updates after resume"); - results.Data.Count.Should().Be(37, "Should receive data after resume"); - results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void ExistingAndNewDataEmittedAsASingleChangesetOnResume() - { - // Arrange - var suspend = _source.SuspendNotifications(); - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - using var results = _source.Connect().AsAggregator(); - Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); - - // Act - suspend.Dispose(); - - // Assert - results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); - results.Data.Count.Should().Be(37 * 2, "Should receive data after resume"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void PendingNotificationsEmittedAsSingleChangeSetOnResume() - { - // Arrange - var suspend = _source.SuspendNotifications(); - Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - _source.RemoveKey(1); - - // Act - suspend.Dispose(); - - // Assert - _results.Data.Count.Should().Be(36, "Should receive data after resume"); - _results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); - _results.Messages[0].Adds.Should().Be(37, "Should have 37 adds"); - _results.Messages[0].Removes.Should().Be(1, "Should show the remove"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void MultipleSuspendsAreCumulative() - { - // Arrange - var suspend = _source.SuspendNotifications(); - using var suspend2 = _source.SuspendNotifications(); - _source.AddOrUpdate(1); - - // Act - suspend.Dispose(); - - // Assert - _results.Messages.Count.Should().Be(0, "Should have no item updates"); - _results.Data.Count.Should().Be(0, "Should not receive data after suspend"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void MultipleSuspendsCanBeResumed() - { - // Arrange - var suspend = _source.SuspendNotifications(); - var suspend2 = _source.SuspendNotifications(); - _source.AddOrUpdate(1); - suspend.Dispose(); - - // Act - suspend2.Dispose(); - - // Assert - _results.Messages.Count.Should().Be(1, "Should receive updates after resume"); - _results.Data.Count.Should().Be(1, "Should receive data after resume"); - _results.IsCompleted.Should().BeFalse("IsCompleted should not have fired"); - } - - [Fact] - public void OnCompletedFiresIfCacheDisposedWhileSuspended() - { - // Arrange - using var suspend = _source.SuspendNotifications(); - using var results = _source.Connect().AsAggregator(); - Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); - - // Act - _source.Dispose(); - - // Assert - results.IsCompleted.Should().BeTrue("IsCompleted should fire even if Notifications are suspended"); - results.Messages.Count.Should().Be(0, "Shouldn't receive any Changesets"); - results.Data.Count.Should().Be(0, "Shouldn't receive any Data"); - } - - [Fact] - public void CountNotificationsCanBeSuspended() - { - // Arrange - using var suspend = _source.SuspendCount(); - - // Act - _source.AddOrUpdate(1); - - // Assert - _countChangeHistory.Count.Should().Be(1, "Should Not receive count updates"); - _countChangeHistory[0].Should().Be(0, "Should have only received the empty list"); - } - - [Fact] - public void CountNotificationsCanBeResumed() - { - // Arrange - { - using var suspend = _source.SuspendCount(); - } - - // Act - _source.AddOrUpdate(1); - - // Assert - _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); - _countChangeHistory[0].Should().Be(0, "Should have received the empty list"); - _countChangeHistory[1].Should().Be(1, "Should have received the updated count"); - } - - [Fact] - public void CountChangedAlwaysStartsWithInitialEvenWhenSuspended() - { - // Arrange - _source.AddOrUpdate(Enumerable.Range(1, 50)); - var countChangeHistory = new List(); - using var suspend = _source.SuspendCount(); - using var countChangeSubscription = _source.CountChanged.Do(countChangeHistory.Add).Subscribe(); - - // Act - Enumerable.Range(100, 50).ForEach(_source.AddOrUpdate); - - // Assert - countChangeHistory.Count.Should().Be(1, "Should receive initial value"); - countChangeHistory[0].Should().Be(50, "Should have received the correct initial value"); - } - - [Fact] - public void PendingCountNotificationsEmittedOnResume() - { - // Arrange - var suspend = _source.SuspendCount(); - _source.AddOrUpdate(1); - _source.AddOrUpdate(2); - _source.AddOrUpdate(3); - - // Act - suspend.Dispose(); - - // Assert - _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); - _countChangeHistory[0].Should().Be(0, "Should have received the initial 0 count"); - _countChangeHistory[1].Should().Be(3, "Should have received the updated count"); - } - - [Fact] - public void MultipleCountSuspendsAreCumulative() - { - // Arrange - var suspend = _source.SuspendCount(); - using var suspend2 = _source.SuspendCount(); - _source.AddOrUpdate(1); - _source.AddOrUpdate(2); - _source.AddOrUpdate(3); - - // Act - suspend.Dispose(); - - // Assert - _countChangeHistory.Count.Should().Be(1, "Should Not receive count updates"); - _countChangeHistory[0].Should().Be(0, "Should have only received the empty list"); - } - - [Fact] - public void MultipleCountSuspendsCanBeResumed() - { - // Arrange - var suspend = _source.SuspendCount(); - var suspend2 = _source.SuspendCount(); - _source.AddOrUpdate(1); - _source.AddOrUpdate(2); - _source.AddOrUpdate(3); - suspend.Dispose(); - - // Act - suspend2.Dispose(); - - // Assert - _countChangeHistory.Count.Should().Be(2, "Should receive count updates"); - _countChangeHistory[0].Should().Be(0, "Should have received the initial 0 count"); - _countChangeHistory[1].Should().Be(3, "Should have received the updated count"); - } - - [Fact] - public async Task SuspensionsAreThreadSafe() - { - // Arrange - var suspend = _source.SuspendNotifications(); - var tasks = Enumerable.Range(1, 100).Select(x => Task.Run(() => _source.AddOrUpdate(x))).ToArray(); - await Task.WhenAll(tasks); - - // Act - await Task.Run(suspend.Dispose); - - // Assert - _results.Data.Count.Should().Be(100, "Should receive data after resume"); - _results.Messages.Count.Should().Be(1, "Should receive single changeset on resume"); - _results.Messages[0].Adds.Should().Be(100, "Should have 100 adds"); - } - - [Fact] - public void ResumeThenReSuspendDeliversFirstBatchOnly() - { - // Forces the ordering: resume completes before re-suspend. - // The deferred subscriber activates with the first batch snapshot, - // then re-suspend holds the second batch until final resume. - using var cache = new SourceCache(static x => x); - var dataSet1 = Enumerable.Range(0, 100).ToList(); - var dataSet2 = Enumerable.Range(1000, 100).ToList(); - var allData = dataSet1.Concat(dataSet2).ToList(); - - var suspend1 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet1); - - using var results = cache.Connect().AsAggregator(); - results.Messages.Count.Should().Be(0, "no messages during suspension"); - - // Resume first — subscriber activates - suspend1.Dispose(); - - results.Messages.Count.Should().Be(1, "exactly one message after resume"); - results.Messages[0].Adds.Should().Be(dataSet1.Count, $"snapshot should have {dataSet1.Count} adds"); - results.Messages[0].Removes.Should().Be(0, "no removes"); - results.Messages[0].Updates.Should().Be(0, "no updates"); - results.Messages[0].Select(x => x.Key).Should().Equal(dataSet1, "snapshot should contain first batch keys"); - - // Re-suspend, write second batch - var suspend2 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet2); - - results.Messages.Count.Should().Be(1, "still one message — second batch held by suspension"); - results.Summary.Overall.Adds.Should().Be(dataSet1.Count, $"still {dataSet1.Count} adds total"); - - // Final resume - suspend2.Dispose(); - - results.Messages.Count.Should().Be(2, "two messages total"); - results.Messages[1].Adds.Should().Be(dataSet2.Count, $"second message has {dataSet2.Count} adds"); - results.Messages[1].Removes.Should().Be(0, "no removes in second message"); - results.Messages[1].Updates.Should().Be(0, "no updates in second message"); - results.Messages[1].Select(x => x.Key).Should().Equal(dataSet2, "second message should contain second batch keys"); - - results.Summary.Overall.Adds.Should().Be(allData.Count, $"exactly {allData.Count} adds total"); - results.Summary.Overall.Removes.Should().Be(0, "no removes"); - results.Data.Count.Should().Be(allData.Count, $"{allData.Count} items in final state"); - results.Error.Should().BeNull(); - results.IsCompleted.Should().BeFalse(); - } - - [Fact] - public void ReSuspendThenResumeDeliversAllInSingleBatch() - { - // Forces the ordering: re-suspend before resume. - // Suspend count goes 1→2→1, no resume signal fires. - // Both batches accumulate and arrive as a single changeset on final resume. - using var cache = new SourceCache(static x => x); - var dataSet1 = Enumerable.Range(0, 100).ToList(); - var dataSet2 = Enumerable.Range(1000, 100).ToList(); - var allData = dataSet1.Concat(dataSet2).ToList(); - - var suspend1 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet1); - - using var results = cache.Connect().AsAggregator(); - results.Messages.Count.Should().Be(0, "no messages during suspension"); - - // Re-suspend first — count goes 1→2 - var suspend2 = cache.SuspendNotifications(); - - // Resume first suspend — count goes 2→1, still suspended - suspend1.Dispose(); - - results.Messages.Count.Should().Be(0, "no messages — still suspended (count=1)"); - results.Summary.Overall.Adds.Should().Be(0, "no adds — still suspended"); - - // Write second batch while still suspended - cache.AddOrUpdate(dataSet2); - - results.Messages.Count.Should().Be(0, "still no messages"); - - // Final resume — count goes 1→0 - suspend2.Dispose(); - - results.Messages.Count.Should().Be(1, "single message with all data"); - results.Messages[0].Adds.Should().Be(allData.Count, $"all {allData.Count} items in one changeset"); - results.Messages[0].Removes.Should().Be(0, "no removes"); - results.Messages[0].Updates.Should().Be(0, "no updates"); - results.Messages[0].Select(c => c.Key).OrderBy(k => k).Should().Equal(allData, "should contain both batches in order"); - - results.Summary.Overall.Adds.Should().Be(allData.Count, $"exactly {allData.Count} adds total"); - results.Summary.Overall.Removes.Should().Be(0, "no removes"); - results.Summary.Overall.Updates.Should().Be(0, "no updates"); - results.Data.Count.Should().Be(allData.Count, $"{allData.Count} items in final state"); - results.Error.Should().BeNull(); - results.IsCompleted.Should().BeFalse(); - } - - [Fact] - public async Task ConcurrentSuspendDuringResumeDoesNotCorrupt() - { - // Stress test: races resume against re-suspend on two threads. - // Both orderings are correct (tested deterministically above). - // This test verifies no corruption, deadlocks, or data loss under contention. - const int iterations = 200; - var dataSet1 = Enumerable.Range(0, 100).ToList(); - var dataSet2 = Enumerable.Range(1000, 100).ToList(); - var allData = dataSet1.Concat(dataSet2).ToList(); - - for (var iter = 0; iter < iterations; iter++) - { - using var cache = new SourceCache(static x => x); - - var suspend1 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet1); - using var results = cache.Connect().AsAggregator(); - - using var barrier = new Barrier(2); - var resumeTask = Task.Run(() => - { - barrier.SignalAndWait(); - suspend1.Dispose(); - }); - - var reSuspendTask = Task.Run(() => - { - barrier.SignalAndWait(); - return cache.SuspendNotifications(); - }); - - await Task.WhenAll(resumeTask, reSuspendTask); - var suspend2 = await reSuspendTask; - - cache.AddOrUpdate(dataSet2); - suspend2.Dispose(); - - results.Summary.Overall.Adds.Should().Be(allData.Count, $"iteration {iter}: exactly {allData.Count} adds"); - results.Summary.Overall.Removes.Should().Be(0, $"iteration {iter}: no removes"); - results.Summary.Overall.Updates.Should().Be(0, $"iteration {iter}: no updates because keys don't overlap"); - results.Data.Count.Should().Be(allData.Count, $"iteration {iter}: {allData.Count} items in final state"); - results.Data.Keys.OrderBy(k => k).Should().Equal(allData, $"iteration {iter}: all keys present in order"); - results.Error.Should().BeNull($"iteration {iter}: no errors"); - results.IsCompleted.Should().BeFalse($"iteration {iter}: not completed"); - } - } - - [Fact] - public async Task ResumeSignalUnderLockPreventsStaleSnapshotFromReSuspend() - { - // Verifies that a deferred Connect subscriber never sees data written during - // a re-suspension. The resume signal fires under the lock (reentrant), so the - // deferred subscriber activates and takes its snapshot before any other thread - // can re-suspend or write new data. - // - // A slow first subscriber blocks delivery of accumulated changes, creating a - // window where the main thread re-suspends and writes a second batch. The - // deferred subscriber's snapshot must contain only the first batch. - using var cache = new SourceCache(static x => x); - var dataSet1 = Enumerable.Range(0, 100).ToList(); - var dataSet2 = Enumerable.Range(1000, 100).ToList(); - var allData = dataSet1.Concat(dataSet2).ToList(); - - using var delivering = new SemaphoreSlim(0, 1); - using var proceedWithResuspend = new SemaphoreSlim(0, 1); - - var suspend1 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet1); - - // First subscriber blocks on delivery to hold the delivery thread - var firstDelivery = true; - using var slowSub = cache.Connect().Subscribe(_ => - { - if (firstDelivery) - { - firstDelivery = false; - delivering.Release(); - proceedWithResuspend.Wait(TimeSpan.FromSeconds(5)); - } - }); - - // Deferred subscriber — will activate when resume signal fires - using var results = cache.Connect().AsAggregator(); - results.Messages.Count.Should().Be(0, "no messages during suspension"); - - // Resume on background thread — delivery blocks on slow subscriber - var resumeTask = Task.Run(() => suspend1.Dispose()); - (await delivering.WaitAsync(TimeSpan.FromSeconds(5))).Should().BeTrue("delivery should have started"); - - // Re-suspend and write second batch while delivery is blocked - var suspend2 = cache.SuspendNotifications(); - cache.AddOrUpdate(dataSet2); - - // dataSet2 must not appear in any message received so far - foreach (var msg in results.Messages) - { - foreach (var change in msg) - { - change.Key.Should().BeInRange(0, 99, - "deferred subscriber should only have first-batch keys before second resume"); - } - } - - // Unblock delivery - proceedWithResuspend.Release(); - await resumeTask; - - // Only dataSet1 should have been delivered — dataSet2 is held by second suspension - results.Summary.Overall.Adds.Should().Be(dataSet1.Count, - $"exactly {dataSet1.Count} adds before second resume — dataSet2 must be held by suspension"); - results.Messages.Should().HaveCount(1, "exactly one message (snapshot of dataSet1)"); - results.Messages[0].Adds.Should().Be(dataSet1.Count); - results.Messages[0].Select(c => c.Key).Should().Equal(dataSet1, - "snapshot should contain exactly first-batch keys in order"); - - // Resume second suspension — dataSet2 arrives now - suspend2.Dispose(); - - results.Summary.Overall.Adds.Should().Be(allData.Count, $"exactly {allData.Count} adds total"); - results.Summary.Overall.Removes.Should().Be(0, "no removes"); - results.Messages.Should().HaveCount(2, "two messages: snapshot + second batch"); - results.Messages[1].Adds.Should().Be(dataSet2.Count); - results.Messages[1].Select(c => c.Key).Should().Equal(dataSet2, - "second message should contain exactly second-batch keys in order"); - results.Data.Count.Should().Be(allData.Count); - results.Data.Keys.OrderBy(k => k).Should().Equal(allData); - results.Error.Should().BeNull(); - results.IsCompleted.Should().BeFalse(); - } - - public void Dispose() - { - _source.Dispose(); - _results.Dispose(); - _countChangeSubscription.Dispose(); - } -} diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 5cbeca98..e497fd0e 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -352,11 +352,18 @@ private void ResumeNotifications() } // Emit the resume signal after releasing the delivery scope so that - // accumulated changes are delivered first + // accumulated changes are delivered first. Re-check the suspend count + // under the lock: a concurrent SuspendNotifications() may have run in the + // window since the count was decremented, and its state must win. Emitting + // the signal only when still unsuspended keeps _notifySuspendCount and the + // suspended-notification subject from diverging. if (emitResume) { using var readLock = _notifications.AcquireReadLock(); - _suspensionTracker.Value.EmitResumeNotification(); + if (!_suspensionTracker.Value.AreNotificationsSuspended) + { + _suspensionTracker.Value.EmitResumeNotification(); + } } } @@ -476,10 +483,16 @@ public void EnqueueChanges(IEnumerable> changes) public void SuspendNotifications() { - if (++_notifySuspendCount == 1) + ++_notifySuspendCount; + + // Signal suspension based on the subject's own state rather than the count. + // ResumeNotifications() decrements the count and emits its resume signal in + // separate steps, so the count can briefly reach zero before the 'false' signal + // is emitted. Gating on the subject keeps the signal monotonic and avoids a + // redundant 'true' when a suspend interleaves in that window. + if (!_areNotificationsSuspended.IsDisposed && !_areNotificationsSuspended.Value) { Debug.Assert(_pendingChanges.Count == 0, "Shouldn't be any pending values if suspend was just started"); - Debug.Assert(!_areNotificationsSuspended.Value, "SuspendSubject should be false for the first suspend call"); _areNotificationsSuspended.OnNext(true); } }