Guideline 6.4: Protect calls to user code from within an operator

[benjchristensen]

While discussing pull request #212 a possible issue came up with out incoming (not operators) Observer's are handled in relation to wrapping them with AtomicObserver.

Instead of discussing it on that unrelated pull request I am moving it to here. The original comment on the possible problem is from @johngmyers here:

Guideline 6.4 explicitly states:

Note: do not protect calls to Subscribe, Dispose, OnNext, OnError and OnCompleted methods. These calls are on the edge of the monad. Calling the OnError method from these places will lead to unexpected behavior.

This seems to expressly contraindicate the AtomicObserver.onNext() protection code.

[benjchristensen]

@johngmyers This is an interesting observation ... I hadn't noticed that note at the end of guideline 6.4 before. We added this wrapping code close to 1 year ago as a result of dealing with errors being thrown in random places because of bad Observer implementations subscribing to sequences.

I don't yet fully understand the principle this note is trying to say nor the issue it claims it will cause.

The reason why the AtomicObserver exists is because "non-trusted" Observer implementations (via an actual Observer or one created via the use of Action or Func classes or the various language lambas/closures) often are a risk for throwing exceptions.

For example:

public void onNext(String v) {
    int num = Integer.parseInt(v);
    doSomething(num);
}

That code is very easy to have result in a NumberFormatException being thrown.

We had NullPointerExceptions also being thrown, and what was happening is that they throw an exception off on some random thread because it's all asynchronous and things seem to just fail silently.

Operator implementations as part of the RxJava library all should comply with the Rx Guidelines but users of the library don't do that very well and thus attempts have been made to be resilient to incorrect behavior.

The two most obvious examples are:

1) exceptions being thrown from code passed in by users

Guideline 6.4 covers functions, comparators, and other such obvious things (Func* and Action* implementations predominantly), but the Observer is by far the biggest one we had issues with.

2) Observables calling onNext concurrently or from multiple threads even when sequential (not concurrent)

This is why we still implement the operators as if they are going to be called concurrently and use volatile, atomic and java.util.concurrent.* data structures.

We understand Guideline 6.8 and thus do not synchronize anything (except where the operator requires it such as merge/zip) but we still assume we need to handle visibility across threads.

I'm going to ignore the second one for this discussion as that's not related to Guideline 6.4 which this issue is about.

For errors occurring in onNext/onError/onCompleted when they are provided by the user I see 3 options:

1) Throw
We can just let it throw exceptions off on whatever thread occurs

This is bad and not the solution we can accept. The only time I know of when that's the only acceptable option is documented here: #198

2) Try/Catch Everywhere

Every Rx operator implementation needs to try/catch on every call to Observer onNext/onError/onCompleted methods

3) Single Wrapper

Have a single wrapping class such as the current AtomicObserver that is wrapped around Observers passed in from the user code.

Option 2 means a lot of verbose error handling must be reproduced in every operator and it's error prone to make sure every single one does it in all the right places.

What we found in experience was we would have some operators doing okay but then one would not and result in bad error handling for the user.

Option 3 evolved out of wanting a single place to ensure proper error handling occurred regardless of whatever every single Rx operator was implemented without bugs in having a try/catch around every single call to the Observer.

---

All this said I am not beholden to any particular implementation approach. My goals are:

• comply with Rx contract unless the JVM prevents it or it doesn't match Java idioms and in that case match it as closely as possible
  • naming conventions (lower-case vs capitalized)
  • some names used by Rx.Net are reserved (do/finally)
  • lack of extension methods requires a different approach for Observable (that's why it's a class and not an interface)
  • lack of yield means synchronous Observables aren't easily made lazy (and I'd love for someone to implement a lazy Iterable to deal with this)
