This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch fix/tcp-outgoing-connection-retry-swallowed-exception-3135 in repository https://gitbox.apache.org/repos/asf/pekko.git
commit f366ac4702557ec516a159b8c7de7645a1027e30 Author: 虎鸣 <[email protected]> AuthorDate: Tue Jun 23 21:16:03 2026 +0800 fix: use Timers for finishConnect retry instead of raw scheduler callback 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]
