Netty: respond with 408 when the request timeout fires before the body is fully received

doc/server/netty.md

@@ -100,7 +100,7 @@ Tapir's endpoints) are added to a Netty server.
 
 `NettyConfig` exposes a number of configuration options which allows to
 customise the server socket, such as:
-* request timeout
+* request timeout, see [request timeout](#request-timeout) below
 * connection timeout
 * linger timeout
 * graceful shutdown timeout: when stopped e.g. using
@@ -121,6 +121,15 @@ import scala.concurrent.duration.*
 val config = NettyConfig.default.requestTimeout(5.seconds)
 ```
 
+### Request timeout
+
+Spans from receiving the request headers to starting to write a response, so it
+also bounds how long the client has to send the body: keep it higher than your
+longest upload, and lower than `idleTimeout`. When exceeded, an empty response
+with `Connection: close` is sent and the connection is closed - `503` if the
+request had been received in full, `408` if the body was still incomplete.
+Ignored for Web Sockets, once the handshake has been established.
+
 ## Web sockets
 
 ### tapir-netty-server-sync

server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala

@@ -30,8 +30,10 @@ import scala.concurrent.duration._
   *   contains tapir's server processing logic.
   *
   * @param requestTimeout
-  *   The maximum duration to wait for a response to be produced. If exceeded, the server will return a HTTP 503 response and close the
-  *   connection. This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`.
+  *   The maximum duration between receiving the request headers and producing a response; it therefore also bounds how long the client has
+  *   to send the body. If exceeded, an empty response with `Connection: close` is sent and the connection is closed: `503` if the request
+  *   was fully received, `408` if the body was still incomplete. Ignored in Web Sockets (after a handshake is established). Make sure it's
+  *   lower than `idleTimeout`.
   *
   * @param connectionTimeout
   *   Specifies the maximum duration within which a connection between a client and a server must be established.

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala

@@ -68,8 +68,17 @@ class NettyServerHandler[F[_]](
   // if the connection gets closed.
   private[this] val pendingResponses = MutableQueue.empty[() => Future[Unit]]
 
+  // `IdleStateHandler` re-fires `WRITER_IDLE` every `requestTimeout` until a write completes; only the first firing is answered, as the
+  // connection is closed along with the response. A plain var, as it's only touched on the channel's event loop.
+  private[this] var requestTimeoutHandled = false
+
+  // Present only if a request timeout is set, and the pipeline has the HTTP codec registered under `ServerCodecHandlerName`. If absent, an
+  // exceeded request timeout is always reported as 503.
+  private[this] var requestBodyTracker: Option[RequestBodyCompletionTracker] = None
+
   private val logger = LoggerFactory.getLogger(getClass.getName)
   private final val WebSocketAutoPingHandlerName = "wsAutoPingHandler"
+  private final val RequestBodyTrackerHandlerName = "requestBodyTracker"
 
   override def handlerAdded(ctx: ChannelHandlerContext): Unit =
     if (ctx.channel.isActive) {
@@ -85,7 +94,12 @@ class NettyServerHandler[F[_]](
       // Initialize our ExecutionContext
       eventLoopContext = ExecutionContext.fromExecutor(ctx.channel.eventLoop)
       config.idleTimeout.foreach { idleTimeout =>
-        ctx.pipeline().addFirst(new IdleStateHandler(0, 0, idleTimeout.toMillis.toInt, TimeUnit.MILLISECONDS))
+        ctx.pipeline().addFirst(new IdleStateHandler(0, 0, idleTimeout.toMillis, TimeUnit.MILLISECONDS))
+      }
+      if (config.requestTimeout.isDefined && ctx.pipeline().context(ServerCodecHandlerName) != null) {
+        val tracker = new RequestBodyCompletionTracker
+        ctx.pipeline().addAfter(ServerCodecHandlerName, RequestBodyTrackerHandlerName, tracker)
+        requestBodyTracker = Some(tracker)
       }
       // When the channel closes we want to cancel any pending dispatches.
       // Since the listener will be executed from the channels EventLoop everything is thread safe.
@@ -100,27 +114,37 @@ class NettyServerHandler[F[_]](
     }
   }
 
-  def writeError503ThenClose(ctx: ChannelHandlerContext): Unit = {
-    val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.SERVICE_UNAVAILABLE)
+  private def writeErrorThenClose(ctx: ChannelHandlerContext, status: HttpResponseStatus): Unit = {
+    val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, status)
     res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0)
     res.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE)
     val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE)
   }
 
