Repository: grpc/grpc-kotlin (kotlin)

Issue:
Status `cause` is replaced when it shouldn't be
After https://github.com/grpc/grpc-kotlin/pull/400, we're seeing a change in behavior in our interceptors when a call is closed due to a thrown StatusException or StatusRuntimeException. In particular, the status' cause is being replaced: whereas beforehand the cause was the _cause of the exception's status_, it is now the exception itself.

Supposing we have:

```kotlin
val exception = Exception("foo")

// service definition
override suspend fun sayHello(request: HelloRequest): HelloReply {
    throw Status.fromCode(code)
        .withCause(exception)
        .asException()
}
```

In our intercepted call's `close()` method we now get:

```kotlin
override fun close(status: Status, trailers: Metadata) {
     // the thrown exception from sayHello; in grpc-kotlin prior to #400 and in grpc-java this is `exception` itself
    status.cause

    // this is now how we have to get `exception`
    Status.fromThrowable(status.cause).cause 
}
```

This is because https://github.com/grpc/grpc-kotlin/blob/c449254cb1d3d2dcefbb5fb1a79cff5b3ebe1973/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt#L265 _replaces_ the `cause` of the status even when that status had been correctly inferred from a StatusException/StatusRuntimeException.

We noticed this in an interceptor we use for services using both grpc-kotlin and grpc-java service implementations. We can use some clever branching (`Status.fromThrowable(status.cause) == status` means we're in grpc-java or prior versions of grpc-kotlin) but I don't think this is really the best way to handle it; I don't think grpc-kotlin should replace a validly inferred cause from Status.fromThrowable.

I think the solution is to add two branches to the `when` expression in ServerCalls.kt - one each for StatusException and StatusRuntimeException that do not attach `failure` as the cause.  In service of the desired behavior of #400, _not_ replacing the cause for these cases should still preserve caller info in the stack trace since `failure` is still available in the `status` object. Does that sound reasonable?

Pull request description:
Ensures that Statuses derived from StatusExceptions and StatusRuntimeExceptions match their behavior before #400. This is also `grpc-java`'s behavior when `onError` is called with a StatusException or StatusRuntimeException.

Fixes https://github.com/grpc/grpc-kotlin/pull/455.

Patch:
diff --git a/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt b/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt
index 404990da..d2b532b4 100644
--- a/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt
+++ b/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt
@@ -26,13 +26,13 @@ import io.grpc.ServerCallHandler
 import io.grpc.ServerMethodDefinition
 import io.grpc.Status
 import io.grpc.StatusException
+import io.grpc.StatusRuntimeException
 import kotlinx.coroutines.CancellationException
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.onFailure
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.launch
@@ -40,7 +40,6 @@ import kotlinx.coroutines.sync.Mutex
 import kotlinx.coroutines.sync.withLock
 import java.util.concurrent.atomic.AtomicBoolean
 import kotlin.coroutines.CoroutineContext
-import kotlinx.coroutines.channels.onFailure
 import io.grpc.Metadata as GrpcMetadata
 
 /**
@@ -262,6 +261,7 @@ object ServerCalls {
       val closeStatus = when (failure) {
         null -> Status.OK
         is CancellationException -> Status.CANCELLED.withCause(failure)
+        is StatusException, is StatusRuntimeException -> Status.fromThrowable(failure)
         else -> Status.fromThrowable(failure).withCause(failure)
       }
       val trailers = failure?.let { Status.trailersFromThrowable(it) } ?: GrpcMetadata()


Test patch:
diff --git a/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt b/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt
index 4996c2aa..d47a4587 100644
--- a/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt
+++ b/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt
@@ -20,12 +20,10 @@ import com.google.common.truth.Truth.assertThat
 import com.google.common.truth.extensions.proto.ProtoTruth.assertThat
 import io.grpc.*
 import io.grpc.examples.helloworld.GreeterGrpc
-import io.grpc.examples.helloworld.GreeterGrpcKt
 import io.grpc.examples.helloworld.HelloReply
 import io.grpc.examples.helloworld.HelloRequest
 import io.grpc.examples.helloworld.GreeterGrpcKt.GreeterCoroutineStub
 import io.grpc.examples.helloworld.GreeterGrpcKt.GreeterCoroutineImplBase
-import io.grpc.stub.StreamObserver
 import kotlinx.coroutines.CompletableDeferred
 import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -1015,7 +1013,8 @@ class ServerCallsTest : AbstractCallsTest() {
   }
 
   @Test
-  fun testStatusExceptionPropagatesStack() = runBlocking {
+  fun testPropagateStackTraceForStatusException() = runBlocking {
+    val thrownStatusCause = CompletableDeferred<Throwable?>()
 
     val serverImpl = object : GreeterCoroutineImplBase() {
       override suspend fun sayHello(request: HelloRequest): HelloReply {
@@ -1023,7 +1022,9 @@ class ServerCallsTest : AbstractCallsTest() {
       }
 
       private fun internalServerCall(): Nothing {
-        throw StatusException(Status.INTERNAL)
+        val exception = Exception("causal exception")
+        thrownStatusCause.complete(exception)
+        throw Status.INTERNAL.withCause(exception).asException()
       }
     }
 
@@ -1059,7 +1060,111 @@ class ServerCallsTest : AbstractCallsTest() {
     assertThat(clientException.status.code).isEqualTo(Status.Code.INTERNAL)
     val statusCause = receivedStatusCause.await()
     // but the exception should propagate to server interceptors, with stack trace intact
-    assertThat(statusCause).isNotNull()
+    assertThat(statusCause).isEqualTo(thrownStatusCause.await())
+    assertThat(statusCause!!.stackTraceToString()).contains("internalServerCall")
+  }
+
+  @Test
+  fun testPropagateStackTraceForStatusRuntimeException() = runBlocking {
+    val thrownStatusCause = CompletableDeferred<Throwable?>()
+
+    val serverImpl = object : GreeterCoroutineImplBase() {
+      override suspend fun sayHello(request: HelloRequest): HelloReply {
+        internalServerCall()
+      }
+
+      private fun internalServerCall(): Nothing {
+        val exception = Exception("causal exception")
+        thrownStatusCause.complete(exception)
+        throw Status.INTERNAL.withCause(exception).asRuntimeException()
+      }
+    }
+
+    val receivedStatusCause = CompletableDeferred<Throwable?>()
+
+    val interceptor = object : ServerInterceptor {
+      override fun <ReqT, RespT> interceptCall(
+        call: ServerCall<ReqT, RespT>,
+        requestHeaders: Metadata,
+        next: ServerCallHandler<ReqT, RespT>
+      ): ServerCall.Listener<ReqT> =
+        next.startCall(
+          object : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
+            override fun close(status: Status, trailers: Metadata) {
+              receivedStatusCause.complete(status.cause)
+              super.close(status, trailers)
+            }
+          },
+          requestHeaders
+        )
+    }
+
+    val channel = makeChannel(serverImpl, interceptor)
+
+    val stub = GreeterGrpc.newBlockingStub(channel)
+    val clientException = assertThrows<StatusRuntimeException> {
+      stub.sayHello(helloRequest(""))
+    }
+
+    // the exception should not propagate to the client
+    assertThat(clientException.cause).isNull()
+
+    assertThat(clientException.status.code).isEqualTo(Status.Code.INTERNAL)
+    val statusCause = receivedStatusCause.await()
+    // but the exception should propagate to server interceptors, with stack trace intact
+    assertThat(statusCause).isEqualTo(thrownStatusCause.await())
+    assertThat(statusCause!!.stackTraceToString()).contains("internalServerCall")
+  }
+
+  @Test
+  fun testPropagateStackTraceForNonStatusException() = runBlocking {
+    val thrownStatusCause = CompletableDeferred<Throwable?>()
+
+    val serverImpl = object : GreeterCoroutineImplBase() {
+      override suspend fun sayHello(request: HelloRequest): HelloReply {
+        internalServerCall()
+      }
+
+      private fun internalServerCall(): Nothing {
+        val exception = Exception("causal exception")
+        thrownStatusCause.complete(exception)
+        throw exception
+      }
+    }
+
+    val receivedStatusCause = CompletableDeferred<Throwable?>()
+
+    val interceptor = object : ServerInterceptor {
+      override fun <ReqT, RespT> interceptCall(
+        call: ServerCall<ReqT, RespT>,
+        requestHeaders: Metadata,
+        next: ServerCallHandler<ReqT, RespT>
+      ): ServerCall.Listener<ReqT> =
+        next.startCall(
+          object : ForwardingServerCall.SimpleForwardingServerCall<ReqT, RespT>(call) {
+            override fun close(status: Status, trailers: Metadata) {
+              receivedStatusCause.complete(status.cause)
+              super.close(status, trailers)
+            }
+          },
+          requestHeaders
+        )
+    }
+
+    val channel = makeChannel(serverImpl, interceptor)
+
+    val stub = GreeterGrpc.newBlockingStub(channel)
+    val clientException = assertThrows<StatusRuntimeException> {
+      stub.sayHello(helloRequest(""))
+    }
+
+    // the exception should not propagate to the client
+    assertThat(clientException.cause).isNull()
+
+    assertThat(clientException.status.code).isEqualTo(Status.Code.UNKNOWN)
+    val statusCause = receivedStatusCause.await()
+    // but the exception should propagate to server interceptors, with stack trace intact
+    assertThat(statusCause).isEqualTo(thrownStatusCause.await())
     assertThat(statusCause!!.stackTraceToString()).contains("internalServerCall")
   }