• focus on getting the public API right and let the underlying implementations evolve as needed to improve performance, resilience and compliance
• target the JVM and not any specific language so that any and all should be capable of being supported (already significant improvements are coming from community feedback and involvement (More formal support for Scala #213 and Explore code generation for language adaptors #204)

---

On to an example ...

The recently updated (#212) take operator that will blow up if a user passes in a bad Observer ( bug #217 ):

            @Override
            public void onError(Exception e) {
                if (counter.getAndSet(num) < num) {
                    observer.onError(e);
                }
            }

            @Override
            public void onNext(T args) {
                System.out.println("Take: " + args);
                final int count = counter.incrementAndGet();
                if (count <= num) {
                    System.out.println("Take sending to onNext: " + args);
                    observer.onNext(args);
                    System.out.println("Take AFTER sending to onNext: " + args);
                    if (count == num) {
                        observer.onCompleted();
                    }
                }
                if (count >= num) {
                    // this will work if the sequence is asynchronous, it will have no effect on a synchronous observable
                    subscription.unsubscribe();
                }
            }

Here is the unit test that breaks it:

        @Test
        public void testTakeWithError() {
            final AtomicInteger count = new AtomicInteger();
            final AtomicReference<Exception> error = new AtomicReference<Exception>();
            Observable.from("1", "2", "three", "4").take(3).subscribe(new Observer<String>() {

                @Override
                public void onCompleted() {
                    System.out.println("completed");
                }

                @Override
                public void onError(Exception e) {
                    error.set(e);
                    System.out.println("error");
                    e.printStackTrace();
                }

                @Override
                public void onNext(String v) {
                    int num = Integer.parseInt(v);
                    System.out.println(num);
                    // doSomething(num);
                    count.incrementAndGet();
                }

            });
            assertEquals(2, count.get());
            assertNotNull(error.get());
            if (!(error.get() instanceof NumberFormatException)) {
                fail("It should be a NumberFormatException");
            }
        }

That will print out 1 and 2 and then silently fail.

Note how the onNext in the take implementation does not have a try/catch around observer.onNext and then the onError has logic that prevents observer.onError from getting calling after the count increments - which happened in the onNext that failed - thus it skips sending the error to the observer.

Funny enough this test case even shows that the way AtomicObserver is currently being used doesn't handle this scenario! This is because the "isTrusted" decision point is for the Observable not the incoming Observer.

If I force the AtomicObserver to be used for this test case then the onError is correctly called even with the bugs in the take operator.

This particular take operator is heavily unit tested and at least 2 sets of eyes reviewed it and both missed it.

I've had this AtomicObserver in place for over a year and didn't realize this particular type of case was still a gaping hole - probably explains why I still hear in production about "silent failures" occasionally happening - not even a log entry showing up.

As for how having AtomicObserver may cause issues that the Rx Guideline 6.4 note warns of? It must be related to the unsubscribe chain but I haven't thought through all of the implications and it's not jumping out to me.

We obviously need some way of improving the libraries resilience against error handling bugs - even the current attempt with AtomicObserver isn't covering all of the scenarios as I just discovered by doing this test.

It is very easy to have holes in the error handling in all of the Rx operator implementations so I'd like to know how we can achieve Guideline 6.4 while admitting it's near impossible to have bug-free Rx operators and accomplish the goal of never having an exception be swallowed or lost on a random thread.

I'd very much appreciate your or anyone else's assistance in figuring out the approach to take and then getting there.

[benjchristensen]

Here are unit tests demonstrating various use cases.

• Synchronous with error in Observer -> Passes because subscribe method has a try/catch that works with synchronous sequences
• Synchronous with error in Observable -> Passes because subscribe method has a try/catch that works with synchronous sequences
• Asynchronous with error in Observer and no composition -> Passes because AtomicObserver catches the error
• Asynchronous with error in Observer and composition -> Fails because AtomicObserver doesn't get used in the right place ... it wraps the composed operator rather than the user provided Observer
• Asynchronous with error in Observable -> Fails because exception is thrown on the user provided thread. Not sure if there is anything that can be done in that situation.

2 of these 5 can be helped by the AtomicObserver but it's only helping 1 of the 2 right now. These 2 use cases are the most common of the 5 so it's valid to pursue a solution I think. The use case that doesn't work could be fixed by using a better approach for deciding when to wrap (or of course finding a completely different approach which is the point of this issue).

/**
 * The error from the user provided Observable is not handled because this is
 * asynchronous and the Exception throws on the thread.
 * 
 * Result: Fails
 */
@Test
public void testCustomObservableWithErrorInObservableAsynchronous() throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(1);
    final AtomicInteger count = new AtomicInteger();
    final AtomicReference<Exception> error = new AtomicReference<Exception>();
    Observable.create(new Func1<Observer<String>, Subscription>() {

        @Override
        public Subscription call(final Observer<String> observer) {
            final BooleanSubscription s = new BooleanSubscription();
            new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        if (!s.isUnsubscribed()) {
                            observer.onNext("1");
                            observer.onNext("2");
                            throw new NumberFormatException();
                        }
                    } finally {
                        latch.countDown();
                    }
                }
            }).start();
            return s;
        }
    }).subscribe(new Observer<String>() {

        @Override
        public void onCompleted() {
            System.out.println("completed");
        }

        @Override
        public void onError(Exception e) {
            error.set(e);
            System.out.println("error");
            e.printStackTrace();
        }

        @Override
        public void onNext(String v) {
            System.out.println(v);
            count.incrementAndGet();
        }

    });

    // wait for async sequence to complete
    latch.await();

    assertEquals(2, count.get());
    assertNotNull(error.get());
    if (!(error.get() instanceof NumberFormatException)) {
        fail("It should be a NumberFormatException");
    }
}

