[FIX] fast sync fix (#652)

* Rewind N  blocks(fast-sync-block-validation-n) and blacklist that Peer,  in case ParentDifficultyNotFound  happens while fast sync is processing blocks 
* Change log level to warning in case of blacklist

Author: KonradStaniec

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

@@ -22,7 +22,7 @@ trait BlacklistSupport {
         removeOldestPeer()
       }
       undoBlacklist(blacklistId)
-      log.debug(s"Blacklisting peer ($blacklistId), $reason")
+      log.warning(s"Blacklisting peer ($blacklistId), $reason")
       val unblacklistCancellable = scheduler.scheduleOnce(duration, self, UnblacklistPeer(blacklistId))
       blacklistedPeers.put(blacklistId, unblacklistCancellable)
     } else {

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

@@ -296,7 +296,7 @@ class FastSync(
       blockchain.getTotalDifficultyByHash(header.parentHash).toRight(ParentDifficultyNotFound(header))
     }
 
-    private def handleBlockValidationError(header: BlockHeader, peer: Peer, N: Int): Unit = {
+    private def handleRewind(header: BlockHeader, peer: Peer, N: Int): Unit = {
       blacklist(peer.id, blacklistDuration, "block header validation failed")
       if (header.number <= syncState.safeDownloadTarget) {
         discardLastBlocks(header.number, N)
@@ -315,14 +315,18 @@ class FastSync(
       if (checkHeadersChain(headers)) {
         processHeaders(peer, headers) match {
           case ParentDifficultyNotFound(header) =>
-            log.debug("Parent difficulty not found for block {}, not processing rest of headers", header.number)
-            processSyncing()
+            // We could end in wrong fork and get blocked so we should rewind our state a little
+            // we blacklist peer just in case we got malicious peer which would send us bad blocks, forcing us to rollback
+            // to genesis
+            log.warning("Parent difficulty not found for block {}, not processing rest of headers", header.idTag)
+            handleRewind(header, peer, syncConfig.fastSyncBlockValidationN)
           case HeadersProcessingFinished =>
             processSyncing()
           case ImportedTargetBlock  =>
             updateTargetBlock(ImportedLastBlock)
           case ValidationFailed(header, peerToBlackList) =>
-            handleBlockValidationError(header, peerToBlackList, syncConfig.fastSyncBlockValidationN)
+            log.warning(s"validation fo header ${header.idTag} failed")
+            handleRewind(header, peerToBlackList, syncConfig.fastSyncBlockValidationN)
         }
       } else {
         blacklist(peer.id, blacklistDuration, "error in block headers response")
@@ -385,8 +389,8 @@ class FastSync(
     }
 
     private def handleNodeData(peer: Peer, requestedHashes: Seq[HashType], nodeData: NodeData) = {
-      if (nodeData.values.isEmpty) {
-        log.debug(s"got empty mpt node response for known hashes switching to blockchain only: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}")
+      if (nodeData.values.isEmpty && requestedHashes.nonEmpty) {
+        log.info(s"got empty mpt node response for known hashes from peer ${peer.id}: ${requestedHashes.map(h => Hex.toHexString(h.v.toArray[Byte]))}")
         blacklist(peer.id,blacklistDuration, "empty mpt node response for known hashes")
       }
 
@@ -679,24 +683,28 @@ class FastSync(
     }
 
     def requestNodes(peer: Peer): Unit = {
-      val (nonMptNodesToGet, remainingNonMptNodes) = syncState.pendingNonMptNodes.splitAt(nodesPerRequest)
-      val (mptNodesToGet, remainingMptNodes) = syncState.pendingMptNodes.splitAt(nodesPerRequest - nonMptNodesToGet.size)
-      val nodesToGet = nonMptNodesToGet ++ mptNodesToGet
-
-      val handler = context.actorOf(
-        PeerRequestHandler.props[GetNodeData, NodeData](
-          peer, peerResponseTimeout, etcPeerManager, peerEventBus,
-          requestMsg = GetNodeData(nodesToGet.map(_.v)),
-          responseMsgCode = NodeData.code))
-
-      context watch handler
-      assignedHandlers += (handler -> peer)
-      peerRequestsTime += (peer -> Instant.now())
-      syncState = syncState.copy(
-        pendingNonMptNodes = remainingNonMptNodes,
-        pendingMptNodes = remainingMptNodes)
-      requestedMptNodes += handler -> mptNodesToGet
-      requestedNonMptNodes += handler -> nonMptNodesToGet
+      if (syncState.pendingNonMptNodes.nonEmpty || syncState.pendingMptNodes.nonEmpty) {
+        val (nonMptNodesToGet, remainingNonMptNodes) = syncState.pendingNonMptNodes.splitAt(nodesPerRequest)
+        val (mptNodesToGet, remainingMptNodes) = syncState.pendingMptNodes.splitAt(nodesPerRequest - nonMptNodesToGet.size)
+        val nodesToGet = nonMptNodesToGet ++ mptNodesToGet
+        log.info(s"Request ${nodesToGet.size} nodes from peer ${peer.id}")
+        val handler = context.actorOf(
+          PeerRequestHandler.props[GetNodeData, NodeData](
+            peer, peerResponseTimeout, etcPeerManager, peerEventBus,
+            requestMsg = GetNodeData(nodesToGet.map(_.v)),
+            responseMsgCode = NodeData.code))
+
+        context watch handler
+        assignedHandlers += (handler -> peer)
+        peerRequestsTime += (peer -> Instant.now())
+        syncState = syncState.copy(
+          pendingNonMptNodes = remainingNonMptNodes,
+          pendingMptNodes = remainingMptNodes)
+        requestedMptNodes += handler -> mptNodesToGet
+        requestedNonMptNodes += handler -> nonMptNodesToGet
+      } else {
+        log.debug("There is node work to assign for peer")
+      }
     }
 
     def unassignedPeers: Set[Peer] = peersToDownloadFrom.keySet diff assignedHandlers.values.toSet

src/test/scala/io/iohk/ethereum/blockchain/sync/SyncControllerSpec.scala

@@ -173,6 +173,35 @@ class SyncControllerSpec extends FlatSpec with Matchers with BeforeAndAfter with
     syncState.receiptsQueue.isEmpty shouldBe true
   }
 
+  it should "rewind fast-sync state if received header have no known parent" in new TestSetup() {
+
+    val bestBlockNumber = defaultExpectedTargetBlock - 1
+
+    startWithState(defaultState.copy(bestBlockHeaderNumber = bestBlockNumber))
+
+    Thread.sleep(1.seconds.toMillis)
+
+    syncController ! SyncController.Start
+
+    updateHandshakedPeers(HandshakedPeers(singlePeer))
+
+    sendBlockHeaders(
+      defaultTargetBlockHeader.number,
+      Seq(defaultTargetBlockHeader.copy(number = defaultExpectedTargetBlock, parentHash = ByteString(0,1))),
+      peer1,
+      defaultExpectedTargetBlock - bestBlockNumber)
+
+    persistState()
+
+    val syncState = storagesInstance.storages.fastSyncStateStorage.getSyncState().get
+
+    syncState.bestBlockHeaderNumber shouldBe (bestBlockNumber - syncConfig.fastSyncBlockValidationN)
+    syncState.nextBlockToFullyValidate shouldBe (bestBlockNumber - syncConfig.fastSyncBlockValidationN + 1)
+    syncState.blockBodiesQueue.isEmpty shouldBe true
+    syncState.receiptsQueue.isEmpty shouldBe true
+  }
+
+
   it should "not change best block after receiving faraway block" in new TestSetup(_validators = new Mocks.MockValidatorsAlwaysSucceed {
     override val blockHeaderValidator: BlockHeaderValidator = { (blockHeader, getBlockHeaderByHash) => Left(HeaderParentNotFoundError) }
   }) {