Should this spec route errors occurring in the next handler to the error handler?

[benlesh]

I was talking with @trxcllnt today about the behavior the occurs when you throw in an next handler...

For example:

myObservable.subscribe((x) => {
  // ReferenceError:  let's say x.foo is undefined here.
  someSideEffect(x.foo.blah);
}, err => {
   reportAnError(err);
});

Currently, it will just throw, and will not go to the error handler.

This results in people using RxJS like this as a "best practice", which isn't at all a "best practice", because do and doAction will route thrown errors down the error path:

myObservable.doAction((x) => {
  someSideEffect(x.foo.blah);
})
.subscribe(noop, (err) => {
  reportAnError(err);
});

Apparently teams at Netflix even have their own forks of RxJS that have patched the the onNext handler call to route to the onError handler if it throws.

[zenparsing]

Good question.

On one hand, this clearly illustrates the desire to handle errors in one place (the subscription error handler).

On the other hand, error(ex) is supposed to be called when the stream terminates with an error, not because an error happened generally. In particular, if the observer is left in an invalid state when it threw, then it might be a bad idea to subsequently invoke any method on it.

The "doAction" best practice seems better suited to an alternate promise-returning subscription method:

myObservable.forEach(x => {
    blahblahblah();
}).then(null, err => {
    reportAnError(err);
});

Although such a function is currently blocked on async task cancellation stuff.

[benlesh]

Well a sticky wicket about this is if you gracefully handle the error in the next handler, you're forced to unsubscribe yourself:

let errorHandler = (err) => {
    /* do things here */
};

let subscription = myObservable.subscribe(x => {
  try {
    someSideEffect(x.foo.bar);
  } catch (err) {
    errorHandler(err);
    subscription.unsubscribe(); // better unsub!
  }
}, 
errorHandler);

[benlesh]

... that is I presume the spec states that if an unhandled exception happens in the next handler, it will cause unsubscription currently. So if you do handle the error, it's not going to unsubscribe, so you'll have to do that manually.

[zenparsing]

@Blesh Well, you would probably just call SubscriptionObserver.prototype.error, which would do the unsubscription automatically, so that's not a problem.

But the larger problem, to me, is that you don't want to invoke methods on an observer which has failed and may be in an invalid state.

[trxcllnt]

@Blesh an important detail in your example is that if the Observable onNext's synchronously, the subscription value will still be undefined inside the catch.

@zenparsing ...error(ex) is supposed to be called when the stream terminates with an error, not because an error happened generally

I acknowledge and agree 100%... on paper. But having worked the last five years with teams and projects using Rx, I'm inclined to question the pragmatism of this behavior.

...the desire to handle errors in one place...

This is certainly helpful, but I see more here. By encouraging users to move throw-prone side-effects inside the monad, we're encouraging them to mix pure and impure computation. Suddenly, operators that expect to handle legitimate exceptions from the pure computations before it have to understand and account for errors that would normally only have occurred in the subscription.

if the observer is left in an invalid state when it threw, then it might be a bad idea to subsequently invoke any method on it

Agreed. If next throws an error, it'd be incorrect to message the Observer again with next or complete. I'd argue we should invoke the error handler though, because next shouldn't (?) invoke its own error method.

With the behavior Ben's described, next has two options for error handling and stream disposal; it can catch side-effect errors inside next and recover, or it can throw the error back to the Observer, which invokes error and unsubscribes.

source.subscribe(
  (x) => {
    try { sideEffectsAhoy(x); }
    catch(e) {
      if(!(e instanceof RecoverableError)) {
        throw e; // re-throw the error to invoke the `error` handler and unsubscribe from source
      }
      console.log("Recovered from " + e.toString());
    }
  },
  (e) => { console.error("Encountered a nasty error :("); }
);

[trxcllnt]

@Blesh @zenparsing and of course, the default error handler simply re-throws if is isn't overridden, so the error would still propagate out in the worst case (as is the case today).

[benlesh]

an important detail in your example is that if the Observable onNext's synchronously, the subscription value will still be undefined inside the catch.

Haha... Yeah, of course. LOL missed that one because I was thinking about RxNext. This is another place where a Subscriber is better than an Observer, because you're going to have access to unsubscribe of a Subscriber.

[zenparsing]

See #50. Again, I don't think you should call any method on an observer which has failed. It really is "dirty" and can't be trusted at that point. Instead, users should use "do" which returns a promise, and deal with errors by way of that promise.

[trxcllnt]

@zenparsing what do you mean by "trusted?" Observers are stateless, and thus don't have a concept of "failure." I'm curious as to what would constitute a "failed" Observer.

[zenparsing]

@trxcllnt I don't think we can assume that arbitrary objects provided to the API are stateless. In fact, given JS's OO bent, I think we have to assume the opposite.

[benlesh]

@trxcllnt.. I think he means an observer which has already processed an error.

[trxcllnt]

@Blesh @zenparsing The next callback throwing an error and the Observer receiving an error message are two separate behaviors. As described in the Rx Design Guidelines (PDF):

Observables should send zero or more onNext messages, followed by either an onCompleted or onError (but not both), and no more messages should arrive after an onError or onCompleted message.

After an Observer's error callback is invoked, the Observable must not message the Observer again. If the error callback hasn't been invoked yet, as would be the case if next throws an error, it's still valid to invoke the error callback.

Do you have a real-world example in which invoking error after next, even if next throws, isn't safe on a spec-conforming Observer?

[benlesh]

So it seems like the answer is "No, errors in the nextHandler should just throw, and not be sent to the error handler"?

