From bfba39d8e08c94be8329e3b9f0bcb1c50ce2c2f9 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Sun, 20 Sep 2026 22:06:08 -0700 Subject: [PATCH 1/2] Preserve Switch subscription ownership across reentrant handoffs --- .../SwitchFixture.SubscriptionLifetime.cs | 395 ++++++++++++++++++ src/DynamicData.Tests/Cache/SwitchFixture.cs | 2 +- src/DynamicData/Cache/Internal/Switch.cs | 222 ++++++---- .../Cache/ObservableCacheEx.Switch.cs | 8 + 4 files changed, 550 insertions(+), 77 deletions(-) create mode 100644 src/DynamicData.Tests/Cache/SwitchFixture.SubscriptionLifetime.cs diff --git a/src/DynamicData.Tests/Cache/SwitchFixture.SubscriptionLifetime.cs b/src/DynamicData.Tests/Cache/SwitchFixture.SubscriptionLifetime.cs new file mode 100644 index 000000000..f0018956f --- /dev/null +++ b/src/DynamicData.Tests/Cache/SwitchFixture.SubscriptionLifetime.cs @@ -0,0 +1,395 @@ +using System; +using System.Collections.Generic; +using System.Linq; +using System.Reactive; +using System.Reactive.Concurrency; +using System.Reactive.Disposables; +using System.Reactive.Linq; +using System.Reactive.Subjects; +using DynamicData.Tests.Domain; +using DynamicData.Tests.Utilities; +using FluentAssertions; +using Xunit; +using Xunit.Abstractions; + +namespace DynamicData.Tests.Cache; + +public partial class SwitchFixture +{ + private readonly ITestOutputHelper _output; + + public SwitchFixture(ITestOutputHelper output) => _output = output; + + [Fact] + public void ReplacementReleasesPreviousSubscriptionBeforeAcquiringResource() + { + var people = CreateSubscriptionPeople(2); + Person? resourceOwner = null; + var released = new List(); + using var switchable = new Subject>>(); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .RecordCacheItems(out var results); + + switchable.OnNext(CreateExclusiveSource(people[0])); + results.RecordedItemsByKey.Values.Should().Equal(people[0]); + + switchable.OnNext(CreateExclusiveSource(people[1])); + + results.Error.Should().BeNull("the previous subscription must release the resource before its replacement subscribes"); + results.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[1] }.ToDictionary(person => person.Name)); + resourceOwner.Should().BeSameAs(people[1]); + released.Should().Equal(people[0]); + + subscription.Dispose(); + + resourceOwner.Should().BeNull(); + released.Should().Equal(people); + switchable.HasObservers.Should().BeFalse(); + results.HasCompleted.Should().BeFalse("unsubscription is not completion"); + + IObservable> CreateExclusiveSource(Person person) => + Observable.Create>(observer => + { + if (resourceOwner is not null) + { + observer.OnError(new InvalidOperationException("The previous subscription still owns the resource.")); + return Disposable.Empty; + } + + resourceOwner = person; + observer.OnNext(AddSubscriptionPerson(person)); + return Disposable.Create(() => + { + released.Add(person); + resourceOwner = null; + }); + }); + } + + [Theory] + [InlineData(null)] + [InlineData(NotificationKind.OnCompleted)] + [InlineData(NotificationKind.OnError)] + public void ReentrantSelectionDuringSynchronousAddKeepsLatestSubscription(NotificationKind? supersededTermination) + { + var people = CreateSubscriptionPeople(2); + var supersededDisposals = 0; + var supersededError = new InvalidOperationException("The superseded source failed before Subscribe returned."); + using var switchable = new Subject>>(); + using var latest = new Subject>(); + var first = Observable.Create>(observer => + { + // The downstream callback selects latest before this Subscribe can return its disposable. + observer.OnNext(AddSubscriptionPerson(people[0])); + if (supersededTermination is NotificationKind.OnCompleted) + observer.OnCompleted(); + else if (supersededTermination is NotificationKind.OnError) + observer.OnError(supersededError); + + return Disposable.Create(() => supersededDisposals++); + }); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .Do(changes => + { + if (changes.Any(change => change.Reason is ChangeReason.Add && change.Key == people[0].Name)) + switchable.OnNext(latest); + }) + .RecordCacheItems(out var results); + + switchable.OnNext(first); + var latestRemainedSubscribed = latest.HasObservers; + latest.OnNext(AddSubscriptionPerson(people[1])); + switchable.OnCompleted(); + var completedBeforeLatest = results.HasCompleted; + latest.OnCompleted(); + + results.Error.Should().BeNull("termination of a superseded source must not terminate the selected source"); + latestRemainedSubscribed.Should().BeTrue("the late return from the first Subscribe must not dispose the newer subscription"); + supersededDisposals.Should().Be(1); + completedBeforeLatest.Should().BeFalse(); + results.HasCompleted.Should().BeTrue(); + results.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[1] }.ToDictionary(person => person.Name)); + results.RecordedChangeSets.SelectMany(changes => changes) + .Select(change => (change.Reason, change.Key, change.Current)) + .Should().Equal( + (ChangeReason.Add, people[0].Name, people[0]), + (ChangeReason.Remove, people[0].Name, people[0]), + (ChangeReason.Add, people[1].Name, people[1])); + } + + [Fact] + public void ReentrantSelectionDuringResetDoesNotSubscribeSupersededReplacement() + { + var people = CreateSubscriptionPeople(3); + var supersededSubscriptions = 0; + using var first = new SourceCache(person => person.Name); + using var latest = new TestSourceCache(person => person.Name); + using var switchable = new Subject>>(); + first.AddOrUpdate(people[0]); + latest.AddOrUpdate(people[1]); + var superseded = Observable.Defer(() => + { + supersededSubscriptions++; + return Observable.Never>(); + }); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .Do(changes => + { + if (changes.Any(change => change.Reason is ChangeReason.Remove && change.Key == people[0].Name)) + switchable.OnNext(latest.Connect()); + }) + .RecordCacheItems(out var results); + + switchable.OnNext(first.Connect()); + switchable.OnNext(superseded); + latest.AddOrUpdate(people[2]); + switchable.OnCompleted(); + latest.Complete(); + + supersededSubscriptions.Should().Be(0, "the reset callback selected a newer source before the pending subscription started"); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.RecordedItemsByKey.Should().BeEquivalentTo(people.Skip(1).ToDictionary(person => person.Name)); + results.RecordedChangeSets.SelectMany(changes => changes) + .Select(change => (change.Reason, change.Key, change.Current)) + .Should().Equal( + (ChangeReason.Add, people[0].Name, people[0]), + (ChangeReason.Remove, people[0].Name, people[0]), + (ChangeReason.Add, people[1].Name, people[1]), + (ChangeReason.Add, people[2].Name, people[2])); + } + + [Fact] + public void ReentrantSelectionDuringDisposalDoesNotSubscribeSupersededReplacement() + { + var people = CreateSubscriptionPeople(2); + var firstDisposals = 0; + var supersededSubscriptions = 0; + using var latest = new Subject>(); + using var switchable = new Subject>>(); + var first = Observable.Create>(observer => + { + observer.OnNext(AddSubscriptionPerson(people[0])); + return Disposable.Create(() => + { + firstDisposals++; + switchable.OnNext(latest); + }); + }); + var superseded = Observable.Defer(() => + { + supersededSubscriptions++; + return Observable.Never>(); + }); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .RecordCacheItems(out var results); + + switchable.OnNext(first); + switchable.OnNext(superseded); + latest.OnNext(AddSubscriptionPerson(people[1])); + switchable.OnCompleted(); + latest.OnCompleted(); + + firstDisposals.Should().Be(1); + supersededSubscriptions.Should().Be(0, "releasing the old subscription selected latest before the replacement could start"); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[1] }.ToDictionary(person => person.Name)); + results.RecordedChangeSets.SelectMany(changes => changes) + .Select(change => (change.Reason, change.Key, change.Current)) + .Should().Equal( + (ChangeReason.Add, people[0].Name, people[0]), + (ChangeReason.Remove, people[0].Name, people[0]), + (ChangeReason.Add, people[1].Name, people[1])); + } + + [Fact] + public void DisposalDuringResetDoesNotSubscribeReplacement() + { + var people = CreateSubscriptionPeople(2); + var replacementSubscriptions = 0; + using var first = new BehaviorSubject>(AddSubscriptionPerson(people[0])); + using var replacementChanges = new Subject>(); + using var switchable = new Subject>>(); + var replacement = Observable.Defer(() => + { + replacementSubscriptions++; + return replacementChanges; + }); + var results = new CacheItemRecordingObserver(Scheduler.Immediate); + IObserver> recorder = results; + using var subscription = new SingleAssignmentDisposable(); + subscription.Disposable = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .Subscribe( + changes => + { + // Record the reset before disposing, so the expected observable state is unambiguous. + recorder.OnNext(changes); + if (changes.Removes != 0) + subscription.Dispose(); + }, + recorder.OnError, + recorder.OnCompleted); + + switchable.OnNext(first); + switchable.OnNext(replacement); + var notificationsAtDisposal = results.Notifications.ToArray(); + replacementChanges.OnNext(AddSubscriptionPerson(people[1])); + replacementChanges.OnCompleted(); + + replacementSubscriptions.Should().Be(0, "disposal in the reset callback must cancel the pending subscription, not just immediately dispose it"); + first.HasObservers.Should().BeFalse(); + replacementChanges.HasObservers.Should().BeFalse(); + switchable.HasObservers.Should().BeFalse(); + results.RecordedItemsByKey.Should().BeEmpty(); + results.Notifications.Should().Equal(notificationsAtDisposal); + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeFalse(); + } + + [Fact] + public void OuterErrorDuringResetDoesNotSubscribeReplacement() + { + var person = CreateSubscriptionPeople(1)[0]; + var error = new InvalidOperationException("The outer source failed during reset."); + var replacementSubscriptions = 0; + using var first = new BehaviorSubject>(AddSubscriptionPerson(person)); + using var switchable = new Subject>>(); + var replacement = Observable.Defer(() => + { + replacementSubscriptions++; + return Observable.Never>(); + }); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(value => value.Name) + .Do(changes => + { + if (changes.Removes != 0) + switchable.OnError(error); + }) + .RecordCacheItems(out var results); + + switchable.OnNext(first); + switchable.OnNext(replacement); + + replacementSubscriptions.Should().Be(0, "a pending subscription must not start after terminal delivery"); + results.Error.Should().BeSameAs(error); + results.HasCompleted.Should().BeFalse(); + results.RecordedItemsByKey.Should().BeEmpty(); + results.Notifications.Select(notification => notification.Value.Kind) + .Should().Equal(NotificationKind.OnNext, NotificationKind.OnNext, NotificationKind.OnError); + first.HasObservers.Should().BeFalse(); + switchable.HasObservers.Should().BeFalse(); + } + + [Fact] + public void OuterCompletionDuringResetWaitsForReplacement() + { + var people = CreateSubscriptionPeople(2); + using var first = new BehaviorSubject>(AddSubscriptionPerson(people[0])); + using var replacement = new TestSourceCache(person => person.Name); + using var switchable = new Subject>>(); + using var subscription = switchable.Switch() + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .Do(changes => + { + if (changes.Removes != 0) + switchable.OnCompleted(); + }) + .RecordCacheItems(out var results); + + switchable.OnNext(first); + switchable.OnNext(replacement.Connect()); + results.HasCompleted.Should().BeFalse("outer completion must not cancel the already-selected inner source"); + + replacement.AddOrUpdate(people[1]); + replacement.Complete(); + + results.Error.Should().BeNull(); + results.HasCompleted.Should().BeTrue(); + results.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[1] }.ToDictionary(person => person.Name)); + first.HasObservers.Should().BeFalse(); + } + + [Fact] + public void SynchronousOuterSubscriptionFailureReleasesActivatedInner() + { + var person = CreateSubscriptionPeople(1)[0]; + var error = new InvalidOperationException("The outer Subscribe failed after activating an inner source."); + var innerDisposals = 0; + using var inner = new BehaviorSubject>(AddSubscriptionPerson(person)); + var source = RawAnonymousObservable.Create>>(observer => + { + observer.OnNext(inner.Finally(() => innerDisposals++)); + throw error; + }); + using var subscription = source.Switch() + .ValidateSynchronization() + .ValidateChangeSets(value => value.Name) + .RecordCacheItems(out var results); + + results.Error.Should().BeSameAs(error); + results.HasCompleted.Should().BeFalse(); + results.RecordedItemsByKey.Should().BeEquivalentTo(new[] { person }.ToDictionary(value => value.Name)); + results.Notifications.Select(notification => notification.Value.Kind) + .Should().Equal(NotificationKind.OnNext, NotificationKind.OnError); + inner.HasObservers.Should().BeFalse("a failed outer activation must release the inner it already subscribed"); + innerDisposals.Should().Be(1); + } + + [Fact] + public void SubscriptionsKeepIndependentCurrentSources() + { + var people = CreateSubscriptionPeople(2); + using var first = new BehaviorSubject>(AddSubscriptionPerson(people[0])); + using var second = new BehaviorSubject>(AddSubscriptionPerson(people[1])); + using var switchable = new Subject>>(); + var switched = switchable.Switch(); + using var firstSubscription = switched + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .RecordCacheItems(out var firstResults); + using var secondSubscription = switched + .ValidateSynchronization() + .ValidateChangeSets(person => person.Name) + .RecordCacheItems(out var secondResults); + + switchable.OnNext(first); + firstSubscription.Dispose(); + switchable.OnNext(second); + switchable.OnCompleted(); + second.OnCompleted(); + + firstResults.Error.Should().BeNull(); + firstResults.HasCompleted.Should().BeFalse(); + firstResults.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[0] }.ToDictionary(person => person.Name)); + secondResults.Error.Should().BeNull(); + secondResults.HasCompleted.Should().BeTrue(); + secondResults.RecordedItemsByKey.Should().BeEquivalentTo(new[] { people[1] }.ToDictionary(person => person.Name)); + first.HasObservers.Should().BeFalse(); + second.HasObservers.Should().BeFalse(); + switchable.HasObservers.Should().BeFalse(); + } + + private static IChangeSet AddSubscriptionPerson(Person person) => + new ChangeSet { new(ChangeReason.Add, person.Name, person) }; + + private Person[] CreateSubscriptionPeople(int count) + { + const int seed = 0x51A7; + _output.WriteLine("Subscription lifetime data: seed={0}, count={1}", seed, count); + return Fakers.Person.Clone().UseSeed(seed).Generate(count).ToArray(); + } +} diff --git a/src/DynamicData.Tests/Cache/SwitchFixture.cs b/src/DynamicData.Tests/Cache/SwitchFixture.cs index d98bf2ad2..03c41b14b 100644 --- a/src/DynamicData.Tests/Cache/SwitchFixture.cs +++ b/src/DynamicData.Tests/Cache/SwitchFixture.cs @@ -11,7 +11,7 @@ namespace DynamicData.Tests.Cache; -public class SwitchFixture +public partial class SwitchFixture { [Fact] public void ClearsForNewSource() diff --git a/src/DynamicData/Cache/Internal/Switch.cs b/src/DynamicData/Cache/Internal/Switch.cs index 28eef90a2..1dc34b43a 100644 --- a/src/DynamicData/Cache/Internal/Switch.cs +++ b/src/DynamicData/Cache/Internal/Switch.cs @@ -13,8 +13,9 @@ internal sealed class Switch(IObservable>> _sources = sources ?? throw new ArgumentNullException(nameof(sources)); - public IObservable> Run() => Observable.Create>( - observer => + public IObservable> Run() => Observable.Using( + static () => new SingleAssignmentDisposable(), + lifetime => Observable.Create>(observer => { // 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 @@ -24,101 +25,170 @@ public IObservable> Run() => Observable.Create(); - var subscription = new SerialDisposable(); + var outer = new SingleAssignmentDisposable(); - // 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 activeSourceId = 0; + // The holder is also the generation token. Publish it before calling any user code, so + // reentrant selection can cancel a subscription whose Subscribe has not yet returned. + SingleAssignmentDisposable? activeSubscription = null; var isSourceRunning = false; var areSourcesComplete = false; + var isStopped = false; - var outer = _sources.SubscribeSafe( - source => + // Using owns this slot before activation, including synchronous subscription failures. + lifetime.Disposable = Disposable.Create(() => + { + SingleAssignmentDisposable? subscription; + + // This scope does not drain. Invalidate pending work before waiting for delivery, + // and never run subscription teardown under the queue's gate. + using (queue.AcquireReadLock()) + { + isStopped = true; + subscription = activeSubscription; + activeSubscription = null; + } + + // Finish downstream delivery before tearing down the sources feeding it. + queue.Dispose(); + try + { + outer.Dispose(); + } + finally { - int sourceId; + subscription?.Dispose(); + } + }); - using (var scope = queue.AcquireLock()) + outer.Disposable = _sources.SubscribeSafe( + SwitchSource, + error => Fail(error, null), + () => + { + using var scope = queue.AcquireLock(); + + if (isStopped) { - sourceId = ++activeSourceId; - isSourceRunning = true; + return; + } - if (current.Count != 0) - { - scope.EnqueueNext(new ChangeSet( - current.KeyValues.Select(static pair => new Change(ChangeReason.Remove, pair.Key, pair.Value)))); + areSourcesComplete = true; - current.Clear(); - } + // A selected source counts as running even while its subscription is pending. + if (!isSourceRunning) + { + isStopped = true; + scope.EnqueueCompleted(); } + }); - // Subscribed outside the lock. The source may deliver synchronously, and that - // delivery takes the lock for itself. - subscription.Disposable = source.SubscribeSafe( - changes => - { - using var scope = queue.AcquireLock(); + return Disposable.Empty; - if (sourceId != activeSourceId) - { - return; - } + void SwitchSource(IObservable> source) + { + SingleAssignmentDisposable subscription; + SingleAssignmentDisposable? previous; - current.Clone(changes); + // Publish without draining notifications: even a reset callback must not get ahead + // of releasing the previous source's resources. + using (queue.AcquireReadLock()) + { + if (isStopped || areSourcesComplete) + { + return; + } - if (changes.Count != 0) - { - scope.EnqueueNext(changes); - } - }, - error => - { - using var scope = queue.AcquireLock(); + subscription = new SingleAssignmentDisposable(); + previous = activeSubscription; + activeSubscription = subscription; + isSourceRunning = true; + } - if (sourceId != activeSourceId) - { - return; - } + // Disposal is user code too: it can select another source or terminate the result. + previous?.Dispose(); - scope.EnqueueError(error); - }, - () => - { - using var scope = queue.AcquireLock(); - - if (sourceId != activeSourceId) - { - return; - } - - isSourceRunning = false; - - if (areSourcesComplete) - { - scope.EnqueueCompleted(); - } - }); - }, - queue.OnError, - () => + using (var scope = queue.AcquireLock()) { - using var scope = queue.AcquireLock(); + if (!IsCurrent(subscription)) + { + return; + } - areSourcesComplete = true; + if (current.Count != 0) + { + scope.EnqueueNext(new ChangeSet( + current.KeyValues.Select(static pair => new Change(ChangeReason.Remove, pair.Key, pair.Value)))); - // The current source may still be running, and the result ends only once both have. - if (!isSourceRunning) + current.Clear(); + } + } + + // Reset delivery is another reentrant boundary. Outer completion alone does not + // cancel this source, but disposal, failure, or a newer selection does. + using (queue.AcquireReadLock()) + { + if (!IsCurrent(subscription)) { - scope.EnqueueCompleted(); + return; } - }); + } - // 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(() => + // Subscribe outside the gate and assign only to this generation's holder. If a + // synchronous notification selected a newer source, this holder is already disposed + // and disposes the late-returning subscription without touching the newer one. + subscription.Disposable = source.SubscribeSafe( + changes => + { + using var scope = queue.AcquireLock(); + + if (!IsCurrent(subscription)) + { + return; + } + + current.Clone(changes); + + if (changes.Count != 0) + { + scope.EnqueueNext(changes); + } + }, + error => Fail(error, subscription), + () => + { + using var scope = queue.AcquireLock(); + + if (!IsCurrent(subscription)) + { + return; + } + + isSourceRunning = false; + + if (areSourcesComplete) + { + isStopped = true; + scope.EnqueueCompleted(); + } + }); + } + + void Fail(Exception error, SingleAssignmentDisposable? subscription) { - queue.Dispose(); - outer.Dispose(); - subscription.Dispose(); - }); - }); + using var scope = queue.AcquireLock(); + + if (isStopped || (subscription is not null && !ReferenceEquals(subscription, activeSubscription))) + { + return; + } + + // Stop as soon as the terminal notification is queued, not only when it is delivered. + isStopped = true; + scope.EnqueueError(error); + } + + // All callers hold the queue gate; notification handlers and handoffs use the same state. + bool IsCurrent(SingleAssignmentDisposable subscription) => + !isStopped && ReferenceEquals(subscription, activeSubscription); + })); } diff --git a/src/DynamicData/Cache/ObservableCacheEx.Switch.cs b/src/DynamicData/Cache/ObservableCacheEx.Switch.cs index 9c5d06c9e..228713291 100644 --- a/src/DynamicData/Cache/ObservableCacheEx.Switch.cs +++ b/src/DynamicData/Cache/ObservableCacheEx.Switch.cs @@ -47,6 +47,14 @@ public static IObservable> Switch(this /// A changeset stream reflecting the items from the most recently emitted inner source. /// /// On switch: Remove is emitted for all items from the previous source, then Add for all items from the new source. + /// + /// The previous inner subscription is disposed before the replacement subscribes. If disposal or synchronous + /// notification callbacks select a newer source, a superseded pending source cannot replace that newer subscription. + /// + /// + /// Completion of the outer source waits for the selected inner source to complete. An error terminates the result, + /// and disposing the result cancels pending source activation as well as the active subscriptions. + /// /// Worth noting: Each switch clears the entire downstream cache before populating from the new source. Subscribers see a full remove-then-add reset on every switch. /// Also worth noting:This operator intentionally shadows the native operator, as downstream listeners will generally become corrupt when the native operator is used. This is due to its lack of the automatic-clearing behavior mentioned above. /// From 0eac5f26b75b8169fffd5010dafd9603a7bf9023 Mon Sep 17 00:00:00 2001 From: "Darrin W. Cullop" Date: Mon, 21 Sep 2026 07:32:18 -0700 Subject: [PATCH 2/2] Document Switch subscription ownership semantics --- .github/instructions/dynamicdata-cache.instructions.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/instructions/dynamicdata-cache.instructions.md b/.github/instructions/dynamicdata-cache.instructions.md index 45b1874bb..dfd054233 100644 --- a/.github/instructions/dynamicdata-cache.instructions.md +++ b/.github/instructions/dynamicdata-cache.instructions.md @@ -792,6 +792,8 @@ FIFO eviction when cache exceeds a size limit. `IObservable>>` → subscribes to the latest inner observable, disposing previous. +The previous subscription is released before its replacement starts. Reentrant selection during reset, disposal, or initial delivery cannot activate a superseded source or dispose the newest subscription. Outer completion waits for the selected inner source, while errors and disposal cancel pending activation. + ### RefCount Shares the upstream subscription with reference counting.