From 8b72e7979da5d35a7b6c34ecb75446f63b60430a Mon Sep 17 00:00:00 2001 From: rwalerow Date: Tue, 11 Aug 2026 14:11:24 +0200 Subject: [PATCH 01/41] Add test reproducing the issue of invalid response --- .../NettyFutureRequestTimeoutTests.scala | 26 +++++++++++++++++++ 1 file changed, 26 insertions(+) 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..173b4b1f0a 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 @@ -85,6 +85,32 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } } .unsafeToFuture() + }, + Test("respond with status 400 when not all declared bytes are received within time window") { + val e = endpoint.put + .in(stringBody) + .out(stringBody) + .serverLogicSuccess[Future] { body => + Future.successful(body) + } + + val config: NettyConfig = NettyConfig.default.requestTimeout(1.second) + + val bind = IO.fromFuture(IO.delay(NettyFutureServer(config).addEndpoints(List(e)).start())) + + Resource + .make(bind)(server => IO.fromFuture(IO.delay(server.stop()))) + .map(_.port) + .use { port => + basicRequest + .put(uri"http://localhost:$port") + .contentLength(10000L) + .body("test") + .send(backend).map { response => + response.code shouldBe StatusCode.BadRequest + } + } + .unsafeToFuture() } ) } From d120bb3bc70cac02fe03e378ae1176aca31a49f8 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Thu, 13 Aug 2026 14:38:46 +0200 Subject: [PATCH 02/41] Change test to plain java socket implementation --- .../NettyFutureRequestTimeoutTests.scala | 27 ++++++++++++------- 1 file changed, 18 insertions(+), 9 deletions(-) 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 173b4b1f0a..472ede0b18 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 @@ -1,7 +1,8 @@ package sttp.tapir.server.netty -import sttp.tapir._ +import sttp.tapir.* import sttp.tapir.tests.Test + import scala.concurrent.Future import java.util.concurrent.atomic.AtomicInteger import scala.concurrent.duration.DurationInt @@ -11,15 +12,18 @@ import sttp.tapir.server.metrics.EndpointMetric import io.netty.channel.EventLoopGroup import cats.effect.IO import cats.effect.kernel.Resource + import scala.concurrent.ExecutionContext -import sttp.client4._ +import sttp.client4.* import sttp.capabilities.fs2.Fs2Streams import org.scalatest.concurrent.Eventually import org.scalatest.concurrent.Eventually.eventually -import org.scalatest.matchers.should.Matchers._ +import org.scalatest.matchers.should.Matchers.* import cats.effect.unsafe.implicits.global import sttp.model.StatusCode +import java.net.Socket + class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: WebSocketStreamBackend[IO, Fs2Streams[IO]])(implicit ec: ExecutionContext ) { @@ -102,12 +106,17 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We .make(bind)(server => IO.fromFuture(IO.delay(server.stop()))) .map(_.port) .use { port => - basicRequest - .put(uri"http://localhost:$port") - .contentLength(10000L) - .body("test") - .send(backend).map { response => - response.code shouldBe StatusCode.BadRequest + val bytes = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\ntest".getBytes + + for { + socket <- IO(new Socket("localhost", port)) + _ <- IO(socket.getOutputStream.write(bytes)) + _ <- IO(socket.getOutputStream.flush()) + _ <- IO.sleep(4.second) + response <- IO(new String(socket.getInputStream.readAllBytes())) + _ <- IO(socket.close()) + } yield { + response should include ("400 Bad Request") } } .unsafeToFuture() From b45cbe60b70b28688b9a14a978ae15ed47b1af4b Mon Sep 17 00:00:00 2001 From: rwalerow Date: Thu, 13 Aug 2026 14:40:02 +0200 Subject: [PATCH 03/41] Add request body completed tracker to track whether entire body was already read --- .../sttp/tapir/server/netty/NettyConfig.scala | 2 ++ .../netty/internal/NettyServerHandler.scala | 9 +++++++- .../netty/internal/RequestBodyTracker.scala | 23 +++++++++++++++++++ 3 files changed, 33 insertions(+), 1 deletion(-) create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala 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..83428a9cf1 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 @@ -11,6 +11,7 @@ import io.netty.handler.ssl.SslContext import org.playframework.netty.http.HttpStreamsServerHandler import sttp.tapir.server.netty.NettyConfig.EventLoopConfig import sttp.tapir.server.netty.internal._ +import sttp.tapir.server.netty.internal.RequestBodyTracker import scala.concurrent.duration._ @@ -148,6 +149,7 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } + pipeline.addLast(new RequestBodyTrackerHandler) pipeline.addLast(new HttpStreamsServerHandler()) pipeline.addLast(handler) if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler()) 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..7aa090da95 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 @@ -107,6 +107,12 @@ class NettyServerHandler[F[_]]( val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) } + def writeError400ThenClose(ctx: ChannelHandlerContext): Unit = { + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST) + res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0) + val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) + } + override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit = { evt match { case e: IdleStateEvent => @@ -114,7 +120,8 @@ class NettyServerHandler[F[_]]( logger.error( s"Closing connection due to exceeded response timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) - writeError503ThenClose(ctx) + if(ctx.channel().attr(RequestBodyTracker.BodyComplete).get()) writeError503ThenClose(ctx) + else writeError400ThenClose(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)")}") diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala new file mode 100644 index 0000000000..581158de6c --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala @@ -0,0 +1,23 @@ +package sttp.tapir.server.netty.internal + +import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandler, ChannelInboundHandlerAdapter} +import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} +import io.netty.util.AttributeKey + +object RequestBodyTracker { + val key = "tapir.requestbody.completed" + + val BodyComplete: AttributeKey[Boolean] = AttributeKey.valueOf[Boolean](key) +} + +private[netty] class RequestBodyTrackerHandler extends ChannelInboundHandlerAdapter { + + override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { + msg match { + case _: LastHttpContent => ctx.channel().attr(RequestBodyTracker.BodyComplete).set(true) + case _: HttpRequest => ctx.channel().attr(RequestBodyTracker.BodyComplete).set(false) + case _ => () + } + ctx.fireChannelRead(msg) + } +} From 2ad1f490bbd66f91c8a5b3908b92bb972e789ac3 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 10:38:10 +0200 Subject: [PATCH 04/41] Rename Body tracker to express intention --- .../scala/sttp/tapir/server/netty/NettyConfig.scala | 4 ++-- ...racker.scala => RequestBodyCompletedTracker.scala} | 11 +++++++---- 2 files changed, 9 insertions(+), 6 deletions(-) rename server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/{RequestBodyTracker.scala => RequestBodyCompletedTracker.scala} (51%) 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 83428a9cf1..e5549ff421 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 @@ -11,7 +11,7 @@ import io.netty.handler.ssl.SslContext import org.playframework.netty.http.HttpStreamsServerHandler import sttp.tapir.server.netty.NettyConfig.EventLoopConfig import sttp.tapir.server.netty.internal._ -import sttp.tapir.server.netty.internal.RequestBodyTracker +import sttp.tapir.server.netty.internal.RequestBodyCompletedTracker import scala.concurrent.duration._ @@ -149,7 +149,7 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - pipeline.addLast(new RequestBodyTrackerHandler) + pipeline.addLast(new RequestBodyCompletedTracker) pipeline.addLast(new HttpStreamsServerHandler()) pipeline.addLast(handler) if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler()) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala similarity index 51% rename from server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala rename to server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index 581158de6c..71edcde3a4 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -4,18 +4,21 @@ import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandler, ChannelIn import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} import io.netty.util.AttributeKey -object RequestBodyTracker { +object RequestBodyCompletedTracker { val key = "tapir.requestbody.completed" val BodyComplete: AttributeKey[Boolean] = AttributeKey.valueOf[Boolean](key) + + def wasBodyCompletelySend(ctx: ChannelHandlerContext): Boolean = + Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).contains(true) } -private[netty] class RequestBodyTrackerHandler extends ChannelInboundHandlerAdapter { +private[netty] class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { msg match { - case _: LastHttpContent => ctx.channel().attr(RequestBodyTracker.BodyComplete).set(true) - case _: HttpRequest => ctx.channel().attr(RequestBodyTracker.BodyComplete).set(false) + case _: LastHttpContent => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(true) + case _: HttpRequest => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(false) case _ => () } ctx.fireChannelRead(msg) From f0dd8e95e24ced673d4103e62dbddb95fa51f9cb Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 10:38:53 +0200 Subject: [PATCH 05/41] Improve the way user events are handled --- .../server/netty/internal/NettyServerHandler.scala | 11 ++++++++--- 1 file changed, 8 insertions(+), 3 deletions(-) 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 7aa090da95..e5cb484f11 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 @@ -116,12 +116,17 @@ class NettyServerHandler[F[_]]( override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit = { evt match { case e: IdleStateEvent => - if (e.state() == IdleState.WRITER_IDLE) { + if (e.state() == IdleState.WRITER_IDLE && RequestBodyCompletedTracker.wasBodyCompletelySend(ctx)) { logger.error( s"Closing connection due to exceeded response timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) - if(ctx.channel().attr(RequestBodyTracker.BodyComplete).get()) writeError503ThenClose(ctx) - else writeError400ThenClose(ctx) + writeError503ThenClose(ctx) + } + if(e.state == IdleState.WRITER_IDLE) { + logger.error( + s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" + ) + writeError400ThenClose(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)")}") From 3d36b48bfd33f0b0601a001b156720ad75a834fe Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 11:03:54 +0200 Subject: [PATCH 06/41] Fix multiple event branches call on Idle event and add missing connection close header --- .../tapir/server/netty/internal/NettyServerHandler.scala | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) 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 e5cb484f11..85a4da7d27 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 @@ -32,6 +32,7 @@ import scala.collection.mutable.{Queue => MutableQueue} import scala.concurrent.{ExecutionContext, Future} import scala.util.control.NonFatal import scala.util.{Failure, Success} +import RequestBodyCompletedTracker._ /** @param unsafeRunAsync * Function which dispatches given effect to run asynchronously, returning its result as a Future, and function of type `() => @@ -110,19 +111,20 @@ class NettyServerHandler[F[_]]( def writeError400ThenClose(ctx: ChannelHandlerContext): Unit = { val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST) res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0) + res.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) } override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit = { evt match { case e: IdleStateEvent => - if (e.state() == IdleState.WRITER_IDLE && RequestBodyCompletedTracker.wasBodyCompletelySend(ctx)) { + if (e.state() == IdleState.WRITER_IDLE && wasBodyCompletelySend(ctx)) { logger.error( s"Closing connection due to exceeded response timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) writeError503ThenClose(ctx) } - if(e.state == IdleState.WRITER_IDLE) { + if(e.state == IdleState.WRITER_IDLE && !wasBodyCompletelySend(ctx)) { logger.error( s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) From 1b151a8dd511c3f52a00b1d980aa1d9690798a34 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 11:11:48 +0200 Subject: [PATCH 07/41] Randomize port in a new test and lower the timeouts --- .../tapir/server/netty/NettyFutureRequestTimeoutTests.scala | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) 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 472ede0b18..0daf893482 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 @@ -98,7 +98,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We Future.successful(body) } - val config: NettyConfig = NettyConfig.default.requestTimeout(1.second) + val config: NettyConfig = NettyConfig.default.randomPort.requestTimeout(500.millis) val bind = IO.fromFuture(IO.delay(NettyFutureServer(config).addEndpoints(List(e)).start())) @@ -112,7 +112,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We socket <- IO(new Socket("localhost", port)) _ <- IO(socket.getOutputStream.write(bytes)) _ <- IO(socket.getOutputStream.flush()) - _ <- IO.sleep(4.second) + _ <- IO.sleep(1.second) response <- IO(new String(socket.getInputStream.readAllBytes())) _ <- IO(socket.close()) } yield { From c9082df9721583c1f487cad5b44e038514468d01 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 11:47:13 +0200 Subject: [PATCH 08/41] Report 400 only on READ timeouts with uncomplete body send --- .../tapir/server/netty/internal/NettyServerHandler.scala | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) 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 85a4da7d27..082afe241b 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 @@ -118,13 +118,13 @@ class NettyServerHandler[F[_]]( override def userEventTriggered(ctx: ChannelHandlerContext, evt: Any): Unit = { evt match { case e: IdleStateEvent => - if (e.state() == IdleState.WRITER_IDLE && wasBodyCompletelySend(ctx)) { + 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.WRITER_IDLE && !wasBodyCompletelySend(ctx)) { + if(e.state == IdleState.READER_IDLE && !wasBodyCompletelySend(ctx)) { logger.error( s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) @@ -152,7 +152,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(requestTimeout.toMillis.toInt, requestTimeout.toMillis.toInt, 0, TimeUnit.MILLISECONDS) } requestTimeoutHandler.foreach(h => ctx.pipeline().addFirst(h)) val (runningFuture, cancellationSwitch) = unsafeRunAsync { () => From 65860922c2704000182b6b4ad7fc838179962c9e Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 14 Aug 2026 12:43:03 +0200 Subject: [PATCH 09/41] Add a new behavior description to the netty conifg --- .../src/main/scala/sttp/tapir/server/netty/NettyConfig.scala | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) 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 e5549ff421..cd780539ea 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 @@ -32,7 +32,9 @@ import scala.concurrent.duration._ * * @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`. + * connection. + * If exceeded when request was partially send and then stalled the server will return a HTTP 400 and close the connection. + * This timeout is 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. From 7bce6605bb7a83fbb0f72b74c9b0bfa96a74f386 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Mon, 17 Aug 2026 10:14:31 +0200 Subject: [PATCH 10/41] Reformat code --- .../server/netty/NettyFutureRequestTimeoutTests.scala | 8 ++++---- 1 file changed, 4 insertions(+), 4 deletions(-) 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 0daf893482..8ea4852039 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 @@ -1,6 +1,6 @@ package sttp.tapir.server.netty -import sttp.tapir.* +import sttp.tapir._ import sttp.tapir.tests.Test import scala.concurrent.Future @@ -14,11 +14,11 @@ import cats.effect.IO import cats.effect.kernel.Resource import scala.concurrent.ExecutionContext -import sttp.client4.* +import sttp.client4._ import sttp.capabilities.fs2.Fs2Streams import org.scalatest.concurrent.Eventually import org.scalatest.concurrent.Eventually.eventually -import org.scalatest.matchers.should.Matchers.* +import org.scalatest.matchers.should.Matchers._ import cats.effect.unsafe.implicits.global import sttp.model.StatusCode @@ -116,7 +116,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We response <- IO(new String(socket.getInputStream.readAllBytes())) _ <- IO(socket.close()) } yield { - response should include ("400 Bad Request") + response should include("400 Bad Request") } } .unsafeToFuture() From 3908f64c39f2bbb7d0a4563edaf62b4cbbf254e5 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 19 Aug 2026 10:28:05 +0200 Subject: [PATCH 11/41] Remove unused import and rename method to be more intuitive --- .../tapir/server/netty/internal/NettyServerHandler.scala | 2 +- .../server/netty/internal/RequestBodyCompletedTracker.scala | 6 +++--- 2 files changed, 4 insertions(+), 4 deletions(-) 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 082afe241b..ce9cf67456 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 @@ -124,7 +124,7 @@ class NettyServerHandler[F[_]]( ) writeError503ThenClose(ctx) } - if(e.state == IdleState.READER_IDLE && !wasBodyCompletelySend(ctx)) { + if(e.state == IdleState.READER_IDLE && !wasRequestBodyFullyReceived(ctx)) { logger.error( s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index 71edcde3a4..1674b25b8d 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -1,6 +1,6 @@ package sttp.tapir.server.netty.internal -import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandler, ChannelInboundHandlerAdapter} +import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} import io.netty.util.AttributeKey @@ -9,7 +9,7 @@ object RequestBodyCompletedTracker { val BodyComplete: AttributeKey[Boolean] = AttributeKey.valueOf[Boolean](key) - def wasBodyCompletelySend(ctx: ChannelHandlerContext): Boolean = + def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).contains(true) } @@ -21,6 +21,6 @@ private[netty] class RequestBodyCompletedTracker extends ChannelInboundHandlerAd case _: HttpRequest => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(false) case _ => () } - ctx.fireChannelRead(msg) + val _ = ctx.fireChannelRead(msg) } } From 429b054fd733a47656c9fd2f98e2e99e79e957c2 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 11:17:04 +0200 Subject: [PATCH 12/41] Unify error write methods and change logging type for not fully send request --- .../netty/internal/NettyServerHandler.scala | 21 +++++++------------ 1 file changed, 7 insertions(+), 14 deletions(-) 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 ce9cf67456..d073297931 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 @@ -101,15 +101,8 @@ class NettyServerHandler[F[_]]( } } - def writeError503ThenClose(ctx: ChannelHandlerContext): Unit = { - val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.SERVICE_UNAVAILABLE) - res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0) - res.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) - val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) - } - - def writeError400ThenClose(ctx: ChannelHandlerContext): Unit = { - val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST) + def writeErrorThenClose(ctx: ChannelHandlerContext, errorResponseStatus: HttpResponseStatus): Unit = { + val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, errorResponseStatus) res.headers().set(HttpHeaderNames.CONTENT_LENGTH, 0) res.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.CLOSE) val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) @@ -122,13 +115,13 @@ class NettyServerHandler[F[_]]( logger.error( s"Closing connection due to exceeded response timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) - writeError503ThenClose(ctx) + writeErrorThenClose(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE) } - if(e.state == IdleState.READER_IDLE && !wasRequestBodyFullyReceived(ctx)) { - logger.error( + if (e.state == IdleState.READER_IDLE && !wasRequestBodyFullyReceived(ctx)) { + logger.debug( s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) - writeError400ThenClose(ctx) + writeErrorThenClose(ctx, HttpResponseStatus.BAD_REQUEST) } if (e.state() == IdleState.ALL_IDLE) { logger.debug(s"Closing connection due to exceeded idle timeout of ${config.idleTimeout.map(_.toString).getOrElse("(not set)")}") @@ -236,7 +229,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)) () From f644f0c2c41c848c4add7a30b0afcb1c3162dcde Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 12:04:23 +0200 Subject: [PATCH 13/41] Rephrase param documentation --- .../src/main/scala/sttp/tapir/server/netty/NettyConfig.scala | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) 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 cd780539ea..a08b0a5699 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 @@ -31,9 +31,7 @@ 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. - * If exceeded when request was partially send and then stalled the server will return a HTTP 400 and close the connection. + * The maximum duration to wait for a response to be produced, which includes receiving the request. * This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`. * * @param connectionTimeout From eeb50f9831569cbc03e9249fa90245708d860062 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 12:05:18 +0200 Subject: [PATCH 14/41] Tracking body completed would work properly also for provided pipelines --- .../server/netty/internal/RequestBodyCompletedTracker.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index 1674b25b8d..579ef2af86 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -10,7 +10,7 @@ object RequestBodyCompletedTracker { val BodyComplete: AttributeKey[Boolean] = AttributeKey.valueOf[Boolean](key) def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = - Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).contains(true) + Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).getOrElse(true) } private[netty] class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { From 9a6c5b5a223d268d89fbe2b8bb8705c40bbcb86a Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 12:48:38 +0200 Subject: [PATCH 15/41] Change test to check whethere 503 was not additionally writen and use Resource to close a socket --- .../NettyFutureRequestTimeoutTests.scala | 21 +++++++++++-------- 1 file changed, 12 insertions(+), 9 deletions(-) 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 8ea4852039..1c999e8677 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 @@ -108,15 +108,18 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We .use { port => val bytes = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\ntest".getBytes - for { - socket <- IO(new Socket("localhost", port)) - _ <- IO(socket.getOutputStream.write(bytes)) - _ <- IO(socket.getOutputStream.flush()) - _ <- IO.sleep(1.second) - response <- IO(new String(socket.getInputStream.readAllBytes())) - _ <- IO(socket.close()) - } yield { - response should include("400 Bad Request") + Resource + .make(IO(new Socket("localhost", port)))(socket => IO(socket.close())) + .use { socket => + for { + _ <- IO(socket.getOutputStream.write(bytes)) + _ <- IO(socket.getOutputStream.flush()) + _ <- IO.sleep(1.second) + response <- IO(new String(socket.getInputStream.readAllBytes())) + } yield { + response should include("400 Bad Request") + response should not include("503") + } } } .unsafeToFuture() From 114ed8620316613c3b66d9f579a73900d8a25084 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 12:57:56 +0200 Subject: [PATCH 16/41] Request completed tracker makes sens only if request timeout is set --- .../main/scala/sttp/tapir/server/netty/NettyConfig.scala | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) 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 a08b0a5699..c055a3fee6 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 @@ -31,8 +31,8 @@ import scala.concurrent.duration._ * contains tapir's server processing logic. * * @param requestTimeout - * The maximum duration to wait for a response to be produced, which includes receiving the request. - * This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`. + * The maximum duration to wait for a response to be produced, which includes receiving the request. This timeout is 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. @@ -149,7 +149,9 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - pipeline.addLast(new RequestBodyCompletedTracker) + if (cfg.requestTimeout.isDefined) { + pipeline.addLast(new RequestBodyCompletedTracker) + } pipeline.addLast(new HttpStreamsServerHandler()) pipeline.addLast(handler) if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler()) From b7aecaa2b58d0881534836c473d21a13281f001b Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 13:01:24 +0200 Subject: [PATCH 17/41] Added an information where in the pipeline should RequestBodyCompletedTracker be placed --- .../server/netty/internal/RequestBodyCompletedTracker.scala | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index 579ef2af86..f388deac3e 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -4,6 +4,10 @@ import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} import io.netty.util.AttributeKey +/** + * Has to be included in the pipeline before HttpStreamsServerHandler + * to observe LastHttpContent and HttpRequest + */ object RequestBodyCompletedTracker { val key = "tapir.requestbody.completed" From a6b9a6d4413c7053f28cf24ff276ada40aa41ec9 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 13:22:39 +0200 Subject: [PATCH 18/41] Remove unnecesary import --- .../src/main/scala/sttp/tapir/server/netty/NettyConfig.scala | 1 - 1 file changed, 1 deletion(-) 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 c055a3fee6..e95c1d66e4 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 @@ -11,7 +11,6 @@ import io.netty.handler.ssl.SslContext import org.playframework.netty.http.HttpStreamsServerHandler import sttp.tapir.server.netty.NettyConfig.EventLoopConfig import sttp.tapir.server.netty.internal._ -import sttp.tapir.server.netty.internal.RequestBodyCompletedTracker import scala.concurrent.duration._ From 38168cd8575085f27a59382cac9d176a17834abb Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 15:30:26 +0200 Subject: [PATCH 19/41] Reformat missed files --- .../netty/internal/RequestBodyCompletedTracker.scala | 10 ++++------ .../server/netty/NettyFutureRequestTimeoutTests.scala | 4 ++-- 2 files changed, 6 insertions(+), 8 deletions(-) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index f388deac3e..a8b9886d34 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -4,10 +4,8 @@ import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} import io.netty.util.AttributeKey -/** - * Has to be included in the pipeline before HttpStreamsServerHandler - * to observe LastHttpContent and HttpRequest - */ +/** Has to be included in the pipeline before HttpStreamsServerHandler to observe LastHttpContent and HttpRequest + */ object RequestBodyCompletedTracker { val key = "tapir.requestbody.completed" @@ -22,8 +20,8 @@ private[netty] class RequestBodyCompletedTracker extends ChannelInboundHandlerAd override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { msg match { case _: LastHttpContent => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(true) - case _: HttpRequest => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(false) - case _ => () + case _: HttpRequest => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(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 1c999e8677..da0cd5fdbc 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 @@ -118,9 +118,9 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We response <- IO(new String(socket.getInputStream.readAllBytes())) } yield { response should include("400 Bad Request") - response should not include("503") + response should not include ("503") } - } + } } .unsafeToFuture() } From f8d53eba5f0970e8779ad917263f89cd9aac5e42 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 15:59:06 +0200 Subject: [PATCH 20/41] Rephrase requestTimeout documentation --- .../scala/sttp/tapir/server/netty/NettyConfig.scala | 11 +++++++++-- 1 file changed, 9 insertions(+), 2 deletions(-) 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 e95c1d66e4..252d52f0d9 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,15 @@ import scala.concurrent.duration._ * contains tapir's server processing logic. * * @param requestTimeout - * The maximum duration to wait for a response to be produced, which includes receiving the request. This timeout is ignored in Web - * Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`. + * The maximum duration to wait for the request to be received in full, and for the response to be produced. When exceeded, an empty + * error response with a `Connection: close` header is sent, and the connection is closed. The status code depends on which side stalled: + * + * - `503 Service Unavailable`, if no part of the response is written within this duration. The server did not produce, or start + * streaming, a response in time. + * - `400 Bad Request`, if no data is received within this duration while the request body is still incomplete. The client sent the + * headers, and possibly part of the body, then stalled before sending the final chunk. + * + * This timeout is 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. From 35850c84f22304a9bc9c330dc1b9121b47eebc46 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 26 Aug 2026 16:12:21 +0200 Subject: [PATCH 21/41] Rephrase reading debug message, grammar fix --- .../sttp/tapir/server/netty/internal/NettyServerHandler.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 d073297931..1c3c6b1f30 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 @@ -119,7 +119,7 @@ class NettyServerHandler[F[_]]( } if (e.state == IdleState.READER_IDLE && !wasRequestBodyFullyReceived(ctx)) { logger.debug( - s"Closing connection due to partially send request with pause exceeded request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" + s"Closing connection: the request body was not fully received within the request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) writeErrorThenClose(ctx, HttpResponseStatus.BAD_REQUEST) } From 67f072e0e811cdac88efe50caf09c6077a502c77 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Thu, 27 Aug 2026 16:07:15 +0200 Subject: [PATCH 22/41] Open up Request completed Tracker --- .../server/netty/internal/RequestBodyCompletedTracker.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index a8b9886d34..de039a9d3a 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -15,7 +15,7 @@ object RequestBodyCompletedTracker { Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).getOrElse(true) } -private[netty] class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { +class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { msg match { From 060b819982acdbf7acc5f756ae00f21f210b2137 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Thu, 27 Aug 2026 16:07:31 +0200 Subject: [PATCH 23/41] Remove unnecesary wait witihn a test case --- .../sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala | 1 - 1 file changed, 1 deletion(-) 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 da0cd5fdbc..e44ff4bb84 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 @@ -114,7 +114,6 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We for { _ <- IO(socket.getOutputStream.write(bytes)) _ <- IO(socket.getOutputStream.flush()) - _ <- IO.sleep(1.second) response <- IO(new String(socket.getInputStream.readAllBytes())) } yield { response should include("400 Bad Request") From da05efe054addbd3f7e7d17bd9f1f344c17653fb Mon Sep 17 00:00:00 2001 From: rwalerow Date: Thu, 27 Aug 2026 16:31:14 +0200 Subject: [PATCH 24/41] Add so timeout to socket --- .../server/netty/NettyFutureRequestTimeoutTests.scala | 8 +++++++- 1 file changed, 7 insertions(+), 1 deletion(-) 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 e44ff4bb84..9c17d169c6 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 @@ -102,6 +102,12 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We val bind = IO.fromFuture(IO.delay(NettyFutureServer(config).addEndpoints(List(e)).start())) + val createSocket: Int => Socket = port => { + val s = new Socket("localhost", port) + s.setSoTimeout(1000) + s + } + Resource .make(bind)(server => IO.fromFuture(IO.delay(server.stop()))) .map(_.port) @@ -109,7 +115,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We val bytes = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\ntest".getBytes Resource - .make(IO(new Socket("localhost", port)))(socket => IO(socket.close())) + .make(IO(createSocket(port)))(socket => IO(socket.close())) .use { socket => for { _ <- IO(socket.getOutputStream.write(bytes)) From 75c63b5daa583ae2eaf251fd10f4f96cac645ec5 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 28 Aug 2026 12:07:23 +0200 Subject: [PATCH 25/41] Find and remove request body tracker when upgrading connection to a websocket --- .../netty/internal/NettyServerHandler.scala | 15 ++++++++------- 1 file changed, 8 insertions(+), 7 deletions(-) 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 1c3c6b1f30..a804c1f26f 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 @@ -1,9 +1,9 @@ package sttp.tapir.server.netty.internal import io.netty.buffer.{ByteBuf, Unpooled} -import io.netty.channel._ +import io.netty.channel.* import io.netty.channel.group.ChannelGroup -import io.netty.handler.codec.http._ +import io.netty.handler.codec.http.* import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory import io.netty.handler.stream.{ChunkedFile, ChunkedStream} import io.netty.handler.timeout.{IdleState, IdleStateEvent, IdleStateHandler} @@ -12,7 +12,7 @@ import org.reactivestreams.{Publisher, Subscriber, Subscription} import org.slf4j.LoggerFactory import sttp.model.StatusCode import sttp.monad.MonadError -import sttp.monad.syntax._ +import sttp.monad.syntax.* import sttp.tapir.server.model.ServerResponse import sttp.tapir.server.netty.NettyResponseContent.{ ByteBufNettyResponseContent, @@ -27,12 +27,12 @@ import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean -import scala.collection.JavaConverters._ -import scala.collection.mutable.{Queue => MutableQueue} +import scala.collection.JavaConverters.* +import scala.collection.mutable.Queue as MutableQueue import scala.concurrent.{ExecutionContext, Future} import scala.util.control.NonFatal -import scala.util.{Failure, Success} -import RequestBodyCompletedTracker._ +import scala.util.{Failure, Success, Try} +import RequestBodyCompletedTracker.* /** @param unsafeRunAsync * Function which dispatches given effect to run asynchronously, returning its result as a Future, and function of type `() => @@ -362,6 +362,7 @@ class NettyServerHandler[F[_]]( handshakeReq: HttpRequest ) = { ctx.pipeline().remove(this) + Option(ctx.pipeline().get(classOf[RequestBodyCompletedTracker])).foreach(ctx.pipeline().remove) ctx .pipeline() .addAfter( From a7ff81189f950d24d838e38de0250f7cf54bf887 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Fri, 28 Aug 2026 12:14:17 +0200 Subject: [PATCH 26/41] Reorganize imports --- .../netty/internal/NettyServerHandler.scala | 20 +++++++------------ 1 file changed, 7 insertions(+), 13 deletions(-) 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 a804c1f26f..49b8eef91d 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 @@ -1,9 +1,9 @@ package sttp.tapir.server.netty.internal import io.netty.buffer.{ByteBuf, Unpooled} -import io.netty.channel.* +import io.netty.channel._ import io.netty.channel.group.ChannelGroup -import io.netty.handler.codec.http.* +import io.netty.handler.codec.http._ import io.netty.handler.codec.http.websocketx.WebSocketServerHandshakerFactory import io.netty.handler.stream.{ChunkedFile, ChunkedStream} import io.netty.handler.timeout.{IdleState, IdleStateEvent, IdleStateHandler} @@ -12,27 +12,21 @@ import org.reactivestreams.{Publisher, Subscriber, Subscription} import org.slf4j.LoggerFactory import sttp.model.StatusCode import sttp.monad.MonadError -import sttp.monad.syntax.* +import sttp.monad.syntax._ import sttp.tapir.server.model.ServerResponse -import sttp.tapir.server.netty.NettyResponseContent.{ - ByteBufNettyResponseContent, - ChunkedFileNettyResponseContent, - ChunkedStreamNettyResponseContent, - ReactivePublisherNettyResponseContent, - ReactiveWebSocketProcessorNettyResponseContent -} +import sttp.tapir.server.netty.NettyResponseContent._ +import sttp.tapir.server.netty.internal.RequestBodyCompletedTracker._ import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean -import scala.collection.JavaConverters.* +import scala.collection.JavaConverters._ import scala.collection.mutable.Queue as MutableQueue import scala.concurrent.{ExecutionContext, Future} import scala.util.control.NonFatal -import scala.util.{Failure, Success, Try} -import RequestBodyCompletedTracker.* +import scala.util.{Failure, Success} /** @param unsafeRunAsync * Function which dispatches given effect to run asynchronously, returning its result as a Future, and function of type `() => From ed782d227a2e38b1fa29553da465f0c36c85218b Mon Sep 17 00:00:00 2001 From: rwalerow Date: Mon, 31 Aug 2026 11:31:49 +0200 Subject: [PATCH 27/41] Fix renaming import --- .../sttp/tapir/server/netty/internal/NettyServerHandler.scala | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 49b8eef91d..fe24d5b6ed 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 @@ -23,7 +23,7 @@ import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean import scala.collection.JavaConverters._ -import scala.collection.mutable.Queue as MutableQueue +import scala.collection.mutable.{Queue => MutableQueue} import scala.concurrent.{ExecutionContext, Future} import scala.util.control.NonFatal import scala.util.{Failure, Success} From 96529ba81748746f0037c089781b2f586f04e8e4 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Mon, 31 Aug 2026 13:24:30 +0200 Subject: [PATCH 28/41] Use stateful handler instead of channle attribute --- .../internal/RequestBodyCompletedTracker.scala | 14 ++++++++------ 1 file changed, 8 insertions(+), 6 deletions(-) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index de039a9d3a..e74b40c83b 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -7,20 +7,22 @@ import io.netty.util.AttributeKey /** Has to be included in the pipeline before HttpStreamsServerHandler to observe LastHttpContent and HttpRequest */ object RequestBodyCompletedTracker { - val key = "tapir.requestbody.completed" - - val BodyComplete: AttributeKey[Boolean] = AttributeKey.valueOf[Boolean](key) def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = - Option(ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).get()).getOrElse(true) + ctx.pipeline().get(classOf[RequestBodyCompletedTracker]) match { + case t: RequestBodyCompletedTracker => t.bodyFullyReceived + case _ => true + } } class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { + private[internal] var bodyFullyReceived: Boolean = true + override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { msg match { - case _: LastHttpContent => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(true) - case _: HttpRequest => ctx.channel().attr(RequestBodyCompletedTracker.BodyComplete).set(false) + case _: LastHttpContent => bodyFullyReceived = true + case _: HttpRequest => bodyFullyReceived = false case _ => () } val _ = ctx.fireChannelRead(msg) From 345cf0cd7f37591edbd6eeef5375267c530f5acd Mon Sep 17 00:00:00 2001 From: rwalerow Date: Mon, 31 Aug 2026 13:53:15 +0200 Subject: [PATCH 29/41] Return 408 instead of 400 in incomplete request --- .../sttp/tapir/server/netty/internal/NettyServerHandler.scala | 2 +- .../tapir/server/netty/NettyFutureRequestTimeoutTests.scala | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) 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 fe24d5b6ed..37baf54255 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 @@ -115,7 +115,7 @@ class NettyServerHandler[F[_]]( logger.debug( s"Closing connection: the request body was not fully received within the request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" ) - writeErrorThenClose(ctx, HttpResponseStatus.BAD_REQUEST) + writeErrorThenClose(ctx, HttpResponseStatus.REQUEST_TIMEOUT) } if (e.state() == IdleState.ALL_IDLE) { logger.debug(s"Closing connection due to exceeded idle timeout of ${config.idleTimeout.map(_.toString).getOrElse("(not set)")}") 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 9c17d169c6..6461d8c681 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 @@ -122,7 +122,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We _ <- IO(socket.getOutputStream.flush()) response <- IO(new String(socket.getInputStream.readAllBytes())) } yield { - response should include("400 Bad Request") + response should include("408 Request Timeout") response should not include ("503") } } From c23839ffdfa3faf6a2ea67a0aeafba3c5ac6d8cf Mon Sep 17 00:00:00 2001 From: rwalerow Date: Mon, 31 Aug 2026 17:54:01 +0200 Subject: [PATCH 30/41] Move docs to 408 status and remove unused import --- .../src/main/scala/sttp/tapir/server/netty/NettyConfig.scala | 2 +- .../server/netty/internal/RequestBodyCompletedTracker.scala | 1 - .../tapir/server/netty/NettyFutureRequestTimeoutTests.scala | 2 +- 3 files changed, 2 insertions(+), 3 deletions(-) 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 252d52f0d9..704e07c513 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 @@ -35,7 +35,7 @@ import scala.concurrent.duration._ * * - `503 Service Unavailable`, if no part of the response is written within this duration. The server did not produce, or start * streaming, a response in time. - * - `400 Bad Request`, if no data is received within this duration while the request body is still incomplete. The client sent the + * - `408 Request Timeout`, if no data is received within this duration while the request body is still incomplete. The client sent the * headers, and possibly part of the body, then stalled before sending the final chunk. * * This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`. diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala index e74b40c83b..35b7866067 100644 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala @@ -2,7 +2,6 @@ package sttp.tapir.server.netty.internal import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} -import io.netty.util.AttributeKey /** Has to be included in the pipeline before HttpStreamsServerHandler to observe LastHttpContent and HttpRequest */ 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 6461d8c681..8fd60e921d 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 @@ -90,7 +90,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } .unsafeToFuture() }, - Test("respond with status 400 when not all declared bytes are received within time window") { + Test("respond with status 408 when not all declared bytes are received within time window") { val e = endpoint.put .in(stringBody) .out(stringBody) From 69232bcddb0a90d9ae801d314ba8277ca9e40d58 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Tue, 1 Sep 2026 15:51:01 +0200 Subject: [PATCH 31/41] Rename body completed tracker and rollback to single timeout to avoid double notification --- doc/server/netty.md | 29 ++++- .../sttp/tapir/server/netty/NettyConfig.scala | 20 ++-- .../netty/internal/NettyServerHandler.scala | 48 ++++---- .../RequestBodyCompletedTracker.scala | 29 ----- .../RequestBodyCompletionTracker.scala | 31 +++++ .../NettyFutureRequestTimeoutTests.scala | 106 +++++++++++++----- 6 files changed, 179 insertions(+), 84 deletions(-) delete mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala diff --git a/doc/server/netty.md b/doc/server/netty.md index 6249ea65c6..f4841f9935 100644 --- a/doc/server/netty.md +++ b/doc/server/netty.md @@ -100,7 +100,9 @@ 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: bounds the time between receiving the request headers and + producing a response, and so also bounds how long a client may take to upload + a request body; see [request timeout](#request-timeout) below * connection timeout * linger timeout * graceful shutdown timeout: when stopped e.g. using @@ -121,6 +123,31 @@ import scala.concurrent.duration.* val config = NettyConfig.default.requestTimeout(5.seconds) ``` +### Request timeout + +The request timeout starts when the request headers are received, and is only +satisfied once a response starts being written. When it is exceeded, the server +sends an empty response with a `Connection: close` header and closes the +connection. The status code says which side ran out of time, which is decided by +whether the request body had been received in full when the timeout expired: + +* `503 Service Unavailable`, if the request had been received in full, but no + part of a response was produced in time — the endpoint's logic is too slow. +* `408 Request Timeout`, if the request body was still incomplete — the client + sent the headers, and possibly part of the body, then either stalled or kept + sending too slowly to finish in time. A client declaring + `Content-Length: 10000` and then sending only the first few bytes is answered + this way. + +Because the timeout also covers receiving the body, it has to be higher than the +longest upload you want to accept, and lower than `idleTimeout`. It is ignored +for Web Sockets, once the handshake has been established. + +Telling the two cases apart relies on a handler which +`NettyConfig.defaultInitPipeline` adds when `requestTimeout` is set. A custom +`initPipeline` which does not add it still gets the timeout, but always reports +it as `503`; see the `initPipeline` scaladoc for the placement requirement. + ## 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 704e07c513..39e9c5f1ed 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 @@ -27,16 +27,18 @@ import scala.concurrent.duration._ * * @param initPipeline * The function to create the Netty pipeline, using the configuration instance, the pipeline created so far, and the handler which - * contains tapir's server processing logic. + * contains tapir's server processing logic. A custom pipeline which omits the handlers added by [[NettyConfig.defaultInitPipeline]] also + * gives up the behaviour which depends on them; in particular, without a `RequestBodyCompletionTracker` placed after the HTTP codec and + * before `HttpStreamsServerHandler`, an exceeded `requestTimeout` is always reported as `503`, never as `408`. * * @param requestTimeout - * The maximum duration to wait for the request to be received in full, and for the response to be produced. When exceeded, an empty - * error response with a `Connection: close` header is sent, and the connection is closed. The status code depends on which side stalled: + * The maximum duration between receiving the request headers and producing a response, which therefore also bounds the time the client + * has to send the request body. When exceeded, an empty error response with a `Connection: close` header is sent, and the connection is + * closed. The status code is decided by whether the request body had been received in full when the timeout expired: * - * - `503 Service Unavailable`, if no part of the response is written within this duration. The server did not produce, or start - * streaming, a response in time. - * - `408 Request Timeout`, if no data is received within this duration while the request body is still incomplete. The client sent the - * headers, and possibly part of the body, then stalled before sending the final chunk. + * - `503 Service Unavailable`, if the request had been received in full, but no part of a response was produced in time. + * - `408 Request Timeout`, if the request body was still incomplete: the client sent the headers, and possibly part of the body, then + * either stalled or kept sending too slowly to finish in time. * * This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than `idleTimeout`. * @@ -155,8 +157,10 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } + // tracks request body completion, so that an exceeded requestTimeout can be reported as 408 rather than 503 when it's the client + // which stalled mid-upload; has to come before HttpStreamsServerHandler if (cfg.requestTimeout.isDefined) { - pipeline.addLast(new RequestBodyCompletedTracker) + pipeline.addLast(new RequestBodyCompletionTracker) } pipeline.addLast(new HttpStreamsServerHandler()) pipeline.addLast(handler) 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 37baf54255..2cc89d0009 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 @@ -14,8 +14,14 @@ import sttp.model.StatusCode import sttp.monad.MonadError import sttp.monad.syntax._ import sttp.tapir.server.model.ServerResponse -import sttp.tapir.server.netty.NettyResponseContent._ -import sttp.tapir.server.netty.internal.RequestBodyCompletedTracker._ +import sttp.tapir.server.netty.NettyResponseContent.{ + ByteBufNettyResponseContent, + ChunkedFileNettyResponseContent, + ChunkedStreamNettyResponseContent, + ReactivePublisherNettyResponseContent, + ReactiveWebSocketProcessorNettyResponseContent +} +import sttp.tapir.server.netty.internal.RequestBodyCompletionTracker.wasRequestBodyFullyReceived import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} @@ -102,26 +108,28 @@ class NettyServerHandler[F[_]]( val _ = ctx.writeAndFlush(res).addListener(ChannelFutureListener.CLOSE) } + private def handleRequestTimeout(ctx: ChannelHandlerContext): Unit = { + val timeoutDescription = config.requestTimeout.map(_.toString).getOrElse("(not set)") + if (wasRequestBodyFullyReceived(ctx)) { + logger.error(s"Closing connection due to exceeded response timeout of $timeoutDescription") + writeErrorThenClose(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE) + } else { + logger.debug(s"Closing connection: the request body was 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)")}" - ) - writeErrorThenClose(ctx, HttpResponseStatus.SERVICE_UNAVAILABLE) - } - if (e.state == IdleState.READER_IDLE && !wasRequestBodyFullyReceived(ctx)) { - logger.debug( - s"Closing connection: the request body was not fully received within the request timeout of ${config.requestTimeout.map(_.toString).getOrElse("(not set)")}" - ) - writeErrorThenClose(ctx, HttpResponseStatus.REQUEST_TIMEOUT) - } - 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 => 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) } } @@ -139,7 +147,7 @@ class NettyServerHandler[F[_]]( def runRoute(req: HttpRequest, releaseReq: () => Any = () => ()): Unit = { val requestTimeoutHandler = config.requestTimeout.map { requestTimeout => - new IdleStateHandler(requestTimeout.toMillis.toInt, 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 { () => @@ -356,7 +364,7 @@ class NettyServerHandler[F[_]]( handshakeReq: HttpRequest ) = { ctx.pipeline().remove(this) - Option(ctx.pipeline().get(classOf[RequestBodyCompletedTracker])).foreach(ctx.pipeline().remove) + Option(ctx.pipeline().get(classOf[RequestBodyCompletionTracker])).foreach(ctx.pipeline().remove) ctx .pipeline() .addAfter( diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala deleted file mode 100644 index 35b7866067..0000000000 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletedTracker.scala +++ /dev/null @@ -1,29 +0,0 @@ -package sttp.tapir.server.netty.internal - -import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} -import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} - -/** Has to be included in the pipeline before HttpStreamsServerHandler to observe LastHttpContent and HttpRequest - */ -object RequestBodyCompletedTracker { - - def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = - ctx.pipeline().get(classOf[RequestBodyCompletedTracker]) match { - case t: RequestBodyCompletedTracker => t.bodyFullyReceived - case _ => true - } -} - -class RequestBodyCompletedTracker extends ChannelInboundHandlerAdapter { - - private[internal] var bodyFullyReceived: Boolean = true - - override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { - msg match { - case _: LastHttpContent => bodyFullyReceived = true - case _: HttpRequest => bodyFullyReceived = false - case _ => () - } - val _ = ctx.fireChannelRead(msg) - } -} 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..16c69adfad --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala @@ -0,0 +1,31 @@ +package sttp.tapir.server.netty.internal + +import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} +import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} + +object RequestBodyCompletionTracker { + def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { + val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) + tracker == null || tracker.bodyFullyReceived + } +} + +/** Tracks whether the body of the request currently being handled has been received in full, so that a request timeout can tell a client + * which stalled mid-upload (408) from server logic which is too slow to respond (503). + * + * Has to be included in the pipeline after the HTTP codec and before `HttpStreamsServerHandler`, which replaces the individual + * [[HttpRequest]] / [[LastHttpContent]] messages this relies on with a single streamed request. + */ +class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { + + private[internal] var bodyFullyReceived: Boolean = true + + override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { + 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 8fd60e921d..f3db032b03 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 @@ -2,7 +2,6 @@ package sttp.tapir.server.netty import sttp.tapir._ import sttp.tapir.tests.Test - import scala.concurrent.Future import java.util.concurrent.atomic.AtomicInteger import scala.concurrent.duration.DurationInt @@ -12,7 +11,6 @@ import sttp.tapir.server.metrics.EndpointMetric import io.netty.channel.EventLoopGroup import cats.effect.IO import cats.effect.kernel.Resource - import scala.concurrent.ExecutionContext import sttp.client4._ import sttp.capabilities.fs2.Fs2Streams @@ -21,8 +19,9 @@ import org.scalatest.concurrent.Eventually.eventually import org.scalatest.matchers.should.Matchers._ import cats.effect.unsafe.implicits.global import sttp.model.StatusCode - +import java.io.OutputStream import java.net.Socket +import scala.concurrent.duration.FiniteDuration class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: WebSocketStreamBackend[IO, Fs2Streams[IO]])(implicit ec: ExecutionContext @@ -90,44 +89,99 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } .unsafeToFuture() }, - Test("respond with status 408 when not all declared bytes are received within time window") { + Test(s"respond with status 408 when not all declared body bytes are received, in a single write") { + val requestTimeout = 500.millis + val pauseBeforeBodyFragment = requestTimeout / 2 + val dribbleInterval = requestTimeout / 3 + + def requestHead(port: Int): Array[Byte] = + s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes + + val bodyFragment: Array[Byte] = "test".getBytes + val e = endpoint.put .in(stringBody) .out(stringBody) - .serverLogicSuccess[Future] { body => - Future.successful(body) - } - - val config: NettyConfig = NettyConfig.default.randomPort.requestTimeout(500.millis) + .serverLogicSuccess[Future](body => Future.successful(body)) - val bind = IO.fromFuture(IO.delay(NettyFutureServer(config).addEndpoints(List(e)).start())) + val serverConfig = NettyConfig.default + .eventLoopGroup(eventLoopGroup) + .randomPort + .withDontShutdownEventLoopGroupOnClose + .noGracefulShutdown + .requestTimeout(requestTimeout) - val createSocket: Int => Socket = port => { - val s = new Socket("localhost", port) - s.setSoTimeout(1000) - s - } + 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 => - val bytes = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\ntest".getBytes - - Resource - .make(IO(createSocket(port)))(socket => IO(socket.close())) - .use { socket => - for { - _ <- IO(socket.getOutputStream.write(bytes)) - _ <- IO(socket.getOutputStream.flush()) - response <- IO(new String(socket.getInputStream.readAllBytes())) - } yield { - response should include("408 Request Timeout") + Resource.fromAutoCloseable(IO(clientSocket(port, requestTimeout))).use { socket => + (for { + _ <- IO.blocking(socket.getOutputStream.write(requestHead(port) ++ bodyFragment)) + _ <- IO.blocking(socket.getOutputStream.flush()) + resp <- IO.blocking(new String(socket.getInputStream.readAllBytes())) + } yield resp) + .map { response => + // asserting on the status line, rather than just containment, also covers a second response written behind the first + response should startWith("HTTP/1.1 408 Request Timeout") response should not include ("503") } + } + } + .unsafeToFuture() + }, + Test( + s"respond with status 408 when not all declared body bytes are received, with the body fragment in a separate write, then a stall" + ) { + val requestTimeout = 500.millis + val dribbleInterval = requestTimeout / 3 + + def requestHead(port: Int): Array[Byte] = + s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes + + val bodyFragment: Array[Byte] = "test".getBytes + + val e = endpoint.put + .in(stringBody) + .out(stringBody) + .serverLogicSuccess[Future](body => Future.successful(body)) + + val serverConfig = NettyConfig.default + .eventLoopGroup(eventLoopGroup) + .randomPort + .withDontShutdownEventLoopGroupOnClose + .noGracefulShutdown + .requestTimeout(requestTimeout) + + 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, requestTimeout))).use { socket => + (for { + _ <- IO.blocking(socket.getOutputStream.write(requestHead(port))) + _ <- IO.blocking(socket.getOutputStream.flush()) + _ <- IO.blocking(socket.getOutputStream.write(bodyFragment)) + _ <- IO.blocking(socket.getOutputStream.flush()) + _ <- IO.sleep(dribbleInterval) + response <- IO.blocking(new String(socket.getInputStream.readAllBytes())) + } yield response).map { response => + response should startWith("HTTP/1.1 408 Request Timeout") + response should not include ("503") } + } } .unsafeToFuture() } ) + + private def clientSocket(port: Int, requestTimeout: FiniteDuration): Socket = { + val socket = new Socket("localhost", port) + socket.setSoTimeout((requestTimeout * 20).toMillis.toInt) + socket + } } From c6332a94b8c32acf1b1d146e046625e9435b2c7f Mon Sep 17 00:00:00 2001 From: rwalerow Date: Tue, 1 Sep 2026 16:18:00 +0200 Subject: [PATCH 32/41] Remove unused variables and imports --- .../tapir/server/netty/internal/NettyServerHandler.scala | 2 +- .../server/netty/NettyFutureRequestTimeoutTests.scala | 7 ++----- 2 files changed, 3 insertions(+), 6 deletions(-) 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 2cc89d0009..28701ff8a6 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 @@ -21,7 +21,7 @@ import sttp.tapir.server.netty.NettyResponseContent.{ ReactivePublisherNettyResponseContent, ReactiveWebSocketProcessorNettyResponseContent } -import sttp.tapir.server.netty.internal.RequestBodyCompletionTracker.wasRequestBodyFullyReceived +import RequestBodyCompletionTracker.wasRequestBodyFullyReceived import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} 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 f3db032b03..7b423c146b 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 @@ -19,7 +19,6 @@ import org.scalatest.concurrent.Eventually.eventually import org.scalatest.matchers.should.Matchers._ import cats.effect.unsafe.implicits.global import sttp.model.StatusCode -import java.io.OutputStream import java.net.Socket import scala.concurrent.duration.FiniteDuration @@ -89,10 +88,8 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } .unsafeToFuture() }, - Test(s"respond with status 408 when not all declared body bytes are received, in a single write") { + Test("respond with status 408 when not all declared body bytes are received, in a single write") { val requestTimeout = 500.millis - val pauseBeforeBodyFragment = requestTimeout / 2 - val dribbleInterval = requestTimeout / 3 def requestHead(port: Int): Array[Byte] = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes @@ -133,7 +130,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We .unsafeToFuture() }, Test( - s"respond with status 408 when not all declared body bytes are received, with the body fragment in a separate write, then a stall" + "respond with status 408 when not all declared body bytes are received, with the body fragment in a separate write, then a stall" ) { val requestTimeout = 500.millis val dribbleInterval = requestTimeout / 3 From 1db0ff07ad7c8d4729df397c50fcaf2274e60be7 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Tue, 1 Sep 2026 17:07:02 +0200 Subject: [PATCH 33/41] Additional comments and removed unnecessary sleep in test --- .../main/scala/sttp/tapir/server/netty/NettyConfig.scala | 7 +++---- .../netty/internal/RequestBodyCompletionTracker.scala | 3 +++ .../server/netty/NettyFutureRequestTimeoutTests.scala | 1 - 3 files changed, 6 insertions(+), 5 deletions(-) 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 39e9c5f1ed..a2c51b8eb7 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 @@ -33,15 +33,14 @@ import scala.concurrent.duration._ * * @param requestTimeout * The maximum duration between receiving the request headers and producing a response, which therefore also bounds the time the client - * has to send the request body. When exceeded, an empty error response with a `Connection: close` header is sent, and the connection is - * closed. The status code is decided by whether the request body had been received in full when the timeout expired: + * has to send the request body. This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than + * `idleTimeout`. When exceeded, an empty error response with a `Connection: close` header is sent, and the connection is closed. The + * status code is decided by whether the request body had been received in full when the timeout expired: * * - `503 Service Unavailable`, if the request had been received in full, but no part of a response was produced in time. * - `408 Request Timeout`, if the request body was still incomplete: the client sent the headers, and possibly part of the body, then * either stalled or kept sending too slowly to finish in time. * - * This timeout is 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/RequestBodyCompletionTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala index 16c69adfad..aa5742df37 100644 --- 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 @@ -18,6 +18,9 @@ object RequestBodyCompletionTracker { */ class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { + /** A plain var, as it's only ever touched on the channel's event loop: written by channelRead below, read (through + * wasRequestBodyFullyReceived) by NettyServerHandler.userEventTriggered when the request timeout fires. + */ private[internal] var bodyFullyReceived: Boolean = true override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { 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 7b423c146b..6298b65f35 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 @@ -164,7 +164,6 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We _ <- IO.blocking(socket.getOutputStream.flush()) _ <- IO.blocking(socket.getOutputStream.write(bodyFragment)) _ <- IO.blocking(socket.getOutputStream.flush()) - _ <- IO.sleep(dribbleInterval) response <- IO.blocking(new String(socket.getInputStream.readAllBytes())) } yield response).map { response => response should startWith("HTTP/1.1 408 Request Timeout") From 67dc5dc4c7ea37736b1dfdb532280b5cef80810b Mon Sep 17 00:00:00 2001 From: rwalerow Date: Tue, 1 Sep 2026 17:30:03 +0200 Subject: [PATCH 34/41] Remove unused variable --- .../sttp/tapir/server/netty/NettyFutureRequestTimeoutTests.scala | 1 - 1 file changed, 1 deletion(-) 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 6298b65f35..d2a72c31d0 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 @@ -133,7 +133,6 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We "respond with status 408 when not all declared body bytes are received, with the body fragment in a separate write, then a stall" ) { val requestTimeout = 500.millis - val dribbleInterval = requestTimeout / 3 def requestHead(port: Int): Array[Byte] = s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes From 5345881912427ea4281d8ff2a35bc52af968ee14 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 2 Sep 2026 10:40:58 +0200 Subject: [PATCH 35/41] Fixes after a next round --- doc/server/netty.md | 35 ++--- .../sttp/tapir/server/netty/NettyConfig.scala | 20 +-- .../netty/internal/NettyServerHandler.scala | 12 +- .../RequestBodyCompletionTracker.scala | 1 + .../NettyFutureRequestTimeoutTests.scala | 143 ++++++++---------- 5 files changed, 92 insertions(+), 119 deletions(-) diff --git a/doc/server/netty.md b/doc/server/netty.md index f4841f9935..35405c8269 100644 --- a/doc/server/netty.md +++ b/doc/server/netty.md @@ -100,9 +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: bounds the time between receiving the request headers and - producing a response, and so also bounds how long a client may take to upload - a request body; see [request timeout](#request-timeout) below +* request timeout, see [request timeout](#request-timeout) below * connection timeout * linger timeout * graceful shutdown timeout: when stopped e.g. using @@ -125,28 +123,15 @@ val config = NettyConfig.default.requestTimeout(5.seconds) ### Request timeout -The request timeout starts when the request headers are received, and is only -satisfied once a response starts being written. When it is exceeded, the server -sends an empty response with a `Connection: close` header and closes the -connection. The status code says which side ran out of time, which is decided by -whether the request body had been received in full when the timeout expired: - -* `503 Service Unavailable`, if the request had been received in full, but no - part of a response was produced in time — the endpoint's logic is too slow. -* `408 Request Timeout`, if the request body was still incomplete — the client - sent the headers, and possibly part of the body, then either stalled or kept - sending too slowly to finish in time. A client declaring - `Content-Length: 10000` and then sending only the first few bytes is answered - this way. - -Because the timeout also covers receiving the body, it has to be higher than the -longest upload you want to accept, and lower than `idleTimeout`. It is ignored -for Web Sockets, once the handshake has been established. - -Telling the two cases apart relies on a handler which -`NettyConfig.defaultInitPipeline` adds when `requestTimeout` is set. A custom -`initPipeline` which does not add it still gets the timeout, but always reports -it as `503`; see the `initPipeline` scaladoc for the placement requirement. +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. + +The `408` relies on a handler added by `NettyConfig.defaultInitPipeline`; a +custom `initPipeline` which omits it always reports `503`. ## Web sockets 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 a2c51b8eb7..c7be5a9f48 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 @@ -27,19 +27,14 @@ import scala.concurrent.duration._ * * @param initPipeline * The function to create the Netty pipeline, using the configuration instance, the pipeline created so far, and the handler which - * contains tapir's server processing logic. A custom pipeline which omits the handlers added by [[NettyConfig.defaultInitPipeline]] also - * gives up the behaviour which depends on them; in particular, without a `RequestBodyCompletionTracker` placed after the HTTP codec and - * before `HttpStreamsServerHandler`, an exceeded `requestTimeout` is always reported as `503`, never as `408`. + * contains tapir's server processing logic. A custom pipeline which omits `RequestBodyCompletionTracker` reports an exceeded + * `requestTimeout` as `503`, never as `408`. * * @param requestTimeout - * The maximum duration between receiving the request headers and producing a response, which therefore also bounds the time the client - * has to send the request body. This timeout is ignored in Web Sockets (after a handshake is established). Make sure it's lower than - * `idleTimeout`. When exceeded, an empty error response with a `Connection: close` header is sent, and the connection is closed. The - * status code is decided by whether the request body had been received in full when the timeout expired: - * - * - `503 Service Unavailable`, if the request had been received in full, but no part of a response was produced in time. - * - `408 Request Timeout`, if the request body was still incomplete: the client sent the headers, and possibly part of the body, then - * either stalled or kept sending too slowly to finish in time. + * 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 a `Connection: close` header is sent and the connection is closed - `503` if the + * request had been 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. @@ -156,8 +151,7 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - // tracks request body completion, so that an exceeded requestTimeout can be reported as 408 rather than 503 when it's the client - // which stalled mid-upload; has to come before HttpStreamsServerHandler + // has to come before HttpStreamsServerHandler, see RequestBodyCompletionTracker if (cfg.requestTimeout.isDefined) { pipeline.addLast(new RequestBodyCompletionTracker) } 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 28701ff8a6..0e0bc528eb 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 @@ -69,6 +69,10 @@ class NettyServerHandler[F[_]]( // if the connection gets closed. private[this] val pendingResponses = MutableQueue.empty[() => Future[Unit]] + // Guards against `IdleStateHandler` re-firing `WRITER_IDLE` every `requestTimeout` until a write completes: only the first request + // timeout is answered, the connection is closed afterwards. A plain var, as it's only touched on the channel's event loop. + private[this] var requestTimeoutHandled = false + private val logger = LoggerFactory.getLogger(getClass.getName) private final val WebSocketAutoPingHandlerName = "wsAutoPingHandler" @@ -86,7 +90,7 @@ 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)) } // 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. @@ -109,6 +113,7 @@ class NettyServerHandler[F[_]]( } private def handleRequestTimeout(ctx: ChannelHandlerContext): Unit = { + requestTimeoutHandled = true val timeoutDescription = config.requestTimeout.map(_.toString).getOrElse("(not set)") if (wasRequestBodyFullyReceived(ctx)) { logger.error(s"Closing connection due to exceeded response timeout of $timeoutDescription") @@ -123,8 +128,9 @@ class NettyServerHandler[F[_]]( evt match { case e: IdleStateEvent => e.state() match { - case IdleState.WRITER_IDLE => handleRequestTimeout(ctx) - case IdleState.ALL_IDLE => + case IdleState.WRITER_IDLE if !requestTimeoutHandled => + 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 _ => () 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 index aa5742df37..c8c36301a8 100644 --- 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 @@ -24,6 +24,7 @@ class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { private[internal] var bodyFullyReceived: Boolean = true override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { + // the order of the cases matters: FullHttpRequest is both an HttpRequest and a LastHttpContent, and has to match the latter msg match { case _: LastHttpContent => bodyFullyReceived = true case _: HttpRequest => bodyFullyReceived = false 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 d2a72c31d0..32c1d6a646 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 @@ -88,91 +88,78 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } .unsafeToFuture() }, - Test("respond with status 408 when not all declared body bytes are received, in a single write") { - val requestTimeout = 500.millis - - def requestHead(port: Int): Array[Byte] = - s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes - - val bodyFragment: Array[Byte] = "test".getBytes - - val e = endpoint.put - .in(stringBody) - .out(stringBody) - .serverLogicSuccess[Future](body => Future.successful(body)) - - val serverConfig = NettyConfig.default - .eventLoopGroup(eventLoopGroup) - .randomPort - .withDontShutdownEventLoopGroupOnClose - .noGracefulShutdown - .requestTimeout(requestTimeout) - - 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, requestTimeout))).use { socket => - (for { - _ <- IO.blocking(socket.getOutputStream.write(requestHead(port) ++ bodyFragment)) - _ <- IO.blocking(socket.getOutputStream.flush()) - resp <- IO.blocking(new String(socket.getInputStream.readAllBytes())) - } yield resp) - .map { response => - // asserting on the status line, rather than just containment, also covers a second response written behind the first - response should startWith("HTTP/1.1 408 Request Timeout") - response should not include ("503") - } - } - } - .unsafeToFuture() + Test("respond with status 408 when not all declared body bytes are received, body fragment in a separate read") { + responseToTimingOutRequest { (socket, port) => + for { + _ <- send(socket, requestHead(port)) + // the pause makes the fragment arrive as a separate read, which is what a stalled upload actually looks like + _ <- IO.sleep(incompleteBodyTimeout / 5) + _ <- send(socket, bodyFragment) + } yield () + }.map { response => + // asserting on the status line, rather than just containment, also covers a second response written behind the first + response should startWith("HTTP/1.1 408 Request Timeout") + response should not include ("503") + }.unsafeToFuture() }, - Test( - "respond with status 408 when not all declared body bytes are received, with the body fragment in a separate write, then a stall" - ) { - val requestTimeout = 500.millis - - def requestHead(port: Int): Array[Byte] = - s"PUT / HTTP/1.1\r\nHost: localhost:$port\r\nContent-Type: text/plain\r\nContent-Length: 10000\r\n\r\n".getBytes + Test("respond with status 408 for an incomplete request following a complete one on the same connection") { + responseToTimingOutRequest { (socket, port) => + for { + _ <- send(socket, requestHead(port, bodyFragment.length) ++ bodyFragment) + _ <- IO.sleep(incompleteBodyTimeout / 5) + _ <- send(socket, requestHead(port)) + } yield () + }.map { response => + // the first request is answered normally; the body-completion flag has to be reset for the second one to be reported as 408 + response should startWith("HTTP/1.1 200 OK") + response should include("408 Request Timeout") + }.unsafeToFuture() + } + ) - val bodyFragment: Array[Byte] = "test".getBytes + private val incompleteBodyTimeout = 500.millis - val e = endpoint.put - .in(stringBody) - .out(stringBody) - .serverLogicSuccess[Future](body => Future.successful(body)) + private val bodyFragment: Array[Byte] = "test".getBytes - val serverConfig = NettyConfig.default - .eventLoopGroup(eventLoopGroup) - .randomPort - .withDontShutdownEventLoopGroupOnClose - .noGracefulShutdown - .requestTimeout(requestTimeout) + private def requestHead(port: Int, contentLength: Int = 10000): 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 - val bind = IO.fromFuture(IO.delay(NettyFutureServer(serverConfig).addEndpoints(List(e)).start())) + private def send(socket: Socket, bytes: Array[Byte]): IO[Unit] = + IO.blocking { + socket.getOutputStream.write(bytes) + socket.getOutputStream.flush() + } - Resource - .make(bind)(server => IO.fromFuture(IO.delay(server.stop()))) - .map(_.port) - .use { port => - Resource.fromAutoCloseable(IO(clientSocket(port, requestTimeout))).use { socket => - (for { - _ <- IO.blocking(socket.getOutputStream.write(requestHead(port))) - _ <- IO.blocking(socket.getOutputStream.flush()) - _ <- IO.blocking(socket.getOutputStream.write(bodyFragment)) - _ <- IO.blocking(socket.getOutputStream.flush()) - response <- IO.blocking(new String(socket.getInputStream.readAllBytes())) - } yield response).map { response => - response should startWith("HTTP/1.1 408 Request Timeout") - response should not include ("503") - } - } + /** Starts a server with an echo endpoint and a short request timeout, runs `writeRequest` against it using a plain socket (so that a + * partially sent request can be simulated), and returns everything the server wrote back before closing the connection. + */ + private def responseToTimingOutRequest(writeRequest: (Socket, Int) => IO[Unit]): IO[String] = { + val e = endpoint.put + .in(stringBody) + .out(stringBody) + .serverLogicSuccess[Future](body => Future.successful(body)) + + val serverConfig = NettyConfig.default + .eventLoopGroup(eventLoopGroup) + .randomPort + .withDontShutdownEventLoopGroupOnClose + .noGracefulShutdown + .requestTimeout(incompleteBodyTimeout) + + 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, incompleteBodyTimeout))).use { socket => + for { + _ <- writeRequest(socket, port) + response <- IO.blocking(new String(socket.getInputStream.readAllBytes())) + } yield response } - .unsafeToFuture() - } - ) + } + } private def clientSocket(port: Int, requestTimeout: FiniteDuration): Socket = { val socket = new Socket("localhost", port) From 90b8e9a25ea4e6f2f003e8bc5ea5f340c32aa7a7 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 2 Sep 2026 12:02:35 +0200 Subject: [PATCH 36/41] Review fixes and test refactor - document RequestBodyCompletionTracker's absent-tracker fallback and its initial state; move the companion below the class it describes - make both request-timeout log messages name the same timeout and state the status they send - qualify the tracker reference in NettyConfig's scaladoc, and explain the requestTimeout guard and the Web Socket removal - assert on the exact sequence of response status lines, pin the test charset, and widen the timing margins for slow CI - extract the socket-level test data & helpers into TimingOutRequestSpecData Co-Authored-By: Claude Opus 5 --- .../sttp/tapir/server/netty/NettyConfig.scala | 6 +- .../netty/internal/NettyServerHandler.scala | 16 ++-- .../RequestBodyCompletionTracker.scala | 22 ++++-- .../NettyFutureRequestTimeoutTests.scala | 76 +++---------------- .../netty/TimingOutRequestSpecData.scala | 67 ++++++++++++++++ 5 files changed, 104 insertions(+), 83 deletions(-) create mode 100644 server/netty-server/src/test/scala/sttp/tapir/server/netty/TimingOutRequestSpecData.scala 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 c7be5a9f48..7180ac076c 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 @@ -27,8 +27,8 @@ import scala.concurrent.duration._ * * @param initPipeline * The function to create the Netty pipeline, using the configuration instance, the pipeline created so far, and the handler which - * contains tapir's server processing logic. A custom pipeline which omits `RequestBodyCompletionTracker` reports an exceeded - * `requestTimeout` as `503`, never as `408`. + * contains tapir's server processing logic. A custom pipeline which omits + * [[sttp.tapir.server.netty.internal.RequestBodyCompletionTracker]] reports an exceeded `requestTimeout` as `503`, never as `408`. * * @param requestTimeout * The maximum duration between receiving the request headers and producing a response; it therefore also bounds how long the client has @@ -151,7 +151,7 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - // has to come before HttpStreamsServerHandler, see RequestBodyCompletionTracker + // only of use when a request timeout can fire; has to come before HttpStreamsServerHandler, see RequestBodyCompletionTracker if (cfg.requestTimeout.isDefined) { pipeline.addLast(new RequestBodyCompletionTracker) } 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 0e0bc528eb..32e1e925b6 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 @@ -21,7 +21,6 @@ import sttp.tapir.server.netty.NettyResponseContent.{ ReactivePublisherNettyResponseContent, ReactiveWebSocketProcessorNettyResponseContent } -import RequestBodyCompletionTracker.wasRequestBodyFullyReceived import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} @@ -105,21 +104,20 @@ class NettyServerHandler[F[_]]( } } - def writeErrorThenClose(ctx: ChannelHandlerContext, errorResponseStatus: HttpResponseStatus): Unit = { - val res = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, errorResponseStatus) + 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 = { - requestTimeoutHandled = true val timeoutDescription = config.requestTimeout.map(_.toString).getOrElse("(not set)") - if (wasRequestBodyFullyReceived(ctx)) { - logger.error(s"Closing connection due to exceeded response timeout of $timeoutDescription") + if (RequestBodyCompletionTracker.wasRequestBodyFullyReceived(ctx)) { + 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: the request body was not fully received within the request timeout of $timeoutDescription") + logger.debug(s"Closing connection with 408: request body not fully received within the request timeout of $timeoutDescription") writeErrorThenClose(ctx, HttpResponseStatus.REQUEST_TIMEOUT) } } @@ -129,6 +127,7 @@ class NettyServerHandler[F[_]]( case e: IdleStateEvent => 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)")}") @@ -370,7 +369,8 @@ class NettyServerHandler[F[_]]( handshakeReq: HttpRequest ) = { ctx.pipeline().remove(this) - Option(ctx.pipeline().get(classOf[RequestBodyCompletionTracker])).foreach(ctx.pipeline().remove) + // the HTTP request framing it keys off is gone after the upgrade, so it would only be dead weight in a Web Socket pipeline + Option(ctx.pipeline().get(classOf[RequestBodyCompletionTracker])).foreach(tracker => ctx.pipeline().remove(tracker)) 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 index c8c36301a8..0fb7c42f7e 100644 --- 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 @@ -3,13 +3,6 @@ package sttp.tapir.server.netty.internal import io.netty.channel.{ChannelHandlerContext, ChannelInboundHandlerAdapter} import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} -object RequestBodyCompletionTracker { - def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { - val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) - tracker == null || tracker.bodyFullyReceived - } -} - /** Tracks whether the body of the request currently being handled has been received in full, so that a request timeout can tell a client * which stalled mid-upload (408) from server logic which is too slow to respond (503). * @@ -21,7 +14,7 @@ class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { /** A plain var, as it's only ever touched on the channel's event loop: written by channelRead below, read (through * wasRequestBodyFullyReceived) by NettyServerHandler.userEventTriggered when the request timeout fires. */ - private[internal] var bodyFullyReceived: Boolean = true + private var bodyFullyReceived: Boolean = true override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { // the order of the cases matters: FullHttpRequest is both an HttpRequest and a LastHttpContent, and has to match the latter @@ -33,3 +26,16 @@ class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { val _ = ctx.fireChannelRead(msg) } } + +object RequestBodyCompletionTracker { + + /** Whether the request currently being handled on `ctx`'s channel has been received in full. + * + * Answers `true` if the pipeline has no [[RequestBodyCompletionTracker]] - an unknown state is reported as "received in full", so that a + * timeout is blamed on the server rather than on the client. + */ + def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { + val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) + tracker == null || tracker.bodyFullyReceived + } +} 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 32c1d6a646..b95ffadef2 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 @@ -19,8 +19,6 @@ import org.scalatest.concurrent.Eventually.eventually import org.scalatest.matchers.should.Matchers._ import cats.effect.unsafe.implicits.global import sttp.model.StatusCode -import java.net.Socket -import scala.concurrent.duration.FiniteDuration class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: WebSocketStreamBackend[IO, Fs2Streams[IO]])(implicit ec: ExecutionContext @@ -31,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 @@ -88,82 +89,29 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We } .unsafeToFuture() }, - Test("respond with status 408 when not all declared body bytes are received, body fragment in a separate read") { - responseToTimingOutRequest { (socket, port) => + Test("respond with status 408 when not all declared request body bytes are received") { + statusLinesForTimingOutRequest { (socket, port) => for { _ <- send(socket, requestHead(port)) // the pause makes the fragment arrive as a separate read, which is what a stalled upload actually looks like - _ <- IO.sleep(incompleteBodyTimeout / 5) + _ <- IO.sleep(pauseBetweenWrites) _ <- send(socket, bodyFragment) } yield () - }.map { response => - // asserting on the status line, rather than just containment, also covers a second response written behind the first - response should startWith("HTTP/1.1 408 Request Timeout") - response should not include ("503") + }.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") { - responseToTimingOutRequest { (socket, port) => + statusLinesForTimingOutRequest { (socket, port) => for { _ <- send(socket, requestHead(port, bodyFragment.length) ++ bodyFragment) - _ <- IO.sleep(incompleteBodyTimeout / 5) + _ <- IO.sleep(pauseBetweenWrites) _ <- send(socket, requestHead(port)) } yield () - }.map { response => + }.map { statusLines => // the first request is answered normally; the body-completion flag has to be reset for the second one to be reported as 408 - response should startWith("HTTP/1.1 200 OK") - response should include("408 Request Timeout") + statusLines shouldBe List("HTTP/1.1 200 OK", "HTTP/1.1 408 Request Timeout") }.unsafeToFuture() } ) - - private val incompleteBodyTimeout = 500.millis - - private val bodyFragment: Array[Byte] = "test".getBytes - - private def requestHead(port: Int, contentLength: Int = 10000): 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 - - private def send(socket: Socket, bytes: Array[Byte]): IO[Unit] = - IO.blocking { - socket.getOutputStream.write(bytes) - socket.getOutputStream.flush() - } - - /** Starts a server with an echo endpoint and a short request timeout, runs `writeRequest` against it using a plain socket (so that a - * partially sent request can be simulated), and returns everything the server wrote back before closing the connection. - */ - private def responseToTimingOutRequest(writeRequest: (Socket, Int) => IO[Unit]): IO[String] = { - val e = endpoint.put - .in(stringBody) - .out(stringBody) - .serverLogicSuccess[Future](body => Future.successful(body)) - - val serverConfig = NettyConfig.default - .eventLoopGroup(eventLoopGroup) - .randomPort - .withDontShutdownEventLoopGroupOnClose - .noGracefulShutdown - .requestTimeout(incompleteBodyTimeout) - - 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, incompleteBodyTimeout))).use { socket => - for { - _ <- writeRequest(socket, port) - response <- IO.blocking(new String(socket.getInputStream.readAllBytes())) - } yield response - } - } - } - - private def clientSocket(port: Int, requestTimeout: FiniteDuration): Socket = { - val socket = new Socket("localhost", port) - socket.setSoTimeout((requestTimeout * 20).toMillis.toInt) - socket - } } 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..3595abadd5 --- /dev/null +++ b/server/netty-server/src/test/scala/sttp/tapir/server/netty/TimingOutRequestSpecData.scala @@ -0,0 +1,67 @@ +package sttp.tapir.server.netty + +import cats.effect.IO +import cats.effect.kernel.Resource +import io.netty.channel.EventLoopGroup +import sttp.tapir._ + +import java.net.Socket +import java.nio.charset.StandardCharsets.US_ASCII +import scala.concurrent.duration.{DurationInt, FiniteDuration} +import scala.concurrent.{ExecutionContext, Future} + +class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: ExecutionContext) { + + private val shortRequestTimeout = 1.second + + val pauseBetweenWrites: FiniteDuration = shortRequestTimeout / 10 + + val bodyFragment: Array[Byte] = "test".getBytes(US_ASCII) + + private val StatusLine = """HTTP/1\.1 \d{3} [^\r\n]*""".r + + def requestHead(port: Int, contentLength: Int = 10000): 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 send(socket: Socket, bytes: Array[Byte]): IO[Unit] = + IO.blocking { + socket.getOutputStream.write(bytes) + socket.getOutputStream.flush() + } + + def statusLinesForTimingOutRequest(writeRequest: (Socket, Int) => IO[Unit]): IO[List[String]] = { + val e = endpoint.put + .in(stringBody) + .out(stringBody) + .serverLogicSuccess[Future](body => 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 => + for { + _ <- writeRequest(socket, port) + written <- IO.blocking(new String(socket.getInputStream.readAllBytes(), US_ASCII)) + } yield StatusLine.findAllIn(written).toList + } + } + } + + private def clientSocket(port: Int): Socket = { + val socket = new Socket("localhost", port) + // only reached if the server neither responds nor closes the connection, i.e. if the request timeout didn't fire at all + socket.setSoTimeout((shortRequestTimeout * 20).toMillis.toInt) + socket + } +} From a5952ef18b7bad22762fddaf01d21b6c6aa61200 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 2 Sep 2026 14:05:34 +0200 Subject: [PATCH 37/41] Extend comment --- .../server/netty/internal/RequestBodyCompletionTracker.scala | 2 ++ 1 file changed, 2 insertions(+) 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 index 0fb7c42f7e..43578e3115 100644 --- 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 @@ -8,6 +8,8 @@ import io.netty.handler.codec.http.{HttpRequest, LastHttpContent} * * Has to be included in the pipeline after the HTTP codec and before `HttpStreamsServerHandler`, which replaces the individual * [[HttpRequest]] / [[LastHttpContent]] messages this relies on with a single streamed request. + * + * Tracked per connection, not per request: headers of a following request reset it, so a slow handler can report 408 instead of 503. */ class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { From 4e22608844df96677a4a726dc61adf31c0bf9015 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 2 Sep 2026 17:00:36 +0200 Subject: [PATCH 38/41] Extend documentation --- doc/server/netty.md | 7 ++- .../sttp/tapir/server/netty/NettyConfig.scala | 12 +++--- .../netty/RequestBodyCompletionTracker.scala | 41 ++++++++++++++++++ .../netty/internal/NettyServerHandler.scala | 7 ++- .../RequestBodyCompletionTracker.scala | 43 ------------------- .../NettyFutureRequestTimeoutTests.scala | 3 +- .../netty/TimingOutRequestSpecData.scala | 1 - 7 files changed, 56 insertions(+), 58 deletions(-) create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala delete mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala diff --git a/doc/server/netty.md b/doc/server/netty.md index 35405c8269..b1b97ec5cc 100644 --- a/doc/server/netty.md +++ b/doc/server/netty.md @@ -130,8 +130,11 @@ 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. -The `408` relies on a handler added by `NettyConfig.defaultInitPipeline`; a -custom `initPipeline` which omits it always reports `503`. +The `408` relies on a `RequestBodyCompletionTracker` handler, which +`NettyConfig.defaultInitPipeline` adds whenever a request timeout is set. A +custom `initPipeline` has to add it itself - after the HTTP codec and before +`HttpStreamsServerHandler` - or an exceeded timeout is always reported as +`503`. ## Web sockets 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 7180ac076c..9b5722dfea 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 @@ -27,14 +27,14 @@ import scala.concurrent.duration._ * * @param initPipeline * The function to create the Netty pipeline, using the configuration instance, the pipeline created so far, and the handler which - * contains tapir's server processing logic. A custom pipeline which omits - * [[sttp.tapir.server.netty.internal.RequestBodyCompletionTracker]] reports an exceeded `requestTimeout` as `503`, never as `408`. + * contains tapir's server processing logic. A pipeline without [[RequestBodyCompletionTracker]] reports an exceeded `requestTimeout` as + * `503`, never `408`. * * @param requestTimeout * 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 a `Connection: close` header is sent and the connection is closed - `503` if the - * request had been 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`. + * 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. @@ -151,7 +151,7 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - // only of use when a request timeout can fire; has to come before HttpStreamsServerHandler, see RequestBodyCompletionTracker + // only read when a request timeout fires; placement is significant, see RequestBodyCompletionTracker if (cfg.requestTimeout.isDefined) { pipeline.addLast(new RequestBodyCompletionTracker) } diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala new file mode 100644 index 0000000000..5ba98c1c14 --- /dev/null +++ b/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala @@ -0,0 +1,41 @@ +package sttp.tapir.server.netty + +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). + * + * Belongs after the HTTP codec and before `HttpStreamsServerHandler`, which replaces the [[HttpRequest]] / [[LastHttpContent]] messages + * this relies on with a single streamed request. [[NettyConfig.defaultInitPipeline]] adds it whenever `requestTimeout` is set; a custom + * [[NettyConfig.initPipeline]] has to add it itself. + * + * 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. + */ +class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { + + private var bodyFullyReceived: Boolean = true + + 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) + } +} + +object RequestBodyCompletionTracker { + + /** Whether the request currently being handled on `ctx`'s channel has been received in full. `true` if the pipeline has no + * [[RequestBodyCompletionTracker]]: an unknown state is blamed on the server rather than on the client. + */ + private[netty] def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { + val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) + tracker == null || tracker.bodyFullyReceived + } +} 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 32e1e925b6..5d5e32418d 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 @@ -23,7 +23,7 @@ import sttp.tapir.server.netty.NettyResponseContent.{ } import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} -import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} +import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, RequestBodyCompletionTracker, Route} import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean @@ -68,8 +68,8 @@ class NettyServerHandler[F[_]]( // if the connection gets closed. private[this] val pendingResponses = MutableQueue.empty[() => Future[Unit]] - // Guards against `IdleStateHandler` re-firing `WRITER_IDLE` every `requestTimeout` until a write completes: only the first request - // timeout is answered, the connection is closed afterwards. A plain var, as it's only touched on the channel's event loop. + // `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 private val logger = LoggerFactory.getLogger(getClass.getName) @@ -369,7 +369,6 @@ class NettyServerHandler[F[_]]( handshakeReq: HttpRequest ) = { ctx.pipeline().remove(this) - // the HTTP request framing it keys off is gone after the upgrade, so it would only be dead weight in a Web Socket pipeline Option(ctx.pipeline().get(classOf[RequestBodyCompletionTracker])).foreach(tracker => ctx.pipeline().remove(tracker)) ctx .pipeline() 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 deleted file mode 100644 index 43578e3115..0000000000 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala +++ /dev/null @@ -1,43 +0,0 @@ -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 request timeout can tell a client - * which stalled mid-upload (408) from server logic which is too slow to respond (503). - * - * Has to be included in the pipeline after the HTTP codec and before `HttpStreamsServerHandler`, which replaces the individual - * [[HttpRequest]] / [[LastHttpContent]] messages this relies on with a single streamed request. - * - * Tracked per connection, not per request: headers of a following request reset it, so a slow handler can report 408 instead of 503. - */ -class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { - - /** A plain var, as it's only ever touched on the channel's event loop: written by channelRead below, read (through - * wasRequestBodyFullyReceived) by NettyServerHandler.userEventTriggered when the request timeout fires. - */ - private var bodyFullyReceived: Boolean = true - - override def channelRead(ctx: ChannelHandlerContext, msg: Any): Unit = { - // the order of the cases matters: FullHttpRequest is both an HttpRequest and a LastHttpContent, and has to match the latter - msg match { - case _: LastHttpContent => bodyFullyReceived = true - case _: HttpRequest => bodyFullyReceived = false - case _ => () - } - val _ = ctx.fireChannelRead(msg) - } -} - -object RequestBodyCompletionTracker { - - /** Whether the request currently being handled on `ctx`'s channel has been received in full. - * - * Answers `true` if the pipeline has no [[RequestBodyCompletionTracker]] - an unknown state is reported as "received in full", so that a - * timeout is blamed on the server rather than on the client. - */ - def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { - val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) - tracker == null || tracker.bodyFullyReceived - } -} 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 b95ffadef2..a942e9f97b 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 @@ -93,7 +93,7 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We statusLinesForTimingOutRequest { (socket, port) => for { _ <- send(socket, requestHead(port)) - // the pause makes the fragment arrive as a separate read, which is what a stalled upload actually looks like + // the pause makes the fragment arrive as its own read, as a stalled upload would _ <- IO.sleep(pauseBetweenWrites) _ <- send(socket, bodyFragment) } yield () @@ -109,7 +109,6 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We _ <- send(socket, requestHead(port)) } yield () }.map { statusLines => - // the first request is answered normally; the body-completion flag has to be reset for the second one to be reported as 408 statusLines shouldBe List("HTTP/1.1 200 OK", "HTTP/1.1 408 Request Timeout") }.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 index 3595abadd5..d24c1ab6dd 100644 --- 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 @@ -60,7 +60,6 @@ class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: Exec private def clientSocket(port: Int): Socket = { val socket = new Socket("localhost", port) - // only reached if the server neither responds nor closes the connection, i.e. if the request timeout didn't fire at all socket.setSoTimeout((shortRequestTimeout * 20).toMillis.toInt) socket } From fbb3b6487da54645433db398d6b8e3752e2a0d32 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 23 Sep 2026 14:15:04 +0200 Subject: [PATCH 39/41] Address latest CR issues --- .../NettyFutureRequestTimeoutTests.scala | 33 ++++++++----- .../netty/TimingOutRequestSpecData.scala | 46 +++++++++++++------ 2 files changed, 55 insertions(+), 24 deletions(-) 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 a942e9f97b..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 @@ -90,27 +90,38 @@ class NettyFutureRequestTimeoutTests(eventLoopGroup: EventLoopGroup, backend: We .unsafeToFuture() }, Test("respond with status 408 when not all declared request body bytes are received") { - statusLinesForTimingOutRequest { (socket, port) => + statusLinesFromShortTimeoutServer { (socket, port) => for { - _ <- send(socket, requestHead(port)) - // the pause makes the fragment arrive as its own read, as a stalled upload would - _ <- IO.sleep(pauseBetweenWrites) - _ <- send(socket, bodyFragment) - } yield () + _ <- 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") { - statusLinesForTimingOutRequest { (socket, port) => + statusLinesFromShortTimeoutServer { (socket, port) => for { - _ <- send(socket, requestHead(port, bodyFragment.length) ++ bodyFragment) - _ <- IO.sleep(pauseBetweenWrites) - _ <- send(socket, requestHead(port)) - } yield () + _ <- 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 index d24c1ab6dd..803debbfd9 100644 --- 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 @@ -5,36 +5,57 @@ 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, FiniteDuration} +import scala.concurrent.duration.DurationInt import scala.concurrent.{ExecutionContext, Future} class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: ExecutionContext) { private val shortRequestTimeout = 1.second - val pauseBetweenWrites: FiniteDuration = shortRequestTimeout / 10 + val completeBody: Array[Byte] = "test".getBytes(US_ASCII) - val bodyFragment: Array[Byte] = "test".getBytes(US_ASCII) + val slowBody: Array[Byte] = "slow".getBytes(US_ASCII) + private val slowBodyString = new String(slowBody, US_ASCII) - private val StatusLine = """HTTP/1\.1 \d{3} [^\r\n]*""".r - - def requestHead(port: Int, contentLength: Int = 10000): Array[Byte] = + 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 statusLinesForTimingOutRequest(writeRequest: (Socket, Int) => IO[Unit]): IO[List[String]] = { + 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() // the chunk's data + chunkSize = in.readLine() + } + in.readLine() // blank line trailing the terminal chunk + } + statusLine + } + + def statusLinesFromShortTimeoutServer(interact: (Socket, Int) => IO[List[String]]): IO[List[String]] = { val e = endpoint.put .in(stringBody) .out(stringBody) - .serverLogicSuccess[Future](body => Future.successful(body)) + .serverLogicSuccess[Future] { body => + if (body == slowBodyString) Thread.sleep((shortRequestTimeout * 2).toMillis) + Future.successful(body) + } val serverConfig = NettyConfig.default .eventLoopGroup(eventLoopGroup) @@ -50,17 +71,16 @@ class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: Exec .map(_.port) .use { port => Resource.fromAutoCloseable(IO(clientSocket(port))).use { socket => - for { - _ <- writeRequest(socket, port) - written <- IO.blocking(new String(socket.getInputStream.readAllBytes(), US_ASCII)) - } yield StatusLine.findAllIn(written).toList + interact(socket, port) } } } + private val socketReadTimeout = shortRequestTimeout * 20 + private def clientSocket(port: Int): Socket = { val socket = new Socket("localhost", port) - socket.setSoTimeout((shortRequestTimeout * 20).toMillis.toInt) + socket.setSoTimeout(socketReadTimeout.toMillis.toInt) socket } } From 39f1495acd43d62ff3b109fa97d113fb57f96ac4 Mon Sep 17 00:00:00 2001 From: rwalerow Date: Wed, 23 Sep 2026 14:40:01 +0200 Subject: [PATCH 40/41] Rearrange variables --- .../server/netty/TimingOutRequestSpecData.scala | 12 +++++------- 1 file changed, 5 insertions(+), 7 deletions(-) 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 index 803debbfd9..ddf4081e9f 100644 --- 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 @@ -13,11 +13,11 @@ import scala.concurrent.{ExecutionContext, Future} class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: ExecutionContext) { - private val shortRequestTimeout = 1.second - 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] = @@ -40,10 +40,10 @@ class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: Exec if (headers.exists(_.toLowerCase.contains("chunked"))) { var chunkSize = in.readLine() while (chunkSize != "0") { - in.readLine() // the chunk's data + in.readLine() chunkSize = in.readLine() } - in.readLine() // blank line trailing the terminal chunk + in.readLine() } statusLine } @@ -76,8 +76,6 @@ class TimingOutRequestSpecData(eventLoopGroup: EventLoopGroup)(implicit ec: Exec } } - private val socketReadTimeout = shortRequestTimeout * 20 - private def clientSocket(port: Int): Socket = { val socket = new Socket("localhost", port) socket.setSoTimeout(socketReadTimeout.toMillis.toInt) From 5d2f893a6e033832370ef227e59abb5f84377b5b Mon Sep 17 00:00:00 2001 From: Adam Warski Date: Thu, 24 Sep 2026 14:45:26 +0000 Subject: [PATCH 41/41] Netty: install the request body tracker from NettyServerHandler, keep it internal --- doc/server/netty.md | 6 --- .../sttp/tapir/server/netty/NettyConfig.scala | 7 +--- .../netty/RequestBodyCompletionTracker.scala | 41 ------------------- .../netty/internal/NettyServerHandler.scala | 16 ++++++-- .../RequestBodyCompletionTracker.scala | 32 +++++++++++++++ 5 files changed, 46 insertions(+), 56 deletions(-) delete mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala create mode 100644 server/netty-server/src/main/scala/sttp/tapir/server/netty/internal/RequestBodyCompletionTracker.scala diff --git a/doc/server/netty.md b/doc/server/netty.md index b1b97ec5cc..4ef4f94f54 100644 --- a/doc/server/netty.md +++ b/doc/server/netty.md @@ -130,12 +130,6 @@ 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. -The `408` relies on a `RequestBodyCompletionTracker` handler, which -`NettyConfig.defaultInitPipeline` adds whenever a request timeout is set. A -custom `initPipeline` has to add it itself - after the HTTP codec and before -`HttpStreamsServerHandler` - or an exceeded timeout is always reported as -`503`. - ## 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 9b5722dfea..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 @@ -27,8 +27,7 @@ import scala.concurrent.duration._ * * @param initPipeline * The function to create the Netty pipeline, using the configuration instance, the pipeline created so far, and the handler which - * contains tapir's server processing logic. A pipeline without [[RequestBodyCompletionTracker]] reports an exceeded `requestTimeout` as - * `503`, never `408`. + * contains tapir's server processing logic. * * @param requestTimeout * The maximum duration between receiving the request headers and producing a response; it therefore also bounds how long the client has @@ -151,10 +150,6 @@ object NettyConfig { if (cfg.compressionConfig.enabled) { pipeline.addLast(new HttpContentCompressor()) } - // only read when a request timeout fires; placement is significant, see RequestBodyCompletionTracker - if (cfg.requestTimeout.isDefined) { - pipeline.addLast(new RequestBodyCompletionTracker) - } pipeline.addLast(new HttpStreamsServerHandler()) pipeline.addLast(handler) if (cfg.addLoggingHandler) pipeline.addLast(new LoggingHandler()) diff --git a/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala b/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala deleted file mode 100644 index 5ba98c1c14..0000000000 --- a/server/netty-server/src/main/scala/sttp/tapir/server/netty/RequestBodyCompletionTracker.scala +++ /dev/null @@ -1,41 +0,0 @@ -package sttp.tapir.server.netty - -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). - * - * Belongs after the HTTP codec and before `HttpStreamsServerHandler`, which replaces the [[HttpRequest]] / [[LastHttpContent]] messages - * this relies on with a single streamed request. [[NettyConfig.defaultInitPipeline]] adds it whenever `requestTimeout` is set; a custom - * [[NettyConfig.initPipeline]] has to add it itself. - * - * 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. - */ -class RequestBodyCompletionTracker extends ChannelInboundHandlerAdapter { - - private var bodyFullyReceived: Boolean = true - - 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) - } -} - -object RequestBodyCompletionTracker { - - /** Whether the request currently being handled on `ctx`'s channel has been received in full. `true` if the pipeline has no - * [[RequestBodyCompletionTracker]]: an unknown state is blamed on the server rather than on the client. - */ - private[netty] def wasRequestBodyFullyReceived(ctx: ChannelHandlerContext): Boolean = { - val tracker = ctx.pipeline().get(classOf[RequestBodyCompletionTracker]) - tracker == null || tracker.bodyFullyReceived - } -} 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 5d5e32418d..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 @@ -23,7 +23,7 @@ import sttp.tapir.server.netty.NettyResponseContent.{ } import sttp.tapir.server.netty.internal.reactivestreams.{CancellingSubscriber, SubscribeTrackingStreamedHttpRequest} import sttp.tapir.server.netty.internal.ws.{WebSocketAutoPingHandler, WebSocketPingPongFrameHandler} -import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, RequestBodyCompletionTracker, Route} +import sttp.tapir.server.netty.{NettyConfig, NettyResponse, NettyServerRequest, Route} import java.util.concurrent.TimeUnit import java.util.concurrent.atomic.AtomicBoolean @@ -72,8 +72,13 @@ class NettyServerHandler[F[_]]( // 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) { @@ -91,6 +96,11 @@ class NettyServerHandler[F[_]]( config.idleTimeout.foreach { idleTimeout => 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. val _ = ctx.channel.closeFuture.addListener { (_: ChannelFuture) => @@ -113,7 +123,7 @@ class NettyServerHandler[F[_]]( private def handleRequestTimeout(ctx: ChannelHandlerContext): Unit = { val timeoutDescription = config.requestTimeout.map(_.toString).getOrElse("(not set)") - if (RequestBodyCompletionTracker.wasRequestBodyFullyReceived(ctx)) { + 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 { @@ -369,7 +379,7 @@ class NettyServerHandler[F[_]]( handshakeReq: HttpRequest ) = { ctx.pipeline().remove(this) - Option(ctx.pipeline().get(classOf[RequestBodyCompletionTracker])).foreach(tracker => ctx.pipeline().remove(tracker)) + 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) + } +}