Minor cleanup (#417)

* minor cleanup

Signed-off-by: [email protected] <[email protected]>

* spotless

Signed-off-by: [email protected] <[email protected]>

* make it work

Signed-off-by: [email protected] <[email protected]>

* make search interval and search timeout in discovery configurable

Signed-off-by: [email protected] <[email protected]>

* add exception handler

Signed-off-by: [email protected] <[email protected]>

* updates after comments

Signed-off-by: [email protected] <[email protected]>

* job and logs

Signed-off-by: [email protected] <[email protected]>

* try to fix flaky test

Signed-off-by: [email protected] <[email protected]>

* Removed unneeded syncStatusProviderProvider

* Removed unneeded syncStatusProviderProvider

* Merged with main one more time

* Cleaned unused argument

* Removed MaruDiscoveryPeer, replaced with Teku's DiscoveryPeer

* Fixed assertions

* Fixed the test

---------

Signed-off-by: [email protected] <[email protected]>
Co-authored-by: Roman <[email protected]>

Author: pinges

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

@@ -26,7 +26,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 net.consensys.linea.metrics.MetricsFacade
 import org.apache.logging.log4j.LogManager
 import org.assertj.core.api.Assertions.assertThat
@@ -185,7 +184,6 @@ class MaruPeerScoringTest {
           forkIdHashManager: ForkPeeringManager,
           isBlockImportEnabledProvider: () -> Boolean,
           p2pState: P2PState,
-          syncStatusProviderProvider: () -> SyncStatusProvider,
           ->
           MisbehavingP2PNetwork(
             privateKeyBytes = privateKeyBytes,
@@ -199,7 +197,6 @@ class MaruPeerScoringTest {
             forkIdHashManager = forkIdHashManager,
             isBlockImportEnabledProvider = isBlockImportEnabledProvider,
             p2pState = p2pState,
-            syncStatusProviderProvider = syncStatusProviderProvider,
             blockRetrievalStrategy = blockRetrievalStrategy,
           ).p2pNetwork
         },

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 =

p2p/src/main/kotlin/maru/p2p/MaruPeerManager.kt

@@ -15,10 +15,8 @@ import java.util.concurrent.TimeUnit
 import java.util.concurrent.atomic.AtomicBoolean
 import maru.config.P2PConfig
 import maru.p2p.discovery.MaruDiscoveryService
-import maru.syncing.SyncStatusProvider
 import org.apache.logging.log4j.LogManager
 import org.apache.logging.log4j.Logger
-import tech.pegasys.teku.infrastructure.async.SafeFuture
 import tech.pegasys.teku.networking.p2p.network.P2PNetwork
 import tech.pegasys.teku.networking.p2p.network.PeerHandler
 import tech.pegasys.teku.networking.p2p.peer.DisconnectReason
@@ -27,10 +25,9 @@ import tech.pegasys.teku.networking.p2p.peer.Peer
 
 class MaruPeerManager(
   private val maruPeerFactory: MaruPeerFactory,
-  p2pConfig: P2PConfig,
+  private val p2pConfig: P2PConfig,
   private val reputationManager: MaruReputationManager,
   private val isStaticPeer: (NodeId) -> Boolean,
-  private val syncStatusProviderProvider: () -> SyncStatusProvider,
 ) : PeerHandler,
   PeerLookup {
   private val log: Logger = LogManager.getLogger(this.javaClass)
@@ -43,8 +40,6 @@ class MaruPeerManager(
   private val peers: ConcurrentHashMap<NodeId, MaruPeer> = ConcurrentHashMap()
 
   private var scheduler: ScheduledExecutorService? = null
-  private lateinit var p2pNetwork: P2PNetwork<Peer>
-  private lateinit var syncStatusProvider: SyncStatusProvider
 
   val peerCount: Int
     get() = connectedPeers().size
@@ -61,8 +56,6 @@ class MaruPeerManager(
       return
     }
     this.discoveryService = discoveryService
-    this.p2pNetwork = p2pNetwork
-    this.syncStatusProvider = syncStatusProviderProvider()
     scheduler = Executors.newSingleThreadScheduledExecutor(Thread.ofPlatform().daemon().factory())
     scheduler!!.scheduleAtFixedRate({
       logConnectedPeers()
@@ -75,32 +68,26 @@ class MaruPeerManager(
           reputationManager = reputationManager,
           maxPeers = maxPeers,
           getPeerCount = { peerCount },
+          discoveryConfig = p2pConfig.discovery!!,
         )
       discoveryTask!!.start()
     }
   }
 
-  fun stop(): SafeFuture<Unit> {
+  fun stop() {
     if (!started.compareAndSet(true, false)) {
       log.warn("Trying to stop stopped MaruPeerManager")
-      return SafeFuture.completedFuture(Unit)
+      return
     }
     discoveryTask?.stop()
     discoveryTask = null
     scheduler!!.shutdown()
     scheduler = null
-    return SafeFuture.completedFuture(Unit)
+    peers.values.forEach { it.disconnectCleanly(DisconnectReason.SHUTTING_DOWN) }
   }
 
   private fun logConnectedPeers() {
-    connectedPeers().keys.toList().also { peers ->
-      log.info(
-        "currently connected peers: peerCount={} peers={}",
-        peers.size,
-        peers,
-      )
-    }
-
+    log.info("Currently connected peers={}", connectedPeers().keys.toList())
     if (log.isDebugEnabled) {
       discoveryService?.getKnownPeers()?.forEach { peer ->
         log.debug("discovered peer={}", peer)
@@ -121,7 +108,7 @@ class MaruPeerManager(
       log.debug("Connected to peer={}, static=$isAStaticPeer", peer.id)
     } else { // not static and not allowed to connect -> disconnect
       peer.disconnectCleanly(DisconnectReason.RATE_LIMITING)
-      log.debug("Peer={} is not allowed to connect yet", peer.id)
+      log.trace("Peer={} is not allowed to connect yet", peer.id)
     }
   }
 

p2p/src/main/kotlin/maru/p2p/MaruRpcResponseCallback.kt

@@ -65,7 +65,7 @@ class MaruRpcResponseCallback<TResponse : Message<*, *>>(
           ),
         ).finishWarn(log)
     } catch (e: StreamClosedException) {
-      log.debug(
+      log.trace(
         "Unable to send error message ({}) to peer, rpc stream already closed: {}",
         error,
         rpcStream,

p2p/src/main/kotlin/maru/p2p/P2PNetworkImpl.kt

@@ -33,7 +33,6 @@ import maru.p2p.topics.QbftMessageSerDe
 import maru.p2p.topics.TopicHandlerWithInOrderDelivering
 import maru.serialization.SerDe
 import maru.serialization.rlp.MaruCompressorRLPSerDe
-import maru.syncing.SyncStatusProvider
 import net.consensys.linea.metrics.MetricsFacade
 import net.consensys.linea.metrics.Tag
 import org.apache.logging.log4j.LogManager
@@ -66,7 +65,6 @@ class P2PNetworkImpl(
   private val forkIdHashManager: ForkPeeringManager,
   isBlockImportEnabledProvider: () -> Boolean,
   private val p2PState: P2PState,
-  private val syncStatusProviderProvider: () -> SyncStatusProvider,
   // for testing:
   private val rpcMethodsFactory: (
     StatusManager,
@@ -141,7 +139,6 @@ class P2PNetworkImpl(
         p2pConfig = p2pConfig,
         reputationManager = reputationManager,
         isStaticPeer = this::isStaticPeer,
-        syncStatusProviderProvider = syncStatusProviderProvider,
       )
 
     return Libp2pNetworkFactory(LINEA_DOMAIN).build(
@@ -250,10 +247,9 @@ class P2PNetworkImpl(
 
   override fun stop(): SafeFuture<Unit> {
     log.info("Stopping={}", this::class.simpleName)
-    val pmStop = maruPeerManager.stop()
+    maruPeerManager.stop()
     discoveryService?.stop()
-    val p2pStop = p2pNetwork.stop()
-    return SafeFuture.allOf(p2pStop, pmStop).thenApply {}
+    return p2pNetwork.stop().thenApply {}
   }
 
   override fun close() {