SelfAwareJoinDecider: remove infinite timeout on http lookup

management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/ClusterBootstrap.scala

@@ -126,14 +126,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten
    *
    * We give the required selfContactPoint some time to be set asynchronously, or else log an error.
    */
-  @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit = system.scheduler.scheduleOnce(10.seconds) {
-    if (!selfContactPoint.isCompleted) {
-      _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out."))
-      log.error(
-        "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " +
-        "if binding bootstrap routes manually and not via pekko-management.")
+  @InternalApi private[bootstrap] def ensureSelfContactPoint(): Unit =
+    system.scheduler.scheduleOnce(ClusterBootstrap.SelfContactPointTimeout) {
+      if (!selfContactPoint.isCompleted) {
+        _selfContactPointUri.failure(new TimeoutException("Awaiting Bootstrap.selfContactPoint timed out."))
+        log.error(
+          "'Bootstrap.selfContactPoint' was NOT set, but is required for the bootstrap to work " +
+          "if binding bootstrap routes manually and not via pekko-management.")
+      }
     }
-  }
 
   /**
    * INTERNAL API
@@ -153,6 +154,15 @@ final class ClusterBootstrap(implicit system: ExtendedActorSystem) extends Exten
 
 object ClusterBootstrap extends ExtensionId[ClusterBootstrap] with ExtensionIdProvider {
 
+  /**
+   * INTERNAL API
+   *
+   * How long `selfContactPoint` is given to be set before the promise behind it is failed. Anything
+   * waiting on that promise has to outlast this, so it is shared rather than restated - see
+   * `SelfAwareJoinDecider.selfContactPointTimeout`.
+   */
+  @InternalApi private[bootstrap] val SelfContactPointTimeout: FiniteDuration = 10.seconds
+
   override def lookup: ClusterBootstrap.type = ClusterBootstrap
 
   override def get(system: ActorSystem): ClusterBootstrap = super.get(system)

management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/SelfAwareJoinDecider.scala

@@ -13,11 +13,15 @@
 
 package org.apache.pekko.management.cluster.bootstrap
 
+import java.util.concurrent.TimeoutException
+import java.util.concurrent.atomic.{ AtomicBoolean, AtomicReference }
+
 import org.apache.pekko
 import pekko.actor.ActorSystem
 import pekko.annotation.InternalApi
 import pekko.discovery.ServiceDiscovery.ResolvedTarget
 import pekko.event.{ LogSource, Logging }
+import pekko.http.scaladsl.model.Uri
 
 import scala.concurrent.Await
 import scala.concurrent.duration._
@@ -42,18 +46,59 @@ import scala.concurrent.duration._
   protected def contactPointString(contactPoint: ResolvedTarget): String =
     s"${contactPoint.host}:${contactPoint.port.getOrElse("0")}"
 
+  /**
+   * How long to block waiting for the self contact point to be set. `ClusterBootstrap` fails the
+   * promise itself after `ClusterBootstrap.SelfContactPointTimeout`, so this only has to outlast
+   * that; it is derived from it rather than restated so that the two cannot drift apart. Overridable
+   * for tests, which cannot afford to wait this long.
+   */
+  protected def selfContactPointTimeout: FiniteDuration = ClusterBootstrap.SelfContactPointTimeout * 3
+
+  // `decide` is called from the BootstrapCoordinator actor and guarded by its `decisionInProgress`
+  // flag, so in-tree callers are already serialised. `JoinDecider` is a user-pluggable interface
+  // though, and a custom one is free to resolve the contact point inside the Future it returns, so
+  // these stay atomic rather than plain vars.
+  //
+  // `cachedSelfContactPoint` holds the resolved value, `null` until there is one. Failures are not
+  // cached, so a contact point set later is still picked up.
+  private val cachedSelfContactPoint = new AtomicReference[(String, Int)]()
+  // Set by whichever caller takes the one permitted blocking wait; everyone else fails fast.
+  private val hasBlockedForSelfContactPoint = new AtomicBoolean(false)
+
   /**
    * The value `ClusterBootstrap(system).selfContactPoints` is set prior
    * to HTTP binding, during [[pekko.management.scaladsl.PekkoManagement.start()]], hence we
    * accept blocking on this initialization. If no value is received, the future will fail with
    * a `TimeoutException` and ClusterBootstrap will log an explanatory error to the user.
+   *
+   * A resolved value is cached. A failure is not, so that a contact point set later is still picked
+   * up, but the blocking wait is only ever paid once: once `ClusterBootstrap.start()` has run, the
+   * promise is always completed within `ClusterBootstrap.SelfContactPointTimeout`, so reaching
+   * the timeout at all means it never ran and no amount of further waiting will help.
    */
-  private[bootstrap] def selfContactPoint: (String, Int) =
-    Await.result(
-      ClusterBootstrap(system).selfContactPoint
-        .map(uri => (uri.authority.host.toString, uri.authority.port))(system.dispatcher),
-      Duration.Inf // the future has a timeout
-    )
+  private[bootstrap] def selfContactPoint: (String, Int) = {
+    val cached = cachedSelfContactPoint.get()
+    if (cached ne null) cached
+    else {
+      val pending = ClusterBootstrap(system).selfContactPoint
+      val uri = pending.value match {
+        case Some(completed)                                                  => completed.get
+        case None if hasBlockedForSelfContactPoint.compareAndSet(false, true) =>
+          Await.result(pending, selfContactPointTimeout)
+        case None =>
+          throw new TimeoutException(
+            "'Bootstrap.selfContactPoint' is still not set after waiting " +
+            s"[$selfContactPointTimeout] for it once. It is required for the bootstrap to work " +
+            "if binding bootstrap routes manually and not via pekko-management.")
+      }
+      val resolved = toContactPoint(uri)
+      cachedSelfContactPoint.compareAndSet(null, resolved)
+      resolved
+    }
+  }
+
+  private def toContactPoint(uri: Uri): (String, Int) =
+    (uri.authority.host.toString, uri.authority.port)
 
   /**
    * Determines whether it has the need and ability to join self and create a new cluster.
@@ -66,7 +111,7 @@ import scala.concurrent.duration._
         log.warning(
           BootstrapLogMarker.inProgress(info.contactPoints.map(contactPointString), info.allSeedNodes),
           "Self contact point [{}] not found in targets {}",
-          contactPointString(selfContactPoint),
+          contactPointString(self),
           info.contactPoints.mkString(", "))
       }
       false

management-cluster-bootstrap/src/main/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinator.scala

@@ -16,7 +16,7 @@ package org.apache.pekko.management.cluster.bootstrap.internal
 import java.time.LocalDateTime
 import java.util.concurrent.ThreadLocalRandom
 import scala.collection.immutable
-import scala.concurrent.ExecutionContext
+import scala.concurrent.{ ExecutionContext, Future }
 import org.apache.pekko
 import pekko.actor.Actor
 import pekko.actor.ActorRef
@@ -34,6 +34,7 @@ import pekko.pattern.pipe
 
 import scala.concurrent.duration._
 import scala.util.Try
+import scala.util.control.NonFatal
 import pekko.event.Logging
 import pekko.management.cluster.bootstrap.{
   BootstrapLogMarker,
@@ -392,8 +393,16 @@ private[pekko] class BootstrapCoordinator(
 
         decisionInProgress = true
 
-        joinDecider
-          .decide(info)
+        // `decide` may throw synchronously before it ever builds a Future (it resolves the self
+        // contact point eagerly), and `recover` would not see that. Letting it escape would leave
+        // `decisionInProgress` stuck true and stop this coordinator deciding ever again.
+        val decision =
+          try joinDecider.decide(info)
+          catch {
+            case NonFatal(e) => Future.failed(e)
+          }
+
+        decision
           .recover {
             case e =>
               log.error(e, "Join decision failed: {}", e)

management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/SelfContactPointResolutionSpec.scala

@@ -0,0 +1,145 @@
+/*
+ * 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.management.cluster.bootstrap
+
+import java.util.concurrent.TimeoutException
+
+import com.typesafe.config.ConfigFactory
+import org.apache.pekko
+import pekko.actor.ActorSystem
+import pekko.event.NoLogging
+import pekko.testkit.TestKit
+import org.scalatest.BeforeAndAfterAll
+import org.scalatest.matchers.should.Matchers
+import org.scalatest.wordspec.AnyWordSpec
+
+import scala.concurrent.duration._
+
+object SelfContactPointResolutionSpec {
+
+  /** Waits a fraction of the real timeout, so that the blocking path can be exercised in a test. */
+  class ShortTimeoutJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings)
+      extends LowestAddressJoinDecider(system, settings) {
+    override protected def selfContactPointTimeout: FiniteDuration = 500.millis
+  }
+
+  /** Reads back the production timeout, which is otherwise protected. */
+  class TimeoutReadingJoinDecider(system: ActorSystem, settings: ClusterBootstrapSettings)
+      extends LowestAddressJoinDecider(system, settings) {
+    def configuredTimeout: FiniteDuration = selfContactPointTimeout
+  }
+
+}
+
+class SelfContactPointResolutionSpec extends AnyWordSpec with Matchers with BeforeAndAfterAll {
+
+  import SelfContactPointResolutionSpec._
+
+  private val config = ConfigFactory.parseString("""
+      pekko {
+        loglevel = INFO
+        remote.artery.canonical.port = 0
+        management.http {
+          hostname = "10.0.0.2"
+          base-path = "test"
+        }
+      }
+      """).withFallback(ConfigFactory.load())
+
+  // a system per test, because setSelfContactPoint completes a promise that cannot be reset
+  private var systems: List[ActorSystem] = Nil
+
+  private def newDecider(name: String): (ActorSystem, ShortTimeoutJoinDecider) = {
+    val system = ActorSystem(name, config)
+    systems = system :: systems
+    val settings = ClusterBootstrapSettings(system.settings.config, NoLogging)
+    (system, new ShortTimeoutJoinDecider(system, settings))
+  }
+
+  private def elapsed(body: => Unit): FiniteDuration = {
+    val started = System.nanoTime()
+    body
+    (System.nanoTime() - started).nanos
+  }
+
+  "SelfAwareJoinDecider.selfContactPoint" should {
+
+    "resolve the contact point once it has been set" in {
+      val (system, decider) = newDecider("self-contact-point-resolves")
+      ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test")
+
+      decider.selfContactPoint should ===(("10.0.0.2", 8558))
+    }
+
+    "keep returning the resolved contact point on later calls" in {
+      val (system, decider) = newDecider("self-contact-point-caches")
+      ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test")
+
+      val first = decider.selfContactPoint
+      val second = decider.selfContactPoint
+
+      first should ===(("10.0.0.2", 8558))
+      second should ===(first)
+    }
+
+    "time out instead of blocking forever when the contact point is never set" in {
+      val (_, decider) = newDecider("self-contact-point-times-out")
+
+      a[TimeoutException] should be thrownBy decider.selfContactPoint
+    }
+
+    "block for the timeout only once, then fail fast" in {
+      val (_, decider) = newDecider("self-contact-point-fails-fast")
+
+      val firstCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint)
+      val secondCall = elapsed(a[TimeoutException] should be thrownBy decider.selfContactPoint)
+
+      firstCall should be >= 500.millis
+      // without this the bootstrap coordinator would park a dispatcher thread for the full
+      // timeout on every probe, for as long as the contact point stays unset
+      secondCall should be < 250.millis
+    }
+
+    "still pick up a contact point that is set after a timeout" in {
+      val (system, decider) = newDecider("self-contact-point-set-late")
+
+      a[TimeoutException] should be thrownBy decider.selfContactPoint
+
+      ClusterBootstrap(system).setSelfContactPoint("http://10.0.0.2:8558/test")
+
+      decider.selfContactPoint should ===(("10.0.0.2", 8558))
+    }
+
+  }
+
+  "ClusterBootstrap.SelfContactPointTimeout" should {
+
+    "be outlasted by the decider's own wait, so that the promise fails first" in {
+      val (system, _) = newDecider("self-contact-point-timeout-ordering")
+      val settings = ClusterBootstrapSettings(system.settings.config, NoLogging)
+      val decider = new TimeoutReadingJoinDecider(system, settings)
+
+      decider.configuredTimeout should be > ClusterBootstrap.SelfContactPointTimeout
+    }
+
+  }
+
+  override def afterAll(): Unit =
+    systems.foreach(TestKit.shutdownActorSystem(_, 5.seconds))
+
+}

