3.x: Fix scheduled tasks' fatal exception behavior

[akarnokd]

Fatal exceptions may be lost with scheduled direct & scheduled periodic direct tasks because FutureTask simply treats them as exceptional outcomes. For regular tasks, ScheduledRunnable already avoids rethrowing fatal errors as it would go nowhere.

This PR adds this behavior to the direct runnable tasks.

Resolves #6954

[jxdabc]

But it seems make #748 a problem again. It seems tricky to balance this issue and #748. After all, if we passed StackOverflowError to RxJavaPlugins.onError(), RxJavaPlugins.onError() should throw it and it will be swallowed again.

I think customizing the future task will completely solve #748 and this issue. JDK supports that and at least as I know Android are using that approach. And I think it is not complex as the core code is only implementing the JDK's interface. It even simplifies the schedulers implementation, as I see ; )

[jxdabc]

I'm sorry, I think I made a mistake.
I'll check it.

[akarnokd]

this will halt the application when stack overflow:

I don't understand. As far as I know, calling the uncaught exception handler on Android kills your app. On desktop, such calls do nothing but the repetition of the schedulePeriodicallyDirect should end.

[jxdabc]

this will halt the application when stack overflow:

I don't understand. As far as I know, calling the uncaught exception handler on Android kills your app. On desktop, such calls do nothing but the repetition of the schedulePeriodicallyDirect should end.

I am sorry. I think I made a mistake when simplifies #748 case. The key point is when StackOverflowError ocurrs calling error hander without throw may cause stack full again. That is also why I selected a complex approach. I'll check it.

[jxdabc]

this will halt the application when stack overflow:

I don't understand. As far as I know, calling the uncaught exception handler on Android kills your app. On desktop, such calls do nothing but the repetition of the schedulePeriodicallyDirect should end.

Hi, akarnokd, sorry to the above mistake.

Personally, I think this approach has some flaws.

1 - This approach means we'll catch fatal exceptions instead of throwing it through the call stack. But the functions on the call stack may need this exception to clean/stop itself. For example, this change will make periodic tasks unable to stop (at least on desktop or server):

@Test // Fail
public void periodicTaskShouldStopOnError() throws Exception {
    AtomicInteger repeatCount = new AtomicInteger();
    new ComputationScheduler().schedulePeriodicallyDirect(new Runnable() {
        @Override
        public void run() {
            repeatCount.incrementAndGet();
            if (true) {
                throw new OutOfMemoryError();
            }
        }
    }, 0, 500, TimeUnit.MILLISECONDS);

    Thread.sleep(1000 * 3);
    assertEquals(1, repeatCount.get());
}

2 - To solve the above problem, we need to rethrow the exception after RxJavaPlugins.onError() or make RxJavaPlugins.onError() to throw after calling the uncaughtExceptionHandler of the thread. This actullay makes some exceptions to be handled more than once. And I think it will make things more and more complex as we solve those cases one by one.

3 - If we are able to check exceptions in the Future as #6954 mentioned (and JDK has given us all we need). Exception handing in RxJava will be simpler and easier as we do elsewhere: make RxJavaPlugins.onError() handle/propagate the exception and continue the work, or make RxJavaPlugins.onError() throws and exceptions go through the stack to do whatever clean/stop needed and finally to the uncaughtExceptionHandler of the thread.

[codecov]

Codecov Report

Merging #6956 into 3.x will increase coverage by 0.01%.
The diff coverage is 100.00%.

@@             Coverage Diff              @@
##                3.x    #6956      +/-   ##
============================================
+ Coverage     99.64%   99.65%   +0.01%     
- Complexity     6667     6668       +1     
============================================
  Files           742      742              
  Lines         47228    47248      +20     
  Branches       6369     6369              
============================================
+ Hits          47059    47084      +25     
+ Misses           51       50       -1     
+ Partials        118      114       -4     

