List: cross-cache deadlock immunity (parity with cache PR #1079)

src/DynamicData/List/Internal/BufferIf.cs

@@ -7,6 +7,8 @@
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
+using DynamicData.Internal;
+
 namespace DynamicData.List.Internal;
 
 internal sealed class BufferIf<T>(IObservable<IChangeSet<T>> source, IObservable<bool> pauseIfTrueSelector, bool initialPauseState = false, TimeSpan? timeOut = null, IScheduler? scheduler = null)
@@ -23,13 +25,16 @@ internal sealed class BufferIf<T>(IObservable<IChangeSet<T>> source, IObservable
     public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
             observer =>
             {
-                var locker = InternalEx.NewLock();
+                // SharedDeliveryQueue + SynchronizeSafe replaces Synchronize(locker) so the
+                // gate lock is released before downstream OnNext. Closes the cross-cache
+                // deadlock window.
+                var queue = new SharedDeliveryQueue();
                 var paused = initialPauseState;
                 var buffer = new ChangeSet<T>();
                 var timeoutSubscriber = new SerialDisposable();
                 var timeoutSubject = new Subject<bool>();
 
-                var bufferSelector = Observable.Return(initialPauseState).Concat(_pauseIfTrueSelector.Merge(timeoutSubject)).ObserveOn(_scheduler).Synchronize(locker).Publish();
+                var bufferSelector = Observable.Return(initialPauseState).Concat(_pauseIfTrueSelector.Merge(timeoutSubject)).ObserveOn(_scheduler).SynchronizeSafe(queue).Publish();
 
                 var pause = bufferSelector.Where(state => state).Subscribe(
                     _ =>
@@ -61,7 +66,7 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                         timeoutSubscriber.Disposable = Disposable.Empty;
                     });
 
-                var updateSubscriber = _source.Synchronize(locker).Subscribe(
+                var updateSubscriber = _source.SynchronizeSafe(queue).Subscribe(
                     updates =>
                     {
                         if (paused)
@@ -85,6 +90,7 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                         updateSubscriber.Dispose();
                         timeoutSubject.OnCompleted();
                         timeoutSubscriber.Dispose();
+                        queue.Dispose();
                     });
             });
 }

src/DynamicData/List/Internal/Combiner.cs

@@ -1,51 +1,47 @@
-// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
+// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
 // Roland Pheasant licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for full license information.
 
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 
 using DynamicData.Cache.Internal;
