Merge pull request #137 from moleike/add-status-codes

Add gRPC status codes

codegen/testing/src/test/scala/hello/world/TestServiceSuite.scala

@@ -6,6 +6,7 @@ import fs2.Stream
 import munit._
 import org.http4s._
 import org.http4s.client.Client
+import org.http4s.grpc.GrpcStatus._
 import org.http4s.syntax.all._
 import org.scalacheck._
 import org.scalacheck.effect.PropF.forAllF
@@ -127,7 +128,7 @@ class TestServiceSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
         status.pure[IO]
       }
       .assertEquals(
-        Some(org.http4s.grpc.codecs.NamedHeaders.GrpcStatus(12))
+        Some(org.http4s.grpc.codecs.NamedHeaders.GrpcStatus(Unimplemented))
       )
   }
 
@@ -136,7 +137,7 @@ class TestServiceSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
       val route = org.http4s.HttpRoutes.of[IO] { case _ =>
         Response(Status.Ok)
           .putHeaders(
-            org.http4s.grpc.codecs.NamedHeaders.GrpcStatus(12)
+            org.http4s.grpc.codecs.NamedHeaders.GrpcStatus(Unimplemented)
           )
           .pure[IO]
       }
@@ -148,7 +149,7 @@ class TestServiceSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
         .`export`(msg, Headers.empty)
         .attemptNarrow[org.http4s.grpc.GrpcExceptions.StatusRuntimeException]
         .map(_.leftMap(grpcFailed => grpcFailed.status))
-        .assertEquals(Either.left(12))
+        .assertEquals(Either.left(Unimplemented))
     }
   }
 
@@ -179,7 +180,43 @@ class TestServiceSuite extends CatsEffectSuite with ScalaCheckEffectSuite {
         .noStreaming(msg, Headers.empty)
         .attemptNarrow[org.http4s.grpc.GrpcExceptions.StatusRuntimeException]
         .map(_.leftMap(grpcFailed => grpcFailed.status))
-        .assertEquals(Either.left(2))
+        .assertEquals(Either.left(Unknown))
+    }
+
+  }
+
+  test("Server Fails with an Status Code") {
+
+    implicit val arbitraryStatusCode: Arbitrary[Code] = Arbitrary(
+      Gen.oneOf(codeValues.filter(_ != Ok))
+    )
+
+    forAllF { (msg: TestMessage, statusCode: Code) =>
+      val ts = new TestService[IO] {
+        def noStreaming(request: TestMessage, ctx: Headers): IO[TestMessage] =
+          IO(request) <* IO.raiseError(statusCode.asStatusRuntimeException())
+
+        def clientStreaming(request: Stream[IO, TestMessage], ctx: Headers): IO[TestMessage] =
+          request.compile.lastOrError
+
+        def serverStreaming(request: TestMessage, ctx: Headers): Stream[IO, TestMessage] =
+          Stream.emit(request)
+
+        def bothStreaming(request: Stream[IO, TestMessage], ctx: Headers): Stream[IO, TestMessage] =
+          request
+
+        def `export`(request: TestMessage, ctx: Headers): IO[TestMessage] = IO(request)
+      }
+      val client = TestService.fromClient[IO](
+        Client.fromHttpApp(TestService.toRoutes[IO](ts).orNotFound),
+        Uri(),
+      )
+
+      client
+        .noStreaming(msg, Headers.empty)
+        .attemptNarrow[org.http4s.grpc.GrpcExceptions.StatusRuntimeException]
+        .map(_.leftMap(grpcFailed => grpcFailed.status))
+        .assertEquals(Either.left(statusCode))
     }
 
   }

core/src/main/scala/org/http4s/grpc/ClientGrpc.scala

@@ -6,6 +6,7 @@ import cats.syntax.all._
 import fs2._
 import org.http4s._
 import org.http4s.client.Client
+import org.http4s.grpc.GrpcStatus._
 import org.http4s.grpc.codecs.NamedHeaders
 import org.http4s.h2.H2Keys
 import scodec.Decoder