confirm? @jhusain? @trxcllnt? @zenparsing?

[zenparsing]

@Blesh That would be my answer.

[benlesh]

Sounds okay to me honestly. Closing for now.

[trxcllnt]

@Blesh this is in direct conflict with our teams' internal needs, so I disagree 100%.

[benlesh]

Oh? I guess I had forgotten you had hit this issue. Reopening.

[jhusain]

Why should we change the underlying semantics of the type? The internal teams requirements can be accommodated with a method:

We could just add a method listen. Observable.listen() would translates to observable.doAction(onNext, onError, onComplete).subscribe(...)? Changing the types semantics seems high risk. Curious why the decision was made in the first place. Any feedback @headinthebox @mattpodwysocki ?

JH

On Aug 8, 2015, at 8:59 AM, Ben Lesh notifications@github.com wrote:

Reopened #47.

—
Reply to this email directly or view it on GitHub.

[headinthebox]

I am a bit lost what exactly the issue is. Try the following in RxJava and it prints wham, replace the test with x == 2 and it prints boom. Same in Rx.NET.

class MainJava {
  public static void main(String[] args) throws InterruptedException {
    Observable.from(1,2,3,4).subscribe(
            x -> {
                if(x == 10) {
                    throw new RuntimeException("boom");
                }
            },
            e -> System.out.println(">>> "+e.getMessage()),
            () -> {
                throw new RuntimeException("wham");
            });
  }
}

[trxcllnt]

@headinthebox if that is the case in RxJava/Rx.NET, then that's the behavior I'm arguing for here. That is not the present behavior in RxJS, as illustrated in this jsbin example.

[zenparsing]

Looks like that is the case for the SafeSubscriber in RxJava:

https://github.com/ReactiveX/RxJava/blob/1.x/src/main/java/rx/observers/SafeSubscriber.java#L137

But not the case for AnonymousSafeObserver of Rx.NET:

https://github.com/Reactive-Extensions/Rx.NET/blob/master/Rx.NET/Source/System.Reactive.Core/Reactive/AnonymousSafeObserver.cs#L44

Am I reading the code correctly?

[zenparsing]

Still opposed to this - RxJava's semantics here seem questionable to me (although I understand the practicalities).

[jhusain]

Reopening this issue, because I ran across the following use case. The following code prints "error" to the console:

function test() {
  try {
    [1,2,3].forEach(x => {
      throw "error";
    })
  }
  catch(e) {
    console.log(e);  
  }
}

The way the current Observable spec is written, the following code prints nothing to the console:

async function test() {
  try {
    await Observable.from([1,2,3]).forEach(x => {
      throw "error";
    });
  }
  catch(e) {
    console.log(e);  
  }
}

This is definitely wrong because it is a refactoring hazard. Furthermore the Promise returned by forEach becomes a Promise.never, because after next throws no further notifications are delivered to the Observer. This seems like a big footgun.

If we want to mimic the behavior of sync forEach, we have two choices:

1. deliver errors thrown from next to error
2. Special case forEach by putting a try/catch block around the nextFn. If nextFn throws, reject the Promise

I'm pretty uncomfortable with option #2. Seems to me that forEach and subscribe should be semantically equivalent in every way. For JavaScript, option #1 seems correct to me.

[zenparsing]

@jhusain We removed forEach from this repo until we had the desired semantics better established, but I've implemented it in zen-observable using option 2.

I'm not sure I understand the objection to option 2, though.

If we went with this change, then wouldn't it force our hand with respect to #109? DOM wants to be able to keep sending next even if the observer throws, and it wouldn't be able to do so if we routed errors to error (since there can only be one error per stream).

[jhusain]

You make a valid point with respect to #109. I've previously advocated for the on method swallowing errors from next to avoid closing the Subscription, but @domenic raised a valid point with respect that this feels like a more fundamental mismatch. If we feel as though its the right behavior to close subscriptions when throwing, it will be necessary to rationalize what's so special about EventTargets that they don't need this behavior. Furthermore should it really apply to all event targets? Will add more comments in that thread.

[mattpodwysocki]

@jhusain I'm concerned here after our conversation yesterday with regards to error semantics closing versus not closing based upon error in the stream. If it's unhandled meaning a throw happens in a method, then is it routed to the error and then not torn down, or is it torn down? And does that differ from regular error semantics?

[trxcllnt]

@zenparsing I'm not clear why #109 is an issue, since Observables are referentially transparent. Subscriptions can synchronously .retry() if they error. If the consensus is that's too much work to push onto the end consumer, then DOM Observables (or DOM Subscribers) can be special subtypes that automatically retry on error.

The idea that the subscription is cleaned up on error/complete is central to Rx's guaranteed deterministic memory management.

[trxcllnt]

To be clear, the behavior I'm advocating for here is:

1. Forward errors thrown from next to error.
2. Let errors thrown from error or complete propagate globally.
3. Subjects should catch errors thrown from Observers' next and forward to current + remaining Observers' error.
   3a. Subject should catch errors thrown from Observers' error, and complete, and forward on to remaining Observers' error. Catch there, and use latest caught error to continue notifying. If last Observer's error throws, re-throw globally.

p.s. we will likely have a way to configure this behavior in RxJS to allow errors to propagate globally instead of being caught, so Alex Liu and the node team can extract accurate core dumps (doing anything else risks corrupting VM state that's essential to postmortem investigative analysis).

[jhusain]

@zenparsing I see the wisdom of the approach you've taken. I'm content to remove autoclose behavior when Observers throw, but add it in for forEach. I will add my rationale to #109.