[ETCM-316] Fast-sync branch resolver (#887)

* init new actor, FastSyncBranchResolver

* added searching mode in fast sync branch resolver

* fix style

* add schedule when don't get peers

* Added ut in class FastSyncBranchResolver

* change messages

* fix case object...

* add new unit test

* add new unit test

* change name getFirstCommonBlock

* change name to batch

* init new actor, FastSyncBranchResolver

* added searching mode in fast sync branch resolver

* fix style

* add schedule when don't get peers

* Added ut in class FastSyncBranchResolver

* change messages

* fix case object...

* add new unit test

* add new unit test

* create actor FastSyncBranchResolver

* change name getFirstCommonBlock

* change name to batch

* Reformat triggered by sbt pp

* Cleanup and simplify

* Handle error cases

* Fix tests

* [ETCM-316] Add more tests and fix binary search logic

* [ETCM-316] Finish tests for branch resolving

* [ETCM-316] Cleanup

* [ETCM-316] Small test improvements

* [ETCM-316] Log binary search state

* [ETCM-316] Move some logging to improve readability

* [ETCM-316] Remove unneeded errors and reformat

* [ETCM-316] Handle branch resolution failure

* [ETCM-316] Address PR comments

* [ETCM-316] Remove unnecessary string interpolation

Co-authored-by: Petra Bierleutgeb <[email protected]>

Author: biandratti

src/it/scala/io/iohk/ethereum/ledger/BlockImporterItSpec.scala

@@ -21,7 +21,12 @@ import org.scalatest.matchers.should.Matchers
 
 import scala.concurrent.duration._
 
-class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators with AsyncFlatSpecLike with Matchers with BeforeAndAfterAll {
+class BlockImporterItSpec
+    extends MockFactory
+    with TestSetupWithVmAndValidators
+    with AsyncFlatSpecLike
+    with Matchers
+    with BeforeAndAfterAll {
 
   implicit val testScheduler = Scheduler.fixedPool("test", 32)
 
@@ -57,11 +62,15 @@ class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators
       ethCompatibleStorage = true
     )
 
-  override lazy val ledger = new TestLedgerImpl(successValidators)  {
-    override private[ledger] lazy val blockExecution = new BlockExecution(blockchain, blockchainConfig, consensus.blockPreparator, blockValidation) {
-      override def executeAndValidateBlock(block: Block, alreadyValidated: Boolean = false): Either[BlockExecutionError, Seq[Receipt]] =
-        Right(BlockResult(emptyWorld).receipts)
-    }
+  override lazy val ledger = new TestLedgerImpl(successValidators) {
+    override private[ledger] lazy val blockExecution =
+      new BlockExecution(blockchain, blockchainConfig, consensus.blockPreparator, blockValidation) {
+        override def executeAndValidateBlock(
+            block: Block,
+            alreadyValidated: Boolean = false
+        ): Either[BlockExecutionError, Seq[Receipt]] =
+          Right(BlockResult(emptyWorld).receipts)
+      }
   }
 
   val blockImporter = system.actorOf(
@@ -75,7 +84,8 @@ class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators
       pendingTransactionsManagerProbe.ref,
       checkpointBlockGenerator,
       supervisor.ref
-    ))
+    )
+  )
 
   val genesisBlock = blockchain.genesisBlock
   val block1: Block = getBlock(genesisBlock.number + 1, parent = genesisBlock.header.hash)
@@ -119,7 +129,8 @@ class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators
         pendingTransactionsManagerProbe.ref,
         checkpointBlockGenerator,
         supervisor.ref
-      ))
+      )
+    )
 
     blockImporter ! BlockImporter.Start
     blockImporter ! BlockFetcher.PickedBlocks(NonEmptyList.fromListUnsafe(newBranch))
@@ -142,7 +153,6 @@ class BlockImporterItSpec extends MockFactory with TestSetupWithVmAndValidators
     blockchain.getBestBlock().get shouldEqual newBlock3
   }
 
