diff --git a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs
index 89f274619..23fe05d6a 100644
--- a/src/DynamicData.Tests/AggregationTests/MaxFixture.cs
+++ b/src/DynamicData.Tests/AggregationTests/MaxFixture.cs
@@ -1,4 +1,5 @@
-using System;
+using System;
+using System.Reactive.Subjects;
 
 using DynamicData.Aggregation;
 using DynamicData.Tests.Domain;
@@ -70,4 +71,30 @@ public void RemoveItems()
         result.Should().Be(20, "Max value should be 20 after remove");
         accumulator.Dispose();
     }
+
+    [Fact]
+    public void MaximumCompletesWhenTheSourceCompletes()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("Maximum is built on QueryWhenChanged");
+    }
+
+    [Fact]
+    public void MaximumDeliversTheErrorWithoutThrowing()
+    {
+        Exception? error = null;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.Maximum(p => p.Age).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/AndFixture.cs b/src/DynamicData.Tests/List/AndFixture.cs
index 1d18596ad..fef62fb8c 100644
--- a/src/DynamicData.Tests/List/AndFixture.cs
+++ b/src/DynamicData.Tests/List/AndFixture.cs
@@ -1,7 +1,9 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -97,4 +99,34 @@ public void StartingWithNonEmptySourceProducesNoResult()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = ObservableListEx.And(first, second).Subscribe(_ => { }, () => completed = true);
+
+        first.OnCompleted();
+        completed.Should().BeFalse("the second source is still live");
+
+        second.OnCompleted();
+        completed.Should().BeTrue("every source has now finished");
+    }
+
+    [Fact]
+    public void DeliversAnErrorFromAnySource()
+    {
+        Exception? error = null;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = ObservableListEx.And(first, second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/BufferFixture.cs b/src/DynamicData.Tests/List/BufferFixture.cs
index e02330f98..ef3dfbad3 100644
--- a/src/DynamicData.Tests/List/BufferFixture.cs
+++ b/src/DynamicData.Tests/List/BufferFixture.cs
@@ -1,5 +1,8 @@
-using System;
+using System;
+using System.Collections.Generic;
+using System.Reactive.Concurrency;
 using System.Reactive.Linq;
+using System.Reactive.Subjects;
 
 using DynamicData.Tests.Domain;
 
@@ -48,4 +51,50 @@ public void ResultsWillBeReceivedAfterClosingBuffer()
         _scheduler.AdvanceBy(TimeSpan.FromSeconds(61).Ticks);
         _results.Messages.Count.Should().Be(1, "Should be 1 update");
     }
+
+    [Fact]
+    public void BufferIfCompletesWhenTheSourceCompletes()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.BufferIf(Observable.Return(false), Scheduler.Immediate).Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void BufferIfFlushesHeldChangesBeforeCompleting()
+    {
+        var received = 0;
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.BufferIf(Observable.Return(true), Scheduler.Immediate).Subscribe(_ => received++, () => completed = true);
+
+        source.OnNext(new ChangeSet<Person> { new Change<Person>(ListChangeReason.Add, new Person("a", 1), 0) });
+        source.OnCompleted();
+
+        received.Should().Be(1, "changes held back by the pause would otherwise be lost");
+        completed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void BufferIfFailsWhenThePauseSelectorFails()
+    {
+        var expectedError = new Exception("Test Exception");
+        var actualError = default(Exception);
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var pause = new Subject<bool>();
+        using var subscription = source.BufferIf(pause, Scheduler.Immediate).Subscribe(_ => { }, error => actualError = error, () => completed = true);
+
+        pause.OnError(expectedError);
+
+        actualError.Should().BeSameAs(expectedError, "a failure of the pause selector belongs to the subscriber, not to whichever thread happened to raise it");
+        completed.Should().BeFalse("the pause selector failed, it did not complete");
+    }
 }
diff --git a/src/DynamicData.Tests/List/ExceptFixture.cs b/src/DynamicData.Tests/List/ExceptFixture.cs
index d74eff07b..eab8cd97a 100644
--- a/src/DynamicData.Tests/List/ExceptFixture.cs
+++ b/src/DynamicData.Tests/List/ExceptFixture.cs
@@ -1,7 +1,9 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -106,4 +108,34 @@ public void NothingFromOther()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Except(second).Subscribe(_ => { }, () => completed = true);
+
+        first.OnCompleted();
+        completed.Should().BeFalse("the second source is still live");
+
+        second.OnCompleted();
+        completed.Should().BeTrue("every source has now finished");
+    }
+
+    [Fact]
+    public void DeliversAnErrorFromAnySource()
+    {
+        Exception? error = null;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Except(second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/GroupImmutableFixture.cs b/src/DynamicData.Tests/List/GroupImmutableFixture.cs
index cbf674654..75b541691 100644
--- a/src/DynamicData.Tests/List/GroupImmutableFixture.cs
+++ b/src/DynamicData.Tests/List/GroupImmutableFixture.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Generic;
 using System.Linq;
 using System.Reactive;
 using System.Reactive.Subjects;
@@ -181,4 +182,17 @@ public void UpdatesArePermissible()
         var group = _results.Data.Items[0];
         group.Count.Should().Be(2);
     }
+
+    [Fact]
+    public void CompletesWhenNoRegrouperIsSupplied()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.GroupWithImmutableState(p => p.Age).Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open");
+    }
 }
