[Bug]: DisposeMany() suppresses exceptions

[seanreid1201]

Describe the bug 🐞

Calling DisposeMany() on an IObservable<IChangeSet> prevents exceptions propagating out to the caller.
See the test case provided below.
If the call to DisposeMany is omitted, the test passes.

Step to reproduce

[Fact]
public void ExceptionsArePropagated()
{
    var source = new SourceList<int>();
    source.Add(1);

    var transformed = source
        .Connect()
        .Transform<int, int>(_ => throw new InvalidCastException())
        .DisposeMany();

    FluentActions.Invoking(() => transformed.Subscribe()).Should().Throw<InvalidCastException>();
}

Reproduction repository

My work VPN prevents access to github via git

Expected behavior

When an exception is thrown from within the observable, and not handled by the caller, it should propagate out.

Screenshots 🖼️

No response

IDE

Visual Studio 2022

Operating system

No response

Version

No response

Device

No response

DynamicData Version

7.12.1

Additional information ℹ️

No response

[RolandPheasant]

Thanks for reporting this. I'll take a look soon

[seanreid1201]

I've taken some time to look into this.
In DynamicData.List.Internal.OnBeingRemoved<T>, line 35, it does this:

var subscriber = _source.Synchronize(locker).Do(changes => RegisterForRemoval(items, changes), observer.OnError).SubscribeSafe(observer);

If this is changed to:

var subscriber = _source.Synchronize(locker).Do(changes => RegisterForRemoval(items, changes), ).Subscribe(observer);

then the exception bubbles out.

However, from a quick search of the code, it appears that SubscribeSafe is used in many places so other methods will also exhibit this behaviour (for example, I found that RefCount also suppresses errors).

I imagine it is likely then that you will say that this behaviour will not change.

If that's the case, is there something I can do on the outside to make the exceptions on the OnError channel throw?
I've tried various things, but they all seem to just get suppressed.
For example, in the test case I provided, if you subscribe like:

transformed.Subscribe(x => { }, ex => throw new Exception(), () => { });

the exception is still swallowed.

[RolandPheasant]

Exceptions should propagate. Let's not rush into changing everything, but it probably should be done.

[JakenVeina]

Exceptions aren't not-propagating, they're just being emitted via OnError() within the stream, instead of throwing out of .Subscribe(). I'm convinced this is the correct approach.

In my mind, .Subscribe() should not throw this exception, since it's the result of emitting values into the stream, not a result of some failure to construct observable objects that make up the stream. I.E. it's not the responsibility of .Subscribe() to notify the caller of data issues, it's the responsibility of stream operators to handle data issues.

Alternatively, consider this realistic scenario:

var subscription = source
    .Transform(item => new MyDisposable(name: item.Name.ToUpper()))
    .DisposeMany();

Let's say that we have a bug where item.Name might be null, so this transform could throw NullReferenceException. Today, we can be assured that such an exception will always be published via OnError(). If we were to make the requested change here, that would no longer be guaranteed. Instead, the exception might manifest two different ways, depending on the state of the data in the source at the time of .Subscribe(), making it more cumbersome for consumers to deal with.

Additionally, if the exception throws out of .Subscribe() it becomes essentially impossible to recover from. The stream that the caller was trying to construct can never be constructed, without clearing out the bogus data from the source. If the exceptions are published via OnError() there's mechanisms to handle those errors without killing the entire stream.

@seanreid1201

If that's the case, is there something I can do on the outside to make the exceptions on the OnError channel throw?

Where would you want them to throw? With the proposed change, you would get some exceptions to throw upon .Subscribe() but if you care about these exceptions, then you HAVE to also care about the exceptions published via OnError(), because the same exception could throw both ways. I.E. If it's possible for your stream to emit errors, you have to add operators to the stream to handle them. If you have those operators, then what's the point of duplicating that logic to catch the same errors being thrown out of .Subscribe()? If you don't have those operators, then your stream is already swallowing exceptions, why does it matter that .Subscribe() does too?

[JakenVeina]

@RolandPheasant this should now be resolved.

[github-actions]

This issue has been automatically locked since there has not been any recent activity after it was closed. Please open a new issue for related bugs.