-
   it should "switch to a branch with a checkpoint" in {
 
     val checkpoint = ObjectGenerators.fakeCheckpointGen(3, 3).sample.get

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

@@ -81,7 +81,7 @@ object Blacklist {
 
     case object WrongBlockHeaders extends BlacklistReason {
       val reasonType: BlacklistReasonType = WrongBlockHeadersType
-      val description: String = "Wrong blockheaders response (empty or not chain forming)"
+      val description: String = "Wrong blockheaders response: Peer didn't respond with requested block headers."
     }
     case object BlockHeaderValidationFailed extends BlacklistReason {
       val reasonType: BlacklistReasonType = BlockHeaderValidationFailedType

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

@@ -17,6 +17,8 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
 
   private implicit val ec: ExecutionContext = context.dispatcher
 
+  private val bigIntReverseOrdering: Ordering[BigInt] = Ordering[BigInt].reverse
+
   def etcPeerManager: ActorRef
   def peerEventBus: ActorRef
   def blacklist: Blacklist
@@ -44,6 +46,11 @@ trait PeerListSupportNg { self: Actor with ActorLogging =>
 
   def getPeerById(peerId: PeerId): Option[Peer] = handshakedPeers.get(peerId).map(_.peer)
 
+  def getPeerWithHighestBlock: Option[PeerWithInfo] =
+    peersToDownloadFrom.values.toList.sortBy { case PeerWithInfo(_, peerInfo) =>
+      peerInfo.maxBlockNumber
+    }(bigIntReverseOrdering).headOption
+
   def blacklistIfHandshaked(peerId: PeerId, duration: FiniteDuration, reason: BlacklistReason): Unit =
     handshakedPeers.get(peerId).foreach(_ => blacklist.add(peerId, duration, reason))
 

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

@@ -24,15 +24,15 @@ class PeerRequestHandler[RequestMsg <: Message, ResponseMsg <: Message: ClassTag
 
   import PeerRequestHandler._
 
-  val initiator: ActorRef = context.parent
+  private val initiator: ActorRef = context.parent
 
-  val timeout: Cancellable = scheduler.scheduleOnce(responseTimeout, self, Timeout)
+  private val timeout: Cancellable = scheduler.scheduleOnce(responseTimeout, self, Timeout)
 
-  val startTime: Long = System.currentTimeMillis()
+  private val startTime: Long = System.currentTimeMillis()
 
   private def subscribeMessageClassifier = MessageClassifier(Set(responseMsgCode), PeerSelector.WithId(peer.id))
 
-  def timeTakenSoFar(): Long = System.currentTimeMillis() - startTime
+  private def timeTakenSoFar(): Long = System.currentTimeMillis() - startTime
 
   override def preStart(): Unit = {
     etcPeerManager ! EtcPeerManagerActor.SendMessage(toSerializable(requestMsg), peer.id)
@@ -79,8 +79,8 @@ object PeerRequestHandler {
   )(implicit scheduler: Scheduler, toSerializable: RequestMsg => MessageSerializable): Props =
     Props(new PeerRequestHandler(peer, responseTimeout, etcPeerManager, peerEventBus, requestMsg, responseMsgCode))
 
-  case class RequestFailed(peer: Peer, reason: String)
-  case class ResponseReceived[T](peer: Peer, response: T, timeTaken: Long)
+  final case class RequestFailed(peer: Peer, reason: String)
+  final case class ResponseReceived[T](peer: Peer, response: T, timeTaken: Long)
 
   private case object Timeout
 }

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

@@ -136,7 +136,11 @@ class FastSync(
     private val syncStateStorageActor = context.actorOf(Props[StateStorageActor](), "state-storage")
     syncStateStorageActor ! fastSyncStateStorage
 
-    private val branchResolver = context.actorOf(Props[FastSyncBranchResolver](), "fast-sync-branch-resolver")
+    private val branchResolver = context.actorOf(
+      FastSyncBranchResolverActor
+        .props(self, peerEventBus, etcPeerManager, blockchain, blacklist, syncConfig, appStateStorage, scheduler),
+      "fast-sync-branch-resolver"
+    )
 
     private val syncStateScheduler = context.actorOf(
       SyncStateSchedulerActor
@@ -288,13 +292,13 @@ class FastSync(
 
         // Start branch resolution and wait for response from the FastSyncBranchResolver actor.
         context become waitingForBranchResolution
-        branchResolver ! FastSyncBranchResolver.StartBranchResolver
+        branchResolver ! FastSyncBranchResolverActor.StartBranchResolver
         currentSkeleton = None
       }
     }
 
     private def waitingForBranchResolution: Receive = handleStatus orElse {
-      case FastSyncBranchResolver.BranchResolvedSuccessful(firstCommonBlockNumber, resolvedPeer) =>
+      case FastSyncBranchResolverActor.BranchResolvedSuccessful(firstCommonBlockNumber, resolvedPeer) =>
         // Reset the batch failures count
         batchFailuresCount = 0
 
@@ -303,6 +307,9 @@ class FastSync(
         masterPeer = Some(resolvedPeer)
         context become receive
         processSyncing()
+      case _: FastSyncBranchResolverActor.BranchResolutionFailed =>
+        // there isn't much we can do if we don't find a branch/peer to continue syncing, so let's try again
+        branchResolver ! FastSyncBranchResolverActor.StartBranchResolver
     }
 
     private def blockHeadersError(peer: Peer): Unit = {

src/main/scala/io/iohk/ethereum/blockchain/sync/fast/FastSyncBranchResolver.scala

@@ -1,18 +1,116 @@
 package io.iohk.ethereum.blockchain.sync.fast
 
-import akka.actor.Actor
+import cats.data.NonEmptyList
+import io.iohk.ethereum.domain.{BlockHeader, Blockchain}
 import io.iohk.ethereum.network.Peer
+import io.iohk.ethereum.utils.Logger
 
-class FastSyncBranchResolver extends Actor {
+trait FastSyncBranchResolver {
 
-  override def receive: Receive = ???
+  import FastSyncBranchResolver._
+
+  protected def blockchain: Blockchain
+
+  // TODO [ETCM-676] move to [[Blockchain]] and make sure it's atomic
+  def discardBlocksAfter(lastValidBlock: BigInt): Unit =
+    discardBlocks(lastValidBlock, blockchain.getBestBlockNumber())
+
+  // TODO [ETCM-676] move to [[Blockchain]] and make sure it's atomic
+  private def discardBlocks(fromBlock: BigInt, toBlock: BigInt): Unit = {
+    val blocksToBeRemoved = childOf(fromBlock).to(toBlock).reverse.toList
+    blocksToBeRemoved.foreach { toBeRemoved =>
+      blockchain
+        .getBlockHeaderByNumber(toBeRemoved)
+        .foreach(header => blockchain.removeBlock(header.hash, withState = false))
+    }
+  }
 
 }
 
 object FastSyncBranchResolver {
-  sealed trait BranchResolverRequest
-  case object StartBranchResolver extends BranchResolverRequest
 
-  sealed trait BranchResolverResponse
-  case class BranchResolvedSuccessful(firstCommonBlockNumber: BigInt, masterPeer: Peer) extends BranchResolverResponse
+  /**
+    * Stores the current search state for binary search.
+    * Meaning we know the first common block lies between minBlockNumber and maxBlockNumber.
+    */
+  final case class SearchState(minBlockNumber: BigInt, maxBlockNumber: BigInt, masterPeer: Peer)
+
+  def parentOf(blockHeaderNumber: BigInt): BigInt = blockHeaderNumber - 1
+  def childOf(blockHeaderNumber: BigInt): BigInt = blockHeaderNumber + 1
+}
+
+/**
+  * Attempt to find last common block within recent blocks by looking for a parent/child
+  * relationship between our block headers and remote peer's block headers.
+  */
+class RecentBlocksSearch(blockchain: Blockchain) {
+
+  /**
+    * Find the highest common block by trying to find a block so that our block n is the parent of remote candidate block n + 1
+    */
+  def getHighestCommonBlock(
+      candidateHeaders: Seq[BlockHeader],
+      bestBlockNumber: BigInt
+  ): Option[BigInt] = {
+    def isParent(potentialParent: BigInt, childCandidate: BlockHeader): Boolean =
+      blockchain.getBlockHeaderByNumber(potentialParent).exists { _.isParentOf(childCandidate) }
+    NonEmptyList.fromList(candidateHeaders.reverse.toList).flatMap { remoteHeaders =>
+      val blocksToBeCompared = bestBlockNumber.until(bestBlockNumber - remoteHeaders.size).by(-1).toList
+      remoteHeaders.toList
+        .zip(blocksToBeCompared)
+        .collectFirst {
+          case (childCandidate, parent) if isParent(parent, childCandidate) => parent
+        }
+    }
+  }
+
+}
+
+object BinarySearchSupport extends Logger {
+  import FastSyncBranchResolver._
+
+  sealed trait BinarySearchResult
+  final case class BinarySearchCompleted(highestCommonBlockNumber: BigInt) extends BinarySearchResult
+  final case class ContinueBinarySearch(searchState: SearchState) extends BinarySearchResult
+  case object NoCommonBlock extends BinarySearchResult
+
+  /**
+    * Returns the block number in the middle between min and max.
+    * If there is no middle, it will return the lower value.
+    *
+    * E.g. calling this method with min = 3 and max = 6 will return 4
+    */
+  def middleBlockNumber(min: BigInt, max: BigInt): BigInt = (min + max) / 2
+
+  def blockHeaderNumberToRequest(min: BigInt, max: BigInt): BigInt =
+    childOf(middleBlockNumber(min, max))
+
+  def validateBlockHeaders(
+      parentBlockHeader: BlockHeader,
+      childBlockHeader: BlockHeader,
+      searchState: SearchState
+  ): BinarySearchResult = {
+    val childNum = childBlockHeader.number
+    val parentNum = parentBlockHeader.number
+    val min = searchState.minBlockNumber
+    val max = searchState.maxBlockNumber
+
+    log.debug(
+      "Validating block headers (binary search) for parentBlockHeader {}, remote childBlockHeader {} and search state {}",
+      parentBlockHeader.number,
+      childBlockHeader.number,
+      searchState
+    )
+
+    if (parentBlockHeader.isParentOf(childBlockHeader)) { // chains are still aligned but there might be an even better block
+      if (parentNum == max) BinarySearchCompleted(parentNum)
+      else if (parentNum == min && childNum == max) ContinueBinarySearch(searchState.copy(minBlockNumber = childNum))
+      else ContinueBinarySearch(searchState.copy(minBlockNumber = parentNum))
+    } else { // no parent/child -> chains have diverged before parent block
+      if (min == 1 && max <= 2) NoCommonBlock
+      else if (min == max) BinarySearchCompleted(parentOf(parentNum))
+      else ContinueBinarySearch(searchState.copy(maxBlockNumber = parentOf(parentNum).max(1)))
+    }
+  }
+
 }