Fix pipeline stack trace

ktor-utils/common/src/io/ktor/util/pipeline/PipelineContext.kt

@@ -103,7 +103,7 @@ private class SuspendFunctionGun<TSubject : Any, TContext : Any>(
 
         @Suppress("UNCHECKED_CAST")
         override val context: CoroutineContext
-            get () {
+            get() {
                 val cont = rootContinuation
                 return when (cont) {
                     null -> throw IllegalStateException("Not started")
@@ -212,7 +212,12 @@ private class SuspendFunctionGun<TSubject : Any, TContext : Any>(
             else -> unexpectedRootContinuationValue(rootContinuation)
         } as Continuation<TSubject>
 
-        next.resumeWith(result)
+        if (!result.isFailure) {
+            next.resumeWith(result)
+        } else {
+            val exception = recoverStackTraceBridge(result.exceptionOrNull()!!, next)
+            next.resumeWithException(exception)
+        }
     }
 
     private fun discardLastRootContinuation() {

ktor-utils/common/src/io/ktor/util/pipeline/StackTraceRecover.kt

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.util.pipeline
+
+import kotlin.coroutines.*
+import kotlinx.coroutines.internal.*
+
+
+/**
+ * Recreates the exception with the original cause to keep exception structure.
+ *
+ * Notice: This method breaks the [exception] identity.
+ */
+internal fun recoverStackTraceBridge(exception: Throwable, continuation: Continuation<*>): Throwable = try {
+    @Suppress("INVISIBLE_MEMBER")
+    recoverStackTrace(exception, continuation).withCause(exception.cause)
+} catch (_: Throwable) {
+    exception
+}
+
+internal expect fun Throwable.withCause(cause: Throwable?): Throwable

ktor-utils/js/src/io/ktor/util/pipeline/StackTraceRecoverJs.kt

@@ -0,0 +1,7 @@
+/*
+ * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.util.pipeline
+
+internal actual fun Throwable.withCause(cause: Throwable?): Throwable = this

ktor-utils/jvm/src/io/ktor/util/pipeline/ExceptionConstructor.kt

@@ -0,0 +1,87 @@
+/*
+ * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+package io.ktor.util.pipeline
+
+import kotlinx.coroutines.*
+import java.lang.reflect.*
+import java.util.*
+import java.util.concurrent.locks.*
+import kotlin.concurrent.*
+
+/**
+ * This file is a copy of the [kotlinx.coroutines.internal.ExceptionConstructor] with a single difference:
+ * [tryCopyException] takes additional argument with cause to use in recovered exception.
+ */
+
+private val throwableFields = Throwable::class.java.fieldsCountOrDefault(-1)
+private val cacheLock = ReentrantReadWriteLock()
+private typealias Ctor = (Throwable) -> Throwable?
+
+// Replace it with ClassValue when Java 6 support is over
+private val exceptionCtors: WeakHashMap<Class<out Throwable>, Ctor> = WeakHashMap()
+
+@Suppress("UNCHECKED_CAST")
+internal fun <E : Throwable> tryCopyException(exception: E, cause: Throwable): E? {
+    // Fast path for CopyableThrowable
+    if (exception is CopyableThrowable<*>) {
+        return runCatching { exception.createCopy() as E? }.getOrNull()
+    }
+    // Use cached ctor if found
+    cacheLock.read { exceptionCtors[exception.javaClass] }?.let { cachedCtor ->
+        return cachedCtor(exception) as E?
+    }
+    /*
+     * Skip reflective copy if an exception has additional fields (that are usually populated in user-defined constructors)
+     */
+    if (throwableFields != exception.javaClass.fieldsCountOrDefault(0)) {
+        cacheLock.write { exceptionCtors[exception.javaClass] = { null } }
+        return null
+    }
+    /*
+     * Try to reflectively find constructor(), constructor(message, cause), constructor(cause) or constructor(message).
+     * Exceptions are shared among coroutines, so we should copy exception before recovering current stacktrace.
+     */
+    var ctor: Ctor? = null
+    val constructors = exception.javaClass.constructors.sortedByDescending { it.parameterTypes.size }
+    for (constructor in constructors) {
+        ctor = createConstructor(constructor)
+        if (ctor != null) break
+    }
+    // Store the resulting ctor to cache
+    cacheLock.write { exceptionCtors[exception.javaClass] = ctor ?: { null } }
+    return ctor?.invoke(cause) as E?
+}
+
+private fun createConstructor(constructor: Constructor<*>): Ctor? {
+    val p = constructor.parameterTypes
+    return when (p.size) {
+        2 -> when {
+            p[0] == String::class.java && p[1] == Throwable::class.java ->
+                safeCtor { e -> constructor.newInstance(e.message, e) as Throwable }
+            else -> null
+        }
+        1 -> when (p[0]) {
+            Throwable::class.java ->
+                safeCtor { e -> constructor.newInstance(e) as Throwable }
+            String::class.java ->
+                safeCtor { e -> (constructor.newInstance(e.message) as Throwable).also { it.initCause(e) } }
+            else -> null
+        }
+        0 -> safeCtor { e -> (constructor.newInstance() as Throwable).also { it.initCause(e) } }
+        else -> null
+    }
+}
+
+private inline fun safeCtor(crossinline block: (Throwable) -> Throwable): Ctor =
+    { e -> runCatching { block(e) }.getOrNull() }
+
+private fun Class<*>.fieldsCountOrDefault(defaultValue: Int) =
+    kotlin.runCatching { fieldsCount() }.getOrDefault(defaultValue)
+
+private tailrec fun Class<*>.fieldsCount(accumulator: Int = 0): Int {
+    val fieldsCount = declaredFields.count { !Modifier.isStatic(it.modifiers) }
+    val totalFields = accumulator + fieldsCount
+    val superClass = superclass ?: return totalFields
+    return superClass.fieldsCount(totalFields)
+}

ktor-utils/jvm/src/io/ktor/util/pipeline/StackTraceRecoverJvm.kt

@@ -0,0 +1,16 @@
+/*
+ * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.util.pipeline
+
+internal actual fun Throwable.withCause(cause: Throwable?): Throwable {
+    if (cause == null || this.cause == cause) {
+        return this
+    }
+
+    val result = tryCopyException(this, cause) ?: return this
+    result.stackTrace = stackTrace
+
+    return result
+}

ktor-utils/jvm/test/io/ktor/tests/utils/PipelineTest.kt

@@ -111,7 +111,7 @@ class PipelineTest {
                 p2.execute(Unit, "p2")
                 proceed()
                 events.add("success-p1-1 $subject")
-            } catch(t: Throwable) {
+            } catch (t: Throwable) {
                 events.add("fail-p1-1 $subject")
                 throw t
             }
@@ -123,7 +123,8 @@ class PipelineTest {
         }
 
         p1.executeBlocking("p1")
-        assertEquals(listOf(
+        assertEquals(
+            listOf(
                 "intercept-p1-1 p1",
                 "intercept-p2-1 p2",
                 "intercept-p3-1 p3",
@@ -131,7 +132,8 @@ class PipelineTest {
                 "success-p2-1 p2",
                 "intercept-p1-2 p1",
                 "success-p1-1 p1"
-        ), events)
+            ), events
+        )
     }
 
     @Test
@@ -278,8 +280,12 @@ class PipelineTest {
         assertFailsWith<UnsupportedOperationException> {
             pipeline.executeBlocking("some")
         }
-        assertEquals(listOf("intercept1 some", "intercept2 some", "intercept3 another",
-                "intercept4 some", "fail1 some"), events)
+        assertEquals(
+            listOf(
+                "intercept1 some", "intercept2 some", "intercept3 another",
+                "intercept4 some", "fail1 some"
+            ), events
+        )
     }
 
     @Test
@@ -367,7 +373,11 @@ class PipelineTest {
         assertEquals(listOf("intercept1 some", "future1 some", "intercept2 another", "success1 some"), events)
     }
 
-    private fun checkBeforeAfterPipeline(after: PipelinePhase, before: PipelinePhase, pipeline: Pipeline<String, Unit>) {
+    private fun checkBeforeAfterPipeline(
+        after: PipelinePhase,
+        before: PipelinePhase,
+        pipeline: Pipeline<String, Unit>
+    ) {
         var value = false
         pipeline.intercept(after) {
             value = true
@@ -418,4 +428,41 @@ class PipelineTest {
         pipeline.insertPhaseAfter(before, after)
         checkBeforeAfterPipeline(after, before, pipeline)
     }
+
+    @Test
+    fun testStackTraceWithMultipleInterceptors() {
+        val pipeline = pipeline()
+
+        pipeline.intercept(::interceptor1)
+        pipeline.intercept(::interceptor2)
+        pipeline.intercept(::interceptor3)
+
+        try {
+            pipeline.executeBlocking("start")
+        } catch (cause: Throwable) {
+            val stackTrace = cause.stackTrace
+            assertEquals(6, stackTrace.size)
+
+            assertTrue("interceptor3" in stackTrace[0].toString())
+            assertTrue("interceptor3" in stackTrace[1].toString())
+            assertEquals("\b\b\b(Coroutine boundary.\b(\b)", stackTrace[2].toString())
+            assertTrue("interceptor2" in stackTrace[3].toString())
+            assertTrue("interceptor1" in stackTrace[4].toString())
+        }
+    }
+}
+
+private suspend fun interceptor1(context: PipelineContext<String, Unit>, content: String) {
+    yield()
+    context.proceedWith("$content first")
+}
+
+private suspend fun interceptor2(context: PipelineContext<String, Unit>, content: String) {
+    yield()
+    context.proceedWith("$content second")
+}
+
+private suspend fun interceptor3(context: PipelineContext<String, Unit>, content: String) {
+    yield()
+    error(content)
 }

ktor-utils/posix/src/io/ktor/util/pipeline/StackTraceRecoverNative.kt

@@ -0,0 +1,7 @@
+/*
+ * Copyright 2014-2019 JetBrains s.r.o and contributors. Use of this source code is governed by the Apache 2.0 license.
+ */
+
+package io.ktor.util.pipeline
+
+internal actual fun Throwable.withCause(cause: Throwable?): Throwable = this