This is an automated email from the ASF dual-hosted git repository.
pjfanning pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko-http.git
The following commit(s) were added to refs/heads/main by this push:
new 69aa0303e fix: close pool connection after an early response, abort
the upload only if rejected (#1350)
69aa0303e is described below
commit 69aa0303eca72d7ac589e6557a7a8fe2c68304be
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Oct 23 17:58:18 2026 +0100
fix: close pool connection after an early response, abort the upload only
if rejected (#1350)
* fix: close pool connection after an early response
Motivation:
When the response completed before the streamed request entity was fully
sent, the pool slot went back to Idle once the request entity completed and
reused the connection. The client cannot tell whether the server read the
rest of the request entity, and a server that did not could misinterpret the
remaining data as the next request. The Pekko HTTP server already closes the
connection after an early response for the same reason.
Modification:
WaitingForEndOfRequestEntity now moves to ToBeClosed when the request entity
completes or fails, instead of returning the slot to Idle.
Result:
A connection that saw an early response is not reused; the next request goes
over a new connection.
Tests:
- sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.*" -
passed (179 succeeded, 0 failed)
- New and updated SlotStateSpec and HostConnectionPoolSpec tests fail
without the fix
References:
None - client pool hardening for early responses
* fix: abort request entity when the pool sees an early response
Motivation:
After an early response (the response completed before the streamed request
entity was fully sent), the pool kept sending the rest of the request entity
and only closed the connection once it was done. A server that does not read
the rest of the request entity could still misinterpret that data as a
further request. HTTP/1.1 suggests that a client stop sending the body when
the server indicates it does not want it.
Modification:
WaitingForEndOfResponseEntity moves straight to ToBeClosed when the response
entity completes while the request entity is still being sent. The pool now
wraps streamed request entities with a kill switch and aborts it when the
connection is closed, so the rest of the request entity is not sent and the
request is not framed as complete.
Result:
After an early response, the pool stops sending the request entity and
closes
the connection immediately instead of after the upload finishes.
Tests:
- sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.*" -
passed (178 succeeded, 0 failed)
- Updated SlotStateSpec and HostConnectionPoolSpec tests fail without the
change
- sbt "http-core/mimaReportBinaryIssues" - passed
References:
Refs #1349
* fix: only abort the request entity after an early rejecting response
Motivation:
Aborting the request entity after every early response truncates uploads to
servers that answer early (e.g. 200 or 202) but keep reading the request
entity. RFC 9112, section 9.3 only asks a client to stop sending when the
response indicates that the server does not want the body.
Modification:
After an early response, the pool closes the connection right away and
aborts
the request entity only when the response is an error (4xx/5xx) or carries
Connection: close. Otherwise, it finishes sending the request entity and
then
closes the connection without reusing it.
Result:
Uploads to servers that answer early with a success response complete as
before; uploads rejected early are cancelled.
Tests:
- sbt "http-core/testOnly org.apache.pekko.http.impl.engine.client.*" -
passed (182 succeeded, 0 failed)
- New success-response tests in SlotStateSpec and HostConnectionPoolSpec
fail without the change
- sbt "http-core/mimaReportBinaryIssues" - passed
References:
Refs #1350
---
.../engine/client/pool/NewHostConnectionPool.scala | 21 ++++++--
.../http/impl/engine/client/pool/SlotState.scala | 38 ++++++++++----
.../engine/client/HostConnectionPoolSpec.scala | 60 +++++++++++++++++++++
.../impl/engine/client/pool/SlotStateSpec.scala | 61 ++++++++++++++--------
4 files changed, 145 insertions(+), 35 deletions(-)
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
index eda9dff81..0eb3266aa 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPool.scala
@@ -457,6 +457,7 @@ private[client] object NewHostConnectionPool {
responseIn: SubSinkInlet[HttpResponse]) extends InHandler with
OutHandler { connection =>
var ongoingResponseEntity: Option[HttpEntity] = None
var ongoingResponseEntityKillSwitch: Option[KillSwitch] = None
+ var ongoingRequestEntityKillSwitch: Option[KillSwitch] = None
var connectionEstablished: Boolean = false
/** Will only be executed if this connection is still the current
connection for its slot */
@@ -468,10 +469,16 @@ private[client] object NewHostConnectionPool {
request.entity match {
case _: HttpEntity.Strict => request
case e =>
- val (newEntity, entityComplete) =
HttpEntity.captureTermination(request.entity)
- entityComplete.onComplete(safely {
- case Success(_) =>
withSlot(_.onRequestEntityCompleted())
- case Failure(cause) =>
withSlot(_.onRequestEntityFailed(cause))
+ val (newEntity, StreamUtils.StreamControl(_, entityComplete,
entityKillSwitch)) =
+ StreamUtils.transformEntityStream(request.entity,
+ StreamUtils.CaptureMaterializationAndTerminationOp)
+ ongoingRequestEntityKillSwitch = entityKillSwitch
+ entityComplete.onComplete(safely { (res: Try[Unit]) =>
+ ongoingRequestEntityKillSwitch = None
+ res match {
+ case Success(_) =>
withSlot(_.onRequestEntityCompleted())
+ case Failure(cause) =>
withSlot(_.onRequestEntityFailed(cause))
+ }
})(ExecutionContext.parasitic)
request.withEntity(newEntity)
}
@@ -497,6 +504,12 @@ private[client] object NewHostConnectionPool {
failure.getOrElse(new IllegalStateException("Connection was
closed while response was still in-flight"))
ongoingResponseEntity.foreach(_.dataBytes.runWith(Sink.cancelled)(subFusingMaterializer))
ongoingResponseEntityKillSwitch.foreach(_.abort(exception))
+ // Abort rather than complete a request entity that is still being
sent, so that the remaining data is
+ // never sent and the request is not framed as complete.
+ ongoingRequestEntityKillSwitch.foreach(_.abort(failure.getOrElse(
+ new IllegalStateException("Connection was closed while request
entity was still being sent")
+ with NoStackTrace)))
+ ongoingRequestEntityKillSwitch = None
}
def isClosed: Boolean = requestOut.isClosed || responseIn.isClosed
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/SlotState.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/SlotState.scala
index 0f45cec2c..0af14351d 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/SlotState.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/pool/SlotState.scala
@@ -21,7 +21,7 @@ import pekko.annotation.InternalApi
import pekko.http.impl.engine.client.PoolFlow.RequestContext
import pekko.http.impl.util._
import pekko.http.scaladsl.Http
-import pekko.http.scaladsl.model.HttpResponse
+import pekko.http.scaladsl.model.{ headers, HttpResponse }
import pekko.http.scaladsl.settings.ConnectionPoolSettings
import pekko.macros.LogHelper
@@ -108,6 +108,15 @@ private[pool] sealed abstract class SlotState extends
Product {
*/
@InternalApi
private[pool] object SlotState {
+
+ /**
+ * Whether an early response indicates that the server does not want the
rest of the request entity: an error
+ * response or one that closes the connection. Not
`SlotContext.willCloseAfter`, which also covers the client-side
+ * max-connection-lifetime.
+ */
+ private def serverRejectsRequestEntity(response: HttpResponse): Boolean =
+ response.status.isFailure ||
response.header[headers.Connection].exists(_.hasClose)
+
sealed private[pool] /* to avoid warnings */ abstract class ConnectedState
extends SlotState {
def isConnected: Boolean = true
@@ -346,9 +355,16 @@ private[pool] object SlotState {
waitingForEndOfRequestEntity: Boolean) extends ConnectedState with
BusyWithResultAlreadyDetermined {
override def onResponseEntityCompleted(ctx: SlotContext): SlotState =
- if (waitingForEndOfRequestEntity)
- WaitingForEndOfRequestEntity
- else if (ctx.willCloseAfter(ongoingResponse) || ctx.isConnectionClosed)
+ if (waitingForEndOfRequestEntity) {
+ // The server sent an "early" response before the request entity was
fully sent. We cannot know whether the
+ // server will read the rest of the request entity, and a server that
does not could misinterpret the remaining
+ // data as the next request, so the connection is never reused.
+ // If the response indicates that the server does not want the rest of
the request entity, close the
+ // connection right away, which also aborts sending the rest of the
request entity (RFC 9112, section 9.3).
+ // Otherwise, the server may still be reading the request entity, so
finish sending it before closing.
+ if (serverRejectsRequestEntity(ongoingResponse) ||
ctx.isConnectionClosed) ToBeClosed
+ else WaitingForEndOfRequestEntity
+ } else if (ctx.willCloseAfter(ongoingResponse) || ctx.isConnectionClosed)
ToBeClosed // when would ctx.isConnectionClose be true? what that mean
that the connection has already failed before? do we need that state at all?
else
idle(ctx)
@@ -358,15 +374,17 @@ private[pool] object SlotState {
WaitingForEndOfResponseEntity(ongoingRequest, ongoingResponse,
waitingForEndOfRequestEntity = false)
}
}
+
+ /**
+ * An early response was received while the request entity was still being
sent, or the connection failed while the
+ * request entity was still being sent and the request has already been
dispatched for retry. The connection is not
+ * reused after the request entity completes.
+ */
case object WaitingForEndOfRequestEntity extends ConnectedState {
final override def isIdle = false
- override def onRequestEntityCompleted(ctx: SlotContext): SlotState =
- if (ctx.isConnectionClosed) ToBeClosed
- else idle(ctx)
- override def onRequestEntityFailed(ctx: SlotContext, cause: Throwable):
SlotState =
- if (ctx.isConnectionClosed) ToBeClosed // ignore error here
- else idle(ctx)
+ override def onRequestEntityCompleted(ctx: SlotContext): SlotState =
ToBeClosed
+ override def onRequestEntityFailed(ctx: SlotContext, cause: Throwable):
SlotState = ToBeClosed // ignore error here
override def onConnectionCompleted(ctx: SlotContext): SlotState =
ToBeClosed
override def onConnectionFailed(ctx: SlotContext, cause: Throwable):
SlotState = Failed(cause)
}
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HostConnectionPoolSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HostConnectionPoolSpec.scala
index 9b71936aa..3ba91081b 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HostConnectionPoolSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/HostConnectionPoolSpec.scala
@@ -172,6 +172,66 @@ class HostConnectionPoolSpec extends
PekkoSpecWithMaterializer(
conn2.pushResponse()
expectResponse()
})
+ if (clientServerImplementation == PassThrough) {
+ // the Pekko HTTP server engines close the connection themselves after
an early response,
+ // so only PassThrough shows whether the pool would reuse the
connection
+ "finish the request entity and not reuse the connection after an early
successful response".inWithShutdown(
+ new SetupWithServerProbes {
+ val reqEntityProbe = pushChunkedRequest()
+ responseOut.request(1)
+
+ val conn1 = expectNextConnection()
+ val reqInProbe = conn1.expectChunkedRequestBytesAsProbe()
+
+ reqEntityProbe.sendNext(ByteString("test"))
+ reqInProbe.expectUtf8EncodedString("test")
+
+ // the server responds before the request entity has been sent
completely
+ conn1.pushResponse(HttpResponse(entity = "early"))
+ expectResponseEntityAsString() shouldEqual "early"
+
+ // the server may still be reading, so the rest of the request
entity is sent
+ reqEntityProbe.sendNext(ByteString("test2"))
+ reqInProbe.expectUtf8EncodedString("test2")
+ reqEntityProbe.sendComplete()
+ reqInProbe.request(1)
+ reqInProbe.expectComplete()
+
+ // the next request must not be sent over the connection that saw
the early response
+ pushRequest(HttpRequest(uri = "/next"))
+ val conn2 = expectNextConnection()
+ conn2.expectRequestToPath("/next")
+ conn2.pushResponse(HttpResponse())
+ expectResponse()
+ })
+ "abort the request entity and not reuse the connection after an early
error response".inWithShutdown(
+ new SetupWithServerProbes {
+ val reqEntityProbe = pushChunkedRequest()
+ responseOut.request(1)
+
+ val conn1 = expectNextConnection()
+ val reqInProbe = conn1.expectChunkedRequestBytesAsProbe()
+
+ reqEntityProbe.sendNext(ByteString("test"))
+ reqInProbe.expectUtf8EncodedString("test")
+
+ // the server rejects the request before the request entity has
been sent completely
+ conn1.pushResponse(HttpResponse(StatusCodes.PayloadTooLarge,
entity = "too large"))
+ expectResponseEntityAsString() shouldEqual "too large"
+
+ // the rest of the request entity is not sent
+ reqEntityProbe.expectCancellation()
+ reqInProbe.request(1)
+ reqInProbe.expectError()
+
+ // the next request must not be sent over the connection that saw
the early response
+ pushRequest(HttpRequest(uri = "/next"))
+ val conn2 = expectNextConnection()
+ conn2.expectRequestToPath("/next")
+ conn2.pushResponse(HttpResponse())
+ expectResponse()
+ })
+ }
"open up to max-connections when enough requests are
pending".inWithShutdown(
new SetupWithServerProbes(_.withMaxConnections(2)) {
pushRequest(HttpRequest(uri = "/1"))
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/SlotStateSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/SlotStateSpec.scala
index 65c250c80..ae9f76569 100644
---
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/SlotStateSpec.scala
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/SlotStateSpec.scala
@@ -22,7 +22,7 @@ import pekko.http.impl.engine.client.pool.SlotState._
import pekko.http.scaladsl.Http
import pekko.http.scaladsl.model.ContentTypes
import pekko.http.scaladsl.model.HttpEntity
-import pekko.http.scaladsl.model.{ headers, HttpRequest, HttpResponse }
+import pekko.http.scaladsl.model.{ headers, HttpRequest, HttpResponse,
StatusCodes }
import pekko.http.scaladsl.settings.ConnectionPoolSettings
import pekko.stream.scaladsl.Source
import pekko.testkit.PekkoSpec
@@ -137,32 +137,51 @@ class SlotStateSpec extends PekkoSpec {
state should be(Idle(Duration.Inf))
}
- "consider a slot 'idle' only when the request has been successfully sent"
in {
- var state: SlotState = Unconnected
+ "finish sending the request and then close the connection after an early
successful response" in {
val context = new MockSlotContext(system.log)
- state = context.expectOpenConnection {
- state.onPreConnect(context)
- }
- state = state.onNewRequest(context, TheRequestContext)
+ var state = earlyResponseCompleted(context, HttpResponse())
+ // The response was completed before the request was sent completely (an
"early" response). The server may
+ // still be reading the request entity, so it is sent completely, but
the connection is not reused.
+ state should be(WaitingForEndOfRequestEntity)
- state = state.onConnectionAttemptSucceeded(context, outgoingConnection)
+ state = state.onRequestEntityCompleted(context)
+ state should be(ToBeClosed)
+ }
- state should be(PushingRequestToConnection(TheRequestContext))
- state = state.onRequestDispatched(context)
+ "close the connection when the request entity fails after an early
successful response" in {
+ val context = new MockSlotContext(system.log)
+ var state = earlyResponseCompleted(context, HttpResponse())
+ state should be(WaitingForEndOfRequestEntity)
- state = state.onResponseReceived(context, HttpResponse())
- state = state.onResponseDispatchable(context)
- state = state.onResponseEntitySubscribed(context)
- state = state.onResponseEntityCompleted(context)
- // We don't want this slot to become idle yet, because consuming the
request might still take
- // a while, in which case putting this request on another connection
would be lower latency.
- // In theory we could get higher throughput by allowing requests on busy
connections, but
- // there are also situations where it would have an adverse effect, so
better to be safe.
- state.isIdle should be(false)
+ state = state.onRequestEntityFailed(context, new
RuntimeException("request entity failed"))
+ state should be(ToBeClosed)
+ }
- state = state.onRequestEntityCompleted(context)
- state.isIdle should be(true)
+ "close the connection right away after an early error response" in {
+ earlyResponseCompleted(new MockSlotContext(system.log),
HttpResponse(StatusCodes.PayloadTooLarge)) should
+ be(ToBeClosed)
+ }
+
+ "close the connection right away after an early response that closes the
connection" in {
+ earlyResponseCompleted(new MockSlotContext(system.log),
+ HttpResponse(headers = headers.Connection("close") :: Nil)) should
be(ToBeClosed)
+ }
+ }
+
+ /** Runs a slot through a streamed request whose response entity completes
before the request entity */
+ private def earlyResponseCompleted(context: MockSlotContext, response:
HttpResponse): SlotState = {
+ var state: SlotState = Unconnected
+ state = context.expectOpenConnection {
+ state.onPreConnect(context)
}
+ state = state.onNewRequest(context, TheRequestContext)
+ state = state.onConnectionAttemptSucceeded(context, outgoingConnection)
+ state = state.onRequestDispatched(context)
+
+ state = state.onResponseReceived(context, response)
+ state = state.onResponseDispatchable(context)
+ state = state.onResponseEntitySubscribed(context)
+ state.onResponseEntityCompleted(context)
}
class MockSlotContext(
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