management-cluster-bootstrap/src/test/scala/org/apache/pekko/management/cluster/bootstrap/internal/BootstrapCoordinatorSpec.scala

@@ -13,15 +13,21 @@
 
 package org.apache.pekko.management.cluster.bootstrap.internal
 
-import java.util.concurrent.atomic.AtomicReference
+import java.util.concurrent.TimeoutException
+import java.util.concurrent.atomic.{ AtomicInteger, AtomicReference }
 
 import org.apache.pekko
 import pekko.actor.{ ActorRef, ActorSystem, Props }
 import pekko.discovery.ServiceDiscovery.{ Resolved, ResolvedTarget }
 import pekko.discovery.{ Lookup, MockDiscovery }
 import pekko.http.scaladsl.model.Uri
 import pekko.management.cluster.bootstrap.internal.BootstrapCoordinator.Protocol.InitiateBootstrapping
-import pekko.management.cluster.bootstrap.{ ClusterBootstrapSettings, LowestAddressJoinDecider }
+import pekko.management.cluster.bootstrap.{
+  ClusterBootstrapSettings,
+  JoinDecision,
+  LowestAddressJoinDecider,
+  SeedNodesInformation
+}
 import com.typesafe.config.ConfigFactory
 import org.scalatest.BeforeAndAfterAll
 import org.scalatest.concurrent.Eventually
