[ETCM-275] Improve synccontroller tests

Author: KonradStaniec

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

@@ -90,36 +90,37 @@ class SyncStateSchedulerActor(
       self ! BloomFilterResult(value)
   }
 
-  def waitingForBloomFilterToLoad(lastReceivedCommand: Option[(SyncStateSchedulerActorCommand, ActorRef)]): Receive = {
-    case BloomFilterResult(result) =>
-      log.debug(
-        s"Loaded ${result.writtenElements} already known elements from storage to bloom filter the error while loading " +
-          s"was ${result.error}"
-      )
-      lastReceivedCommand match {
-        case Some((startSignal: StartSyncingTo, sender)) =>
-          val initStats = ProcessingStatistics().addSaved(result.writtenElements)
-          val initState = startSyncing(startSignal.stateRoot, startSignal.blockNumber)
-          context become (syncing(
-            initState,
-            DownloaderState(),
-            initStats,
-            startSignal.blockNumber,
-            sender,
-            Queue(),
-            processing = false,
-            None
-          ))
-        case Some((restartSignal: RestartRequested.type, sender)) =>
-          sender ! WaitingForNewTargetBlock
-          context.become(idle(ProcessingStatistics().addSaved(result.writtenElements)))
-        case _ =>
-          context.become(idle(ProcessingStatistics().addSaved(result.writtenElements)))
-      }
+  def waitingForBloomFilterToLoad(lastReceivedCommand: Option[(SyncStateSchedulerActorCommand, ActorRef)]): Receive =
+    handleCommonMessages orElse {
+      case BloomFilterResult(result) =>
+        log.debug(
+          s"Loaded ${result.writtenElements} already known elements from storage to bloom filter the error while loading " +
+            s"was ${result.error}"
+        )
+        lastReceivedCommand match {
+          case Some((startSignal: StartSyncingTo, sender)) =>
+            val initStats = ProcessingStatistics().addSaved(result.writtenElements)
+            val initState = startSyncing(startSignal.stateRoot, startSignal.blockNumber)
+            context become (syncing(
+              initState,
+              DownloaderState(),
+              initStats,
+              startSignal.blockNumber,
+              sender,
+              Queue(),
+              processing = false,
+              None
+            ))
+          case Some((restartSignal: RestartRequested.type, sender)) =>
+            sender ! WaitingForNewTargetBlock
+            context.become(idle(ProcessingStatistics().addSaved(result.writtenElements)))
+          case _ =>
+            context.become(idle(ProcessingStatistics().addSaved(result.writtenElements)))
+        }
 
-    case command: SyncStateSchedulerActorCommand =>
-      context.become(waitingForBloomFilterToLoad(Some((command, sender()))))
-  }
+      case command: SyncStateSchedulerActorCommand =>
+        context.become(waitingForBloomFilterToLoad(Some((command, sender()))))
+    }
 
   private def startSyncing(root: ByteString, bn: BigInt): SchedulerState = {
     timers.startTimerAtFixedRate(PrintInfoKey, PrintInfo, 30.seconds)
@@ -130,7 +131,7 @@ class SyncStateSchedulerActor(
     initState
   }
 
-  def idle(processingStatistics: ProcessingStatistics): Receive = {
+  def idle(processingStatistics: ProcessingStatistics): Receive = handleCommonMessages orElse {
     case StartSyncingTo(root, bn) =>
       val state1 = startSyncing(root, bn)
       context become (syncing(

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

@@ -45,7 +45,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
   }
 
   after {
-    Await.result(system.terminate(), 1.seconds)
+    Await.result(system.terminate(), 10.seconds)
   }
 
   "SyncController" should "download pivot block and request block headers" in new TestSetup() {
@@ -249,7 +249,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
       storagesInstance.storages.fastSyncStateStorage.getSyncState().get.pivotBlock shouldBe defaultPivotBlockHeader
     }
 
-    setupAutoPilot(etcPeerManager, newHanshaked, newPivot, BlockchainData(newBlocks))
+    updateAutoPilot(etcPeerManager.ref, newHanshaked, newPivot, BlockchainData(newBlocks))
     val watcher = TestProbe()
     watcher.watch(syncController)
 
@@ -287,7 +287,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
       storagesInstance.storages.fastSyncStateStorage.getSyncState().get.pivotBlockUpdateFailures shouldBe 1
     }
 
-    setupAutoPilot(etcPeerManager, freshHandshakedPeers, freshHeader, BlockchainData(newBlocks), onlyPivot = true)
+    updateAutoPilot(etcPeerManager.ref, freshHandshakedPeers, freshHeader, BlockchainData(newBlocks), onlyPivot = true)
 
     eventually(timeout = eventuallyTimeOut) {
       storagesInstance.storages.fastSyncStateStorage.getSyncState().get.pivotBlock shouldBe defaultPivotBlockHeader
@@ -320,7 +320,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
     val freshHandshakedPeers1 = HandshakedPeers(Map(peer1 -> freshPeerInfo1a))
 
     // set up new received header previously received header will need update
-    setupAutoPilot(etcPeerManager, freshHandshakedPeers1, freshHeader1, BlockchainData(newBlocks))
+    updateAutoPilot(etcPeerManager.ref, freshHandshakedPeers1, freshHeader1, BlockchainData(newBlocks))
 
     eventually(timeout = longeventuallyTimeOut) {
       storagesInstance.storages.fastSyncStateStorage
@@ -397,7 +397,13 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
     val newHandshakedPeers = HandshakedPeers(Map(peer1 -> peerWithBetterBlock))
     val newPivot = defaultPivotBlockHeader.copy(number = defaultPivotBlockHeader.number + syncConfig.maxPivotBlockAge)
 
-    setupAutoPilot(etcPeerManager, newHandshakedPeers, newPivot, BlockchainData(newBlocks), failedNodeRequest = true)
+    updateAutoPilot(
+      etcPeerManager.ref,
+      newHandshakedPeers,
+      newPivot,
+      BlockchainData(newBlocks),
+      failedNodeRequest = true
+    )
 
     // sync to new pivot
     eventually(timeout = eventuallyTimeOut) {
@@ -406,7 +412,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
     }
 
     // enable peer to respond with mpt nodes
-    setupAutoPilot(etcPeerManager, newHandshakedPeers, newPivot, BlockchainData(newBlocks))
+    updateAutoPilot(etcPeerManager.ref, newHandshakedPeers, newPivot, BlockchainData(newBlocks))
 
     val watcher = TestProbe()
     watcher.watch(syncController)
@@ -426,6 +432,8 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
   ) extends EphemBlockchainTestSetup
       with TestSyncPeers
       with TestSyncConfig {
+
+    @volatile
     var stateDownloadStarted = false
 
     val eventuallyTimeOut: Timeout = Timeout(Span(10, Seconds))
@@ -482,7 +490,7 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
           ommersPool.ref,
           etcPeerManager.ref,
           syncConfig,
-          externalSchedulerOpt = None
+          externalSchedulerOpt = Some(system.scheduler)
         )
       )
     )
@@ -511,6 +519,78 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
         }
       }
     }
