From f691d4d67f8cf7d75188de268de92859f7ee8e8a Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sat, 4 Jul 2026 02:06:17 -0500 Subject: [PATCH 1/8] Fixed that subscriptions occurring during an `.Edit()` operation upon a `SourceCache<>` or `SourceList<>` would still publish an initial notification to the subscriber. This resulted in the potential for notifications to be duplicated, for items added or manipulated during the edit. Instead, such subscriptions now have their notifications deferred until the `.Edit()` is complete. Resolves #1129. --- .../Cache/SourceCacheFixture.cs | 53 ++++++++++++- .../List/SourceListFixture.cs | 50 ++++++++++++ src/DynamicData/Cache/ObservableCache.cs | 44 +++++++---- src/DynamicData/List/SourceList.cs | 79 +++++++++++++------ 4 files changed, 187 insertions(+), 39 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs index 99b79fbfd..24b628e32 100644 --- a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs @@ -1,12 +1,13 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; using System.Threading; using System.Threading.Tasks; using DynamicData.Tests.Domain; - +using DynamicData.Tests.Utilities; using FluentAssertions; using Xunit; @@ -354,5 +355,55 @@ public void ConnectDuringDeliveryDoesNotDuplicate() addCounts.GetValueOrDefault("k2").Should().Be(1, "k2 should appear once, not duplicated from snapshot + queued delivery"); } + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditDoesNotDuplicate() + { + using var items = new SourceCache(static item => item); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + CacheItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.AddOrUpdate(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets(static item => item) + .RecordCacheItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.AddOrUpdate(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItemsByKey.Should().BeEquivalentTo( + new Dictionary() + { + [1] = 1, + [2] = 2 + }, + options => options.WithoutStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + private sealed record TestItem(string Key, string Value); } diff --git a/src/DynamicData.Tests/List/SourceListFixture.cs b/src/DynamicData.Tests/List/SourceListFixture.cs index dc37bdaed..9323113b7 100644 --- a/src/DynamicData.Tests/List/SourceListFixture.cs +++ b/src/DynamicData.Tests/List/SourceListFixture.cs @@ -1,13 +1,63 @@ using System; using System.Collections.Generic; using System.Linq; +using System.Reactive.Disposables; + using FluentAssertions; using Xunit; +using DynamicData.Tests.Utilities; + namespace DynamicData.Tests.List; public class SourceListFixture { + // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 + [Fact] + public void ConnectDuringEditDoesNotDuplicate() + { + using var items = new SourceList(); + + using var subscriptions = new CompositeDisposable(); + + // An initial subscription is required to initiate internal buffering of changes, during the upcoming .Edit(). + // That is, we want there to be changes buffered, internally, when the mid-edit subscription comes in, to + // ensure that they don't get duplicated. This is the scenario that came in up #1129. + subscriptions.Add(items + .Connect() + .Subscribe()); + + ListItemRecordingObserver? results = null; + + items.Edit(inner => + { + inner.Add(1); + + subscriptions.Add(items + .Connect() + .ValidateChangeSets() + .RecordListItems(out results)); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + inner.Add(2); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + }); + + results.Should().NotBeNull("the edit delegate should have been invoked"); + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); + results.RecordedItems.Should().BeEquivalentTo( + new[] { 1, 2, }, + options => options.WithStrictOrdering(), + "all items in the source should have propagated downstream"); + + results.HasCompleted.Should().BeFalse("the source has not yet completed"); + } + [Fact] public void InitialChangeIsRange() { diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 60642ee29..579348805 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -6,9 +6,8 @@ using System.Reactive.Disposables; using System.Reactive.Linq; using System.Reactive.Subjects; -using System.Threading; + using DynamicData.Binding; -using DynamicData.Cache; using DynamicData.Cache.Internal; // ReSharper disable once CheckNamespace @@ -42,6 +41,9 @@ internal sealed class ObservableCache : IObservableCache _notifications; + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposed with _cleanUp")] + private readonly BehaviorSubject _isEditInProgress; + private int _editLevel; // The level of recursion in editing. private long _currentVersion; // Monotonic counter incremented under lock for each enqueued change notification. @@ -53,6 +55,7 @@ public ObservableCache(IObservable> source) _readerWriter = new ReaderWriter(); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(false); var loader = source.Subscribe( changeSet => @@ -83,6 +86,7 @@ public ObservableCache(Func? keySelector = null) _readerWriter = new ReaderWriter(keySelector); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); + _isEditInProgress = new(false); _cleanUp = Disposable.Create(NotifyCompleted); } @@ -115,17 +119,18 @@ public IObservable> Connect(Func? predi { lock (_locker) { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + && (!_isEditInProgress.Value)) // Create the Connection Observable ? CreateConnectObservable(predicate, suppressEmptyChangeSets) - // Defer until notifications are no longer suspended. Take(1) means there is only - // ever one inner sequence, so SelectMany carries the terminal event of the gate - // through on its own: the connection ends when the cache does, and fails when it - // fails, rather than reporting a failure as a successful completion. - : _suspensionTracker.Value.NotificationsSuspendedObservable - .Where(static areNotificationsSuspended => !areNotificationsSuspended) + // Defer until notifications are no longer suspended + : Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) .Take(1) .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); @@ -144,15 +149,18 @@ public IObservable> Watch(TKey key) => { lock (_locker) { - var observable = (!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) + && (!_isEditInProgress.Value)) // Create the Watch Observable ? CreateWatchObservable(key) - // Defer until notifications are no longer suspended. See Connect() for why - // SelectMany is used here. - : _suspensionTracker.Value.NotificationsSuspendedObservable - .Where(static areNotificationsSuspended => !areNotificationsSuspended) + // Defer until notifications are no longer suspended + : Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) .Take(1) .SelectMany(_ => CreateWatchObservable(key)); @@ -189,6 +197,7 @@ internal void UpdateFromIntermediate(Action> update ChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; @@ -205,6 +214,8 @@ internal void UpdateFromIntermediate(Action> update { notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } + + _isEditInProgress.OnNext(_editLevel is not 0); } internal void UpdateFromSource(Action> updateAction) @@ -216,6 +227,7 @@ internal void UpdateFromSource(Action> updateActio ChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; @@ -232,6 +244,8 @@ internal void UpdateFromSource(Action> updateActio { notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } + + _isEditInProgress.OnNext(_editLevel is not 0); } private IObservable> CreateConnectObservable(Func? predicate, bool suppressEmptyChangeSets) => @@ -394,6 +408,7 @@ public void OnError(Exception error) { cache._changesPreview.OnError(error); cache._changes.OnError(error); + cache._isEditInProgress.OnError(error); if (cache._countChanged.IsValueCreated) { @@ -410,6 +425,7 @@ public void OnCompleted() { cache._changes.OnCompleted(); cache._changesPreview.OnCompleted(); + cache._isEditInProgress.OnCompleted(); if (cache._countChanged.IsValueCreated) { diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 3db42bc11..12037d02b 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -36,6 +36,9 @@ public sealed class SourceList : ISourceList private readonly ReaderWriter _readerWriter = new(); + [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposal is superfluous after completion, and causes a bunch of test failures")] + private readonly BehaviorSubject _isEditInProgress; + private int _editLevel; /// @@ -44,6 +47,8 @@ public sealed class SourceList : ISourceList /// The source. public SourceList(IObservable>? source = null) { + _isEditInProgress = new(false); + var loader = source is null ? Disposable.Empty : LoadFromSource(source); _cleanUp = Disposable.Create( @@ -78,40 +83,31 @@ public SourceList(IObservable>? source = null) /// public IObservable> Connect(Func? predicate = null) - { - var observable = Observable.Create>( - observer => + => Observable.Create>(observer => + { + lock (_locker) { - lock (_locker) - { - if (_readerWriter.Items.Length > 0) - { - observer.OnNext( - new ChangeSet - { - new(ListChangeReason.AddRange, _readerWriter.Items, 0) - }); - } + var observable = !_isEditInProgress.Value - var source = _changes.Finally(observer.OnCompleted); + // Create the Connection Observable + ? CreateConnectObservable(predicate) - return source.SubscribeSafe(observer); - } - }); - - if (predicate is not null) - { - observable = new FilterStatic(observable, predicate).Run(); - } + // Defer until notifications are no longer suspended + : _isEditInProgress + .Where(static isEditInProgress => !isEditInProgress) + .Take(1) + .SelectMany(_ => CreateConnectObservable(predicate)); - return observable; - } + return observable.SubscribeSafe(observer); + } + }); /// public void Dispose() { _cleanUp.Dispose(); _changesPreview.Dispose(); + _isEditInProgress.OnCompleted(); } /// @@ -124,6 +120,7 @@ public void Edit(Action> updateAction) IChangeSet? changes = null; _editLevel++; + _isEditInProgress.OnNext(_editLevel is not 0); if (_editLevel == 1) { @@ -140,6 +137,8 @@ public void Edit(Action> updateAction) { InvokeNext(changes); } + + _isEditInProgress.OnNext(_editLevel is not 0); } } @@ -156,6 +155,36 @@ public IObservable> Preview(Func? predicate = null) return observable; } + private IObservable> CreateConnectObservable(Func? predicate) + { + var observable = Observable.Create>( + observer => + { + lock (_locker) + { + if (_readerWriter.Items.Length > 0) + { + observer.OnNext( + new ChangeSet + { + new(ListChangeReason.AddRange, _readerWriter.Items, 0) + }); + } + + var source = _changes.Finally(observer.OnCompleted); + + return source.SubscribeSafe(observer); + } + }); + + if (predicate is not null) + { + observable = new FilterStatic(observable, predicate).Run(); + } + + return observable; + } + private void InvokeNext(IChangeSet changes) { if (changes.Count == 0) @@ -195,6 +224,7 @@ private void OnCompleted() { _changesPreview.OnCompleted(); _changes.OnCompleted(); + _isEditInProgress.OnCompleted(); } } @@ -204,6 +234,7 @@ private void OnError(Exception exception) { _changesPreview.OnError(exception); _changes.OnError(exception); + _isEditInProgress.OnError(exception); } } } From 742b99f62254c7f6afde0ba0b61c9e20e5012c24 Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sun, 26 Jul 2026 04:44:17 -0500 Subject: [PATCH 2/8] Added exception guarantees to edit level tracking, to ensure that tracking doesn't fall apart if user-injected code throws. --- .../Cache/SourceCacheFixture.cs | 20 ++++++- .../List/SourceListFixture.cs | 17 ++++++ src/DynamicData/Cache/ObservableCache.cs | 58 +++++++++++-------- src/DynamicData/List/SourceList.cs | 37 +++++++----- 4 files changed, 91 insertions(+), 41 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs index 24b628e32..4cfea0ba9 100644 --- a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs @@ -163,8 +163,6 @@ public void EmptyChangesWithFilter() change!.Count.Should().Be(0); } - - [Fact] public void StaticFilterRemove() { @@ -192,7 +190,6 @@ public void StaticFilterRemove() public record class SomeObject(int Id, int Value); - [Fact] public async Task MultiCacheFanInDoesNotDeadlock() { @@ -405,5 +402,22 @@ public void ConnectDuringEditDoesNotDuplicate() results.HasCompleted.Should().BeFalse("the source has not yet completed"); } + [Fact] + public void ConnectContinuesToWorkNormallyAfterAFailedEdit() + { + using var source = new SourceCache(static item => item); + + source.AddOrUpdate(1); + + source.Invoking(source => source.Edit(_ => throw new Exception("Test"))) + .Should().Throw() + .WithMessage("Test"); + + using var subscription = source.Connect().RecordCacheItems(out var results); + + results.Error.Should().BeNull("new subscribers should not receive previous errors"); + results.RecordedChangeSets.Should().ContainSingle("an initial changeset should have been published."); + } + private sealed record TestItem(string Key, string Value); } diff --git a/src/DynamicData.Tests/List/SourceListFixture.cs b/src/DynamicData.Tests/List/SourceListFixture.cs index 9323113b7..06905c5a4 100644 --- a/src/DynamicData.Tests/List/SourceListFixture.cs +++ b/src/DynamicData.Tests/List/SourceListFixture.cs @@ -71,4 +71,21 @@ public void InitialChangeIsRange() changeSets[0].First().Type.Should().Be(ChangeType.Range); changeSets[0].First().Range.Index.Should().Be(0); } + + [Fact] + public void ConnectContinuesToWorkNormallyAfterAFailedEdit() + { + using var source = new SourceList(); + + source.Add(1); + + source.Invoking(source => source.Edit(_ => throw new Exception("Test"))) + .Should().Throw() + .WithMessage("Test"); + + using var subscription = source.Connect().RecordListItems(out var results); + + results.Error.Should().BeNull("new subscribers should not receive previous errors"); + results.RecordedChangeSets.Should().ContainSingle("an initial changeset should have been published."); + } } diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 579348805..57d9c245b 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -198,24 +198,29 @@ internal void UpdateFromIntermediate(Action> update _editLevel++; _isEditInProgress.OnNext(_editLevel is not 0); - if (_editLevel == 1) + try { - var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; - changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + if (_editLevel == 1) + { + var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; + changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + } + else + { + _readerWriter.WriteNested(updateAction); + } } - else + finally { - _readerWriter.WriteNested(updateAction); - } + _editLevel--; - _editLevel--; + if (changes is not null && _editLevel == 0) + { + notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + } - if (changes is not null && _editLevel == 0) - { - notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + _isEditInProgress.OnNext(_editLevel is not 0); } - - _isEditInProgress.OnNext(_editLevel is not 0); } internal void UpdateFromSource(Action> updateAction) @@ -228,24 +233,29 @@ internal void UpdateFromSource(Action> updateActio _editLevel++; _isEditInProgress.OnNext(_editLevel is not 0); - if (_editLevel == 1) + try { - var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; - changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + if (_editLevel == 1) + { + var previewHandler = _changesPreview.HasObservers ? (Action>)InvokePreview : null; + changes = _readerWriter.Write(updateAction, previewHandler, _changes.HasObservers); + } + else + { + _readerWriter.WriteNested(updateAction); + } } - else + finally { - _readerWriter.WriteNested(updateAction); - } + _editLevel--; - _editLevel--; + if (changes is not null && _editLevel == 0) + { + notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + } - if (changes is not null && _editLevel == 0) - { - notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); + _isEditInProgress.OnNext(_editLevel is not 0); } - - _isEditInProgress.OnNext(_editLevel is not 0); } private IObservable> CreateConnectObservable(Func? predicate, bool suppressEmptyChangeSets) => diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 12037d02b..2ff9c3f4e 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -121,24 +121,33 @@ public void Edit(Action> updateAction) _editLevel++; _isEditInProgress.OnNext(_editLevel is not 0); - - if (_editLevel == 1) - { - changes = _changesPreview.HasObservers ? _readerWriter.WriteWithPreview(updateAction, InvokeNextPreview) : _readerWriter.Write(updateAction); - } - else + try { - _readerWriter.WriteNested(updateAction); - } - - _editLevel--; + try + { + if (_editLevel == 1) + { + changes = _changesPreview.HasObservers ? _readerWriter.WriteWithPreview(updateAction, InvokeNextPreview) : _readerWriter.Write(updateAction); + } + else + { + _readerWriter.WriteNested(updateAction); + } + } + finally + { + _editLevel--; + } - if (changes is not null && _editLevel == 0) + if (changes is not null && (_editLevel is 0)) + { + InvokeNext(changes); + } + } + finally { - InvokeNext(changes); + _isEditInProgress.OnNext(_editLevel is not 0); } - - _isEditInProgress.OnNext(_editLevel is not 0); } } From 95e2d32967bec2dcec2edb6f0f85f6280c533633 Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Sun, 26 Jul 2026 04:46:54 -0500 Subject: [PATCH 3/8] Adjusted `.Connect()` to ensure that the internal system to support `.SuspendNotifications()` does not initialize unless necessary. --- .../Miscellaneous/LockImplementations.cs | 42 +++++++++++++++++++ src/DynamicData/Cache/ObservableCache.cs | 35 ++++++++++++---- 2 files changed, 69 insertions(+), 8 deletions(-) create mode 100644 src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs diff --git a/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs b/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs new file mode 100644 index 000000000..cce51279f --- /dev/null +++ b/src/DynamicData.Benchmarks/Miscellaneous/LockImplementations.cs @@ -0,0 +1,42 @@ +using System.Threading; +using BenchmarkDotNet.Attributes; + +namespace DynamicData.Benchmarks.Miscellaneous; + +[MemoryDiagnoser] +[MarkdownExporterAttribute.GitHub] +public class LockImplementations +{ + public LockImplementations() + { + _objectGate = new(); + _threadingGate = new(); + } + + [Benchmark(Baseline = true)] + public int NoLock() + { + return 0; + } + + [Benchmark] + public int ObjectLock() + { + lock (_objectGate) + { + return 0; + } + } + + [Benchmark] + public int ThreadingLock() + { + lock (_threadingGate) + { + return 0; + } + } + + private readonly object _objectGate; + private readonly Lock _threadingGate; +} diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 57d9c245b..6b76034aa 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -119,22 +119,41 @@ public IObservable> Connect(Func? predi { lock (_locker) { - var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) - && (!_isEditInProgress.Value)) + var observable = ( + _suspensionTracker.IsValueCreated, + _isEditInProgress.Value) + switch + { + // No suspensions or edits active, create the connection immediately. + (false, false) => CreateConnectObservable(predicate, suppressEmptyChangeSets), + + // Edit in progress, but suspension system is inactive + // Need to avoid activating the suspension system if we don't absolutely need to, as it adds + // locking overhead to edit operations. But then when the edit is done, we need to check if a + // suspension came in. If so, do a followup wait with the full logic for both systems. It + // needs to be the full logic, in case an edit comes in during the suspension, and so on. + (false, true) => _isEditInProgress + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) + .Take(1) + .Select(_ => CreateFullyDeferredConnection()) + .Switch(), + + // If the suspension system is already active, we can monitor both systems simultaneously, and + // make the connection as soon as both are idle at the same time. + _ => CreateFullyDeferredConnection() + }; - // Create the Connection Observable - ? CreateConnectObservable(predicate, suppressEmptyChangeSets) + return observable.SubscribeSafe(observer); - // Defer until notifications are no longer suspended - : Observable.CombineLatest( + IObservable> CreateFullyDeferredConnection() + => Observable.CombineLatest( _suspensionTracker.Value.NotificationsSuspendedObservable, _isEditInProgress, static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Do(static _ => { }, observer.OnCompleted) .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) .Take(1) .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); - - return observable.SubscribeSafe(observer); } }); From 6cbe1c95e0bbef7567bd930cc9de69c819d5d91a Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Mon, 27 Jul 2026 01:57:51 -0500 Subject: [PATCH 4/8] Extended the strategy of lazy-initialization for the suspension system to the edit-tracking system as well. Also extended the proper strategy for lazy-initialization of both systems from `.Connect()` to `.Watch()`. --- src/DynamicData/Cache/ObservableCache.cs | 160 ++++++++++++----------- 1 file changed, 86 insertions(+), 74 deletions(-) diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 6b76034aa..8c96d063e 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -42,7 +42,7 @@ internal sealed class ObservableCache : IObservableCache _notifications; [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposed with _cleanUp")] - private readonly BehaviorSubject _isEditInProgress; + private readonly Lazy> _isEditInProgress; private int _editLevel; // The level of recursion in editing. @@ -55,7 +55,7 @@ public ObservableCache(IObservable> source) _readerWriter = new ReaderWriter(); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); - _isEditInProgress = new(false); + _isEditInProgress = new(() => new(_editLevel is not 0)); var loader = source.Subscribe( changeSet => @@ -86,7 +86,7 @@ public ObservableCache(Func? keySelector = null) _readerWriter = new ReaderWriter(keySelector); _notifications = new DeliveryQueue(_locker, new CacheUpdateObserver(this)); _suspensionTracker = new(() => new SuspensionTracker()); - _isEditInProgress = new(false); + _isEditInProgress = new(() => new(_editLevel is not 0)); _cleanUp = Disposable.Create(NotifyCompleted); } @@ -114,48 +114,10 @@ public ObservableCache(Func? keySelector = null) public IReadOnlyDictionary KeyValues => _readerWriter.KeyValues; - public IObservable> Connect(Func? predicate = null, bool suppressEmptyChangeSets = true) => - Observable.Create>(observer => - { - lock (_locker) - { - var observable = ( - _suspensionTracker.IsValueCreated, - _isEditInProgress.Value) - switch - { - // No suspensions or edits active, create the connection immediately. - (false, false) => CreateConnectObservable(predicate, suppressEmptyChangeSets), - - // Edit in progress, but suspension system is inactive - // Need to avoid activating the suspension system if we don't absolutely need to, as it adds - // locking overhead to edit operations. But then when the edit is done, we need to check if a - // suspension came in. If so, do a followup wait with the full logic for both systems. It - // needs to be the full logic, in case an edit comes in during the suspension, and so on. - (false, true) => _isEditInProgress - .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) - .Take(1) - .Select(_ => CreateFullyDeferredConnection()) - .Switch(), - - // If the suspension system is already active, we can monitor both systems simultaneously, and - // make the connection as soon as both are idle at the same time. - _ => CreateFullyDeferredConnection() - }; - - return observable.SubscribeSafe(observer); - - IObservable> CreateFullyDeferredConnection() - => Observable.CombineLatest( - _suspensionTracker.Value.NotificationsSuspendedObservable, - _isEditInProgress, - static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) - .Do(static _ => { }, observer.OnCompleted) - .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) - .Take(1) - .SelectMany(_ => CreateConnectObservable(predicate, suppressEmptyChangeSets)); - } - }); + public IObservable> Connect( + Func? predicate = null, + bool suppressEmptyChangeSets = true) + => CreateDeferredNotificationObservable(() => CreateConnectObservable(predicate, suppressEmptyChangeSets)); public void Dispose() => _cleanUp.Dispose(); @@ -163,29 +125,8 @@ IObservable> CreateFullyDeferredConnection() public IObservable> Preview(Func? predicate = null) => predicate is null ? _changesPreview : _changesPreview.Filter(predicate); - public IObservable> Watch(TKey key) => - Observable.Create>(observer => - { - lock (_locker) - { - var observable = ((!_suspensionTracker.IsValueCreated || !_suspensionTracker.Value.AreNotificationsSuspended) - && (!_isEditInProgress.Value)) - - // Create the Watch Observable - ? CreateWatchObservable(key) - - // Defer until notifications are no longer suspended - : Observable.CombineLatest( - _suspensionTracker.Value.NotificationsSuspendedObservable, - _isEditInProgress, - static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) - .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) - .Take(1) - .SelectMany(_ => CreateWatchObservable(key)); - - return observable.SubscribeSafe(observer); - } - }); + public IObservable> Watch(TKey key) + => CreateDeferredNotificationObservable(() => CreateWatchObservable(key)); public IDisposable SuspendCount() { @@ -216,7 +157,8 @@ internal void UpdateFromIntermediate(Action> update ChangeSet? changes = null; _editLevel++; - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnNext(_editLevel is not 0); try { if (_editLevel == 1) @@ -238,7 +180,8 @@ internal void UpdateFromIntermediate(Action> update notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnNext(_editLevel is not 0); } } @@ -251,7 +194,8 @@ internal void UpdateFromSource(Action> updateActio ChangeSet? changes = null; _editLevel++; - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnNext(_editLevel is not 0); try { if (_editLevel == 1) @@ -273,7 +217,8 @@ internal void UpdateFromSource(Action> updateActio notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnNext(_editLevel is not 0); } } @@ -307,6 +252,65 @@ private IObservable> CreateConnectObservable(Func CreateDeferredNotificationObservable(Func> factory) + => Observable.Create(observer => + { + lock (_locker) + { + var observable = ( + // A suspension can't be in-progress if the suspension system hasn't been activated. + _suspensionTracker.IsValueCreated, + // An edit can be in-progress before the edit-tracking notification system is activated, so this + // one needs an extra check. + _isEditInProgress.IsValueCreated || (_editLevel is not 0)) + switch + { + // Neither the suspension system nor the edit system is active, create the connection + // immediately. + (false, false) => factory.Invoke(), + + // Edit system is active, suspension system isn't + // Need to avoid activating the suspension system if we don't absolutely need to, as it adds + // locking overhead to edit operations. But then when the edit is done, we need to check if a + // suspension came in. If so, do a followup wait with the full logic for both systems. It + // needs to be the full logic, in case an edit comes in during the suspension, and so on. + (false, true) => _isEditInProgress.Value + .Where(static isEditInProgress => !isEditInProgress) + .Take(1) + .SelectMany(_ => (_suspensionTracker.IsValueCreated && _suspensionTracker.Value.AreNotificationsSuspended) + ? CreateFullyDeferredConnection() + : factory.Invoke()), + + // Suspension system is active, edit system isn't + // Same case as above, but reversed. Just wait on the suspension system, but then do a followup + // wait if needed. + (true, false) => _suspensionTracker.Value.NotificationsSuspendedObservable + .Where(static isSuspensionInProgress => !isSuspensionInProgress) + .Take(1) + .SelectMany(_ => (_isEditInProgress.IsValueCreated && _isEditInProgress.Value.Value) + ? CreateFullyDeferredConnection() + : factory.Invoke()), + + // If both systems are already active, we can monitor both systems simultaneously, and make the + // connection as soon as both are idle at the same time. + _ => CreateFullyDeferredConnection() + }; + + return observable.SubscribeSafe(observer); + + IObservable CreateFullyDeferredConnection() + => Observable.CombineLatest( + _suspensionTracker.Value.NotificationsSuspendedObservable, + _isEditInProgress.Value, + static (areNotificationsSuspended, isEditInProgress) => areNotificationsSuspended || isEditInProgress) + .Do(static _ => { }, observer.OnCompleted) + .Where(static shouldConnectionBeDeferred => !shouldConnectionBeDeferred) + .Take(1) + .Select(_ => factory.Invoke()) + .Switch(); + } + }); + private IObservable> CreateWatchObservable(TKey key) => Observable.Create>( observer => @@ -437,7 +441,11 @@ public void OnError(Exception error) { cache._changesPreview.OnError(error); cache._changes.OnError(error); - cache._isEditInProgress.OnError(error); + + if (cache._isEditInProgress.IsValueCreated) + { + cache._isEditInProgress.Value.OnError(error); + } if (cache._countChanged.IsValueCreated) { @@ -454,7 +462,11 @@ public void OnCompleted() { cache._changes.OnCompleted(); cache._changesPreview.OnCompleted(); - cache._isEditInProgress.OnCompleted(); + + if (cache._isEditInProgress.IsValueCreated) + { + cache._isEditInProgress.Value.OnCompleted(); + } if (cache._countChanged.IsValueCreated) { From 41c0dbe9f0cd4c01849d28cd0214ee0cc393649b Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Mon, 27 Jul 2026 02:28:39 -0500 Subject: [PATCH 5/8] Added logic to avoid redundant `_isEditInProgress` notifications. --- src/DynamicData/Cache/ObservableCache.cs | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/src/DynamicData/Cache/ObservableCache.cs b/src/DynamicData/Cache/ObservableCache.cs index 8c96d063e..745e06ba1 100644 --- a/src/DynamicData/Cache/ObservableCache.cs +++ b/src/DynamicData/Cache/ObservableCache.cs @@ -157,8 +157,8 @@ internal void UpdateFromIntermediate(Action> update ChangeSet? changes = null; _editLevel++; - if (_isEditInProgress.IsValueCreated) - _isEditInProgress.Value.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); try { if (_editLevel == 1) @@ -180,8 +180,8 @@ internal void UpdateFromIntermediate(Action> update notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } - if (_isEditInProgress.IsValueCreated) - _isEditInProgress.Value.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } @@ -194,8 +194,8 @@ internal void UpdateFromSource(Action> updateActio ChangeSet? changes = null; _editLevel++; - if (_isEditInProgress.IsValueCreated) - _isEditInProgress.Value.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); try { if (_editLevel == 1) @@ -217,8 +217,8 @@ internal void UpdateFromSource(Action> updateActio notifications.EnqueueNext(new CacheUpdate(changes, _readerWriter.Count, ++_currentVersion)); } - if (_isEditInProgress.IsValueCreated) - _isEditInProgress.Value.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } From 597f8d469ca72b0ac91a00803ae041a4a1ba641e Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Mon, 27 Jul 2026 02:37:46 -0500 Subject: [PATCH 6/8] Removed redundant completion signal on _isEditInProgress, within SourceList, with some context clarification instead. --- src/DynamicData/List/SourceList.cs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 2ff9c3f4e..24df26eea 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -107,7 +107,10 @@ public void Dispose() { _cleanUp.Dispose(); _changesPreview.Dispose(); - _isEditInProgress.OnCompleted(); + // Intentionally skipping disposal for _isEditInProgress, as it's technically redundant after _cleanUp.Dispose() + // calls .OnCompleted(), and doing disposal anyway causes a whole bunch of test failures. That really suggests + // we need to rework the lifecycle mechanics of this class, as a whole, but that's probably going to involve + // breaking changes. } /// From 1c5ef790a5b1a6c05fcf926ee07248632ca917d4 Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Mon, 27 Jul 2026 02:50:37 -0500 Subject: [PATCH 7/8] Adjusted the new test for covering #1129, to also exercise removals and nested edits. --- src/DynamicData.Tests/Cache/SourceCacheFixture.cs | 15 +++++++++------ 1 file changed, 9 insertions(+), 6 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs index 4cfea0ba9..38fc02e73 100644 --- a/src/DynamicData.Tests/Cache/SourceCacheFixture.cs +++ b/src/DynamicData.Tests/Cache/SourceCacheFixture.cs @@ -354,7 +354,7 @@ public void ConnectDuringDeliveryDoesNotDuplicate() // Covers https://github.com/reactivemarbles/DynamicData/issues/1129 [Fact] - public void ConnectDuringEditDoesNotDuplicate() + public void ConnectDuringEditsDoesNotDuplicate() { using var items = new SourceCache(static item => item); @@ -385,17 +385,20 @@ public void ConnectDuringEditDoesNotDuplicate() results.Error.Should().BeNull("no errors should have occurred"); results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); + + // Explicitly doing a nested edit, as that system is closely intertwined with the edit-tracking system that + // .Connect() uses. + items.Remove(item: 1); + + results.Error.Should().BeNull("no errors should have occurred"); + results.RecordedChangeSets.Should().BeEmpty("no changes should be published in the middle of an edit"); }); results.Should().NotBeNull("the edit delegate should have been invoked"); results.Error.Should().BeNull("no errors should have occurred"); results.RecordedChangeSets.Should().ContainSingle("subscribers should only receive a single initial changeset"); results.RecordedItemsByKey.Should().BeEquivalentTo( - new Dictionary() - { - [1] = 1, - [2] = 2 - }, + new Dictionary() { [2] = 2 }, options => options.WithoutStrictOrdering(), "all items in the source should have propagated downstream"); From f130a97f88a3420ad0d8f98d7e544178c6c41dfe Mon Sep 17 00:00:00 2001 From: Jake Meiergerd Date: Thu, 30 Jul 2026 01:28:56 -0500 Subject: [PATCH 8/8] Extended the strategy of lazy-initialization for the edit-tracking system in `ObservableCache` to the edit-tracking system in `SourceList`. --- src/DynamicData/List/SourceList.cs | 30 +++++++++++++++++------------- 1 file changed, 17 insertions(+), 13 deletions(-) diff --git a/src/DynamicData/List/SourceList.cs b/src/DynamicData/List/SourceList.cs index 24df26eea..a159af490 100644 --- a/src/DynamicData/List/SourceList.cs +++ b/src/DynamicData/List/SourceList.cs @@ -37,7 +37,7 @@ public sealed class SourceList : ISourceList private readonly ReaderWriter _readerWriter = new(); [System.Diagnostics.CodeAnalysis.SuppressMessage("Usage", "CA2213:Disposable fields should be disposed", Justification = "Disposal is superfluous after completion, and causes a bunch of test failures")] - private readonly BehaviorSubject _isEditInProgress; + private readonly Lazy> _isEditInProgress; private int _editLevel; @@ -47,7 +47,7 @@ public sealed class SourceList : ISourceList /// The source. public SourceList(IObservable>? source = null) { - _isEditInProgress = new(false); + _isEditInProgress = new(() => new(_editLevel is not 0)); var loader = source is null ? Disposable.Empty : LoadFromSource(source); @@ -87,16 +87,16 @@ public IObservable> Connect(Func? predicate = null) { lock (_locker) { - var observable = !_isEditInProgress.Value + var observable = _isEditInProgress.IsValueCreated || (_editLevel is not 0) - // Create the Connection Observable - ? CreateConnectObservable(predicate) - - // Defer until notifications are no longer suspended - : _isEditInProgress + // Defer connection until there is no longer an in-progress edit. + ? _isEditInProgress.Value .Where(static isEditInProgress => !isEditInProgress) .Take(1) - .SelectMany(_ => CreateConnectObservable(predicate)); + .SelectMany(_ => CreateConnectObservable(predicate)) + + // Otherwise, just connect immediately, and avoid forcing the edit-tracking system to initialize. + : CreateConnectObservable(predicate); return observable.SubscribeSafe(observer); } @@ -123,7 +123,8 @@ public void Edit(Action> updateAction) IChangeSet? changes = null; _editLevel++; - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 1)) + _isEditInProgress.Value.OnNext(true); try { try @@ -149,7 +150,8 @@ public void Edit(Action> updateAction) } finally { - _isEditInProgress.OnNext(_editLevel is not 0); + if (_isEditInProgress.IsValueCreated && (_editLevel is 0)) + _isEditInProgress.Value.OnNext(false); } } } @@ -236,7 +238,8 @@ private void OnCompleted() { _changesPreview.OnCompleted(); _changes.OnCompleted(); - _isEditInProgress.OnCompleted(); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnCompleted(); } } @@ -246,7 +249,8 @@ private void OnError(Exception exception) { _changesPreview.OnError(exception); _changes.OnError(exception); - _isEditInProgress.OnError(exception); + if (_isEditInProgress.IsValueCreated) + _isEditInProgress.Value.OnError(exception); } } }