Fix a race in Job.join that sporadically results in normal completion

The race happens in the slow-path of 'join' implementation when
parent invokes join on a child coroutines that crashes and cancels
the parent.

Fixes #1123

Author: elizarov

kotlinx-coroutines-core/common/src/CancellableContinuationImpl.kt

@@ -170,6 +170,11 @@ internal open class CancellableContinuationImpl<in T>(
         // otherwise, onCompletionInternal was already invoked & invoked tryResume, and the result is in the state
         val state = this.state
         if (state is CompletedExceptionally) throw recoverStackTrace(state.cause, this)
+        // if the parent job was already cancelled, then throw the corresponding cancellation exception
+        // otherwise, there is a race is suspendCancellableCoroutine { cont -> ... } does cont.resume(...)
+        // before the block returns. This getResult would return a result as opposed to cancellation
+        // exception that should have happened if the continuation is dispatched for execution later.
+        if (resumeMode == MODE_CANCELLABLE) context.checkCompletion()
         return getSuccessfulResult(state)
     }
 

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

@@ -0,0 +1,37 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.*
+
+/**
+ * Test a race between job failure and join.
+ *
+ * See [#1123](https://github.com/Kotlin/kotlinx.coroutines/issues/1123).
+ */
+class JobStructuredJoinStressTest : TestBase() {
+    private val nRepeats = 10_000 * stressTestMultiplier
+
+    @Test
+    fun testStress() {
+        repeat(nRepeats) {
+            assertFailsWith<TestException> {
+                runBlocking {
+                    // launch in background
+                    val job = launch(Dispatchers.Default) {
+                        throw TestException("OK") // crash
+                    }
+                    try {
+                        job.join()
+                    } catch (e: CancellationException) {
+                        // expected -- proceed
+                        throw e
+                    }
+                    error("job.join returned normally, but it should not have")
+                }
+            }
+        }
+    }
+}