diff --git a/src/DynamicData.Tests/List/GroupOnFixture.cs b/src/DynamicData.Tests/List/GroupOnFixture.cs
index 61164d3c3..693f883ae 100644
--- a/src/DynamicData.Tests/List/GroupOnFixture.cs
+++ b/src/DynamicData.Tests/List/GroupOnFixture.cs
@@ -1,5 +1,7 @@
-using System;
+using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
 using DynamicData.Tests.Domain;
 
@@ -74,4 +76,17 @@ public void UpdateWillChangeTheGroup()
         var firstGroup = _results.Data.Items[0].List.Items.ToArray();
         firstGroup[0].Should().Be(amended, "Should be same person");
     }
+
+    [Fact]
+    public void CompletesWhenNoRegrouperIsSupplied()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.GroupOn(p => p.Age).Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("an absent regrouper can never fire and so must not hold the result open");
+    }
 }
diff --git a/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs
index 253a70070..ce11ca522 100644
--- a/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs
+++ b/src/DynamicData.Tests/List/MergeChangeSetsFixture.cs
@@ -1,4 +1,5 @@
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Linq;
 using System.Reactive.Concurrency;
@@ -12,7 +13,6 @@
 using FluentAssertions;
 using Microsoft.Reactive.Testing;
 using Xunit;
-using System.Collections.Concurrent;
 
 namespace DynamicData.Tests.List;
 
@@ -439,4 +439,21 @@ private Animal UseInsert(AnimalOwner owner)
 
     private IEnumerable<IObservable<IChangeSet<Animal>>> GetEnumerableObservable() => _animalOwners.Select(owner => owner.Animals.Connect());
     private IObservable<IObservable<IChangeSet<Animal>>> GetObservableObservable() => GetEnumerableObservable().ToObservable();
+
+    [Fact]
+    public void WithAComparerDoesNotRecurse()
+    {
+        var completed = false;
+
+        using var sources = new SourceList<IObservable<IChangeSet<int, int>>>();
+
+        // This overload used to bind to itself and exhaust the stack before returning.
+        using var subscription = sources.Connect()
+            .MergeChangeSets(Comparer<int>.Default)
+            .Subscribe(_ => { }, () => completed = true);
+
+        sources.Dispose();
+
+        completed.Should().BeTrue();
+    }
 }
diff --git a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs
index ea8019ea4..f8fbed867 100644
--- a/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs
+++ b/src/DynamicData.Tests/List/MergeManyChangeSetsFixture.cs
@@ -1,4 +1,9 @@
+using System;
+using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -52,4 +57,19 @@ public void MergeManyShouldWork()
 
         new[] { 2, 100, 10,11,12,13,14 }.Should().BeEquivalentTo(d.Items);
     }
+
+    [Fact]
+    public void DeliversTheErrorWithoutThrowing()
+    {
+        Exception? error = null;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source
+            .MergeManyChangeSets(_ => Observable.Empty<IChangeSet<Person>>(), EqualityComparer<Person>.Default)
+            .Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/MergeManyFixture.cs b/src/DynamicData.Tests/List/MergeManyFixture.cs
index 997e91f9f..438f2e845 100644
--- a/src/DynamicData.Tests/List/MergeManyFixture.cs
+++ b/src/DynamicData.Tests/List/MergeManyFixture.cs
@@ -1,7 +1,9 @@
-using System;
+using System;
+using System.Collections.Generic;
 using System.Reactive.Linq;
 using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -114,7 +116,7 @@ public void MergedStreamCompletesWhenSourceAndItemsComplete()
     /// Stream completes even if one of the children fails.
     /// </summary>
     [Fact]
-    public void MergedStreamCompletesIfLastItemFails()
+    public void MergedStreamFailsIfLastItemFails()
     {
         var receivedError = default(Exception);
         var streamCompleted = false;
@@ -129,9 +131,9 @@ public void MergedStreamCompletesIfLastItemFails()
         _source.Dispose();
         item.FailObservable(new Exception("Test exception"));
 
-        receivedError.Should().Be(default);
         sourceCompleted.Should().BeTrue();
-        streamCompleted.Should().BeTrue();
+        receivedError.Should().NotBeNull("Merge propagates a failure from any inner stream");
+        streamCompleted.Should().BeFalse("a failure and a completion are mutually exclusive");
     }
 
     /// <summary>
@@ -174,4 +176,19 @@ public void InvokeObservable(bool value)
             _changed.OnNext(value);
         }
     }
+
+    [Fact]
+    public void DeliversAnErrorRaisedByAChild()
+    {
+        Exception? error = null;
+
+        using var source = new SourceList<Person>();
+        using var child = new Subject<int>();
+        using var subscription = source.Connect().MergeMany(_ => child).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.Add(new Person("a", 1));
+        child.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>("Merge propagates a failure from any inner stream");
+    }
 }
