Bug Fix: Synchronization Issues in MergeManyChangeSet operators

src/DynamicData.Tests/Cache/MergeManyChangeSetsCacheFixture.cs

@@ -53,9 +53,11 @@ public MergeManyChangeSetsCacheFixture()
     [Theory]
     [InlineData(5, 7)]
     [InlineData(10, 50)]
-    [InlineData(5, 100)]
+#if !DEBUG
+    [InlineData(10, 1_000)]
     [InlineData(200, 500)]
-    [InlineData(100, 5)]
+    [InlineData(1_000, 10)]
+#endif
     public async Task MultiThreadedStressTest(int marketCount, int priceCount)
     {
         var MaxAddTime = TimeSpan.FromSeconds(0.250);
@@ -65,23 +67,19 @@ public async Task MultiThreadedStressTest(int marketCount, int priceCount)
 
         IObservable<Unit> AddRemoveStress(int marketCount, int priceCount, int parallel, IScheduler scheduler) =>
             Observable.Create<Unit>(observer => new CompositeDisposable
-                {
+                (
                     AddRemoveMarkets(marketCount, parallel, scheduler)
-                            .Subscribe(
-                                onNext: _ => { },
-                                onError: ex => observer.OnError(ex)),
+                        .Subscribe(
+                            onNext: static _ => { },
+                            onError: observer.OnError),
 
                     _marketCache.Connect()
-                            .MergeMany(market => AddRemovePrices((Market)market, priceCount, parallel, scheduler))
-                            .Subscribe(
-                                onNext: _ => { },
-                                onError: ex => observer.OnError(ex),
-                                onCompleted: () =>
-                                    {
-                                        observer.OnNext(Unit.Default);
-                                        observer.OnCompleted();
-                                    })
-                });
+                        .MergeMany(market => AddRemovePrices((Market)market, priceCount, parallel, scheduler))
+                        .Subscribe(
+                            onNext: static _ => { },
+                            onError: observer.OnError,
+                            onCompleted: observer.OnCompleted)
+                ));
 
         IObservable<IMarket> AddRemoveMarkets(int ownerCount, int parallel, IScheduler scheduler) =>
             _marketFaker.IntervalGenerate(MaxAddTime, scheduler)

src/DynamicData.Tests/Cache/MergeManyChangeSetsListFixture.cs

@@ -51,9 +51,11 @@ public MergeManyChangeSetsListFixture()
     [Theory]
     [InlineData(5, 7)]
     [InlineData(10, 50)]
+#if !DEBUG
     [InlineData(10, 1_000)]
     [InlineData(200, 500)]
     [InlineData(1_000, 10)]
+#endif
     public async Task MultiThreadedStressTest(int ownerCount, int animalCount)
     {
         var MaxAddTime = TimeSpan.FromSeconds(0.250);
@@ -63,23 +65,19 @@ public async Task MultiThreadedStressTest(int ownerCount, int animalCount)
 
         IObservable<Unit> AddRemoveAnimalsStress(int ownerCount, int animalCount, int parallel, IScheduler scheduler) =>
             Observable.Create<Unit>(observer => new CompositeDisposable
-            {
-                AddRemoveOwners(ownerCount, parallel, scheduler)
-                    .Subscribe(
-                        onNext: static _ => { },
-                        onError: ex => observer.OnError(ex)),
+                (
+                    AddRemoveOwners(ownerCount, parallel, scheduler)
+                        .Subscribe(
+                            onNext: static _ => { },
+                            onError: observer.OnError),
 
-                _animalOwners.Connect()
+                    _animalOwners.Connect()
                         .MergeMany(owner => AddRemoveAnimals(owner, animalCount, parallel, scheduler))
                         .Subscribe(
                             onNext: static _ => { },
-                            onError: ex => observer.OnError(ex),
-                            onCompleted: () =>
-                                {
-                                    observer.OnNext(Unit.Default);
-                                    observer.OnCompleted();
-                                })
-            });
+                            onError: observer.OnError,
+                            onCompleted: observer.OnCompleted)
+                ));
 
         IObservable<AnimalOwner> AddRemoveOwners(int ownerCount, int parallel, IScheduler scheduler) =>
             _animalOwnerFaker.IntervalGenerate(_randomizer, MaxAddTime, scheduler)

src/DynamicData/Cache/Internal/MergeManyCacheChangeSets.cs

