Ix.Net ToObservableObservable throwing System.NotSupportedException: Specified method is not supported.

[cowlinb6]

When using an IObservable we are intermittently seeing exceptions like:

at System.Linq.AsyncEnumerable.ToObservableObservable1.<>c__DisplayClass2_0.<<Subscribe>g__Core|0>d.MoveNext() in /_/Ix.NET/Source/System.Linq.Async/System/Linq/Operators/ToObservable.cs:line 74" at Cloud.Common.Database.DocumentDbRepositoryBase1.GetItemsStream[T](String sqlString, CancellationToken cancellationToken, Int32 batchSize, Int32 priority)+System.IAsyncDisposable.DisposeAsync()"
Unhandled exception. System.NotSupportedException: Specified method is not supported.

Assembly: System.Linq.Asynq
Version: 6.0.3

Not able to reliable reproduce this, but the basic usage pattern we have is:

    IObservable<SomeType> source = ..

    var reportData = source
        .TakeWhile(_ => !cts.Token.IsCancellationRequested)
        .Do(_ =>
        {
            count++;

            if (count > settings.RowLimit)
            {
                cts.Cancel();
            }
        })
        .Select(call => Process(call,))
        .Where(filterFuncSummary)
        .ToList()
        .Subscribe(
            tcs.SetResult,
            tcs.SetException,
            () => tcs.TrySetResult([])
        );

    var results = tcs.Task.Result;

This is running in a Linux container (seen in Docker and in Kubernetes) running Alpine 3.2.1.

We don't seem to be able to catch this so it is crashing the pod.

Is this something we should be handling or an issue in the library?

Thanks

[fedeAlterio]

I see you are converting from IAsyncEnumerable<T> to IObservable<T> and vice versa using ToObservable() and ToAsyncEnumerable().

Both of conversion methods currently have some issues.
ToObservable can indeed throw an unhandled exception if the exception is thrown on DisposeAsync(). Since its unhandled, the default behavior is to terminate the process crashing the pod. See this PR.

ToAsyncEnumerable() seems to have a rare race condition on Cancellation/Dispose. See this pr.

In your stacktrace I see the exception is thrown on DisposeAsync() so probably the first Fix can prevent at least the pod to crash.

Maybe the second fix can prevent the exception to be thrown on DisposeAsync() also

Hope it helps :)

[cowlinb6]

Thanks, worked around this with your suggestion.