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,18 @@ 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) {
+            val job = context[Job]
+            if (job != null && !job.isActive) {
+                val cause = job.getCancellationException()
+                cancelResult(state, cause)
+                throw cause
+            }
+        }
         return getSuccessfulResult(state)
     }
 

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

@@ -0,0 +1,59 @@
+/*
+ * Copyright 2016-2019 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import kotlinx.atomicfu.*
+import org.junit.*
+import java.util.concurrent.*
+import kotlin.test.*
+import kotlin.test.Test
+
+class CancellableContinuationResumeCloseStressTest : TestBase() {
+    private val dispatcher =
+        newFixedThreadPoolContext(2, "CancellableContinuationResumeCloseStressTest")
+    private val barrier = CyclicBarrier(3)
+    private val nRepeats = 1_000 * stressTestMultiplier
+
+    private val closed = atomic(false)
+
+    @After
+    fun tearDown() {
+        dispatcher.close()
+    }
+
+    @Test
+    fun testStress() = runTest {
+        repeat(nRepeats) {
+            closed.value = false
+            val job = launch(dispatcher) {
+                try {
+                    val ok = resumeClose()
+                    assertEquals("OK", ok)
+                    assertFalse(closed.value)
+                } catch (e: CancellationException) {
+                    assertTrue(closed.value)
+                }
+            }
+            @Suppress("BlockingMethodInNonBlockingContext")
+            barrier.await()
+            job.cancel() // (1) cancel job
+            job.join()
+        }
+    }
+
+    private suspend fun resumeClose() = suspendCancellableCoroutine<String> { cont ->
+        dispatcher.executor.execute {
+            barrier.await() // (2) resume at the same time
+            cont.resume("OK") {
+                close()
+            }
+        }
+        barrier.await() // (3) return at the same time
+    }
+
+    fun close() {
+        assertFalse(closed.getAndSet(true))
+    }
+}

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 = 1_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")
+                }
+            }
+        }
+    }
+}