Fixed #1131 - Race condition between SuspendNotifications() and .ResumeNotifications()

src/DynamicData.Tests/Cache/SuspendNotificationsFixture.IntegrationTests.cs

@@ -0,0 +1,290 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
+
+using FluentAssertions;
+using Xunit;
+
+namespace DynamicData.Tests.Cache;
+
+public static partial class SuspendNotificationsFixture
+{
+    public sealed class IntegrationTests
+        : IntegrationTestFixtureBase
+    {
+        [Fact]
+        public void ResumeDeliversPendingChangesWithoutHoldingTheLock()
+        {
+            // On resume, the changes accumulated while suspended must be delivered to
+            // subscribers WITHOUT the cache lock held. A subscriber that blocks mid-delivery
+            // must therefore not stall an unrelated, lock-requiring operation on another
+            // thread. If the lock were held during delivery, the concurrent operation would
+            // block for the full duration of the blocked subscriber.
+            //
+            // Dedicated threads are used (rather than the thread pool) so the test is immune
+            // to pool starvation when run alongside other test collections.
+            using var cache = new SourceCache<int, int>(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<int, int>(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 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<int, int>(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()
+        {
+            // Arrange
+            using var source = new SourceCache<int, int>(static x => x);
+            var results = source.Connect().AsAggregator();
+            var countChangeHistory = new List<int>();
+            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()
+        {
+            // 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<int, int>(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");
+            }
+        }
+    }
+}
+