diff --git a/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs b/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs
new file mode 100644
index 000000000..a5180a053
--- /dev/null
+++ b/src/DynamicData.Tests/List/MultiSourceSerializationFixture.cs
@@ -0,0 +1,246 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Reactive;
+using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using System.Reactive.Threading.Tasks;
+using System.Threading;
+using System.Threading.Tasks;
+
+using DynamicData.Binding;
+using DynamicData.Tests.Domain;
+using DynamicData.Tests.Utilities;
+
+using FluentAssertions;
+
+using Xunit;
+
+using Randomizer = Bogus.Randomizer;
+
+namespace DynamicData.Tests.List;
+
+/// <summary>
+/// Serialization coverage for the list operators that take more than one input. A single input is
+/// serialized by whatever feeds it, so these are the operators where two producers can reach the
+/// same observer at once, and where a missing gate shows up as overlapping delivery rather than as
+/// a wrong value. Every pipeline is checked both for overlap and for change set integrity, since
+/// list change sets carry indices and a torn read usually surfaces as a structurally invalid change
+/// set rather than as a visible race.
+/// </summary>
+public class MultiSourceSerializationFixture
+{
+    private const int Seed = 0x1147;
+
+    private static readonly TimeSpan DrainTimeout = TimeSpan.FromMinutes(2);
+
+    [Fact]
+    public async Task OrDeliversSeriallyWhileBothSourcesAreWritten()
+    {
+        using var left = new SourceList<Person>();
+        using var right = new SourceList<Person>();
+
+        await RunAsync(
+            left.Connect().Or(right.Connect()),
+            writers:
+            [
+                randomizer => left.Add(new Person("L" + randomizer.Int(1, 500), randomizer.Int(1, 80))),
+                randomizer => right.Add(new Person("R" + randomizer.Int(1, 500), randomizer.Int(1, 80))),
+                randomizer =>
+                {
+                    var items = left.Items.ToArray();
+                    if (items.Length > 0)
+                    {
+                        left.Remove(items[randomizer.Int(0, items.Length - 1)]);
+                    }
+                },
+            ],
+            complete: () =>
+            {
+                left.Dispose();
+                right.Dispose();
+            });
+    }
+
+    [Fact]
+    public async Task SortDeliversSeriallyWhileTheComparerChanges()
+    {
+        using var source = new SourceList<Person>();
+        using var comparers = new Subject<IComparer<Person>>();
+        using var resort = new Subject<Unit>();
+
+        var byAge = SortExpressionComparer<Person>.Ascending(p => p.Age);
+        var byName = SortExpressionComparer<Person>.Ascending(p => p.Name);
+
+        await RunAsync(
+            source.Connect().Sort(comparers, resetThreshold: 25, resort: resort),
+            writers:
+            [
+                randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), randomizer.Int(1, 80))),
+                randomizer => comparers.OnNext(randomizer.Bool() ? byAge : byName),
+                _ => resort.OnNext(Unit.Default),
+            ],
+            complete: () =>
+            {
+                comparers.OnCompleted();
+                resort.OnCompleted();
+                source.Dispose();
+            },
+            seedAction: () => comparers.OnNext(byAge));
+    }
+
+    [Fact]
+    public async Task GroupOnDeliversSeriallyWhileRegroupingConcurrently()
+    {
+        using var source = new SourceList<Person>();
+        using var regroup = new Subject<Unit>();
+
+        await RunAsync(
+            source.Connect().GroupOn(p => p.Age % 5, regroup).Transform(group => new Person(group.GroupKey.ToString(), group.GroupKey)),
+            writers:
+            [
+                randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), randomizer.Int(1, 80))),
+                randomizer =>
+                {
+                    var items = source.Items.ToArray();
+                    if (items.Length > 0)
+                    {
+                        source.Remove(items[randomizer.Int(0, items.Length - 1)]);
+                    }
+                },
+                _ => regroup.OnNext(Unit.Default),
+            ],
+            complete: () =>
+            {
+                regroup.OnCompleted();
+                source.Dispose();
+            });
+    }
+
+    [Fact]
+    public async Task BufferIfDeliversSeriallyWhilePausingConcurrently()
+    {
+        using var source = new SourceList<Person>();
+        using var pause = new Subject<bool>();
+
+        await RunAsync(
+            source.Connect().BufferIf(pause),
+            writers:
+            [
+                randomizer => source.Add(new Person("P" + randomizer.Int(1, 500), randomizer.Int(1, 80))),
+                randomizer =>
+                {
+                    var items = source.Items.ToArray();
+                    if (items.Length > 0)
+                    {
+                        source.Remove(items[randomizer.Int(0, items.Length - 1)]);
+                    }
+                },
+                randomizer => pause.OnNext(randomizer.Bool()),
+            ],
+            complete: () =>
+            {
+                // Leave the buffer open, so anything held back has to be flushed on completion.
+                pause.OnNext(false);
+                pause.OnCompleted();
+                source.Dispose();
+            });
+    }
+
+    [Fact]
+    public async Task MergeManyChangeSetsDeliversSeriallyWhileChildrenAreWritten()
+    {
+        using var owners = new SourceList<AnimalOwner>();
+        var created = new List<AnimalOwner>();
+
+        for (var i = 0; i < 5; i++)
+        {
+            var owner = new AnimalOwner("Owner" + i);
+            created.Add(owner);
+            owners.Add(owner);
+        }
+
+        var nextName = 0;
+
+        await RunAsync(
+            owners.Connect().MergeManyChangeSets(o => o.Animals.Connect()).Transform(a => new Person(a.Name, a.Name.Length)),
+            writers:
+            [
+                randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("A" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Mammal)),
+                randomizer =>
+                {
+                    var owner = created[randomizer.Int(0, created.Count - 1)];
+                    var animals = owner.Animals.Items.ToArray();
+                    if (animals.Length > 0)
+                    {
+                        owner.Animals.Remove(animals[randomizer.Int(0, animals.Length - 1)]);
+                    }
+                },
+                randomizer => created[randomizer.Int(0, created.Count - 1)].Animals.Add(new Animal("B" + Interlocked.Increment(ref nextName), "Type", AnimalFamily.Bird)),
+            ],
+            complete: () =>
+            {
+                // The merged result completes only once the parent and every child have.
+                foreach (var owner in created)
+                {
+                    owner.Dispose();
+                }
+
+                owners.Dispose();
+            });
+    }
+
+    /// <summary>
+    /// Drives every writer from its own thread against a shared barrier, then waits for the pipeline
+    /// to drain and asserts that it ended cleanly. A serialization failure surfaces as an
+    /// <c>UnsynchronizedNotificationException</c> on the terminal notification rather than as a
+    /// wrong count, so the assertion is on how the sequence ended.
+    /// </summary>
+    private static async Task RunAsync(
+        IObservable<IChangeSet<Person>> pipeline,
+        Action<Randomizer>[] writers,
+        Action complete,
+        Action? seedAction = null)
+    {
+        var randomizer = new Randomizer(Seed);
+        var iterations = randomizer.Int(150, 250);
+
+        var published = pipeline
+            .ValidateSynchronization()
+            .ValidateChangeSets()
+
+            // Holding each notification briefly makes an overlapping delivery observable instead of
+            // something that has to be caught inside a window of a few instructions.
+            .Do(static _ => Thread.SpinWait(500))
+            .Publish();
+
+        var terminal = published.Materialize().LastAsync().ToTask();
+        using var subscription = published.Connect();
+
+        seedAction?.Invoke();
+
+        using var barrier = new Barrier(writers.Length + 1);
+
+        var tasks = writers.Select((writer, index) => Task.Run(() =>
+        {
+            var threadRandomizer = new Randomizer(Seed + index + 1);
+            barrier.SignalAndWait();
+
+            for (var i = 0; i < iterations; i++)
+            {
+                writer(threadRandomizer);
+            }
+        })).ToArray();
+
+        barrier.SignalAndWait();
+        await Task.WhenAll(tasks);
+
+        complete();
+
+        (await Task.WhenAny(terminal, Task.Delay(DrainTimeout))).Should().BeSameAs(terminal, "the pipeline should drain rather than deadlock");
+
+        var lastNotification = await terminal;
+
+        lastNotification.Exception.Should().BeNull("notifications must not overlap, and every change set must be structurally valid, however the writers interleave");
+    }
+}
diff --git a/src/DynamicData.Tests/List/OrFixture.cs b/src/DynamicData.Tests/List/OrFixture.cs
index c4c973aa2..91c5fafe7 100644
--- a/src/DynamicData.Tests/List/OrFixture.cs
+++ b/src/DynamicData.Tests/List/OrFixture.cs
@@ -1,7 +1,9 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -136,4 +138,34 @@ public void RemovedWhenNoLongerInEither()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Or(second).Subscribe(_ => { }, () => completed = true);
+
+        first.OnCompleted();
+        completed.Should().BeFalse("the second source is still live");
+
+        second.OnCompleted();
+        completed.Should().BeTrue("every source has now finished");
+    }
+
+    [Fact]
+    public void DeliversAnErrorFromAnySource()
+    {
+        Exception? error = null;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Or(second).Subscribe(_ => { }, ex => error = ex, () => { });
+
+        second.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs
index 49c5cd27e..569451f64 100644
--- a/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs
+++ b/src/DynamicData.Tests/List/QueryWhenChangedFixture.cs
@@ -1,4 +1,6 @@
-using System;
+using System;
+using System.Collections.Generic;
+using System.Reactive.Subjects;
 
 using DynamicData.Tests.Domain;
 
@@ -89,4 +91,30 @@ public void Dispose()
         _source.Dispose();
         _results.Dispose();
     }
