List operators drop OnCompleted and mishandle OnError

Start by reproducing the ToCollection() completion failure described in the issue, then trace the affected operators and overloads listed under “Never completes” and “Error path” to their implementations. The issue provides a harness method that exercises empty and one-changeset sources with completion and error signals; use it to verify each behavior. Done means the listed overloads forward terminal events correctly and regression tests cover the cases.

While fixing Switch (#1136, #1137) I wanted to know whether that was a one off or the tip of something, so I wrote a harness that drives every operator overload with a source that either completes or errors, and records what actually comes out the other end.

It reached 476 of the 499 extension methods that hang off IObservable. The rest do not return an observable (AsObservableCache, the Bind overloads returning IDisposable, ChangeSetAggregator and so on) so there is nothing to assert about.

Two things are worth saying before the list.

Overloads of the same operator do not behave alike. Sort is the clearest case:

source.Sort(comparer)                        // completes
source.Sort(Observable.Return(comparer))     // never completes
source.Sort(comparer, resorter)              // never completes

Testing one overload and assuming the rest are fine would have missed most of this, so every overload was driven separately.

"Never completes" means with an empty source. An operator that merges a stream per item is entitled to stay open while those streams are open, so I checked completion against a source that never produced an item. Nothing can legitimately block completion in that case. AutoRefresh, WhenAnyPropertyChanged and GroupOnPropertyWithImmutableState all complete on an empty source and are not listed here.

On the list side 65 overloads either never complete or get the error path wrong. Grouped by the implementation they share.

Repro

var source = new SourceList<Person>();
var completed = false;

using var sub = source.Connect()
    .ToCollection()
    .Subscribe(_ => { }, () => completed = true);

source.Dispose();

// completed is false

Never completes

BatchIf / BufferIf

BufferIf(IObservable<bool>, bool, IScheduler)
BufferIf(IObservable<bool>, bool, TimeSpan?, IScheduler)
BufferIf(IObservable<bool>, IScheduler)
BufferIf(IObservable<bool>, TimeSpan?, IScheduler)

Combiner (Combine)

And(params IObservable<IChangeSet<TObject, TKey>>[])
Except(params IObservable<IChangeSet<TObject, TKey>>[])
Or(params IObservable<IChangeSet<TObject, TKey>>[])
Xor(params IObservable<IChangeSet<TObject, TKey>>[])

DeferUntilLoaded

DeferUntilLoaded()
SkipInitial()

Grouping

GroupOn(Func<IObservable<IChangeSet<TObject,TKey>>,TObject>, IObservable<Unit>)
GroupOn(Func<IObservable<IChangeSet<TObject>>,TObject>, IObservable<Unit>)
GroupOn(Func<Object,TObject>, IObservable<Unit>)
GroupOn(Func<TObject,TObject>, IObservable<Unit>)
GroupOnProperty(Expression<Func<TObject,TObject>>, TimeSpan?, IScheduler)
GroupOnPropertyWithImmutableState(Expression<Func<TObject,TObject>>, TimeSpan?, IScheduler)
GroupWithImmutableState(Func<IObservable<IChangeSet<TObject,TKey>>,TKey>, IObservable<Unit>)
GroupWithImmutableState(Func<IObservable<IChangeSet<TObject>>,TKey>, IObservable<Unit>)
GroupWithImmutableState(Func<Object,TKey>, IObservable<Unit>)
GroupWithImmutableState(Func<TObject,TKey>, IObservable<Unit>)

MergeChangeSets

MergeChangeSets(IEnumerable<IObservable<IChangeSet<IObservable<IChangeSet<TObject,TKey>>>>>, IEqualityComparer<IObservable<IChangeSet<TObject,TKey>>>, IScheduler, bool)
MergeChangeSets(IEnumerable<IObservable<IChangeSet<IObservable<IChangeSet<TObject>>>>>, IEqualityComparer<IObservable<IChangeSet<TObject>>>, IScheduler, bool)
MergeChangeSets(IEnumerable<IObservable<IChangeSet<Object>>>, IEqualityComparer<Object>, IScheduler, bool)
MergeChangeSets(IEnumerable<IObservable<IChangeSet<TObject>>>, IEqualityComparer<TObject>, IScheduler, bool)
MergeChangeSets(IObservable<IChangeSet<IObservable<IChangeSet<TObject,TKey>>>>, IEqualityComparer<IObservable<IChangeSet<TObject,TKey>>>, IScheduler, bool)
MergeChangeSets(IObservable<IChangeSet<IObservable<IChangeSet<TObject>>>>, IEqualityComparer<IObservable<IChangeSet<TObject>>>, IScheduler, bool)
MergeChangeSets(IObservable<IChangeSet<Object>>, IEqualityComparer<Object>, IScheduler, bool)
MergeChangeSets(IObservable<IChangeSet<TObject>>, IEqualityComparer<TObject>, IScheduler, bool)

MonitorStatus

MonitorStatus()

QueryWhenChanged

QueryWhenChanged()
QueryWhenChanged(Func<IReadOnlyCollection<IObservable<IChangeSet<TObject,TKey>>>,TObject>)
QueryWhenChanged(Func<IReadOnlyCollection<IObservable<IChangeSet<TObject>>>,TObject>)
QueryWhenChanged(Func<IReadOnlyCollection<Object>,TObject>)
QueryWhenChanged(Func<IReadOnlyCollection<TObject>,TObject>)

Switch (already covered by #1137 / #1139)

Switch()

ToCollection / ToSortedCollection

ToCollection()
ToSortedCollection(Func<IObservable<IChangeSet<TObject,TKey>>,TKey>, SortDirection)
ToSortedCollection(Func<IObservable<IChangeSet<TObject>>,TKey>, SortDirection)
ToSortedCollection(Func<Object,TKey>, SortDirection)
ToSortedCollection(Func<TObject,TKey>, SortDirection)
ToSortedCollection(IComparer<IObservable<IChangeSet<TObject,TKey>>>)
ToSortedCollection(IComparer<IObservable<IChangeSet<TObject>>>)
ToSortedCollection(IComparer<Object>)
ToSortedCollection(IComparer<TObject>)

Error path

Aggregation Scan

Maximum(Func<IObservable<IChangeSet<TObject,TKey>>,int>, int)
Maximum(Func<IObservable<IChangeSet<TObject>>,int>, int)
Maximum(Func<Object,int>, int)
Maximum(Func<TObject,int>, int)
Minimum(Func<IObservable<IChangeSet<TObject,TKey>>,int>, int)
Minimum(Func<IObservable<IChangeSet<TObject>>,int>, int)
Minimum(Func<Object,int>, int)
Minimum(Func<TObject,int>, int)

MergeChangeSets

MergeChangeSets(IEqualityComparer<TObject>, IComparer<TObject>)

MergeManyChangeSets

MergeManyChangeSets(Func<IObservable<IChangeSet<TObject,TKey>>,IObservable<IChangeSet<TObject,TKey>>>, IComparer<TObject>)
MergeManyChangeSets(Func<IObservable<IChangeSet<TObject,TKey>>,IObservable<IChangeSet<TObject,TKey>>>, IEqualityComparer<TObject>, IComparer<TObject>)
MergeManyChangeSets(Func<IObservable<IChangeSet<TObject,TKey>>,IObservable<IChangeSet<TObject>>>, IEqualityComparer<TObject>)
MergeManyChangeSets(Func<IObservable<IChangeSet<TObject>>,IObservable<IChangeSet<TObject,TKey>>>, IComparer<TObject>)
MergeManyChangeSets(Func<IObservable<IChangeSet<TObject>>,IObservable<IChangeSet<TObject,TKey>>>, IEqualityComparer<TObject>, IComparer<TObject>)
MergeManyChangeSets(Func<IObservable<IChangeSet<TObject>>,IObservable<IChangeSet<TObject>>>, IEqualityComparer<TObject>)
MergeManyChangeSets(Func<Object,IObservable<IChangeSet<TObject,TKey>>>, IComparer<TObject>)
MergeManyChangeSets(Func<Object,IObservable<IChangeSet<TObject,TKey>>>, IEqualityComparer<TObject>, IComparer<TObject>)
MergeManyChangeSets(Func<Object,IObservable<IChangeSet<TObject>>>, IEqualityComparer<TObject>)
MergeManyChangeSets(Func<TObject,IObservable<IChangeSet<TObject,TKey>>>, IComparer<TObject>)
MergeManyChangeSets(Func<TObject,IObservable<IChangeSet<TObject,TKey>>>, IEqualityComparer<TObject>, IComparer<TObject>)
MergeManyChangeSets(Func<TObject,IObservable<IChangeSet<TObject>>>, IEqualityComparer<TObject>)

Why this matters

Anything that waits for a terminal event breaks against these: TakeUntil, Concat, LastAsync, ToTask, Amb. The completion pattern in the repo's own testing guidance (Publish, then LastOrDefaultAsync().ToTask(), then dispose the source) hangs forever rather than failing, so tests written that way around these operators are not asserting what they look like they assert.

The error cases are worse than the completion ones. Where an operator throws out of Subscribe instead of calling OnError, the exception surfaces on whichever thread happened to be writing to the source, which is usually nowhere near the subscription. Where the error is swallowed entirely the subscriber simply never learns the stream is dead.

Method

Each overload was subscribed and then given, in turn: OnCompleted with no items, OnError with no items, one changeset followed by OnCompleted, one changeset followed by OnError, and a dispose followed by more writes. Terminal events were delivered both during subscription and afterwards, since several of these only misbehave when the terminal event arrives synchronously, which is what Switch did.