Fix that the cache Switch operator never completes

[dwcullop]

Fixes #1136.

The cache Switch operator could never complete, could retain data from a source it had already switched away from, and routed delivery through a lock.

Never completing

It relayed changes through a private LockFreeObservableCache:

var destination = new LockFreeObservableCache<TObject, TKey>();
...
.PopulateInto(destination);

destination.Connect().Merge(errors).SubscribeSafe(observer)

A relay only ends when it is disposed, so the terminal event of the source had nowhere to go. Errors were carried across by hand through the merged errors subject; completion had no equivalent path and was silently dropped. A consumer of this operator never received OnCompleted, ever.

The lock

The obvious fix is to let Observable.Switch sit between the sources and the observer and propagate terminal events on its own. That is wrong here, and it took a measurement to see why.

Observable.Switch 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. That is precisely the cross cache deadlock shape the delivery queue exists to avoid, and the operator on main avoids it today by keeping the observer on the far side of the relay cache.

Measured with a subscriber that blocks inside OnNext, writing to the source from another thread:

	time to return
main	0 ms
via Observable.Switch	748 ms
this PR	1 ms

So the operator is now written by hand. A SerialDisposable holds the current inner subscription, and every subscription 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, fixed by construction rather than by timing. All state changes and all delivery go through the queue, which releases its lock before handing anything downstream, so a second producer enqueues and returns instead of waiting.

The reset that clears the previous source's contribution is enqueued under the same lock acquisition that takes the new identity, which keeps it ordered against the changes that follow it.

Terminal semantics

Completes once the sources and the current inner have both completed. Fails as soon as either does. That matches Observable.Switch, which is what callers reasonably expect from something called Switch.

Tests

Cache/SwitchFixture:

• CompletesWhenSourcesAndInnerComplete
• DoesNotCompleteWhileInnerIsStillRunning
• DoesNotCompleteWhenOnlyASupersededInnerCompletes
• CompletesWhenSourcesAndInnerCompleteSynchronously
• DeliversChangesEmittedBeforeSynchronousCompletion
• PropagatesInnerErrorsRaisedSynchronously
• IgnoresChangesFromASupersededSource
• DoesNotHoldALockWhileDeliveringDownstream

That last one blocks a subscriber inside OnNext and asserts that writing to the source from another thread still returns. It fails against an Observable.Switch implementation and passes against this one, so the lock behaviour cannot quietly come back.

SuspendNotificationsFixture, covering #1136 directly:

• OnCompletedFiresIfCacheDisposedAfterConnectingWhileSuspended
• OnErrorFiresIfCacheFailsAfterConnectingWhileSuspended
• OnCompletedFiresIfCacheDisposedAfterWatchingWhileSuspended

Notes

Rebased onto current main and squashed. The branch previously carried a workaround in ObservableCache.Connect() that was later reverted in favour of fixing the operator, and replaying that history over #1132 conflicted repeatedly for no benefit. Nothing outside Switch and its tests is touched now.

The list Switch has the same three defects and is fixed the same way in #1139. The two are independent.

Behavioural change to a public operator: sequences that previously never terminated now terminate when their sources do. main is on 10.0-preview.

Unrelated, and not addressed here: AggregationEx.InvalidateWhen still uses Rx's Switch and therefore still holds a gate during delivery. It operates on plain value streams rather than changesets, so the blast radius is smaller, but it is the same pattern if anyone is worried about it.

[Copilot]

Pull request overview

This PR fixes a long-standing termination bug in DynamicData’s cache Switch operator, restoring Rx-like Observable.Switch completion semantics for IObservable<IObservable<IChangeSet<TObject, TKey>>> pipelines (notably affecting deferred Connect() subscriptions during notification suspension, per #1136).

Changes:

• Propagate terminal events through an explicit “terminal” channel and end the downstream destination.Connect() stream via TakeUntil, enabling completion to flow correctly.
• Add operator-level tests for completion behavior (sources vs current/superseded inner sequences).
• Add regression tests covering deferred Connect() / Watch() during SuspendNotifications(), ensuring completion/error propagation after activation.

Reviewed changes

Copilot reviewed 3 out of 3 changed files in this pull request and generated 1 comment.

File	Description
src/DynamicData/Cache/Internal/Switch.cs	Reworks termination propagation so the cache Switch can complete/error correctly despite relaying via a cache that otherwise only ends on disposal.
src/DynamicData.Tests/Cache/SwitchFixture.cs	Adds targeted unit tests validating correct completion semantics of the cache Switch operator.
src/DynamicData.Tests/Cache/SuspendNotificationsFixture.cs	Adds regression coverage for #1136 ensuring deferred subscriptions complete/error once activated.