RequestStreaming.Enabled: connection stops reading when a response is written without consuming the request body — next keep-alive request is silently dropped

[BalduinLandolt]

Describe the bug

When the server runs with RequestStreaming.Enabled and a handler answers a request without consuming its body (in our case: tapir security logic rejecting with 401 before body decoding, but an unmatched route or an endpoint that just ignores the body behaves the same), the connection can end up in a state where it never reads from the socket again.

The response to that request is still written correctly. But if the request's body bytes arrive in a later TCP read burst than the headers, the AsyncBodyReader that was installed for the request stays in Buffering state with autoRead=false, and nothing ever calls ctx.read() again. The next request on the same keep-alive connection is then never read and never answered — the client just times out.

As far as I can tell from the sources: AsyncBodyReader.handlerAdded sets autoRead=false; it is only undone when the reader sees LastHttpContent or the body is consumed (connect() → ctx.read()). ServerInboundHandler.writeResponse neither drains the remaining body nor closes the connection when the response completes while the reader is still buffering.

To Reproduce

Self-contained scala-cli file (below). It starts a server with request streaming enabled and one route that returns 401 without touching the body, then runs the same scenario twice over a raw socket:

1. control: PUT headers+body in one write, then a GET on the same connection → both answered
2. split: PUT headers, 300 ms pause, then the body, then the same GET → the PUT gets its 401, the GET is never answered

//> using scala 3.3.7
//> using dep dev.zio::zio-http:3.11.3

import zio.*
import zio.http.*
import zio.http.Server.Config

import java.io.{InputStream, OutputStream}
import java.net.Socket

object LostRequestRepro extends ZIOAppDefault {

  val routes: Routes[Any, Response] = Routes(
    // answers WITHOUT touching the request body — like a security failure in tapir
    Method.PUT / "reject"  -> handler((_: Request) => Response.status(Status.Unauthorized)),
    Method.GET / "version" -> handler(Response.text("ok")),
  )

  def readResponse(in: InputStream, maxWaitMs: Int): String = {
    val deadline = java.lang.System.currentTimeMillis() + maxWaitMs
    val sb       = new StringBuilder
    while (java.lang.System.currentTimeMillis() < deadline && !sb.toString.contains("\r\n\r\n")) {
      if (in.available() > 0) {
        val buf = new Array[Byte](8192)
        val n   = in.read(buf)
        if (n > 0) sb.append(new String(buf, 0, n))
      } else Thread.sleep(10)
    }
    sb.toString
  }

  def runScenario(port: Int, splitBody: Boolean): String = {
    val socket = new Socket("127.0.0.1", port)
    socket.setTcpNoDelay(true)
    val out: OutputStream = socket.getOutputStream
    val in: InputStream   = socket.getInputStream

    val body    = """{"x":"yyyy"}"""
    val headers =
      s"PUT /reject HTTP/1.1\r\nHost: localhost\r\nContent-Type: application/json\r\nContent-Length: ${body.length}\r\n\r\n"

    if (splitBody) {
      out.write(headers.getBytes); out.flush()
      Thread.sleep(300) // body arrives in a second read burst
      out.write(body.getBytes); out.flush()
    } else {
      out.write((headers + body).getBytes); out.flush()
    }
    val resp1 = readResponse(in, 2000).linesIterator.nextOption().getOrElse("<no response>")

    // second request on the SAME connection:
    out.write("GET /version HTTP/1.1\r\nHost: localhost\r\n\r\n".getBytes); out.flush()
    val resp2 = readResponse(in, 3000)
    socket.close()
    val r2 = if (resp2.isEmpty) "<NO RESPONSE within 3s — request lost>" else resp2.linesIterator.next()
    s"first(PUT->401): $resp1 | second(GET, same connection): $r2"
  }

  override def run =
    (for {
      port <- Server.install(routes)
      ctrl <- ZIO.attemptBlocking(runScenario(port, splitBody = false))
      _    <- Console.printLine(s"CONTROL (headers+body in one segment): $ctrl")
      bug  <- ZIO.attemptBlocking(runScenario(port, splitBody = true))
      _    <- Console.printLine(s"SPLIT   (body in second segment):      $bug")
    } yield ()).provide(
      Server.live,
      ZLayer.succeed(Config.default.onAnyOpenPort.enableRequestStreaming),
    )
}

Output:

CONTROL (headers+body in one segment): first(PUT->401): HTTP/1.1 401 Unauthorized | second(GET, same connection): HTTP/1.1 200 OK
SPLIT   (body in second segment):      first(PUT->401): HTTP/1.1 401 Unauthorized | second(GET, same connection): <NO RESPONSE within 3s — request lost>

The raw socket is only there to make the segment split deterministic. The race also fires with a plain java.net.http.HttpClient (it writes headers and body as separate segments): looping PUT-with-body→401 followed by a GET on the pooled connection loses a request within a few dozen iterations on an idle machine.

Expected behaviour

The GET on the same connection gets its 200 in both scenarios. When a response completes while the request body is unconsumed, I'd expect the server to either drain the remaining body or close the connection (many HTTP servers do the latter) — not to keep the connection open but stop reading it.

Environment

• zio-http 3.8.0 and 3.11.3 (both reproduce; only requests that go through the async reader are affected — Disabled is fine, and Hybrid is fine for bodies under its aggregation threshold, since aggregated requests never install the async reader; bodies above the threshold still stream and are presumably affected)
• reproduced on macOS/JDK 17; originally observed on Linux/JDK 25

Additional context

We found this hunting down long-standing CI flakiness in dasch-swiss/dsp-api — a tapir-on-zio-http API where tests would intermittently hang for exactly the client timeout on trivial 401/403 checks and then pass on the next request. dasch-swiss/dsp-api#4179 has the details; we've worked around it by switching to RequestStreaming.Hybrid.

The investigation and reproduction were done with AI assistance, under close review on my side. I may be able to offer an (AI-assisted) PR for the drain-or-close fix, if that's welcome.

[wwbakker]

We are running into the same issue, and for us, also causing flaky integration tests.

we've worked around it by switching to RequestStreaming.Hybrid.

In our tests though, if the upload has no content-length, and the actual upload is bigger than the hybrid cut-off, it results in a 413 http error. For us at least, this is not a valid workaround.