List operators drop OnCompleted and mishandle OnError

[dwcullop]

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.

[dwcullop]

Found another one while wiring the audit into the test project, and this one is worse than anything in the list above because it takes the process with it.

ObservableListEx.MergeChangeSets.cs, line 193:

public static IObservable<IChangeSet<TObject, TKey>> MergeChangeSets<TObject, TKey>(this IObservable<IChangeSet<IObservable<IChangeSet<TObject, TKey>>>> source, IComparer<TObject> comparer)
    where TObject : notnull
    where TKey : notnull
{
    comparer.ThrowArgumentNullExceptionIfNull(nameof(comparer));

    return source.MergeChangeSets(comparer);
}

It calls itself. The intent was clearly the overload below it that takes an optional equality comparer and an optional comparer, but a lone IComparer<TObject> argument is an exact match for this method's own signature, so that is what it binds to. Calling it is an immediate stack overflow, which cannot be caught and kills the process rather than throwing.

The IObservableList overload at line 162 goes the same way, since it is just source.Connect().MergeChangeSets(comparer).

Repro, and this will take the test runner down with it:

var sources = new SourceList<IObservable<IChangeSet<Person, string>>>();

sources.Connect().MergeChangeSets(SomeComparer);

Naming the argument is enough to fix it:

return source.MergeChangeSets(equalityComparer: null, comparer: comparer);

I swept the rest of the library for the same shape, a public method whose body forwards to itself with the arguments unchanged, and this is the only one. There are around sixty other self-referencing delegations but they all reshape an argument on the way through, so they land on a different overload.

Both overloads are excluded from the conformance sweep for now, since a stack overflow cannot be trapped. The exclusion carries a comment pointing here, and removing it once this is fixed puts them back under test automatically.

[dwcullop]

Correction, and it is my fault rather than the library's.

Every MergeChangeSets entry in the list above is wrong. Those overloads take bool completable = true, and the harness was manufacturing a value for every parameter rather than honouring declared defaults, so it was passing false. That is an explicit request not to complete, and the operator was doing exactly what it was asked. All twelve cache entries now come back clean.

The harness now uses a parameter's default value wherever one is declared, which is the more faithful thing to do anyway since it tests the API the way it is documented. Re-running with that change moves the numbers:

• cache: 38 affected overloads, not 42
• list: 79 affected overloads, not 65

The list count went up because honouring defaults also stopped the harness from accidentally suppressing behaviour elsewhere, which surfaced Sort and GroupOn overloads it had previously bound differently.

One genuine MergeChangeSets defect does survive on the list side, and it is an error path rather than a completion one:

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

It throws out of Subscribe instead of delivering OnError. That is separate from the recursion reported above.

Sorry for the noise. The root cause table in the issue body still holds for everything else.