+
+    [Fact]
+    public void CompletesWhenTheSourceCompletes()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue();
+    }
+
+    [Fact]
+    public void DeliversTheError()
+    {
+        Exception? error = null;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.QueryWhenChanged().Subscribe(_ => { }, ex => error = ex, () => { });
+
+        source.OnError(new InvalidOperationException("boom"));
+
+        error.Should().BeOfType<InvalidOperationException>();
+    }
 }
diff --git a/src/DynamicData.Tests/List/SortFixture.cs b/src/DynamicData.Tests/List/SortFixture.cs
index 7f2a7dea3..0fe9a3e17 100644
--- a/src/DynamicData.Tests/List/SortFixture.cs
+++ b/src/DynamicData.Tests/List/SortFixture.cs
@@ -1,6 +1,7 @@
 using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Linq;
 using System.Reactive.Subjects;
 using DynamicData.Binding;
 using DynamicData.Tests.Domain;
@@ -182,4 +183,19 @@ public void SortInitialBatch()
 
         actualResult.Should().BeEquivalentTo(expectedResult);
     }
+
+    [Fact]
+    public void CompletesWhenGivenAComparerObservable()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source
+            .Sort(Observable.Return(SortExpressionComparer<Person>.Ascending(p => p.Name)))
+            .Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("an absent resort signal can never fire and so must not hold the result open");
+    }
 }
