Add read-only lock for DeliveryQueue and improve safety

Introduce ReadOnlyScopedAccess to DeliveryQueue<TItem> for safe, read-only access to queue state under lock. Update tests and ObservableCache<TObject, TKey> to use AcquireReadLock() for reading PendingCount, replacing direct property access and manual locking. Make PendingCount private and encapsulate lock release logic. Wrap _suspensionTracker disposal in a lock for thread safety. These changes improve thread safety and clarify access patterns for queue state.

Author: dwcullop

src/DynamicData.Tests/Internal/DeliveryQueueFixture.cs

@@ -316,9 +316,9 @@ public void PendingCountPreservedOnException()
 
         act.Should().Throw<InvalidOperationException>();
 
-        lock (_gate)
+        using (var rl = queue.AcquireReadLock())
         {
-            queue.PendingCount.Should().Be(1, "only the dequeued item should be decremented");
+            rl.PendingCount.Should().Be(1, "only the dequeued item should be decremented");
         }
     }
 
@@ -334,7 +334,10 @@ public void PendingCountClearedOnTermination()
             notifications.Enqueue("STOP");
         }
 
-        queue.PendingCount.Should().Be(0);
+        using (var rl = queue.AcquireReadLock())
+        {
+            rl.PendingCount.Should().Be(0);
+        }
     }
 
     // Category 6: Stress / Thread Safety

src/DynamicData/Cache/ObservableCache.cs

@@ -92,11 +92,10 @@ public ObservableCache(Func<TObject, TKey>? keySelector = null)
         Observable.Create<int>(
             observer =>
             {
-                lock (_locker)
-                {
-                    var source = _countChanged.Value.StartWith(_readerWriter.Count).DistinctUntilChanged();
-                    return source.SubscribeSafe(observer);
-                }
+                using var readLock = _notifications.AcquireReadLock();
+
+                var source = _countChanged.Value.StartWith(_readerWriter.Count).DistinctUntilChanged();
+                return source.SubscribeSafe(observer);
             });
 
     public IReadOnlyList<TObject> Items => _readerWriter.Items;
@@ -233,56 +232,54 @@ private IObservable<IChangeSet<TObject, TKey>> CreateConnectObservable(Func<TObj
         Observable.Create<IChangeSet<TObject, TKey>>(
             observer =>
             {
-                lock (_locker)
-                {
-                    // Skip pending notifications to avoid duplicating items already in the snapshot.
-                    var skipCount = _notifications.PendingCount;
+                using var readLock = _notifications.AcquireReadLock();
 
-                    var initial = InternalEx.Return(() => (IChangeSet<TObject, TKey>)GetInitialUpdates(predicate));
-                    var changesStream = skipCount > 0 ? _changes.Skip(skipCount) : _changes;
-                    var changes = initial.Concat(changesStream);
+                // Skip pending notifications to avoid duplicating items already in the snapshot.
+                var skipCount = readLock.PendingCount;
 
-                    if (predicate != null)
-                    {
-                        changes = changes.Filter(predicate, suppressEmptyChangeSets);
-                    }
-                    else if (suppressEmptyChangeSets)
-                    {
-                        changes = changes.NotEmpty();
-                    }
+                var initial = InternalEx.Return(() => (IChangeSet<TObject, TKey>)GetInitialUpdates(predicate));
+                var changesStream = skipCount > 0 ? _changes.Skip(skipCount) : _changes;
+                var changes = initial.Concat(changesStream);
 
-                    return changes.SubscribeSafe(observer);
+                if (predicate != null)
+                {
+                    changes = changes.Filter(predicate, suppressEmptyChangeSets);
+                }
+                else if (suppressEmptyChangeSets)
+                {
+                    changes = changes.NotEmpty();
                 }
+
+                return changes.SubscribeSafe(observer);
             });
 
     private IObservable<Change<TObject, TKey>> CreateWatchObservable(TKey key) =>
         Observable.Create<Change<TObject, TKey>>(
             observer =>
             {
-                lock (_locker)
+                using var readLock = _notifications.AcquireReadLock();
+
+                var skipCount = readLock.PendingCount;
+
+                var initial = _readerWriter.Lookup(key);
+                if (initial.HasValue)
                 {
-                    var skipCount = _notifications.PendingCount;
+                    observer.OnNext(new Change<TObject, TKey>(ChangeReason.Add, key, initial.Value));
+                }
 
-                    var initial = _readerWriter.Lookup(key);
-                    if (initial.HasValue)
+                var changesStream = skipCount > 0 ? _changes.Skip(skipCount) : _changes;
+                return changesStream.Finally(observer.OnCompleted).Subscribe(
+                    changes =>
                     {
-                        observer.OnNext(new Change<TObject, TKey>(ChangeReason.Add, key, initial.Value));
-                    }
-
-                    var changesStream = skipCount > 0 ? _changes.Skip(skipCount) : _changes;
-                    return changesStream.Finally(observer.OnCompleted).Subscribe(
-                        changes =>
+                        foreach (var change in changes.ToConcreteType())
                         {
-                            foreach (var change in changes.ToConcreteType())
+                            var match = EqualityComparer<TKey>.Default.Equals(change.Key, key);
+                            if (match)
                             {
-                                var match = EqualityComparer<TKey>.Default.Equals(change.Key, key);
-                                if (match)
-                                {
-                                    observer.OnNext(change);
-                                }
+                                observer.OnNext(change);
                             }
-                        });
-                }
+                        }
+                    });
             });
 
     /// <summary>
