From 8a3cfcbbf6e3849f9dae94ff031855c7827c2b9a Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 4 Jul 2026 00:36:23 -0500 Subject: [PATCH 1/2] Fixed that `ObservableCache.SuspendNotifications()` uses a `lock` to synchronize updates to internal state for tracking notification suspensions, but `.ResumeNotifications()` did not use any synchronization when updating that same state, allowing for corruption of state if the two methods happen to overlap, concurrently. Also fixed that the test for this behavior was intermittent, specifically that it tended to falsely pass when running in parallel with many other tests. The test has now been moved to its own `IntegrationTests` fixture, to guarantee it is never parallelized with other tests. Resolves #1131. --- ...ndNotificationsFixture.IntegrationTests.cs | 147 +++++ .../SuspendNotificationsFixture.UnitTests.cs | 464 ++++++++++++++ .../Cache/SuspendNotificationsFixture.cs | 590 +----------------- src/DynamicData/Cache/ObservableCache.cs | 8 +- 4 files changed, 620 insertions(+), 589 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs create mode 100644 src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs new file mode 100644 index 000000000..8d254e273 --- /dev/null +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs @@ -0,0 +1,147 @@ +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 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(); + } + } +} diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs new file mode 100644 index 000000000..b2e8947ce --- /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 db39604a5..e490eb968 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 5cbeca988..786652f41 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -166,7 +166,13 @@ public IDisposable SuspendNotifications() lock (_locker) { _suspensionTracker.Value.SuspendNotifications(); - return Disposable.Create(this, static cache => cache.ResumeNotifications()); + return Disposable.Create(this, static cache => + { + lock (cache._locker) + { + cache.ResumeNotifications(); + } + }); } } From 6911074dc1907537845d962274bc183252fdeeae Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Thu, 9 Jul 2026 19:44:25 -0700 Subject: [PATCH 2/2] Deliver resumed notifications off the lock, closing the #1131 race without holding it ResumeNotifications() was wrapped in lock(_locker), which via lock reentrancy held the cache lock across the DeliveryQueue drain, delivering accumulated changes to subscribers while the lock was held. Restore off-lock delivery and close the suspend/resume state-divergence race a different way: - SuspendNotifications() no longer wraps the resume callback in the cache lock; queued changes drain outside the lock, as the DeliveryQueue intends. - ResumeNotifications() re-checks the suspend count under the lock before emitting the resume signal, so a concurrent SuspendNotifications() in the decrement/emit window wins and the count and the subject cannot diverge. - SuspensionTracker.SuspendNotifications() gates the suspended signal on the subject's own value rather than the count, keeping it monotonic while a resume's signal is still in flight. Tests: - Add StaleResumeSignalIsSuppressedByConcurrentReSuspend: deterministic proof that a connection made during a racing re-suspend does not activate on a stale resume signal (fails without the count re-check). - Add ResumeDeliversPendingChangesWithoutHoldingTheLock: proves a blocked subscriber does not stall a concurrent lock-requiring operation during resume delivery. - Remove ResumeSignalUnderLockPreventsStaleSnapshotFromReSuspend: redundant coverage that only passed by timing out a deadlock. --- ...ndNotificationsFixture.IntegrationTests.cs | 201 ++++++++++-------- src/DynamicData/Cache/ObservableCache.cs | 29 ++- 2 files changed, 136 insertions(+), 94 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs index 8d254e273..659e73b0d 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs @@ -13,6 +13,124 @@ 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() { @@ -60,88 +178,5 @@ public async Task ConcurrentSuspendDuringResumeDoesNotCorrupt() 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(); - } } } diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 786652f41..e497fd0ea 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -166,13 +166,7 @@ public IDisposable SuspendNotifications() lock (_locker) { _suspensionTracker.Value.SuspendNotifications(); - return Disposable.Create(this, static cache => - { - lock (cache._locker) - { - cache.ResumeNotifications(); - } - }); + return Disposable.Create(this, static cache => cache.ResumeNotifications()); } } @@ -358,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(); + } } } @@ -482,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); } }