[ETCM-313] and [ETCM-316]: Header skeleton using new branch resolver

src/ets/scala/io/iohk/ethereum/ets/blockchain/EthashTestBlockHeaderValidator.scala

@@ -15,7 +15,7 @@ class EthashTestBlockHeaderValidator(blockchainConfig: BlockchainConfig) extends
 
   protected def difficulty: DifficultyCalculator = DifficultyCalculator(blockchainConfig)
 
-  def validateEvenMore(blockHeader: BlockHeader, parentHeader: BlockHeader): Either[BlockHeaderError, BlockHeaderValid] =
+  override def validateEvenMore(blockHeader: BlockHeader): Either[BlockHeaderError, BlockHeaderValid] =
     Right(BlockHeaderValid)
 
 }

src/it/scala/io/iohk/ethereum/sync/FastSyncItSpec.scala

@@ -59,49 +59,53 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfterAll {
 
   it should "sync blockchain with state nodes when peer do not response with full responses" in
     customTestCaseResourceM(
-      FakePeer.start3FakePeersRes(
+      FakePeer.start4FakePeersRes(
         fakePeerCustomConfig2 = FakePeerCustomConfig(HostConfig()),
         fakePeerCustomConfig3 = FakePeerCustomConfig(HostConfig())
       )
-    ) { case (peer1, peer2, peer3) =>
+    ) { case (peer1, peer2, peer3, peer4) =>
       for {
         _ <- peer2.importBlocksUntil(1000)(updateStateAtBlock(500))
         _ <- peer3.importBlocksUntil(1000)(updateStateAtBlock(500))
-        _ <- peer1.connectToPeers(Set(peer2.node, peer3.node))
+        _ <- peer4.importBlocksUntil(1000)(updateStateAtBlock(500))
+
+        _ <- peer1.connectToPeers(Set(peer2.node, peer3.node, peer4.node))
         _ <- peer1.startFastSync().delayExecution(50.milliseconds)
         _ <- peer1.waitForFastSyncFinish()
       } yield {
         val trie = peer1.getBestBlockTrie()
         val synchronizingPeerHaveAllData = peer1.containsExpectedDataUpToAccountAtBlock(1000, 500)
-        // due to the fact that function generating state is deterministic both peer2 and peer3 ends up with exactly same
+        // due to the fact that function generating state is deterministic both peer3 and peer4 ends up with exactly same
         // state, so peer1 can get whole trie from both of them.
-        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
         assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
+        assert(peer1.bl.getBestBlockNumber() == peer4.bl.getBestBlockNumber() - peer4.testSyncConfig.pivotBlockOffset)
         assert(trie.isDefined)
         assert(synchronizingPeerHaveAllData)
       }
     }
 
   it should "sync blockchain with state nodes when one of the peers send empty state responses" in
     customTestCaseResourceM(
-      FakePeer.start3FakePeersRes(
+      FakePeer.start4FakePeersRes(
         fakePeerCustomConfig2 = FakePeerCustomConfig(HostConfig()),
         fakePeerCustomConfig3 = FakePeerCustomConfig(HostConfig().copy(maxMptComponentsPerMessage = 0))
       )
-    ) { case (peer1, peer2, peer3) =>
+    ) { case (peer1, peer2, peer3, peer4) =>
       for {
         _ <- peer2.importBlocksUntil(1000)(updateStateAtBlock(500))
         _ <- peer3.importBlocksUntil(1000)(updateStateAtBlock(500))
-        _ <- peer1.connectToPeers(Set(peer2.node, peer3.node))
+        _ <- peer4.importBlocksUntil(1000)(updateStateAtBlock(500))
+
+        _ <- peer1.connectToPeers(Set(peer2.node, peer3.node, peer4.node))
         _ <- peer1.startFastSync().delayExecution(50.milliseconds)
         _ <- peer1.waitForFastSyncFinish()
       } yield {
         val trie = peer1.getBestBlockTrie()
         val synchronizingPeerHaveAllData = peer1.containsExpectedDataUpToAccountAtBlock(1000, 500)
-        // due to the fact that function generating state is deterministic both peer2 and peer3 ends up with exactly same
+        // due to the fact that function generating state is deterministic both peer3 and peer4 ends up with exactly same
         // state, so peer1 can get whole trie from both of them.
-        assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
         assert(peer1.bl.getBestBlockNumber() == peer3.bl.getBestBlockNumber() - peer3.testSyncConfig.pivotBlockOffset)
+        assert(peer1.bl.getBestBlockNumber() == peer4.bl.getBestBlockNumber() - peer4.testSyncConfig.pivotBlockOffset)
         assert(trie.isDefined)
         assert(synchronizingPeerHaveAllData)
       }

src/it/scala/io/iohk/ethereum/sync/util/CommonFakePeer.scala

@@ -222,7 +222,8 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
     startRetryInterval = 50.milliseconds,
     nodesPerRequest = 200,
     maxTargetDifference = 1,
-    syncRetryInterval = 50.milliseconds
+    syncRetryInterval = 50.milliseconds,
+    blacklistDuration = 100.seconds
   )
 
   lazy val broadcaster = new BlockBroadcast(etcPeerManager)

src/main/resources/application.conf

@@ -369,7 +369,9 @@ mantis {
     max-concurrent-requests = 50
 
     # Requested number of block headers when syncing from other peers
-    block-headers-per-request = 200
+    # Will cause an error if it's higher than max-blocks-headers-per-message of the peer we're requesting from,
+    # so this number should not be set very high.
+    block-headers-per-request = 100
 
     # Requested number of block bodies when syncing from other peers
     block-bodies-per-request = 128
@@ -469,6 +471,10 @@ mantis {
     # (peer.bestKnownBlock - pivot-block-offset) - node.curentPivotBlock > max-pivot-age
     # it fast sync pivot block has become stale and it needs update
     max-pivot-block-age = 96
+
+    # Maximum number of retries performed by fast sync when the master peer sends invalid block headers.
+    # On reaching this limit, it will perform branch resolving.
+    fast-sync-max-batch-retries = 5
   }
 
   pruning {

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

@@ -93,7 +93,7 @@ object Blacklist {
 
     case object WrongBlockHeaders extends BlacklistReason {
       val reasonType: BlacklistReasonType = WrongBlockHeadersType
-      val description: String = "Wrong blockheaders response (empty or not chain forming)"
+      val description: String = "Wrong blockheaders response: Peer didn't respond with requested block headers."
     }
     case object BlockHeaderValidationFailed extends BlacklistReason {
       val reasonType: BlacklistReasonType = BlockHeaderValidationFailedType

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

@@ -17,6 +17,8 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
 
   private implicit val ec: ExecutionContext = context.dispatcher
 
+  protected val bigIntReverseOrdering: Ordering[BigInt] = Ordering[BigInt].reverse
+
   def etcPeerManager: ActorRef
   def peerEventBus: ActorRef
   def blacklist: Blacklist
@@ -44,6 +46,9 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
 
   def getPeerById(peerId: PeerId): Option[Peer] = handshakedPeers.get(peerId).map(_.peer)
 
+  def getPeerWithHighestBlock: Option[PeerWithInfo] =
+    peersToDownloadFrom.values.toList.sortBy(_.peerInfo.maxBlockNumber)(bigIntReverseOrdering).headOption
+
   def blacklistIfHandshaked(peerId: PeerId, duration: FiniteDuration, reason: BlacklistReason): Unit =
     handshakedPeers.get(peerId).foreach(_ => blacklist.add(peerId, duration, reason))
 

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

@@ -24,15 +24,15 @@ class PeerRequestHandler[RequestMsg <: Message, ResponseMsg <: Message: ClassTag
 
   import PeerRequestHandler._
 
-  val initiator: ActorRef = context.parent
+  private val initiator: ActorRef = context.parent
 
-  val timeout: Cancellable = scheduler.scheduleOnce(responseTimeout, self, Timeout)
+  private val timeout: Cancellable = scheduler.scheduleOnce(responseTimeout, self, Timeout)
 
-  val startTime: Long = System.currentTimeMillis()
+  private val startTime: Long = System.currentTimeMillis()
 
   private def subscribeMessageClassifier = MessageClassifier(Set(responseMsgCode), PeerSelector.WithId(peer.id))
 
-  def timeTakenSoFar(): Long = System.currentTimeMillis() - startTime
+  private def timeTakenSoFar(): Long = System.currentTimeMillis() - startTime
 
   override def preStart(): Unit = {
     etcPeerManager ! EtcPeerManagerActor.SendMessage(toSerializable(requestMsg), peer.id)
@@ -79,8 +79,8 @@ object PeerRequestHandler {
   )(implicit scheduler: Scheduler, toSerializable: RequestMsg => MessageSerializable): Props =
     Props(new PeerRequestHandler(peer, responseTimeout, etcPeerManager, peerEventBus, requestMsg, responseMsgCode))
 
-  case class RequestFailed(peer: Peer, reason: String)
-  case class ResponseReceived[T](peer: Peer, response: T, timeTaken: Long)
+  final case class RequestFailed(peer: Peer, reason: String)
+  final case class ResponseReceived[T](peer: Peer, response: T, timeTaken: Long)
 
   private case object Timeout
 }