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 9861865b3 fix: cancel the pending slot state timeout when a pool slot
is shut down (#1283)
9861865b3 is described below
commit 9861865b37710310b1cbce45a324505e341eaf9a
Author: PJ Fanning <[email protected]>
AuthorDate: Fri Sep 11 09:43:29 2026 +0100
fix: cancel the pending slot state timeout when a pool slot is shut down
(#1283)
Motivation:
Every state transition in `NewHostConnectionPool`'s slot state machine
cancels the timeout of the state it leaves and, if the new state defines
one, schedules a fresh `materializer.scheduleOnce` task for it. Those
tasks are not bound to the stage lifecycle, so they keep running after
the stage has stopped.
`Slot.shutdown()`, which `postStop` calls for every slot, closed the
connection and ran `state.onShutdown` but never cancelled that timeout.
A pool that stops while a slot is in a state with a finite timeout -
waiting for a connection, for a response entity subscription, or for the
keep-alive timeout - therefore leaves a scheduled task behind that keeps
the slot, and through it the whole pool logic with its connections and
request contexts, reachable until the timeout elapses. Firing does
nothing useful either: the task only invokes an async callback on a
stage that is already gone.
Modification:
Cancel the current timeout at the start of `Slot.shutdown()`, like every
state transition and the slot error path already do.
Result:
Shutting a pool down no longer leaves scheduled state timeouts, and the
pool becomes collectable as soon as it stops.
Tests:
- sbt "http-core/testOnly
org.apache.pekko.http.impl.engine.client.pool.NewHostConnectionPoolSpec" - pass
(1 test). Reverting only the `cancelCurrentTimeout()` call makes it fail, as
the scheduled task is still live after the pool stopped.
- sbt "http-core/mimaReportBinaryIssues" - pass.
- scalafmt --mode diff-ref=upstream/main --test - pass.
References:
None - found while auditing `src/main` for resource leaks
---
.../engine/client/pool/NewHostConnectionPool.scala | 3 +
.../client/pool/NewHostConnectionPoolSpec.scala | 110 +++++++++++++++++++++
2 files changed, 113 insertions(+)
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 dd7e1fcdb..d125bb244 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
@@ -226,6 +226,9 @@ private[client] object NewHostConnectionPool {
def isIdle: Boolean = state.isIdle
def isConnected: Boolean = state.isConnected
def shutdown(): Unit = {
+ // the state timeout is scheduled on the materializer, so it
outlives this stage unless it is cancelled here
+ cancelCurrentTimeout()
+
// if the connection is idle, we just complete it regularly,
otherwise, we forcibly tear it down
// with an error (which will be logged in
OutgoingConnectionBlueprint, see `mapError` there).
val reason =
diff --git
a/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPoolSpec.scala
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPoolSpec.scala
new file mode 100644
index 000000000..139ac9d68
--- /dev/null
+++
b/http-core/src/test/scala/org/apache/pekko/http/impl/engine/client/pool/NewHostConnectionPoolSpec.scala
@@ -0,0 +1,110 @@
+/*
+ * 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.pool
+
+import java.net.InetSocketAddress
+import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.ThreadFactory
+
+import com.typesafe.config.Config
+
+import org.apache.pekko
+import pekko.actor.{ Cancellable, LightArrayRevolverScheduler }
+import pekko.event.LoggingAdapter
+import pekko.http.impl.engine.client.PoolFlow.{ RequestContext,
ResponseContext }
+import pekko.http.impl.util._
+import pekko.http.scaladsl.Http
+import pekko.http.scaladsl.model.{ ContentTypes, HttpEntity, HttpRequest,
HttpResponse }
+import pekko.http.scaladsl.settings.ConnectionPoolSettings
+import pekko.stream.scaladsl.{ Flow, Keep, Sink, Source }
+import pekko.stream.testkit.{ TestPublisher, TestSubscriber }
+import pekko.stream.testkit.scaladsl.{ TestSink, TestSource }
+import pekko.util.ByteString
+
+import scala.concurrent.{ ExecutionContext, Future, Promise }
+import scala.concurrent.duration._
+import scala.jdk.CollectionConverters._
+
+object NewHostConnectionPoolSpec {
+
+ /** A delay no other part of the machinery schedules with, so that slot
state timeouts can be told apart */
+ val SlotStateTimeout = 7331.millis
+
+ val slotStateTimeouts = new CopyOnWriteArrayList[Cancellable]
+
+ /** Records the tasks scheduled for [[SlotStateTimeout]] so that a test can
check whether they are cancelled */
+ final class TrackingScheduler(config: Config, log: LoggingAdapter,
threadFactory: ThreadFactory)
+ extends LightArrayRevolverScheduler(config, log, threadFactory) {
+ override def scheduleOnce(delay: FiniteDuration, runnable: Runnable)(
+ implicit executor: ExecutionContext): Cancellable = {
+ val cancellable = super.scheduleOnce(delay, runnable)
+ if (delay == SlotStateTimeout) slotStateTimeouts.add(cancellable)
+ cancellable
+ }
+ }
+}
+
+class NewHostConnectionPoolSpec extends PekkoSpecWithMaterializer(
+ """
+ pekko.scheduler.implementation =
"org.apache.pekko.http.impl.engine.client.pool.NewHostConnectionPoolSpec$TrackingScheduler"
+ """) {
+ import NewHostConnectionPoolSpec._
+
+ "The host connection pool" should {
+
+ "cancel a pending slot state timeout when the pool is shut down" in {
+ slotStateTimeouts.clear()
+
+ val settings =
+ ConnectionPoolSettings(system)
+ .withMaxConnections(1)
+ .withMinConnections(0)
+ .withResponseEntitySubscriptionTimeout(SlotStateTimeout)
+
+ val connectionRequests = TestSubscriber.probe[HttpRequest]()
+ val connectionResponses = TestPublisher.probe[HttpResponse]()
+ val connectionFlow =
+ Flow.fromSinkAndSource(Sink.fromSubscriber(connectionRequests),
Source.fromPublisher(connectionResponses))
+ .mapMaterializedValue(_ =>
Future.successful(Http.OutgoingConnection(address, address)))
+
+ val (requestsIn, responsesOut) =
+ TestSource[RequestContext]()
+ .via(NewHostConnectionPool(connectionFlow, settings, system.log))
+ .toMat(TestSink[ResponseContext]())(Keep.both)
+ .run()
+
+ responsesOut.request(1)
+ requestsIn.sendNext(RequestContext(HttpRequest(uri = "/"),
Promise[HttpResponse](), 0))
+
+ connectionRequests.requestNext()
+ // a response whose entity is never subscribed to, which is what puts
the slot on a state timeout
+ connectionResponses.sendNext(
+ HttpResponse(entity =
HttpEntity.Chunked.fromData(ContentTypes.`text/plain(UTF-8)`,
Source.maybe[ByteString])))
+ responsesOut.expectNext()
+
+ awaitCond(slotStateTimeouts.size == 1)
+
+ // tear the pool down while the slot is still waiting for that
subscription
+ responsesOut.cancel()
+
+ awaitCond(slotStateTimeouts.asScala.forall(_.isCancelled))
+ }
+ }
+
+ private def address = new InetSocketAddress("127.0.0.1", 1)
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
