[ETCM-186] Fix strict pick with too few blocks (#723)

Author: Nicolás Tallar

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

@@ -68,14 +68,17 @@ class BlockFetcher(
   private def handleCommands(state: BlockFetcherState): Receive = {
     case PickBlocks(amount) => state.pickBlocks(amount) |> handlePickedBlocks(state) |> fetchBlocks
     case StrictPickBlocks(from, atLeastWith) =>
-      val minBlock = from.min(atLeastWith).max(1)
-      log.debug("Strict Pick blocks from {} to {}", from, atLeastWith)
+      // FIXME: Consider having StrictPickBlocks calls guaranteeing this
+      // from parameter could be negative or 0 so we should cap it to 1 if that's the case
+      val fromCapped = from.max(1)
+      val minBlock = fromCapped.min(atLeastWith).max(1)
+      log.debug("Strict Pick blocks from {} to {}", fromCapped, atLeastWith)
       log.debug("Lowest available block is {}", state.lowestBlock)
 
       val newState = if (minBlock < state.lowestBlock) {
         state.invalidateBlocksFrom(minBlock, None)._2
       } else {
-        state.strictPickBlocks(from, atLeastWith) |> handlePickedBlocks(state)
+        state.strictPickBlocks(fromCapped, atLeastWith) |> handlePickedBlocks(state)
       }
 
       fetchBlocks(newState)

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

@@ -107,9 +107,15 @@ case class BlockFetcherState(
       None
     }
 
+  /**
+    * Returns all the ready blocks but only if it includes blocks with number:
+    * - lower = min(from, atLeastWith)
+    * - upper = max(from, atLeastWith)
+    */
   def strictPickBlocks(from: BigInt, atLeastWith: BigInt): Option[(NonEmptyList[Block], BlockFetcherState)] = {
     val lower = from.min(atLeastWith)
     val upper = from.max(atLeastWith)
+
     readyBlocks.some
       .filter(_.headOption.exists(block => block.number <= lower))
       .filter(_.lastOption.exists(block => block.number >= upper))

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

@@ -105,7 +105,11 @@ class BlockImporter(
 
   private def importBlocks(blocks: NonEmptyList[Block]): ImportFn =
     importWith {
-      log.debug("Attempting to import blocks starting from {}", blocks.head.number)
+      log.debug(
+        "Attempting to import blocks starting from {} and ending with {}",
+        blocks.head.number,
+        blocks.last.number
+      )
       Future
         .successful(resolveBranch(blocks))
         .flatMap {

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

@@ -27,6 +27,7 @@ case class BlockHeader(
 
   override def toString: String = {
     s"""BlockHeader {
+       |hash: $hashAsHexString
        |parentHash: ${Hex.toHexString(parentHash.toArray[Byte])}
        |ommersHash: ${Hex.toHexString(ommersHash.toArray[Byte])}
        |beneficiary: ${Hex.toHexString(beneficiary.toArray[Byte])}

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

@@ -33,6 +33,5 @@ class BlockFetcherStateSpec extends TestKit(ActorSystem()) with AnyWordSpecLike
         newState.knownTop shouldEqual newBestBlock.number
       }
     }
-
   }
 }

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

@@ -134,38 +134,6 @@ class RegularSyncSpec extends RegularSyncFixtures with AnyWordSpecLike with Befo
     }
 
     "resolving branches" should {
-      trait FakeLedger { self: Fixture =>
-        class FakeLedgerImpl extends TestLedgerImpl {
-          override def importBlock(block: Block)(
-              implicit blockExecutionContext: ExecutionContext): Future[BlockImportResult] = {
-            val result: BlockImportResult = if (didTryToImportBlock(block)) {
-              DuplicateBlock
-            } else {
-              if (importedBlocks.isEmpty || bestBlock.isParentOf(block) || importedBlocks.exists(_.isParentOf(block))) {
-                importedBlocks.add(block)
-                BlockImportedToTop(List(BlockData(block, Nil, block.header.difficulty)))
-              } else if (block.number > bestBlock.number) {
-                importedBlocks.add(block)
-                BlockEnqueued
-              } else {
-                BlockImportFailed("foo")
-              }
-            }
-
-            Future.successful(result)
-          }
-
-          override def resolveBranch(headers: scala.Seq[BlockHeader]): BranchResolutionResult = {
-            val importedHashes = importedBlocks.map(_.hash).toSet
-
-            if (importedBlocks.isEmpty || (importedHashes.contains(headers.head.parentHash) && headers.last.number > bestBlock.number)) {
-              NewBetterBranch(Nil)
-            } else {
-              UnknownBranch
-            }
-          }
-        }
-      }
 
       "go back to earlier block in order to find a common parent with new branch" in new Fixture(testSystem)
       with FakeLedger {
@@ -216,6 +184,55 @@ class RegularSyncSpec extends RegularSyncFixtures with AnyWordSpecLike with Befo
       }
     }
 
+    "go back to earlier positive block in order to resolve a fork when branch smaller than branch resolution size" in new Fixture(testSystem)
+      with FakeLedger {
+      implicit val ec: ExecutionContext = system.dispatcher
+      override lazy val blockchain: BlockchainImpl = stub[BlockchainImpl]
+      (blockchain.getBestBlockNumber _).when().onCall(() => ledger.bestBlock.number)
+      override lazy val ledger: TestLedgerImpl = new FakeLedgerImpl()
+      override lazy val syncConfig = defaultSyncConfig.copy(
+        syncRetryInterval = 1.second,
+        printStatusInterval = 0.5.seconds,
+        branchResolutionRequestSize = 12, // Over the original branch size
+
+        // Big so that they don't impact the test
+        blockHeadersPerRequest = 50,
+        blockBodiesPerRequest = 50,
+        blocksBatchSize = 50
+      )
+
+      val originalBranch = getBlocks(10, genesis)
+      val betterBranch = getBlocks(originalBranch.size * 2, genesis)
+
+      class ForkingAutoPilot(blocksToRespond: List[Block], forkedBlocks: Option[List[Block]])
+        extends PeersClientAutoPilot(blocksToRespond) {
+        override def overrides(sender: ActorRef): PartialFunction[Any, Option[AutoPilot]] = {
+          case req @ PeersClient.Request(GetBlockBodies(hashes), _, _) =>
+            val defaultResult = defaultHandlers(sender)(req)
+            if (forkedBlocks.nonEmpty && hashes.contains(blocksToRespond.last.hash)) {
+              Some(new ForkingAutoPilot(forkedBlocks.get, None))
+            } else
+              defaultResult
+        }
+      }
+
+      peersClient.setAutoPilot(new ForkingAutoPilot(originalBranch, Some(betterBranch)))
+
+      Await.result(ledger.importBlock(genesis), remainingOrDefault)
+
+      regularSync ! RegularSync.Start
+
+      peerEventBus.expectMsgClass(classOf[Subscribe])
+      val blockFetcher = peerEventBus.sender()
+      peerEventBus.reply(MessageFromPeer(NewBlock(originalBranch.last, originalBranch.last.number), defaultPeer.id))
+
+      awaitCond(ledger.bestBlock == originalBranch.last, 5.seconds)
+
+      // As node will be on top, we have to re-trigger the fetching process by simulating a block from the fork being broadcasted
+      blockFetcher ! MessageFromPeer(NewBlock(betterBranch.last, betterBranch.last.number), defaultPeer.id)
+      awaitCond(ledger.bestBlock == betterBranch.last, 5.seconds)
+    }
+
     "fetching state node" should {
       abstract class MissingStateNodeFixture(system: ActorSystem) extends Fixture(system) {
         val failingBlock: Block = testBlocksChunked.head.head
@@ -457,4 +474,36 @@ class RegularSyncSpec extends RegularSyncFixtures with AnyWordSpecLike with Befo
       }
     }
   }
+
+  trait FakeLedger { self: Fixture =>
+    class FakeLedgerImpl extends TestLedgerImpl {
+      override def importBlock(block: Block)(
+        implicit blockExecutionContext: ExecutionContext): Future[BlockImportResult] = {
+        val result: BlockImportResult = if (didTryToImportBlock(block)) {
+          DuplicateBlock
+        } else {
+          if (importedBlocks.isEmpty || bestBlock.isParentOf(block) || importedBlocks.exists(_.isParentOf(block))) {
+            importedBlocks.add(block)
+            BlockImportedToTop(List(BlockData(block, Nil, block.header.difficulty)))
+          } else if (block.number > bestBlock.number) {
+            importedBlocks.add(block)
+            BlockEnqueued
+          } else {
+            BlockImportFailed("foo")
+          }
+        }
+
+        Future.successful(result)
+      }
+
+      override def resolveBranch(headers: Seq[BlockHeader]): BranchResolutionResult = {
+        val importedHashes = importedBlocks.map(_.hash).toSet
+
+        if (importedBlocks.isEmpty || (importedHashes.contains(headers.head.parentHash) && headers.last.number > bestBlock.number))
+          NewBetterBranch(Nil)
+        else
+          UnknownBranch
+      }
+    }
+  }
 }