Impacted Files	Coverage Δ	Complexity Δ
...main/java/io/reactivex/rxjava3/core/Scheduler.java	100.00% <100.00%> (ø)	11.00 <0.00> (ø)
...rxjava3/internal/schedulers/ExecutorScheduler.java	100.00% <100.00%> (ø)	10.00 <0.00> (ø)
...java3/internal/schedulers/InstantPeriodicTask.java	100.00% <100.00%> (ø)	20.00 <0.00> (ø)
...ternal/schedulers/ScheduledDirectPeriodicTask.java	100.00% <100.00%> (ø)	2.00 <0.00> (ø)
...java3/internal/schedulers/ScheduledDirectTask.java	100.00% <100.00%> (ø)	2.00 <0.00> (ø)
...rxjava3/internal/schedulers/ScheduledRunnable.java	100.00% <100.00%> (ø)	30.00 <0.00> (ø)
...tivex/rxjava3/internal/jdk8/ParallelCollector.java	93.57% <0.00%> (-4.59%)	2.00% <0.00%> (ø%)
...l/operators/observable/ObservableFlatMapMaybe.java	91.54% <0.00%> (-3.53%)	2.00% <0.00%> (ø%)
.../internal/disposables/ListCompositeDisposable.java	98.00% <0.00%> (-2.00%)	34.00% <0.00%> (-1.00%)
...a3/internal/operators/flowable/FlowableCreate.java	98.05% <0.00%> (-0.98%)	6.00% <0.00%> (ø%)
... and 13 more

---

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update b47783e...48bf526. Read the comment docs.

[jxdabc]

Disposing the periodic task on error may be a little tricky workaround.

1 - The periodic FutureTask is canceled asynchronously, until PeriodicDirectTask.setFuture is done. This may cause the periodic task to run some more rounds before actually stop. The following test almost can't success on my laptop:

@Test
public void periodicTaskShouldStopOnError() throws InterruptedException {
    try {
        // suppress log
        RxJavaPlugins.setErrorHandler((e) -> {});

        CountDownLatch latch = new CountDownLatch(1000);

        AtomicInteger assertCount = new AtomicInteger();
        for (int i = 0; i < 1000; i++) {
            new Thread(new Runnable() {
                @Override
                public void run() {
                    try {
                        periodicTaskShouldStopOnErrorOnce();
                    } catch (AssertionError e) {
                        e.printStackTrace();
                        assertCount.incrementAndGet();
                    } catch (Exception e) {
                    } finally {
                        latch.countDown();
                    }
                }
            }).start();
        }

        latch.await();
        assertEquals(0, assertCount.get());
    } finally {
        RxJavaPlugins.reset();
    }
}

private void periodicTaskShouldStopOnErrorOnce() throws Exception {
    AtomicInteger repeatCount = new AtomicInteger();

    Schedulers.computation().schedulePeriodicallyDirect(new Runnable() {
        @Override
        public void run() {
            repeatCount.incrementAndGet();
            if (true) {
                throw new OutOfMemoryError();
            }
        }
    }, 0, 1, TimeUnit.NANOSECONDS);

    Thread.sleep(200);

    assertEquals(1, repeatCount.get());
}

2 - I wonder whether we should route InterruptibleRunnable's & BooleanRunnable's errors to RxJavaPlugins.onError() as well. If not, I have to handle errors both in RxJavaPlugins.onError() and thread's uncaught exception handlers. When schedulers are involved, errors may be pass to either of the two.

3 - My original PR was actually intending to make things easy for my team: we only need to set an error handle via RxJavaPlugins.setErrorHandler which prints a log for not-a-matter error and rethrow for fatal erros and every thing goes fine. However, maybe I could replace all schedulers with RxJavaPlugins.setXXXSchedulerHandler without change the main stream though I think my original PR simplifies implemetation of the main stream without side effect as I could see.

[akarnokd]

1. Looks like the periodic task should be crashed nonetheless.
2. Yes, task wrapper behavior should be consistent across.
3. "make things easy for my team" - that's why I encourage people to write their own hooks and components to workaround very specific corner cases.

[jxdabc]

2. Yes, task wrapper behavior should be consistent across.

Would you like to do this in this PR or later?

[akarnokd]

@JakeWharton Does this syntax work with Android desugar (rethrowing final Throwables)?

[JakeWharton]

Hmm this looks like a normal throw. I don't understand what might be a compatibility concern here.

[akarnokd]

The Java 7 compiler feature that let's you rethrow constant cheched or Throwable exceptions from within a catch block, even if the method didn't specify a throws clause.

[JakeWharton]

Ah, missed it was Throwable. Checked exceptions are only a feature of the Java compiler not the Java bytecode so yeah the Android toolchain won't care 👍 .

[jxdabc]

Great, that fixed all corner cases I could see.
But I wonder whether calling RxJavaPlugins.onError(ex) (calling uncaught exception handler) in the middle of exception throw chain is good and intuitional. A little tricky as I see.
If you'd have a look on #6955, the original PR. It requires no more change than this one, handling Future related erros in ONE place and not calling uncaught exception handler in the middle of exception throw chain. It should pass all tests in this PR as well without modification. :)