Async Create hanging while publishing observable

[yuramag]

Consider the following code:

var xs = Observable.Create<Unit>(async o => 
{
    await Task.Delay(10);
    o.OnError(new Exception());
}).Replay().RefCount();

xs.Subscribe(x => Console.WriteLine(x));
xs.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message));
await xs.DefaultIfEmpty();

The sequence above doesn't throw any exceptions and never completes.

I made the following observations:

1. Removing the first subscription enables error propagation - exception is thrown in Subscribe context (last line)
2. Removing .Replay().RefCount() enables error propagation - exception is thrown in Subscribe context (last line)
3. Removing await Task.Delay(10) enables error propagation - exception is thrown in OnError call (within Create method). Surprisingly, switching two Subscribe methods makes exception thrown at Subscribe context (last line).

That being said, I am asking whether the following issues are by design:

1. Observable sequence in the above scenario never being completed
2. The fact that exception is sometimes thrown inside Create method, and other times - at Subscribe context.

If this is by design, what would you recommend as a workaround? How do I publish my sequences so that all of my clients (observers) can safely handle exceptions in this case? Current behavior seems so arbitrary, especially for library writers. It also makes debugging very painful. Please advise.

[akarnokd]

Could you try with 4.1 preview?

Also omitting the onError handler in the first subscribe may be the problem. Afaik such subscriptions without an OnError handler rethrow the received exception which breaks the operator logic, preventing further event delivery altogether.

[yuramag]

Tried it on 4.1 preview - still no luck.

I think you are right saying that omitting onError breaks operator logic due to exception being rethrown. The problem is that it's being swallowed sometimes.

Changing o.OnError(new Exception()); to Task.Run(() => o.OnError(new Exception())).ContinueWith(t => Console.WriteLine("ERROR"), TaskContinuationOptions.OnlyOnFaulted); enables exception propagation.

Either way, wouldn't it be more straightforward if exceptions were never thrown in operator context?

[antichaosdb]

Hopefully this is the same kind of issue. We just updated from 4.0 to 4.1 and a bunch of our Rx broke. It seems that the behaviour when you do Observable.Create(async obs => has changed. As an example, what is the final state of the bools in this code?

bool started = false;
bool message = false;
bool completed = false;
bool disposed = false;
var obs = Observable.Create<bool>(async observer =>
{
    started = true;
    await Task.Delay(1);
    observer.OnNext(true);
    return Disposable.Create(() =>
    {
        disposed = true;
    });
})
var sub = obs.Subscribe(x => message = x, () => completed = true);

Answer is it depends on if you're debugging or not, and how long you wait, but the change is that in 4.1 this observable completes and disposes itself which it didn't before.

I appreciate that this may be bad code for some good reasons, and I've found more sensible ways to do what we're trying to do, but having hunted though the changelogs and searched the docs, I wanted to check whether this was always bad but just happened to work, or whether this is a deliberate change in order to support something else.

[danielcweber]

When do you expect the bools in the code to be in a consistent final state? Once you subscribe, the notifications that will eventually set your bools will fire asynchronously (possibly not even that, as you may have observed when debugging). You can't assume the bools to have any determined state after Subscribe completes.

The observable completing is by design, it has been subject of some debate but the behaviour hasn't changed from 4.0 to 4.1. What's important is that OnNext happens before OnCompleted. You might add some code to assert that

If you continue to experience strange behaviour, please provide a failing unit test asserting your expectation.

[antichaosdb]

Thanks very much for the clarification :)

[danielcweber]

@yuramag Could you please provide a working example with some assertions (a unit test in the best case) to clarify your expectation?

[danielcweber]

@yuramag @antichaosdb Can you confirm this is fixed with v4.1.2 ?

[yuramag]

The issue is still present. Unfortunately, behavior is not predictable. Consider the following test:

[TestMethod]
public async Task TestRx1()
{
    const int timeout = 5;

    var xs = Observable.Create<Unit>(async o =>
    {
        await Task.Delay(timeout);
        o.OnError(new Exception());
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message), () => Console.WriteLine("Completed"));
    
    var completion = Task.Run(async () => await xs.DefaultIfEmpty());
    completion.ContinueWith(x => Console.WriteLine($"Eception: {x.Exception.Message}"));

    await Task.Delay(timeout * 2);

    Assert.IsTrue(completion.IsCompleted);
}

I expect the assertion to always pass, which happens half of the time. Sometimes though I am getting exceptions printed out and assertion failing.

[danielcweber]

Your assertion is definitely incorrect. It's a task, it's run asynchronously, even with a delay, you just can't make that assumption.

[yuramag]

Well, I agree that it is not entirely correct. I was just trying to test around hanging on await of xs. Consider these two tests:

[TestMethod]
public async Task TestRxPasses()
{
    var xs = Observable.Create<Unit>(async o =>
    {
        await Task.Delay(5);
        o.OnError(new Exception());
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message), () => Console.WriteLine("Completed"));

    await xs.DefaultIfEmpty();
}

[TestMethod]
public async Task TestRxHangs()
{
    var xs = Observable.Create<Unit>(async o =>
    {
        await Task.Delay(5);
        o.OnError(new Exception());
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x));
    xs.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message), () => Console.WriteLine("Completed"));

    await xs.DefaultIfEmpty();
}

The last one never ends!

[danielcweber]

That's due to the first call to Subscribe, which doesn't observe the exception but only rethrows it. That'll break the whole chain, every operator will unsubscribe from it's source and you won't get anything out of xs again. It's unrelated to that specific overload of Create. This code doesn't terminate as well:

[TestMethod]
public async Task TestRxHangs()
{
    var xs = Observable.Create<Unit>(o =>
    {
        Task.Delay(5).ContinueWith(_ => o.OnError(new Exception()));
        return Disposable.Empty;
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x));
    xs.Subscribe(x => Console.WriteLine(x), ex => Console.WriteLine(ex.Message), () => Console.WriteLine("Completed"));

    await xs.DefaultIfEmpty();
}

[yuramag]

What bothers me is that these two semantically equivalent tests behave differently - one passes, but the other does not:

[TestMethod]
[ExpectedException(typeof(Exception))]
public async Task TestRxPasses2()
{
    var xs = Observable.Create<Unit>(o =>
    {
        o.OnError(new Exception());
        return Disposable.Empty;
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x));

    await xs.DefaultIfEmpty();
}

[TestMethod]
[ExpectedException(typeof(Exception))]
public async Task TestRxHangs2()
{
    var xs = Observable.Create<Unit>(async o =>
    {
        Task.Delay(1).ContinueWith(_ => o.OnError(new Exception()));
        return Disposable.Empty;
    }).Replay().RefCount();

    xs.Subscribe(x => Console.WriteLine(x));

    await xs.DefaultIfEmpty();
}

Again, if you believe that this is expected by design, I won't argue.

[danielcweber]

The first one is synchronous, you get the exception directly on subscription. The second is asynchronous, so the test won't observe the exception, but the subscription chain still will be broken (standard Rx operators unsubscribe and clean up when a downstream-call to OnError throws). As a rule of thumb, don't use the Subscribe-extension that doesn't handle exceptions because you might just run into situations just like this, which is especially harmful if the observable is shared through Replay and RefCount.

[bartdesmet]

The analysis by @danielcweber on this thread is correct. I'm not sure if there's still anything actionable with this issue given that the behavior is as intended?