fix: Replace lock-during-notification with queue-based drain to prevent cross-cache deadlock

The original code held _locker while calling _changes.OnNext(), so subscriber
callbacks that propagated to other caches created ABBA deadlocks when concurrent
writes were happening on those caches.

New design:
- Single _locker protects mutation and queue state
- Write paths: lock, mutate, enqueue changeset, release lock, then drain
- DrainOutsideLock delivers notifications with no lock held
- _isDraining flag ensures only one thread drains at a time, preserving
  Rx serialization contract
- Re-entrant writes enqueue and return; the outer drain loop delivers
  them sequentially
- Connect/Watch/CountChanged use Skip(pendingCount) to avoid duplicating
  items already in the snapshot, with no delivery under lock
- Terminal events (OnCompleted/OnError) routed through drain queue
- Preview remains synchronous under _locker (required by ReaderWriter)
- Suspension state captured at enqueue time; re-checked at delivery
- try/catch resets _isDraining on exception
- volatile _isTerminated prevents post-dispose delivery

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,43 @@ public void StaticFilterRemove()
 
     public record class SomeObject(int Id, int Value);
 
+
+    [Fact]
+    public async Task ConcurrentEditsShouldNotDeadlockWithSubscribersThatModifyOtherCaches()
+    {
+        const int itemCount = 100;
+
+        using var cacheA = new SourceCache<TestItem, string>(static x => x.Key);
+        using var cacheB = new SourceCache<TestItem, string>(static x => x.Key);
+        using var destination = new SourceCache<TestItem, string>(static x => x.Key);
+        using var subA = cacheA.Connect().PopulateInto(destination);
+        using var subB = cacheB.Connect().PopulateInto(destination);
+        using var results = destination.Connect().AsAggregator();
+
+        var taskA = Task.Run(() =>
+        {
+            for (var i = 0; i < itemCount; i++)
+            {
+                cacheA.AddOrUpdate(new TestItem($"a-{i}", $"ValueA-{i}"));
+            }
+        });
+
+        var taskB = Task.Run(() =>
+        {
+            for (var i = 0; i < itemCount; i++)
+            {
+                cacheB.AddOrUpdate(new TestItem($"b-{i}", $"ValueB-{i}"));
+            }
+        });
+
+        var completed = Task.WhenAll(taskA, taskB);
+        var finished = await Task.WhenAny(completed, Task.Delay(TimeSpan.FromSeconds(10)));
+
+        finished.Should().BeSameAs(completed, "concurrent edits with cross-cache subscribers should not deadlock");
+        results.Error.Should().BeNull();
+        results.Data.Count.Should().Be(itemCount * 2, "all items from both caches should arrive in the destination");
+        results.Data.Items.Should().BeEquivalentTo([.. cacheA.Items, .. cacheB.Items], "all items should be in the destination");
+    }
+
+    private sealed record TestItem(string Key, string Value);
 }