Move change batching out of the delivery queue

src/DynamicData/Internal/CacheParentSubscription.cs

@@ -13,7 +13,8 @@ namespace DynamicData.Internal;
 /// when either the parent or child gets a new value.
 /// Uses a <see cref="SharedDeliveryQueue"/> for serialization and lock-free delivery.
 /// Same-thread reentrant delivery preserves child-during-parent ordering.
-/// OnDrainComplete calls EmitChanges after the outermost delivery, outside the lock.
+/// Accumulated changes are emitted once per delivery frame, where a frame is one
+/// notification plus anything delivered synchronously beneath it on the same thread.
 /// </summary>
 /// <typeparam name="TParent">Type of the Parent ChangeSet.</typeparam>
 /// <typeparam name="TKey">Type for the Parent ChangeSet Key.</typeparam>
@@ -29,6 +30,7 @@ internal abstract class CacheParentSubscription<TParent, TKey, TChild, TObserver
     private readonly SharedDeliveryQueue _queue;
     private readonly IObserver<TObserver> _observer;
     private int _subscriptionCounter = 1; // Starts at 1 for the parent subscription
+    private int _frameDepth;
     private bool _isCompleted;
     private bool _hasTerminated;
     private bool _disposedValue;
@@ -40,7 +42,7 @@ internal abstract class CacheParentSubscription<TParent, TKey, TChild, TObserver
     protected CacheParentSubscription(IObserver<TObserver> observer)
     {
         _observer = observer;
-        _queue = new SharedDeliveryQueue(onDrainComplete: OnDrainComplete);
+        _queue = new SharedDeliveryQueue();
     }
 
     /// <inheritdoc/>
@@ -76,9 +78,9 @@ protected void AddChildSubscription(IObservable<TChild> observable, TKey parentK
         disposableContainer.Disposable = observable
             .Finally(CheckCompleted)
             .SubscribeSafe(
-                onNext: val => ChildOnNext(val, parentKey),
+                onNext: val => DeliverChild(val, parentKey),
                 onError: TerminalError,
-                onCompleted: () => RemoveChildSubscription(parentKey));
+                onCompleted: () => CompleteChild(parentKey));
     }
 
     protected void RemoveChildSubscription(TKey parentKey) => _childSubscriptions.Remove(parentKey);
@@ -88,9 +90,9 @@ protected void CreateParentSubscription(IObservable<IChangeSet<TParent, TKey>> s
             source
                 .SynchronizeSafe(_queue)
                 .SubscribeSafe(
-                    onNext: ParentOnNext,
+                    onNext: DeliverParent,
                     onError: TerminalError,
-                    onCompleted: CheckCompleted);
+                    onCompleted: CompleteParent);
 
     protected virtual void Dispose(bool disposing)
     {
@@ -116,8 +118,72 @@ protected virtual void Dispose(bool disposing)
     protected IObservable<T> MakeChildObservable<T>(IObservable<T> observable) =>
         observable.SynchronizeSafe(_queue);
 
-    private void OnDrainComplete()
+    private void DeliverParent(IChangeSet<TParent, TKey> changes)
     {
+        ++_frameDepth;
+        try
+        {
+            ParentOnNext(changes);
+        }
+        finally
+        {
+            EndFrame();
+        }
+    }
+
+    private void DeliverChild(TChild child, TKey parentKey)
+    {
+        ++_frameDepth;
+        try
+        {
+            ChildOnNext(child, parentKey);
+        }
+        finally
+        {
+            EndFrame();
+        }
+    }
+
+    private void CompleteParent()
+    {
+        ++_frameDepth;
+        try
+        {
+            CheckCompleted();
+        }
+        finally
+        {
+            EndFrame();
+        }
+    }
+
+    private void CompleteChild(TKey parentKey)
+    {
+        ++_frameDepth;
+        try
+        {
+            RemoveChildSubscription(parentKey);
+        }
+        finally
+        {
+            EndFrame();
+        }
+    }
+
+    /// <summary>
+    /// Closes the current delivery frame. Deliveries nested beneath this one, which the queue
+    /// runs inline on the same thread, close their own frame first and leave the emit to the
+    /// outermost, so one upstream notification and everything it triggers synchronously produce
+    /// a single downstream changeset. No lock is needed around the depth because the queue has
+    /// already serialized delivery.
+    /// </summary>
+    private void EndFrame()
+    {
+        if (--_frameDepth != 0)
+        {
+            return;
+        }
+
         EmitChanges(_observer);
 
         if (Volatile.Read(ref _isCompleted) && !_hasTerminated)

src/DynamicData/Internal/SharedDeliveryQueue.cs

@@ -1,4 +1,4 @@
-// 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.
 
@@ -16,7 +16,6 @@ namespace DynamicData.Internal;
 internal sealed class SharedDeliveryQueue : IDisposable
 {
     private readonly List<DrainableBase> _sources = [];
-    private readonly Action? _onDrainComplete;
 
 #if NET9_0_OR_GREATER
     private readonly Lock _gate;
@@ -31,22 +30,12 @@ internal sealed class SharedDeliveryQueue : IDisposable
 
     /// <summary>Initializes a new instance of the <see cref="SharedDeliveryQueue"/> class with its own internal lock.</summary>
     public SharedDeliveryQueue()
-        : this(onDrainComplete: null)
-    {
-    }
-
-    /// <summary>
-    /// Initializes a new instance of the <see cref="SharedDeliveryQueue"/> class with its own internal lock
-    /// and a callback that fires outside the lock after each drain cycle completes.
-    /// </summary>
-    public SharedDeliveryQueue(Action? onDrainComplete)
     {
 #if NET9_0_OR_GREATER
         _gate = new Lock();
 #else
         _gate = new object();
 #endif
-        _onDrainComplete = onDrainComplete;
     }
 
 #if NET9_0_OR_GREATER
@@ -198,11 +187,6 @@ private void DrainAll()
                     return;
                 }
 
-                if (_onDrainComplete is not null)
-                {
-                    _onDrainComplete();
-                }
-
                 // Atomically check for pending items and release drain ownership
                 // if empty. This closes the TOCTOU window: if we checked and released
                 // in separate lock scopes, Thread B could enqueue between them,
@@ -212,7 +196,7 @@ private void DrainAll()
 
                 if (_activeBits.HasAny() && !_isTerminated)
                 {
-                    // Items arrived during _onDrainComplete. Loop back to drain them.
+                    // Items arrived while we were delivering. Loop back to drain them.
                     ExitLock();
                     continue;
                 }