diff --git a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs
index 521fbebc3..9b3dec0e5 100644
--- a/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs
+++ b/src/DynamicData.Tests/Cache/SuspendNotificationsFixture.UnitTests.cs
@@ -454,64 +454,80 @@ public void ReSuspendThenResumeDeliversAllInSingleBatch()
             results.IsCompleted.Should().BeFalse();
         }
 
-        [Fact]
-        public void OnErrorFiresIfCacheFailsWhileSuspended()
-        {
-            // A connection made while suspended is deferred, and must still be told when the source
-            // fails. Reporting the failure as a successful completion leaves the subscriber's error
-            // handling unrun and its data looking complete.
-            using var source = new Subject<IChangeSet<int, int>>();
-            using var cache = new IntermediateCache<int, int>(source);
-
-            using var suspend = cache.SuspendNotifications();
-            using var results = cache.Connect().AsAggregator();
-
-            var expectedError = new Exception("Test Exception");
-            source.OnError(expectedError);
-
-            results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail");
-            results.IsCompleted.Should().BeFalse("the source failed, it did not complete");
-        }
-
-        [Fact]
-        public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended()
-        {
-            // Watch defers the same way Connect does, and has the same obligation.
-            using var source = new Subject<IChangeSet<int, int>>();
-            using var cache = new IntermediateCache<int, int>(source);
-
-            using var suspend = cache.SuspendNotifications();
-            Exception? actualError = null;
-            var isCompleted = false;
-            using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true);
-
-            var expectedError = new Exception("Test Exception");
-            source.OnError(expectedError);
-
-            actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail");
-            isCompleted.Should().BeFalse("the source failed, it did not complete");
-        }
-
-        [Fact]
-        public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended()
-        {
-            // The deferred connection has activated by the time the failure arrives, so this covers
-            // the path through the connection itself rather than through the suspension gate.
-            using var source = new Subject<IChangeSet<int, int>>();
-            using var cache = new IntermediateCache<int, int>(source);
-
-            var suspend = cache.SuspendNotifications();
-            using var results = cache.Connect().AsAggregator();
-            source.OnNext(new ChangeSet<int, int> { new(ChangeReason.Add, 1, 1) });
-
-            suspend.Dispose();
-            var expectedError = new Exception("Test Exception");
-            source.OnError(expectedError);
-
-            results.Error.Should().Be(expectedError, "an activated connection should still see the source fail");
-            results.Data.Count.Should().Be(1, "the data written before the failure should have arrived");
-        }
-
+        [Fact]
+        public void OnErrorFiresIfCacheFailsWhileSuspended()
+        {
+            // A connection made while suspended is deferred, and must still be told when the source
+            // fails. Reporting the failure as a successful completion leaves the subscriber's error
+            // handling unrun and its data looking complete.
+            using var source = new Subject<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(source);
+
+            using var suspend = cache.SuspendNotifications();
+            using var results = cache.Connect().AsAggregator();
+
+            var expectedError = new Exception("Test Exception");
+            source.OnError(expectedError);
+
+            results.Error.Should().Be(expectedError, "a connection deferred by a suspension should see the source fail");
+            results.IsCompleted.Should().BeFalse("the source failed, it did not complete");
+        }
+
+        [Fact]
+        public void OnErrorFiresIfCacheFailsWhileWatchIsSuspended()
+        {
+            // Watch defers the same way Connect does, and has the same obligation.
+            using var source = new Subject<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(source);
+
+            using var suspend = cache.SuspendNotifications();
+            Exception? actualError = null;
+            var isCompleted = false;
+            using var subscription = cache.Watch(1).Subscribe(static _ => { }, error => actualError = error, () => isCompleted = true);
+
+            var expectedError = new Exception("Test Exception");
+            source.OnError(expectedError);
+
+            actualError.Should().Be(expectedError, "a watch deferred by a suspension should see the source fail");
+            isCompleted.Should().BeFalse("the source failed, it did not complete");
+        }
+
+        [Fact]
+        public void OnErrorFiresIfCacheFailsAfterResumingWhileConnectionWasSuspended()
+        {
+            // The deferred connection has activated by the time the failure arrives, so this covers
+            // the path through the connection itself rather than through the suspension gate.
+            using var source = new Subject<IChangeSet<int, int>>();
+            using var cache = new IntermediateCache<int, int>(source);
+
+            var suspend = cache.SuspendNotifications();
+            using var results = cache.Connect().AsAggregator();
+            source.OnNext(new ChangeSet<int, int> { new(ChangeReason.Add, 1, 1) });
+
+            suspend.Dispose();
+            var expectedError = new Exception("Test Exception");
+            source.OnError(expectedError);
+
+            results.Error.Should().Be(expectedError, "an activated connection should still see the source fail");
+            results.Data.Count.Should().Be(1, "the data written before the failure should have arrived");
+        }
+
+        [Fact]
+        public void OnCompletedFiresIfCacheDisposedAfterResumingWhileWatchWasSuspended()
+        {
+            // The tests above cover failure. Completion has to reach an activated watch too, and
+            // this covers the path through the watch itself rather than through the suspension gate.
+            var suspend = _source.SuspendNotifications();
+            var isCompleted = false;
+            using var subscription = _source.Watch(1).Subscribe(static _ => { }, () => isCompleted = true);
+            _source.AddOrUpdate(1);
+
+            suspend.Dispose();
+            _source.Dispose();
+
+            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..d98bf2ad2 100644
--- a/src/DynamicData.Tests/Cache/SwitchFixture.cs
+++ b/src/DynamicData.Tests/Cache/SwitchFixture.cs
@@ -1,11 +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;
@@ -89,4 +90,179 @@ public void PropagatesInnerErrors()
 
         results.Error.Should().Be(error);
     }
