Etcm 290 invalid branches built by fetcher (#787)

- Fix for invalid chains passed to importer
- Add chain validation on headers added to queue, remove pending requests check when checking for top on fetch side
- Temporarily remove peers blacklisting after receiving useless response, improve peer message logging
- Fix problem with fetcher being stuck with headers from rejected fork

Author: kapke

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

@@ -114,12 +114,12 @@ class BlockFetcher(
         } else {
           log.debug("Fetched {} headers starting from block {}", headers.size, headers.headOption.map(_.number))
 
-          state.validatedHeaders(headers) match {
+          state.appendHeaders(headers) match {
             case Left(err) =>
-              peersClient ! BlacklistPeer(peer.id, err)
+              log.info("Dismissed received headers due to: {}", err)
               state.withHeaderFetchReceived
-            case Right(validHeaders) =>
-              state.withHeaderFetchReceived.appendHeaders(validHeaders)
+            case Right(updatedState) =>
+              updatedState.withHeaderFetchReceived
           }
         }
 
@@ -144,7 +144,7 @@ class BlockFetcher(
           state.withBodiesFetchReceived
         } else {
           log.debug("Fetched {} block bodies", bodies.size)
-          state.withBodiesFetchReceived.addBodies(peer, bodies)
+          state.withBodiesFetchReceived.addBodies(peer.id, bodies)
         }
 
       fetchBlocks(newState)
@@ -188,40 +188,24 @@ class BlockFetcher(
         case Left(_) => state
         case Right(validHashes) => state.withPossibleNewTopAt(validHashes.lastOption.map(_.number))
       }
-
       supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-
       fetchBlocks(newState)
     case MessageFromPeer(NewBlock(_, block, _), peerId) =>
+      //TODO ETCM-389: Handle mined, checkpoint and new blocks uniformly
+      log.debug("Received NewBlock {}", block.idTag)
       val newBlockNr = block.number
       val nextExpectedBlock = state.lastFullBlockNumber + 1
 
-      log.debug("Received NewBlock nr {}", newBlockNr)
-
-      // we're on top, so we can pass block directly to importer
-      if (newBlockNr == nextExpectedBlock && state.isOnTop) {
-        log.debug("Pass block directly to importer")
+      if (state.isOnTop && newBlockNr == nextExpectedBlock) {
+        log.debug("Passing block directly to importer")
         val newState = state.withPeerForBlocks(peerId, Seq(newBlockNr)).withKnownTopAt(newBlockNr)
         state.importer ! OnTop
         state.importer ! ImportNewBlock(block, peerId)
         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) {
-        log.debug("Enqueue new block for import")
-        val newState = state.appendNewBlock(block, peerId)
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        context become started(newState)
-        // waiting for some bodies but we don't have this header yet - at least we can use new block header
-      } else if (newBlockNr == state.nextToLastBlock && !state.isFetchingHeaders) {
-        log.debug("Waiting for bodies. Add only headers")
-        val newState = state.appendHeaders(List(block.header))
-        supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
-        fetchBlocks(newState)
-        // we're far from top
-      } else if (newBlockNr > nextExpectedBlock) {
-        log.debug("Far from top")
-        val newState = state.withKnownTopAt(newBlockNr)
+      } else {
+        log.debug("Ignoring received block as it doesn't match local state or fetch side is not on top")
+        val newState = state.withPossibleNewTopAt(block.number)
         supervisor ! ProgressProtocol.GotNewBlock(newState.knownTop)
         fetchBlocks(newState)
       }
@@ -236,17 +220,17 @@ class BlockFetcher(
     //ex. After a successful handshake, fetcher will receive the info about the header of the peer best block
     case MessageFromPeer(BlockHeaders(headers), _) =>
       headers.lastOption.map { bh =>
-        log.debug(s"Candidate for new top at block ${bh.number}, current know top ${state.knownTop}")
+        log.debug(s"Candidate for new top at block ${bh.number}, current known top ${state.knownTop}")
         val newState = state.withPossibleNewTopAt(bh.number)
         fetchBlocks(newState)
       }
     //keep fetcher state updated in case new checkpoint block or mined block was imported
-    case InternalLastBlockImport(blockNr) => {
+    case InternalLastBlockImport(blockNr) =>
       log.debug(s"New last block $blockNr imported from the inside")
       val newLastBlock = blockNr.max(state.lastBlock)
       val newState = state.withLastBlock(newLastBlock).withPossibleNewTopAt(blockNr)
+
       fetchBlocks(newState)
-    }
   }
 
   private def handlePickedBlocks(
@@ -277,7 +261,7 @@ class BlockFetcher(
       .getOrElse(fetcherState)
 
   private def fetchHeaders(state: BlockFetcherState): Unit = {
-    val blockNr = state.nextToLastBlock
+    val blockNr = state.nextBlockToFetch
     val amount = syncConfig.blockHeadersPerRequest
 
     fetchHeadersFrom(blockNr, amount) pipeTo self

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

@@ -3,12 +3,11 @@ package io.iohk.ethereum.blockchain.sync.regular
 import akka.actor.ActorRef
 import akka.util.ByteString
 import cats.data.NonEmptyList
-import io.iohk.ethereum.domain.{Block, BlockHeader, BlockBody, HeadersSeq}
-import io.iohk.ethereum.network.{Peer, PeerId}
+import cats.implicits._
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcherState._
+import io.iohk.ethereum.domain.{Block, BlockBody, BlockHeader, HeadersSeq}
+import io.iohk.ethereum.network.PeerId
 import io.iohk.ethereum.network.p2p.messages.PV62.BlockHash
-import BlockFetcherState._
-import cats.syntax.either._
-import cats.syntax.option._
 
 import scala.collection.immutable.Queue
 
@@ -52,7 +51,7 @@ case class BlockFetcherState(
 
   def hasFetchedTopHeader: Boolean = lastBlock == knownTop
 
-  def isOnTop: Boolean = !isFetching && hasFetchedTopHeader && hasEmptyBuffer
+  def isOnTop: Boolean = hasFetchedTopHeader && hasEmptyBuffer
 
   def hasReachedSize(size: Int): Boolean = (readyBlocks.size + waitingHeaders.size) >= size
 
@@ -68,26 +67,43 @@ case class BlockFetcherState(
       .orElse(waitingHeaders.headOption.map(_.number))
       .getOrElse(lastBlock)
 
-  def nextToLastBlock: BigInt = lastBlock + 1
+  /**
+    * Next block number to be fetched, calculated in a way to maintain local queues consistency,
+    * even if `lastBlock` property is much higher - it's more important to have this consistency
+    * here and allow standard rollback/reorganization mechanisms to kick in if we get too far with mining,
+    * therefore `lastBlock` is used here only if blocks and headers queues are empty
+    */
+  def nextBlockToFetch: BigInt = waitingHeaders.lastOption
+    .map(_.number)
+    .orElse(readyBlocks.lastOption.map(_.number))
+    .getOrElse(lastBlock) + 1
 
   def takeHashes(amount: Int): Seq[ByteString] = waitingHeaders.take(amount).map(_.hash)
 
-  def appendHeaders(headers: Seq[BlockHeader]): BlockFetcherState =
-    withPossibleNewTopAt(headers.lastOption.map(_.number))
-      .copy(
-        waitingHeaders = waitingHeaders ++ headers.filter(_.number > lastBlock).sortBy(_.number),
-        lastBlock = HeadersSeq.lastNumber(headers).getOrElse(lastBlock)
-      )
+  def appendHeaders(headers: Seq[BlockHeader]): Either[String, BlockFetcherState] =
+    validatedHeaders(headers.sortBy(_.number)).map(validHeaders => {
+      val lastNumber = HeadersSeq.lastNumber(validHeaders)
+      withPossibleNewTopAt(lastNumber)
+        .copy(
+          waitingHeaders = waitingHeaders ++ validHeaders,
+          lastBlock = lastNumber.getOrElse(lastBlock)
+        )
+    })
 
-  def validatedHeaders(headers: Seq[BlockHeader]): Either[String, Seq[BlockHeader]] =
+  /**
+    * Validates received headers consistency and their compatibilty with the state
+    * TODO ETCM-370: This needs to be more fine-grained and detailed so blacklisting can be re-enabled
+    */
+  private def validatedHeaders(headers: Seq[BlockHeader]): Either[String, Seq[BlockHeader]] =
     if (headers.isEmpty) {
       Right(headers)
     } else {
       headers
         .asRight[String]
-        .ensure("Given headers are not sequence with already fetched ones")(_.head.number <= nextToLastBlock)
-        .ensure("Given headers aren't better than already fetched ones")(_.last.number > lastBlock)
         .ensure("Given headers should form a sequence without gaps")(HeadersSeq.areChain)
+        .ensure("Given headers do not form a chain with already stored ones")(headers =>
+          (waitingHeaders.lastOption, headers.headOption).mapN(_ isParentOf _).getOrElse(true)
+        )
     }
 
   def validateNewBlockHashes(hashes: Seq[BlockHash]): Either[String, Seq[BlockHash]] =
@@ -100,15 +116,26 @@ case class BlockFetcherState(
         }
       )
 
-  def addBodies(peer: Peer, bodies: Seq[BlockBody]): BlockFetcherState = {
-    val (matching, waiting) = waitingHeaders.splitAt(bodies.length)
-    val blocks = matching.zip(bodies).map((Block.apply _).tupled)
-
-    withPeerForBlocks(peer.id, blocks.map(_.header.number))
-      .copy(
-        readyBlocks = readyBlocks.enqueue(blocks),
-        waitingHeaders = waiting
-      )
+  /**
+    * Matches bodies with headers in queue and adding matched bodies to the blocks.
+    * If bodies is empty collection - headers in queue are removed as the cause is:
+    *   - the headers are from rejected fork and therefore it won't be possible to resolve bodies for them
+    *   - given peer is still syncing (quite unlikely due to preference of peers with best total difficulty
+    *     when making a request)
+    */
+  def addBodies(peerId: PeerId, bodies: Seq[BlockBody]): BlockFetcherState = {
+    if (bodies.isEmpty) {
+      copy(waitingHeaders = Queue.empty)
+    } else {
+      val (matching, waiting) = waitingHeaders.splitAt(bodies.length)
+      val blocks = matching.zip(bodies).map((Block.apply _).tupled)
+
+      withPeerForBlocks(peerId, blocks.map(_.header.number))
+        .copy(
+          readyBlocks = readyBlocks.enqueue(blocks),
+          waitingHeaders = waiting
+        )
+    }
   }
 
   def appendNewBlock(block: Block, fromPeer: PeerId): BlockFetcherState =

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

@@ -70,11 +70,13 @@ class BlockImporter(
       SignedTransaction.retrieveSendersInBackGround(blocks.toList.map(_.body))
       importBlocks(blocks)(state)
 
+    //TODO ETCM-389: Handle mined, checkpoint and new blocks uniformly
     case MinedBlock(block) =>
       if (!state.importing) {
         importMinedBlock(block, state)
       }
 
+    //TODO ETCM-389: Handle mined, checkpoint and new blocks uniformly
     case nc @ NewCheckpoint(parentHash, signatures) =>
       if (state.importing) {
         //We don't want to lose a checkpoint

src/main/scala/io/iohk/ethereum/domain/Block.scala

@@ -23,7 +23,7 @@ case class Block(header: BlockHeader, body: BlockBody) {
 
   val hasCheckpoint: Boolean = header.hasCheckpoint
 
-  def isParentOf(child: Block): Boolean = number + 1 == child.number && child.header.parentHash == hash
+  def isParentOf(child: Block): Boolean = header.isParentOf(child.header)
 }
 
 object Block {

src/main/scala/io/iohk/ethereum/domain/BlockHeader.scala

@@ -47,6 +47,8 @@ case class BlockHeader(
 
   val hasCheckpoint: Boolean = checkpoint.isDefined
 
+  def isParentOf(child: BlockHeader): Boolean = number + 1 == child.number && child.parentHash == hash
+
   override def toString: String = {
     val (treasuryOptOutString: String, checkpointString: String) = extraFields match {
       case HefPostEcip1097(definedOptOut, maybeCheckpoint) =>

src/test/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherSpec.scala

@@ -1,16 +1,15 @@
 package io.iohk.ethereum.blockchain.sync.regular
 
 import java.net.InetSocketAddress
-
 import akka.actor.ActorSystem
 import akka.testkit.{TestKit, TestProbe}
 import com.miguno.akka.testing.VirtualTime
 import io.iohk.ethereum.BlockHelpers
 import io.iohk.ethereum.Fixtures.{Blocks => FixtureBlocks}
 import io.iohk.ethereum.blockchain.sync.PeersClient.BlacklistPeer
-import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.InvalidateBlocksFrom
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.{InternalLastBlockImport, InvalidateBlocksFrom, PickBlocks}
 import io.iohk.ethereum.blockchain.sync.{PeersClient, TestSyncConfig}
-import io.iohk.ethereum.domain.{Block, ChainWeight}
+import io.iohk.ethereum.domain.{Block, ChainWeight, HeadersSeq}
 import io.iohk.ethereum.network.Peer
 import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
 import io.iohk.ethereum.network.PeerEventBusActor.{PeerSelector, Subscribe}
@@ -124,6 +123,78 @@ class BlockFetcherSpec extends TestKit(ActorSystem("BlockFetcherSpec_System")) w
       peersClient.expectMsgClass(classOf[BlacklistPeer])
       peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
     }
+
+    "should ensure blocks passed to importer are always forming chain" in new TestSetup {
+      startFetcher()
+
+      triggerFetching()
+
+      val firstBlocksBatch = BlockHelpers.generateChain(syncConfig.blockHeadersPerRequest, FixtureBlocks.Genesis.block)
+      val secondBlocksBatch = BlockHelpers.generateChain(syncConfig.blockHeadersPerRequest, firstBlocksBatch.last)
+      val alternativeSecondBlocksBatch =
+        BlockHelpers.generateChain(syncConfig.blockHeadersPerRequest, firstBlocksBatch.last)
+
+      // Fetcher requests for headers
+      val firstGetBlockHeadersRequest =
+        GetBlockHeaders(Left(1), syncConfig.blockHeadersPerRequest, skip = 0, reverse = false)
+      peersClient.expectMsgPF() { case PeersClient.Request(msg, _, _) if msg == firstGetBlockHeadersRequest => () }
+
+      // Respond first headers request
+      val firstGetBlockHeadersResponse = BlockHeaders(firstBlocksBatch.map(_.header))
+      peersClient.reply(PeersClient.Response(fakePeer, firstGetBlockHeadersResponse))
+
+      // Second headers request with response pending
+      val secondGetBlockHeadersRequest = GetBlockHeaders(
+        Left(secondBlocksBatch.head.number),
+        syncConfig.blockHeadersPerRequest,
+        skip = 0,
+        reverse = false
+      )
+      // Save the reference to respond to the ask pattern on fetcher
+      val refForAnswerSecondHeaderReq = peersClient.expectMsgPF() {
+        case PeersClient.Request(msg, _, _) if msg == secondGetBlockHeadersRequest => peersClient.lastSender
+      }
+
+      // First bodies request
+      val firstGetBlockBodiesRequest = GetBlockBodies(firstBlocksBatch.map(_.hash))
+      val refForAnswerFirstBodiesReq = peersClient.expectMsgPF() {
+        case PeersClient.Request(msg, _, _) if msg == firstGetBlockBodiesRequest => peersClient.lastSender
+      }
+
+      // Block 16 is mined (we could have reached this stage due to invalidation messages sent to the fetcher)
+      val minedBlock = alternativeSecondBlocksBatch.drop(5).head
+      val minedBlockNumber = minedBlock.number
+      blockFetcher ! InternalLastBlockImport(minedBlockNumber)
+
+      // Answer pending requests: first block bodies request + second block headers request
+      val secondGetBlockHeadersResponse = BlockHeaders(secondBlocksBatch.map(_.header))
+      peersClient.send(refForAnswerSecondHeaderReq, PeersClient.Response(fakePeer, secondGetBlockHeadersResponse))
+
+      val firstGetBlockBodiesResponse = BlockBodies(firstBlocksBatch.map(_.body))
+      peersClient.send(refForAnswerFirstBodiesReq, PeersClient.Response(fakePeer, firstGetBlockBodiesResponse))
+
+      // Third headers request with response pending
+      peersClient.expectMsgPF() { case PeersClient.Request(GetBlockHeaders(_, _, _, _), _, _) =>
+        peersClient.lastSender
+      }
+
+      // Second bodies request
+      val refForAnswerSecondBodiesReq = peersClient.expectMsgPF() { case PeersClient.Request(GetBlockBodies(_), _, _) =>
+        peersClient.lastSender
+      }
+      peersClient.send(
+        refForAnswerSecondBodiesReq,
+        PeersClient.Response(fakePeer, BlockBodies(alternativeSecondBlocksBatch.drop(6).map(_.body)))
+      )
+
+      importer.send(blockFetcher, PickBlocks(syncConfig.blocksBatchSize))
+      importer.ignoreMsg({ case BlockImporter.NotOnTop => true })
+      importer.expectMsgPF() { case BlockFetcher.PickedBlocks(blocks) =>
+        val headers = blocks.map(_.header).toList
+
+        assert(HeadersSeq.areChain(headers))
+      }
+    }
   }
 
   trait TestSetup extends TestSyncConfig {

src/test/scala/io/iohk/ethereum/blockchain/sync/regular/BlockFetcherStateSpec.scala

@@ -2,12 +2,15 @@ package io.iohk.ethereum.blockchain.sync.regular
 
 import akka.actor.ActorSystem
 import akka.testkit.{TestKit, TestProbe}
-import io.iohk.ethereum.domain.Block
+import io.iohk.ethereum.BlockHelpers
 import io.iohk.ethereum.Fixtures.Blocks.ValidBlock
+import io.iohk.ethereum.domain.Block
 import io.iohk.ethereum.network.PeerId
 import org.scalatest.matchers.should.Matchers
 import org.scalatest.wordspec.AnyWordSpecLike
 
+import scala.collection.immutable.Queue
+
 class BlockFetcherStateSpec extends TestKit(ActorSystem()) with AnyWordSpecLike with Matchers {
   "BlockFetcherState" when {
     "invalidating blocks" should {
@@ -33,5 +36,20 @@ class BlockFetcherStateSpec extends TestKit(ActorSystem()) with AnyWordSpecLike
         newState.knownTop shouldEqual newBestBlock.number
       }
     }
+
+    "handling new block bodies" should {
+      "clear headers queue if got empty list of bodies" in {
+        val headers = BlockHelpers.generateChain(5, BlockHelpers.genesis).map(_.header)
+        val peer = PeerId("foo")
+
+        val result = BlockFetcherState
+          .initial(TestProbe().ref, 0)
+          .appendHeaders(headers)
+          .map(_.addBodies(peer, List()))
+          .map(_.waitingHeaders)
+
+        assert(result === Right(Queue.empty))
+      }
+    }
   }
 }