From 031aafed6ff4e84eff396ae504c827d6c809000a Mon Sep 17 00:00:00 2001 From: davidl Date: Fri, 27 Mar 2026 13:22:11 +0100 Subject: [PATCH 1/3] Fix hanging request streaming under concurrent load --- .../netty/server/ServerInboundHandler.scala | 27 +++- .../RequestStreamingConcurrencySpec.scala | 148 ++++++++++++++++++ 2 files changed, 168 insertions(+), 7 deletions(-) create mode 100644 zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala 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 f8a940d6c5..0637b418bc 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 @@ -196,13 +196,26 @@ private[zio] final case class ServerInboundHandler( ctx.writeAndFlush(jResponse) NettyBodyWriter.writeAndFlush(response.body, contentLength, ctx) match { case Some(bodyTask) => - val channelConfig = ctx.channel().config() - val previousAutoRead = channelConfig.isAutoRead - if (previousAutoRead) channelConfig.setAutoRead(false) - Some(bodyTask.ensuring(ZIO.succeed { - if (previousAutoRead) { - channelConfig.setAutoRead(true) - ctx.channel().read() + // Disable auto-read while the response body is being streamed to prevent + // the next request's headers from being read and interleaved. + // All autoRead manipulation must happen on the event loop thread to avoid + // racing with AsyncBodyReader (used for request streaming) + val channel = ctx.channel() + val previousAutoRead = new java.util.concurrent.atomic.AtomicBoolean(true) + channel + .eventLoop() + .execute(() => { + val prev = channel.config().isAutoRead + previousAutoRead.set(prev) + if (prev) channel.config().setAutoRead(false): Unit + }) + Some(bodyTask.ensuring(ZIO.async[Any, Nothing, Unit] { cb => + channel.eventLoop().execute { () => + if (previousAutoRead.get()) { + channel.config().setAutoRead(true) + channel.read() + } + cb(Exit.unit) } })) case None => None diff --git a/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala b/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala new file mode 100644 index 0000000000..43679d672f --- /dev/null +++ b/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala @@ -0,0 +1,148 @@ +/* + * 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 + +import zio._ +import zio.http.Server.RequestStreaming +import zio.http.netty.NettyConfig +import zio.stream.ZStream +import zio.test.TestAspect._ +import zio.test._ + +import java.util.UUID + +/** + * Regression test for request streaming + concurrent load. + * + * 3.9.0 introduced autoRead toggling in ServerInboundHandler.attemptFullWrite + * that races with AsyncBodyReader's autoRead management when requestStreaming + * is enabled. Under concurrent load this causes channels to get stuck in an + * unreadable state, making the server stop responding to some requests. + */ +object RequestStreamingConcurrencySpec extends ZIOSpecDefault { + + private val Parallelism = 100 + private val OpsPerFiber = 20 + private val PayloadSize = 4096 + + // In-memory store: UUID -> bytes + private val store = new java.util.concurrent.ConcurrentHashMap[UUID, Array[Byte]]() + + private val routes = Routes( + // POST /store — consume streamed body, store it, return the id + Method.POST / "store" -> handler { (req: Request) => + for { + bytes <- req.body.asChunk + id = UUID.randomUUID() + _ = store.put(id, bytes.toArray) + } yield Response( + Status.Created, + Headers(Header.Custom("X-Id", id.toString)), + ) + }, + + // GET /fetch/:id — return stored bytes as a streamed response + Method.GET / "fetch" / string("id") -> handler { (id: String, _: Request) => + val uuid = UUID.fromString(id) + val bytes = store.get(uuid) + if (bytes eq null) ZIO.succeed(Response.notFound) + else ZIO.succeed(Response(body = Body.fromStreamChunked(ZStream.fromChunk(Chunk.fromArray(bytes))))) + }, + ).sandbox + + private def server: ZIO[Any, Throwable, Int] = + for { + portPromise <- Promise.make[Throwable, Int] + _ <- Server + .installRoutes(routes) + .intoPromise(portPromise) + .zipRight(ZIO.never) + .provide( + ZLayer.succeed(NettyConfig.defaultWithFastShutdown), + ZLayer.succeed( + Server.Config.default.onAnyOpenPort + .requestStreaming(RequestStreaming.Enabled) + .idleTimeout(100.seconds), + ), + Server.customized, + ) + .fork + port <- portPromise.await + } yield port + + private def storePayload(port: Int, payload: Chunk[Byte]): RIO[Scope with Client, UUID] = + for { + resp <- Client.streaming( + Request.post( + URL.decode(s"http://localhost:$port/store").toOption.get, + Body.fromChunk(payload), + ), + ) + _ <- ZIO.when(resp.status != Status.Created)( + resp.body.asString.flatMap(body => ZIO.fail(new RuntimeException(s"Store failed: ${resp.status} $body"))), + ) + id <- ZIO + .fromOption(resp.headers.get("X-Id")) + .mapBoth( + _ => new RuntimeException("Missing X-Id header"), + h => UUID.fromString(h), + ) + } yield id + + private def fetchPayload(port: Int, id: UUID): RIO[Scope with Client, Chunk[Byte]] = + for { + resp <- Client.streaming( + Request.get(URL.decode(s"http://localhost:$port/fetch/$id").toOption.get), + ) + _ <- ZIO.when(resp.status != Status.Ok)( + ZIO.fail(new RuntimeException(s"Fetch failed: ${resp.status}")), + ) + bytes <- resp.body.asChunk + } yield bytes + + override def spec: Spec[TestEnvironment with Scope, Any] = + suite("RequestStreamingConcurrencySpec")( + test(s"${Parallelism}x$OpsPerFiber concurrent store+fetch with requestStreaming(Enabled) should not hang") { + for { + port <- server + payload <- Random.nextBytes(PayloadSize) + results <- ZIO + .foreachPar((1 to Parallelism).toList) { _ => + ZIO.foreach((1 to OpsPerFiber).toList) { _ => + for { + id <- storePayload(port, payload) + bytes <- fetchPayload(port, id) + } yield (payload.length, bytes.length) + } + } + .timeoutFail(new RuntimeException("TIMEOUT"))(120.seconds) + res = results.flatten + } yield assertTrue( + res.size == Parallelism * OpsPerFiber, + res.forall { case (expected, actual) => expected == actual }, + ) + }, + ) + .provideSome[Client](Scope.default) + .provideShared( + DnsResolver.default, + ZLayer.succeed(NettyConfig.defaultWithFastShutdown), + ZLayer.succeed(Client.Config.default.connectionTimeout(100.seconds).idleTimeout(100.seconds)), + Client.live, + ) @@ withLiveClock @@ sequential @@ timeout(180.seconds) @@ + TestAspect.after(ZIO.succeed(store.clear())) +} From 101a6563fde522c77342eef0233f2028957ff2dc Mon Sep 17 00:00:00 2001 From: davidl Date: Fri, 27 Mar 2026 13:47:49 +0100 Subject: [PATCH 2/3] fmt --- .../zio/http/RequestStreamingConcurrencySpec.scala | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala b/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala index 43679d672f..f58031da60 100644 --- a/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala +++ b/zio-http/jvm/src/test/scala/zio/http/RequestStreamingConcurrencySpec.scala @@ -16,14 +16,16 @@ package zio.http +import java.util.UUID + import zio._ -import zio.http.Server.RequestStreaming -import zio.http.netty.NettyConfig -import zio.stream.ZStream import zio.test.TestAspect._ import zio.test._ -import java.util.UUID +import zio.stream.ZStream + +import zio.http.Server.RequestStreaming +import zio.http.netty.NettyConfig /** * Regression test for request streaming + concurrent load. From 1b51256ef89604308f742d08b7517e0d868a5f94 Mon Sep 17 00:00:00 2001 From: "github-actions[bot]" <41898282+github-actions[bot]@users.noreply.github.com> Date: Tue, 7 Apr 2026 01:53:38 +0000 Subject: [PATCH 3/3] Update zio, zio-streams, zio-test, ... to 2.1.25 --- project/Dependencies.scala | 2 +- zio-http-example-basic-auth/build.sbt | 10 +++++----- zio-http-example-cookie-auth/build.sbt | 11 +++++------ zio-http-example-digest-auth/build.sbt | 4 ++-- .../build.sbt | 6 +++--- zio-http-example-jwt-bearer-token-auth/build.sbt | 8 ++++---- zio-http-example-oauth-bearer-token-auth/build.sbt | 4 ++-- zio-http-example-opaque-bearer-token-auth/build.sbt | 4 ++-- zio-http-example-webauthn/build.sbt | 2 +- 9 files changed, 25 insertions(+), 26 deletions(-) diff --git a/project/Dependencies.scala b/project/Dependencies.scala index d8dfb43a0a..d573a3d051 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -4,7 +4,7 @@ object Dependencies { val JwtCoreVersion = "11.0.3" val NettyVersion = "4.2.12.Final" val ScalaCompatCollectionVersion = "2.14.0" - val ZioVersion = "2.1.24" + val ZioVersion = "2.1.25" val ZioCliVersion = "0.8.0" val ZioJsonVersion = "0.9.0" val ZioSchemaVersion = "1.8.3" diff --git a/zio-http-example-basic-auth/build.sbt b/zio-http-example-basic-auth/build.sbt index db4565f53f..786327d076 100644 --- a/zio-http-example-basic-auth/build.sbt +++ b/zio-http-example-basic-auth/build.sbt @@ -1,13 +1,13 @@ -name := "zio-http-example-basic-auth" -version := "0.1.0" +name := "zio-http-example-basic-auth" +version := "0.1.0" scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", ) @@ -16,5 +16,5 @@ enablePlugins(DockerPlugin) Compile / mainClass := Some("example.auth.basic.AuthenticationServer") -dockerBaseImage := "eclipse-temurin:21-jre" +dockerBaseImage := "eclipse-temurin:21-jre" dockerExposedPorts := Seq(8080) diff --git a/zio-http-example-cookie-auth/build.sbt b/zio-http-example-cookie-auth/build.sbt index b7a8606683..dbe552649d 100644 --- a/zio-http-example-cookie-auth/build.sbt +++ b/zio-http-example-cookie-auth/build.sbt @@ -1,13 +1,13 @@ -name := "zio-http-example-cookie-auth" -version := "0.1.0" +name := "zio-http-example-cookie-auth" +version := "0.1.0" scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", ) @@ -16,6 +16,5 @@ enablePlugins(DockerPlugin) Compile / mainClass := Some("example.auth.session.cookie.AuthenticationServer") -dockerBaseImage := "eclipse-temurin:21-jre" +dockerBaseImage := "eclipse-temurin:21-jre" dockerExposedPorts := Seq(8080) - diff --git a/zio-http-example-digest-auth/build.sbt b/zio-http-example-digest-auth/build.sbt index dd6b0852fc..0c71363bb2 100644 --- a/zio-http-example-digest-auth/build.sbt +++ b/zio-http-example-digest-auth/build.sbt @@ -4,10 +4,10 @@ scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", ) diff --git a/zio-http-example-jwt-bearer-refresh-token-auth/build.sbt b/zio-http-example-jwt-bearer-refresh-token-auth/build.sbt index 0c3e54a5cf..e95a91a325 100644 --- a/zio-http-example-jwt-bearer-refresh-token-auth/build.sbt +++ b/zio-http-example-jwt-bearer-refresh-token-auth/build.sbt @@ -4,11 +4,11 @@ scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", - "dev.zio" %% "zio-http" % "3.5.1", + "dev.zio" %% "zio" % "2.1.25", + "dev.zio" %% "zio-http" % "3.5.1", "com.github.jwt-scala" %% "jwt-core" % "11.0.3", "com.github.jwt-scala" %% "jwt-zio-json" % "11.0.3", ) diff --git a/zio-http-example-jwt-bearer-token-auth/build.sbt b/zio-http-example-jwt-bearer-token-auth/build.sbt index 1d933e1cf9..e2ca5bbba1 100644 --- a/zio-http-example-jwt-bearer-token-auth/build.sbt +++ b/zio-http-example-jwt-bearer-token-auth/build.sbt @@ -2,13 +2,13 @@ name := "zio-http-example-jwt-bearer-token-auth" version := "0.1.0" scalaVersion := "2.13.18" -publish / skip := true +publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", - "dev.zio" %% "zio-http" % "3.5.1", + "dev.zio" %% "zio" % "2.1.25", + "dev.zio" %% "zio-http" % "3.5.1", "com.github.jwt-scala" %% "jwt-core" % "11.0.3", "com.github.jwt-scala" %% "jwt-zio-json" % "11.0.3", ) diff --git a/zio-http-example-oauth-bearer-token-auth/build.sbt b/zio-http-example-oauth-bearer-token-auth/build.sbt index 41df4d800c..69e23b8f64 100644 --- a/zio-http-example-oauth-bearer-token-auth/build.sbt +++ b/zio-http-example-oauth-bearer-token-auth/build.sbt @@ -4,10 +4,10 @@ scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", "com.github.jwt-scala" %% "jwt-core" % "11.0.3", "com.github.jwt-scala" %% "jwt-zio-json" % "11.0.3", diff --git a/zio-http-example-opaque-bearer-token-auth/build.sbt b/zio-http-example-opaque-bearer-token-auth/build.sbt index 8d5b585943..698e72a897 100644 --- a/zio-http-example-opaque-bearer-token-auth/build.sbt +++ b/zio-http-example-opaque-bearer-token-auth/build.sbt @@ -4,10 +4,10 @@ scalaVersion := "2.13.18" publish / skip := true publishArtifact := false -run / fork := true +run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", ) diff --git a/zio-http-example-webauthn/build.sbt b/zio-http-example-webauthn/build.sbt index db73e5173c..01b6f08476 100644 --- a/zio-http-example-webauthn/build.sbt +++ b/zio-http-example-webauthn/build.sbt @@ -7,7 +7,7 @@ publishArtifact := false run / fork := true libraryDependencies ++= Seq( - "dev.zio" %% "zio" % "2.1.24", + "dev.zio" %% "zio" % "2.1.25", "dev.zio" %% "zio-http" % "3.5.1", "dev.zio" %% "zio-config" % "4.0.7", "com.yubico" % "webauthn-server-core" % "2.7.0",