From dc1185dbfb1024c90fbcfae93720925737c691a3 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 11:20:48 +0100 Subject: [PATCH 1/3] feat: optional strict response entities for the HTTP/1.1 client Motivation: Client responses are dispatched with streamed entities, so applications always have to consume or discard the entity before the connection can be reused. Applications that only ever work with fully buffered responses have to call `toStrict` on every response themselves. Modification: Add `pekko.http.client.strict-response-entity-timeout` (`off` by default) and `pekko.http.client.strict-response-entity-max-bytes` (8m by default). When a timeout is configured, `OutgoingConnectionBlueprint` collects every response entity into an `HttpEntity.Strict` before the response leaves the connection layer, which also covers the connection pool behind the host-level and request-level APIs. Entities that are already strict pass through untouched. Result: The HTTP/1.1 client can be configured to hand out strict response entities. The default behaviour is unchanged: entities stay streamed. Tests: - sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.LowLevelOutgoingConnectionSpec" - pass - sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.HttpConfigurationSpec" - pass - sbt http-core/mimaReportBinaryIssues - pass - sbt ++3.3.8 http-core/Test/compile - pass - sbt docs/paradox - pass - scalafmt --list --mode diff-ref=origin/main - no changes References: None - follow-up on making client response entity handling configurable --- .../main/paradox/client-side/configuration.md | 20 +++ http-core/src/main/resources/reference.conf | 24 ++++ .../client/OutgoingConnectionBlueprint.scala | 27 +++- .../ClientConnectionSettingsImpl.scala | 11 ++ .../settings/ClientConnectionSettings.scala | 25 ++++ .../settings/ClientConnectionSettings.scala | 23 ++++ .../engine/client/HttpConfigurationSpec.scala | 20 +++ .../LowLevelOutgoingConnectionSpec.scala | 115 ++++++++++++++++++ 8 files changed, 264 insertions(+), 1 deletion(-) diff --git a/docs/src/main/paradox/client-side/configuration.md b/docs/src/main/paradox/client-side/configuration.md index 3e1066f897..521a20721e 100644 --- a/docs/src/main/paradox/client-side/configuration.md +++ b/docs/src/main/paradox/client-side/configuration.md @@ -20,6 +20,26 @@ 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 setting only applies to the HTTP/1.1 client, which includes the connection pool backing +@ref[request-level](request-level.md) and @ref[host-level](host-level.md) APIs. + ## Pool Settings Pool settings influence the behavior of client connection pools as used with APIs like `Http.singleRequest` diff --git a/http-core/src/main/resources/reference.conf b/http-core/src/main/resources/reference.conf index 1c7019879b..534a7ec7b1 100644 --- a/http-core/src/main/resources/reference.conf +++ b/http-core/src/main/resources/reference.conf @@ -447,6 +447,30 @@ 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 setting only applies to the HTTP/1.1 client (which includes the + # connection pool used by `singleRequest` and the host-level API). + 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. # diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala index 6672d74461..9be002b48e 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala @@ -29,7 +29,7 @@ import pekko.stream._ import pekko.stream.scaladsl._ import pekko.http.scaladsl.Http import pekko.http.scaladsl.model.headers -import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse, IllegalResponseException, ResponseEntity } +import pekko.http.scaladsl.model.{ HttpEntity, HttpRequest, HttpResponse, IllegalResponseException, ResponseEntity } import pekko.http.impl.engine.rendering.{ HttpRequestRendererFactory, RequestRenderingContext } import pekko.http.impl.engine.parsing._ import pekko.http.impl.util._ @@ -103,6 +103,7 @@ private[http] object OutgoingConnectionBlueprint { val responsePrep = Flow[List[ParserOutput.ResponseOutput]] .mapConcat(ConstantFun.scalaIdentityFunction) .via(new PrepareResponse(parserSettings)) + .via(strictifyResponseEntities(settings)) val terminationFanout = b.add(Broadcast[HttpResponse](2)) @@ -137,6 +138,30 @@ private[http] object OutgoingConnectionBlueprint { logTLSBidiBySetting("client-plain-text", settings.logUnencryptedNetworkBytes)) } + /** + * Collects response entities into `HttpEntity.Strict` entities if + * `pekko.http.client.strict-response-entity-timeout` is configured, otherwise passes responses through unchanged. + * + * Responses on a single HTTP/1.1 connection are strictly sequential, so collecting the entity of one response + * before the next one is emitted does not hold up any other response. + */ + private def strictifyResponseEntities( + settings: ClientConnectionSettings): Flow[HttpResponse, HttpResponse, NotUsed] = + settings.strictResponseEntityTimeout match { + case None => Flow[HttpResponse] + case Some(timeout) => + val maxBytes = settings.strictResponseEntityMaxBytes + Flow[HttpResponse].flatMapConcat { response => + 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)) + } + }.named("strictifyResponseEntities") + } + // a simple merge stage that simply forwards its first input and ignores its second input // (the terminationBackchannelInput), but applies a special completion handling private object TerminationMerge diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala index 9da508eac8..439c81602e 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/settings/ClientConnectionSettingsImpl.scala @@ -44,6 +44,8 @@ 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) @@ -51,6 +53,10 @@ private[pekko] final case class ClientConnectionSettingsImpl( 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) diff --git a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala index fcb567be86..2976077c8f 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/javadsl/settings/ClientConnectionSettings.scala +++ b/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) } diff --git a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala b/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala index 69c99aa3b1..0b7960c097 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/scaladsl/settings/ClientConnectionSettings.scala +++ b/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) diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala index a23cd996b7..d309f1e3d5 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HttpConfigurationSpec.scala +++ b/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) diff --git a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala index fe48e62051..ad0cd246d2 100644 --- a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala +++ b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/LowLevelOutgoingConnectionSpec.scala @@ -1003,6 +1003,121 @@ class LowLevelOutgoingConnectionSpec extends PekkoSpecWithMaterializer with Insi netInSub.sendComplete() responses.expectComplete() } + + "collect response entities into strict entities if configured".which { + val strictConfig = "pekko.http.client.strict-response-entity-timeout = 2s" + + "collects a chunked response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |4 + |DEFG + |0 + | + |""") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "collects a default response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Content-Length: 7 + | + |ABC""") + sendWireData("DEFG") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "collects a close-delimited response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + | + |ABCDEFG""") + closeNetworkInput() + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + } + + "passes through an already strict response entity" in new TestSetup(config = strictConfig) { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Content-Length: 7 + | + |ABCDEFG""") + + expectResponse().entity shouldEqual + HttpEntity.Strict(ContentTypes.`text/plain(UTF-8)`, ByteString("ABCDEFG")) + + requestsSub.sendComplete() + netOut.expectComplete() + netInSub.sendComplete() + responses.expectComplete() + } + + "fails the connection if the response entity exceeds strict-response-entity-max-bytes" in new TestSetup( + config = strictConfig + "\npekko.http.client.strict-response-entity-max-bytes = 5") { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |4 + |DEFG + |0 + | + |""") + + responsesSub.request(1) + responses.expectError().getMessage should include("longer than the maximum of 5") + } + + "leaves response entities streamed by default" in new TestSetup { + sendStandardRequest() + sendWireData( + """HTTP/1.1 200 OK + |Content-Type: text/plain; charset=UTF-8 + |Transfer-Encoding: chunked + | + |3 + |ABC + |""") + + inside(expectResponse()) { + case HttpResponse(_, _, entity: HttpEntity.Chunked, _) => + entity.contentType shouldEqual ContentTypes.`text/plain(UTF-8)` + } + } + } } class TestSetup(maxResponseContentLength: Int = -1, config: String = "") { From 440b7f56a6cfdd9d7b0f925fff4fcb6c0e1a0640 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 11:37:25 +0100 Subject: [PATCH 2/3] feat: extend strict response entities to the HTTP/2 client Motivation: `pekko.http.client.strict-response-entity-timeout` only affected the HTTP/1.1 client, so HTTP/2 users could not get the same behaviour. Modification: Move the strictify flow into `StreamUtils.strictifyResponseEntities`, taking a parallelism, and apply it in `Http2Blueprint.httpLayerClient` as well. The HTTP/1.1 client keeps parallelism 1, since responses on a connection are sequential anyway; the HTTP/2 client uses `max-concurrent-streams` so that a big response does not delay smaller ones on other streams. Document the HTTP/2 specific trade-offs in `reference.conf` and in the client configuration docs: responses are emitted in the order their entities complete, worst case memory is `max-concurrent-streams` * `strict-response-entity-max-bytes`, a failing entity fails the whole connection with every stream in flight on it, and entity data is read at the peer's pace rather than the application's. Result: The setting now covers both the HTTP/1.1 and the HTTP/2 client, with the HTTP/2 caveats spelled out. Tests: - sbt "http2-tests/testOnly org.apache.pekko.http.impl.engine.http2.Http2ClientSpec" - pass - sbt "http2-tests/testOnly org.apache.pekko.http.impl.engine.http2.Http2ClientServerSpec org.apache.pekko.http.impl.engine.http2.Http2PersistentClientTlsSpec org.apache.pekko.http.impl.engine.http2.Http2PersistentClientPlaintextSpec" - pass - sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.LowLevelOutgoingConnectionSpec org.apache.pekko.http.impl.engine.client.HttpConfigurationSpec" - pass - sbt http-core/mimaReportBinaryIssues - pass - sbt ++3.3.8 http-core/Test/compile http2-tests/Test/compile - pass - sbt docs/paradox - pass - scalafmt --list --mode diff-ref=upstream/main - no changes References: None - follow-up to the HTTP/1.1 support in the previous commit --- .../main/paradox/client-side/configuration.md | 25 +++++- http-core/src/main/resources/reference.conf | 14 +++- .../client/OutgoingConnectionBlueprint.scala | 33 ++------ .../impl/engine/http2/Http2Blueprint.scala | 22 +++++- .../pekko/http/impl/util/StreamUtils.scala | 29 ++++++- .../impl/engine/http2/Http2ClientSpec.scala | 79 +++++++++++++++++++ 6 files changed, 170 insertions(+), 32 deletions(-) diff --git a/docs/src/main/paradox/client-side/configuration.md b/docs/src/main/paradox/client-side/configuration.md index 521a20721e..609f6a944c 100644 --- a/docs/src/main/paradox/client-side/configuration.md +++ b/docs/src/main/paradox/client-side/configuration.md @@ -37,8 +37,29 @@ configured duration fails with a `TimeoutException`, and one that exceeds cases the connection is failed. Trailing headers of chunked responses are dropped, as they are with `HttpEntity.toStrict`. -This setting only applies to the HTTP/1.1 client, which includes the connection pool backing -@ref[request-level](request-level.md) and @ref[host-level](host-level.md) APIs. +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 diff --git a/http-core/src/main/resources/reference.conf b/http-core/src/main/resources/reference.conf index 534a7ec7b1..d87810e2c8 100644 --- a/http-core/src/main/resources/reference.conf +++ b/http-core/src/main/resources/reference.conf @@ -461,8 +461,18 @@ pekko.http { # Trailing headers of chunked responses are dropped, as they are with # `HttpEntity.toStrict`. # - # This setting only applies to the HTTP/1.1 client (which includes the - # connection pool used by `singleRequest` and the host-level API). + # 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 diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala index 9be002b48e..df83567871 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala +++ b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/OutgoingConnectionBlueprint.scala @@ -29,7 +29,7 @@ import pekko.stream._ import pekko.stream.scaladsl._ import pekko.http.scaladsl.Http import pekko.http.scaladsl.model.headers -import pekko.http.scaladsl.model.{ HttpEntity, HttpRequest, HttpResponse, IllegalResponseException, ResponseEntity } +import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse, IllegalResponseException, ResponseEntity } import pekko.http.impl.engine.rendering.{ HttpRequestRendererFactory, RequestRenderingContext } import pekko.http.impl.engine.parsing._ import pekko.http.impl.util._ @@ -103,7 +103,12 @@ private[http] object OutgoingConnectionBlueprint { val responsePrep = Flow[List[ParserOutput.ResponseOutput]] .mapConcat(ConstantFun.scalaIdentityFunction) .via(new PrepareResponse(parserSettings)) - .via(strictifyResponseEntities(settings)) + .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)) @@ -138,30 +143,6 @@ private[http] object OutgoingConnectionBlueprint { logTLSBidiBySetting("client-plain-text", settings.logUnencryptedNetworkBytes)) } - /** - * Collects response entities into `HttpEntity.Strict` entities if - * `pekko.http.client.strict-response-entity-timeout` is configured, otherwise passes responses through unchanged. - * - * Responses on a single HTTP/1.1 connection are strictly sequential, so collecting the entity of one response - * before the next one is emitted does not hold up any other response. - */ - private def strictifyResponseEntities( - settings: ClientConnectionSettings): Flow[HttpResponse, HttpResponse, NotUsed] = - settings.strictResponseEntityTimeout match { - case None => Flow[HttpResponse] - case Some(timeout) => - val maxBytes = settings.strictResponseEntityMaxBytes - Flow[HttpResponse].flatMapConcat { response => - 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)) - } - }.named("strictifyResponseEntities") - } - // a simple merge stage that simply forwards its first input and ignores its second input // (the terminationBackchannelInput), but applies a special completion handling private object TerminationMerge diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala index 31454417f6..9d0cb77b27 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/http2/Http2Blueprint.scala +++ b/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 { diff --git a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala b/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala index 85479451c6..d12346beb7 100644 --- a/http-core/src/main/scala/org/apache/pekko/http/impl/util/StreamUtils.scala +++ b/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 => /** diff --git a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala index d17c0185f3..82e3346787 100644 --- a/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala +++ b/http2-tests/src/test/scala/org/apache/pekko/http/impl/engine/http2/Http2ClientSpec.scala @@ -717,6 +717,85 @@ class Http2ClientSpec extends PekkoSpecWithMaterializer(""" } + "support strict response entities" should { + abstract class StrictEntitySetup extends TestSetup with NetProbes { + override def settings = + super.settings.withStrictResponseEntityTimeout(Some(2.seconds.dilated)) + } + + "collect a response entity into a strict entity".inAssertAllStagesStopped(new StrictEntitySetup { + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(streamId, endStream = false, ByteString("abc")) + network.sendDATA(streamId, endStream = true, ByteString("def")) + + user.expectResponse().entity shouldBe + HttpEntity.Strict(ContentTypes.`application/octet-stream`, ByteString("abcdef")) + }) + + "emit responses in the order in which their entities complete".inAssertAllStagesStopped( + new StrictEntitySetup { + user.emitRequest(Get("/slow")) + network.expectDecodedHEADERS(0x1, endStream = true) + user.emitRequest(Get("/fast")) + network.expectDecodedHEADERS(0x3, endStream = true) + + // the first response starts first but its entity stays incomplete + network.sendHEADERS(0x1, endStream = false, + Seq(RawHeader(":status", "200"), RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(0x1, endStream = false, ByteString("in")) + + // while the second response completes right away + network.sendHEADERS(0x3, endStream = false, + Seq(RawHeader(":status", "201"), RawHeader("content-type", "application/octet-stream"))) + network.sendDATA(0x3, endStream = true, ByteString("complete")) + + user.expectResponse().status shouldBe StatusCodes.Created + + network.sendDATA(0x1, endStream = true, ByteString("complete")) + user.expectResponse().status shouldBe StatusCodes.OK + }) + + "fail the connection if a response entity exceeds strict-response-entity-max-bytes".inAssertAllStagesStopped( + new StrictEntitySetup { + override def settings = super.settings.withStrictResponseEntityMaxBytes(5) + + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + EventFilter.error(pattern = "HTTP2 connection failed with error .*", occurrences = 1).intercept { + network.sendDATA(streamId, endStream = true, ByteString("abcdef")) + + user.responseIn.expectSubscriptionAndError().getMessage should include( + "longer than the maximum of 5") + } + }) + + "leave response entities streamed by default".inAssertAllStagesStopped(new TestSetup with NetProbes { + val streamId = 0x1 + user.emitRequest(Get("/")) + network.expectDecodedHEADERS(streamId, endStream = true) + + network.sendHEADERS(streamId, endStream = false, + Seq( + RawHeader(":status", "200"), + RawHeader("content-type", "application/octet-stream"))) + + user.expectResponse().entity shouldBe a[Chunked] + }) + } + "expose synthetic headers" should { "expose Tls-Session-Info".inAssertAllStagesStopped(new TestSetup { lazy val expectedSession = SSLContext.getDefault.createSSLEngine.getSession From 4609c7e5feee8309a47db4a9eb664ba2586b27f2 Mon Sep 17 00:00:00 2001 From: PJ Fanning Date: Tue, 25 Aug 2026 12:18:16 +0100 Subject: [PATCH 3/3] Create client-strict-entities.excludes --- .../client-strict-entities.excludes | 21 +++++++++++++++++++ 1 file changed, 21 insertions(+) create mode 100644 http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes diff --git a/http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes b/http-core/src/main/mima-filters/2.0.x.backwards.excludes/client-strict-entities.excludes new file mode 100644 index 0000000000..60ab6e579f --- /dev/null +++ b/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")