From 638454a8342f0a5a0ea2e657f5c043acb81a2161 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 08:53:26 -0700 Subject: [PATCH 1/4] Fix that the cache Switch operator never completes, without taking a lock Switch could never complete, could retain data from a source it had already switched away from, and routed delivery through a lock. It relayed changes through a private LockFreeObservableCache. That cache only ends when it is disposed, so the terminal event of the source had nowhere to go. Errors were carried across by hand through a merged subject; completion had no equivalent path and was silently dropped. A consumer never received OnCompleted, ever. Switching is now explicit rather than delegated to Observable.Switch, which holds its gate for the whole of the downstream OnNext call. A pipeline that crosses into another cache runs that work under the gate, and a producer on another thread blocks behind it, which is the cross cache deadlock shape the delivery queue exists to avoid. Measured against a subscriber that blocks inside OnNext, writing to the source from another thread took 748ms through Observable.Switch and 1ms through the queue, which enqueues and returns. A SerialDisposable holds the current inner subscription, and each one carries an identity, so a superseded source still delivering concurrently is dropped rather than applied on top of the state its replacement has already established. That is the second defect above. All state changes and all delivery happen through the queue lock, which is released before anything is handed downstream. The result completes once the sources and the current inner have both completed, and fails as soon as either does. --- .../SuspendNotificationsFixture.UnitTests.cs | 64 +++++++- src/DynamicData.Tests/Cache/SwitchFixture.cs | 143 ++++++++++++++++++ src/DynamicData/Cache/Internal/Switch.cs | 114 ++++++++++---- 3 files changed, 294 insertions(+), 27 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs index ddc9d6c07..9b6372476 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -1,7 +1,8 @@ -using System; +using System; using System.Collections.Generic; using System.Linq; using System.Reactive.Linq; +using System.Reactive.Subjects; using System.Threading.Tasks; using FluentAssertions; @@ -453,6 +454,67 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch() results.IsCompleted.Should().BeFalse(); } + [Fact] + public void OnCompletedFiresIfCacheDisposedAfterConnectingWhileSuspended() + { + // A connection made while suspended is deferred until the suspension lifts. Once it + // activates it must behave exactly like any other subscriber, including terminating + // when the source does. Previously the deferral dropped the completion, leaving the + // subscriber alive forever. + var suspend = _source.SuspendNotifications(); + using var results = _source.Connect().AsAggregator(); + Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); + + // Act + suspend.Dispose(); + _source.AddOrUpdate(1000); + _source.Dispose(); + + // Assert + results.IsCompleted.Should().BeTrue("a connection deferred by a suspension should still complete when the source does"); + results.Error.Should().BeNull("no error should have occurred"); + results.Data.Count.Should().Be(38, "all data written before disposal should have arrived"); + } + + [Fact] + public void OnErrorFiresIfCacheFailsAfterConnectingWhileSuspended() + { + // The same applies to the error case: a deferred connection that has activated must + // still see a failure of the source. + using var source = new Subject>(); + using var cache = new ObservableCache(source); + + var suspend = cache.SuspendNotifications(); + using var results = cache.Connect().AsAggregator(); + source.OnNext(new ChangeSet { new(ChangeReason.Add, 1, 1) }); + + // Act + suspend.Dispose(); + var expectedError = new Exception("Test Exception"); + source.OnError(expectedError); + + // Assert + results.Error.Should().Be(expectedError, "a connection deferred by a suspension should still see the source fail"); + results.Data.Count.Should().Be(1, "the data written before the failure should have arrived"); + } + + [Fact] + public void OnCompletedFiresIfCacheDisposedAfterWatchingWhileSuspended() + { + // Watch() defers the same way Connect() does, and has the same obligation. + var suspend = _source.SuspendNotifications(); + var isCompleted = false; + using var subscription = _source.Watch(1).Subscribe(static _ => { }, () => isCompleted = true); + _source.AddOrUpdate(1); + + // Act + suspend.Dispose(); + _source.Dispose(); + + // Assert + isCompleted.Should().BeTrue("a watch deferred by a suspension should still complete when the source does"); + } + public void Dispose() { _source.Dispose(); diff --git a/src/DynamicData.Tests/Cache/SwitchFixture.cs b/src/DynamicData.Tests/Cache/SwitchFixture.cs index 5a5b76555..a0a6cd83b 100644 --- a/src/DynamicData.Tests/Cache/SwitchFixture.cs +++ b/src/DynamicData.Tests/Cache/SwitchFixture.cs @@ -1,6 +1,8 @@ using System; using System.Linq; +using System.Reactive.Linq; using System.Reactive.Subjects; +using System.Threading; using DynamicData.Tests.Domain; @@ -89,4 +91,145 @@ public void PropagatesInnerErrors() results.Error.Should().Be(error); } + + [Fact] + public void CompletesWhenSourcesAndInnerComplete() + { + using var source = new SourceCache(p => p.Name); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + source.AddOrUpdate(Enumerable.Range(1, 100).Select(i => new Person("Person" + i, i)).ToArray()); + + switchable.OnCompleted(); + results.IsCompleted.Should().BeFalse("the inner sequence is still running"); + + source.Dispose(); + + results.IsCompleted.Should().BeTrue("both the sources and the inner sequence have completed"); + results.Error.Should().BeNull(); + results.Data.Count.Should().Be(100, "all data should have been received before completion"); + } + + [Fact] + public void DoesNotCompleteWhileInnerIsStillRunning() + { + using var source = new SourceCache(p => p.Name); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + switchable.OnCompleted(); + source.AddOrUpdate(new Person("Person1", 1)); + + results.IsCompleted.Should().BeFalse("the inner sequence has not completed"); + results.Data.Count.Should().Be(1, "changes should still flow after the sources sequence completes"); + } + + [Fact] + public void DoesNotCompleteWhenOnlyASupersededInnerCompletes() + { + using var first = new SourceCache(p => p.Name); + using var second = new SourceCache(p => p.Name); + using var switchable = new BehaviorSubject>>(first.Connect()); + using var results = switchable.Switch().AsAggregator(); + + switchable.OnNext(second.Connect()); + switchable.OnCompleted(); + + first.Dispose(); + + results.IsCompleted.Should().BeFalse("the superseded sequence is not the current one"); + + second.AddOrUpdate(new Person("Person1", 1)); + results.Data.Count.Should().Be(1, "the current sequence should still be delivering"); + + second.Dispose(); + results.IsCompleted.Should().BeTrue("the current sequence has now completed"); + } + + [Fact] + public void CompletesWhenSourcesAndInnerCompleteSynchronously() + { + using var results = Observable.Return(Observable.Empty>()).Switch().AsAggregator(); + + results.IsCompleted.Should().BeTrue("everything completed during subscription"); + results.Error.Should().BeNull(); + } + + [Fact] + public void DeliversChangesEmittedBeforeSynchronousCompletion() + { + var change = new ChangeSet { new(ChangeReason.Add, "Person1", new Person("Person1", 1)) }; + using var results = Observable.Return(Observable.Return((IChangeSet)change)).Switch().AsAggregator(); + + results.Data.Count.Should().Be(1, "changes emitted before a synchronous completion must not be lost"); + results.IsCompleted.Should().BeTrue("the source completed"); + results.Error.Should().BeNull(); + } + + [Fact] + public void IgnoresChangesFromASupersededSource() + { + using var first = new Subject>(); + using var second = new Subject>(); + using var switchable = new BehaviorSubject>>(first); + using var results = switchable.Switch().AsAggregator(); + + first.OnNext(new ChangeSet { new(ChangeReason.Add, "Person1", new Person("Person1", 1)) }); + results.Data.Count.Should().Be(1); + + switchable.OnNext(second); + results.Data.Count.Should().Be(0, "moving to a new source drops what the previous one contributed"); + + first.OnNext(new ChangeSet { new(ChangeReason.Add, "Person2", new Person("Person2", 2)) }); + + results.Data.Count.Should().Be(0, "a superseded source must not be able to write into the result"); + results.Error.Should().BeNull(); + } + + [Fact] + public void PropagatesInnerErrorsRaisedSynchronously() + { + var error = new Exception("Test"); + using var results = Observable.Return(Observable.Throw>(error)).Switch().AsAggregator(); + + results.Error.Should().Be(error, "the error was raised during subscription"); + } + + [Fact] + public void DoesNotHoldALockWhileDeliveringDownstream() + { + // Observable.Switch holds its gate for the whole of downstream delivery, which is the shape that + // deadlocks when a pipeline crosses into another cache. Delivery has to go through the queue, which + // enqueues and returns, so a producer is never held up by whatever a subscriber is doing. + using var switchable = new Subject>>(); + using var first = new Subject>(); + + using var isDelivering = new ManualResetEventSlim(false); + using var release = new ManualResetEventSlim(false); + + using var subscription = switchable.Switch().Subscribe(_ => + { + isDelivering.Set(); + release.Wait(TimeSpan.FromSeconds(10)); + }); + + switchable.OnNext(first); + + var deliverer = new Thread(() => first.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) })) { IsBackground = true }; + deliverer.Start(); + + isDelivering.Wait(TimeSpan.FromSeconds(10)).Should().BeTrue("the subscriber should have been handed the change"); + + var producer = new Thread(() => switchable.OnNext(new Subject>())) { IsBackground = true }; + producer.Start(); + + var producerFinished = producer.Join(TimeSpan.FromSeconds(2)); + + release.Set(); + deliverer.Join(TimeSpan.FromSeconds(10)); + producer.Join(TimeSpan.FromSeconds(10)); + + producerFinished.Should().BeTrue("writing to the source must not block while a subscriber holds onto a notification"); + } } diff --git a/src/DynamicData/Cache/Internal/Switch.cs b/src/DynamicData/Cache/Internal/Switch.cs index 766f9f14a..539afe074 100644 --- a/src/DynamicData/Cache/Internal/Switch.cs +++ b/src/DynamicData/Cache/Internal/Switch.cs @@ -4,7 +4,6 @@ using System.Reactive.Disposables; using System.Reactive.Linq; -using System.Reactive.Subjects; namespace DynamicData.Cache.Internal; @@ -17,30 +16,93 @@ internal sealed class Switch(IObservable> Run() => Observable.Create>( observer => { - var queue = new SharedDeliveryQueue(); - - var destination = new LockFreeObservableCache(); - - var errors = new Subject>(); - - var populator = Observable.Switch( - _sources - .SynchronizeSafe(queue) - .Do(onNext: _ => destination.Clear(), - onError: error => errors.OnError(error))) - .SynchronizeSafe(queue) - .Do(onNext: static _ => { }, - onError: error => errors.OnError(error)) - .PopulateInto(destination); - - return new CompositeDisposable( - destination, - errors, - populator, - destination - .Connect() - .Merge(errors) - .SubscribeSafe(observer), - queue); + // Switching is done by hand rather than with Observable.Switch, which holds its gate for + // the whole of downstream delivery. The queue enqueues and returns instead, so a producer + // is never held up by whatever a subscriber does with the notification, and a pipeline + // crossing into another cache cannot deadlock against it. + var queue = new DeliveryQueue>(observer); + + // What the current source has contributed, so that switching away can take it back out. + var current = new Cache(); + var subscription = new SerialDisposable(); + + // Identifies the current source. A superseded one may still be mid-delivery, and anything + // it produces after this point belongs to a source that has already been switched away from. + var active = 0; + var isSourceRunning = false; + var areSourcesComplete = false; + + var outer = _sources.Subscribe( + source => + { + int id; + + using (var scope = queue.AcquireLock()) + { + id = ++active; + isSourceRunning = true; + + if (current.Count != 0) + { + scope.EnqueueNext(new ChangeSet( + current.KeyValues.Select(static pair => new Change(ChangeReason.Remove, pair.Key, pair.Value)))); + + current.Clear(); + } + } + + // Subscribed outside the lock. The source may deliver synchronously, and that + // delivery takes the lock for itself. + subscription.Disposable = source.Subscribe( + changes => + { + using var scope = queue.AcquireLock(); + + if (id != active) + { + return; + } + + current.Clone(changes); + + if (changes.Count != 0) + { + scope.EnqueueNext(changes); + } + }, + queue.OnError, + () => + { + using var scope = queue.AcquireLock(); + + if (id != active) + { + return; + } + + isSourceRunning = false; + + if (areSourcesComplete) + { + scope.EnqueueCompleted(); + } + }); + }, + queue.OnError, + () => + { + using var scope = queue.AcquireLock(); + + areSourcesComplete = true; + + // The current source may still be running, and the result ends only once both have. + if (!isSourceRunning) + { + scope.EnqueueCompleted(); + } + }); + + // Queue first, so that delivery is finished before the subscriptions are torn down. + return new CompositeDisposable(queue, outer, subscription); }); } From 75163806d90094e17d166d8b7a1f89d3050f20f6 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 15:37:54 -0700 Subject: [PATCH 2/4] Ignore errors from a superseded source in the cache Switch operator OnNext and OnCompleted both check the captured id against the active source, because a source that has been switched away from may still be mid-delivery. OnError went straight to the queue without that check, so a late failure from a source no longer selected could terminate the output. Rx's own Switch discards it, so the hand-rolled version was a regression on that point. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/Cache/SwitchFixture.cs | 23 ++++++++++++++++++++ src/DynamicData/Cache/Internal/Switch.cs | 12 +++++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/src/DynamicData.Tests/Cache/SwitchFixture.cs b/src/DynamicData.Tests/Cache/SwitchFixture.cs index a0a6cd83b..fa1bd2344 100644 --- a/src/DynamicData.Tests/Cache/SwitchFixture.cs +++ b/src/DynamicData.Tests/Cache/SwitchFixture.cs @@ -232,4 +232,27 @@ public void DoesNotHoldALockWhileDeliveringDownstream() producerFinished.Should().BeTrue("writing to the source must not block while a subscriber holds onto a notification"); } + + [Fact] + public void IgnoresErrorsFromASupersededSource() + { + // Switching away from a source means everything it produces afterwards belongs to a source + // that is no longer selected, and that includes its failures. + using var switchable = new Subject>>(); + using var superseded = new Subject>(); + using var current = new Subject>(); + + using var results = switchable.Switch().AsAggregator(); + + switchable.OnNext(superseded); + switchable.OnNext(current); + + superseded.OnError(new Exception("Test")); + + results.Error.Should().BeNull("the failed source had already been switched away from"); + + current.OnNext(new ChangeSet { new(ChangeReason.Add, "a", new Person("a", 1)) }); + + results.Data.Count.Should().Be(1, "the selected source should still be delivering"); + } } diff --git a/src/DynamicData/Cache/Internal/Switch.cs b/src/DynamicData/Cache/Internal/Switch.cs index 539afe074..f5174d470 100644 --- a/src/DynamicData/Cache/Internal/Switch.cs +++ b/src/DynamicData/Cache/Internal/Switch.cs @@ -70,7 +70,17 @@ public IObservable> Run() => Observable.Create + { + using var scope = queue.AcquireLock(); + + if (id != active) + { + return; + } + + scope.EnqueueError(error); + }, () => { using var scope = queue.AcquireLock(); From 8d9bdd93b7142847fcd456e440d230ee096fda47 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 16:11:18 -0700 Subject: [PATCH 3/4] Make the superseded-source test actually exercise the guard The first version subscribed a Subject and failed it after the switch, but SerialDisposable had already disposed that subscription, so Rx suppressed the notification and the test passed with or without the guard. Disposal cannot reach a notification already in flight, which is the case the guard exists for. RawAnonymousObservable hands back the observer directly, so the failure can be delivered after the switch without needing a race to land. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/Cache/SwitchFixture.cs | 24 ++++++++++++++------ 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SwitchFixture.cs b/src/DynamicData.Tests/Cache/SwitchFixture.cs index fa1bd2344..d98bf2ad2 100644 --- a/src/DynamicData.Tests/Cache/SwitchFixture.cs +++ b/src/DynamicData.Tests/Cache/SwitchFixture.cs @@ -1,13 +1,12 @@ using System; using System.Linq; +using System.Reactive.Disposables; using System.Reactive.Linq; using System.Reactive.Subjects; using System.Threading; - using DynamicData.Tests.Domain; - +using DynamicData.Tests.Utilities; using FluentAssertions; - using Xunit; namespace DynamicData.Tests.Cache; @@ -236,10 +235,20 @@ public void DoesNotHoldALockWhileDeliveringDownstream() [Fact] public void IgnoresErrorsFromASupersededSource() { - // Switching away from a source means everything it produces afterwards belongs to a source - // that is no longer selected, and that includes its failures. + // Switching away from a source means anything it produces afterwards belongs to a source that + // is no longer selected, and that includes its failures. Ordinarily disposal stops a + // superseded source being heard from again, but disposal cannot reach a notification that is + // already in flight, so the operator has to discard it on arrival. The raw observable hands + // back the observer directly, which is how that in-flight failure is reproduced here without + // needing a race to land. + var supersededObserver = default(IObserver>); + var superseded = RawAnonymousObservable.Create>(observer => + { + supersededObserver = observer; + return Disposable.Empty; + }); + using var switchable = new Subject>>(); - using var superseded = new Subject>(); using var current = new Subject>(); using var results = switchable.Switch().AsAggregator(); @@ -247,7 +256,8 @@ public void IgnoresErrorsFromASupersededSource() switchable.OnNext(superseded); switchable.OnNext(current); - superseded.OnError(new Exception("Test")); + supersededObserver.Should().NotBeNull("the superseded source should have been subscribed"); + supersededObserver!.OnError(new Exception("Test")); results.Error.Should().BeNull("the failed source had already been switched away from"); From 27b8a42e79de5feeac3d96a9fa6977e43c52f143 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Tue, 28 Jul 2026 08:01:53 -0700 Subject: [PATCH 4/4] Address review feedback Rename active to activeSourceId and id to sourceId, which say what they are. Use SubscribeSafe throughout, so a source that throws out of its subscribe call is reported through the queue rather than escaping to whoever happened to be subscribing. Replace the CompositeDisposable with an explicit closure. Disposal order matters here, the queue has to drain before the subscriptions feeding it go away, and CompositeDisposable does not specify an order. Drop OnCompletedFiresIfCacheDisposedAfterConnectingWhileSuspended, which is covered more thoroughly by the test in #1145. --- .../SuspendNotificationsFixture.UnitTests.cs | 22 ---------------- src/DynamicData/Cache/Internal/Switch.cs | 26 ++++++++++++------- 2 files changed, 16 insertions(+), 32 deletions(-) diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs index 9b6372476..a338aed3b 100644 --- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs +++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs @@ -454,28 +454,6 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch() results.IsCompleted.Should().BeFalse(); } - [Fact] - public void OnCompletedFiresIfCacheDisposedAfterConnectingWhileSuspended() - { - // A connection made while suspended is deferred until the suspension lifts. Once it - // activates it must behave exactly like any other subscriber, including terminating - // when the source does. Previously the deferral dropped the completion, leaving the - // subscriber alive forever. - var suspend = _source.SuspendNotifications(); - using var results = _source.Connect().AsAggregator(); - Enumerable.Range(101, 37).ForEach(_source.AddOrUpdate); - - // Act - suspend.Dispose(); - _source.AddOrUpdate(1000); - _source.Dispose(); - - // Assert - results.IsCompleted.Should().BeTrue("a connection deferred by a suspension should still complete when the source does"); - results.Error.Should().BeNull("no error should have occurred"); - results.Data.Count.Should().Be(38, "all data written before disposal should have arrived"); - } - [Fact] public void OnErrorFiresIfCacheFailsAfterConnectingWhileSuspended() { diff --git a/src/DynamicData/Cache/Internal/Switch.cs b/src/DynamicData/Cache/Internal/Switch.cs index f5174d470..28eef90a2 100644 --- a/src/DynamicData/Cache/Internal/Switch.cs +++ b/src/DynamicData/Cache/Internal/Switch.cs @@ -28,18 +28,18 @@ public IObservable> Run() => Observable.Create { - int id; + int sourceId; using (var scope = queue.AcquireLock()) { - id = ++active; + sourceId = ++activeSourceId; isSourceRunning = true; if (current.Count != 0) @@ -53,12 +53,12 @@ public IObservable> Run() => Observable.Create { using var scope = queue.AcquireLock(); - if (id != active) + if (sourceId != activeSourceId) { return; } @@ -74,7 +74,7 @@ public IObservable> Run() => Observable.Create> Run() => Observable.Create> Run() => Observable.Create + { + queue.Dispose(); + outer.Dispose(); + subscription.Dispose(); + }); }); }