Inconsistent handling of undeliverable errors in RX builders

[1zaman]

While looking at the implementations of the RX coroutines builders in the kotlinx-coroutines-rx* modules, I noticed some inconsistencies in the handling of undeliverable errors between the rxObservable implementation on the one hand, and the rxSingle, rxMaybe, and rxCompletable implementations on the other.

In the rxSingle, rxMaybe, and rxCompletable implementations, when the coroutine is cancelled with an exception, it first tries to deliver it to the observer, and if that's not possible then it is reported to the RX global exception handler. This behaviour is the same regardless of whether the error is fatal or not.

Whereas in the rxObservable implementation, it drops undeliverable non-fatal errors, but always reports fatal errors to the RX global exception handler after attempting to deliver it to the observer, regardless of whether the observer could handle it or not.

The rxFlowable/publish implementation seems to be a combination of both, as it reports all undeliverable or fatal exceptions to the global exception handler. This seems to be the best and safest approach.

[1zaman]

So the different handling of fatal exceptions in rxObservable seems to have been due to a potential issue with rxFlowable that was reported in #1297, where the subscriber throws an exception from the onNext callback, which leaves the operators that were processing the emission queue in an inconsistent state, and thus it does not handle the exception when the coroutine tries to emit it after being cancelled due to it. The standard subscriber that's used when subscribing with lambda callbacks catches any non-fatal exceptions that are emitted by the onNext callback and forwards them to onError, but does not catch fatal exceptions, thus causing this issue. The standard flowable builders don't catch any exceptions that are emitted by the subscriber, and just let fatal exceptions propagate and crash the app, so they don't encounter this problem.

This problem seems to be specific to the rxFlowable/publish implementation due to it's usage of a locked emission queue, and not present in rxObservable, but the fix was applied consistently to both. However, it does not apply in the rxSingle, rxMaybe, and rxCompletable implementations, since they don't have any non-terminal emissions that can throw an exception while the coroutine is still running, and any exception from a terminal emission can be caught and handled right there by passing it to the global exception handler.

This explains the special handing of fatal exceptions in rxObservable and rxFlowable. However, it still results in inconsistency with the other RX builders in the case where the coroutine itself throws a fatal exception. We can consider changing the other implementations to have the same behaviour as rxObservable for consistency.

This would also be in line with the standard RX builders, which also let fatal exceptions thrown by the emitter function propagate to the uncaught exception handler. However, it seems that coroutines don't have the concept of fatal exceptions that should not be caught, so all exceptions are caught and emitted by them, and therefore the current behaviour of rxObservable in handling them when the coroutine completes by reporting them to the RX global exception handler makes sense I suppose

[dkhalanskyjb]

Thank you for raising this!

Yes, your analysis is spot-on, the root cause for the original problem was that onNext rethrows fatal exceptions but not the other ones and that we consistently send the errors flying around to onError, which landed in the emission queue.

Looking at the way RxJava treats fatal exceptions, it doesn't seem though like we could behave consistently with them in spirit: they attempt to let fatal exceptions bubble up to the surface as fast as possible, except in scenarios when the fatal exception is being passed to onError, in which case RxJava treats these exceptions like any others. Especially with the recent introduction of trySend, a non-throwing attempt at sending an element through a channel, we won't be able to arbitrarily throw fatal exceptions without violating the existing contracts.

However, it seems like there is a simple solution imposed by the Reactive Streams specification itself: https://github.com/reactive-streams/reactive-streams-jvm/blob/v1.0.3/README.md#2.13 It says that in case any methods of Subscriber throw, the publisher must report the errors not to onError but somewhere else. Following this, we can neatly solve the original problem: the error doesn't become lost in the emission queue, as it gets passed to the handler for undeliverable exceptions.