@@ -30,19 +30,19 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() => Observabl
                 var changeTracker = new ChangeSetMergeTracker<TDestination, TDestinationKey>(() => sourceCacheOfCaches.Items, comparer, equalityComparer);
 
                 // Share a connection to the source cache
-                var shared = sourceCacheOfCaches.Connect().Publish();
+                var shared = sourceCacheOfCaches.Connect().Synchronize(locker).Publish();
 
                 // Merge the child changeset changes together and apply to the tracker
-                var allChanges = shared.MergeMany(mc => mc.Source)
-                                                 .Subscribe(
-                                                        changes => changeTracker.ProcessChangeSet(changes, observer),
-                                                        observer.OnError,
-                                                        observer.OnCompleted);
+                var allChanges = shared
+                    .MergeMany(mc => mc.Source)
+                    .Subscribe(
+                        changes => changeTracker.ProcessChangeSet(changes, observer),
+                        observer.OnError,
+                        observer.OnCompleted);
 
                 // When a source item is removed, all of its sub-items need to be removed
                 var removedItems = shared
-                    .Synchronize(locker)
-                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer))
+                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer), invokeOnUnsubscribe: false)
                     .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.Cache.KeyValues, observer))
                     .Subscribe();
 

src/DynamicData/Cache/Internal/MergeManyCacheChangeSetsSourceCompare.cs

@@ -28,27 +28,28 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() => Observabl
             {
                 var locker = new object();
 
-                // Transform to an observable cache of merge containers.
+                // Transform to an cache of child caches of ParentChildEntry
                 var sourceCacheOfCaches = source
-                                            .Transform((obj, key) => new ChangeSetCache<ParentChildEntry, TDestinationKey>(_changeSetSelector(obj, key).Synchronize(locker)))
-                                            .AsObservableCache();
-
-                // Share a single connection to the cache
-                var shared = sourceCacheOfCaches.Connect().Synchronize(locker).Publish();
+                    .Transform((obj, key) => new ChangeSetCache<ParentChildEntry, TDestinationKey>(_changeSetSelector(obj, key).Synchronize(locker)))
+                    .AsObservableCache();
 
                 // This is manages all of the changes
                 var changeTracker = new ChangeSetMergeTracker<ParentChildEntry, TDestinationKey>(() => sourceCacheOfCaches.Items, _comparer, _equalityComparer);
 
+                // Share a single connection to the cache
+                var shared = sourceCacheOfCaches.Connect().Synchronize(locker).Publish();
+
                 // Merge the child changeset changes together and apply to the tracker
-                var allChanges = shared.MergeMany(mc => mc.Source)
-                                                 .Subscribe(
-                                                        changes => changeTracker.ProcessChangeSet(changes, observer),
-                                                        observer.OnError,
-                                                        observer.OnCompleted);
+                var allChanges = shared
+                    .MergeMany(mc => mc.Source)
+                    .Subscribe(
+                        changes => changeTracker.ProcessChangeSet(changes, observer),
+                        observer.OnError,
+                        observer.OnCompleted);
 
                 // When a source item is removed, all of its sub-items need to be removed
                 var removedItems = shared
-                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer))
+                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer), invokeOnUnsubscribe: false)
                     .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.Cache.KeyValues, observer))
                     .Subscribe();
 

src/DynamicData/Cache/Internal/MergeManyListChangeSets.cs

