Minor cleanup

app/src/integrationTest/kotlin/maru/app/MaruPeerScoringTest.kt

@@ -14,6 +14,7 @@ import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.minutes
 import kotlin.time.Duration.Companion.seconds
 import kotlin.time.toJavaDuration
+import kotlinx.coroutines.CoroutineExceptionHandler
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.Job
@@ -30,7 +31,6 @@ import maru.p2p.messages.BlockRetrievalStrategy
 import maru.p2p.messages.DefaultBlockRetrievalStrategy
 import maru.p2p.messages.StatusManager
 import maru.serialization.SerDe
-import maru.syncing.SyncStatusProvider
 import maru.test.util.NetworkUtil.findFreePort
 import net.consensys.linea.metrics.MetricsFacade
 import org.apache.logging.log4j.LogManager
@@ -45,6 +45,7 @@ import org.hyperledger.besu.tests.acceptance.dsl.node.cluster.ClusterConfigurati
 import org.hyperledger.besu.tests.acceptance.dsl.transaction.net.NetTransactions
 import org.junit.jupiter.api.AfterEach
 import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.fail
 import testutils.FourEmptyResponsesStrategy
 import testutils.MisbehavingP2PNetwork
 import testutils.PeeringNodeNetworkStack
@@ -53,7 +54,6 @@ import testutils.besu.BesuFactory
 import testutils.besu.BesuTransactionsHelper
 import testutils.besu.ethGetBlockByNumber
 import testutils.maru.MaruFactory