@@ -177,7 +178,7 @@ object ClientGrpc {
     val reason = headers.get[NamedHeaders.GrpcMessage]
 
     status match {
-      case Some(NamedHeaders.GrpcStatus(0)) => ().pure[F]
+      case Some(NamedHeaders.GrpcStatus(Ok)) => ().pure[F]
       case Some(NamedHeaders.GrpcStatus(status)) =>
         GrpcExceptions.StatusRuntimeException(status, reason.map(_.message)).raiseError[F, Unit]
       case None => ().pure[F]

core/src/main/scala/org/http4s/grpc/GrpcExceptions.scala

@@ -1,9 +1,11 @@
 package org.http4s.grpc
 
 object GrpcExceptions {
-  final case class StatusRuntimeException(status: Int, message: Option[String])
+  final case class StatusRuntimeException(status: GrpcStatus.Code, message: Option[String])
       extends RuntimeException({
         val me = message.fold("")((m: String) => s", Message-${m}")
-        s"Grpc Failed: Status-$status${me}"
-      })
+        s"Grpc Failed: Status-${status.value}${me}"
+      }) {
+    assert(status != GrpcStatus.Ok)
+  }
 }

core/src/main/scala/org/http4s/grpc/GrpcStatus.scala

@@ -0,0 +1,68 @@
+package org.http4s.grpc
+
+import GrpcExceptions.StatusRuntimeException
+
+object GrpcStatus {
+
+  sealed abstract class Code(val value: Int) extends Product with Serializable {
+    def asStatusRuntimeException(message: Option[String] = None): StatusRuntimeException =
+      StatusRuntimeException(this, message)
+  }
+
+  case object Ok extends Code(0)
+
+  case object Cancelled extends Code(1)
+
+  case object Unknown extends Code(2)
+
+  case object InvalidArgument extends Code(3)
+
+  case object DeadlineExceeded extends Code(4)
+
+  case object NotFound extends Code(5)
+
+  case object AlreadyExists extends Code(6)
+
+  case object PermissionDenied extends Code(7)
+
+  case object ResourceExhausted extends Code(8)
+
+  case object FailedPrecondition extends Code(9)
+
+  case object Aborted extends Code(10)
+
+  case object OutOfRange extends Code(11)
+
+  case object Unimplemented extends Code(12)
+
+  case object Internal extends Code(13)
+
+  case object Unavailable extends Code(14)
+
+  case object DataLoss extends Code(15)
+
+  case object Unauthenticated extends Code(16)
+
+  def fromCodeValue(value: Int): Option[Code] = codeValues.find(_.value == value)
+
+  val codeValues: List[Code] = List(
+    Ok,
+    Cancelled,
+    Unknown,
+    InvalidArgument,
+    DeadlineExceeded,
+    NotFound,
+    AlreadyExists,
+    PermissionDenied,
+    ResourceExhausted,
+    FailedPrecondition,
+    Aborted,
+    OutOfRange,
+    Unimplemented,
+    Internal,
+    Unavailable,
+    DataLoss,
+    Unauthenticated,
+  )
+
+}

core/src/main/scala/org/http4s/grpc/ServerGrpc.scala

@@ -6,6 +6,8 @@ import cats.syntax.all._
 import fs2._
 import org.http4s._
 import org.http4s.dsl.request._
+import org.http4s.grpc.GrpcExceptions.StatusRuntimeException
+import org.http4s.grpc.GrpcStatus._
 import org.http4s.grpc.codecs.NamedHeaders
 import org.http4s.headers.Allow
 import org.http4s.headers.Trailer