+    // scalastyle:off method.length
+    case class SyncStateAutoPilot(
+        handshakedPeers: HandshakedPeers,
+        pivotHeader: BlockHeader,
+        blockchainData: BlockchainData,
+        failedReceiptsTries: Int,
+        failedBodiesTries: Int,
+        onlyPivot: Boolean,
+        failedNodeRequest: Boolean
+    ) extends AutoPilot {
+      override def run(sender: ActorRef, msg: Any): AutoPilot = {
+        msg match {
+          case EtcPeerManagerActor.GetHandshakedPeers =>
+            sender ! handshakedPeers
+            this
+
+          case SendMessage(msg: GetBlockHeadersEnc, peer) =>
+            val underlyingMessage = msg.underlyingMsg
+            if (underlyingMessage.maxHeaders == 1) {
+              // pivot block
+              sender ! MessageFromPeer(BlockHeaders(Seq(pivotHeader)), peer)
+              this
+            } else {
+              if (!onlyPivot) {
+                val start = msg.underlyingMsg.block.left.get
+                val stop = start + msg.underlyingMsg.maxHeaders
+                val headers = (start until stop).flatMap(i => blockchainData.headers.get(i))
+                sender ! MessageFromPeer(BlockHeaders(headers), peer)
+                this
+              } else {
+                this
+              }
+            }
+
+          case SendMessage(msg: GetReceiptsEnc, peer) if !onlyPivot =>
+            val underlyingMessage = msg.underlyingMsg
+            if (failedReceiptsTries > 0) {
+              sender ! MessageFromPeer(Receipts(Seq()), peer)
+              this.copy(failedReceiptsTries = failedReceiptsTries - 1)
+            } else {
+              val rec = msg.underlyingMsg.blockHashes.flatMap(h => blockchainData.receipts.get(h))
+              sender ! MessageFromPeer(Receipts(rec), peer)
+              this
+            }
+
+          case SendMessage(msg: GetBlockBodiesEnc, peer) if !onlyPivot =>
+            val underlyingMessage = msg.underlyingMsg
+            if (failedBodiesTries > 0) {
+              sender ! MessageFromPeer(BlockBodies(Seq()), peer)
+              this.copy(failedBodiesTries = failedBodiesTries - 1)
+            } else {
+              val bod = msg.underlyingMsg.hashes.flatMap(h => blockchainData.bodies.get(h))
+              sender ! MessageFromPeer(BlockBodies(bod), peer)
+              this
+            }
+
+          case SendMessage(msg: GetNodeDataEnc, peer) if !onlyPivot =>
+            stateDownloadStarted = true
+            val underlyingMessage = msg.underlyingMsg
+            if (!failedNodeRequest) {
+              sender ! MessageFromPeer(NodeData(Seq(defaultStateMptLeafWithAccount)), peer)
+              this
+            } else {
+              this
+            }
+
+          case AutoPilotUpdateData(peers, pivot, data, failedReceipts, failedBodies, onlyPivot, failedNode) =>
+            sender ! DataUpdated
+            this.copy(peers, pivot, data, failedReceipts, failedBodies, onlyPivot, failedNode)
+        }
+      }
+    }
 
     // scalastyle:off method.length parameter.number
     def setupAutoPilot(
@@ -523,59 +603,54 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
         onlyPivot: Boolean = false,
         failedNodeRequest: Boolean = false
     ): Unit = {
-      testProbe.setAutoPilot(new AutoPilot {
-        var failedReceipts = 0
-        var failedBodies = 0
-        def run(sender: ActorRef, msg: Any): AutoPilot = {
-          msg match {
-            case EtcPeerManagerActor.GetHandshakedPeers =>
-              sender ! handshakedPeers
-
-            case SendMessage(msg: GetBlockHeadersEnc, peer) =>
-              val underlyingMessage = msg.underlyingMsg
-              if (underlyingMessage.maxHeaders == 1) {
-                // pivot block
-                sender ! MessageFromPeer(BlockHeaders(Seq(pivotHeader)), peer)
-              } else {
-                if (!onlyPivot) {
-                  val start = msg.underlyingMsg.block.left.get
-                  val stop = start + msg.underlyingMsg.maxHeaders
-                  val headers = (start until stop).flatMap(i => blockchainData.headers.get(i))
-                  sender ! MessageFromPeer(BlockHeaders(headers), peer)
-                }
-              }
-
-            case SendMessage(msg: GetReceiptsEnc, peer) if !onlyPivot =>
-              val underlyingMessage = msg.underlyingMsg
-              if (failedReceipts < failedReceiptsTries) {
-                sender ! MessageFromPeer(Receipts(Seq()), peer)
-                failedReceipts = failedReceipts + 1
-              } else {
-                val rec = msg.underlyingMsg.blockHashes.flatMap(h => blockchainData.receipts.get(h))
-                sender ! MessageFromPeer(Receipts(rec), peer)
-              }
-
-            case SendMessage(msg: GetBlockBodiesEnc, peer) if !onlyPivot =>
-              val underlyingMessage = msg.underlyingMsg
-              if (failedBodies < failedBodiesTries) {
-                sender ! MessageFromPeer(BlockBodies(Seq()), peer)
-                failedBodies = failedBodies + 1
-              } else {
-                val bod = msg.underlyingMsg.hashes.flatMap(h => blockchainData.bodies.get(h))
-                sender ! MessageFromPeer(BlockBodies(bod), peer)
-              }
+      testProbe.setAutoPilot(
+        SyncStateAutoPilot(
+          handshakedPeers,
+          pivotHeader,
+          blockchainData,
+          failedReceiptsTries,
+          failedBodiesTries,
+          onlyPivot,
+          failedNodeRequest
+        )
+      )
+    }
 
-            case SendMessage(msg: GetNodeDataEnc, peer) if !onlyPivot =>
-              stateDownloadStarted = true
-              val underlyingMessage = msg.underlyingMsg
-              if (!failedNodeRequest) {
-                sender ! MessageFromPeer(NodeData(Seq(defaultStateMptLeafWithAccount)), peer)
-              }
-          }
+    case class AutoPilotUpdateData(
+        handshakedPeers: HandshakedPeers,
+        pivotHeader: BlockHeader,
+        blockchainData: BlockchainData,
+        failedReceiptsTries: Int = 0,
+        failedBodiesTries: Int = 0,
+        onlyPivot: Boolean = false,
+        failedNodeRequest: Boolean = false
+    )
+    case object DataUpdated
 
-          this
-        }
-      })
+    def updateAutoPilot(
+        probeWIthPilot: ActorRef,
+        handshakedPeers: HandshakedPeers,
+        pivotHeader: BlockHeader,
+        blockchainData: BlockchainData,
+        failedReceiptsTries: Int = 0,
+        failedBodiesTries: Int = 0,
+        onlyPivot: Boolean = false,
+        failedNodeRequest: Boolean = false
+    ): Unit = {
+      val sender = TestProbe()
+      probeWIthPilot.tell(
+        AutoPilotUpdateData(
+          handshakedPeers,
+          pivotHeader,
+          blockchainData,
+          failedReceiptsTries,
+          failedBodiesTries,
+          onlyPivot,
+          failedNodeRequest
+        ),
+        sender.ref
+      )
+      sender.expectMsg(DataUpdated)
     }
 
     val defaultExpectedPivotBlock = 399500
@@ -645,26 +720,4 @@ class SyncControllerSpec extends AnyFlatSpec with Matchers with BeforeAndAfter w
       Thread.sleep(300)
     }
   }
-
-  class TestWithRegularSyncOnSetup extends TestSetup() {
-    val syncControllerWithRegularSync = TestActorRef(
-      Props(
-        new SyncController(
-          storagesInstance.storages.appStateStorage,
-          blockchain,
-          blockchainConfig,
-          storagesInstance.storages.fastSyncStateStorage,
-          ledger,
-          new Mocks.MockValidatorsAlwaysSucceed,
-          peerMessageBus.ref,
-          pendingTransactionsManager.ref,
-          checkpointBlockGenerator,
-          ommersPool.ref,
-          etcPeerManager.ref,
-          syncConfig,
-          externalSchedulerOpt = None
-        )
-      )
-    )
-  }
 }