From c529c898c5fb944b469fb4de1f9adca02a9adff3 Mon Sep 17 00:00:00 2001 From: davidl Date: Wed, 29 Apr 2026 13:03:41 +0200 Subject: [PATCH 1/3] Cap pre-connect request body buffer to prevent OOM for request streaming (#3173) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR #3174 stopped `AsyncBodyReader` from auto-reading after the consumer connects, but the `State.Buffering` window — used before the route handler first pulls the body — still returned `readMore = true` unconditionally. With request streaming enabled and an async handler, a fast client could fill the heap with `buffer0` bytes during the gap between `channelRead` and the handler calling `connect`. Gate the `Buffering` arm on a configurable cap (`Server.Config.requestBodyPreConnectBufferSize`, default 64 KB). Once the buffer reaches the cap, `ctx.read()` is no longer called; Netty's TCP back-pressure stalls the client until the consumer connects and the post-transition `ctx.read()` in `connect` resumes reading. Only the streaming path is affected — `RequestStreaming.Disabled` uses `HttpObjectAggregator` and never instantiates `AsyncBodyReader`. --- .../zio/http/netty/AsyncBodyReader.scala | 24 +++- .../netty/server/ServerAsyncBodyHandler.scala | 3 +- .../netty/server/ServerInboundHandler.scala | 2 +- .../zio/http/netty/AsyncBodyReaderSpec.scala | 106 ++++++++++++++++++ .../src/main/scala/zio/http/Server.scala | 22 +++- 5 files changed, 150 insertions(+), 7 deletions(-) create mode 100644 zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala diff --git a/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala b/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala index 9d06f20749..5ac956b202 100644 --- a/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala +++ b/zio-http/jvm/src/main/scala/zio/http/netty/AsyncBodyReader.scala @@ -32,12 +32,18 @@ import io.netty.channel.{ChannelHandlerContext, SimpleChannelInboundHandler} import io.netty.handler.codec.http.{HttpContent, LastHttpContent} import io.netty.util.concurrent.ScheduledFuture -private[netty] abstract class AsyncBodyReader(timeoutMillis: Option[Long]) - extends SimpleChannelInboundHandler[HttpContent](true) { self => +private[netty] abstract class AsyncBodyReader( + timeoutMillis: Option[Long], + maxPreConnectBufferSize: Int = AsyncBodyReader.DefaultMaxPreConnectBufferSize, +) extends SimpleChannelInboundHandler[HttpContent](true) { self => import zio.http.netty.AsyncBodyReader._ private var state: State = State.Buffering private val buffer = new mutable.ArrayBuilder.ofByte() + // Tracks `buffer` size while in `State.Buffering` so we can apply back-pressure + // before the consumer connects. `ArrayBuilder#knownSize` is unreliable on + // Scala 2.12, so we count bytes ourselves. + private var bufferedBytes: Int = 0 private var previousAutoRead: Boolean = false private var readingDone: Boolean = false private var ctx: ChannelHandlerContext = _ @@ -114,9 +120,12 @@ private[netty] abstract class AsyncBodyReader(timeoutMillis: Option[Long]) val readMore = state match { case State.Buffering => - // `connect` method hasn't been called yet, add all incoming content to the buffer + // `connect` method hasn't been called yet, add all incoming content to the buffer. + // Cap the pre-connect buffer to avoid unbounded heap growth when a fast producer + // outpaces a slow-to-start consumer (see issue #3173). buffer0.addAll(content) - true + bufferedBytes += content.length + bufferedBytes < maxPreConnectBufferSize case State.Direct(callback) if isLast && buffer0.knownSize == 0 => // Buffer is empty, we can just use the array directly callback(Chunk.fromArray(content), isLast = true) @@ -219,6 +228,13 @@ private[netty] abstract class AsyncBodyReader(timeoutMillis: Option[Long]) private[netty] object AsyncBodyReader { + /** + * Default cap on bytes that may be buffered before the consumer calls + * `connect`. Once the buffer reaches this size we stop calling `ctx.read()`, + * relying on Netty's TCP back-pressure until the consumer drains the buffer. + */ + final val DefaultMaxPreConnectBufferSize: Int = 64 * 1024 + sealed trait State object State { diff --git a/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerAsyncBodyHandler.scala b/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerAsyncBodyHandler.scala index ade061bbea..700b539845 100644 --- a/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerAsyncBodyHandler.scala +++ b/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerAsyncBodyHandler.scala @@ -18,4 +18,5 @@ package zio.http.netty.server import zio.http.netty.AsyncBodyReader -private[zio] final class ServerAsyncBodyHandler extends AsyncBodyReader(timeoutMillis = None) {} +private[zio] final class ServerAsyncBodyHandler(maxPreConnectBufferSize: Int) + extends AsyncBodyReader(timeoutMillis = None, maxPreConnectBufferSize = maxPreConnectBufferSize) {} diff --git a/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala b/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala index 0637b418bc..84f3b0ed6c 100644 --- a/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala +++ b/zio-http/jvm/src/main/scala/zio/http/netty/server/ServerInboundHandler.scala @@ -131,7 +131,7 @@ private[zio] final case class ServerInboundHandler( } private def addAsyncBodyHandler(ctx: ChannelHandlerContext): AsyncBodyReader = { - val handler = new ServerAsyncBodyHandler + val handler = new ServerAsyncBodyHandler(config.requestBodyPreConnectBufferSize) ctx .channel() .pipeline() diff --git a/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala b/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala new file mode 100644 index 0000000000..5db941834e --- /dev/null +++ b/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala @@ -0,0 +1,106 @@ +/* + * Copyright 2021 - 2023 Sporta Technologies PVT LTD & the ZIO HTTP contributors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package zio.http.netty + +import java.util.concurrent.atomic.AtomicInteger + +import zio._ +import zio.test._ + +import zio.http.ZIOHttpSpec + +import io.netty.buffer.Unpooled +import io.netty.channel.embedded.EmbeddedChannel +import io.netty.channel.{ChannelHandlerContext, ChannelOutboundHandlerAdapter} +import io.netty.handler.codec.http.DefaultHttpContent + +/** + * Regression test for the pre-connect Buffering window in `AsyncBodyReader`. + * + * Before any consumer calls `connect`, the reader buffers incoming body chunks + * on the heap and asks Netty for more (`ctx.read()`). Without a cap, a fast + * producer (e.g. a localhost upload) can fill the heap during the few + * milliseconds it takes the route handler to call `request.body.asStream` and + * start pulling. See issue #3173. + * + * `EmbeddedChannel.writeInbound` ignores `autoRead`, so we cannot verify the + * fix by checking that buffered bytes stay below the cap. Instead we count the + * outbound `read()` requests AsyncBodyReader makes — the cap should make those + * calls stop, which is what enforces TCP back-pressure on a real socket. + */ +object AsyncBodyReaderSpec extends ZIOHttpSpec { + + private final class TestReader(maxPreConnectBufferSize: Int) + extends AsyncBodyReader(timeoutMillis = None, maxPreConnectBufferSize = maxPreConnectBufferSize) + + /** Outbound handler placed before AsyncBodyReader to count `ctx.read()`. */ + private final class ReadCounter extends ChannelOutboundHandlerAdapter { + val count = new AtomicInteger(0) + override def read(ctx: ChannelHandlerContext): Unit = { + count.incrementAndGet() + ctx.read(): Unit + } + } + + private def setup(cap: Int): (ReadCounter, EmbeddedChannel) = { + val counter = new ReadCounter + val ch = new EmbeddedChannel(new TestReader(maxPreConnectBufferSize = cap)) + // addFirst places counter at the head of the pipeline, so outbound `read` + // events from AsyncBodyReader pass through it on their way to the channel. + ch.pipeline().addFirst(counter): Unit + (counter, ch) + } + + private def feed(ch: EmbeddedChannel, chunkBytes: Int, count: Int): Unit = { + val payload = new Array[Byte](chunkBytes) + var i = 0 + while (i < count) { + ch.writeInbound(new DefaultHttpContent(Unpooled.wrappedBuffer(payload))): Unit + i += 1 + } + } + + override def spec: Spec[TestEnvironment with Scope, Any] = + suite("AsyncBodyReader")( + test("Buffering state stops requesting reads once pre-connect cap is reached") { + ZIO.attempt { + val cap = 64 * 1024 + val chunkSz = 8 * 1024 + val (counter, ch) = setup(cap) + // Feed 100 chunks of 8 KB = 800 KB, far above the 64 KB cap. + feed(ch, chunkBytes = chunkSz, count = 100) + + // Without the fix, AsyncBodyReader would request a read after every + // chunk (~100 reads). With the cap, it stops once bufferedBytes + // reaches the cap — that's after ceil(cap / chunkSz) = 8 chunks. + val reads = counter.count.get() + assertTrue(reads <= cap / chunkSz, reads < 100) + } + }, + test("Buffering state keeps requesting reads while under cap") { + ZIO.attempt { + val cap = 64 * 1024 + val chunkSz = 1024 + val (counter, ch) = setup(cap) + feed(ch, chunkBytes = chunkSz, count = 4) + // 4 KB buffered, well below 64 KB cap → reader still asking for more. + val reads = counter.count.get() + assertTrue(reads == 4) + } + }, + ) +} diff --git a/zio-http/shared/src/main/scala/zio/http/Server.scala b/zio-http/shared/src/main/scala/zio/http/Server.scala index 6d22498271..1097d183af 100644 --- a/zio-http/shared/src/main/scala/zio/http/Server.scala +++ b/zio-http/shared/src/main/scala/zio/http/Server.scala @@ -76,6 +76,8 @@ object Server extends ServerPlatformSpecific { tcpNoDelay: Boolean, @unroll generateHeadRoutes: Boolean = false, + @unroll + requestBodyPreConnectBufferSize: Int = 64 * 1024, ) { self => /** @@ -197,6 +199,18 @@ object Server extends ServerPlatformSpecific { def requestStreaming(requestStreaming: RequestStreaming): Config = self.copy(requestStreaming = requestStreaming) + /** + * Maximum number of bytes the server will buffer from a streamed request + * body before the route handler starts consuming it. When the buffer + * reaches this size, Netty's TCP back-pressure kicks in and the client is + * paused until the handler calls `request.body.asStream` (or similar) and + * begins reading. Prevents heap exhaustion when an async handler defers + * consuming the body and the client uploads faster than the handler can + * connect. Only applies when [[RequestStreaming.Enabled]] is configured. + */ + def requestBodyPreConnectBufferSize(size: Int): Config = + self.copy(requestBodyPreConnectBufferSize = size) + /** * Sets the maximum number of connection requests that will be queued before * being rejected @@ -232,7 +246,10 @@ object Server extends ServerPlatformSpecific { zio.Config.duration("idle-timeout").optional.withDefault(Config.default.idleTimeout) ++ zio.Config.boolean("avoid-context-switching").withDefault(Config.default.avoidContextSwitching) ++ zio.Config.int("so-backlog").withDefault(Config.default.soBacklog) ++ - zio.Config.boolean("tcp-nodelay").withDefault(Config.default.tcpNoDelay) + zio.Config.boolean("tcp-nodelay").withDefault(Config.default.tcpNoDelay) ++ + zio.Config + .int("request-body-pre-connect-buffer-size") + .withDefault(Config.default.requestBodyPreConnectBufferSize) }.map { case ( @@ -252,6 +269,7 @@ object Server extends ServerPlatformSpecific { avoidCtxSwitch, soBacklog, tcpNoDelay, + requestBodyPreConnectBufferSize, ) => default.copy( sslConfig = sslConfig, @@ -269,6 +287,7 @@ object Server extends ServerPlatformSpecific { avoidContextSwitching = avoidCtxSwitch, soBacklog = soBacklog, tcpNoDelay = tcpNoDelay, + requestBodyPreConnectBufferSize = requestBodyPreConnectBufferSize, ) } @@ -289,6 +308,7 @@ object Server extends ServerPlatformSpecific { avoidContextSwitching = false, soBacklog = 100, tcpNoDelay = true, + requestBodyPreConnectBufferSize = 64 * 1024, ) final case class ResponseCompressionConfig( From 7d69683479870b7d0cc0a16b7ce1fd68475ab859 Mon Sep 17 00:00:00 2001 From: davidl Date: Wed, 29 Apr 2026 16:27:06 +0200 Subject: [PATCH 2/3] Scalafmt --- .../zio/http/netty/AsyncBodyReaderSpec.scala | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala b/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala index 5db941834e..a315bef96b 100644 --- a/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala +++ b/zio-http/jvm/src/test/scala/zio/http/netty/AsyncBodyReaderSpec.scala @@ -49,7 +49,7 @@ object AsyncBodyReaderSpec extends ZIOHttpSpec { /** Outbound handler placed before AsyncBodyReader to count `ctx.read()`. */ private final class ReadCounter extends ChannelOutboundHandlerAdapter { - val count = new AtomicInteger(0) + val count = new AtomicInteger(0) override def read(ctx: ChannelHandlerContext): Unit = { count.incrementAndGet() ctx.read(): Unit @@ -78,9 +78,9 @@ object AsyncBodyReaderSpec extends ZIOHttpSpec { suite("AsyncBodyReader")( test("Buffering state stops requesting reads once pre-connect cap is reached") { ZIO.attempt { - val cap = 64 * 1024 - val chunkSz = 8 * 1024 - val (counter, ch) = setup(cap) + val cap = 64 * 1024 + val chunkSz = 8 * 1024 + val (counter, ch) = setup(cap) // Feed 100 chunks of 8 KB = 800 KB, far above the 64 KB cap. feed(ch, chunkBytes = chunkSz, count = 100) @@ -93,12 +93,12 @@ object AsyncBodyReaderSpec extends ZIOHttpSpec { }, test("Buffering state keeps requesting reads while under cap") { ZIO.attempt { - val cap = 64 * 1024 - val chunkSz = 1024 - val (counter, ch) = setup(cap) + val cap = 64 * 1024 + val chunkSz = 1024 + val (counter, ch) = setup(cap) feed(ch, chunkBytes = chunkSz, count = 4) // 4 KB buffered, well below 64 KB cap → reader still asking for more. - val reads = counter.count.get() + val reads = counter.count.get() assertTrue(reads == 4) } }, From 9dc477f046e1764fbcd98d71eecb2f0cf209c2be Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Sun, 3 May 2026 02:14:15 +0000 Subject: [PATCH 3/3] Update sbt, scripted-plugin to 1.12.11 --- project/build.properties | 2 +- zio-http-example-basic-auth/project/build.properties | 2 +- zio-http-example-cookie-auth/project/build.properties | 2 +- zio-http-example-digest-auth/project/build.properties | 2 +- .../project/build.properties | 2 +- zio-http-example-jwt-bearer-token-auth/project/build.properties | 2 +- .../project/build.properties | 2 +- .../project/build.properties | 2 +- zio-http-example-webauthn/project/build.properties | 2 +- 9 files changed, 9 insertions(+), 9 deletions(-) diff --git a/project/build.properties b/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/project/build.properties +++ b/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-basic-auth/project/build.properties b/zio-http-example-basic-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-basic-auth/project/build.properties +++ b/zio-http-example-basic-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-cookie-auth/project/build.properties b/zio-http-example-cookie-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-cookie-auth/project/build.properties +++ b/zio-http-example-cookie-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-digest-auth/project/build.properties b/zio-http-example-digest-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-digest-auth/project/build.properties +++ b/zio-http-example-digest-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-jwt-bearer-refresh-token-auth/project/build.properties b/zio-http-example-jwt-bearer-refresh-token-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-jwt-bearer-refresh-token-auth/project/build.properties +++ b/zio-http-example-jwt-bearer-refresh-token-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-jwt-bearer-token-auth/project/build.properties b/zio-http-example-jwt-bearer-token-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-jwt-bearer-token-auth/project/build.properties +++ b/zio-http-example-jwt-bearer-token-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-oauth-bearer-token-auth/project/build.properties b/zio-http-example-oauth-bearer-token-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-oauth-bearer-token-auth/project/build.properties +++ b/zio-http-example-oauth-bearer-token-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-opaque-bearer-token-auth/project/build.properties b/zio-http-example-opaque-bearer-token-auth/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-opaque-bearer-token-auth/project/build.properties +++ b/zio-http-example-opaque-bearer-token-auth/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11 diff --git a/zio-http-example-webauthn/project/build.properties b/zio-http-example-webauthn/project/build.properties index df061f4fbf..dabdb15903 100644 --- a/zio-http-example-webauthn/project/build.properties +++ b/zio-http-example-webauthn/project/build.properties @@ -1 +1 @@ -sbt.version=1.12.9 +sbt.version=1.12.11