/**
 * The error from the user provided Observer is not handled by the subscribe method try/catch.
 * 
 * It is handled by the AtomicObserver that wraps the provided Observer.
 * 
 * It doesn't compose well though ... if `take(1)` is put between the Observable and
 * subscribe then it still fails.
 * 
 * Result: Passes (if AtomicObserver functionality exists)
 */
@Test
public void testCustomObservableWithErrorInObserverAsynchronous() throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(1);
    final AtomicInteger count = new AtomicInteger();
    final AtomicReference<Exception> error = new AtomicReference<Exception>();
    Observable.create(new Func1<Observer<String>, Subscription>() {

        @Override
        public Subscription call(final Observer<String> observer) {
            final BooleanSubscription s = new BooleanSubscription();
            new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        if (!s.isUnsubscribed()) {
                            observer.onNext("1");
                            observer.onNext("2");
                            observer.onNext("three");
                            observer.onNext("4");
                            observer.onCompleted();
                        }
                    } finally {
                        latch.countDown();
                    }
                }
            }).start();
            return s;
        }
    }).subscribe(new Observer<String>() {

        @Override
        public void onCompleted() {
            System.out.println("completed");
        }

        @Override
        public void onError(Exception e) {
            error.set(e);
            System.out.println("error");
            e.printStackTrace();
        }

        @Override
        public void onNext(String v) {
            int num = Integer.parseInt(v);
            System.out.println(num);
            // doSomething(num);
            count.incrementAndGet();
        }

    });

    // wait for async sequence to complete
    latch.await();

    assertEquals(2, count.get());
    assertNotNull(error.get());
    if (!(error.get() instanceof NumberFormatException)) {
        fail("It should be a NumberFormatException");
    }
}

/**
 * The error from the user provided Observer is not handled by the subscribe method try/catch.
 * 
 * It is also not handled by the AtomicObserver because composition defeats it.
 * 
 * Result: Fails
 */
@Test
public void testCustomObservableWithErrorInObserverAsynchronousWithComposition() throws InterruptedException {
    final CountDownLatch latch = new CountDownLatch(1);
    final AtomicInteger count = new AtomicInteger();
    final AtomicReference<Exception> error = new AtomicReference<Exception>();
    Observable.create(new Func1<Observer<String>, Subscription>() {

        @Override
        public Subscription call(final Observer<String> observer) {
            final BooleanSubscription s = new BooleanSubscription();
            new Thread(new Runnable() {

                @Override
                public void run() {
                    try {
                        if (!s.isUnsubscribed()) {
                            observer.onNext("1");
                            observer.onNext("2");
                            observer.onNext("three");
                            observer.onNext("4");
                            observer.onCompleted();
                        }
                    } finally {
                        latch.countDown();
                    }
                }
            }).start();
            return s;
        }
    }).take(1).subscribe(new Observer<String>() {

        @Override
        public void onCompleted() {
            System.out.println("completed");
        }

        @Override
        public void onError(Exception e) {
            error.set(e);
            System.out.println("error");
            e.printStackTrace();
        }

        @Override
        public void onNext(String v) {
            int num = Integer.parseInt(v);
            System.out.println(num);
            // doSomething(num);
            count.incrementAndGet();
        }

    });

    // wait for async sequence to complete
    latch.await();

    assertEquals(2, count.get());
    assertNotNull(error.get());
    if (!(error.get() instanceof NumberFormatException)) {
        fail("It should be a NumberFormatException");
    }
}

/**
 * The error from the user provided Observer is handled by the subscribe 
 * try/catch because this is synchronous
 * 
 * Result: Passes
 */
@Test
public void testCustomObservableWithErrorInObserverSynchronous() {
    final AtomicInteger count = new AtomicInteger();
    final AtomicReference<Exception> error = new AtomicReference<Exception>();
    Observable.create(new Func1<Observer<String>, Subscription>() {

        @Override
        public Subscription call(Observer<String> observer) {
            observer.onNext("1");
            observer.onNext("2");
            observer.onNext("three");
            observer.onNext("4");
            observer.onCompleted();
            return Subscriptions.empty();
        }
    }).subscribe(new Observer<String>() {

        @Override
        public void onCompleted() {
            System.out.println("completed");
        }

        @Override
        public void onError(Exception e) {
            error.set(e);
            System.out.println("error");
            e.printStackTrace();
        }

        @Override
        public void onNext(String v) {
            int num = Integer.parseInt(v);
            System.out.println(num);
            // doSomething(num);
            count.incrementAndGet();
        }

    });
    assertEquals(2, count.get());
    assertNotNull(error.get());
    if (!(error.get() instanceof NumberFormatException)) {
        fail("It should be a NumberFormatException");
    }
}

