Cache operators drop OnCompleted and mishandle OnError

Start with the listed operator overloads grouped by implementation, especially BatchIf/BufferIf, Bind, Combine, Grouping, joins, MergeChangeSets, and the error-path cases. Reproduce the terminal-event behavior using the issue’s harness approach: test empty completion and error, then a changeset followed by each terminal event. Done means the affected overloads complete and propagate errors correctly, with tests that cover the cases described.

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 cache side 42 overloads either never complete or get the error path wrong. Grouped by the implementation they share, since several of these are one bug wearing different names.

Repro

var source = new SourceCache<Person, string>(p => p.Name);
var completed = false;

using var sub = source.Connect()
    .Or(Observable.Empty<IChangeSet<Person, string>>())
    .Subscribe(_ => { }, () => completed = true);

source.Dispose();

// completed is false

Never completes

BatchIf / BufferIf

BatchIf(IObservable<bool>, bool, IObservable<Unit>, IScheduler)
BatchIf(IObservable<bool>, bool, IScheduler)
BatchIf(IObservable<bool>, bool, TimeSpan?, IScheduler)
BatchIf(IObservable<bool>, IScheduler)
BatchIf(IObservable<bool>, TimeSpan?, IScheduler)

Bind (SortAndPage / SortAndVirtualize output)

Bind(IList<TObject>)
Bind(out ReadOnlyObservableCollection<TObject>)

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

Group(Func<TObject,TKey>, IObservable<IDistinctChangeSet<TKey>>)

Join family

FullJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<Optional<TObject>,IGrouping<TObject,TKey,TKey>,TObject>)
FullJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TKey,Optional<TObject>,IGrouping<TObject,TKey,TKey>,TObject>)
InnerJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TKey,TObject,IGrouping<TObject,TKey,TKey>,TObject>)
InnerJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TObject,IGrouping<TObject,TKey,TKey>,TObject>)
LeftJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TKey,TObject,IGrouping<TObject,TKey,TKey>,TObject>)
LeftJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TObject,IGrouping<TObject,TKey,TKey>,TObject>)
RightJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<Optional<TObject>,IGrouping<TObject,TKey,TKey>,TObject>)
RightJoinMany(IObservable<IChangeSet<TObject,TKey>>, Func<TObject,TKey>, Func<TKey,Optional<TObject>,IGrouping<TObject,TKey,TKey>,TObject>)

MergeChangeSets

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

MonitorStatus

MonitorStatus()

Sort (dynamic overloads only)

Sort(IComparer<TObject>, IObservable<Unit>, SortOptimisations, int)
Sort(IObservable<IComparer<TObject>>, SortOptimisations, int)

Switch (already covered by #1137 / #1139)

Switch()

Error path

Bind (SortAndPage / SortAndVirtualize output)

Bind(IList<TObject>, SortAndBindOptions)
Bind(out ReadOnlyObservableCollection<TObject>, SortAndBindOptions)

Grouping

Group(Func<TObject,TKey>)
Group(Func<TObject,TKey>, IObservable<Unit>)

MergeManyItems

MergeManyItems(Func<TObject,IObservable<TObject>>)
MergeManyItems(Func<TObject,TKey,IObservable<TObject>>)

TreeBuilder

SelectTree(Func<TObject,TKey>)
TransformToTree(Func<TObject,TKey>, IObservable<Func<Node<TObject,TKey>,bool>>)

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.