Merge pull request http4s#1784 from rossabaker/issue-1783

Tighten up logging of errors thrown by services

Author: rossabaker

blaze-server/src/main/scala/org/http4s/server/blaze/Http1ServerStage.scala

@@ -2,7 +2,8 @@ package org.http4s
 package server
 package blaze
 
-import cats.effect.{Effect, IO}
+import cats.data.OptionT
+import cats.effect.{Effect, IO, Sync}
 import cats.implicits._
 import fs2._
 import java.nio.ByteBuffer
@@ -60,6 +61,7 @@ private[blaze] class Http1ServerStage[F[_]](
 
   // micro-optimization: unwrap the service and call its .run directly
   private[this] val serviceFn = service.run
+  private[this] val optionTSync = Sync[OptionT[F, ?]]
 
   // both `parser` and `isClosed` are protected by synchronization on `parser`
   private[this] val parser = new Http1ServerParser[F](logger, maxRequestLineLen, maxHeadersLen)
@@ -140,19 +142,21 @@ private[blaze] class Http1ServerStage[F[_]](
     parser.collectMessage(body, requestAttrs) match {
       case Right(req) =>
         executionContext.execute(new Runnable {
-          def run(): Unit =
-            F.runAsync {
-                try serviceFn(req)
-                  .getOrElse(Response.notFound)
-                  .handleErrorWith(serviceErrorHandler(req))
-                catch serviceErrorHandler(req)
-              } {
-                case Right(resp) =>
-                  IO(renderResponse(req, resp, cleanup))
+          def run(): Unit = {
+            val action = optionTSync
+              .suspend(serviceFn(req))
+              .getOrElse(Response.notFound)
+              .recoverWith(serviceErrorHandler(req))
+              .flatMap(resp => F.delay(renderResponse(req, resp, cleanup)))
+
+            F.runAsync(action) {
+                case Right(()) => IO.unit
                 case Left(t) =>
-                  IO(internalServerError(s"Error running route: $req", t, req, cleanup))
+                  IO(logger.error(t)(s"Error running request: $req")).attempt *> IO(
+                    closeConnection())
               }
               .unsafeRunSync()
+          }
         })
       case Left((e, protocol)) =>
         badMessage(e.details, new BadRequest(e.sanitized), Request[F]().withHttpVersion(protocol))

blaze-server/src/main/scala/org/http4s/server/blaze/Http2NodeStage.scala

@@ -2,14 +2,14 @@ package org.http4s
 package server
 package blaze
 
-import cats.effect.{Effect, IO}
+import cats.data.OptionT
+import cats.effect.{Effect, IO, Sync}
 import cats.implicits._
 import fs2._
 import fs2.Stream._
 import java.util.Locale
 import org.http4s.{Headers => HHeaders, Method => HMethod}
 import org.http4s.Header.Raw
-import org.http4s.Status._
 import org.http4s.blaze.http.Headers
 import org.http4s.blaze.http.http20.{Http2StageTools, NodeMsg}
 import org.http4s.blaze.http.http20.Http2Exception._
@@ -33,6 +33,10 @@ private class Http2NodeStage[F[_]](
   import Http2StageTools._
   import NodeMsg.{DataFrame, HeadersFrame}
 
+  // micro-optimization: unwrap the service and call its .run directly
+  private[this] val serviceFn = service.run
+  private[this] val optionTSync = Sync[OptionT[F, ?]]
+
   override def name = "Http2NodeStage"
 
   override protected def stageStartup(): Unit = {
@@ -185,21 +189,20 @@ private class Http2NodeStage[F[_]](
       val hs = HHeaders(headers.result())
       val req = Request(method, path, HttpVersion.`HTTP/2.0`, hs, body, attributes)
       executionContext.execute(new Runnable {
-        def run(): Unit =
-          F.runAsync {
-              try service(req)
-                .getOrElse(Response.notFound)
-                .recoverWith(serviceErrorHandler(req))
-                .handleError(_ => Response(InternalServerError, req.httpVersion))
-                .map(renderResponse)
-              catch serviceErrorHandler(req)
-            } {
-              case Right(_) =>
-                IO.unit
-              case Left(t) =>
-                IO(logger.error(t)("Error rendering response"))
-            }
-            .unsafeRunSync()
+        def run(): Unit = {
+          val action = optionTSync
+            .suspend(serviceFn(req))
+            .getOrElse(Response.notFound)
+            .recoverWith(serviceErrorHandler(req))
+            .flatMap(resp => renderResponse(resp))
+
+          F.runAsync(action) {
+            case Right(()) => IO.unit
+            case Left(t) =>
+              IO(logger.error(t)(s"Error running request: $req")).attempt *> IO(
+                shutdownWithCommand(Cmd.Disconnect))
+          }
+        }.unsafeRunSync()
       })
     }
   }

server/src/main/scala/org/http4s/server/package.scala

@@ -7,7 +7,6 @@ import cats.implicits._
 import org.http4s.headers.{Connection, `Content-Length`}
 import org.http4s.syntax.string._
 import org.log4s.getLogger
-import scala.util.control.NonFatal
 
 package object server {
 
@@ -101,7 +100,7 @@ package object server {
         s"""Message failure handling request: ${req.method} ${req.pathInfo} from ${req.remoteAddr
           .getOrElse("<unknown>")}""")
       mf.toHttpResponse(req.httpVersion)
-    case NonFatal(t) =>
+    case t if !t.isInstanceOf[VirtualMachineError] =>
       serviceErrorLogger.error(t)(
         s"""Error servicing request: ${req.method} ${req.pathInfo} from ${req.remoteAddr.getOrElse(
           "<unknown>")}""")

servlet/src/main/scala/org/http4s/servlet/Http4sServlet.scala

@@ -1,6 +1,7 @@
 package org.http4s
 package servlet
 
+import cats.data.OptionT
 import cats.effect._
 import cats.implicits.{catsSyntaxEither => _, _}
 import fs2.async
@@ -29,6 +30,7 @@ class Http4sServlet[F[_]](
 
   // micro-optimization: unwrap the service and call its .run directly
   private[this] val serviceFn = service.run
+  private[this] val optionTSync = Sync[OptionT[F, ?]]
 
   object ServletRequestKeys {
     val HttpSession: AttributeKey[Option[HttpSession]] = AttributeKey[Option[HttpSession]]
@@ -97,10 +99,11 @@ class Http4sServlet[F[_]](
       bodyWriter: BodyWriter[F]): F[Unit] = {
     ctx.addListener(new AsyncTimeoutHandler(request, bodyWriter))
     // Note: We're catching silly user errors in the lift => flatten.
-    val response = Async.shift(executionContext) *> F
-      .delay(serviceFn(request).getOrElse(Response.notFound))
-      .flatten
-      .handleErrorWith(serviceErrorHandler(request))
+    val response = Async.shift(executionContext) *>
+      optionTSync
+        .suspend(serviceFn(request))
+        .getOrElse(Response.notFound)
+        .recoverWith(serviceErrorHandler(request))
 
     val servletResponse = ctx.getResponse.asInstanceOf[HttpServletResponse]
     renderResponse(response, servletResponse, bodyWriter)
@@ -122,9 +125,9 @@ class Http4sServlet[F[_]](
             s"Async context timed out, but response was already committed: ${request.method} ${request.uri.path}")
           F.pure(())
         }
-      } { _ =>
-        ctx.complete()
-        IO.unit
+      } {
+        case Right(()) => IO(ctx.complete())
+        case Left(t) => IO(logger.error(t)("Error timing out async context")) *> IO(ctx.complete())
       }
     }
   }