@@ -21,27 +21,31 @@ public IObservable<IChangeSet<TDestination>> Run() => Observable.Create<IChangeS
             {
                 var locker = new object();
 
+                // Transform to a Cloned List and create an Observable Cache
+                var sourceCacheOfLists = source
+                    .Transform((obj, key) => new ClonedListChangeSet<TDestination>(selector(obj, key).Synchronize(locker), equalityComparer))
+                    .Synchronize(locker)
+                    .AsObservableCache();
+
                 // This is manages all of the changes
                 var changeTracker = new ChangeSetMergeTracker<TDestination>();
 
-                // Transform to a changeset of Cloned Child Lists and Share
-                var shared = source
-                                            .Transform((obj, key) => new ClonedListChangeSet<TDestination>(selector(obj, key).Synchronize(locker), equalityComparer))
-                                            .Publish();
+                // Share a connection to the source cache
+                var shared = sourceCacheOfLists.Connect().Synchronize(locker).Publish();
 
                 // Merge all of the children changesets together and apply to the tracker
-                var allChanges = shared.MergeMany(clonedList => clonedList.Source.RemoveIndex())
-                                                 .Subscribe(
-                                                        changes => changeTracker.ProcessChangeSet(changes, observer),
-                                                        observer.OnError,
-                                                        observer.OnCompleted);
+                var allChanges = shared
+                    .MergeMany(clonedList => clonedList.Source.RemoveIndex())
+                    .Subscribe(
+                        changes => changeTracker.ProcessChangeSet(changes, observer),
+                        observer.OnError,
+                        observer.OnCompleted);
 
                 // When a source item is removed, all of its sub-items need to be removed
                 var removedItems = shared
-                        .Synchronize(locker)
-                        .OnItemRemoved(clonedList => changeTracker.RemoveItems(clonedList.List, observer), invokeOnUnsubscribe: false)
-                        .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.List, observer))
-                        .Subscribe();
+                    .OnItemRemoved(clonedList => changeTracker.RemoveItems(clonedList.List, observer), invokeOnUnsubscribe: false)
+                    .OnItemUpdated((_, prev) => changeTracker.RemoveItems(prev.List, observer))
+                    .Subscribe();
 
                 return new CompositeDisposable(allChanges, removedItems, shared.Connect());
             });

src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs

@@ -22,7 +22,7 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() =>
             {
                 var locker = new object();
 
-                // Transform to an observable list of merge containers.
+                // Transform to a List of child caches
                 var sourceListOfCaches = source
                     .Transform(obj => new ChangeSetCache<TDestination, TDestinationKey>(changeSetSelector(obj).Synchronize(locker)))
                     .AsObservableList();
@@ -31,11 +31,10 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() =>
                 var changeTracker = new ChangeSetMergeTracker<TDestination, TDestinationKey>(() => sourceListOfCaches.Items.ToArray(), comparer, equalityComparer);
 
                 // Share a connection to the source list
-                var shared = sourceListOfCaches.Connect().Publish();
+                var shared = sourceListOfCaches.Connect().Synchronize(locker).Publish();
 
                 // Merge the child changeset changes together and apply to the tracker
                 var allChanges = shared
-                    .Synchronize(locker)
                     .MergeMany(mc => mc.Source)
                     .Subscribe(
                         changes => changeTracker.ProcessChangeSet(changes, observer),
@@ -44,8 +43,7 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() =>
 
                 // When a source item is removed, all of its sub-items need to be removed
                 var removedItems = shared
-                    .Synchronize(locker)
-                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer))
+                    .OnItemRemoved(mc => changeTracker.RemoveItems(mc.Cache.KeyValues, observer), invokeOnUnsubscribe: false)
                     .Subscribe();
 
                 return new CompositeDisposable(sourceListOfCaches, allChanges, removedItems, shared.Connect());

src/DynamicData/List/Internal/MergeManyListChangeSets.cs

@@ -20,20 +20,19 @@ public IObservable<IChangeSet<TDestination>> Run() =>
             {
                 var locker = new object();
 
-                // Transform to a changeset of Cloned Child Lists and then Share
+                // Transform to a list of child lists
                 var sourceListOfLists = source
-                                            .Transform(obj => new ClonedListChangeSet<TDestination>(selector(obj).Synchronize(locker), equalityComparer))
-                                            .AsObservableList();
+                    .Transform(obj => new ClonedListChangeSet<TDestination>(selector(obj).Synchronize(locker), equalityComparer))
+                    .AsObservableList();
 
                 // This is manages all of the changes
                 var changeTracker = new ChangeSetMergeTracker<TDestination>();
 
-                // Share a connection to the source cache
-                var shared = sourceListOfLists.Connect().Publish();
+                // Share a connection to the source list
+                var shared = sourceListOfLists.Connect().Synchronize(locker).Publish();
 
                 // Merge the items back together
                 var allChanges = shared
-                    .Synchronize(locker)
                     .MergeMany(clonedList => clonedList.Source.RemoveIndex())
                     .Subscribe(
                         changes => changeTracker.ProcessChangeSet(changes, observer),
@@ -42,7 +41,6 @@ public IObservable<IChangeSet<TDestination>> Run() =>
 
                 // When a source item is removed, all of its sub-items need to be removed
                 var removedItems = shared
-                    .Synchronize(locker)
                     .OnItemRemoved(mc => changeTracker.RemoveItems(mc.List, observer), invokeOnUnsubscribe: false)
                     .Subscribe();