diff --git a/src/DynamicData.Tests/List/ToCollectionFixture.cs b/src/DynamicData.Tests/List/ToCollectionFixture.cs
index a46aa790f..05a2dd6a5 100644
--- a/src/DynamicData.Tests/List/ToCollectionFixture.cs
+++ b/src/DynamicData.Tests/List/ToCollectionFixture.cs
@@ -1,6 +1,9 @@
 using System;
 using System.Collections.Generic;
 using System.Reactive.Linq;
+using System.Reactive.Subjects;
+using DynamicData.Binding;
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 using Xunit;
 
@@ -23,4 +26,32 @@ public void ToCollectionTest()
         res1?.Count.Should().Be(2);
         res2?.Count.Should().Be(2);
     }
+
+    [Fact]
+    public void CompletesWhenTheSourceCompletes()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source.ToCollection().Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue("ToCollection is built on QueryWhenChanged");
+    }
+
+    [Fact]
+    public void ToSortedCollectionCompletesWhenTheSourceCompletes()
+    {
+        var completed = false;
+
+        using var source = new Subject<IChangeSet<Person>>();
+        using var subscription = source
+            .ToSortedCollection(SortExpressionComparer<Person>.Ascending(p => p.Name))
+            .Subscribe(_ => { }, () => completed = true);
+
+        source.OnCompleted();
+
+        completed.Should().BeTrue();
+    }
 }
diff --git a/src/DynamicData.Tests/List/XOrFixture.cs b/src/DynamicData.Tests/List/XOrFixture.cs
index 8f66b0d20..f52b6a6c4 100644
--- a/src/DynamicData.Tests/List/XOrFixture.cs
+++ b/src/DynamicData.Tests/List/XOrFixture.cs
@@ -1,7 +1,9 @@
-using System;
+using System;
 using System.Collections.Generic;
 using System.Linq;
+using System.Reactive.Subjects;
 
+using DynamicData.Tests.Domain;
 using FluentAssertions;
 
 using Xunit;
@@ -108,4 +110,20 @@ public void RemovedWhenNoLongerInEither()
     }
 
     protected abstract IObservable<IChangeSet<int>> CreateObservable();
+
+    [Fact]
+    public void CompletesOnlyWhenEverySourceCompletes()
+    {
+        var completed = false;
+
+        using var first = new Subject<IChangeSet<Person>>();
+        using var second = new Subject<IChangeSet<Person>>();
+        using var subscription = first.Xor(second).Subscribe(_ => { }, () => completed = true);
+
+        first.OnCompleted();
+        completed.Should().BeFalse("the second source is still live");
+
+        second.OnCompleted();
+        completed.Should().BeTrue("every source has now finished");
+    }
 }
diff --git a/src/DynamicData/List/Internal/BufferIf.cs b/src/DynamicData/List/Internal/BufferIf.cs
index 8f7d1993c..cb51c99fa 100644
--- a/src/DynamicData/List/Internal/BufferIf.cs
+++ b/src/DynamicData/List/Internal/BufferIf.cs
@@ -31,21 +31,25 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
 
                 var bufferSelector = Observable.Return(initialPauseState).Concat(_pauseIfTrueSelector.Merge(timeoutSubject)).ObserveOn(_scheduler).Synchronize(locker).Publish();
 
