From 91c9badbd4245124dd5df77864751b44c2318236 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Tue, 6 Oct 2026 17:42:23 +0200 Subject: [PATCH 1/4] Decide streamed-body resumes through ResponseBodyControl.execute A resume() from a subscriber thread only takes effect later on AHC's event loop, after parts that the loop may still be decoding, and can then let one more socket read through with no demand left. Play WS prevented that by capturing the event loop through Netty's internal ThreadExecutorMap and re-running its check there. AHC 3.0.15 adds ResponseBodyControl.execute, which runs a task in sequence with the body callbacks, so upgrade to it and pass the check to execute instead. A check that runs on the thread that asked for it, inline on the event loop or with a control that applies calls synchronously, resumes directly; one queued from another thread runs a drain() pass on the event loop, as before. At most one queued check is pending at a time. The fake controls in the publisher tests now model execute as well. --- README.md | 4 +- .../scala/play/api/libs/ws/ahc/Streamed.scala | 59 ++++++++----------- .../ResponseBodyPublisherLostResumeSpec.scala | 14 +++-- ...sponseBodyPublisherOffLoopResumeSpec.scala | 12 ++-- ...sponseBodyPublisherSuspendOnLoopSpec.scala | 12 ++-- project/Dependencies.scala | 2 +- 6 files changed, 49 insertions(+), 54 deletions(-) diff --git a/README.md b/README.md index 08d5c745..e8f2ea26 100644 --- a/README.md +++ b/README.md @@ -55,7 +55,7 @@ Specifically, shading AsyncHttpClient means that its Netty version does not conf ### AHC 3 migration notes -Play WS 3.1 uses shaded AsyncHttpClient 3.0.14. Most Play WS request, response, streaming, and OAuth entry points remain available, but redirect, cookie, and streamed-body behavior is not identical to the AHC 2-backed releases. Code that uses shaded AHC types directly through an `underlying` escape hatch must be updated for AHC 3. A client supplied directly as an `AsyncHttpClient` also uses that client's own defaults rather than all of Play WS's configuration defaults. +Play WS 3.1 uses shaded AsyncHttpClient 3.0.15. Most Play WS request, response, streaming, and OAuth entry points remain available, but redirect, cookie, and streamed-body behavior is not identical to the AHC 2-backed releases. Code that uses shaded AHC types directly through an `underlying` escape hatch must be updated for AHC 3. A client supplied directly as an `AsyncHttpClient` also uses that client's own defaults rather than all of Play WS's configuration defaults. There are two additional compatibility changes for APIs that exposed AHC 2 implementation details: @@ -80,7 +80,7 @@ Play WS defaults `play.ws.ahc.http2Enabled` to `false` to preserve HTTP/1.1 beha The AHC defaults are a 16 MiB per-stream window and no client-side concurrent-stream cap (`http2MaxConcurrentStreams = -1` leaves the limit server-controlled). Play WS also defaults its connection limits to unlimited, so there is no hard client-side aggregate buffering bound when HTTP/2 is enabled. A rough upper estimate for received data queued by suspended HTTP/2 responses is `connections × effective concurrent streams × http2InitialWindowSize`, excluding network and decoder overhead; decoded data from compressed responses can be much larger. A finite aggregate policy needs limits on all three factors, but connection caps also apply to HTTP/1.1 hosts: with AHC's default zero wait for a free connection, requests exceeding a cap fail immediately. Do not apply a low `maxConnectionsPerHost` merely as an HTTP/2 memory setting without assessing HTTP/1.1 concurrency. -With AHC 3.0.14, a TLS connection to a proxy of type `HTTPS` also offers `h2` when `http2Enabled = true`. If the proxy selects `h2`, the request fails, because AHC still sends its `CONNECT` or proxied request over HTTP/1.1. Play WS's own proxy settings always connect to the proxy without TLS: setting `WSProxyServer.protocol` to `"https"` (`Some("https")` in Scala) only selects the authentication scheme. So this affects only an `HTTPS` proxy configured on AHC directly, for example through `AhcConfigBuilder.modifyUnderlying`; keep HTTP/2 disabled for such a client. +With AHC 3.0.15, a TLS connection to a proxy of type `HTTPS` also offers `h2` when `http2Enabled = true`. If the proxy selects `h2`, the request fails, because AHC still sends its `CONNECT` or proxied request over HTTP/1.1. Play WS's own proxy settings always connect to the proxy without TLS: setting `WSProxyServer.protocol` to `"https"` (`Some("https")` in Scala) only selects the authentication scheme. So this affects only an `HTTPS` proxy configured on AHC directly, for example through `AhcConfigBuilder.modifyUnderlying`; keep HTTP/2 disabled for such a client. Set `play.ws.ahc.maxDecompressedResponseSize` to apply one decompressed-response limit to both HTTP/1.1 and HTTP/2. AHC's defaults apply when it is unset. This limits decoded bytes per response, not total client heap use: each socket read of a compressed body (up to 64 KiB) is inflated in full before downstream demand is checked, so a highly compressible response can queue many MiB of body parts beyond demand, roughly 64 KiB times the compression ratio per read. AHC 2 behaves the same way. AHC inflates any `Content-Encoding` a server sends, even when Play WS did not request compression, so assess memory use for clients that stream responses from untrusted servers. Setting the limit to zero disables decompression-bomb protection and is not recommended for untrusted responses. diff --git a/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala b/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala index d3c65673..91a8b9f6 100644 --- a/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala +++ b/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala @@ -7,7 +7,6 @@ package play.api.libs.ws.ahc import java.net.URI import java.nio.ByteBuffer import java.util.concurrent.ConcurrentLinkedQueue -import java.util.concurrent.RejectedExecutionException import java.util.concurrent.atomic.AtomicBoolean import java.util.concurrent.atomic.AtomicInteger import java.util.concurrent.atomic.AtomicLong @@ -20,8 +19,6 @@ import org.reactivestreams.Subscription import play.shaded.ahc.io.netty.buffer.ByteBuf import play.shaded.ahc.io.netty.buffer.Unpooled import play.shaded.ahc.io.netty.handler.codec.http.HttpHeaders -import play.shaded.ahc.io.netty.util.concurrent.EventExecutor -import play.shaded.ahc.io.netty.util.internal.ThreadExecutorMap import play.shaded.ahc.org.asynchttpclient.AsyncHandler import play.shaded.ahc.org.asynchttpclient.AsyncHandler.State import play.shaded.ahc.org.asynchttpclient.HttpResponseBodyPart @@ -126,9 +123,8 @@ private final class ResponseBodyPublisher(control: ResponseBodyControl) extends private val resumeCheckPending = new AtomicBoolean() @volatile private var failure: Throwable = _ - // AHC 3 creates the control and starts the response body on the channel event loop. ThreadExecutorMap is internal - // to Netty, but Play WS shades a fixed Netty version. - private val eventLoop: EventExecutor = ThreadExecutorMap.currentExecutor() + // AHC creates this publisher from onResponseBodyStart on the channel event loop, so this suspension takes effect + // before the first body part. control.suspend() override def subscribe(nextSubscriber: Subscriber[? >: HttpResponseBodyPart]): Unit = { @@ -163,7 +159,7 @@ private final class ResponseBodyPublisher(control: ResponseBodyControl) extends } } if (accepted) { - // AHC delivers body parts on that event loop. Suspending from drain() instead could queue an off-loop + // AHC delivers body parts on the channel event loop. Suspending from drain() instead could queue an off-loop // command after a newer inline resume and leave a demanded response stalled. control.suspend() drain() @@ -235,38 +231,31 @@ private final class ResponseBodyPublisher(control: ResponseBodyControl) extends } /** - * Resumes transport reads for unmet demand, from the channel event loop whenever that loop is known. + * Resumes transport reads for unmet demand, deciding on the thread that delivers the body parts. * - * AHC 3.0.14 applies a control call made on the event loop inline, but queues a call made on another thread and - * applies it later without checking again. A drain() on a subscriber thread can find the queue empty while the loop - * is still decoding a read. A resume() queued then only runs after the loop has buffered that read's parts, when - * the demand may already be met, and lets one more socket read through with no demand left. For a compressed body, - * that read can inflate to tens of MiB. + * A resume() called from another thread only takes effect later, on the AHC event loop. A drain() on a subscriber + * thread can find the queue empty while the loop is still decoding a read; a resume() queued then would run after + * the loop has buffered that read's parts, when the demand may already be met, and let one more socket read through + * with no demand left. For a compressed body, that read can inflate to tens of MiB. * - * So a drain() outside the loop schedules a single drain() on it instead, which resumes inline only if demand is - * still unmet. AHC delivers body parts, which offer() buffers, only on that loop, and running the check as a drain() - * pass keeps another thread from delivering the last parts between the check and resume(). Without a known loop, for - * example when the publisher was not created on a Netty event loop, resume directly. - * - * This indirection can be dropped once a future AHC release defines the order in which control calls from different - * threads take effect, or offers a resume that is evaluated on the event loop. + * So the decision runs through ResponseBodyControl.execute, which runs a task in sequence with the body callbacks. + * When the task runs on the thread that called resumeReads(), that is, inline on the event loop or with a control + * that applies calls synchronously, it resumes directly: this drain() already decided on the current state. When it + * runs on the event loop after being queued from another thread, it runs a drain() pass there instead, which resumes + * inline only if demand is still unmet. Running the check as a drain() pass keeps another thread from delivering the + * last parts between the check and resume(). At most one such pass is pending at a time. */ private def resumeReads(): Unit = { - if (eventLoop == null || eventLoop.inEventLoop()) { - control.resume() - } else if (resumeCheckPending.compareAndSet(false, true)) { - try { - eventLoop.execute(new Runnable { - override def run(): Unit = { - // Clear the flag before draining, so that unmet demand found by a concurrent drain() can schedule another - // check. Clearing it afterwards could drop that check and leave reads suspended. - resumeCheckPending.set(false) - drain() - } - }) - } catch { - case _: RejectedExecutionException => () // The event loop is shutting down, so there is nothing to resume. - } + val caller = Thread.currentThread() + if (resumeCheckPending.compareAndSet(false, true)) { + control.execute(new Runnable { + override def run(): Unit = { + // Clear the flag first, so that unmet demand found by a concurrent drain() can schedule another check. + // Clearing it afterwards could drop that check and leave reads suspended. + resumeCheckPending.set(false) + if (Thread.currentThread() eq caller) control.resume() else drain() + } + }) } } diff --git a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherLostResumeSpec.scala b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherLostResumeSpec.scala index 6f66c01d..15756aea 100644 --- a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherLostResumeSpec.scala +++ b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherLostResumeSpec.scala @@ -87,8 +87,8 @@ class ResponseBodyPublisherLostResumeSpec extends Specification { } /** - * Models AHC 3.0.14 `NettyResponseBodyControl`: a call runs inline when made on the channel event loop and is - * queued to the event loop otherwise; `suspended` is the applied state (suspended + autoRead(false)). + * Models AHC's `NettyResponseBodyControl`: a call, including `execute`, runs inline when made on the channel event loop + * and is queued to the event loop otherwise; `suspended` is the applied state (suspended + autoRead(false)). */ final class EventLoopResponseBodyControl extends ResponseBodyControl { private val loopThread = new AtomicReference[Thread]() @@ -114,28 +114,30 @@ final class EventLoopResponseBodyControl extends ResponseBodyControl { /** Waits until every task queued so far (and any queued by them) has run. */ def quiesce(): Unit = { onLoop(()); onLoop(()) } - private def execute(task: () => Unit): Unit = + private def onLoopOrQueue(task: () => Unit): Unit = if (inEventLoop) task() else try loop.execute(new Runnable { def run(): Unit = task() }) catch { case _: java.util.concurrent.RejectedExecutionException => () } // as AHC: loop is gone + override def execute(task: Runnable): Unit = onLoopOrQueue(() => task.run()) + override def suspend(): Unit = { val inline = inEventLoop - execute { () => + onLoopOrQueue { () => if (active && !suspended) suspended = true if (inline) onInlineSuspend(inlineSuspends.incrementAndGet()) } } - override def resume(): Unit = execute { () => + override def resume(): Unit = onLoopOrQueue { () => if (active && suspended) { suspended = false onResumed() } } - override def cancel(): Unit = execute { () => + override def cancel(): Unit = onLoopOrQueue { () => if (active) { active = false suspended = false diff --git a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherOffLoopResumeSpec.scala b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherOffLoopResumeSpec.scala index 2c4887c6..fa1c51a2 100644 --- a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherOffLoopResumeSpec.scala +++ b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherOffLoopResumeSpec.scala @@ -168,7 +168,7 @@ object ResponseBodyPublisherOffLoopResumeSpec { } } - /** Models AHC 3.0.14 `NettyResponseBodyControl`: inline on the event loop, queued to it otherwise. */ + /** Models AHC's `NettyResponseBodyControl`, including `execute`: inline on the event loop, queued to it otherwise. */ final class LoopControl(loop: EventExecutor) extends ResponseBodyControl { @volatile var active = true @volatile var suspended = false @@ -177,15 +177,17 @@ object ResponseBodyPublisherOffLoopResumeSpec { def offLoopResumes: List[String] = offLoop.asScala.toList - private def execute(task: => Unit): Unit = + private def onLoopOrQueue(task: => Unit): Unit = if (loop.inEventLoop()) task else loop.execute(new Runnable { def run(): Unit = task }) - override def suspend(): Unit = execute(if (active) suspended = true) + override def execute(task: Runnable): Unit = onLoopOrQueue(task.run()) + + override def suspend(): Unit = onLoopOrQueue(if (active) suspended = true) override def resume(): Unit = { resumeCalls.incrementAndGet() if (!loop.inEventLoop()) offLoop.add(Thread.currentThread().getName) - execute(if (active && suspended) suspended = false) + onLoopOrQueue(if (active && suspended) suspended = false) } - override def cancel(): Unit = execute(if (active) { active = false; suspended = false }) + override def cancel(): Unit = onLoopOrQueue(if (active) { active = false; suspended = false }) } } diff --git a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala index e838a078..e38ecbf5 100644 --- a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala +++ b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala @@ -163,21 +163,23 @@ object ResponseBodyPublisherSuspendOnLoopSpec { def shutdown(): Unit = Seq(loop, other, timer, helper).foreach(_.shutdownNow()) } - /** Models AHC 3.0.14 `NettyResponseBodyControl`: inline on the event loop, queued to it otherwise. */ + /** Models AHC's `NettyResponseBodyControl`, including `execute`: inline on the event loop, queued to it otherwise. */ final class LoopControl(rig: Rig) extends ResponseBodyControl { @volatile var active = true @volatile var suspended = false @volatile var cancelled = false - private def execute(task: => Unit): Unit = + private def onLoopOrQueue(task: => Unit): Unit = if (rig.inEventLoop) task else rig.loop.execute(() => task) + override def execute(task: Runnable): Unit = onLoopOrQueue(task.run()) + override def suspend(): Unit = { if (!rig.inEventLoop) rig.recordOffLoopSuspend(Thread.currentThread().getName) - execute(if (active) suspended = true) + onLoopOrQueue(if (active) suspended = true) } - override def resume(): Unit = execute(if (active && suspended) suspended = false) - override def cancel(): Unit = execute(if (active) { deactivate(); cancelled = true }) + override def resume(): Unit = onLoopOrQueue(if (active && suspended) suspended = false) + override def cancel(): Unit = onLoopOrQueue(if (active) { deactivate(); cancelled = true }) def deactivate(): Unit = { active = false; suspended = false } } diff --git a/project/Dependencies.scala b/project/Dependencies.scala index da3cbb9f..564ea9b0 100644 --- a/project/Dependencies.scala +++ b/project/Dependencies.scala @@ -52,7 +52,7 @@ object Dependencies { val cachecontrol = Seq("org.playframework" %% "cachecontrol" % "3.1.0-M3+2-5a8b88cf-SNAPSHOT") - val asyncHttpClient = Seq("org.asynchttpclient" % "async-http-client" % "3.0.14") + val asyncHttpClient = Seq("org.asynchttpclient" % "async-http-client" % "3.0.15") // Keep in sync with the Netty version tested by the AHC version above. Netty // is bundled into Play WS's shaded AHC jar, so applications cannot override it. From 4abdf9edea1ef87276b700aa6b82b13cd1aaee54 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Tue, 6 Oct 2026 17:42:23 +0200 Subject: [PATCH 2/4] Document that raw Cookie headers survive cookie-store hops With play.ws.ahc.useCookieStore = true, AHC 3.0.14 re-encoded a raw Cookie header on redirects and authentication retries: pairs could be reordered, valueless or unbalanced-quote pairs were dropped, and a value that AHC's strict encoder rejected failed the hop. AHC 3.0.15 keeps the header as written and only takes out the pairs whose cookie a response replaced or deleted. Update the README, and test a redirect that replaces one raw pair and keeps a valueless one. --- README.md | 2 +- .../api/libs/ws/ahc/AhcWSClientSpec.scala | 25 +++++++++++++++++++ 2 files changed, 26 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index e8f2ea26..1bff11ef 100644 --- a/README.md +++ b/README.md @@ -68,7 +68,7 @@ Play WS adapts `Source`, `File`, and input-stream-supplier request bodies to a o AHC 3.0.14 preserves the method and body of non-`POST` requests across 301 and 302 redirects; the legacy rewrite to a bodyless `GET` now applies only to `POST`. A 303 still switches to a bodyless `GET`, except that `HEAD` and `OPTIONS` keep their method, while 307 and 308 preserve method and body. This can turn a former bodyless-`GET` success into an explicit replay failure for a one-shot Play WS request body. `QUERY` follows the same non-`POST` rule. -Cookie handling also differs from the AHC 2-based Play WS. Cookies added through the Play WS cookie API (`addCookies`, `withCookies`) now follow same-origin redirects, whereas AHC 2 dropped them on every redirect; cross-origin redirects still strip them. A raw `Cookie` header is now merged with API and cookie-store cookies on the initial request and on every redirect, and a raw pair wins over an API cookie of the same name. AHC 2 instead let API cookies replace the raw header on the initial request, sent only the raw header on redirects, and, with a cookie store, replaced the raw header with the stored cookies for the URL. Several raw `Cookie` headers are now folded into one. With the opt-in `play.ws.ahc.useCookieStore = true`, redirects and authentication retries re-encode the raw header: pairs can be reordered, valueless or unbalanced-quote pairs are dropped, and a value that AHC's strict encoder rejects fails the redirect or retry even though the initial request was sent. Prefer valid cookie syntax and the Play WS cookie API; `useLaxCookieEncoder = true` changes validation more broadly and is not a general safety workaround. The cookie store remains disabled by default. +Cookie handling also differs from the AHC 2-based Play WS. Cookies added through the Play WS cookie API (`addCookies`, `withCookies`) now follow same-origin redirects, whereas AHC 2 dropped them on every redirect; cross-origin redirects still strip them. A raw `Cookie` header is now merged with API and cookie-store cookies on the initial request and on every redirect, and a raw pair wins over an API cookie of the same name. AHC 2 instead let API cookies replace the raw header on the initial request, sent only the raw header on redirects, and, with a cookie store, replaced the raw header with the stored cookies for the URL. Several raw `Cookie` headers are now folded into one. With the opt-in `play.ws.ahc.useCookieStore = true`, redirects and authentication retries keep the raw header as written, except for pairs whose cookie a response replaced or deleted; the stored cookie, if any, is sent instead. The cookie store remains disabled by default. Connections authenticated with NTLM, SPNEGO or Kerberos, or through a proxy login that applies to the whole connection, are now pooled per principal, so AHC no longer reuses them for requests made with other credentials. diff --git a/integration-tests/src/test/scala/play/api/libs/ws/ahc/AhcWSClientSpec.scala b/integration-tests/src/test/scala/play/api/libs/ws/ahc/AhcWSClientSpec.scala index e58b381b..449ed16d 100644 --- a/integration-tests/src/test/scala/play/api/libs/ws/ahc/AhcWSClientSpec.scala +++ b/integration-tests/src/test/scala/play/api/libs/ws/ahc/AhcWSClientSpec.scala @@ -88,6 +88,16 @@ class AhcWSClientSpec(implicit val executionEnv: ExecutionEnv) case _ => Results.NotFound } + case GET(p"/cookie-echo-redirect") => + components.defaultActionBuilder { + Results + .Redirect(url = "/cookie-echo", status = MOVED_PERMANENTLY) + .withCookies(Cookie(name = "flash", value = "redirect-cookie")) + } + case GET(p"/cookie-echo") => + components.defaultActionBuilder { req => + Results.Ok(req.headers.getAll("Cookie").mkString("\n")) + } case p"/redirect/${status}" => components.defaultActionBuilder { Results.Redirect("/index", status.toInt) @@ -310,6 +320,21 @@ class AhcWSClientSpec(implicit val executionEnv: ExecutionEnv) } } + "keep a raw Cookie header as written when following a redirect with a cookie store" in { + withClientFollowingRedirect(AhcWSClientConfigFactory.forConfig().copy(useCookieStore = true)) { client => + val result = Await.result( + client + .url(s"http://localhost:$testServerPort/cookie-echo-redirect") + .addHttpHeaders("Cookie" -> "z=1; flash=stale; a=2; flag") + .get() + .map(res => res.body[String]), + defaultTimeout + ) + // Only the pair that the redirect replaced is taken out; the stored cookie is sent instead. + result must beEqualTo("z=1; a=2; flag; flash=redirect-cookie") + } + } + "not keep cookie when repeating the request" in { withClientFollowingRedirect() { client => // First let's do a request that sets a cookie From d1716984dab22d45ce5cb89ce592b3ecdbee2ee4 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Tue, 6 Oct 2026 17:42:24 +0200 Subject: [PATCH 3/4] Document demand-bounded HTTP/1.1 decompression AHC 3.0.15 hands on decompressed HTTP/1.1 body parts of at most 64 KiB and pauses decompression while the response is suspended. Before, every socket read of up to 64 KiB was inflated and delivered in full, so a highly compressible body queued many MiB beyond demand. Update the README; HTTP/2 still inflates each received DATA frame in full. StreamedGzipBufferingSpec now requires that nothing is delivered beyond demand: with early demand for one part, AHC delivers exactly that part. Before, 1,875,916 bytes were delivered beyond demand. --- README.md | 2 +- .../play/api/libs/ws/ahc/StreamedGzipBufferingSpec.scala | 7 ++++--- 2 files changed, 5 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index 1bff11ef..aeccfb7e 100644 --- a/README.md +++ b/README.md @@ -82,7 +82,7 @@ The AHC defaults are a 16 MiB per-stream window and no client-side concurrent-st With AHC 3.0.15, a TLS connection to a proxy of type `HTTPS` also offers `h2` when `http2Enabled = true`. If the proxy selects `h2`, the request fails, because AHC still sends its `CONNECT` or proxied request over HTTP/1.1. Play WS's own proxy settings always connect to the proxy without TLS: setting `WSProxyServer.protocol` to `"https"` (`Some("https")` in Scala) only selects the authentication scheme. So this affects only an `HTTPS` proxy configured on AHC directly, for example through `AhcConfigBuilder.modifyUnderlying`; keep HTTP/2 disabled for such a client. -Set `play.ws.ahc.maxDecompressedResponseSize` to apply one decompressed-response limit to both HTTP/1.1 and HTTP/2. AHC's defaults apply when it is unset. This limits decoded bytes per response, not total client heap use: each socket read of a compressed body (up to 64 KiB) is inflated in full before downstream demand is checked, so a highly compressible response can queue many MiB of body parts beyond demand, roughly 64 KiB times the compression ratio per read. AHC 2 behaves the same way. AHC inflates any `Content-Encoding` a server sends, even when Play WS did not request compression, so assess memory use for clients that stream responses from untrusted servers. Setting the limit to zero disables decompression-bomb protection and is not recommended for untrusted responses. +Set `play.ws.ahc.maxDecompressedResponseSize` to apply one decompressed-response limit to both HTTP/1.1 and HTTP/2. AHC's defaults apply when it is unset. This limits decoded bytes per response, not total client heap use. Over HTTP/1.1, AHC hands on decompressed body parts of at most 64 KiB and pauses decompression while the stream has no unmet demand, so little decoded data queues beyond demand; only when the connection closes is the rest of what was already received decompressed at once. AHC 2 instead inflated every socket read in full. Over HTTP/2, AHC still inflates each received `DATA` frame in full before downstream demand is checked, so a highly compressible response can queue many MiB of body parts beyond demand. AHC inflates any `Content-Encoding` a server sends, even when Play WS did not request compression, so assess memory use for clients that stream responses from untrusted servers. Setting the limit to zero disables decompression-bomb protection and is not recommended for untrusted responses. `play.ws.ahc.shutdownQuietPeriod` and `play.ws.ahc.shutdownTimeout` control AHC event-loop shutdown. Both must be non-negative, and the timeout must be at least as long as the quiet period. Both default to zero to preserve Play WS's existing immediate-shutdown behavior. diff --git a/integration-tests/src/test/scala/play/api/libs/ws/ahc/StreamedGzipBufferingSpec.scala b/integration-tests/src/test/scala/play/api/libs/ws/ahc/StreamedGzipBufferingSpec.scala index 6b517b62..6689ff16 100644 --- a/integration-tests/src/test/scala/play/api/libs/ws/ahc/StreamedGzipBufferingSpec.scala +++ b/integration-tests/src/test/scala/play/api/libs/ws/ahc/StreamedGzipBufferingSpec.scala @@ -54,12 +54,13 @@ class StreamedGzipBufferingSpec(implicit val executionEnv: ExecutionEnv) } "A streamed gzip response" should { - "not read past the demanded socket read when demand arrives from another thread" in { + "not deliver past the demanded part when demand arrives from another thread" in { val client = new DefaultAsyncHttpClient(new AhcConfigBuilder(AhcWSClientConfigFactory.forConfig()).build()) try { - // The first read, which also carries the headers, inflates to about 2 MiB; one more read to tens of MiB. + // AHC delivers no further HTTP/1.1 body part while the stream is suspended, so only the demanded part arrives. + // Before AHC 3.0.15, the first read, which also carries the headers, was delivered in full, about 2 MiB. val bufferedBeyondDemand = (1 to 3).map(_ => bufferedAfterEarlyDemand(client)) - bufferedBeyondDemand must contain(beLessThan(10L * 1024 * 1024)).forall + bufferedBeyondDemand must contain(be_==(0L)).forall } finally { client.close() } From 7af726665926d055016b38c0b737318c82b4da61 Mon Sep 17 00:00:00 2001 From: Matthias Kurz Date: Wed, 7 Oct 2026 10:36:54 +0200 Subject: [PATCH 4/4] Update the reason for suspending only on the event loop Before AHC 3.0.15, a ResponseBodyControl.suspend() queued from another thread could take effect after a newer resume() made on the event loop and stall a demanded response. That was the reason ResponseBodyPublisher suspends only on the event loop. AHC 3.0.15 skips such a suspend. Suspending on the event loop is still the better place: it takes effect before the next body part, whereas a suspend from another thread would only take effect after further parts. Update the comment and the description of ResponseBodyPublisherSuspendOnLoopSpec; the code and the tests are unchanged. --- .../main/scala/play/api/libs/ws/ahc/Streamed.scala | 5 +++-- .../ResponseBodyPublisherSuspendOnLoopSpec.scala | 14 +++++++++----- 2 files changed, 12 insertions(+), 7 deletions(-) diff --git a/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala b/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala index 91a8b9f6..f24cc626 100644 --- a/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala +++ b/play-ahc-ws-standalone/src/main/scala/play/api/libs/ws/ahc/Streamed.scala @@ -159,8 +159,9 @@ private final class ResponseBodyPublisher(control: ResponseBodyControl) extends } } if (accepted) { - // AHC delivers body parts on the channel event loop. Suspending from drain() instead could queue an off-loop - // command after a newer inline resume and leave a demanded response stalled. + // AHC delivers body parts on the channel event loop, so suspending here takes effect before the next part. A + // suspend from drain() on another thread would only take effect later, after further parts. AHC 3.0.15 skips a + // queued suspend once a newer resume has overtaken it, so that could no longer stall a demanded response. control.suspend() drain() } diff --git a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala index e38ecbf5..9daf41d6 100644 --- a/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala +++ b/play-ahc-ws-standalone/src/test/scala/play/api/libs/ws/ahc/ResponseBodyPublisherSuspendOnLoopSpec.scala @@ -21,10 +21,11 @@ import play.shaded.ahc.org.asynchttpclient.ResponseBodyControl import scala.jdk.CollectionConverters._ /** - * The lost-resume fix relies on `ResponseBodyPublisher` calling `ResponseBodyControl.suspend()` only on the channel - * event loop: AHC applies a call made on the loop inline and queues a call made elsewhere, so an off-loop suspend could - * take effect after a newer inline resume and stall the stream. Each step below runs to completion, including the loop - * tasks it queues, before the next one starts, so these checks do not depend on thread timing. + * `ResponseBodyPublisher` calls `ResponseBodyControl.suspend()` only on the channel event loop, so that each suspension + * takes effect before the next body part. AHC applies a call made on the loop inline and queues a call made elsewhere. + * Before AHC 3.0.15, an off-loop suspend could also take effect after a newer inline resume and stall the stream; AHC + * 3.0.15 skips such a suspend. Each step below runs to completion, including the loop tasks it queues, before the next + * one starts, so these checks do not depend on thread timing. */ class ResponseBodyPublisherSuspendOnLoopSpec extends Specification { import ResponseBodyPublisherSuspendOnLoopSpec._ @@ -163,7 +164,10 @@ object ResponseBodyPublisherSuspendOnLoopSpec { def shutdown(): Unit = Seq(loop, other, timer, helper).foreach(_.shutdownNow()) } - /** Models AHC's `NettyResponseBodyControl`, including `execute`: inline on the event loop, queued to it otherwise. */ + /** + * Models AHC's `NettyResponseBodyControl`, including `execute`: inline on the event loop, queued to it otherwise. It + * leaves out AHC 3.0.15's rule for skipping a queued suspend, which only matters for off-loop suspends. + */ final class LoopControl(rig: Rig) extends ResponseBodyControl { @volatile var active = true @volatile var suspended = false