fix: Separate mutation and notification locks in ObservableCache to prevent cross-cache deadlock

Splits _locker into _locker (mutation) and _notificationLocker (notification
delivery) so that InvokeNext no longer holds the mutation lock during
_changes.OnNext. Adds re-entrant notification queue to preserve Rx
serialization contract when subscribers trigger nested writes.

Author: dwcullop

src/DynamicData.Tests/Cache/SourceCacheFixture.cs

@@ -1,6 +1,8 @@
-using System;
+using System;
 using System.Linq;
 using System.Reactive.Linq;
+using System.Threading;
+using System.Threading.Tasks;
 
 using DynamicData.Tests.Domain;
 
@@ -188,4 +190,39 @@ public void StaticFilterRemove()
 
     public record class SomeObject(int Id, int Value);
 
+
+    [Fact]
+    public void ConcurrentEditsShouldNotDeadlockWithSubscribersThatModifyOtherCaches()
+    {
+        using var cacheA = new SourceCache<string, int>(static x => x.GetHashCode());
+        using var cacheB = new SourceCache<string, int>(static x => x.GetHashCode());
+        using var destination = new SourceCache<string, int>(static x => x.GetHashCode());
+        using var subA = cacheA.Connect().PopulateInto(destination);
+        using var subB = cacheB.Connect().PopulateInto(destination);
+
+        var count = 0;
+        using var feedback = destination.Connect().Subscribe(_ => Interlocked.Increment(ref count));
+
+        var completed = Task.WaitAll(
+            [
+                Task.Run(() =>
+                {
+                    for (var i = 0; i < 100; i++)
+                    {
+                        cacheA.AddOrUpdate($"A-{i}");
+                    }
+                }),
+                Task.Run(() =>
+                {
+                    for (var i = 0; i < 100; i++)
+                    {
+                        cacheB.AddOrUpdate($"B-{i}");
+                    }
+                }),
+            ],
+            TimeSpan.FromSeconds(10));
+
+        completed.Should().BeTrue("concurrent edits with cross-cache subscribers should not deadlock");
+        count.Should().BeGreaterThan(0, "destination should have received changeset notifications");
+    }
 }

src/DynamicData/Cache/ObservableCache.cs

@@ -32,17 +32,23 @@ internal sealed class ObservableCache<TObject, TKey> : IObservableCache<TObject,
 
 #if NET9_0_OR_GREATER
     private readonly Lock _locker = new();
+    private readonly Lock _notificationLocker = new();
 #else
     private readonly object _locker = new();
+    private readonly object _notificationLocker = new();
 #endif
 
     private readonly ReaderWriter<TObject, TKey> _readerWriter;
 
+    private readonly Queue<ChangeSet<TObject, TKey>> _pendingNotifications = new();
+
     private int _editLevel; // The level of recursion in editing.
 
