fix(ServerCalls): Ensure failure cause is propagated in Status to int…

…erceptors
grpc · Aug 22, 2023 · 075a4a6 · 075a4a6
1 parent e12115c
commit 075a4a6
Show file tree

Hide file tree

Showing 2 changed files with 57 additions and 4 deletions.
diff --git a/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt b/stub/src/main/java/io/grpc/kotlin/ServerCalls.kt
@@ -262,7 +262,7 @@ object ServerCalls {
       val closeStatus = when (failure) {
         null -> Status.OK
         is CancellationException -> Status.CANCELLED.withCause(failure)
-        else -> Status.fromThrowable(failure)
+        else -> Status.fromThrowable(failure).withCause(failure)
       }
       val trailers = failure?.let { Status.trailersFromThrowable(it) } ?: GrpcMetadata()
       mutex.withLock { call.close(closeStatus, trailers) }

diff --git a/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt b/stub/src/test/java/io/grpc/kotlin/ServerCallsTest.kt
@@ -20,26 +20,26 @@ 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
 import kotlinx.coroutines.FlowPreview
 import kotlinx.coroutines.Job
 import kotlinx.coroutines.asCoroutineDispatcher
 import kotlinx.coroutines.async
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.channels.trySendBlocking
 import kotlinx.coroutines.channels.toList
 import kotlinx.coroutines.delay
 import kotlinx.coroutines.flow.asFlow
 import kotlinx.coroutines.flow.buffer
 import kotlinx.coroutines.flow.channelFlow
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.consumeAsFlow
 import kotlinx.coroutines.flow.flow
 import kotlinx.coroutines.flow.flowOf
@@ -197,7 +197,10 @@ class ServerCallsTest : AbstractCallsTest() {
     call.request(1)
     call.sendMessage(helloRequest("Garnet"))
     call.halfClose()
-    assertThat(closeTrailers.await()[key]).isEqualTo("value")
+
+    val closedTrailers = closeTrailers.await()
+
+    assertThat(closedTrailers[key]).isEqualTo("value")
   }
 
   @Test
@@ -1010,4 +1013,54 @@ class ServerCallsTest : AbstractCallsTest() {
 
     return config
   }
+
+  @Test
+  fun testStatusExceptionPropagatesStack() = runBlocking {
+
+    val serverImpl = object : GreeterCoroutineImplBase() {
+      override suspend fun sayHello(request: HelloRequest): HelloReply {
+        internalServerCall()
+      }
+
+      private fun internalServerCall(): Nothing {
+        throw StatusException(Status.INTERNAL)
+      }
+    }
+
+    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).isNotNull()
+    assertThat(statusCause!!.stackTraceToString()).contains("internalServerCall")
+  }
+
 }