-                var pause = bufferSelector.Where(state => state).Subscribe(
-                    _ =>
+                // Handled as a single subscription rather than two filtered ones, so that a failure
+                // of the pause selector has one unambiguous path to the observer instead of either
+                // arriving twice or, as before, escaping unhandled.
+                var pauseOrResume = bufferSelector.Subscribe(
+                    state =>
                     {
-                        paused = true;
-
-                        // add pause timeout if required
-                        if (_timeOut != TimeSpan.Zero)
+                        if (state)
                         {
-                            timeoutSubscriber.Disposable = Observable.Timer(_timeOut, _scheduler).Select(_ => false).SubscribeSafe(timeoutSubject);
+                            paused = true;
+
+                            // add pause timeout if required
+                            if (_timeOut != TimeSpan.Zero)
+                            {
+                                timeoutSubscriber.Disposable = Observable.Timer(_timeOut, _scheduler).Select(static _ => false).SubscribeSafe(timeoutSubject);
+                            }
+
+                            return;
                         }
-                    });
 
-                var resume = bufferSelector.Where(state => !state).Subscribe(
-                    _ =>
-                    {
                         paused = false;
 
                         // publish changes and clear buffer
@@ -59,7 +63,8 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
 
                         // kill off timeout if required
                         timeoutSubscriber.Disposable = Disposable.Empty;
-                    });
+                    },
+                    observer.OnError);
 
                 var updateSubscriber = _source.Synchronize(locker).Subscribe(
                     updates =>
@@ -72,6 +77,18 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                         {
                             observer.OnNext(updates);
                         }
+                    },
+                    observer.OnError,
+                    () =>
+                    {
+                        // Anything still buffered would otherwise be lost, so flush before finishing.
+                        if (buffer.Count > 0)
+                        {
+                            observer.OnNext(buffer);
+                            buffer = [];
+                        }
+
+                        observer.OnCompleted();
                     });
 
                 var connected = bufferSelector.Connect();
@@ -80,8 +97,7 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                     () =>
                     {
                         connected.Dispose();
-                        pause.Dispose();
-                        resume.Dispose();
+                        pauseOrResume.Dispose();
                         updateSubscriber.Dispose();
                         timeoutSubject.OnCompleted();
                         timeoutSubscriber.Dispose();
diff --git a/src/DynamicData/List/Internal/Combiner.cs b/src/DynamicData/List/Internal/Combiner.cs
index 64882a3ba..f74e66941 100644
--- a/src/DynamicData/List/Internal/Combiner.cs
+++ b/src/DynamicData/List/Internal/Combiner.cs
@@ -27,6 +27,14 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
 
                 var resultList = new ChangeAwareListWithRefCounts<T>();
 
+                // Merging semantics: the result finishes only once every source has.
+                var pending = _source.Count;
+                if (pending == 0)
+                {
+                    observer.OnCompleted();
+                    return disposable;
+                }
+
                 lock (_locker)
                 {
                     var sourceLists = Enumerable.Range(0, _source.Count).Select(_ => new ReferenceCountTracker<T>()).ToList();
@@ -44,6 +52,14 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                                     {
                                         observer.OnNext(notifications);
                                     }
+                                },
+                                observer.OnError,
+                                () =>
+                                {
+                                    if (Interlocked.Decrement(ref pending) == 0)
+                                    {
+                                        observer.OnCompleted();
+                                    }
                                 }));
                     }
                 }
diff --git a/src/DynamicData/List/Internal/DynamicCombiner.cs b/src/DynamicData/List/Internal/DynamicCombiner.cs
index bb3091378..73316d4e6 100644
--- a/src/DynamicData/List/Internal/DynamicCombiner.cs
+++ b/src/DynamicData/List/Internal/DynamicCombiner.cs
@@ -1,4 +1,4 @@
-// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
+// Copyright (c) 2011-2025 Roland Pheasant. All rights reserved.
 // Roland Pheasant licenses this file to you under the MIT license.
 // See the LICENSE file in the project root for full license information.
 
@@ -40,7 +40,9 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                         {
                             observer.OnNext(notifications);
                         }
-                    });
+                    },
+                    observer.OnError,
+                    observer.OnCompleted);
 
                 // When a list is removed, update all items that were in that list
                 var removedItem = sourceLists.Connect().OnItemRemoved(
@@ -63,7 +65,8 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                                 observer.OnNext(notification2);
                             }
                         }
-                    }).Subscribe();
+                    })
+                    .Subscribe(static _ => { }, static _ => { });
 
                 // When a list is added, update all items that are in that list
                 var sourceChanged = sourceLists.Connect().WhereReasonsAre(ListChangeReason.Add, ListChangeReason.AddRange).ForEachItemChange(
@@ -84,7 +87,8 @@ public IObservable<IChangeSet<T>> Run() => Observable.Create<IChangeSet<T>>(
                                 observer.OnNext(notification2);
                             }
                         }
-                    }).Subscribe();
+                    })
+                    .Subscribe(static _ => { }, static _ => { });
 
                 return new CompositeDisposable(sourceLists, allChanges, removedItem, sourceChanged);
             });
diff --git a/src/DynamicData/List/Internal/GroupOn.cs b/src/DynamicData/List/Internal/GroupOn.cs
index 6ea7deccd..620b05896 100644
--- a/src/DynamicData/List/Internal/GroupOn.cs
+++ b/src/DynamicData/List/Internal/GroupOn.cs
@@ -32,8 +32,9 @@ public IObservable<IChangeSet<IGroup<TObject, TGroupKey>>> Run() => Observable.C
 
                 var grouper = shared.Select(changes => Process(groupings, groupCache, changes));
 
