Upgrade to AHC 3.0.15 and decide streamed-body resumes through ResponseBodyControl.execute

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:
 
@@ -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.
 
@@ -80,9 +80,9 @@ 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.
+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.
 

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

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()
       }

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,8 +159,9 @@ 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
-      // 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()
     }
@@ -235,38 +232,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()
+        }
+      })
     }
   }
 

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