/**
 * The error from the user provided Observable is handled by the 
 * subscribe try/catch because this is synchronous
 * 
 * 
 * Result: Passes
 */
@Test
public void testCustomObservableWithErrorInObservableSynchronous() {
    final AtomicInteger count = new AtomicInteger();
    final AtomicReference<Exception> error = new AtomicReference<Exception>();
    Observable.create(new Func1<Observer<String>, Subscription>() {

        @Override
        public Subscription call(Observer<String> observer) {
            observer.onNext("1");
            observer.onNext("2");
            throw new NumberFormatException();
        }
    }).subscribe(new Observer<String>() {

        @Override
        public void onCompleted() {
            System.out.println("completed");
        }

        @Override
        public void onError(Exception e) {
            error.set(e);
            System.out.println("error");
            e.printStackTrace();
        }

        @Override
        public void onNext(String v) {
            System.out.println(v);
            count.incrementAndGet();
        }

    });
    assertEquals(2, count.get());
    assertNotNull(error.get());
    if (!(error.get() instanceof NumberFormatException)) {
        fail("It should be a NumberFormatException");
    }
}

[johngmyers]

If this semantic is to be kept, it should be implemented by having Observable.subscribe() wrap a ProtectObserver around trusted Observables, not by trying to have each protected operation independently implement and test this semantic.

I think what the guidelines are getting at is that protecting onNext() is going to produce onError() notifications that the Observer isn't going to expect to be recieving (especially since the Observer is by definition buggy).

For example, consider an Observer that subscribes to materialize(). It isn't going to expect to get any onError() calls at all, as materialize() has a contract stating that it never emits onError(). The implementations of materialize() consumers that I've seen do nothing on onError() calls*, so that means the exception would be silently ignored and the bug likely never discovered.

* In my opinion, they should instead throw UnsupportedOperationExcetion()

Put another way, I think the guidelines are trying to say that attempting to protect notification callbacks leads you down a rabbit hole, which I think you're starting to discover.

It is, however, hard to argue against operational experience. It doesn't work to monitor the exceptions coming out of "random" threads, using them to identify the Observer bugs so they can be fixed?

Perhaps a ProtectObserver that Observable.subscribe() wraps around all subscriptions should feed a copy of all Throwables thrown by Observer notification callbacks to a plugin. Monitoring could then use the plugin to identify Observer bugs that need to be fixed.

[benjchristensen]

If this semantic is to be kept, it should be implemented by having Observable.subscribe() wrap a ProtectObserver around trusted Observables, not by trying to have each protected operation independently implement and test this semantic.

I agree which is what was seeking to be accomplished with AtomicObserver. It's concept is similar to how every operator shouldn't have to synchronize everything - error handling should if possible be abstracted to a single place.

I was looking back through the history of our refactoring from internal code to open source RxJava and found where it broke down. We wrapped every Observer passed in to subscribe so it handled both of the "asynchronous observer with errors" use cases.

