Ensure that all coroutines throwables in core are serializable

Fixes #3328

integration/kotlinx-coroutines-guava/test/ListAllCoroutineThrowableSubclassesTest.kt

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines.guava
+
+import com.google.common.reflect.*
+import kotlinx.coroutines.*
+import org.junit.Test
+import kotlin.test.*
+
+class ListAllCoroutineThrowableSubclassesTest : TestBase() {
+
+    /*
+     * These are all known throwables in kotlinx.coroutines.
+     * If you have added one, this test will fail to make
+     * you ensure your exception type is java.io.Serializable.
+     *
+     * We do not have means to check it automatically, so checks are delegated to humans.
+     * Also, this test meant to be in kotlinx-coroutines-core, but properly scanning classpath
+     * requires guava which is toxic dependency that we'd like to avoid even in tests.
+     *
+     * See #3328 for serialization rationale.
+     */
+    private val knownThrowables = setOf(
+        "kotlinx.coroutines.TimeoutCancellationException",
+        "kotlinx.coroutines.JobCancellationException",
+        "kotlinx.coroutines.internal.UndeliveredElementException",
+        "kotlinx.coroutines.CompletionHandlerException",
+        "kotlinx.coroutines.DiagnosticCoroutineContextException",
+        "kotlinx.coroutines.CoroutinesInternalError",
+        "kotlinx.coroutines.channels.ClosedSendChannelException",
+        "kotlinx.coroutines.channels.ClosedReceiveChannelException",
+        "kotlinx.coroutines.flow.internal.ChildCancelledException",
+        "kotlinx.coroutines.flow.internal.AbortFlowException",
+
+        )
+
+    @Test
+    fun testThrowableSubclassesAreSerializable() {
+        var throwables = 0
+        val classes = ClassPath.from(this.javaClass.classLoader)
+            .getTopLevelClassesRecursive("kotlinx.coroutines");
+        classes.forEach {
+            try {
+                if (Throwable::class.java.isAssignableFrom(it.load())) {
+                    // Skip classes from test sources
+                    if (it.load().protectionDomain.codeSource.location.toString().contains("/test/")) {
+                        return@forEach
+                    }
+                    ++throwables
+                    // println(""""$it",""")
+                    assertTrue(knownThrowables.contains(it.toString()))
+                }
+            } catch (e: Throwable) {
+                // Ignore unloadable classes
+            }
+        }
+
+        assertEquals(knownThrowables.size, throwables)
+    }
+}

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

@@ -163,7 +163,7 @@ private class TimeoutCoroutine<U, in T: U>(
  */
 public class TimeoutCancellationException internal constructor(
     message: String,
-    @JvmField internal val coroutine: Job?
+    @JvmField @Transient internal val coroutine: Job?
 ) : CancellationException(message), CopyableThrowable<TimeoutCancellationException> {
     /**
      * Creates a timeout exception with the given message.
@@ -173,7 +173,7 @@ public class TimeoutCancellationException internal constructor(
     internal constructor(message: String) : this(message, null)
 
     // message is never null in fact
-    override fun createCopy(): TimeoutCancellationException? =
+    override fun createCopy(): TimeoutCancellationException =
         TimeoutCancellationException(message ?: "", coroutine).also { it.initCause(this) }
 }
 

kotlinx-coroutines-core/jvm/src/Exceptions.kt

@@ -29,7 +29,7 @@ public actual fun CancellationException(message: String?, cause: Throwable?) : C
 internal actual class JobCancellationException public actual constructor(
     message: String,
     cause: Throwable?,
-    @JvmField internal actual val job: Job
+    @JvmField @Transient internal actual val job: Job
 ) : CancellationException(message), CopyableThrowable<JobCancellationException> {
 
     init {

kotlinx-coroutines-core/jvm/src/flow/internal/FlowExceptions.kt

@@ -8,7 +8,7 @@ import kotlinx.coroutines.*
 import kotlinx.coroutines.flow.*
 
 internal actual class AbortFlowException actual constructor(
-    actual val owner: FlowCollector<*>
+    @JvmField @Transient actual val owner: FlowCollector<*>
 ) : CancellationException("Flow was aborted, no more elements needed") {
 
     override fun fillInStackTrace(): Throwable {

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

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2016-2022 JetBrains s.r.o. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package kotlinx.coroutines
+
+import org.junit.*
+import java.io.*
+
+
+@Suppress("BlockingMethodInNonBlockingContext")
+class JobCancellationExceptionSerializerTest : TestBase() {
+
+    @Test
+    fun testSerialization() = runTest {
+        try {
+            coroutineScope {
+                expect(1)
+
+                launch {
+                    expect(2)
+                    try {
+                        hang {}
+                    } catch (e: CancellationException) {
+                        throw RuntimeException("RE2", e)
+                    }
+                }
+
+                launch {
+                    expect(3)
+                    throw RuntimeException("RE1")
+                }
+            }
+        } catch (e: Throwable) {
+            // Should not fail
+            ObjectOutputStream(ByteArrayOutputStream()).use {
+                it.writeObject(e)
+            }
+            finish(4)
+        }
+    }
+}