+                // An absent regrouper never fires, so Empty rather than Never keeps the merge able to complete.
                 var regrouperFunc = _regrouper is null ?
-                    Observable.Never<IChangeSet<IGroup<TObject, TGroupKey>>>() :
+                    Observable.Empty<IChangeSet<IGroup<TObject, TGroupKey>>>() :
                     _regrouper.Synchronize(locker).CombineLatest(shared.ToCollection(), (_, collection) => Regroup(groupings, groupCache, collection));
 
                 var publisher = grouper.Merge(regrouperFunc).DisposeMany() // dispose removes as the grouping is disposable
diff --git a/src/DynamicData/List/Internal/GroupOnImmutable.cs b/src/DynamicData/List/Internal/GroupOnImmutable.cs
index 2b4a08be8..45a03aeeb 100644
--- a/src/DynamicData/List/Internal/GroupOnImmutable.cs
+++ b/src/DynamicData/List/Internal/GroupOnImmutable.cs
@@ -35,8 +35,9 @@ public IObservable<IChangeSet<IGrouping<TObject, TGroupKey>>> Run() => Observabl
 
                 var grouper = shared.Select(changes => Process(groupings, groupCache, changes));
 
+                // An absent regrouper never fires, so Empty rather than Never keeps the merge able to complete.
                 var reGroupFunc = _reGrouper is null ?
-                    Observable.Never<IChangeSet<IGrouping<TObject, TGroupKey>>>() :
+                    Observable.Empty<IChangeSet<IGrouping<TObject, TGroupKey>>>() :
                     _reGrouper.Synchronize(locker).CombineLatest(shared.ToCollection(), (_, collection) => Regroup(groupings, groupCache, collection));
 
                 var publisher = grouper.Merge(reGroupFunc).NotEmpty().SubscribeSafe(observer);
diff --git a/src/DynamicData/List/Internal/MergeMany.cs b/src/DynamicData/List/Internal/MergeMany.cs
index 65b238068..40cb63e23 100644
--- a/src/DynamicData/List/Internal/MergeMany.cs
+++ b/src/DynamicData/List/Internal/MergeMany.cs
@@ -24,7 +24,7 @@ public IObservable<TDestination> Run() => Observable.Create<TDestination>(
                                                 .SubscribeMany(t =>
                                                 {
                                                     counter.Added();
-                                                    return _observableSelector(t).Synchronize(locker).Finally(() => counter.Finally()).Subscribe(observer.OnNext, _ => { }, () => { });
+                                                    return _observableSelector(t).Synchronize(locker).Finally(() => counter.Finally()).Subscribe(observer.OnNext, observer.OnError, () => { });
                                                 })
                                                 .Subscribe(_ => { }, observer.OnError, observer.OnCompleted);
 
diff --git a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs
index 3b63ea231..238614ff8 100644
--- a/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs
+++ b/src/DynamicData/List/Internal/MergeManyCacheChangeSets.cs
@@ -51,7 +51,7 @@ public IObservable<IChangeSet<TDestination, TDestinationKey>> Run() => Observabl
                     changeTracker.EmitChanges(observer);
                     parentUpdate = false;
                 })
-                .Subscribe();
+                .Subscribe(static _ => { }, static _ => { });
 
             return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove);
         });
diff --git a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs
index 482c4f46b..5c52f6729 100644
--- a/src/DynamicData/List/Internal/MergeManyListChangeSets.cs
+++ b/src/DynamicData/List/Internal/MergeManyListChangeSets.cs
@@ -47,7 +47,7 @@ public IObservable<IChangeSet<TDestination>> Run() => Observable.Create<IChangeS
                     changeTracker.EmitChanges(observer);
                     parentUpdate = false;
                 })
-                .Subscribe();
+                .Subscribe(static _ => { }, static _ => { });
 
             return new CompositeDisposable(shared.Connect(), subMergeMany, subRemove);
         });
diff --git a/src/DynamicData/List/Internal/QueryWhenChanged.cs b/src/DynamicData/List/Internal/QueryWhenChanged.cs
index ef58d4925..ab36511e6 100644
--- a/src/DynamicData/List/Internal/QueryWhenChanged.cs
+++ b/src/DynamicData/List/Internal/QueryWhenChanged.cs
@@ -11,14 +11,14 @@ internal sealed class QueryWhenChanged<T>(IObservable<IChangeSet<T>> source)
 {
     private readonly IObservable<IChangeSet<T>> _source = source ?? throw new ArgumentNullException(nameof(source));
 
-    public IObservable<IReadOnlyCollection<T>> Run() => Observable.Create<IReadOnlyCollection<T>>(observer =>
-                                                             {
-                                                                 var list = new List<T>();
+    public IObservable<IReadOnlyCollection<T>> Run() => Observable.Defer(() =>
+    {
+        var list = new List<T>();
 
-                                                                 return _source.Subscribe(changes =>
-                                                                 {
-                                                                     list.Clone(changes);
-                                                                     observer.OnNext(new ReadOnlyCollectionLight<T>(list));
-                                                                 });
-                                                             });
+        return _source.Select(changes =>
+        {
+            list.Clone(changes);
+            return (IReadOnlyCollection<T>)new ReadOnlyCollectionLight<T>(list);
+        });
+    });
 }