This code got mangled while doing performance optimizations (#104) by reducing extraneous wrapping and synchronization and unit tests didn't exist for the error handling use case we lost and I rediscovered while investigating this issue.

Perhaps a ProtectObserver that Observable.subscribe() wraps around all subscriptions should feed a copy of all Throwables thrown by Observer notification callbacks to a plugin. Monitoring could then use the plugin to identify Observer bugs that need to be fixed.

This is the reason for the RxJavaErrorHandler plugin. We use this in production to capture any errors that occur so we see them even if an Observer implementation doesn't correctly implement onError.

---

I'll work on this further to get improved unit testing around error handling and explore a clean way to provide the protection without hurting the performance benefits we got in #104.

[johngmyers]

This is the reason for the RxJavaErrorHandler plugin. We use this in production to capture any errors that occur so we see them even if an Observer implementation doesn't correctly implement onError.

That's not the same thing. RxJavaErrorHandler captures exceptions passed through onError() notifications to a particular set of Observers. I'm talking about something that captures exceptions thrown from any Observer notification.

[johngmyers]

...an exception passed through onError() into RxJavaErrorHandler is likely to be an operational error, such as a transient network failure. An exception thrown from an Observer notification is guaranteed to be the result of a coding bug.

[benjchristensen]

So you're suggesting a separate plugin (or method/type on the existing plugin) that specifically handles only errors thrown by the Observer. That's an interesting idea. I can see the value in knowing the difference.

[johngmyers]

This particular take operator is heavily unit tested and at least 2 sets of eyes reviewed it and both missed it.

I hope you're not counting my set of eyes here. I noticed this hole in most trusted operators when writing assertTrustedObservable(). What I didn't notice is that Observable.subscribe() partially papers over the hole for the case of synchronous trusted Observables.

If you want users of an interface to comply with a contract requirement, it helps to either enforce the requirement or at least to provide tools to allow implementers to unit test conformance to the requirement.

[benjchristensen]

Enforcing the requirements is the intent of things like AtomicObserver and AtomicObservableSubscription. Tools for testing conformance is nice and an interesting idea but it obviously does nothing to deal with non-conformance in a running system. We can take that on elsewhere (such as the discussion on the other issue about your dislike of my removal of rx.testing).

I have a question about implementation details for wrapping an incoming Observer ... I've asked around several folks internally as well. All approaches have drawbacks I think.

The simplest and cleanest option is to wrap every Observer passed into subscribe. This is what we did prior to #104. The drawback is an extra wrapping at every level of composition.

Since it is ensuring conformance of the contract and not just try/catch behavior this also means it is doing volatile checks which in a tight-loop and deep composition chain causes performance impacts.

# wrapping at every level of composition (50 deep)
compositionTestTotalTime: 4392
nonCompositionalTestWithDirectLoopTotalTime: 1913
nonCompositionalTestWithArrayOfFunctionsTotalTime: 1826

# wrapping only one Observer of the 50
compositionTestTotalTime: 2248
nonCompositionalTestWithDirectLoopTotalTime: 1887
nonCompositionalTestWithArrayOfFunctionsTotalTime: 1813

The numbers are in milliseconds but only mean anything relative to each other. It makes a difference though: 4392 vs 2248.

So, to avoid wrapping every layer it would be nice to only wrap the final Observer passed in by user code.

To know if an Observer is from internal or external can be done by:

1. Inspecting the package of the Observer implementation

This is the cleanest of the 4 options but I question the performance impact (need to test this) of checking every single time.

2. Using a marker interface on internally implemented Observers (i.e. OperatorObserver extends Observer)

I dislike marker interfaces but it would perform well.
It also requires every operator correctly doing this which is error prone (though limited impact if someone forgets).

3. Annotations on internally implemented Observers

Annotations are basically just a different flavor of a marker interface. It adds more complexity to scanning them.
At runtime I would need to test their performance.
It also requires every operator correctly doing this which is error prone (though limited impact if someone forgets).

4. Separate subscribeNonProtected method on Observable

I don't like this because it would have to be a public method as Java can't allow me to make a method only accessible to certain packages.

Right now I'm pursuing either option 1 or 2 and will report back on my findings on option 1 which I'd prefer if it performs well.

Any other approaches that you feel are better or what is your opinion on these 4?

[benjchristensen]

This code for option 1 appears to perform fine:

            /**
             * See https://github.com/Netflix/RxJava/issues/216 for discussion on "Guideline 6.4: Protect calls to user code from within an operator"
             */
            if (observer.getClass().getPackage().getName().startsWith("rx")) {
                Subscription s = onSubscribe.call(observer);
                if (s == null) {
                    // this generally shouldn't be the case on a 'trusted' onSubscribe but in case it happens
                    // we want to gracefully handle it the same as AtomicObservableSubscription does
                    return Subscriptions.empty();
                } else {
                    return s;
                }
            } else {
                AtomicObservableSubscription subscription = new AtomicObservableSubscription();
                return subscription.wrap(onSubscribe.call(new AtomicObserver<T>(subscription, observer)));
            }

I was expecting observer.getClass().getPackage().getName().startsWith("rx") to be a problem but the performance test shows this:

compositionTestTotalTime3: 2361
nonCompositionalTestWithDirectLoopTotalTime: 1879
nonCompositionalTestWithArrayOfFunctionsTotalTime: 1940

Since it's in the right order of magnitude (and faster than wrapping everything) I'm proceeding with option 1 for now unless someone comes up with a better solution.

[johngmyers]

AtomicObserver and AtomicObservableSubscription have to do synchronization, so have to pay the synchronization cost. A ProtectObserver would not have to do synchronization, so should be much cheaper.

[benjchristensen]

They don't do synchronization. They use java.util.concurrent.atomic.* classes which use compareAndSet or volatile - not synchronization.

[johngmyers]

java.util.concurrent.* and volatile are synchronization.