@dotnet-policy-service agree

Hi @idg10,

Apologies for the ping! Could you take a look at the fix when you have a chance? This bug is quite severe as it's causing the process to completely shut down. We’ve implemented internally a custom alternative as a .ToObservableSafe(), but it’s challenging to enforce this version consistently due to the presence of .ToObservable within System.Linq that seems really authoritative.
Thank you very much for your help!

Sorry for the delay - many of us at endjin have been unwell, and since nobody pays us to maintain Rx (and we only maintain System.Linq.Async because for historical reasons, it lives in the same repo) it tends to drop to the back of the queue under these circumstances.

Looking at this, I think there are 4 obvious options when GetAsyncEnumerator throws:

1. terminate the process abruptly (not good, but apparently the current behaviour)
2. throw the exception out of the call to Subscribe
3. forward the exception to the subscriber's OnError
4. quietly swallow the exceptions.

Although 2 is arguably logically like the right thing to do (because this is a case of failing even to get started) in practice this would involve sync-over-async, so we can't do that. For AsyncRx, SubscribeAsync could actually do this, but then you'd have a problem in which the two libraries ended up handling the same error quite differently.

So it does seem like 3 is the only reasonable option left if we want to report the exception. I see you've also implemented 4, allowing people to say they'd like exceptions to be swallowed. That's the one part of this I'm not sure about. I'm glad you've not made it the default, but I'm not sure whether we really want this at all. It seems to enable a bad practice.

What was your rationale for allowing ignoreExceptionsAfterUnsubscribe: true?

Thank you for your answer.
I was trying to be consistent with Observable.FromAsync() behavior, but maybe i misunderstood it: https://github.com/dotnet/reactive/blob/7259cabf5456a21462caea8cb81da8e5d5405db2/Rx.NET/Source/src/System.Reactive/Concurrency/TaskObservationOptions.cs#L27C4-L50C28

Maybe we could just handle DisposeAsync exceptions like other exceptions? So invoking observer.OnError if the cancellation token is not canceled yet?

Ah yes, I remember that FromAsync change now. With that one, we added the flag so that the existing behaviour would continue to occur unless you asked for the new behaviour - it was essentially there as a backwards compatibility feature.

So firstly, I think that in the case where a subscriber is still attached we definitely want to report it to OnError. (And I believe that's what you've done.)

I'm now looking at this code and I'm a bit confused:

try
{
    e = _source.GetAsyncEnumerator(ctd.Token);
}
catch (Exception ex)
{
    if (!ctd.Token.IsCancellationRequested)
    {
        observer.OnError(ex);
    }

    return;
}

If if block. of ctd.Token.IsCancellationRequested is true, that means the called has unsubscribed. So in that case, we are definitely "after unsubscription" and it looks like we just swallow the exception in that case. But I'm not sure this can ever actually occur: this async method hasn't done an await yet, and GetAsyncEnumerator itself is not async, so I don't think there's any way we can have returned the CancellableDisposable to the called by the time we reach that if can we?

So I think we'd want to make the call to OnError unconditional wouldn't we?

But if I'm wrong about that, that implies that this is a second "after unsubscription" case, in which case we'd need to make this also pay attention to that 'ignore' flag.

I agree that the check for unsubscription is useless if the exception is thrown synchronously. Thank you!

About that ignoreExceptionsAfterUnsubscribe flag, I think there are 2 paths:

• We can remove it completely, swallowing any exception in case the unsubscription already happened (This would be similar to the current implementation)
• We can add it

If we choose the latter, in the case the subscription is already disposed:

• If the flag is falsemaybe we should rethrow all the exceptions (throw instead return in every catch clause). The Core method should also be a ValueTask so that exceptions will not cause the process to crash, but will be routed to TaskScheduler.UnobservedTaskException
• if it is true, we should just return (similarly at what happens now)

Consider also that following that approach, if false is the default, we would be inserting a breaking change (TaskScheduler.UnobservedTaskException will now be called)

Also note with the current implementation, the process crashes also if any exception is thrown by DisposeAsync()

We probably won't be merging this now, because of dotnet/runtime#79782 - Async LINQ is going to be built into .NET from v10, at which point the Async LINQ library in this repo will be deprecated.

I don't believe they are including ToObservable in that implementation, because they are just providing core LINQ features, and not Rx support. So we will actually need to find a new home for this extension. I think it might actually make sense for this to be in the main Rx.NET implementation, now that IAsyncEnumerable is built into .NET. Although that might not accommodate .NET FX users. This needs some thought, but I don't think the form it's in now is the right one.

See #2199

In my opinion it would be great to put this conversion method in .NET.
IObservable interface is already there and
Something similar already exists for dataFlow blocks.