diff --git a/src/DynamicData/List/Internal/Sort.cs b/src/DynamicData/List/Internal/Sort.cs
index ff03e2e70..f366fe02d 100644
--- a/src/DynamicData/List/Internal/Sort.cs
+++ b/src/DynamicData/List/Internal/Sort.cs
@@ -10,8 +10,9 @@ namespace DynamicData.List.Internal;
 internal sealed class Sort<T>(IObservable<IChangeSet<T>> source, IComparer<T>? comparer, SortOptions sortOptions, IObservable<Unit>? resort, IObservable<IComparer<T>>? comparerObservable, int resetThreshold)
     where T : notnull
 {
-    private readonly IObservable<IComparer<T>> _comparerObservable = comparerObservable ?? Observable.Never<IComparer<T>>();
-    private readonly IObservable<Unit> _resort = resort ?? Observable.Never<Unit>();
+    // An absent comparer or resort signal will never fire, so it must not hold the merge open.
+    private readonly IObservable<IComparer<T>> _comparerObservable = comparerObservable ?? Observable.Empty<IComparer<T>>();
+    private readonly IObservable<Unit> _resort = resort ?? Observable.Empty<Unit>();
     private readonly IObservable<IChangeSet<T>> _source = source ?? throw new ArgumentNullException(nameof(source));
 
     private IComparer<T> _comparer = comparer ?? Comparer<T>.Default;
diff --git a/src/DynamicData/List/ObservableListEx.BufferIf.cs b/src/DynamicData/List/ObservableListEx.BufferIf.cs
index 37224290f..0df446229 100644
--- a/src/DynamicData/List/ObservableListEx.BufferIf.cs
+++ b/src/DynamicData/List/ObservableListEx.BufferIf.cs
@@ -76,8 +76,8 @@ public static IObservable<IChangeSet<T>> BufferIf<T>(this IObservable<IChangeSet
     /// <item><term>Any (while active)</term><description>Passed through immediately.</description></item>
     /// <item><term>Pause selector emits false</term><description>All buffered changesets are flushed downstream as one combined changeset.</description></item>
     /// <item><term>Timeout fires</term><description>Automatically resumes and flushes the buffer.</description></item>
-    /// <item><term>OnError</term><description>Forwarded immediately (not buffered).</description></item>
-    /// <item><term>OnCompleted</term><description>Forwarded immediately.</description></item>
+    /// <item><term>OnError</term><description>Forwarded immediately (not buffered), and buffered data is lost. An error from <paramref name="pauseIfTrueSelector"/> terminates the output the same way a source error does.</description></item>
+    /// <item><term>OnCompleted</term><description>Any remaining buffered changesets are flushed before completion is forwarded.</description></item>
     /// </list>
     /// <para><b>Worth noting:</b> Each pause/resume cycle re-arms the timeout. Rapid toggling can create many small buffer windows.</para>
     /// </remarks>
diff --git a/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs b/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs
index 4f58c05cc..42af3ba92 100644
--- a/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs
+++ b/src/DynamicData/List/ObservableListEx.MergeChangeSets.cs
@@ -196,7 +196,9 @@ public static IObservable<IChangeSet<TObject, TKey>> MergeChangeSets<TObject, TK
     {
         comparer.ThrowArgumentNullExceptionIfNull(nameof(comparer));
 
-        return source.MergeChangeSets(comparer);
+        // Named arguments are required: a lone IComparer<TObject> is an exact match for this method's own
+        // signature, so an unnamed call binds here again and recurses until the stack is exhausted.
+        return source.MergeChangeSets(equalityComparer: null, comparer: comparer);
     }
 
     /// <inheritdoc cref="MergeChangeSets{TObject, TKey}(IObservableList{IObservable{IChangeSet{TObject, TKey}}}, IComparer{TObject})"/>
diff --git a/src/DynamicData/List/ObservableListEx.MergeMany.cs b/src/DynamicData/List/ObservableListEx.MergeMany.cs
index 55dfa4f19..a09b56f77 100644
--- a/src/DynamicData/List/ObservableListEx.MergeMany.cs
+++ b/src/DynamicData/List/ObservableListEx.MergeMany.cs
@@ -40,6 +40,7 @@ public static partial class ObservableListEx
     /// <item><term><b>Replace</b></term><description>Old subscription disposed, new subscription created for the replacement item.</description></item>
     /// <item><term><b>Remove</b>/<b>RemoveRange</b>/<b>Clear</b></term><description>Subscription disposed.</description></item>
     /// <item><term><b>Refresh</b>/<b>Moved</b></term><description>No effect on subscriptions.</description></item>
+    /// <item><term>OnError</term><description>An error from a per-item observable, or from the source, terminates the merged output.</description></item>
     /// <item><term>OnCompleted (source)</term><description>Completes only after the source and all active inner observables have completed.</description></item>
     /// </list>
     /// </remarks>