-import testutils.maru.awaitTillMaruHasPeers
 
 class MaruPeerScoringTest {
   private lateinit var cluster: Cluster
@@ -65,9 +65,11 @@ class MaruPeerScoringTest {
   private lateinit var fakeLineaContract: FakeLineaRollupSmartContractClient
   private lateinit var validatorEthApiClient: EthApiClient
   private lateinit var followerEthApiClient: EthApiClient
+  private var job: Job? = null
 
   @AfterEach
   fun tearDown() {
+    job?.cancel()
     followerStack.maruApp.stop()
     validatorStack.maruApp.stop()
     followerStack.maruApp.close()
@@ -77,22 +79,17 @@ class MaruPeerScoringTest {
 
   @Test
   fun `node gets in sync with default block retrieval strategy`() {
-    val maruNodeSetup =
-      setUpNodes(blockRetrievalStrategy = DefaultBlockRetrievalStrategy())
+    setUpNodes(blockRetrievalStrategy = DefaultBlockRetrievalStrategy())
 
-    try {
-      await
-        .atMost(20.seconds.toJavaDuration())
-        .pollInterval(200.milliseconds.toJavaDuration())
-        .ignoreExceptions()
-        .untilAsserted {
-          assertThat(
-            followerEthApiClient.getBlockByNumberWithoutTransactionsData(BlockParameter.Tag.LATEST).get().number,
-          ).isGreaterThanOrEqualTo(15UL)
-        }
-    } finally {
-      maruNodeSetup.job.cancel()
-    }
+    await
+      .atMost(20.seconds.toJavaDuration())
+      .pollInterval(200.milliseconds.toJavaDuration())
+      .ignoreExceptions()
+      .untilAsserted {
+        assertThat(
+          followerEthApiClient.getBlockByNumberWithoutTransactionsData(BlockParameter.Tag.LATEST).get().number,
+        ).isGreaterThanOrEqualTo(15UL)
+      }
   }
 
   @Test
@@ -104,31 +101,27 @@ class MaruPeerScoringTest {
         followerCooldownPeriod = 10.minutes,
       )
 
-    try {
-      // In setUpNodes we have made sure that the validator and the follower have 1 peer
-      // Now wait until it is disconnected because of empty responses
-      await
-        .atMost(2.seconds.toJavaDuration())
-        .pollInterval(250.milliseconds.toJavaDuration())
-        .ignoreExceptions()
-        .untilAsserted {
-          assertThat(
-            maruNodeSetup.followerMaruApp.p2pNetwork.peerCount == 0,
-          )
-        }
-      // reconnects after ban period and finishes syncing
-      await
-        .atMost(20.seconds.toJavaDuration())
-        .pollInterval(200.milliseconds.toJavaDuration())
-        .ignoreExceptions()
-        .untilAsserted {
-          assertThat(
-            followerEthApiClient.getBlockByNumberWithoutTransactionsData(BlockParameter.Tag.LATEST).get().number,
-          ).isGreaterThanOrEqualTo(18UL)
-        }
-    } finally {
-      maruNodeSetup.job.cancel()
-    }
+    // In setUpNodes we have made sure that the validator and the follower have 1 peer
+    // Now wait until it is disconnected because of empty responses
+    await
+      .atMost(2.seconds.toJavaDuration())
+      .pollInterval(250.milliseconds.toJavaDuration())
+      .ignoreExceptions()
+      .untilAsserted {
+        assertThat(
+          maruNodeSetup.followerMaruApp.p2pNetwork.peerCount == 0,
+        )
+      }
+    // reconnects after ban period and finishes syncing
+    await
+      .atMost(20.seconds.toJavaDuration())
+      .pollInterval(200.milliseconds.toJavaDuration())
+      .ignoreExceptions()
+      .untilAsserted {
+        assertThat(
+          followerEthApiClient.getBlockByNumberWithoutTransactionsData(BlockParameter.Tag.LATEST).get().number,
+        ).isGreaterThanOrEqualTo(18UL)
+      }
   }
 
   @Test
@@ -141,22 +134,17 @@ class MaruPeerScoringTest {
         blockRetrievalStrategy = TimeOutResponsesStrategy(delay = delay),
         validatorCooldownPeriod = 20.seconds,
       )
-    try {
-      sleep((delay - 1.seconds).inWholeMilliseconds)
-      assertThat(maruNodeSetup.followerMaruApp.p2pNetwork.peerCount == 1)
+    sleep((timeout - 1.seconds).inWholeMilliseconds)
+    assertThat(maruNodeSetup.followerMaruApp.p2pNetwork.peerCount).isEqualTo(1)
 
-      await.untilAsserted {
-        assertThat(maruNodeSetup.followerMaruApp.p2pNetwork.peerCount == 0)
-      }
-    } finally {
-      maruNodeSetup.job.cancel()
+    await.untilAsserted {
+      assertThat(maruNodeSetup.followerMaruApp.p2pNetwork.peerCount).isEqualTo(0)
     }
   }
 
   data class MaruNodeSetup(
     val validatorMaruApp: MaruApp,
     val followerMaruApp: MaruApp,
-    val job: Job,
   )
 
   fun setUpNodes(
@@ -206,7 +194,6 @@ class MaruPeerScoringTest {
           forkIdHashManager: ForkPeeringManager,
           isBlockImportEnabledProvider: () -> Boolean,
           p2pState: P2PState,
-          syncStatusProviderProvider: () -> SyncStatusProvider,
           ->
           MisbehavingP2PNetwork(
             privateKeyBytes = privateKeyBytes,
@@ -220,7 +207,6 @@ class MaruPeerScoringTest {
             forkIdHashManager = forkIdHashManager,
             isBlockImportEnabledProvider = isBlockImportEnabledProvider,
             p2pState = p2pState,
-            syncStatusProviderProvider = syncStatusProviderProvider,
             blockRetrievalStrategy = blockRetrievalStrategy,
           ).p2pNetwork
         },
@@ -272,8 +258,13 @@ class MaruPeerScoringTest {
       )
     followerStack.setMaruApp(followerMaruApp)
 
-    val job =
-      CoroutineScope(Dispatchers.Default).launch {
+    val handler =
+      CoroutineExceptionHandler { _, exception ->
+        fail("Coroutine failed with exception: $exception")
+      }
+
+    job =
+      CoroutineScope(Dispatchers.Default).launch(handler) {
         while (true) {
           transactionsHelper.run {
             validatorStack.besuNode.sendTransactionAndAssertExecution(
@@ -298,16 +289,14 @@ class MaruPeerScoringTest {
 
       followerStack.maruApp.start()
 
-      validatorStack.maruApp.awaitTillMaruHasPeers(1u)
-      followerStack.maruApp.awaitTillMaruHasPeers(1u)
-
       followerEthApiClient =
         createEthApiClient(
           rpcUrl = followerStack.besuNode.jsonRpcBaseUrl().get(),
           log = LogManager.getLogger("clients.l2.test.follower"),
           requestRetryConfig = null,
           vertx = null,
         )
+
       // wait for Besu to be fully started and synced,
       // to avoid CI flakiness due to low resources sometimes
       await
@@ -319,9 +308,11 @@ class MaruPeerScoringTest {
             followerEthApiClient.getBlockByNumberWithoutTransactionsData(BlockParameter.Tag.LATEST).get().number,
           ).isGreaterThanOrEqualTo(0UL)
         }
+
+      return MaruNodeSetup(validatorMaruApp = validatorMaruApp, followerMaruApp = followerMaruApp)
     } catch (e: Exception) {
-      job.cancel()
+      job?.cancel()
+      throw e
     }
-    return MaruNodeSetup(validatorMaruApp = validatorMaruApp, followerMaruApp = followerMaruApp, job = job)
   }
 }

app/src/integrationTest/kotlin/testutils/MisbehavingP2PNetwork.kt

@@ -20,7 +20,6 @@ import maru.p2p.messages.BeaconBlocksByRangeRequest
 import maru.p2p.messages.BlockRetrievalStrategy
 import maru.p2p.messages.StatusManager
 import maru.serialization.SerDe
-import maru.syncing.SyncStatusProvider
 import net.consensys.linea.metrics.MetricsFacade
 import org.hyperledger.besu.plugin.services.MetricsSystem as BesuMetricsSystem
 
@@ -36,7 +35,6 @@ class MisbehavingP2PNetwork(
   forkIdHashManager: ForkPeeringManager,
   isBlockImportEnabledProvider: () -> Boolean,
   p2pState: P2PState,
-  syncStatusProviderProvider: () -> SyncStatusProvider,
   blockRetrievalStrategy: BlockRetrievalStrategy,
 ) {
   val p2pNetwork: P2PNetworkImpl =
@@ -52,7 +50,6 @@ class MisbehavingP2PNetwork(
       forkIdHashManager = forkIdHashManager,
       isBlockImportEnabledProvider = isBlockImportEnabledProvider,
       p2PState = p2pState,
-      syncStatusProviderProvider = syncStatusProviderProvider,
       rpcMethodsFactory = { statusMessageFactory, lineaRpcProtocolIdGenerator, peerLookup, beaconChain ->
         RpcMethods(statusMessageFactory, lineaRpcProtocolIdGenerator, peerLookup, beaconChain, blockRetrievalStrategy)
       },

app/src/main/kotlin/maru/app/MaruAppFactory.kt

@@ -68,7 +68,6 @@ import maru.syncing.HighestHeadTargetSelector
 import maru.syncing.MostFrequentHeadTargetSelector
 import maru.syncing.PeerChainTracker
 import maru.syncing.SyncController
-import maru.syncing.SyncStatusProvider
 import maru.syncing.SyncTargetSelector
 import maru.syncing.beaconchain.pipeline.BeaconChainDownloadPipelineFactory
 import net.consensys.linea.metrics.MetricsFacade
@@ -103,7 +102,6 @@ interface MaruAppFactoryCreator {
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl = ::P2PNetworkImpl,
   ): LongRunningCloseable
 }
@@ -131,7 +129,6 @@ class MaruAppFactory : MaruAppFactoryCreator {
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl,
   ): MaruApp {
     log.info("configs={}", config)
@@ -219,7 +216,6 @@ class MaruAppFactory : MaruAppFactoryCreator {
           }
         },
         p2PState = kvDatabase,
-        syncStatusProviderProvider = { syncControllerImpl!! },
         clock = clock,
         p2pNetworkFactory = p2pNetworkFactory,
       )
@@ -386,7 +382,6 @@ class MaruAppFactory : MaruAppFactoryCreator {
       besuMetricsSystem: BesuMetricsSystem,
       clock: Clock,
       p2PState: P2PState,
-      syncStatusProviderProvider: () -> SyncStatusProvider,
       p2pNetworkFactory: (
         ByteArray,
         P2PConfig,
@@ -399,7 +394,6 @@ class MaruAppFactory : MaruAppFactoryCreator {
         ForkPeeringManager,
         () -> Boolean,
         P2PState,
-        () -> SyncStatusProvider,
       ) -> P2PNetworkImpl = ::P2PNetworkImpl,
     ): P2PNetwork {
       if (p2pConfig == null) {
@@ -433,7 +427,6 @@ class MaruAppFactory : MaruAppFactoryCreator {
         forkIdHashManager,
         isBlockImportEnabledProvider,
         p2PState,
-        syncStatusProviderProvider,
       )
     }
 

app/src/test/kotlin/maru/app/MaruAppCliTest.kt

@@ -28,7 +28,6 @@ import maru.p2p.messages.StatusManager
 import maru.serialization.SerDe
 import maru.services.LongRunningService
 import maru.services.NoOpLongRunningService
-import maru.syncing.SyncStatusProvider
 import net.consensys.linea.metrics.MetricsFacade
 import org.assertj.core.api.Assertions.assertThat
 import org.junit.jupiter.api.AfterAll
@@ -170,7 +169,6 @@ class MaruAppCliTest {
           ForkPeeringManager,
           () -> Boolean,
           P2PState,
-          () -> SyncStatusProvider,
         ) -> P2PNetworkImpl,
       ): LongRunningCloseable {
         capturedMaruConfig = config

config/src/main/kotlin/maru/config/Config.kt

@@ -65,6 +65,8 @@ data class P2PConfig(
     val port: UInt = 9000u,
     val bootnodes: List<String> = emptyList(),
     val refreshInterval: Duration,
+    val searchInterval: Duration = 1.seconds,
+    val searchTimeout: Duration = 30.seconds,
     val advertisedIp: String? = null,
   ) {
     init {

config/src/test/kotlin/maru/config/HopliteFriendlinessTest.kt

@@ -44,6 +44,8 @@ class HopliteFriendlinessTest {
     port = 3324
     bootnodes = ["enr:-Iu4QHk0YN5IRRnufqsWkbO6Tn0iGTx4H_hnyiIEdXDuhIe0KKrxmaECisyvO40mEmmqKLhz_tdIhx2yFBK8XFKhvxABgmlkgnY0gmlwhH8AAAGJc2VjcDI1NmsxoQOgBvD-dv0cX5szOeEsiAMtwxnP1q5CA5toYDrgUyOhV4N0Y3CCJBKDdWRwgiQT"]
     refresh-interval = "30 seconds"
+    search-timeout = "10 seconds"
+    search-interval = "2 seconds"
     advertised-ip = "13.12.11.10"
 
     [p2p.reputation]
@@ -147,6 +149,8 @@ class HopliteFriendlinessTest {
               "enr:-Iu4QHk0YN5IRRnufqsWkbO6Tn0iGTx4H_hnyiIEdXDuhIe0KKrxmaECisyvO40mEmmqKLhz_tdIhx2yFBK8XFKhvxABgmlkgnY0gmlwhH8AAAGJc2VjcDI1NmsxoQOgBvD-dv0cX5szOeEsiAMtwxnP1q5CA5toYDrgUyOhV4N0Y3CCJBKDdWRwgiQT",
             ),
           refreshInterval = 30.seconds,
+          searchTimeout = 10.seconds,
+          searchInterval = 2.seconds,
           advertisedIp = "13.12.11.10",
         ),
       reputation =

jvm-libs/test-utils/src/main/kotlin/testutils/maru/MaruFactory.kt

@@ -58,7 +58,6 @@ import maru.p2p.P2PNetworkImpl
 import maru.p2p.fork.ForkPeeringManager
 import maru.p2p.messages.StatusManager
 import maru.serialization.SerDe
-import maru.syncing.SyncStatusProvider
 import net.consensys.linea.metrics.MetricsFacade
 import org.hyperledger.besu.plugin.services.MetricsSystem as BesuMetricsSystem
 
@@ -347,7 +346,6 @@ class MaruFactory(
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl = ::P2PNetworkImpl,
     startApiServer: Boolean = false,
   ): MaruApp =
@@ -458,7 +456,6 @@ class MaruFactory(
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl = ::P2PNetworkImpl,
     apiPort: UInt = 0u,
     startApiServer: Boolean = false,
@@ -530,7 +527,6 @@ class MaruFactory(
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl = ::P2PNetworkImpl,
   ): MaruApp {
     val p2pConfig =
@@ -598,7 +594,6 @@ class MaruFactory(
       ForkPeeringManager,
       () -> Boolean,
       P2PState,
-      () -> SyncStatusProvider,
     ) -> P2PNetworkImpl = ::P2PNetworkImpl,
   ): MaruApp {
     val p2pConfig =