+    private bool _isNotifying;
+
     public ObservableCache(IObservable<IChangeSet<TObject, TKey>> source)
     {
-        _suspensionTracker = new(() => new SuspensionTracker(_changes.OnNext, InvokeCountNext));
+        _suspensionTracker = new(() => new SuspensionTracker(InvokeNext, InvokeCountNext));
         _readerWriter = new ReaderWriter<TObject, TKey>();
 
         var loader = source.Synchronize(_locker).Finally(
@@ -83,7 +89,7 @@ public ObservableCache(IObservable<IChangeSet<TObject, TKey>> source)
 
     public ObservableCache(Func<TObject, TKey>? keySelector = null)
     {
-        _suspensionTracker = new(() => new SuspensionTracker(_changes.OnNext, InvokeCountNext));
+        _suspensionTracker = new(() => new SuspensionTracker(InvokeNext, InvokeCountNext));
         _readerWriter = new ReaderWriter<TObject, TKey>(keySelector);
 
         _cleanUp = Disposable.Create(
@@ -188,10 +194,10 @@ internal void UpdateFromIntermediate(Action<ICacheUpdater<TObject, TKey>> update
     {
         updateAction.ThrowArgumentNullExceptionIfNull(nameof(updateAction));
 
+        ChangeSet<TObject, TKey>? changes = null;
+
         lock (_locker)
         {
-            ChangeSet<TObject, TKey>? changes = null;
-
             _editLevel++;
             if (_editLevel == 1)
             {
@@ -205,21 +211,26 @@ internal void UpdateFromIntermediate(Action<ICacheUpdater<TObject, TKey>> update
 
             _editLevel--;
 
-            if (changes is not null && _editLevel == 0)
+            if (_editLevel != 0)
             {
-                InvokeNext(changes);
+                changes = null;
             }
         }
+
+        if (changes is not null)
+        {
+            InvokeNext(changes);
+        }
     }
 
     internal void UpdateFromSource(Action<ISourceUpdater<TObject, TKey>> updateAction)
     {
         updateAction.ThrowArgumentNullExceptionIfNull(nameof(updateAction));
 
+        ChangeSet<TObject, TKey>? changes = null;
+
         lock (_locker)
         {
-            ChangeSet<TObject, TKey>? changes = null;
-
             _editLevel++;
             if (_editLevel == 1)
             {
@@ -233,11 +244,16 @@ internal void UpdateFromSource(Action<ISourceUpdater<TObject, TKey>> updateActio
 
             _editLevel--;
 
-            if (changes is not null && _editLevel == 0)
+            if (_editLevel != 0)
             {
-                InvokeNext(changes);
+                changes = null;
             }
         }
+
+        if (changes is not null)
+        {
+            InvokeNext(changes);
+        }
     }
 
     private IObservable<IChangeSet<TObject, TKey>> CreateConnectObservable(Func<TObject, bool>? predicate, bool suppressEmptyChangeSets) =>
@@ -246,19 +262,22 @@ private IObservable<IChangeSet<TObject, TKey>> CreateConnectObservable(Func<TObj
             {
                 lock (_locker)
                 {
-                    var initial = InternalEx.Return(() => (IChangeSet<TObject, TKey>)GetInitialUpdates(predicate));
-                    var changes = initial.Concat(_changes);
-
-                    if (predicate != null)
+                    lock (_notificationLocker)
                     {
-                        changes = changes.Filter(predicate, suppressEmptyChangeSets);
-                    }
-                    else if (suppressEmptyChangeSets)
-                    {
-                        changes = changes.NotEmpty();
-                    }
+                        var initial = InternalEx.Return(() => (IChangeSet<TObject, TKey>)GetInitialUpdates(predicate));
+                        var changes = initial.Concat(_changes);
+
+                        if (predicate != null)
+                        {
+                            changes = changes.Filter(predicate, suppressEmptyChangeSets);
+                        }
+                        else if (suppressEmptyChangeSets)
+                        {
+                            changes = changes.NotEmpty();
+                        }
 
-                    return changes.SubscribeSafe(observer);
+                        return changes.SubscribeSafe(observer);
+                    }
                 }
             });
 
@@ -268,36 +287,60 @@ private IObservable<Change<TObject, TKey>> CreateWatchObservable(TKey key) =>
             {
                 lock (_locker)
                 {
-                    var initial = _readerWriter.Lookup(key);
-                    if (initial.HasValue)
+                    lock (_notificationLocker)
                     {
-                        observer.OnNext(new Change<TObject, TKey>(ChangeReason.Add, key, initial.Value));
-                    }
-
-                    return _changes.Finally(observer.OnCompleted).Subscribe(
-                        changes =>
+                        var initial = _readerWriter.Lookup(key);
+                        if (initial.HasValue)
                         {
-                            foreach (var change in changes.ToConcreteType())
+                            observer.OnNext(new Change<TObject, TKey>(ChangeReason.Add, key, initial.Value));
+                        }
+
+                        return _changes.Finally(observer.OnCompleted).Subscribe(
+                            changes =>
                             {
-                                var match = EqualityComparer<TKey>.Default.Equals(change.Key, key);
-                                if (match)
+                                foreach (var change in changes.ToConcreteType())
                                 {
-                                    observer.OnNext(change);
+                                    var match = EqualityComparer<TKey>.Default.Equals(change.Key, key);
+                                    if (match)
+                                    {
+                                        observer.OnNext(change);
+                                    }
                                 }
-                            }
-                        });
+                            });
+                    }
                 }
             });
 
     private void InvokeNext(ChangeSet<TObject, TKey> changes)
     {
-        lock (_locker)
+        lock (_notificationLocker)
         {
             // If Notifications are not suspended
             if (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended)
             {
-                // Emit the changes
-                _changes.OnNext(changes);
+                if (_isNotifying)
+                {
+                    // Re-entrant call: queue for delivery after the current OnNext completes.
+                    _pendingNotifications.Enqueue(changes);
+                }
+                else
+                {
+                    _isNotifying = true;
+                    try
+                    {
+                        _changes.OnNext(changes);
+
+                        // Drain any re-entrant notifications queued during OnNext.
+                        while (_pendingNotifications.Count > 0)
+                        {
+                            _changes.OnNext(_pendingNotifications.Dequeue());
+                        }
+                    }
+                    finally
+                    {
+                        _isNotifying = false;
+                    }
+                }
             }
             else
             {
@@ -326,7 +369,7 @@ private void InvokePreview(ChangeSet<TObject, TKey> changes)
 
     private void InvokeCountNext()
     {
-        lock (_locker)
+        lock (_notificationLocker)
         {
             if (_countChanged.IsValueCreated)
             {