[ETCM-247] Fix Fetcher updates

- Fetcher will update last block and know top(only if is needed) if a mined or checkpointed block was imported
- Fetcher will update know top if detect a block header with better top

fix

Author: mirkoAlic

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

@@ -50,6 +50,7 @@ class BlockFetcher(
   private def idle(): Receive = handleCommonMessages(None) orElse { case Start(importer, blockNr) =>
     BlockFetcherState.initial(importer, blockNr) |> fetchBlocks
     peerEventBus ! Subscribe(MessageClassifier(Set(NewBlock.code, NewBlockHashes.code), PeerSelector.AllPeers))
+    peerEventBus ! Subscribe(MessageClassifier(Set(BlockHeaders.code), PeerSelector.AllPeers))
   }
 
   def handleCommonMessages(state: Option[BlockFetcherState]): Receive = { case PrintStatus =>
@@ -63,7 +64,8 @@ class BlockFetcher(
       handleNewBlockMessages(state) orElse
       handleHeadersMessages(state) orElse
       handleBodiesMessages(state) orElse
-      handleStateNodeMessages(state)
+      handleStateNodeMessages(state) orElse
+      handlePossibleTopUpdate(state)
 
   private def handleCommands(state: BlockFetcherState): Receive = {
     case PickBlocks(amount) => state.pickBlocks(amount) |> handlePickedBlocks(state) |> fetchBlocks
@@ -183,6 +185,21 @@ class BlockFetcher(
       fetchBlocks(newState)
   }
 
+  private def handlePossibleTopUpdate(state: BlockFetcherState): Receive = {
+    case MessageFromPeer(BlockHeaders(headers), _) =>
+      headers.headOption.map { bh =>
+        log.debug(s"Possible new top at block ${bh.number}")
+        val newState = state.withPossibleNewTopAt(bh.number)
+        fetchBlocks(newState)
+      }
+    //keep fetcher state updated in case new checkpoint block or mined block was imported
+    case LastBlockChanged(blockNr) => {
+      log.debug(s"New last block $blockNr imported from the inside")
+      val newState = state.withLastBlock(blockNr).withPossibleNewTopAt(blockNr)
+      fetchBlocks(newState)
+    }
+  }
+
   private def handlePickedBlocks(
       state: BlockFetcherState
   )(pickResult: Option[(NonEmptyList[Block], BlockFetcherState)]): BlockFetcherState =
@@ -306,6 +323,7 @@ object BlockFetcher {
       new InvalidateBlocksFrom(from, reason, toBlacklist)
   }
   case class BlockImportFailed(blockNr: BigInt, reason: String) extends FetchMsg
+  case class LastBlockChanged(blockNr: BigInt) extends FetchMsg
   case object RetryBodiesRequest extends FetchMsg
   case object RetryHeadersRequest extends FetchMsg
 

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

@@ -201,15 +201,35 @@ class BlockImporter(
     }
 
   private def importMinedBlock(block: Block, state: ImporterState): Unit =
-    importBlock(block, new MinedBlockImportMessages(block), informFetcherOnFail = false)(state)
+    importBlock(
+      block,
+      new MinedBlockImportMessages(block),
+      informFetcherOnFail = false,
+      informFetcherOnLastBlockChanged = true
+    )(state)
 
   private def importCheckpointBlock(block: Block, state: ImporterState): Unit =
-    importBlock(block, new CheckpointBlockImportMessages(block), informFetcherOnFail = false)(state)
+    importBlock(
+      block,
+      new CheckpointBlockImportMessages(block),
+      informFetcherOnFail = false,
+      informFetcherOnLastBlockChanged = true
+    )(state)
 
   private def importNewBlock(block: Block, peerId: PeerId, state: ImporterState): Unit =
-    importBlock(block, new NewBlockImportMessages(block, peerId), informFetcherOnFail = true)(state)
-
-  private def importBlock(block: Block, importMessages: ImportMessages, informFetcherOnFail: Boolean): ImportFn = {
+    importBlock(
+      block,
+      new NewBlockImportMessages(block, peerId),
+      informFetcherOnFail = true,
+      informFetcherOnLastBlockChanged = false
+    )(state)
+
+  private def importBlock(
+      block: Block,
+      importMessages: ImportMessages,
+      informFetcherOnFail: Boolean,
+      informFetcherOnLastBlockChanged: Boolean
+  ): ImportFn = {
     def doLog(entry: ImportMessages.LogEntry): Unit = log.log(entry._1, entry._2)
 
     importWith {
@@ -222,6 +242,9 @@ class BlockImporter(
             val (blocks, tds) = importedBlocksData.map(data => (data.block, data.td)).unzip
             broadcastBlocks(blocks, tds)
             updateTxPool(importedBlocksData.map(_.block), Seq.empty)
+            if (informFetcherOnLastBlockChanged) {
+              fetcher ! BlockFetcher.LastBlockChanged(blocks.last.number)
+            }
 
           case BlockEnqueued => ()
 
@@ -232,6 +255,9 @@ class BlockImporter(
           case ChainReorganised(oldBranch, newBranch, totalDifficulties) =>
             updateTxPool(newBranch, oldBranch)
             broadcastBlocks(newBranch, totalDifficulties)
+            if (informFetcherOnLastBlockChanged) {
+              fetcher ! BlockFetcher.LastBlockChanged(newBranch.last.number)
+            }
 
           case BlockImportFailed(error) =>
             if (informFetcherOnFail) {