From 80c0eb87e5431585ad11ca9961f9bde48863229e Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 08:54:49 -0700 Subject: [PATCH 1/4] Fix that the list Switch operator drops completion and throws errors, without taking a lock Switch relayed changes through a private SourceList and subscribed the observer to that, so the terminal event of the source had nowhere to go. Completion was dropped outright, and an error was rethrown out of the subscription rather than delivered as OnError. 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 collection 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. 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. --- src/DynamicData.Tests/List/SwitchFixture.cs | 176 ++++++++++++++++++++ src/DynamicData/List/Internal/Switch.cs | 90 ++++++++-- 2 files changed, 256 insertions(+), 10 deletions(-) diff --git a/src/DynamicData.Tests/List/SwitchFixture.cs b/src/DynamicData.Tests/List/SwitchFixture.cs index abcdaa8d3..049d08178 100644 --- a/src/DynamicData.Tests/List/SwitchFixture.cs +++ b/src/DynamicData.Tests/List/SwitchFixture.cs @@ -1,6 +1,8 @@ using System; using System.Linq; +using System.Reactive.Linq; using System.Reactive.Subjects; +using System.Threading; using FluentAssertions; @@ -60,4 +62,178 @@ public void PoulatesFirstSource() inital.Should().BeEquivalentTo(_source.Items); } + + [Fact] + public void PropagatesOuterErrors() + { + using var source = new SourceList(); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + source.AddRange(Enumerable.Range(1, 100)); + + var error = new Exception("Test"); + switchable.OnError(error); + + results.Exception.Should().Be(error); + } + + [Fact] + public void PropagatesInnerErrors() + { + using var source = new SourceList(); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + source.AddRange(Enumerable.Range(1, 100)); + + using var source2 = new Subject>(); + switchable.OnNext(source2); + + var error = new Exception("Test"); + source2.OnError(error); + + results.Exception.Should().Be(error); + } + + [Fact] + public void CompletesWhenSourcesAndInnerComplete() + { + using var source = new SourceList(); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + source.AddRange(Enumerable.Range(1, 100)); + + 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.Exception.Should().BeNull(); + results.Data.Count.Should().Be(100, "all data should have been received before completion"); + } + + [Fact] + public void DoesNotCompleteWhileInnerIsStillRunning() + { + using var source = new SourceList(); + using var switchable = new BehaviorSubject>>(source.Connect()); + using var results = switchable.Switch().AsAggregator(); + + switchable.OnCompleted(); + source.Add(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 SourceList(); + using var second = new SourceList(); + 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.Add(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.Exception.Should().BeNull(); + } + + [Fact] + public void DeliversChangesEmittedBeforeSynchronousCompletion() + { + var change = new ChangeSet { new(ListChangeReason.Add, 42) }; + 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.Exception.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(ListChangeReason.Add, 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(ListChangeReason.Add, 2) }); + + results.Data.Count.Should().Be(0, "a superseded source must not be able to write into the result"); + results.Exception.Should().BeNull(); + } + + [Fact] + public void PropagatesInnerErrorsRaisedSynchronously() + { + var error = new Exception("Test"); + using var results = Observable.Return(Observable.Throw>(error)).Switch().AsAggregator(); + + results.Exception.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 collection. 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 Change(ListChangeReason.Add, 1, 0) })) { 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/List/Internal/Switch.cs b/src/DynamicData/List/Internal/Switch.cs index 9425771dd..4ad40ebc7 100644 --- a/src/DynamicData/List/Internal/Switch.cs +++ b/src/DynamicData/List/Internal/Switch.cs @@ -15,21 +15,91 @@ internal sealed class Switch(IObservable>> sources) public IObservable> Run() => Observable.Create>( observer => { - var locker = InternalEx.NewLock(); + // 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 collection cannot deadlock against it. + var queue = new DeliveryQueue>(observer); - var destination = new SourceList(); + // What the current source has contributed, so that switching away can take it back out. + var current = new List(); + var subscription = new SerialDisposable(); - var populator = Observable.Switch( - _sources.Do( - _ => + // 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()) { - lock (locker) + id = ++active; + isSourceRunning = true; + + if (current.Count != 0) { - destination.Clear(); + scope.EnqueueNext(new ChangeSet { new Change(ListChangeReason.Clear, current.ToArray()) }); + current.Clear(); } - })).Synchronize(locker).PopulateInto(destination); + } + + // 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(); + } + }); - var publisher = destination.Connect().SubscribeSafe(observer); - return new CompositeDisposable(destination, populator, publisher); + // Queue first, so that delivery is finished before the subscriptions are torn down. + return new CompositeDisposable(queue, outer, subscription); }); } From 65dcafd6164bb87f48e22cda08c32fe9037a712f Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 15:39:15 -0700 Subject: [PATCH 2/4] Ignore errors from a superseded source in the list Switch operator The list counterpart of the same gap in the cache operator: OnNext and OnCompleted check the captured id against the active source, but OnError went straight to the queue, so a late failure from a source already switched away from could terminate the output. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/List/SwitchFixture.cs | 23 +++++++++++++++++++++ src/DynamicData/List/Internal/Switch.cs | 12 ++++++++++- 2 files changed, 34 insertions(+), 1 deletion(-) diff --git a/src/DynamicData.Tests/List/SwitchFixture.cs b/src/DynamicData.Tests/List/SwitchFixture.cs index 049d08178..10c7550e5 100644 --- a/src/DynamicData.Tests/List/SwitchFixture.cs +++ b/src/DynamicData.Tests/List/SwitchFixture.cs @@ -236,4 +236,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.Exception.Should().BeNull("the failed source had already been switched away from"); + + current.OnNext(new ChangeSet { new Change(ListChangeReason.Add, 1, 0) }); + + results.Data.Count.Should().Be(1, "the selected source should still be delivering"); + } } diff --git a/src/DynamicData/List/Internal/Switch.cs b/src/DynamicData/List/Internal/Switch.cs index 4ad40ebc7..70dee62b9 100644 --- a/src/DynamicData/List/Internal/Switch.cs +++ b/src/DynamicData/List/Internal/Switch.cs @@ -67,7 +67,17 @@ public IObservable> Run() => Observable.Create>( scope.EnqueueNext(changes); } }, - queue.OnError, + error => + { + using var scope = queue.AcquireLock(); + + if (id != active) + { + return; + } + + scope.EnqueueError(error); + }, () => { using var scope = queue.AcquireLock(); From daadde6a7a4f45a2234447aaed4bd7fb44b258a2 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 27 Jul 2026 16:12:25 -0700 Subject: [PATCH 3/4] Make the superseded-source test actually exercise the guard Same correction as the cache side. The Subject version was disposed by SerialDisposable before the failure was raised, so Rx suppressed it and the test passed with or without the guard. RawAnonymousObservable delivers the failure after the switch, which is the in-flight case the guard is for. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- src/DynamicData.Tests/List/SwitchFixture.cs | 22 +++++++++++++++++---- 1 file changed, 18 insertions(+), 4 deletions(-) diff --git a/src/DynamicData.Tests/List/SwitchFixture.cs b/src/DynamicData.Tests/List/SwitchFixture.cs index 10c7550e5..5b490f84f 100644 --- a/src/DynamicData.Tests/List/SwitchFixture.cs +++ b/src/DynamicData.Tests/List/SwitchFixture.cs @@ -1,9 +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.Utilities; + using FluentAssertions; using Xunit; @@ -240,10 +243,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(); @@ -251,7 +264,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.Exception.Should().BeNull("the failed source had already been switched away from"); From e62ba6bda9c9eca072179f61e0ee4b610aaf0a67 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Thu, 30 Jul 2026 17:16:07 -0700 Subject: [PATCH 4/4] Use SubscribeSafe and explicit disposal in the list Switch operator Matches what the cache version got in #1137. SubscribeSafe on both the outer and inner subscriptions, and an explicit teardown instead of CompositeDisposable, which does not specify a disposal order. The queue has to go first so any delivery in flight finishes before the subscriptions feeding it are torn down. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 9582bb33-26d3-4aa5-8dd7-57dc55304680 --- src/DynamicData/List/Internal/Switch.cs | 14 ++++++++++---- 1 file changed, 10 insertions(+), 4 deletions(-) diff --git a/src/DynamicData/List/Internal/Switch.cs b/src/DynamicData/List/Internal/Switch.cs index 70dee62b9..dcd20c283 100644 --- a/src/DynamicData/List/Internal/Switch.cs +++ b/src/DynamicData/List/Internal/Switch.cs @@ -31,7 +31,7 @@ public IObservable> Run() => Observable.Create>( var isSourceRunning = false; var areSourcesComplete = false; - var outer = _sources.Subscribe( + var outer = _sources.SubscribeSafe( source => { int id; @@ -50,7 +50,7 @@ public IObservable> Run() => Observable.Create>( // Subscribed outside the lock. The source may deliver synchronously, and that // delivery takes the lock for itself. - subscription.Disposable = source.Subscribe( + subscription.Disposable = source.SubscribeSafe( changes => { using var scope = queue.AcquireLock(); @@ -109,7 +109,13 @@ public IObservable> Run() => Observable.Create>( } }); - // Queue first, so that delivery is finished before the subscriptions are torn down. - return new CompositeDisposable(queue, outer, subscription); + // Disposal order matters and CompositeDisposable does not specify one. The queue goes first + // so that any delivery in flight is finished before the subscriptions feeding it are torn down. + return Disposable.Create(() => + { + queue.Dispose(); + outer.Dispose(); + subscription.Dispose(); + }); }); }