@@ -325,7 +322,10 @@ private bool DeliverNotification(NotificationItem item)
 
                 if (_suspensionTracker.IsValueCreated)
                 {
-                    _suspensionTracker.Value.Dispose();
+                    lock (_locker)
+                    {
+                        _suspensionTracker.Value.Dispose();
+                    }
                 }
                 return false;
 
@@ -340,7 +340,10 @@ private bool DeliverNotification(NotificationItem item)
 
                 if (_suspensionTracker.IsValueCreated)
                 {
-                    _suspensionTracker.Value.Dispose();
+                    lock (_locker)
+                    {
+                        _suspensionTracker.Value.Dispose();
+                    }
                 }
                 return false;
 

src/DynamicData/Internal/DeliveryQueue.cs

@@ -51,7 +51,7 @@ public DeliveryQueue(object gate, Func<TItem, bool> deliver)
     /// Gets the number of pending items enqueued with <c>countAsPending: true</c>.
     /// Must be read while the caller holds the gate.
     /// </summary>
-    public int PendingCount => _pendingCount;
+    private int PendingCount => _pendingCount;
 
     /// <summary>
     /// Acquires the gate and returns a scoped ScopedAccess for enqueueing items and
@@ -61,6 +61,13 @@ public DeliveryQueue(object gate, Func<TItem, bool> deliver)
     /// </summary>
     public ScopedAccess AcquireLock() => new(this);
 
+    /// <summary>
+    /// Acquires the gate for read-only access and returns a scoped handle.
+    /// Provides access to queue state (e.g., <see cref="PendingCount"/>) but
+    /// cannot enqueue items and does not trigger delivery on dispose.
+    /// </summary>
+    public ReadOnlyScopedAccess AcquireReadLock() => new(this);
+
     private void EnterLock()
     {
 #if NET9_0_OR_GREATER
@@ -70,6 +77,15 @@ private void EnterLock()
 #endif
     }
 
+    private void ExitLock()
+    {
+#if NET9_0_OR_GREATER
+        _gate.Exit();
+#else
+        Monitor.Exit(_gate);
+#endif
+    }
+
     private void EnqueueItem(TItem item, bool countAsPending)
     {
         if (_isTerminated)
@@ -91,11 +107,7 @@ private void ExitLockAndDeliver()
         var shouldDeliver = TryStartDelivery();
 
         // Now release the lock. We do this before delivering to allow other threads to enqueue items while delivery is in progress.
-#if NET9_0_OR_GREATER
-        _gate.Exit();
-#else
-        Monitor.Exit(_gate);
-#endif
+        ExitLock();
 
         // If this thread has been chosen to deliver, do it now that the lock is released.
         // If not, another thread is already delivering or there are no items to deliver.
@@ -219,4 +231,40 @@ public void Dispose()
             owner.ExitLockAndDeliver();
         }
     }
+
+    /// <summary>
+    /// A read-only scoped handle for reading queue state under the gate lock.
+    /// Cannot enqueue items and does not trigger delivery on dispose.
+    /// </summary>
+    public ref struct ReadOnlyScopedAccess
+    {
+        private DeliveryQueue<TItem>? _owner;
+
+        internal ReadOnlyScopedAccess(DeliveryQueue<TItem> owner)
+        {
+            _owner = owner;
+            owner.EnterLock();
+        }
+
+        /// <summary>
+        /// Gets the number of pending items that were enqueued with
+        /// <c>countAsPending: true</c> and have not yet been dequeued for delivery.
+        /// </summary>
+        public readonly int PendingCount => _owner?._pendingCount ?? 0;
+
+        /// <summary>
+        /// Releases the gate lock. Does not trigger delivery.
+        /// </summary>
+        public void Dispose()
+        {
+            var owner = _owner;
+            if (owner is null)
+            {
+                return;
+            }
+
+            _owner = null;
+            owner.ExitLock();
+        }
+    }
 }