Seemingly inconsistent error handling semantics: do vs operators vs subscription.next

[nelsonlaquet]

This all started out with a StackOverflow question that I had asked:
http://stackoverflow.com/questions/35740821/proper-way-to-deal-with-errors-thrown-in-onnext-for-hot-shared-observables

To sum it up: I need subscriptions that throw in their "next" handler to simply unsubscribe and NOT re-throw the exception back up the callstack.

I've boiled down my problem to an inconsistency between error handling semantics. Basically:

• Most (all?) Operators wrap callbacks in a try/catch block and submit the error object FORWARD without re-throwing.
• The Do operator sets the oddly named "syncErrorThrowable" to true on the subscription which essentially does the same thing.
• Normal subscriptions with syncErrorThrowable set to false (default) re-throw the exception. This causes the exception to bubble up the callstack, and will cause the underlying observable to terminate. EVEN if it's a hot observable with multiple other, unrelated, subscriptions. It essentially completely nukes the stream.

I find the latter case troubling, because an error in one part of an application can cause an entire observable sequence to stop producing values - affecting any unrelated part of the application. I know that we consider unhandled errors as terminal, but this issue really affects me during development when using hot module replacement.

Basically, Module A has a programming error in the "next" handler, and throws. This causes the underlying store to stop producing values. So then I fix the error, hot reload the module; but the when the fixed module subscribes to the underlying observable, it receives no values since the sequence was terminated due to the buggy instance.

Of course, this only applies to hot observables. Cold observables are not affected.

I have two possible solutions:

1. Add "subscription.syncErrorThrowable = true" in my subscription wrapper. I wrote a wrapper for the "subscribe" method that accepts a module as the first parameter, so it automatically unsubscribes when that module is detached. I could alter this to set that flag to true.
2. Create an operator that wraps its call to "destination.next" with a try/catch, and on fail, propagates the error FORWARD instead of bubbling the exception up.

I'm wondering which solution is preferable. "syncErrorThrowable" has no documentation and seems to only be used in the "Do" operator, and I feel dirty messing with that. But at the end of the day, why would "do" have drastically different error handling semantics than the "next" handler for subscriptions?

Another question would be, how do the proposed esvnext Observables handle this case?

[trxcllnt]

@nelsonlaquet here's some context on syncErrorThrowable. Subscribers have a flag that Observable's subscribe checks to decide whether it should re-throw errors caught during synchronous notification. The do operator uses a SafeSubscriber internally to accommodate the various argument combinations (callbacks/observer etc.), but it'll only catch errors in do handlers if syncErrorThrowable is set to true on its private SafeSubscriber.

If your next handler throws and the source is synchronous, you can catch that error by wrapping it in a try-catch:

try {
  Observable.of(1, 2, 3).subscribe((i) => {
    if (i === 2) { throw i; };
  });
} catch (e) { console.log(`caught ${e}`); }

If your next handler throws and the source is asynchronous, you're SOL. I've fought for a long time with people about this, so I feel your pain. Here's the only alternative we've got for now:

Observable.of(1, 2, 3, Scheduler.async)
  // If your `next` handler can throw, move it into a `do` block just above the subscription
  .do((i) => { if (i === 2) { throw i; } })
  .subscribe(null, (e) => {
    console.log(`caught ${e}`);
  });

[trxcllnt]

@nelsonlaquet somewhat related: try to minimize multicasting as much as possible, as this is what they mean when they say "subscription side-effects" (one subscription blowing up can hose all the rest).

[nelsonlaquet]

The issue in this instance is that I have a real-time chatroom like application written against socket.io. I figured modelling this with RxJS would be appropriate, and so far it's been nice to work with.

So basically, I have stores such as ChatStore and UserStore, which are classes that follow this pattern:

const events$ = Rx.Observable.merge(
    server.on$("users:list").map(opList),
    server.on$("users:added").map(opAdd),
    server.on$("users:removed").map(opRemove));

this.state$ = events$
    .scan((state, op) => op(state), {state: defaultState})
    .publishReplay(1);

this.state$.connect();

The "op" functions look like:

function opList(users) {
    return state => {
        return { type: "list", state: { users } };
    };
}

"server" is a wrapper over a socket.io socket, and "on$" is simply:

on$(eventName) {
    return Rx.Observable.fromEvent(socket, eventName);
}

Components are free to subscribe to whichever stores they depend on. For example, in my users component that is responsible for the header of the users panel, I may do this:

usersStore.state$.subscribe(({state}) => {
    $header.text(`${state.users.length} users online`);
});

And elsewhere in my chat messages component, maybe this:

Rx.Observable.merge(
    chatStore.newMessage$.map(createMessage),
    usersStore.state$.filter(a => a.type == "add" || a.type == "remove").map(createUserAction))
    .subscribe($message => {
        $messages.append($message);
    });

