feat: optional strict response entities for the HTTP client

docs/src/main/paradox/client-side/configuration.md

@@ -20,6 +20,47 @@ Basic client settings can be overridden in multiple ways:
 
 @@snip [reference.conf](/http-core/src/main/resources/reference.conf) { #client-settings }
 
+## Strict Response Entities
+
+By default response entities are streamed, so the application has to consume (or discard) each response entity before
+the connection can be used for the next request. Setting `pekko.http.client.strict-response-entity-timeout` to a
+duration makes the client collect every response entity into a strict entity (`HttpEntity.Strict`) before the response is
+dispatched to the application:
+
+```
+pekko.http.client.strict-response-entity-timeout = 10s
+```
+
+Keep in mind that this buffers each complete response body in memory. A response that is not fully received within the
+configured duration fails with a `TimeoutException`, and one that exceeds
+`pekko.http.client.strict-response-entity-max-bytes` (8 MB by default) fails with an `EntityStreamException`; in both
+cases the connection is failed. Trailing headers of chunked responses are dropped, as they are with
+`HttpEntity.toStrict`.
+
+This applies to the HTTP/1.1 client, which includes the connection pool backing the
+@ref[request-level](request-level.md) and @ref[host-level](host-level.md) APIs, and to the
+@ref[HTTP/2 client](http2.md).
+
+### On HTTP/2
+
+Responses on an HTTP/1.1 connection are sequential, so collecting one response entity never delays another. On HTTP/2
+several requests are in flight on one connection at the same time, which brings a few things to be aware of:
+
+ * Entities are collected for up to `pekko.http.client.http2.max-concurrent-streams` responses concurrently, so that a
+   large response does not hold up smaller ones on other streams.
+ * Responses are emitted in the order in which their entities complete, not in the order in which their headers
+   arrived. HTTP/2 responses are unordered anyway and have to be correlated to their request via a
+   @apidoc[RequestResponseAssociation], so this does not break the API contract, but it does change observed ordering.
+ * Worst-case memory usage per connection is `max-concurrent-streams` × `strict-response-entity-max-bytes`, which is
+   256 × 8 MB with the defaults. Tune both settings for the responses you actually expect.
+ * A response entity that times out or exceeds the maximum fails the whole connection, and with it every other stream
+   in flight on that connection. The remaining body cannot be skipped safely, so there is no way to fail only the one
+   response. With `Http().connectionTo(host).managedPersistentHttp2()` the connection is re-established afterwards
+   according to `pekko.http.client.http2.max-persistent-attempts`.
+ * Entity data is read from the network as fast as the peer sends it (up to the configured maximum) instead of at the
+   pace the application consumes it, so HTTP/2 flow control no longer reflects application backpressure for response
+   bodies.
+
 ## Pool Settings
 
 Pool settings influence the behavior of client connection pools as used with APIs like `Http.singleRequest`

http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes

@@ -0,0 +1,21 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you 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.
+
+# optional strict response entities for the HTTP client https://github.com/apache/pekko-http/pull/1233
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.javadsl.settings.ClientConnectionSettings.withStrictResponseEntityMaxBytes")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ClientConnectionSettings.strictResponseEntityTimeout")
+ProblemFilters.exclude[ReversedMissingMethodProblem]("org.apache.pekko.http.scaladsl.settings.ClientConnectionSettings.strictResponseEntityMaxBytes")

http-core/src/main/resources/reference.conf

@@ -447,6 +447,40 @@ pekko.http {
       max-content-length = infinite
     }
 