@@ -38,7 +40,7 @@ object ServerGrpc {
   ): HttpRoutes[F] = HttpRoutes.of[F] {
     case req @ POST -> Root / sN / mN if sN === serviceName && mN === methodName =>
       for {
-        status <- Deferred[F, (Int, Option[String])]
+        status <- Deferred[F, (Code, Option[String])]
         trailers = status.get.map { case (i, message) =>
           Headers(
             NamedHeaders.GrpcStatus(i)
@@ -51,12 +53,7 @@ object ServerGrpc {
           .evalMap(f(_, req.headers))
           .flatMap(codecs.Messages.encodeSingle(encode)(_))
           .through(timeoutStream(_)(timeout.map(_.duration)))
-          .onFinalizeCaseWeak {
-            case Resource.ExitCase.Errored(_: TimeoutException) => status.complete((4, None)).void
-            case Resource.ExitCase.Errored(e) => status.complete((2, e.toString().some)).void
-            case Resource.ExitCase.Canceled => status.complete((1, None)).void
-            case Resource.ExitCase.Succeeded => status.complete((0, None)).void
-          }
+          .onFinalizeCaseWeak(updateStatus(status))
           .mask // ensures body closure without rst-stream
 
         Response[F](Status.Ok, HttpVersion.`HTTP/2`)
@@ -81,7 +78,7 @@ object ServerGrpc {
   ): HttpRoutes[F] = HttpRoutes.of[F] {
     case req @ POST -> Root / sN / mN if sN === serviceName && mN === methodName =>
       for {
-        status <- Deferred[F, (Int, Option[String])]
+        status <- Deferred[F, (Code, Option[String])]
         trailers = status.get.map { case (i, message) =>
           Headers(
             NamedHeaders.GrpcStatus(i)
@@ -94,12 +91,7 @@ object ServerGrpc {
           .flatMap(f(_, req.headers))
           .through(codecs.Messages.encode(encode))
           .through(timeoutStream(_)(timeout.map(_.duration)))
-          .onFinalizeCaseWeak {
-            case Resource.ExitCase.Errored(_: TimeoutException) => status.complete((4, None)).void
-            case Resource.ExitCase.Errored(e) => status.complete((2, e.toString().some)).void
-            case Resource.ExitCase.Canceled => status.complete((1, None)).void
-            case Resource.ExitCase.Succeeded => status.complete((0, None)).void
-          }
+          .onFinalizeCaseWeak(updateStatus(status))
           .mask // ensures body closure without rst-stream
         Response[F](Status.Ok, HttpVersion.`HTTP/2`)
           .putHeaders(
@@ -123,7 +115,7 @@ object ServerGrpc {
   ): HttpRoutes[F] = HttpRoutes.of[F] {
     case req @ POST -> Root / sN / mN if sN === serviceName && mN === methodName =>
       for {
-        status <- Deferred[F, (Int, Option[String])]
+        status <- Deferred[F, (Code, Option[String])]
         trailers = status.get.map { case (i, message) =>
           Headers(
             NamedHeaders.GrpcStatus(i)
@@ -136,12 +128,7 @@ object ServerGrpc {
           .eval(f(codecs.Messages.decode(decode)(req.body), req.headers))
           .flatMap(codecs.Messages.encodeSingle(encode)(_))
           .through(timeoutStream(_)(timeout.map(_.duration)))
-          .onFinalizeCaseWeak {
-            case Resource.ExitCase.Errored(_: TimeoutException) => status.complete((4, None)).void
-            case Resource.ExitCase.Errored(e) => status.complete((2, e.toString().some)).void
-            case Resource.ExitCase.Canceled => status.complete((1, None)).void
-            case Resource.ExitCase.Succeeded => status.complete((0, None)).void
-          }
+          .onFinalizeCaseWeak(updateStatus(status))
           .mask // ensures body closure without rst-stream
 
         Response[F](Status.Ok, HttpVersion.`HTTP/2`)
@@ -166,7 +153,7 @@ object ServerGrpc {
   ): HttpRoutes[F] = HttpRoutes.of[F] {
     case req @ POST -> Root / sN / mN if sN === serviceName && mN === methodName =>
       for {
-        status <- Deferred[F, (Int, Option[String])]
+        status <- Deferred[F, (Code, Option[String])]
         trailers = status.get.map { case (i, message) =>
           Headers(
             NamedHeaders.GrpcStatus(i)
@@ -178,12 +165,7 @@ object ServerGrpc {
         val body = f(codecs.Messages.decode(decode)(req.body), req.headers)
           .through(codecs.Messages.encode(encode))
           .through(timeoutStream(_)(timeout.map(_.duration)))
-          .onFinalizeCaseWeak {
-            case Resource.ExitCase.Errored(_: TimeoutException) => status.complete((4, None)).void
-            case Resource.ExitCase.Errored(e) => status.complete((2, e.toString().some)).void
-            case Resource.ExitCase.Canceled => status.complete((1, None)).void
-            case Resource.ExitCase.Succeeded => status.complete((0, None)).void
-          }
+          .onFinalizeCaseWeak(updateStatus(status))
           .mask // ensures body closure without rst-stream
 
         Response[F](Status.Ok, HttpVersion.`HTTP/2`)
@@ -204,7 +186,7 @@ object ServerGrpc {
         .putHeaders(
           SharedGrpc.ContentType,
           SharedGrpc.TE,
-          NamedHeaders.GrpcStatus(12),
+          NamedHeaders.GrpcStatus(Unimplemented),
           "grpc-message" -> s"unknown method $mN for service $sN",
         )
         .pure[F]
@@ -216,7 +198,7 @@ object ServerGrpc {
         .putHeaders(
           SharedGrpc.ContentType,
           SharedGrpc.TE,
-          NamedHeaders.GrpcStatus(12),
+          NamedHeaders.GrpcStatus(Unimplemented),
           "grpc-message" -> s"unknown service $sN",
         )
         .pure[F]
@@ -225,7 +207,7 @@ object ServerGrpc {
         .putHeaders(
           SharedGrpc.ContentType,
           SharedGrpc.TE,
-          NamedHeaders.GrpcStatus(12),
+          NamedHeaders.GrpcStatus(Unimplemented),
           "grpc-message" -> s"unknown method $other",
         )
         .pure[F]
@@ -234,7 +216,7 @@ object ServerGrpc {
         .putHeaders(
           SharedGrpc.ContentType,
           SharedGrpc.TE,
-          NamedHeaders.GrpcStatus(12),
+          NamedHeaders.GrpcStatus(Unimplemented),
           "grpc-message" -> s"unknown request",
         )
         .pure[F]
@@ -248,4 +230,14 @@ object ServerGrpc {
       case Some(value) => s.timeout(value)
     }
 
+  private def updateStatus[F[_]: Concurrent](
+      status: Deferred[F, (Code, Option[String])]
+  ): Resource.ExitCase => F[Unit] = {
+    case Resource.ExitCase.Errored(StatusRuntimeException(c, m)) => status.complete((c, m)).void
+    case Resource.ExitCase.Errored(_: TimeoutException) =>
+      status.complete((DeadlineExceeded, None)).void
+    case Resource.ExitCase.Errored(e) => status.complete((Unknown, e.toString().some)).void
+    case Resource.ExitCase.Canceled => status.complete((Cancelled, None)).void
+    case Resource.ExitCase.Succeeded => status.complete((Ok, None)).void
+  }
 }

core/src/main/scala/org/http4s/grpc/codecs/NamedHeaders.scala

@@ -4,6 +4,8 @@ import cats.parse.Parser
 import cats.syntax.all._
 import org.http4s.Header
 import org.http4s.ParseResult
+import org.http4s.grpc.GrpcStatus.Code
+import org.http4s.grpc.GrpcStatus.fromCodeValue
 import org.http4s.internal.parsing.CommonRules.ows
 import org.http4s.parser.AdditionalRules
 import org.typelevel.ci.CIString
@@ -55,14 +57,16 @@ object NamedHeaders {
   }
 
   // https://grpc.github.io/grpc/core/md_doc_statuscodes.html
-  final case class GrpcStatus(statusCode: Int)
+  final case class GrpcStatus(statusCode: Code)
 
   object GrpcStatus {
-    private val parser = cats.parse.Numbers.nonNegativeIntString.map(s => GrpcStatus(s.toInt))
+    private val parser = cats.parse.Numbers.nonNegativeIntString
+      .mapFilter(s => fromCodeValue(s.toInt))
+      .map(GrpcStatus(_))
 
     implicit val header: Header[GrpcStatus, Header.Single] = Header.create(
       CIString("grpc-status"),
-      (t: GrpcStatus) => t.statusCode.toString(),
+      (t: GrpcStatus) => t.statusCode.value.toString(),
       (s: String) => ParseResult.fromParser(parser, "Invalid GrpcStatus")(s),
     )
   }