Rethrow dispatcher exceptions during the undispatched coroutines start immediately

Otherwise, in the face of malfunctioning dispatcher, there might be no trace of any error, making it extremely hard to debug.

The root cause should be addressed by #4209

Fixes #4142

Author: qwwdfsad

kotlinx-coroutines-core/common/src/intrinsics/Undispatched.kt

@@ -64,6 +64,10 @@ private fun <T, R> ScopeCoroutine<T>.startUndspatched(
 ): Any? {
     val result = try {
         block.startCoroutineUninterceptedOrReturn(receiver, this)
+    } catch (e: DispatchException) {
+        // Special codepath for failing CoroutineDispatcher: rethrow an exception
+        // immediately without waiting for children to indicate something is wrong
+        dispatchException(e)
     } catch (e: Throwable) {
         CompletedExceptionally(e)
     }
@@ -93,3 +97,8 @@ private fun <T, R> ScopeCoroutine<T>.startUndspatched(
 private fun ScopeCoroutine<*>.notOwnTimeout(cause: Throwable): Boolean {
     return cause !is TimeoutCancellationException || cause.coroutine !== this
 }
+
+private fun ScopeCoroutine<*>.dispatchException(e: DispatchException) {
+    makeCompleting(CompletedExceptionally(e.cause))
+    throw throw recoverStackTrace(e.cause, uCont)
+}

kotlinx-coroutines-core/jvm/test/FailFastOnStartTest.kt

@@ -4,9 +4,17 @@ package kotlinx.coroutines
 
 import kotlinx.coroutines.testing.*
 import kotlinx.coroutines.channels.*
+import kotlinx.coroutines.flow.emptyFlow
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
 import org.junit.*
 import org.junit.Test
 import org.junit.rules.*
+import org.junit.runners.model.TestTimedOutException
+import java.util.concurrent.TimeoutException
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.EmptyCoroutineContext
+import kotlin.coroutines.startCoroutine
 import kotlin.test.*
 
 class FailFastOnStartTest : TestBase() {
@@ -81,4 +89,24 @@ class FailFastOnStartTest : TestBase() {
     fun testAsyncNonChild() = runTest(expected = ::mainException) {
         async<Int>(Job() + Dispatchers.Main) { fail() }
     }
+
+    @Test
+    fun testFlowOn()  {
+        // See #4142, this test ensures that `coroutineScope { produce(failingDispatcher, ATOMIC) }`
+        // rethrows an exception. It does not help with the completion of such a coroutine though.
+        // Hack to avoid waiting for test completion
+        expect(1)
+        val caller = suspend {
+            try {
+                emptyFlow<Int>().flowOn(Dispatchers.Main).collect { fail() }
+            } catch (e: Throwable) {
+                assertTrue(mainException(e))
+                expect(2)
+            }
+        }
+
+        caller.startCoroutine(Continuation(EmptyCoroutineContext) {
+            finish(3)
+        })
+    }
 }