Avoid catching fatal errors as a Failed future

src/main/scala/scala/async/FutureStateMachine.scala

@@ -15,6 +15,7 @@ import java.util.Objects
 
 import scala.util.{Failure, Success, Try}
 import scala.concurrent.{ExecutionContext, Future, Promise}
+import scala.util.control.NonFatal
 
 /** The base class for state machines generated by the `scala.async.Async.async` macro.
  *  Not intended to be directly extended in user-written code.
@@ -34,12 +35,14 @@ abstract class FutureStateMachine(execContext: ExecutionContext) extends Functio
   /** Assign `i` to the state variable */
   protected def state_=(s: Int): Unit = state$async = s
 
+  NonFatal // eagerly classloading NonFatal to reduce the chance of a cascading StackOverflowError in `completeFailure`
+
   /** Complete the state machine with the given failure. */
-  // scala-async accidentally started catching NonFatal exceptions in:
-  //  https://github.com/scala/scala-async/commit/e3ff0382ae4e015fc69da8335450718951714982#diff-136ab0b6ecaee5d240cd109e2b17ccb2R411
-  // This follows the new behaviour but should we fix the regression?
-  protected def completeFailure(t: Throwable): Unit = {
-    result$async.complete(Failure(t))
+  protected def completeFailure(t: Throwable): Unit = t match {
+    case NonFatal(t) =>
+      result$async.complete(Failure(t))
+    case _ =>
+      throw t
   }
 
   /** Complete the state machine with the given value. */

src/test/scala/scala/async/ExceptionalTest.scala

@@ -0,0 +1,42 @@
+package scala.async
+
+import java.util.concurrent.atomic.AtomicReference
+
+import org.junit.{Assert, Ignore, Test}
+
+import scala.async.Async.{async, await}
+import scala.concurrent.{ExecutionContext, Future, _}
+import scala.language.postfixOps
+import scala.util.control.NonFatal
+
+class ExceptionalTest {
+  @Test
+  def nonFatalNotCaughtFutureCombinators(): Unit = {
+    check { implicit ec =>
+      Future.successful(42).map(x => (x, throw fatal))
+    }
+  }
+
+  @Test
+  def nonFatalNotCaughtAsync(): Unit = {
+    check { implicit ec =>
+      async {
+        (await(Future.successful(42)), throw fatal)
+      }
+    }
+  }
+
+  def check(f: ExecutionContext => Future[Any]): Unit = {
+    val lastUncaught = new AtomicReference[Throwable]()
+    implicit val executor: ExecutionContextExecutor = ExecutionContext.fromExecutor(null, lastUncaught.set(_))
+    val future = f(executor)
+    Thread.sleep(100)
+    Assert.assertSame(fatal, lastUncaught.get())
+  }
+
+  private val fatal: Throwable = {
+    val t = new VirtualMachineError() {}
+    Assert.assertTrue(NonFatal.unapply(t).isEmpty)
+    t
+  }
+}