[ETCM-283] Check that blocks correspond to the requested headers that we want to process

Author: mirkoAlic

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcher.scala

@@ -140,24 +140,25 @@ class BlockFetcher(
 
   private def handleBodiesMessages(state: BlockFetcherState): Receive = {
     case Response(peer, BlockBodies(bodies)) if state.isFetchingBodies =>
-      val newState =
-        if (state.fetchingBodiesState == AwaitingBodiesToBeIgnored) {
-          log.debug("Received {} block bodies that will be ignored", bodies.size)
-          state.withBodiesFetchReceived
-        } else {
+      log.debug(s"Received ${bodies.size} block bodies")
+      if (state.fetchingBodiesState == AwaitingBodiesToBeIgnored) {
+        log.debug("Block bodies will be ignored due to an invalidation was requested for them")
+        fetchBlocks(state.withBodiesFetchReceived)
+      } else {
+        val newState =
           state.validateBodies(bodies) match {
             case Left(err) =>
               peersClient ! BlacklistPeer(peer.id, err)
               state.withBodiesFetchReceived
             case Right(newBlocks) =>
-              log.debug("Fetched {} block bodies", newBlocks.size)
-              state.withBodiesFetchReceived.appendNewBlocks(newBlocks, peer.id)
+              state.withBodiesFetchReceived.receiveBlocks(newBlocks, peer.id)
           }
-        }
-      fetchBlocks(newState)
+        val waitingHeadersDequeued = newState.waitingHeaders.size - state.waitingHeaders.size
+        log.debug(s"Processed ${waitingHeadersDequeued} new blocks from received block bodies")
+        fetchBlocks(newState)
+      }
     case RetryBodiesRequest if state.isFetchingBodies =>
       log.debug("Time-out occurred while waiting for bodies")
-
       val newState = state.withBodiesFetchReceived
       fetchBlocks(newState)
   }
@@ -214,7 +215,7 @@ class BlockFetcher(
         supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
         context become started(newState)
         // there are some blocks waiting for import but it seems that we reached top on fetch side so we can enqueue new block for import
-      } else if (newBlockNr == nextExpectedBlock && !state.isFetching && state.waitingHeaders.isEmpty) {
+      } else if (newBlockNr == nextExpectedBlock && !state.isFetching) {
         log.debug("Enqueue new block for import")
         val newState = state.appendNewBlock(block, peerId)
         supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)

src/main/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherState.scala

@@ -105,6 +105,11 @@ case class BlockFetcherState(
         }
       )
 
+  /**
+    * When bodies are requested, the response don't need to be a complete sub chain,
+    * even more, we could receive an empty chain and that will be considered valid. Here we just
+    * validate that the received bodies corresponds to an ordered subset of the requested headers.
+    */
   def validateBodies(receivedBodies: Seq[BlockBody]): Either[String, Seq[Block]] =
     bodiesAreOrderedSubsetOfRequested(waitingHeaders.toList, receivedBodies)
       .toRight(
@@ -129,22 +134,44 @@ case class BlockFetcherState(
           bodiesAreOrderedSubsetOfRequested(remainingHeaders, respondedBodies, matchedBlocks)
     }
 
-  def appendNewBlocks(blocks: Seq[Block], fromPeer: PeerId): BlockFetcherState = {
-    val receivedHeaders = blocks.map(_.header)
-    withPeerForBlocks(fromPeer, blocks.map(_.header.number))
-      .copy(
-        readyBlocks = readyBlocks.enqueue(blocks.toList),
-        waitingHeaders = waitingHeaders.diff(receivedHeaders)
-      )
+  // We could optimize this method by stopping as soon as a block is not appended.
+  def receiveBlocks(blocks: Seq[Block], fromPeer: PeerId): BlockFetcherState = {
+    blocks.foldLeft(this) { case (state, block) =>
+      state.receiveBlock(block, fromPeer)
+    }
   }
 
+  /**
+    * Currently to fill in headers we use a queue, so we if we try to process
+    * a block that has its header in the queue but is not the next in the line,
+    * we opt for not appending it.
+    */
+  def receiveBlock(block: Block, fromPeer: PeerId): BlockFetcherState =
+    waitingHeaders.dequeueOption
+      .map { case (waitingHeader, waitingHeadersTail) =>
+        if (waitingHeader.hash == block.hash)
+          unsafeAppendNewBlock(block, fromPeer).copy(
+            waitingHeaders = waitingHeadersTail
+          )
+        else
+          this
+      }
+      .getOrElse(this)
+
+  // only succeed if there is no waiting headers.
   def appendNewBlock(block: Block, fromPeer: PeerId): BlockFetcherState =
+    if (waitingHeaders.isEmpty)
+      unsafeAppendNewBlock(block, fromPeer)
+    else
+      this
+
+  // unsafe in terms of not checking waiting headers queue
+  private def unsafeAppendNewBlock(block: Block, fromPeer: PeerId): BlockFetcherState =
     withPeerForBlocks(fromPeer, Seq(block.header.number))
       .withPossibleNewTopAt(block.number)
       .withLastBlock(block.number)
       .copy(
-        readyBlocks = readyBlocks.enqueue(block),
-        waitingHeaders = waitingHeaders.filter(block.number != _.number)
+        readyBlocks = readyBlocks.enqueue(block)
       )
 
   def pickBlocks(amount: Int): Option[(NonEmptyList[Block], BlockFetcherState)] =