Turn blacklist reasons into proper types, small improvements based on PR comments

ten15bit · ten15bit · commit d8b5fe9da9db · 2021-02-14T22:02:45.000+01:00
diff --git a/src/it/scala/io/iohk/ethereum/sync/util/FastSyncItSpecUtils.scala b/src/it/scala/io/iohk/ethereum/sync/util/FastSyncItSpecUtils.scala
@@ -17,13 +17,17 @@ import monix.eval.Task
 import scala.annotation.tailrec
 import scala.concurrent.duration._
 import scala.util.Try
+import io.iohk.ethereum.blockchain.sync.CacheBasedBlacklist
 object FastSyncItSpecUtils {
 
   class FakePeer(peerName: String, fakePeerCustomConfig: FakePeerCustomConfig)
       extends CommonFakePeer(peerName, fakePeerCustomConfig) {
 
     lazy val validators = new MockValidatorsAlwaysSucceed
 
+    val maxSize = 1000
+    val blacklist = CacheBasedBlacklist.empty(maxSize)
+
     lazy val fastSync = system.actorOf(
       FastSync.props(
         storagesInstance.storages.fastSyncStateStorage,
@@ -32,6 +36,7 @@ object FastSyncItSpecUtils {
         validators,
         peerEventBus,
         etcPeerManager,
+        blacklist,
         testSyncConfig,
         system.scheduler
       )
diff --git a/src/main/scala/io/iohk/ethereum/blockchain/sync/Blacklist.scala b/src/main/scala/io/iohk/ethereum/blockchain/sync/Blacklist.scala
@@ -10,57 +10,151 @@ import scala.jdk.CollectionConverters._
 import scala.jdk.OptionConverters._
 import scala.jdk.DurationConverters._
 
-import Blacklist.BlacklistId
+import Blacklist._
+import io.iohk.ethereum.network.PeerId
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason.BlacklistReasonType.WrongBlockHeadersType
+import io.iohk.ethereum.consensus.validators.std.StdBlockValidator.BlockError
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason.BlacklistReasonType
 
 trait Blacklist {
   def isBlacklisted(id: BlacklistId): Boolean
-  def add(id: BlacklistId, duration: FiniteDuration, reason: String): Unit
+  def add(id: BlacklistId, duration: FiniteDuration, reason: BlacklistReason): Unit
   def remove(id: BlacklistId): Unit
   def keys: Set[BlacklistId]
 }
 
 object Blacklist {
+  import BlacklistReason._
+  import BlacklistReasonType._
+
   trait BlacklistId {
     def value: String
   }
+
+  sealed trait BlacklistReason {
+    def reasonType: BlacklistReasonType
+    def description: String
+  }
+  object BlacklistReason {
+    trait BlacklistReasonType {
+      def code: Int
+      def name: String
+    }
+    object BlacklistReasonType {
+      case object WrongBlockHeadersType extends BlacklistReasonType {
+        val code: Int = 1
+        val name: String = "WrongBlockHeadersType"
+      }
+      case object BlockHeaderValidationFailedType extends BlacklistReasonType {
+        val code: Int = 2
+        val name: String = "BlockHeaderValidationFailed"
+      }
+      case object ErrorInBlockHeadersType extends BlacklistReasonType {
+        val code: Int = 3
+        val name: String = "ErrorInBlockHeaders"
+      }
+      case object EmptyBlockBodiesType extends BlacklistReasonType {
+        val code: Int = 4
+        val name: String = "EmptyBlockBodies"
+      }
+      case object BlockBodiesNotMatchingHeadersType extends BlacklistReasonType {
+        val code: Int = 5
+        val name: String = "BlockBodiesNotMatchingHeaders"
+      }
+      case object EmptyReceiptsType extends BlacklistReasonType {
+        val code: Int = 6
+        val name: String = "EmptyReceipts"
+      }
+      case object InvalidReceiptsType extends BlacklistReasonType {
+        val code: Int = 7
+        val name: String = "InvalidReceipts"
+      }
+      case object RequestFailedType extends BlacklistReasonType {
+        val code: Int = 8
+        val name: String = "RequestFailed"
+      }
+      case object PeerActorTerminatedType extends BlacklistReasonType {
+        val code: Int = 9
+        val name: String = "PeerActorTerminated"
+      }
+    }
+
+    case object WrongBlockHeaders extends BlacklistReason {
+      val reasonType: BlacklistReasonType = WrongBlockHeadersType
+      val description: String = "Wrong blockheaders response (empty or not chain forming)"
+    }
+    case object BlockHeaderValidationFailed extends BlacklistReason {
+      val reasonType: BlacklistReasonType = BlockHeaderValidationFailedType
+      val description: String = "Block header validation failed"
+    }
+    case object ErrorInBlockHeaders extends BlacklistReason {
+      val reasonType: BlacklistReasonType = ErrorInBlockHeadersType
+      val description: String = "Error in block headers response"
+    }
+    final case class EmptyBlockBodies(knownHashes: Seq[String]) extends BlacklistReason {
+      val reasonType: BlacklistReasonType = EmptyBlockBodiesType
+      val description: String = s"Got empty block bodies response for known hashes: $knownHashes"
+    }
+    case object BlockBodiesNotMatchingHeaders extends BlacklistReason {
+      val reasonType: BlacklistReasonType = BlockBodiesNotMatchingHeadersType
+      val description = "Block bodies not matching block headers"
+    }
+    final case class EmptyReceipts(knownHashes: Seq[String]) extends BlacklistReason {
+      val reasonType: BlacklistReasonType = EmptyReceiptsType
+      val description: String = s"Got empty receipts for known hashes: $knownHashes"
+    }
+    final case class InvalidReceipts(knownHashes: Seq[String], error: BlockError) extends BlacklistReason {
+      val reasonType: BlacklistReasonType = InvalidReceiptsType
+      val description: String = s"Got invalid receipts for known hashes: $knownHashes due to: $error"
+    }
+    final case class RequestFailed(error: String) extends BlacklistReason {
+      val reasonType: BlacklistReasonType = RequestFailedType
+      val description: String = s"Request failed with error: $error"
+    }
+    case object PeerActorTerminated extends BlacklistReason {
+      val reasonType: BlacklistReasonType = PeerActorTerminatedType
+      val description: String = "Peer actor terminated"
+    }
+  }
 }
 
-final case class CacheBasedBlacklist(cache: Cache[BlacklistId, String]) extends Blacklist with Logger {
+final case class CacheBasedBlacklist(cache: Cache[BlacklistId, BlacklistReasonType]) extends Blacklist with Logger {
+
+  import CacheBasedBlacklist._
 
   override def isBlacklisted(id: BlacklistId): Boolean = cache.getIfPresent(id).isDefined
 
-  override def add(id: BlacklistId, duration: FiniteDuration, reason: String): Unit =
-    cache
-      .policy()
-      .expireVariably()
-      .toScala
-      .fold {
-        log.warn(
-          s"Unexpected error while adding peer [${id.value}] to blacklist using custom expiration time. Falling back to default expiration."
-        )
-        cache.put(id, reason)
-      } { varExpirationPolicy =>
-        varExpirationPolicy.put(id, reason, duration.toJava)
-      }
+  override def add(id: BlacklistId, duration: FiniteDuration, reason: BlacklistReason): Unit = {
+    log.debug("Blacklisting peer [{}] for {}. Reason: {}", id, duration, reason)
+    cache.policy().expireVariably().toScala match {
+      case Some(varExpiration) => varExpiration.put(id, reason.reasonType, duration.toJava)
+      case None =>
+        log.warn(customExpirationError(id))
+        cache.put(id, reason.reasonType)
+    }
+  }
   override def remove(id: BlacklistId): Unit = cache.invalidate(id)
 
   override def keys: Set[BlacklistId] = cache.underlying.asMap().keySet().asScala.toSet
 }
 
 object CacheBasedBlacklist {
 
+  def customExpirationError(id: BlacklistId): String =
+    s"Unexpected error while adding peer [${id.value}] to blacklist using custom expiration time. Falling back to default expiration."
+
   def empty(maxSize: Int): CacheBasedBlacklist = {
     val cache =
       Scaffeine()
-        .expireAfter[BlacklistId, String](
+        .expireAfter[BlacklistId, BlacklistReasonType](
           create = (_, _) => 60.minutes,
           update = (_, _, _) => 60.minutes,
           read = (_, _, _) => 60.minutes
         ) // required to enable VarExpiration policy (i.e. set custom expiration time per element)
         .maximumSize(
           maxSize
         ) // uses Window TinyLfu eviction policy, see https://github.com/ben-manes/caffeine/wiki/Efficiency
-        .build[BlacklistId, String]()
+        .build[BlacklistId, BlacklistReasonType]()
     CacheBasedBlacklist(cache)
   }
 
diff --git a/src/main/scala/io/iohk/ethereum/blockchain/sync/PeerListSupportNg.scala b/src/main/scala/io/iohk/ethereum/blockchain/sync/PeerListSupportNg.scala
@@ -13,6 +13,7 @@ import scala.concurrent.duration._
 
 trait PeerListSupportNg { self: Actor with ActorLogging =>
   import PeerListSupportNg._
+  import Blacklist._
 
   private implicit val ec: ExecutionContext = context.dispatcher
 
@@ -33,17 +34,17 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
 
   def handlePeerListMessages: Receive = {
     case EtcPeerManagerActor.HandshakedPeers(peers) => updatePeers(peers)
-    case PeerDisconnected(peerId) => removePeer(peerId)
+    case PeerDisconnected(peerId) => removePeerById(peerId)
   }
 
   def peersToDownloadFrom: Map[PeerId, PeerWithInfo] =
     handshakedPeers.filterNot { case (peerId, _) =>
       blacklist.isBlacklisted(peerId)
     }
 
-  def peerById(peerId: PeerId): Option[Peer] = handshakedPeers.get(peerId).map(_.peer)
+  def getPeerById(peerId: PeerId): Option[Peer] = handshakedPeers.get(peerId).map(_.peer)
 
-  def blacklistIfHandshaked(peerId: PeerId, duration: FiniteDuration, reason: String): Unit =
+  def blacklistIfHandshaked(peerId: PeerId, duration: FiniteDuration, reason: BlacklistReason): Unit =
     handshakedPeers.get(peerId).foreach(_ => blacklist.add(peerId, duration, reason))
 
   private def updatePeers(peers: Map[Peer, PeerInfo]): Unit = {
@@ -56,7 +57,7 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
     handshakedPeers = updated
   }
 
-  private def removePeer(peerId: PeerId): Unit = {
+  private def removePeerById(peerId: PeerId): Unit = {
     if (handshakedPeers.keySet.contains(peerId)) {
       peerEventBus ! Unsubscribe(PeerDisconnectedClassifier(PeerSelector.WithId(peerId)))
       blacklist.remove(peerId)
diff --git a/src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSync.scala b/src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSync.scala
@@ -8,6 +8,8 @@ import io.iohk.ethereum.blockchain.sync.PeerListSupportNg.PeerWithInfo
 import io.iohk.ethereum.blockchain.sync.PeerRequestHandler.ResponseReceived
 import io.iohk.ethereum.blockchain.sync.SyncProtocol.Status.Progress
 import io.iohk.ethereum.blockchain.sync._
+import io.iohk.ethereum.blockchain.sync.Blacklist._
+import io.iohk.ethereum.blockchain.sync.Blacklist.BlacklistReason._
 import io.iohk.ethereum.blockchain.sync.fast.ReceiptsValidator.ReceiptsValidationResult
 import io.iohk.ethereum.blockchain.sync.fast.SyncBlocksValidator.BlockBodyValidationResult
 import io.iohk.ethereum.blockchain.sync.fast.SyncStateSchedulerActor.{
@@ -183,7 +185,7 @@ class FastSync(
           )
             handleBlockHeaders(peer, blockHeaders)
           else
-            blacklist.add(peer.id, blacklistDuration, "wrong blockheaders response (empty or not chain forming)")
+            blacklist.add(peer.id, blacklistDuration, WrongBlockHeaders)
         }
 
       case ResponseReceived(peer, BlockBodies(blockBodies), timeTaken) =>
@@ -205,10 +207,10 @@ class FastSync(
         handleReceipts(peer, requestedHashes, receipts)
 
       case PeerRequestHandler.RequestFailed(peer, reason) =>
-        handleRequestFailure(peer, sender(), reason)
+        handleRequestFailure(peer, sender(), RequestFailed(reason))
 
       case Terminated(ref) if assignedHandlers.contains(ref) =>
-        handleRequestFailure(assignedHandlers(ref), ref, "Unexpected error")
+        handleRequestFailure(assignedHandlers(ref), ref, PeerActorTerminated)
     }
 
     def askForPivotBlockUpdate(updateReason: PivotBlockUpdateReason): Unit = {
@@ -423,7 +425,7 @@ class FastSync(
     }
 
     private def handleRewind(header: BlockHeader, peer: Peer, N: Int, duration: FiniteDuration): Unit = {
-      blacklist.add(peer.id, duration, "block header validation failed")
+      blacklist.add(peer.id, duration, BlockHeaderValidationFailed)
       if (header.number <= syncState.safeDownloadTarget) {
         discardLastBlocks(header.number, N)
         syncState = syncState.updateDiscardedBlocks(header, N)
@@ -461,27 +463,22 @@ class FastSync(
             )
         }
       } else {
-        blacklist.add(peer.id, blacklistDuration, "error in block headers response")
+        blacklist.add(peer.id, blacklistDuration, ErrorInBlockHeaders)
         processSyncing()
       }
     }
 
     private def handleBlockBodies(peer: Peer, requestedHashes: Seq[ByteString], blockBodies: Seq[BlockBody]): Unit = {
       if (blockBodies.isEmpty) {
-        val reason =
-          s"got empty block bodies response for known hashes: ${requestedHashes.map(ByteStringUtils.hash2string)}"
-        blacklist.add(peer.id, blacklistDuration, reason)
+        val knownHashes = requestedHashes.map(ByteStringUtils.hash2string)
+        blacklist.add(peer.id, blacklistDuration, EmptyBlockBodies(knownHashes))
         syncState = syncState.enqueueBlockBodies(requestedHashes)
       } else {
         validateBlocks(requestedHashes, blockBodies) match {
           case BlockBodyValidationResult.Valid =>
             insertBlocks(requestedHashes, blockBodies)
           case BlockBodyValidationResult.Invalid =>
-            blacklist.add(
-              peer.id,
-              blacklistDuration,
-              s"responded with block bodies not matching block headers, blacklisting for $blacklistDuration"
-            )
+            blacklist.add(peer.id, blacklistDuration, BlockBodiesNotMatchingHeaders)
             syncState = syncState.enqueueBlockBodies(requestedHashes)
           case BlockBodyValidationResult.DbError =>
             redownloadBlockchain()
@@ -493,8 +490,8 @@ class FastSync(
 
     private def handleReceipts(peer: Peer, requestedHashes: Seq[ByteString], receipts: Seq[Seq[Receipt]]): Unit = {
       if (receipts.isEmpty) {
-        val reason = s"got empty receipts for known hashes: ${requestedHashes.map(ByteStringUtils.hash2string)}"
-        blacklist.add(peer.id, blacklistDuration, reason)
+        val knownHashes = requestedHashes.map(ByteStringUtils.hash2string)
+        blacklist.add(peer.id, blacklistDuration, EmptyReceipts(knownHashes))
         syncState = syncState.enqueueReceipts(requestedHashes)
       } else {
         validateReceipts(requestedHashes, receipts) match {
@@ -515,10 +512,8 @@ class FastSync(
             }
 
           case ReceiptsValidationResult.Invalid(error) =>
-            val reason =
-              s"got invalid receipts for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))}" +
-                s" due to: $error"
-            blacklist.add(peer.id, blacklistDuration, reason)
+            val knownHashes = requestedHashes.map(h => Hex.toHexString(h.toArray[Byte]))
+            blacklist.add(peer.id, blacklistDuration, InvalidReceipts(knownHashes, error))
             syncState = syncState.enqueueReceipts(requestedHashes)
 
           case ReceiptsValidationResult.DbError =>
@@ -529,7 +524,7 @@ class FastSync(
       processSyncing()
     }
 
-    private def handleRequestFailure(peer: Peer, handler: ActorRef, reason: String): Unit = {
+    private def handleRequestFailure(peer: Peer, handler: ActorRef, reason: BlacklistReason): Unit = {
       removeRequestHandler(handler)
 
       syncState = syncState
diff --git a/src/test/scala/io/iohk/ethereum/blockchain/sync/CacheBasedBlacklistSpec.scala b/src/test/scala/io/iohk/ethereum/blockchain/sync/CacheBasedBlacklistSpec.scala
