[ETCM-313] and [ETCM-316]: Header skeleton using new branch resolver (#892)

* create actor FastSyncBranchResolver

* ETCM-313 Download skeleton and then batch headers in parallel

* Implement HeaderSkeleton class

* Validate if skeleton header matches downloaded batch

* Improve validations

* Handle wrong skeleton from master peer

* Fix incorrect example

* Validate PoW of skeleton headers

* Fix bugs found with tests

* Add call to the branch resolver

* Add missing config entries

* Fix unit tests

* Apply scalafmt

* Cleanup tests

* Fastsync: stick with the same master peer while requesting skeleton headers.

* [ETCM-313] Integrate branch resolver actor with fast sync

* [ETCM-313] Fix integration tests, format error messages

* create actor FastSyncBranchResolver

* create actor FastSyncBranchResolver

* [ETCM-316] Fast-sync branch resolver (#887)

* init new actor, FastSyncBranchResolver

* added searching mode in fast sync branch resolver

* fix style

* add schedule when don't get peers

* Added ut in class FastSyncBranchResolver

* change messages

* fix case object...

* add new unit test

* add new unit test

* change name getFirstCommonBlock

* change name to batch

* init new actor, FastSyncBranchResolver

* added searching mode in fast sync branch resolver

* fix style

* add schedule when don't get peers

* Added ut in class FastSyncBranchResolver

* change messages

* fix case object...

* add new unit test

* add new unit test

* create actor FastSyncBranchResolver

* change name getFirstCommonBlock

* change name to batch

* Reformat triggered by sbt pp

* Cleanup and simplify

* Handle error cases

* Fix tests

* [ETCM-316] Add more tests and fix binary search logic

* [ETCM-316] Finish tests for branch resolving

* [ETCM-316] Cleanup

* [ETCM-316] Small test improvements

* [ETCM-316] Log binary search state

* [ETCM-316] Move some logging to improve readability

* [ETCM-316] Remove unneeded errors and reformat

* [ETCM-316] Handle branch resolution failure

* [ETCM-316] Address PR comments

* [ETCM-316] Remove unnecessary string interpolation

Co-authored-by: Petra Bierleutgeb <[email protected]>

* [ETCM-313] Reworked header skeleton (still needs refactoring)

* [ETCM-313] Remove empty method

* Fix SyncController tests, add more logging

* Remove logging that broke integration tests (timeout)

* [ETCM-313] More refactorings

* [ETCM-313] Fix integration tests

* [ETCM-313] Re-request header skeleton in case of errors

* [ETCM-313] Remove skeleton handler name

* [ETCM-313] Small fixes and better logs

* [ETCM-313] Update the default number of requested block headers to not be higher than the default max number of headers returned

* [ETCM-313] Adapt branchresolver recent blocks request.

Co-authored-by: Maximiliano Biandratti <[email protected]>
Co-authored-by: Robin Raju <[email protected]>
Co-authored-by: Petra Bierleutgeb <[email protected]>
Co-authored-by: biandratti <[email protected]>
Co-authored-by: Petra Bierleutgeb <[email protected]>

Author: 4 people

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
 }