go back and redownload if found missing header

Author: adamsmo

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

@@ -2,14 +2,15 @@ package io.iohk.ethereum.blockchain.sync
 
 import akka.actor._
 import akka.util.ByteString
-import io.iohk.ethereum.domain.BlockHeader
+import io.iohk.ethereum.domain.{Block, BlockHeader}
 import io.iohk.ethereum.network.EtcPeerManagerActor.PeerInfo
 import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
 import io.iohk.ethereum.network.PeerEventBusActor.SubscriptionClassifier.MessageClassifier
 import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe, Unsubscribe}
 import io.iohk.ethereum.network.p2p.messages.PV62.{BlockBody, BlockHeaders, GetBlockHeaders}
 import io.iohk.ethereum.network.{EtcPeerManagerActor, Peer}
 import io.iohk.ethereum.utils.Config.Sync._
+import io.iohk.ethereum.validators.BlockValidator
 
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.concurrent.duration.FiniteDuration
@@ -197,11 +198,15 @@ trait FastSync {
         downloadedNodesCount += num
 
       case BlockBodiesReceived(peer, requestedHashes, blockBodies) =>
-        if (validateBlocks(requestedHashes, blockBodies)) {
-          insertBlocks(requestedHashes, blockBodies)
-        } else {
-          blacklist(peer.id, blacklistDuration, s"responded with block bodies not matching block headers, blacklisting for $blacklistDuration")
-          self ! FastSync.EnqueueBlockBodies(requestedHashes)
+        validateBlocks(requestedHashes, blockBodies) match{
+          case Valid =>
+            insertBlocks(requestedHashes, blockBodies)
+          case Invalid =>
+            blacklist(peer.id, blacklistDuration, s"responded with block bodies not matching block headers, blacklisting for $blacklistDuration")
+            self ! FastSync.EnqueueBlockBodies(requestedHashes)
+          case DbError =>
+            log.debug("missing block header for known hash")
+            self ! ProcessSyncing
         }
 
       case BlockHeadersReceived(_, headers) =>
@@ -276,14 +281,24 @@ trait FastSync {
       requestedReceipts = requestedReceipts - handler
     }
 
-    private def validateBlocks(requestedHashes: Seq[ByteString], blockBodies: Seq[BlockBody]): Boolean = (requestedHashes zip blockBodies)
-      .map { case (hash, body) => (blockchain.getBlockHeaderByHash(hash), body) }
-      .forall {
-        case (Some(header), body) =>
-          val result = validators.blockValidator.validateHeaderAndBody(header, body)
-          result.isRight
-        case _ => false
-      }
+    private def validateBlocks(requestedHashes: Seq[ByteString], blockBodies: Seq[BlockBody]): ValidationResult = {
+      var result: ValidationResult = Valid
+      (requestedHashes zip blockBodies)
+        .map { case (hash, body) => (blockchain.getBlockHeaderByHash(hash), body) }
+        .forall {
+          case (Some(header), body) =>
+            val validationResult: Either[BlockValidator.BlockError, Block] = validators.blockValidator.validateHeaderAndBody(header, body)
+            result = validationResult.fold(_ => Invalid, _ => Valid)
+            validationResult.isRight
+          case _ =>
+            blockBodiesQueue = Seq.empty
+            receiptsQueue = Seq.empty
+            bestBlockHeaderNumber = bestBlockHeaderNumber - 2 * blockHeadersPerRequest
+            result = DbError
+            false
+        }
+      result
+    }
 
     private def insertBlocks(requestedHashes: Seq[ByteString], blockBodies: Seq[BlockBody]): Unit = {
       (requestedHashes zip blockBodies).foreach { case (hash, body) =>
@@ -473,6 +488,11 @@ object FastSync {
   private case object PersistSyncState
   case class MarkPeerBlockchainOnly(peer: Peer)
 
+  private sealed trait ValidationResult
+  private case object Valid extends ValidationResult
+  private case object Invalid extends ValidationResult
+  private case object DbError extends ValidationResult
+
   case class SyncState(
     targetBlock: BlockHeader,
     mptNodesQueue: Seq[HashType] = Nil,