From b45c01cfa98f03bcd9c678f51868eeb17e101db4 Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 4 Jul 2026 00:36:23 -0500 Subject: [PATCH 1/5] 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 ce9fbe4b52fe4e737949c3a37ae2502faeb6952e Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Thu, 9 Jul 2026 19:44:25 -0700 Subject: [PATCH 2/5] 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); } } From 2dee6721d554b030d5a0878b6842e1e10b3fd0b9 Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 25 Jul 2026 15:26:45 -0500 Subject: [PATCH 3/5] Adjusted SuspendNotificationsFixture.UnitTests, for consistent styling. --- .../SuspendNotificationsFixture.UnitTests.cs | 789 +++++++++--------- 1 file changed, 394 insertions(+), 395 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs index b2e8947ce..ddc9d6c07 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -9,456 +9,455 @@ using DynamicData.Kernel; -namespace DynamicData.Tests.Cache +namespace DynamicData.Tests.Cache; + +public static partial class SuspendNotificationsFixture { - public static partial class SuspendNotificationsFixture + public sealed class UnitTests + : IDisposable { - public sealed partial class UnitTests - : IDisposable - { - private readonly SourceCache _source = new(static x => x); + private readonly SourceCache _source = new(static x => x); - private readonly ChangeSetAggregator _results; + private readonly ChangeSetAggregator _results; - private readonly List _countChangeHistory = []; + private readonly List _countChangeHistory = []; - private readonly IDisposable _countChangeSubscription; + 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(); + public UnitTests() + { + _results = _source.Connect().AsAggregator(); + _countChangeSubscription = _source.CountChanged.Do(_countChangeHistory.Add).Subscribe(); + } - // Act - _source.AddOrUpdate(1); + [Fact] + public void NotificationsCanBeSuspended() + { + // Arrange + using var suspend = _source.SuspendNotifications(); - // 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"); - } + // Act + _source.AddOrUpdate(1); - [Fact] - public void SuspendingNotificationsDoesNotImpactPreview() - { - // Arrange - using var previewResults = _source.Preview().AsAggregator(); - using var suspend = _source.SuspendNotifications(); + // 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"); + } - // Act - _source.AddOrUpdate(1); + [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"); + } - // 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 SuspendingNotificationsPreventsWatch() + [Fact] + public void NotificationsCanBeResumed() + { + // Arrange { - // 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 + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - // Act - using var results = _source.Connect().AsAggregator(); + // 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"); + } - // 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 ExistingDataNotEmittedWhileSuspended() + { + // Arrange + var suspend = _source.SuspendNotifications(); + Enumerable.Range(1, 37).ForEach(_source.AddOrUpdate); - [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 + using var results = _source.Connect().AsAggregator(); - // 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"); + } - // Assert - gotData.Should().BeTrue("should have received a notice after the suspend was released"); - } + [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(); - [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"); - } + // Act + suspend.Dispose(); - [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"); - } + // Assert + gotData.Should().BeTrue("should have received a notice after the suspend was released"); + } - [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 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 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 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 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 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 OnCompletedFiresIfCacheDisposedWhileSuspended() - { - // Arrange - using var suspend = _source.SuspendNotifications(); - using var results = _source.Connect().AsAggregator(); - Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); + [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"); + } - // Act - _source.Dispose(); + [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"); + } - // 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 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(); + [Fact] + public void CountNotificationsCanBeSuspended() + { + // Arrange + using var suspend = _source.SuspendCount(); - // Act - _source.AddOrUpdate(1); + // 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"); - } + // 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() + [Fact] + public void CountNotificationsCanBeResumed() + { + // Arrange { - // 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"); + using var suspend = _source.SuspendCount(); } - [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 + _source.AddOrUpdate(1); - // Act - Enumerable.Range(100, 50).ForEach(_source.AddOrUpdate); + // 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"); + } - // 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 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 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 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 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 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 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(); + [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); + var suspend1 = cache.SuspendNotifications(); + cache.AddOrUpdate(dataSet1); - using var results = cache.Connect().AsAggregator(); - results.Messages.Count.Should().Be(0, "no messages during suspension"); + 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(); + // Re-suspend first — count goes 1→2 + var suspend2 = cache.SuspendNotifications(); - // Resume first suspend — count goes 2→1, still suspended - suspend1.Dispose(); + // 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"); + 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); + // Write second batch while still suspended + cache.AddOrUpdate(dataSet2); - results.Messages.Count.Should().Be(0, "still no messages"); + results.Messages.Count.Should().Be(0, "still no messages"); - // Final resume — count goes 1→0 - suspend2.Dispose(); + // 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.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(); - } + 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(); - } + public void Dispose() + { + _source.Dispose(); + _results.Dispose(); + _countChangeSubscription.Dispose(); } } } From 01bcb6f601aad2abd0741d100a43672dcf77b3ad Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 25 Jul 2026 15:39:17 -0500 Subject: [PATCH 4/5] Moved an additional test exercising concurrency, that got missed, from SuspendNotificationsFixture.UnitTests to SuspendNotificationsFixture.IntegrationTests. --- ...ndNotificationsFixture.IntegrationTests.cs | 24 +++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs index 659e73b0d..5f5c3864f 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs @@ -1,5 +1,7 @@ using System; +using System.Collections.Generic; using System.Linq; +using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; @@ -131,6 +133,28 @@ public void StaleResumeSignalIsSuppressedByConcurrentReSuspend() lateResults.Messages.Count.Should().Be(1, "a single changeset on the real resume"); } + [Fact] + public async Task SuspensionsAreThreadSafe() + { + // Arrange + using var source = new SourceCache(static x => x); + var results = source.Connect().AsAggregator(); + var countChangeHistory = new List(); + using var countChangeSubscription = source.CountChanged.Do(countChangeHistory.Add).Subscribe(); + + // Act + using var suspend = source.SuspendNotifications(); + var tasks = Enumerable.Range(1, 100).Select(x => Task.Run(() => source.AddOrUpdate(x))).ToArray(); + await Task.WhenAll(tasks); + + 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 async Task ConcurrentSuspendDuringResumeDoesNotCorrupt() { From bf07da524d9c6b2d8311699e47f810fdd3915f58 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Sat, 25 Jul 2026 13:46:49 -0700 Subject: [PATCH 5/5] Close the suspend/resume race by emitting the resume signal in a single lock acquisition ResumeNotifications() was acquiring the lock twice: once to decrement the suspend count and queue the accumulated changes, and again to emit the resume signal. The gap between those two acquisitions was the race window. A SuspendNotifications() landing in it would see a zero count and skip emitting 'true', and the resume would then emit 'false' on top of it, leaving the count and the subject in disagreement. Emitting the signal inside the same scope that cleared the count removes the window entirely, rather than detecting and compensating for it after the fact. Delivery is still performed off the lock: ScopedAccess.Dispose() calls ExitLockAndDeliver(), which releases the lock before draining the queue. The signal now fires before the pending changes are delivered instead of after, which a deferred subscriber handles via the existing HasPending/_currentDeliveryVersion guard in CreateConnectObservable. Also guard EmitResumeNotification() against a disposed subject, since SuspensionTracker.Dispose() runs from the delivery callbacks off the lock and can tear the subject down underneath a concurrent resume. With the window closed, ResumeSignalUnderLockPreventsStaleSnapshotFromReSuspend is restored and passes alongside StaleResumeSignalIsSuppressedByConcurrentReSuspend. --- ...ndNotificationsFixture.IntegrationTests.cs | 84 +++++++++++++++++++ src/DynamicData/Cache/ObservableCache.cs | 53 ++++++------ 2 files changed, 108 insertions(+), 29 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs index 5f5c3864f..3ba800ee3 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs @@ -133,6 +133,89 @@ public void StaleResumeSignalIsSuppressedByConcurrentReSuspend() lateResults.Messages.Count.Should().Be(1, "a single changeset on the real resume"); } + [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(); + } + [Fact] public async Task SuspensionsAreThreadSafe() { @@ -204,3 +287,4 @@ public async Task ConcurrentSuspendDuringResumeDoesNotCorrupt() } } } + diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index e497fd0ea..afce06cc6 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -338,32 +338,25 @@ private void ResumeCount() private void ResumeNotifications() { - bool emitResume; + using var notifications = _notifications.AcquireLock(); - using (var notifications = _notifications.AcquireLock()) - { - Debug.Assert(_suspensionTracker.IsValueCreated, "Should not be Resuming Notifications without Suspend Notifications instance"); + Debug.Assert(_suspensionTracker.IsValueCreated, "Should not be Resuming Notifications without Suspend Notifications instance"); - (var changes, emitResume) = _suspensionTracker.Value.ResumeNotifications(); - if (changes is not null) - { - notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); - } + var (changes, emitResume) = _suspensionTracker.Value.ResumeNotifications(); + if (changes is not null) + { + notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } - // Emit the resume signal after releasing the delivery scope so that - // 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. + // Emit the resume signal in the same lock acquisition that cleared the suspend count, + // so the count and the signal can never disagree: there is no window for a concurrent + // SuspendNotifications() to observe one without the other. The accumulated changes are + // still delivered off the lock, when this scope exits. A deferred subscriber that + // activates on this signal snapshots the current state and skips the still-pending + // delivery via the version guard in CreateConnectObservable. if (emitResume) { - using var readLock = _notifications.AcquireReadLock(); - if (!_suspensionTracker.Value.AreNotificationsSuspended) - { - _suspensionTracker.Value.EmitResumeNotification(); - } + _suspensionTracker.Value.EmitResumeNotification(); } } @@ -483,16 +476,10 @@ public void EnqueueChanges(IEnumerable> changes) public void SuspendNotifications() { - ++_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) + if ((++_notifySuspendCount == 1) && !_areNotificationsSuspended.IsDisposed) { 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); } } @@ -520,7 +507,15 @@ public void SuspendNotifications() return (null, false); } - public void EmitResumeNotification() => _areNotificationsSuspended.OnNext(false); + public void EmitResumeNotification() + { + // Disposal runs from the delivery callbacks, which execute off the lock, so the + // subject can be torn down concurrently with a resume. + if (!_areNotificationsSuspended.IsDisposed) + { + _areNotificationsSuspended.OnNext(false); + } + } public void Dispose() {