+    # If set to a duration, response entities are collected into strict entities
+    # (`HttpEntity.Strict`) before the response is dispatched to the application.
+    # `off` (the default) leaves response entities streamed.
+    #
+    # Note that enabling this means that the complete response body is buffered in
+    # memory. Responses that are not fully received within the given duration are
+    # failed with a `TimeoutException`, responses bigger than
+    # `strict-response-entity-max-bytes` with an `EntityStreamException`. In both
+    # cases the connection is failed as well, since the remaining response body
+    # cannot be skipped safely.
+    #
+    # Trailing headers of chunked responses are dropped, as they are with
+    # `HttpEntity.toStrict`.
+    #
+    # This applies to the HTTP/1.1 client (including the connection pool used by
+    # `singleRequest` and the host-level API) and to the HTTP/2 client. On HTTP/2
+    # the entities of up to `pekko.http.client.http2.max-concurrent-streams`
+    # responses are collected concurrently, so that a big response does not delay
+    # smaller ones on other streams. Note that this means responses are emitted in
+    # the order in which their entities complete rather than in the order their
+    # headers arrived; HTTP/2 responses are unordered anyway and have to be
+    # correlated to their request via a `RequestResponseAssociation`. Also note
+    # that the worst case memory usage per HTTP/2 connection is
+    # `max-concurrent-streams` * `strict-response-entity-max-bytes`, and that
+    # failing one response entity fails the whole connection, and with it every
+    # other stream in flight on it.
+    strict-response-entity-timeout = off
+
+    # The maximum number of bytes collected per response entity when
+    # `strict-response-entity-timeout` is enabled. Ignored otherwise.
+    #
+    # Set to `infinite` to only rely on `pekko.http.client.parsing.max-content-length`.
+    strict-response-entity-max-bytes = ${pekko.http.parsing.max-to-strict-bytes}
+
     # Enables/disables the logging of unencrypted HTTP traffic to and from the HTTP
     # client for debugging reasons.
     #

http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala

@@ -103,6 +103,12 @@ private[http] object OutgoingConnectionBlueprint {
       val responsePrep = Flow[List[ParserOutput.ResponseOutput]]
         .mapConcat(ConstantFun.scalaIdentityFunction)
         .via(new PrepareResponse(parserSettings))
+        .via(settings.strictResponseEntityTimeout match {
+          case Some(timeout) =>
+            // responses on an HTTP/1.1 connection are sequential, so collecting one entity never delays another
+            StreamUtils.strictifyResponseEntities(timeout, settings.strictResponseEntityMaxBytes, parallelism = 1)
+          case None => Flow[HttpResponse]
+        })
 
       val terminationFanout = b.add(Broadcast[HttpResponse](2))
 

http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala

@@ -167,7 +167,27 @@ private[http] object Http2Blueprint {
       StreamUtils.statefulAttrsMap[Http2SubStream, HttpResponse] { attrs =>
         val headerParser = masterHttpHeaderParser.createShallowCopy()
         stream => ResponseParsing.parseResponse(headerParser, settings.parserSettings, attrs)(stream)
-      })
+      }.via(strictifyResponseEntitiesIfConfigured(settings)))
+
+  /**
+   * Collects response entities into strict entities if `pekko.http.client.strict-response-entity-timeout` is
+   * configured.
+   *
+   * Entities are collected for up to `max-concurrent-streams` responses at a time so that a big response does not
+   * delay smaller ones on other streams. Because entities complete in an order of their own, this changes the order
+   * in which responses are emitted; HTTP/2 responses are unordered anyway and have to be correlated to their request
+   * via a `RequestResponseAssociation`.
+   */
+  private def strictifyResponseEntitiesIfConfigured(
+      settings: ClientConnectionSettings): Flow[HttpResponse, HttpResponse, NotUsed] =
+    settings.strictResponseEntityTimeout match {
+      case Some(timeout) =>
+        StreamUtils.strictifyResponseEntities(
+          timeout,
+          settings.strictResponseEntityMaxBytes,
+          parallelism = settings.http2Settings.maxConcurrentStreams)
+      case None => Flow[HttpResponse]
+    }
 
   def idleTimeoutIfConfigured(timeout: Duration): BidiFlow[ByteString, ByteString, ByteString, ByteString, NotUsed] =
     timeout match {

http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala

@@ -44,13 +44,19 @@ private[pekko] final case class ClientConnectionSettingsImpl(
     socketOptions: immutable.Seq[SocketOption],
     parserSettings: ParserSettings,
     streamCancellationDelay: FiniteDuration,
+    strictResponseEntityTimeout: Option[FiniteDuration],
+    strictResponseEntityMaxBytes: Long,
     localAddress: Option[InetSocketAddress],
     http2Settings: Http2ClientSettings,
     transport: ClientTransport)
     extends pekko.http.scaladsl.settings.ClientConnectionSettings {
 
   require(connectingTimeout >= Duration.Zero, "connectingTimeout must be >= 0")
   require(requestHeaderSizeHint > 0, "request-size-hint must be > 0")
+  require(
+    strictResponseEntityTimeout.forall(_ > Duration.Zero),
+    "strict-response-entity-timeout must be > 0 or `off`")
+  require(strictResponseEntityMaxBytes >= 0, "strict-response-entity-max-bytes must be >= 0")
   require(
     Try { parserSettings.maxContentLength }.isSuccess,
     "The provided ParserSettings is a generic object that does not contain the client-specific settings.")
@@ -86,6 +92,11 @@ private[pekko] object ClientConnectionSettingsImpl
       socketOptions = SocketOptionSettings.fromSubConfig(root, c.getConfig("socket-options")),
       parserSettings = ParserSettingsImpl.fromSubConfig(root, c.getConfig("parsing")),
       streamCancellationDelay = c.getFiniteDuration("stream-cancellation-delay"),
+      strictResponseEntityTimeout = c.getString("strict-response-entity-timeout").toRootLowerCase match {
+        case "off" => None
+        case _     => Some(c.getFiniteDuration("strict-response-entity-timeout"))
+      },
+      strictResponseEntityMaxBytes = c.getPossiblyInfiniteBytes("strict-response-entity-max-bytes"),
       localAddress = None,
       http2Settings = Http2ClientSettingsImpl.fromSubConfig(root, c.getConfig("http2")),
       transport = ClientTransport.TCP)

http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala

@@ -17,7 +17,7 @@ import org.apache.pekko
 import pekko.NotUsed
 import pekko.actor.Cancellable
 import pekko.annotation.InternalApi
-import pekko.http.scaladsl.model.HttpEntity
+import pekko.http.scaladsl.model.{ HttpEntity, HttpResponse }
 import pekko.http.scaladsl.util.FastFuture
 import pekko.stream._
 import pekko.stream.impl.fusing.GraphInterpreter
@@ -273,6 +273,33 @@ private[http] object StreamUtils {
   def statefulAttrsMap[T, U](functionConstructor: Attributes => T => U): Flow[T, U, NotUsed] =
     Flow[T].via(ExposeAttributes[T, U](functionConstructor))
 
+  /**
+   * Collects response entities into `HttpEntity.Strict` entities using the given timeout and maximum number of bytes.
+   * Responses that already carry a strict entity are passed through unchanged.
+   *
+   * With `parallelism == 1` responses keep their original order, which is what the HTTP/1.1 client needs where
+   * responses on a connection are sequential anyway. With a bigger value up to `parallelism` entities are collected
+   * concurrently and responses are emitted in the order in which their entities complete.
+   */
+  def strictifyResponseEntities(
+      timeout: FiniteDuration,
+      maxBytes: Long,
+      parallelism: Int): Flow[HttpResponse, HttpResponse, NotUsed] = {
+    def strictify(response: HttpResponse): Source[HttpResponse, Any] =
+      response.entity match {
+        case _: HttpEntity.Strict => Source.single(response)
+        case entity               =>
+          entity.dataBytes
+            .via(new ToStrict(timeout, Some(maxBytes), entity.contentType))
+            .map(strict => response.withEntity(strict))
+      }
+
+    val flow =
+      if (parallelism <= 1) Flow[HttpResponse].flatMapConcat(strictify)
+      else Flow[HttpResponse].flatMapMerge(parallelism, strictify)
+    flow.named("strictifyResponseEntities")
+  }
+
   trait ScheduleSupport extends GraphStageLogic { self =>
 
     /**

http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala

@@ -64,6 +64,21 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti
    */
   final def getStreamCancellationDelay: JDuration = streamCancellationDelay.toJava
   final def getRequestHeaderSizeHint: Int = requestHeaderSizeHint
+
+  /**
+   * If present, response entities are collected into strict entities (using the given timeout)
+   * before the response is dispatched to the application. Only supported by the HTTP/1.1 client.
+   *
+   * @since 2.0.0
+   */
+  final def getStrictResponseEntityTimeout: Optional[JDuration] = strictResponseEntityTimeout.map(_.toJava).toJava
+
+  /**
+   * The maximum number of bytes collected per response entity when a strict response entity timeout is set.
+   *
+   * @since 2.0.0
+   */
+  final def getStrictResponseEntityMaxBytes: Long = strictResponseEntityMaxBytes
   final def getWebsocketSettings: WebSocketSettings = websocketSettings
   final def getWebsocketRandomFactory: Supplier[Random] = () => websocketRandomFactory()
   final def getLocalAddress: Optional[InetSocketAddress] = localAddress.toJava
@@ -79,6 +94,9 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti
   def withRequestHeaderSizeHint(newValue: Int): ClientConnectionSettings
   def withStreamCancellationDelay(newValue: FiniteDuration): ClientConnectionSettings
 
+  /** @since 2.0.0 */
+  def withStrictResponseEntityMaxBytes(newValue: Long): ClientConnectionSettings
+
   // Java API versions of mutators
 
   /**
@@ -114,6 +132,13 @@ abstract class ClientConnectionSettings private[pekko] () { self: ClientConnecti
   def withLocalAddress(newValue: Optional[InetSocketAddress]): ClientConnectionSettings =
     self.copy(localAddress = newValue.toScala)
 
+  /**
+   * Java API
+   * @since 2.0.0
+   */
+  def withStrictResponseEntityTimeout(newValue: Optional[JDuration]): ClientConnectionSettings =
+    self.copy(strictResponseEntityTimeout = newValue.toScala.map(_.toScala))
+
   @ApiMayChange
   def withTransport(newValue: ClientTransport): ClientConnectionSettings = self.copy(transport = newValue.asScala)
 }

http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala

@@ -47,6 +47,21 @@ abstract class ClientConnectionSettings private[pekko] ()
   def parserSettings: ParserSettings
   def logUnencryptedNetworkBytes: Option[Int]
   def streamCancellationDelay: FiniteDuration
+
+  /**
+   * If defined, response entities are collected into `HttpEntity.Strict` entities (using the given timeout)
+   * before the response is dispatched to the application. Only supported by the HTTP/1.1 client.
+   *
+   * @since 2.0.0
+   */
+  def strictResponseEntityTimeout: Option[FiniteDuration]
+
+  /**
+   * The maximum number of bytes collected per response entity when [[strictResponseEntityTimeout]] is defined.
+   *
+   * @since 2.0.0
+   */
+  def strictResponseEntityMaxBytes: Long
   def localAddress: Option[InetSocketAddress]
   def http2Settings: Http2ClientSettings
 
@@ -64,6 +79,14 @@ abstract class ClientConnectionSettings private[pekko] ()
   def withStreamCancellationDelay(newValue: FiniteDuration): ClientConnectionSettings =
     self.copy(streamCancellationDelay = newValue)
 
+  /** @since 2.0.0 */
+  def withStrictResponseEntityTimeout(newValue: Option[FiniteDuration]): ClientConnectionSettings =
+    self.copy(strictResponseEntityTimeout = newValue)
+
+  /** @since 2.0.0 */
+  override def withStrictResponseEntityMaxBytes(newValue: Long): ClientConnectionSettings =
+    self.copy(strictResponseEntityMaxBytes = newValue)
+
   // overloads for idiomatic Scala use
   def withWebsocketSettings(newValue: WebSocketSettings): ClientConnectionSettings =
     self.copy(websocketSettings = newValue)

http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala

@@ -105,6 +105,26 @@ class HttpConfigurationSpec extends PekkoSpec {
       }
     }
 
+    "have strict response entities disabled by default" in {
+      ClientConnectionSettings(system).strictResponseEntityTimeout should ===(None)
+      ClientConnectionSettings(system).strictResponseEntityMaxBytes should ===(8L * 1024 * 1024)
+    }
+
+    "set `pekko.http.client.strict-response-entity-timeout`" in {
+      configuredSystem("""pekko.http.client.strict-response-entity-timeout = 5s
+                         |pekko.http.client.strict-response-entity-max-bytes = 1m""".stripMargin) { sys =>
+        import scala.concurrent.duration._
+
+        val client = ClientConnectionSettings(sys)
+        client.strictResponseEntityTimeout should ===(Some(5.seconds))
+        client.strictResponseEntityMaxBytes should ===(1024L * 1024)
+
+        val pool = ConnectionPoolSettings(sys)
+        pool.connectionSettings.strictResponseEntityTimeout should ===(Some(5.seconds))
+        pool.connectionSettings.strictResponseEntityMaxBytes should ===(1024L * 1024)
+      }
+    }
+
     "change parser settings for all by setting `pekko.http.parsing`" in {
       configuredSystem("""pekko.http.parsing.illegal-header-warnings = off""") { sys =>
         val client = ClientConnectionSettings(sys)