ETCM-760: Move networking and regular sync to new blacklist

Author: AnastasiiaL

src/main/scala/io/iohk/ethereum/blockchain/sync/PeersClient.scala

@@ -1,6 +1,8 @@
 package io.iohk.ethereum.blockchain.sync
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props, Scheduler}
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
+import io.iohk.ethereum.blockchain.sync.PeerListSupportNg.PeerWithInfo
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.p2p.messages.Codes
 import io.iohk.ethereum.network.{Peer, PeerId}
@@ -15,12 +17,12 @@ import scala.reflect.ClassTag
 class PeersClient(
     val etcPeerManager: ActorRef,
     val peerEventBus: ActorRef,
+    val blacklist: Blacklist,
     val syncConfig: SyncConfig,
     implicit val scheduler: Scheduler
 ) extends Actor
     with ActorLogging
-    with PeerListSupport
-    with BlacklistSupport {
+    with PeerListSupportNg {
   import PeersClient._
 
   implicit val ec: ExecutionContext = context.dispatcher
@@ -36,9 +38,9 @@ class PeersClient(
   }
 
   def running(requesters: Requesters): Receive =
-    handleBlacklistMessages orElse handlePeerListMessages orElse {
+    handlePeerListMessages orElse {
       case PrintStatus => printStatus(requesters: Requesters)
-      case BlacklistPeer(peerId, reason) => peerById(peerId).foreach(blacklistIfHandshaked(_, reason))
+      case BlacklistPeer(peerId, reason) => blacklistIfHandshaked(peerId, syncConfig.blacklistDuration, reason)
       case Request(message, peerSelector, toSerializable) =>
         val requester = sender()
         selectPeer(peerSelector) match {
@@ -56,7 +58,7 @@ class PeersClient(
         handleResponse(requesters, Response(peer, message.asInstanceOf[Message]))
       case PeerRequestHandler.RequestFailed(peer, reason) =>
         log.warning(s"Request to peer ${peer.remoteAddress} failed - reason: $reason")
-        handleResponse(requesters, RequestFailed(peer, reason))
+        handleResponse(requesters, RequestFailed(peer, BlacklistReason.RegularSyncRequestFailed(reason)))
     }
 
   private def makeRequest[RequestMsg <: Message, ResponseMsg <: Message](
@@ -108,12 +110,12 @@ class PeersClient(
       peersToDownloadFrom.size
     )
 
-    lazy val handshakedPeersStatus = handshakedPeers.map { case (peer, info) =>
+    lazy val handshakedPeersStatus = handshakedPeers.map { case (peerId, peerWithInfo) =>
       val peerNetworkStatus = PeerNetworkStatus(
-        peer,
-        isBlacklisted = isBlacklisted(peer.id)
+        peerWithInfo.peer,
+        isBlacklisted = blacklist.isBlacklisted(peerId)
       )
-      (peerNetworkStatus, info)
+      (peerNetworkStatus, peerWithInfo.peerInfo)
     }
 
     log.debug(s"Handshaked peers status (number of peers: ${handshakedPeersStatus.size}): $handshakedPeersStatus")
@@ -122,13 +124,19 @@ class PeersClient(
 
 object PeersClient {
 
-  def props(etcPeerManager: ActorRef, peerEventBus: ActorRef, syncConfig: SyncConfig, scheduler: Scheduler): Props =
-    Props(new PeersClient(etcPeerManager, peerEventBus, syncConfig, scheduler))
+  def props(
+      etcPeerManager: ActorRef,
+      peerEventBus: ActorRef,
+      blacklist: Blacklist,
+      syncConfig: SyncConfig,
+      scheduler: Scheduler
+  ): Props =
+    Props(new PeersClient(etcPeerManager, peerEventBus, blacklist, syncConfig, scheduler))
 
   type Requesters = Map[ActorRef, ActorRef]
 
   sealed trait PeersClientMessage
-  case class BlacklistPeer(peerId: PeerId, reason: String) extends PeersClientMessage
+  case class BlacklistPeer(peerId: PeerId, reason: BlacklistReason) extends PeersClientMessage
   case class Request[RequestMsg <: Message](
       message: RequestMsg,
       peerSelector: PeerSelector,
@@ -155,16 +163,16 @@ object PeersClient {
 
   sealed trait ResponseMessage
   case object NoSuitablePeer extends ResponseMessage
-  case class RequestFailed(peer: Peer, reason: String) extends ResponseMessage
+  case class RequestFailed(peer: Peer, reason: BlacklistReason) extends ResponseMessage
   case class Response[T <: Message](peer: Peer, message: T) extends ResponseMessage
 
   sealed trait PeerSelector
   case object BestPeer extends PeerSelector
 
-  def bestPeer(peersToDownloadFrom: Map[Peer, PeerInfo]): Option[Peer] = {
-    val peersToUse = peersToDownloadFrom
-      .collect { case (ref, PeerInfo(_, chainWeight, true, _, _)) =>
-        (ref, chainWeight)
+  def bestPeer(peersToDownloadFrom: Map[PeerId, PeerWithInfo]): Option[Peer] = {
+    val peersToUse = peersToDownloadFrom.values
+      .collect { case PeerWithInfo(peer, PeerInfo(_, chainWeight, true, _, _)) =>
+        (peer, chainWeight)
       }
 
     if (peersToUse.nonEmpty) {

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSyncBranchResolverActor.scala

@@ -192,7 +192,7 @@ class FastSyncBranchResolverActor(
     blacklistIfHandshaked(
       peer.id,
       syncConfig.blacklistDuration,
-      BlacklistReason.RequestFailed(reason)
+      BlacklistReason.FastSyncRequestFailed(reason)
     )
     restart()
   }

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockBroadcasterActor.scala

@@ -2,22 +2,22 @@ package io.iohk.ethereum.blockchain.sync.regular
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props, Scheduler}
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcast.BlockToBroadcast
-import io.iohk.ethereum.blockchain.sync.{BlacklistSupport, PeerListSupport}
+import io.iohk.ethereum.blockchain.sync.{Blacklist, PeerListSupportNg}
 import io.iohk.ethereum.utils.Config.SyncConfig
 
 class BlockBroadcasterActor(
     broadcast: BlockBroadcast,
     val peerEventBus: ActorRef,
     val etcPeerManager: ActorRef,
+    val blacklist: Blacklist,
     val syncConfig: SyncConfig,
     val scheduler: Scheduler
 ) extends Actor
     with ActorLogging
-    with PeerListSupport
-    with BlacklistSupport {
+    with PeerListSupportNg {
   import BlockBroadcasterActor._
 
-  override def receive: Receive = handlePeerListMessages orElse handleBlacklistMessages orElse handleBroadcastMessages
+  override def receive: Receive = handlePeerListMessages orElse handleBroadcastMessages
 
   private def handleBroadcastMessages: Receive = {
     case BroadcastBlock(newBlock) => broadcast.broadcastBlock(newBlock, handshakedPeers)
@@ -33,6 +33,7 @@ object BlockBroadcasterActor {
       broadcast: BlockBroadcast,
       peerEventBus: ActorRef,
       etcPeerManager: ActorRef,
+      blacklist: Blacklist,
       syncConfig: SyncConfig,
       scheduler: Scheduler
   ): Props =
@@ -41,6 +42,7 @@ object BlockBroadcasterActor {
         broadcast = broadcast,
         peerEventBus = peerEventBus,
         etcPeerManager = etcPeerManager,
+        blacklist = blacklist,
         syncConfig = syncConfig,
         scheduler = scheduler
       )

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala

@@ -6,6 +6,7 @@ import akka.actor.{ActorRef => ClassicActorRef}
 import akka.util.{ByteString, Timeout}
 import cats.data.NonEmptyList
 import cats.instances.option._
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.blockchain.sync.PeersClient._
 import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState.{
@@ -126,7 +127,7 @@ class BlockFetcher(
       case InvalidateBlocksFrom(blockNr, reason, withBlacklist) =>
         val (blockProvider, newState) = state.invalidateBlocksFrom(blockNr, withBlacklist)
         log.debug("Invalidate blocks from {}", blockNr)
-        blockProvider.foreach(peersClient ! BlacklistPeer(_, reason))
+        blockProvider.foreach(peersClient ! BlacklistPeer(_, BlacklistReason.BlockImportError(reason)))
         fetchBlocks(newState)
 
       case ReceivedHeaders(headers) if state.isFetchingHeaders =>
@@ -326,7 +327,7 @@ object BlockFetcher {
     def apply(from: BigInt, reason: String, toBlacklist: Option[BigInt]): InvalidateBlocksFrom =
       new InvalidateBlocksFrom(from, reason, toBlacklist)
   }
-  final case class BlockImportFailed(blockNr: BigInt, reason: String) extends FetchCommand
+  final case class BlockImportFailed(blockNr: BigInt, reason: BlacklistReason) extends FetchCommand
   final case class InternalLastBlockImport(blockNr: BigInt) extends FetchCommand
   final case object RetryBodiesRequest extends FetchCommand
   final case object RetryHeadersRequest extends FetchCommand

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherState.scala

@@ -4,6 +4,7 @@ import akka.actor.ActorRef
 import akka.util.ByteString
 import cats.data.NonEmptyList
 import cats.implicits._
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState._
 import io.iohk.ethereum.consensus.validators.BlockValidator
 import io.iohk.ethereum.domain.{Block, BlockBody, BlockHeader, HeadersSeq}
@@ -122,10 +123,10 @@ case class BlockFetcherState(
     * even more, we could receive an empty chain and that will be considered valid. Here we just
     * validate that the received bodies corresponds to an ordered subset of the requested headers.
     */
-  def validateBodies(receivedBodies: Seq[BlockBody]): Either[String, Seq[Block]] =
+  def validateBodies(receivedBodies: Seq[BlockBody]): Either[BlacklistReason, Seq[Block]] =
     bodiesAreOrderedSubsetOfRequested(waitingHeaders.toList, receivedBodies)
       .toRight(
-        "Received unrequested bodies"
+        BlacklistReason.UnrequestedBodies
       )
 
   // Checks that the received block bodies are an ordered subset of the ones requested

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockImporter.scala

@@ -4,6 +4,7 @@ import akka.actor.Actor.Receive
 import akka.actor.{Actor, ActorLogging, ActorRef, NotInfluenceReceiveTimeout, Props, ReceiveTimeout}
 import cats.data.NonEmptyList
 import cats.implicits._
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlocks
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.ProgressProtocol
@@ -20,7 +21,6 @@ import io.iohk.ethereum.utils.Config.SyncConfig
 import io.iohk.ethereum.utils.FunctorOps._
 import monix.eval.Task
 import monix.execution.Scheduler
-import org.bouncycastle.util.encoders.Hex
 
 import scala.concurrent.duration._
 
@@ -263,7 +263,7 @@ class BlockImporter(
               Task.raiseError(missingNodeException)
             case BlockImportFailed(error) =>
               if (informFetcherOnFail) {
-                fetcher ! BlockFetcher.BlockImportFailed(block.number, error)
+                fetcher ! BlockFetcher.BlockImportFailed(block.number, BlacklistReason.BlockImportError(error))
               }
           }
           .map(_ => Running)

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/RegularSync.scala

@@ -3,6 +3,7 @@ package io.iohk.ethereum.blockchain.sync.regular
 import akka.actor.{Actor, ActorLogging, ActorRef, AllForOneStrategy, Cancellable, Props, Scheduler, SupervisorStrategy}
 import akka.actor.typed.{ActorRef => TypedActorRef}
 import io.iohk.ethereum.blockchain.sync.SyncProtocol
+import io.iohk.ethereum.blockchain.sync.{Blacklist, SyncProtocol}
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status.Progress
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.{NewCheckpoint, ProgressProtocol, ProgressState}
@@ -21,6 +22,7 @@ class RegularSync(
     ledger: Ledger,
     blockchain: Blockchain,
     blockValidator: BlockValidator,
+    blacklist: Blacklist,
     syncConfig: SyncConfig,
     ommersPool: ActorRef,
     pendingTransactionsManager: ActorRef,
@@ -38,7 +40,7 @@ class RegularSync(
 
   val broadcaster: ActorRef = context.actorOf(
     BlockBroadcasterActor
-      .props(new BlockBroadcast(etcPeerManager), peerEventBus, etcPeerManager, syncConfig, scheduler),
+      .props(new BlockBroadcast(etcPeerManager), peerEventBus, etcPeerManager, blacklist, syncConfig, scheduler),
     "block-broadcaster"
   )
   val importer: ActorRef =
@@ -126,6 +128,7 @@ object RegularSync {
       ledger: Ledger,
       blockchain: Blockchain,
       blockValidator: BlockValidator,
+      blacklist: Blacklist,
       syncConfig: SyncConfig,
       ommersPool: ActorRef,
       pendingTransactionsManager: ActorRef,
@@ -139,6 +142,7 @@ object RegularSync {
         ledger,
         blockchain,
         blockValidator,
+        blacklist,
         syncConfig,
         ommersPool,
         pendingTransactionsManager,

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/StateNodeFetcher.scala

@@ -12,6 +12,7 @@ import io.iohk.ethereum.network.p2p.Message
 import io.iohk.ethereum.network.p2p.messages.PV63.{GetNodeData, NodeData}
 import io.iohk.ethereum.utils.Config.SyncConfig
 import cats.syntax.either._
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason
 import monix.execution.Scheduler
 
 import scala.util.{Failure, Success}
@@ -46,15 +47,13 @@ class StateNodeFetcher(
         requester
           .collect(stateNodeRequester => {
             val validatedNode = values
-              .asRight[String]
-              .ensure(s"Empty response from peer $peer, blacklisting")(_.nonEmpty)
-              .ensure("Fetched node state hash doesn't match requested one, blacklisting peer")(nodes =>
-                stateNodeRequester.hash == kec256(nodes.head)
-              )
+              .asRight[BlacklistReason]
+              .ensure(BlacklistReason.EmptyStateNodeResponse)(_.nonEmpty)
+              .ensure(BlacklistReason.WrongStateNodeResponse)(nodes => stateNodeRequester.hash == kec256(nodes.head))
 
             validatedNode match {
               case Left(err) =>
-                log.debug(err)
+                log.debug(err.description)
                 peersClient ! BlacklistPeer(peer.id, err)
                 context.self ! StateNodeFetcher.FetchStateNode(stateNodeRequester.hash, stateNodeRequester.replyTo)
                 Behaviors.same[StateNodeFetcherCommand]