@@ -177,6 +183,48 @@ class BootstrapCoordinatorSpec extends AnyWordSpec with Matchers with BeforeAndA
     }
   }
 
+  "The bootstrap coordinator, when a JoinDecider throws synchronously" should {
+
+    // SelfAwareJoinDecider.selfContactPoint resolves eagerly and can throw before `decide` ever
+    // builds a Future, so the failure never reaches `recover`. If that escapes, `decisionInProgress`
+    // is left true and the coordinator stops deciding for the rest of its life.
+    "keep making decisions instead of wedging" in {
+      val decideServiceName = "bootstrap-coordinator-throwing-decider"
+      val decideSettings = ClusterBootstrapSettings(
+        ConfigFactory.parseString(s"""
+          |pekko.management.cluster.bootstrap {
+          | contact-point-discovery.service-name = $decideServiceName
+          | contact-point-discovery.required-contact-point-nr = 1
+          | contact-point-discovery.contact-with-all-contact-points = false
+          |}
+        """.stripMargin).withFallback(system.settings.config),
+        system.log)
+
+      MockDiscovery.set(
+        Lookup(decideServiceName, portName = None, protocol = Some("tcp")),
+        () => Future.successful(Resolved(decideServiceName, List(ResolvedTarget("host1", Some(7626), None)))))
+
+      val decideCalls = new AtomicInteger(0)
+      val throwingDecider = new LowestAddressJoinDecider(system, decideSettings) {
+        override def decide(info: SeedNodesInformation): Future[JoinDecision] = {
+          decideCalls.incrementAndGet()
+          throw new TimeoutException("'Bootstrap.selfContactPoint' was never set")
+        }
+      }
+
+      val coordinator = system.actorOf(Props(new BootstrapCoordinator(discovery, throwingDecider, decideSettings) {
+        override def ensureProbing(selfContactPointScheme: String, contactPoint: ResolvedTarget): Option[ActorRef] =
+          None
+      }))
+      coordinator ! InitiateBootstrapping(selfUri)
+
+      // more than one call proves decisionInProgress was reset after the throw
+      eventually {
+        decideCalls.get should be >= 2
+      }
+    }
+  }
+
   override def afterAll(): Unit = {
     Await.result(system.terminate(), 10.seconds)
     super.afterAll()