[ETCM-103] Refactor handling of checking for stale block

Author: KonradStaniec

src/main/resources/application.conf

@@ -403,6 +403,16 @@ mantis {
     # Max number of mpt nodes held in memory in state sync, before saving them into database
     # 100k is around 60mb (each key-value pair has around 600bytes)
     state-sync-persistBatch-size = 100000
+
+    # If new pivot block received from network will be less than fast sync current pivot block, the re-try to chose new
+    # pivot will be scheduler after this time. Avarage block time in etc/eth is around 15s so after this time, most of
+    # network peers should have new best block
+    pivot-block-reSchedule-interval =  15.seconds
+
+    # If for most network peers, the following condition will be true:
+    # (peer.bestKnownBlock - pivot-block-offset) - node.curentPivotBlock > max-pivot-age
+    # it fast sync pivot block has become stale and it needs update
+    max-pivot-block-age = 96
   }
 
   pruning {

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

@@ -4,6 +4,7 @@ import java.time.Instant
 
 import akka.actor._
 import akka.util.ByteString
+import cats.data.NonEmptyList
 import io.iohk.ethereum.blockchain.sync.FastSyncReceiptsValidator.ReceiptsValidationResult
 import io.iohk.ethereum.blockchain.sync.PeerRequestHandler.ResponseReceived
 import io.iohk.ethereum.blockchain.sync.SyncBlocksValidator.BlockBodyValidationResult
@@ -69,29 +70,9 @@ class FastSync(
   }
 
   def startWithState(syncState: SyncState): Unit = {
-    if (syncState.updatingPivotBlock) {
-      log.info(s"FastSync interrupted during pivot block update, choosing new pivot block")
-      val syncingHandler = new SyncingHandler(syncState)
-      val pivotBlockSelector = context.actorOf(
-        PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self),
-        "pivot-block-selector"
-      )
-      pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
-      context become syncingHandler.waitingForPivotBlockUpdate(ImportedLastBlock)
-    } else {
-      log.info(
-        s"Starting block synchronization (fast mode), pivot block ${syncState.pivotBlock.number}, " +
-          s"block to download to ${syncState.safeDownloadTarget}"
-      )
-      val syncingHandler = new SyncingHandler(syncState)
-      context become syncingHandler.receive
-      if (syncState.isBlockchainWorkFinished && !syncState.stateSyncFinished) {
-        log.info(s"Blockchain sync was completed, starting state sync to block ${syncState.pivotBlock.idTag}")
-        // chain has already been downloaded we can start state sync
-        syncingHandler.startStateSync(syncState.pivotBlock)
-      }
-      syncingHandler.processSyncing()
-    }
+    log.info(s"Starting with existing state and asking for new pivot block")
+    val syncingHandler = new SyncingHandler(syncState)
+    syncingHandler.askForPivotBlockUpdate(NodeRestart)
   }
 
   def startFromScratch(): Unit = {
@@ -115,14 +96,17 @@ class FastSync(
           pivotBlockHeader,
           safeDownloadTarget = pivotBlockHeader.number + syncConfig.fastSyncBlockValidationX
         )
-      startWithState(initialSyncState)
+      val syncingHandler = new SyncingHandler(initialSyncState)
+      context.become(syncingHandler.receive)
+      syncingHandler.processSyncing()
     }
   }
 
   // scalastyle:off number.of.methods
   private class SyncingHandler(initialSyncState: SyncState) {
 
     private val BlockHeadersHandlerName = "block-headers-request-handler"
+    //not part of syncstate as we do not want to persist is.
     private var stateSyncRestartRequested = false
 
     private var requestedHeaders: Map[Peer, BigInt] = Map.empty
@@ -158,13 +142,11 @@ class FastSync(
     private val heartBeat =
       scheduler.scheduleWithFixedDelay(syncRetryInterval, syncRetryInterval * 2, self, ProcessSyncing)
 
-    def startStateSync(targetBlockHeader: BlockHeader): Unit = {
-      syncStateScheduler ! StartSyncingTo(targetBlockHeader.stateRoot, targetBlockHeader.number)
-    }
-
     def receive: Receive = handleCommonMessages orElse {
       case UpdatePivotBlock(state) => updatePivotBlock(state)
-      case WaitingForNewTargetBlock => updatePivotBlock(ImportedLastBlock)
+      case WaitingForNewTargetBlock =>
+        log.info("State sync stopped until receiving new pivot block")
+        updatePivotBlock(ImportedLastBlock)
       case ProcessSyncing => processSyncing()
       case PrintStatus => printStatus()
       case PersistSyncState => persistSyncState()
@@ -206,49 +188,65 @@ class FastSync(
         handleRequestFailure(assignedHandlers(ref), ref, "Unexpected error")
     }
 
-    def waitingForPivotBlockUpdate(processState: FinalBlockProcessingResult): Receive = handleCommonMessages orElse {
+    def askForPivotBlockUpdate(updateReason: PivotBlockUpdateReason): Unit = {
+      syncState = syncState.copy(updatingPivotBlock = true)
+      log.info("Asking for new pivot block")
+      val pivotBlockSelector = {
+        context.actorOf(
+          PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self)
+        )
+      }
+      pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
+      context become waitingForPivotBlockUpdate(updateReason)
+    }
+
+    def reScheduleAskForNewPivot(updateReason: PivotBlockUpdateReason): Unit = {
+      syncState = syncState.copy(pivotBlockUpdateFailures = syncState.pivotBlockUpdateFailures + 1)
+      scheduler.scheduleOnce(syncConfig.pivotBlockReScheduleInterval, self, UpdatePivotBlock(updateReason))
+    }
+
+    def waitingForPivotBlockUpdate(updateReason: PivotBlockUpdateReason): Receive = handleCommonMessages orElse {
       case PivotBlockSelector.Result(pivotBlockHeader) =>
         log.info(s"New pivot block with number ${pivotBlockHeader.number} received")
         if (pivotBlockHeader.number >= syncState.pivotBlock.number) {
-          updatePivotSyncState(processState, pivotBlockHeader)
-          syncState = syncState.copy(updatingPivotBlock = false)
-          context become this.receive
-          processSyncing()
+          if (pivotBlockHeader.number == syncState.pivotBlock.number && updateReason.nodeRestart) {
+            // it can happen after quick node restart than pivot block has not changed in the network. To keep whole
+            // fast sync machinery running as expected we need to make sure that we will receive better pivot than current
+            log.info("Received stale pivot after restart, asking for new pivot")
+            reScheduleAskForNewPivot(updateReason)
+          } else {
+            updatePivotSyncState(updateReason, pivotBlockHeader)
+            syncState = syncState.copy(updatingPivotBlock = false)
+            context become this.receive
+            processSyncing()
+          }
         } else {
-          syncState = syncState.copy(pivotBlockUpdateFailures = syncState.pivotBlockUpdateFailures + 1)
-          scheduler.scheduleOnce(syncRetryInterval, self, UpdatePivotBlock(processState))
+          log.info("Received target block is older than old one, re-scheduling asking for new one")
+          reScheduleAskForNewPivot(updateReason)
         }
 
       case PersistSyncState => persistSyncState()
 
       case UpdatePivotBlock(state) => updatePivotBlock(state)
     }
 
-    private def updatePivotBlock(state: FinalBlockProcessingResult): Unit = {
+    private def updatePivotBlock(state: PivotBlockUpdateReason): Unit = {
       if (syncState.pivotBlockUpdateFailures <= syncConfig.maximumTargetUpdateFailures) {
         if (assignedHandlers.nonEmpty || syncState.blockChainWorkQueued) {
           log.info(s"Still waiting for some responses, rescheduling pivot block update")
           scheduler.scheduleOnce(1.second, self, UpdatePivotBlock(state))
           processSyncing()
         } else {
-          syncState = syncState.copy(updatingPivotBlock = true)
-          log.info("Asking for new pivot block")
-          val pivotBlockSelector = {
-            context.actorOf(
-              PivotBlockSelector.props(etcPeerManager, peerEventBus, syncConfig, scheduler, context.self)
-            )
-          }
-          pivotBlockSelector ! PivotBlockSelector.SelectPivotBlock
-          context become waitingForPivotBlockUpdate(state)
+          askForPivotBlockUpdate(state)
         }
       } else {
         log.warning(s"Sync failure! Number of pivot block update failures reached maximum.")
         sys.exit(1)
       }
     }
 
-    private def updatePivotSyncState(state: FinalBlockProcessingResult, pivotBlockHeader: BlockHeader): Unit =
-      state match {
+    private def updatePivotSyncState(updateReason: PivotBlockUpdateReason, pivotBlockHeader: BlockHeader): Unit =
+      updateReason match {
         case ImportedLastBlock =>
           if (pivotBlockHeader.number - syncState.pivotBlock.number <= syncConfig.maxTargetDifference) {
             log.info(s"Current pivot block is fresh enough, starting state download")
@@ -277,6 +275,17 @@ class FastSync(
           )
           syncState =
             syncState.updatePivotBlock(pivotBlockHeader, syncConfig.fastSyncBlockValidationX, updateFailures = true)
+
+        case NodeRestart =>
+          // in case of node restart we are sure that new pivotBlockHeader > current pivotBlockHeader
+          syncState = syncState.updatePivotBlock(
+            pivotBlockHeader,
+            syncConfig.fastSyncBlockValidationX,
+            updateFailures = false
+          )
+          log.info(
+            s"Changing pivot block to ${pivotBlockHeader.number}, new safe target is ${syncState.safeDownloadTarget}"
+          )
       }
 
     private def removeRequestHandler(handler: ActorRef): Unit = {
@@ -527,31 +536,48 @@ class FastSync(
       }
     }
 
+    def hasBestBlockFreshEnoughToUpdatePivotBlock(info: PeerInfo, state: SyncState, syncConfig: SyncConfig): Boolean = {
+      (info.maxBlockNumber - syncConfig.pivotBlockOffset) - state.pivotBlock.number > syncConfig.maxPivotBlockAge
+    }
+
     private def getPeerWithTooFreshNewBlock(
-        peers: Map[Peer, PeerInfo],
+        peers: NonEmptyList[(Peer, PeerInfo)],
         state: SyncState,
         syncConfig: SyncConfig
-    ): List[Peer] = {
+    ): List[(Peer, BigInt)] = {
       peers.collect {
-        case (peer, info)
-            if (info.maxBlockNumber - syncConfig.pivotBlockOffset) - state.pivotBlock.number > FastSync.maxTargetBlockAge =>
-          peer
-      }.toList
+        case (peer, info) if hasBestBlockFreshEnoughToUpdatePivotBlock(info, state, syncConfig) =>
+          (peer, info.maxBlockNumber)
+      }
     }
 
-    private def shouldUpdateStateTargetBlock(): Boolean = {
-      val availablePeers = peersToDownloadFrom
-      if (availablePeers.isEmpty) {
+    def noBlockchainWorkRemaining: Boolean =
+      syncState.isBlockchainWorkFinished && assignedHandlers.isEmpty
+
+    def notInTheMiddleOfUpdate: Boolean =
+      !(syncState.updatingPivotBlock || stateSyncRestartRequested)
+
+    def pivotBlockIsStale(): Boolean = {
+      val currentPeers = peersToDownloadFrom.toList
+      if (currentPeers.isEmpty) {
         false
       } else {
-        if (availablePeers.size < syncConfig.minPeersToChoosePivotBlock) {
-          getPeerWithTooFreshNewBlock(availablePeers, syncState, syncConfig).size == availablePeers.size
+        val peerWithBestBlockInNetwork = currentPeers.maxBy(peerWithNum => peerWithNum._2.maxBlockNumber)
+
+        val peersWithTooFreshPossiblePivotBlock =
+          getPeerWithTooFreshNewBlock(NonEmptyList.fromListUnsafe(currentPeers), syncState, syncConfig)
+
+        if (peersWithTooFreshPossiblePivotBlock.isEmpty) {
+          log.info(s"There are not peers with to fresh possible pivot block, " +
+            s"best peer has block with number: ${peerWithBestBlockInNetwork._2.maxBlockNumber}")
+          false
         } else {
-          getPeerWithTooFreshNewBlock(
-            availablePeers,
-            syncState,
-            syncConfig
-          ).size >= syncConfig.minPeersToChoosePivotBlock
+          val pivotBlockIsStale = peersWithTooFreshPossiblePivotBlock.size >= minPeersToChoosePivotBlock
+
+          log.info(s"There are ${peersWithTooFreshPossiblePivotBlock.size} peers with possible new pivot block, " +
+            s"best known pivot in current peer list has number ${peerWithBestBlockInNetwork._2.maxBlockNumber}")
+
+          pivotBlockIsStale
         }
       }
     }
@@ -562,23 +588,9 @@ class FastSync(
       } else {
         if (blockchainDataToDownload) {
           processDownloads()
-        } else if (syncState.isBlockchainWorkFinished && assignedHandlers.isEmpty && !syncState.stateSyncFinished) {
-          if (peersToDownloadFrom.nonEmpty) {
-            val bestBLock = peersToDownloadFrom.map { case (p, info) => info.maxBlockNumber }.max
-            log.info(
-              s"BestKnownBlock in network is ${bestBLock}. Target block is ${syncState.pivotBlock.number}." +
-                s"Difference is ${bestBLock - syncState.pivotBlock.number}"
-            )
-          }
-
-          if (shouldUpdateStateTargetBlock() && !stateSyncRestartRequested) {
-            val bestBLock = peersToDownloadFrom.map { case (p, info) => info.maxBlockNumber }.max
-            log.info(
-              s"Updating state sync target block. " +
-                s"BestKnownBlock in network is ${bestBLock}. Target block is ${syncState.pivotBlock.number}." +
-                s"Difference is ${bestBLock - syncState.pivotBlock.number}"
-            )
-
+        } else if (noBlockchainWorkRemaining && !syncState.stateSyncFinished && notInTheMiddleOfUpdate) {
+          if (pivotBlockIsStale()){
+            log.info("Restarting state sync to new pivot block")
             syncStateScheduler ! RestartRequested
             stateSyncRestartRequested = true
           }
@@ -763,7 +775,7 @@ object FastSync {
       )
     )
 
-  private case class UpdatePivotBlock(state: FinalBlockProcessingResult)
+  private case class UpdatePivotBlock(state: PivotBlockUpdateReason)
   private case object ProcessSyncing
 
   private[sync] case object PersistSyncState
@@ -841,8 +853,15 @@ object FastSync {
 
   case object ImportedPivotBlock extends HeaderProcessingResult
 
-  sealed abstract class FinalBlockProcessingResult
+  sealed abstract class PivotBlockUpdateReason {
+    def nodeRestart: Boolean = this match {
+      case ImportedLastBlock => false
+      case LastBlockValidationFailed => false
+      case NodeRestart => true
+    }
+  }
 
-  case object ImportedLastBlock extends FinalBlockProcessingResult
-  case object LastBlockValidationFailed extends FinalBlockProcessingResult
+  case object ImportedLastBlock extends PivotBlockUpdateReason
+  case object LastBlockValidationFailed extends PivotBlockUpdateReason
+  case object NodeRestart extends PivotBlockUpdateReason
 }

src/main/scala/io/iohk/ethereum/utils/Config.scala

@@ -126,6 +126,8 @@ object Config {
       maximumTargetUpdateFailures: Int,
       stateSyncBloomFilterSize: Int,
       stateSyncPersistBatchSize: Int,
+      pivotBlockReScheduleInterval: FiniteDuration,
+      maxPivotBlockAge: Int
   )
 
   object SyncConfig {
@@ -166,7 +168,9 @@ object Config {
         maxTargetDifference = syncConfig.getInt("max-target-difference"),
         maximumTargetUpdateFailures = syncConfig.getInt("maximum-target-update-failures"),
         stateSyncBloomFilterSize = syncConfig.getInt("state-sync-bloomFilter-size"),
-        stateSyncPersistBatchSize = syncConfig.getInt("state-sync-persistBatch-size")
+        stateSyncPersistBatchSize = syncConfig.getInt("state-sync-persistBatch-size"),
+        pivotBlockReScheduleInterval = syncConfig.getDuration("pivot-block-reSchedule-interval").toMillis.millis,
+        maxPivotBlockAge =  syncConfig.getInt("max-pivot-block-age"),
       )
     }
   }

src/test/resources/application.conf

@@ -121,6 +121,10 @@ mantis {
     state-sync-bloomFilter-size = 20000
 
     state-sync-persistBatch-size = 10000
+
+    pivot-block-reSchedule-interval = 15.seconds
+
+    max-pivot-block-age = 96
   }
 
   keyStore {

src/test/scala/io/iohk/ethereum/blockchain/sync/StateSyncUtils.scala

@@ -29,12 +29,12 @@ object StateSyncUtils extends EphemBlockchainTestSetup {
       }
     }
 
-    def buildWorld(accountData: Seq[MptNodeData]): ByteString = {
+    def buildWorld(accountData: Seq[MptNodeData], existingTree: Option[ByteString] = None): ByteString = {
       val init: InMemoryWorldStateProxy = bl
         .getWorldStateProxy(
           blockNumber = 1,
           accountStartNonce = blockchainConfig.accountStartNonce,
-          stateRootHash = None,
+          stateRootHash = existingTree,
           noEmptyAccounts = true,
           ethCompatibleStorage = blockchainConfig.ethCompatibleStorage
         )