+
+    [Fact]
+    public void CompletesWhenSourcesAndInnerComplete()
+    {
+        using var source = new SourceCache<Person, string>(p => p.Name);
+        using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(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<Person, string>(p => p.Name);
+        using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(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<Person, string>(p => p.Name);
+        using var second = new SourceCache<Person, string>(p => p.Name);
+        using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(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<IChangeSet<Person, string>>()).Switch().AsAggregator();
+
+        results.IsCompleted.Should().BeTrue("everything completed during subscription");
+        results.Error.Should().BeNull();
+    }
+
+    [Fact]
+    public void DeliversChangesEmittedBeforeSynchronousCompletion()
+    {
+        var change = new ChangeSet<Person, string> { new(ChangeReason.Add, "Person1", new Person("Person1", 1)) };
+        using var results = Observable.Return(Observable.Return((IChangeSet<Person, string>)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<IChangeSet<Person, string>>();
+        using var second = new Subject<IChangeSet<Person, string>>();
+        using var switchable = new BehaviorSubject<IObservable<IChangeSet<Person, string>>>(first);
+        using var results = switchable.Switch().AsAggregator();
+
+        first.OnNext(new ChangeSet<Person, string> { 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<Person, string> { 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<IChangeSet<Person, string>>(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<IObservable<IChangeSet<Person, string>>>();
+        using var first = new Subject<IChangeSet<Person, string>>();
+
+        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<Person, string> { 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<IChangeSet<Person, string>>())) { 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");
+    }
+
+    [Fact]
+    public void IgnoresErrorsFromASupersededSource()
+    {
+        // 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<IChangeSet<Person, string>>);
+        var superseded = RawAnonymousObservable.Create<IChangeSet<Person, string>>(observer =>
+        {
+            supersededObserver = observer;
+            return Disposable.Empty;
+        });
+
+        using var switchable = new Subject<IObservable<IChangeSet<Person, string>>>();
+        using var current = new Subject<IChangeSet<Person, string>>();
+
+        using var results = switchable.Switch().AsAggregator();
+
+        switchable.OnNext(superseded);
+        switchable.OnNext(current);
+
+        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");
+
+        current.OnNext(new ChangeSet<Person, string> { 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 766f9f14a..28eef90a2 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,109 @@ internal sealed class Switch<TObject, TKey>(IObservable<IObservable<IChangeSet<T
     public IObservable<IChangeSet<TObject, TKey>> Run() => Observable.Create<IChangeSet<TObject, TKey>>(
             observer =>
             {
-                var queue = new SharedDeliveryQueue();
-
-                var destination = new LockFreeObservableCache<TObject, TKey>();
-
-                var errors = new Subject<IChangeSet<TObject, TKey>>();
-
-                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<IChangeSet<TObject, TKey>>(observer);
+
+                // What the current source has contributed, so that switching away can take it back out.
+                var current = new Cache<TObject, TKey>();
+                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 activeSourceId = 0;
+                var isSourceRunning = false;
+                var areSourcesComplete = false;
+
+                var outer = _sources.SubscribeSafe(
+                    source =>
+                    {
+                        int sourceId;
+
+                        using (var scope = queue.AcquireLock())
+                        {
+                            sourceId = ++activeSourceId;
+                            isSourceRunning = true;
+
+                            if (current.Count != 0)
+                            {
+                                scope.EnqueueNext(new ChangeSet<TObject, TKey>(
+                                    current.KeyValues.Select(static pair => new Change<TObject, TKey>(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.SubscribeSafe(
+                            changes =>
+                            {
+                                using var scope = queue.AcquireLock();
+
+                                if (sourceId != activeSourceId)
+                                {
+                                    return;
+                                }
+
+                                current.Clone(changes);
+
+                                if (changes.Count != 0)
+                                {
+                                    scope.EnqueueNext(changes);
+                                }
+                            },
+                            error =>
+                            {
+                                using var scope = queue.AcquireLock();
+
+                                if (sourceId != activeSourceId)
+                                {
+                                    return;
+                                }
+
+                                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();
+
+                        areSourcesComplete = true;
+
+                        // The current source may still be running, and the result ends only once both have.
+                        if (!isSourceRunning)
+                        {
+                            scope.EnqueueCompleted();
+                        }
+                    });
+
+                // 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();
+                });
             });
 }