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,68 @@
+/*
+ * 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 startBarrier = CyclicBarrier(3)
+    private val doneBarrier = CyclicBarrier(2)
+    private val nRepeats = 1_000 * stressTestMultiplier
+
+    private val closed = atomic(false)
+    private var returnedOk = false
+
+    @After
+    fun tearDown() {
+        dispatcher.close()
+    }
+
+    @Test
+    @Suppress("BlockingMethodInNonBlockingContext")
+    fun testStress() = runTest {
+        repeat(nRepeats) {
+            closed.value = false
+            returnedOk = false
+            val job = testJob()
+            startBarrier.await()
+            job.cancel() // (1) cancel job
+            job.join()
+            // check consistency
+            doneBarrier.await()
+            if (returnedOk) {
+                assertFalse(closed.value, "should not have closed resource -- returned Ok")
+            } else {
+                assertTrue(closed.value, "should have closed resource -- was cancelled")
+            }
+        }
+    }
+
+    private fun CoroutineScope.testJob(): Job = launch(dispatcher, start = CoroutineStart.ATOMIC) {
+        val ok = resumeClose() // might be cancelled
+        assertEquals("OK", ok)
+        returnedOk = true
+    }
+
+    private suspend fun resumeClose() = suspendCancellableCoroutine<String> { cont ->
+        dispatcher.executor.execute {
+            startBarrier.await() // (2) resume at the same time
+            cont.resume("OK") {
+                close()
+            }
+            doneBarrier.await()
+        }
+        startBarrier.await() // (3) return at the same time
+    }
+
+    fun close() {
+        assertFalse(closed.getAndSet(true))
+    }
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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
+                    }
+                    assertFailsWith<CancellationException> {
+                        job.join()
+                    }
+                }
+            }
+        }
+    }
+}