This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 72c92cc678 fix: use Timers for finishConnect retry instead of raw
scheduler callback (#3156)
72c92cc678 is described below
commit 72c92cc67822013a68e28152f0613e814bf214a0
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 23 22:53:14 2026 +0800
fix: use Timers for finishConnect retry instead of raw scheduler callback
(#3156)
Motivation:
TcpOutgoingConnection retries a failed finishConnect by scheduling a
callback via context.system.scheduler.scheduleOnce that calls
channelRegistry.register outside the actor's message loop. If that
callback throws, the exception is caught by the scheduler/dispatcher
and never re-enters the actor's supervision, so Tcp.CommandFailed is
never delivered and the commander can hang in the connecting state
indefinitely.
Modification:
Mix in the Timers trait and replace scheduler.scheduleOnce with
timers.startSingleTimer, which sends a private RetryFinishConnect
self-message processed inside the actor's message loop. The retry
handler is wrapped in reportConnectFailure so any exception surfaces
as CommandFailed to the commander. Timers are automatically cancelled
when the actor stops, removing the latent window where a stale
callback could register against a channel being torn down.
Result:
Exceptions thrown during the finishConnect retry path now correctly
surface as Tcp.CommandFailed to the commander, preventing indefinite
hangs in the connecting state.
Tests:
- sbt 'actor-tests / Test / testOnly org.apache.pekko.io.TcpConnectionSpec'
→ 35/35 passed
References:
Fixes #3135
---
.../org/apache/pekko/io/TcpConnectionSpec.scala | 42 ++++++++++++++++++++++
.../apache/pekko/io/TcpOutgoingConnection.scala | 15 +++++---
2 files changed, 52 insertions(+), 5 deletions(-)
diff --git
a/actor-tests/src/test/scala/org/apache/pekko/io/TcpConnectionSpec.scala
b/actor-tests/src/test/scala/org/apache/pekko/io/TcpConnectionSpec.scala
index abf3f1608f..5b0dc66100 100644
--- a/actor-tests/src/test/scala/org/apache/pekko/io/TcpConnectionSpec.scala
+++ b/actor-tests/src/test/scala/org/apache/pekko/io/TcpConnectionSpec.scala
@@ -697,6 +697,48 @@ class TcpConnectionSpec extends PekkoSpec("""
}
}
+ "report failed connection when finishConnect retries are exhausted" in
+ new UnacceptedConnectionTest() {
+ override lazy val connectionActor = createConnectionActor(serverAddress
= UnboundAddress)
+ run {
+ val sel = SelectorProvider.provider().openSelector()
+ try {
+ val key = clientSideChannel.register(sel, SelectionKey.OP_CONNECT |
SelectionKey.OP_READ)
+ sel.select(3000)
+ key.isConnectable should ===(true)
+ connectionActor.toString // force the lazy val
+ Thread.sleep(300)
+
+ // First ChannelConnectable: finishConnect() either throws
(connection refused)
+ // or returns false (connection still pending, schedules timer-based
retry).
+ selector.send(connectionActor, ChannelConnectable)
+
+ // If finishConnect returned false, wait for the timer-based retry
to re-register.
+ // Each retry fires after 1ms via timers.startSingleTimer
(self-message).
+ var retriesObserved = 0
+ var outcome: Option[AnyRef] = None
+ val deadline = System.nanoTime() + 5.seconds.toNanos
+ while (outcome.isEmpty && System.nanoTime() < deadline) {
+ registerCallReceiver.expectNoMessage(50.millis)
+ if (registerCallReceiver.msgAvailable) {
+ registerCallReceiver.receiveWhile(100.millis) {
+ case Registration(_, ops) if ops == OP_CONNECT =>
+ retriesObserved += 1
+ selector.send(connectionActor, ChannelConnectable)
+ }
+ }
+ if (userHandler.msgAvailable) {
+ outcome = Some(userHandler.expectMsgType[CommandFailed])
+ }
+ }
+
+ outcome shouldBe defined
+ watch(connectionActor)
+ expectTerminated(connectionActor)
+ } finally sel.close()
+ }
+ }
+
"close the connection when connection handler dies while connected" in new
EstablishedConnectionTest() {
run {
watch(connectionHandler.ref)
diff --git
a/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
b/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
index 4982f7061a..34c678ce20 100644
--- a/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
+++ b/actor/src/main/scala/org/apache/pekko/io/TcpOutgoingConnection.scala
@@ -20,7 +20,7 @@ import scala.concurrent.duration._
import scala.util.control.{ NoStackTrace, NonFatal }
import org.apache.pekko
-import pekko.actor.{ ActorRef, ReceiveTimeout }
+import pekko.actor.{ ActorRef, ReceiveTimeout, Timers }
import pekko.actor.Status.Failure
import pekko.annotation.InternalApi
import pekko.io.SelectionHandler._
@@ -42,7 +42,8 @@ private[io] class TcpOutgoingConnection(
extends TcpConnection(
_tcp,
SocketChannel.open().configureBlocking(false).asInstanceOf[SocketChannel],
- connect.pullMode) {
+ connect.pullMode)
+ with Timers {
import TcpOutgoingConnection._
import connect._
@@ -126,9 +127,7 @@ private[io] class TcpOutgoingConnection(
completeConnect(registration, commander, options)
} else {
if (remainingFinishConnectRetries > 0) {
- context.system.scheduler.scheduleOnce(1.millisecond) {
- channelRegistry.register(channel, SelectionKey.OP_CONNECT)
- }(context.dispatcher)
+ timers.startSingleTimer(RetryFinishConnect, RetryFinishConnect,
1.millisecond)
context.become(connecting(registration,
remainingFinishConnectRetries - 1))
} else {
log.debug(
@@ -138,6 +137,10 @@ private[io] class TcpOutgoingConnection(
}
}
}
+ case RetryFinishConnect =>
+ reportConnectFailure {
+ channelRegistry.register(channel, SelectionKey.OP_CONNECT)
+ }
case ReceiveTimeout =>
connectionTimeout()
}
@@ -155,6 +158,8 @@ private[io] object TcpOutgoingConnection {
val FinishConnectNeverReturnedTrueException =
new ConnectException("Could not establish connection because finishConnect
never returned true") with NoStackTrace
+ private case object RetryFinishConnect
+
def connectTimeoutExpired(timeout: Option[FiniteDuration]) =
new ConnectException(s"Connect timeout of $timeout expired") with
NoStackTrace
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