+using DynamicData.Internal;
 
 namespace DynamicData.List.Internal;
 
 internal sealed class Combiner<T>(ICollection<IObservable<IChangeSet<T>>> source, CombineOperator type)
     where T : notnull
 {
-#if NET9_0_OR_GREATER
-    private readonly Lock _locker = new();
-#else
-    private readonly object _locker = new();
-#endif
-
     private readonly ICollection<IObservable<IChangeSet<T>>> _source = source ?? throw new ArgumentNullException(nameof(source));
 
     public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
             observer =>
             {
                 var disposable = new CompositeDisposable();
-
                 var resultList = new ChangeAwareListWithRefCounts<T>();
+                var sourceLists = Enumerable.Range(0, _source.Count).Select(_ => new ReferenceCountTracker<T>()).ToList();
 
-                lock (_locker)
-                {
-                    var sourceLists = Enumerable.Range(0, _source.Count).Select(_ => new ReferenceCountTracker<T>()).ToList();
+                // Shared queue serializes the multiple source streams so they appear as a
+                // single sequence to the combiner, but without holding a lock during the
+                // downstream observer.OnNext call.
+                var queue = new SharedDeliveryQueue();
 
-                    foreach (var pair in _source.Zip(sourceLists, (item, list) => new { Item = item, List = list }))
-                    {
-                        disposable.Add(
-                            pair.Item.Synchronize(_locker).Subscribe(
-                                changes =>
+                foreach (var pair in _source.Zip(sourceLists, (item, list) => new { Item = item, List = list }))
+                {
+                    disposable.Add(
+                        pair.Item.SynchronizeSafe(queue).Subscribe(
+                            changes =>
+                            {
+                                CloneSourceList(pair.List, changes);
+
+                                var notifications = UpdateResultList(changes, sourceLists, resultList);
+                                if (notifications.Count != 0)
                                 {
-                                    CloneSourceList(pair.List, changes);
-
-                                    var notifications = UpdateResultList(changes, sourceLists, resultList);
-                                    if (notifications.Count != 0)
-                                    {
-                                        observer.OnNext(notifications);
-                                    }
-                                }));
-                    }
+                                    observer.OnNext(notifications);
+                                }
+                            },
+                            observer.OnError));
                 }
 
                 return disposable;

src/DynamicData/List/Internal/DisposeMany.cs

@@ -6,6 +6,8 @@
 using System.Reactive.Disposables;
 using System.Reactive.Linq;
 
+using DynamicData.Internal;
+
 namespace DynamicData.List.Internal;
 
 internal sealed class DisposeMany<T>(IObservable<IChangeSet<T>> source)
@@ -14,11 +16,13 @@ internal sealed class DisposeMany<T>(IObservable<IChangeSet<T>> source)
     public IObservable<IChangeSet<T>> Run()
         => Observable.Create<IChangeSet<T>>(observer =>
         {
-            // Will be locking on cachedItems directly, instead of using an anonymous gate object. This is acceptable, since it's a privately-held object, there's no risk of deadlock from other consumers locking on it.
             var cachedItems = new List<T>();
 
+            // SynchronizeSafe with queue-first disposal: the DeliveryQueue is terminated and
+            // drained before the source subscription is disposed, ensuring no notifications
+            // can fire while the teardown (per-item Dispose) is in progress.
             var sourceSubscription = source
-                .Synchronize(cachedItems)
+                .SynchronizeSafe()
                 .SubscribeSafe(Observer.Create<IChangeSet<T>>(
                     onNext: changeSet =>
                     {
@@ -76,11 +80,7 @@ public IObservable<IChangeSet<T>> Run()
             return Disposable.Create(() =>
             {
                 sourceSubscription.Dispose();
-
-                lock (cachedItems)
-                {
-                    ProcessFinalization(cachedItems);
-                }
+                ProcessFinalization(cachedItems);
             });
         });
 

src/DynamicData/List/Internal/DynamicCombiner.cs

@@ -6,32 +6,32 @@
 using System.Reactive.Linq;
 
 using DynamicData.Cache.Internal;
+using DynamicData.Internal;
 
 namespace DynamicData.List.Internal;
 
 internal sealed class DynamicCombiner<T>(IObservableList<IObservable<IChangeSet<T>>> source, CombineOperator type)
     where T : notnull
 {
-#if NET9_0_OR_GREATER
-    private readonly Lock _locker = new();
-#else
-    private readonly object _locker = new();
-#endif
-
     private readonly IObservableList<IObservable<IChangeSet<T>>> _source = source ?? throw new ArgumentNullException(nameof(source));
 
     public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
             observer =>
             {
+                // SharedDeliveryQueue + SynchronizeSafe replaces Synchronize(_locker) so the
+                // gate is released before downstream OnNext. Closes the cross-cache deadlock
+                // window.
+                var queue = new SharedDeliveryQueue();
+
                 // this is the resulting list which produces all notifications
                 var resultList = new ChangeAwareListWithRefCounts<T>();
 
                 // Transform to a merge container.
                 // This populates a RefTracker when the original source is subscribed to
-                var sourceLists = _source.Connect().Synchronize(_locker).Transform(changeSet => new MergeContainer(changeSet)).AsObservableList();
+                var sourceLists = _source.Connect().SynchronizeSafe(queue).Transform(changeSet => new MergeContainer(changeSet)).AsObservableList();
 
                 // merge the items back together
-                var allChanges = sourceLists.Connect().MergeMany(mc => mc.Source).Synchronize(_locker).Subscribe(
+                var allChanges = sourceLists.Connect().MergeMany(mc => mc.Source).SynchronizeSafe(queue).Subscribe(
                     changes =>
                     {
                         // Populate result list and check for changes
@@ -86,7 +86,7 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                         }
                     }).Subscribe();
 
-                return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged);
+                return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged, queue);
             });
 
     private bool MatchesConstraint(MergeContainer[] sourceLists, T item)