Which would create rows for things like "user x entered", "chat message 1", "user x left" and so on. For other reasons, I'm not using either React or Angular 2, but mostly following a standard architecture of Store -> Component -> Action.

This works very well, but doesn't this necessarily mean that these observables must be multicasted? I don't want to re-subscribe to the socket for all store-related events every time I need to access that data. I'd like it to be easy and as cheap as possible to have multiple aspects of the user interface driven by these events.

Or is this a situation that I should not be multicasting?

As for your other post... That is disappointing. Moving things into "Do" seems rather silly to me, especially since, as you pointed out, I'd be nooping in my subscription "next" handler. That feels very off to me.

Would it be against the spirit of the library to create a custom operator, such as... I dunno... "protectExceptions" that wraps the call to "destination.next" with a try/catch, and propagates errors into the destination's error signal? Or should I just abuse the syncErrorThrowable property (it is public, after all)?

I really really really do not want to wrap all of my subscription handlers in my own try/catch.

[trxcllnt]

Would it be against the spirit of the library to create a custom operator, such as... I dunno... "protectExceptions" that wraps the call to "destination.next" with a try/catch, and propagates errors into the destination's error signal? Or should I just abuse the syncErrorThrowable property (it is public, after all)?

Yes, I think so. Something as trivial as safeSubscribe(next, error, complete) that returns this.do(next).subscribe(null, error, complete) should do the trick.

I really really really do not want to wrap all of my subscription handlers in my own try/catch.

While I agree with you, it's pretty much us against the world.

[benlesh]

While I agree with you, it's pretty much us against the world.

It really is :P haha

@nelsonlaquet as an explanation, it was determined that having errors thrown in the next handler propagate to the error handler fundamentally changed the semantics of observation via subscribe, such that: 1. Observers now observe themselves, at least for errors, and 2. There would be no guarantee that the errors handled by the error handler came from the producer and not the consumer.

The guarantee is that all errors passed to your error handler in subscribe are errors that originated with the producer.

If you're concerned about errors being thrown in your next handler, you can deal with that using a simple try/catch. Since the subscribe call is the terminus of your observable chain anyway, there isn't much you could do with it in terms of observable composition. That's really what do is for.

[nelsonlaquet]

Since the subscribe call is the terminus of your observable chain anyway

Right, I understand this; however it's actually more than that. The exception terminates that chain (which I'm totally cool with), but it will also nuke the underlying observable, no matter how many operators separate the observable and the subscription, nor how many other subscribers there are.

Say I have a multicast observable that produces events from a socket.io event handler. If I have two subscriptions against it, and one subscription throws an error, the entire observable terminates, causing the second subscription to also terminate.

So I'm not necessarily advocating that an exception in "next" propagates to the error handler; but what if multicast observables catch the error, then re-throw them outside of the current VM tick (think: setTimeout of 0), so that other subscriptions aren't affected, yet the exception is still shown to the developer with the proper call stack?

I think from a design perspective one could make the argument that a subscription only cares about itself; and should not have to consider whether or not it's acting against an observable with 10 million subscribers, or a cold observable fresh off of "range".

The way things currently are, all of my multicast subscriptions have to be "aware" of other potential subscriptions in the form of defensively catching exceptions that shouldn't be caught (since they're programming errors and non-recoverable). This seems contrary to the design of the rest of the library. I think my proposal would add to the stability of unrelated components in the event of a failure, as well as make development with hot-reloading JavaScript so so much easier.

[nelsonlaquet]

@trxcllnt thanks for that suggestion! At the moment that is what I'm doing. I was using a custom subscribe method anyway that auto unsubscribes when the related component detaches, so it fit in well with the rest of my code.

[ccrowhurstram]

Another alternative to creating a custom safeSubscribe would be subscribe using the forEach operator.

EG:

obs$.forEach(val => {
  // do stuff...
  // ... eventually
  throw new Error("Boom");
})
  .catch(ex => {
    // receives Error("Boom")
    // would also receive exceptions thrown by obs$
});

[benlesh]

closing as inactive.

[nelsonlaquet]

Wow, it's been a year, already?

I've been using the solution above since this post (write a custom subscribe method against the Observable prototype which is wrapped in "do") but I'd be interested to know if any further discussion about this topic has been made.

Terminating the underlying hot observable due to a programming error in an unrelated subscribes's "next" handler still feels wrong to me.

Is it simply due to the performance implication of try/catching the "next" callback? Re-reading this thread, I don't find any fundamental design considerations that would prevent this from being proper behavior.

[ccrowhurstram]

It looks like this issue (or very similar) is being actively discussed here: #2145

[lock]

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