diff --git a/doc/server/netty.md b/doc/server/netty.md index 6249ea65c6..4ef4f94f54 100644 --- a/doc/server/netty.md +++ b/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 diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala index 6b9d562506..b3b5ae1ec9 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/NettyConfig.scala +++ b/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. diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala index 1a3aa8d252..824d919c5e 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/NettyServerHandler.scala +++ b/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( diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala new file mode 100644 index 0000000000..5e84311410 --- /dev/null +++ b/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) + } +} diff --git a/server/netty-server/src/test/scala/sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala b/server/netty-server/src/test/scala/sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala index 577b3c2a71..81b530da25 100644 --- a/server/netty-server/src/test/scala/sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala +++ b/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() } ) } diff --git a/server/netty-server/src/test/scala/sttp/tapir/server/netty/TimingOutRequestSpecData.scala b/server/netty-server/src/test/scala/sttp/tapir/server/netty/TimingOutRequestSpecData.scala new file mode 100644 index 0000000000..ddf4081e9f --- /dev/null +++ b/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 + } +}