Support completing the test coroutine from outside the test thread.

This requires a call to invokeOnCompletion to get the completion state
from the other thread.

Changes:
   - runBlockingTest now supports completing the test coroutine from
     another thread
   - success and failure path tests

While fixing this, a subtle non-determinism was discovered that would
lead to flakey tests. If another thread completed the test coroutine
*during* the cleanup checks it was possible to modify the state of the
test during the cleanup checks in a way that could expose false-positive
or false-negative results randomly.

To resolve this, a non-completed coroutine immediately after
advanceUntilIdle is now considered a failure, even if it completes
before the more aggressive cleanup checks. There is still a very brief
window (between advanceTimeBy and getResultIfKnown) that non-determinism
may be introduced, but it will fail with a descriptive error message at
on random executions directing the developer to resolve the
non-determinstic behavior.

Note: testWithOddlyCompletingJob_fails may fail when the implementation
of runBlockingTest changes (it's very tightly coupled).

Author: objcode

kotlinx-coroutines-test/src/TestBuilders.kt

@@ -46,18 +46,56 @@ public fun runBlockingTest(context: CoroutineContext = EmptyCoroutineContext, te
     val (safeContext, dispatcher) = context.checkArguments()
     val startingJobs = safeContext.activeJobs()
     val scope = TestCoroutineScope(safeContext)
+
     val deferred = scope.async {
         scope.testBody()
     }
+
+    // run any outstanding coroutines that can be completed by advancing virtual-time
     dispatcher.advanceUntilIdle()
-    deferred.getCompletionExceptionOrNull()?.let {
-        throw it
-    }
+
+    // fetch results from the coroutine - this may require a thread hop if some child coroutine was *completed* on
+    // another thread during this test so we must use an invokeOnCompletion handler to retrieve the result.
+
+    // There are two code paths for fetching the error:
+    //
+    // 1. The job was already completed (happy path, normal test)
+    //    - invokeOnCompletion was executed immediately and errorThrownByTestOrNull is already at it's final value so
+    //      we can throw it
+    // 2. The job has not already completed (always fail the test due to error or time-based non-determinism)
+    //    - invokeOnCompletion will not be triggered right away. To avoid introducing wall non-deterministic  behavior
+    //      (the deferred may complete between here and the call to activeJobs below) this will always be considered a
+    //      test failure.
+    //    - this will not happen if all coroutines are only waiting to complete due to thread hops, but may happen
+    //      if another thread triggers completion concurrently with this cleanup code.
+    //
+    // give test code errors a priority in the happy path, throw here if the error is already known.
+    val (wasCompletedAfterTest, errorThrownByTestOrNull) = deferred.getResultIfKnown()
+    errorThrownByTestOrNull?.let { throw it }
+
     scope.cleanupTestCoroutines()
     val endingJobs = safeContext.activeJobs()
     if ((endingJobs - startingJobs).isNotEmpty()) {
         throw UncompletedCoroutinesError("Test finished with active jobs: $endingJobs")
     }
+
+    if (!wasCompletedAfterTest) {
+        // Handle path #2, we are going to fail the test in an opinionated way at this point so let the developer know
+        // how to fix it.
+        throw UncompletedCoroutinesError("Test completed all jobs after cleanup code started. This is " +
+                "thrown to avoid non-deterministic behavior in tests (the next execution may fail randomly). Ensure " +
+                "all threads started by the test are completed before returning from runBlockingTest.")
+    }
+}
+
+private fun Deferred<Unit>.getResultIfKnown(): Pair<Boolean, Throwable?> {
+    var testError: Throwable? = null
+    var wasExecuted = false
+    invokeOnCompletion { errorFromTestOrNull ->
+        testError = errorFromTestOrNull
+        wasExecuted = true
+    }.dispose()
+    return Pair(wasExecuted, testError)
 }
 
 private fun CoroutineContext.activeJobs(): Set<Job> {

kotlinx-coroutines-test/test/TestRunBlockingOrderTest.kt

@@ -6,7 +6,9 @@ package kotlinx.coroutines.test
 
 import kotlinx.coroutines.*
 import org.junit.*
+import kotlin.concurrent.thread
 import kotlin.coroutines.*
+import kotlin.test.assertEquals
 
 class TestRunBlockingOrderTest : TestBase() {
     @Test
@@ -67,4 +69,94 @@ class TestRunBlockingOrderTest : TestBase() {
         }
         expect(2)
     }
+
+    @Test
+    fun testNewThread_inSuspendCancellableCoroutine() = runBlockingTest {
+        expect(1)
+        suspendCancellableCoroutine<Unit> { cont ->
+            expect(2)
+            thread {
+                expect(3)
+                cont.resume(Unit)
+            }
+        }
+        finish(4)
+    }
+
+    @Test(expected = UncompletedCoroutinesError::class)
+    fun testWithOddlyCompletingJob_fails() {
+        // this test is suspect since it relies upon the exact ordering of code in runBlockingTest
+        // however, it needs to ensure the job finishes *after* advanceUntilIdle is called in order
+        // to ensure that runBlockingTest errors when presented with threading non-determinism.
+
+        // this test is stable and will always pass unless the implementation changes.
+
+        // If this starts failing because the call to cleanupTestCoroutines changes it will need a similarly
+        // implementation driven test.
+
+        class FakeDispatcher(val delegate: TestCoroutineDispatcher):
+                CoroutineDispatcher(),
+                Delay by delegate,
+                DelayController by delegate {
+            private var cleanupCallback: (() -> Unit)? = null
+
+            override fun dispatch(context: CoroutineContext, block: Runnable) {
+                delegate.dispatch(context, block)
+            }
+
+            fun onCleanup(block: () -> Unit) {
+                cleanupCallback = block
+            }
+
+            override fun cleanupTestCoroutines() {
+                delegate.cleanupTestCoroutines()
+                cleanupCallback?.invoke()
+            }
+        }
+
+        val dispatcher = FakeDispatcher(TestCoroutineDispatcher())
+        val scope = TestCoroutineScope(dispatcher)
+        val resumeAfterTest = CompletableDeferred<Unit>()
+
+        scope.runBlockingTest {
+            expect(1)
+            dispatcher.onCleanup {
+                // after advanceTimeUntilIdle, complete the launched coroutine
+                expect(3)
+                resumeAfterTest.complete(Unit)
+                finish(5)
+            }
+            expect(2)
+            resumeAfterTest.await() // this will resume just before child jobs are checked
+            expect(4)
+        }
+    }
+
+    @Test
+    fun testThrows_throws() {
+        val expected = IllegalStateException("expected")
+        val result = runCatching {
+            expect(1)
+            runBlockingTest {
+                expect(2)
+                throw expected
+            }
+        }
+        finish(3)
+        assertEquals(expected, result.exceptionOrNull())
+    }
+
+    @Test
+    fun testSuspendForever_fails() {
+        val uncompleted = CompletableDeferred<Unit>()
+        val result = runCatching {
+            expect(1)
+            runBlockingTest {
+                expect(2)
+                uncompleted.await()
+            }
+        }
+        finish(3)
+        assertEquals(true, result.isFailure)
+    }
 }