+  private def handleRequestTimeout(ctx: ChannelHandlerContext): Unit = {
+    val timeoutDescription = config.requestTimeout.map(_.toString).getOrElse("(not set)")
+    if (requestBodyTracker.forall(_.bodyFullyReceived)) {
+      logger.error(s"Closing connection with 503: no response produced within the request timeout of $timeoutDescription")
+      writeErrorThenClose(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE)
+    } else {
+      logger.debug(s"Closing connection with 408: request body not fully received within the request timeout of $timeoutDescription")
+      writeErrorThenClose(ctx, HttpResponseStatus.REQUEST_TIMEOUT)
+    }
+  }
+
   override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit = {
     evt match {
       case e: IdleStateEvent =>
-        if (e.state() == IdleState.WRITER_IDLE) {
-          logger.error(
-            s"Closing connection due to exceeded response timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}"
-          )
-          writeError503ThenClose(ctx)
-        }
-        if (e.state() == IdleState.ALL_IDLE) {
-          logger.debug(s"Closing connection due to exceeded idle timeout of ${config.idleTimeout.map(_.toString).getOrElse("(not set)")}")
-          val _ = ctx.close()
+        e.state() match {
+          case IdleState.WRITER_IDLE if !requestTimeoutHandled =>
+            requestTimeoutHandled = true
+            handleRequestTimeout(ctx)
+          case IdleState.ALL_IDLE =>
+            logger.debug(s"Closing connection due to exceeded idle timeout of ${config.idleTimeout.map(_.toString).getOrElse("(not set)")}")
+            val _ = ctx.close()
+          case _ => ()
         }
-      case other =>
+      case _ =>
         super.userEventTriggered(ctx, evt)
     }
   }
@@ -138,7 +162,7 @@ class NettyServerHandler[F[_]](
 
     def runRoute(req: HttpRequest, releaseReq: () => Any = () => ()): Unit = {
       val requestTimeoutHandler = config.requestTimeout.map { requestTimeout =>
-        new IdleStateHandler(0, requestTimeout.toMillis.toInt, 0, TimeUnit.MILLISECONDS)
+        new IdleStateHandler(0, requestTimeout.toMillis, 0, TimeUnit.MILLISECONDS)
       }
       requestTimeoutHandler.foreach(h => ctx.pipeline().addFirst(h))
       val (runningFuture, cancellationSwitch) = unsafeRunAsync { () =>
@@ -222,7 +246,7 @@ class NettyServerHandler[F[_]](
 
     if (isShuttingDown.get()) {
       logger.info("Rejecting request, server is shutting down")
-      writeError503ThenClose(ctx)
+      writeErrorThenClose(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE)
     } else if (HttpUtil.is100ContinueExpected(request)) {
       ctx.writeAndFlush(new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.CONTINUE))
       ()
@@ -355,6 +379,7 @@ class NettyServerHandler[F[_]](
       handshakeReq: HttpRequest
   ) = {
     ctx.pipeline().remove(this)
+    requestBodyTracker.foreach(ctx.pipeline().remove(_))
     ctx
       .pipeline()
       .addAfter(

server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala

@@ -0,0 +1,32 @@
+package sttp.tapir.server.netty.internal
+
+import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter}
+import io.netty.handler.codec.http.{HttpRequest, LastHttpContent}
+
+/** Tracks whether the body of the request currently being handled has been received in full, so that a firing request timeout can report
+  * 408 (client stalled mid-upload) instead of 503 (server too slow to respond). Has to be placed after the HTTP codec and before
+  * `HttpStreamsServerHandler`, which merges the [[HttpRequest]] / [[LastHttpContent]] messages this relies on into a single streamed
+  * request.
+  *
+  * The state is per connection, not per request: with auto-read enabled, a pipelined request's headers can be decoded while the preceding
+  * request is still being handled, resetting it. That preceding request's timeout is then reported as 408 rather than 503; only the status
+  * code is affected.
+  */
+private[netty] class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter {
+
+  // The initial value is never read in practice: the request timeout is armed only once a request's headers have passed through this
+  // handler, which sets the flag. `true` is the safe default, blaming the server (503) rather than the client (408).
+  private var _bodyFullyReceived: Boolean = true
+
+  def bodyFullyReceived: Boolean = _bodyFullyReceived
+
+  override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = {
+    // FullHttpRequest is both an HttpRequest and a LastHttpContent, so LastHttpContent has to be matched first
+    msg match {
+      case _: LastHttpContent => _bodyFullyReceived = true
+      case _: HttpRequest     => _bodyFullyReceived = false
+      case _                  => ()
+    }
+    val _ = ctx.fireChannelRead(msg)
+  }
+}

server/netty-server/src/test/scala/sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala

@@ -29,6 +29,9 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We
     interval = org.scalatest.time.Span(150, org.scalatest.time.Millis)
   )
 
+  private val timingOutRequest = new TimingOutRequestSpecData(eventLoopGroup)
+  import timingOutRequest._
+
   def tests(): List[Test] = List(
     Test("properly update metrics when a request times out") {
       val e = endpoint.post
@@ -85,6 +88,40 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We
           }
         }
         .unsafeToFuture()
+    },
+    Test("respond with status 408 when not all declared request body bytes are received") {
+      statusLinesFromShortTimeoutServer { (socket, port) =>
+        for {
+          _ <- send(socket, incompleteRequestHead(port))
+          status <- readStatusLine(socket)
+        } yield List(status)
+      }.map { statusLines =>
+        statusLines shouldBe List("HTTP/1.1 408 Request Timeout")
+      }.unsafeToFuture()
+    },
+    Test("respond with status 408 for an incomplete request following a complete one on the same connection") {
+      statusLinesFromShortTimeoutServer { (socket, port) =>
+        for {
+          _ <- send(socket, requestHead(port, completeBody.length) ++ completeBody)
+          first <- readStatusLine(socket)
+          _ <- send(socket, incompleteRequestHead(port))
+          second <- readStatusLine(socket)
+        } yield List(first, second)
+      }.map { statusLines =>
+        statusLines shouldBe List("HTTP/1.1 200 OK", "HTTP/1.1 408 Request Timeout")
+      }.unsafeToFuture()
+    },
+    Test("respond with status 503, not 408, for a slow but complete request following a complete fast one on the same connection") {
+      statusLinesFromShortTimeoutServer { (socket, port) =>
+        for {
+          _ <- send(socket, requestHead(port, completeBody.length) ++ completeBody)
+          first <- readStatusLine(socket)
+          _ <- send(socket, requestHead(port, slowBody.length) ++ slowBody)
+          second <- readStatusLine(socket)
+        } yield List(first, second)
+      }.map { statusLines =>
+        statusLines shouldBe List("HTTP/1.1 200 OK", "HTTP/1.1 503 Service Unavailable")
+      }.unsafeToFuture()
     }
   )
 }

server/netty-server/src/test/scala/sttp/tapir/server/netty/TimingOutRequestSpecData.scala

@@ -0,0 +1,84 @@
+package sttp.tapir.server.netty
+
+import cats.effect.IO
+import cats.effect.kernel.Resource
+import io.netty.channel.EventLoopGroup
+import sttp.tapir._
+
+import java.io.{BufferedReader, InputStreamReader}
+import java.net.Socket
+import java.nio.charset.StandardCharsets.US_ASCII
+import scala.concurrent.duration.DurationInt
+import scala.concurrent.{ExecutionContext, Future}
+
+class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: ExecutionContext) {
+
+  val completeBody: Array[Byte] = "test".getBytes(US_ASCII)
+  val slowBody: Array[Byte] = "slow".getBytes(US_ASCII)
+
+  private val shortRequestTimeout = 1.second
+  private val socketReadTimeout = shortRequestTimeout * 20
+  private val slowBodyString = new String(slowBody, US_ASCII)
+
+  def requestHead(port: Int, contentLength: Int): Array[Byte] =
+    s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: $contentLength\r\n\r\n"
+      .getBytes(US_ASCII)
+
+  def incompleteRequestHead(port: Int): Array[Byte] = requestHead(port, contentLength = 10000)
+
+  def send(socket: Socket, bytes: Array[Byte]): IO[Unit] =
+    IO.blocking {
+      socket.getOutputStream.write(bytes)
+      socket.getOutputStream.flush()
+    }
+
+  def readStatusLine(socket: Socket): IO[String] = IO.blocking {
+    val in = new BufferedReader(new InputStreamReader(socket.getInputStream, US_ASCII))
+    val statusLine = in.readLine()
+    val headers = Iterator.continually(in.readLine()).takeWhile(_.nonEmpty).toList
+
+    if (headers.exists(_.toLowerCase.contains("chunked"))) {
+      var chunkSize = in.readLine()
+      while (chunkSize != "0") {
+        in.readLine()
+        chunkSize = in.readLine()
+      }
+      in.readLine()
+    }
+    statusLine
+  }
+
+  def statusLinesFromShortTimeoutServer(interact: (Socket, Int) => IO[List[String]]): IO[List[String]] = {
+    val e = endpoint.put
+      .in(stringBody)
+      .out(stringBody)
+      .serverLogicSuccess[Future] { body =>
+        if (body == slowBodyString) Thread.sleep((shortRequestTimeout * 2).toMillis)
+        Future.successful(body)
+      }
+
+    val serverConfig = NettyConfig.default
+      .eventLoopGroup(eventLoopGroup)
+      .randomPort
+      .withDontShutdownEventLoopGroupOnClose
+      .noGracefulShutdown
+      .requestTimeout(shortRequestTimeout)
+
+    val bind = IO.fromFuture(IO.delay(NettyFutureServer(serverConfig).addEndpoints(List(e)).start()))
+
+    Resource
+      .make(bind)(server => IO.fromFuture(IO.delay(server.stop())))
+      .map(_.port)
+      .use { port =>
+        Resource.fromAutoCloseable(IO(clientSocket(port))).use { socket =>
+          interact(socket, port)
+        }
+      }
+  }
+
+  private def clientSocket(port: Int): Socket = {
+    val socket = new Socket("localhost", port)
+    socket.setSoTimeout(socketReadTimeout.toMillis.toInt)
+    socket
+  }
+}