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 20fb7386e fix: fail requests still queued in a pool interface that
stops (#1282)
20fb7386e is described below
commit 20fb7386e12a988e916b78df29ee2699377a44b2
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 09:41:47 2026 +0100
fix: fail requests still queued in a pool interface that stops (#1282)
Motivation:
`PoolInterface.Logic` keeps requests that the pool cannot take yet in an
`ArrayDeque` of `RequestContext`s, each holding the promise that the
caller of `Http().singleRequest` (or of a cached host connection pool
flow) is waiting on.
When the stage stops without having dispatched them - the pool flow
failing or cancelling, or the materializer shutting down - `postStop`
only failed `shutdownPromise`. Nothing else holds those queued contexts:
they were never handed to a connection, so no response, no error and no
retry through the pool master can ever reach them, and the callers'
`Future[HttpResponse]`s stay uncompleted for the lifetime of the
process.
Modification:
Drain the buffer in `postStop` and fail every queued response promise
with the same `IllegalStateException` that is used for the shutdown
promise.
Result:
A pool that stops unexpectedly fails the requests it never dispatched
instead of leaving their callers waiting forever.
Tests:
- sbt "http-core/testOnly
org.apache.pekko.http.impl.engine.client.PoolInterfaceSpec" - pass (1 test).
Reverting only the `postStop` change makes it fail, as the queued promises are
never completed.
- sbt "http-core/mimaReportBinaryIssues" - pass.
- scalafmt --mode diff-ref=upstream/main --test - pass.
References:
None - found while auditing `src/main` for resource leaks
---
.../http/impl/engine/client/PoolInterface.scala | 9 ++-
.../impl/engine/client/PoolInterfaceSpec.scala | 66 ++++++++++++++++++++++
2 files changed, 74 insertions(+), 1 deletion(-)
diff --git
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolInterface.scala
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolInterface.scala
index 107aeab04..a787c90d7 100644
---
a/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolInterface.scala
+++
b/http-core/src/main/scala/org/apache/pekko/http/impl/engine/client/PoolInterface.scala
@@ -219,7 +219,14 @@ private[http] object PoolInterface {
!shuttingDown && remainingRequested == 0 && idleTimeout.isFinite &&
hcps.setup.settings.minConnections == 0
override def onUpstreamFailure(ex: Throwable): Unit =
shutdownPromise.tryFailure(ex)
- override def postStop(): Unit = shutdownPromise.tryFailure(new
IllegalStateException("Pool shutdown unexpectedly"))
+ override def postStop(): Unit = {
+ val shutdownException = new IllegalStateException("Pool shutdown
unexpectedly")
+ shutdownPromise.tryFailure(shutdownException)
+ // Whatever is still queued here was never dispatched to a connection,
so nothing else is ever going to complete
+ // these promises. Fail them instead of leaving the callers waiting for
a response forever.
+ while (!buffer.isEmpty)
+ buffer.removeFirst().responsePromise.tryFailure(shutdownException)
+ }
// PoolInterface implementations
override def request(request: HttpRequest, responsePromise:
Promise[HttpResponse]): Unit =
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/PoolInterfaceSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/PoolInterfaceSpec.scala
new file mode 100644
index 000000000..f359cf467
--- /dev/null
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/PoolInterfaceSpec.scala
@@ -0,0 +1,66 @@
+/*
+ * 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.
+ */
+
+package org.apache.pekko.http.impl.engine.client
+
+import org.apache.pekko
+import pekko.http.impl.engine.client.PoolFlow.{ RequestContext,
ResponseContext }
+import pekko.http.impl.settings.{ ConnectionPoolSetup, HostConnectionPoolSetup
}
+import pekko.http.impl.util._
+import pekko.http.scaladsl.Http
+import pekko.http.scaladsl.model.{ HttpRequest, HttpResponse }
+import pekko.http.scaladsl.settings.ConnectionPoolSettings
+import pekko.stream.scaladsl.{ Flow, Sink, Source }
+import pekko.stream.testkit.{ TestPublisher, TestSubscriber }
+import pekko.testkit._
+
+import scala.concurrent.{ Await, Promise }
+import scala.concurrent.duration._
+
+class PoolInterfaceSpec extends PekkoSpecWithMaterializer {
+
+ "The pool interface" should {
+
+ "fail requests that are still queued when the pool stops" in {
+ val settings =
ConnectionPoolSettings(system).withMaxConnections(1).withMaxOpenRequests(4)
+ // nothing is ever dispatched to this endpoint, it only identifies the
pool
+ val poolId = new PoolId(
+ HostConnectionPoolSetup("127.0.0.1", 1, ConnectionPoolSetup(settings,
log = system.log)),
+ PoolId.newUniquePool())
+
+ val requestsToPool = TestSubscriber.probe[RequestContext]()
+ val responsesFromPool = TestPublisher.probe[ResponseContext]()
+ val poolInterface =
+ Flow.fromGraph(
+ new PoolInterface.PoolInterfaceStage(poolId, Http().poolMaster,
settings.maxOpenRequests, system.log))
+ .join(Flow.fromSinkAndSource(Sink.fromSubscriber(requestsToPool),
Source.fromPublisher(responsesFromPool)))
+ .run()
+
+ // the pool never signals demand, so these requests only ever reach the
interface's buffer
+ val queuedRequests = Vector.fill(3)(Promise[HttpResponse]())
+ queuedRequests.foreach(poolInterface.request(HttpRequest(uri = "/"), _))
+
+ // the pool flow going away takes the interface with it
+ responsesFromPool.sendComplete()
+ requestsToPool.expectSubscriptionAndComplete()
+
+ queuedRequests.foreach { responsePromise =>
+ Await.result(responsePromise.future.failed, 3.seconds.dilated)
shouldBe an[IllegalStateException]
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
