download mpt first

Author: adamsmo

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

@@ -167,6 +167,8 @@ trait FastSync {
 
     syncStateStorageActor ! fastSyncStateStorage
 
+    private var blockChainOnlyPeers = Set.empty[Peer]
+
     private val syncStatePersistCancellable =
       scheduler.schedule(persistStateSnapshotInterval, persistStateSnapshotInterval) {
         syncStateStorageActor ! SyncState(
@@ -181,7 +183,7 @@ trait FastSync {
 
     private val heartBeat = scheduler.schedule(syncRetryInterval, syncRetryInterval * 2, self, ProcessSyncing)
 
-    def receive: Receive = handlePeerUpdates orElse {
+    def receive: Receive = handlePeerUpdates orElse handleFailingMptPeers orElse {
       case EnqueueNodes(hashes) =>
         hashes.foreach {
           case h: EvmCodeHash => nonMptNodesQueue = h +: nonMptNodesQueue
@@ -227,6 +229,11 @@ trait FastSync {
         printStatus()
     }
 
+    private def handleFailingMptPeers: Receive ={
+      case BlockChainOnlyDownload(peer) =>
+        blockChainOnlyPeers = blockChainOnlyPeers + peer
+    }
+
     private def printStatus() = {
       val totalNodesCount = downloadedNodesCount + mptNodesQueue.size + nonMptNodesQueue.size
       val formatPeer: (Peer) => String = peer => s"${peer.remoteAddress.getAddress.getHostAddress}:${peer.remoteAddress.getPort}"
@@ -316,22 +323,33 @@ trait FastSync {
           scheduler.scheduleOnce(syncRetryInterval, self, ProcessSyncing)
         }
       } else {
-        unassignedPeers
+        val peers = unassignedPeers
+        (peers -- blockChainOnlyPeers)
           .take(maxConcurrentRequests - assignedHandlers.size)
           .foreach(assignWork)
+        peers
+          .intersect(blockChainOnlyPeers)
+          .take(maxConcurrentRequests - assignedHandlers.size)
+          .foreach(assignBlockChainWork)
       }
     }
 
     def assignWork(peer: Peer): Unit = {
+      if (nonMptNodesQueue.nonEmpty || mptNodesQueue.nonEmpty) {
+        requestNodes(peer)
+      } else {
+        assignBlockChainWork(peer)
+      }
+    }
+
+    def assignBlockChainWork(peer: Peer): Unit = {
       if (receiptsQueue.nonEmpty) {
         requestReceipts(peer)
       } else if (blockBodiesQueue.nonEmpty) {
         requestBlockBodies(peer)
       } else if (context.child(blockHeadersHandlerName).isEmpty &&
         initialSyncState.targetBlock.number > bestBlockHeaderNumber) {
         requestBlockHeaders(peer)
-      } else if (nonMptNodesQueue.nonEmpty || mptNodesQueue.nonEmpty) {
-        requestNodes(peer)
       }
     }
 
@@ -423,6 +441,7 @@ object FastSync {
   private case object TargetBlockTimeout
 
   private case object ProcessSyncing
+  case class BlockChainOnlyDownload(peer: Peer)
 
   case class SyncState(
     targetBlock: BlockHeader,

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

@@ -24,8 +24,8 @@ class FastSyncNodesRequestHandler(
 
   override def handleResponseMsg(nodeData: NodeData): Unit = {
     if (nodeData.values.isEmpty) {
-      val reason = s"got empty mpt node response for known hashes: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}"
-      syncController ! BlacklistSupport.BlacklistPeer(peer.id, reason)
+      log.debug(s"got empty mpt node response for known hashes switching to blockchain only: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}")
+      syncController ! BlockChainOnlyDownload(peer)
     }
 
     val receivedHashes = nodeData.values.map(v => ByteString(kec256(v.toArray[Byte])))