[ETCM-739] Refactor BlockFetcher

Change BlockFetcher to typed actor Split fetcher message handling among child actors Abstract the fetch trait Scalafmt Refactor blockFetcherState Fix it tests Fix scalastyle, comments Fix comments
input-output-hk · May 7, 2021 · 5cba7bf · 5cba7bf
1 parent d40732a
commit 5cba7bf
Show file tree

Hide file tree

Showing 15 changed files with 596 additions and 775 deletions.
diff --git a/nix/overlay.nix b/nix/overlay.nix
@@ -3,7 +3,7 @@ rev: final: prev: {
 
   mantis = final.callPackage ./mantis.nix {
     src = ../.;
-    depsSha256 = "sha256-0AeemKFcIU3eVGse8QQGauJeRsF7IgCLo5Yqu2FZsMs=";
+    depsSha256 = "sha256-US4L/xh2otnEfOa05bazb14bgYhQZpF4GfFY30sDkNY=";
   };
 
   mantis-hash = final.mantis.override {

diff --git a/project/Dependencies.scala b/project/Dependencies.scala
@@ -14,7 +14,9 @@ object Dependencies {
     Seq(
       "com.typesafe.akka" %% "akka-actor" % akkaVersion,
       "com.typesafe.akka" %% "akka-slf4j" % akkaVersion,
+      "com.typesafe.akka" %% "akka-actor-typed" % akkaVersion,
       "com.typesafe.akka" %% "akka-testkit" % akkaVersion,
+      "com.typesafe.akka" %% "akka-actor-testkit-typed" % akkaVersion,
       "com.typesafe.akka" %% "akka-stream" % akkaVersion,
       "com.miguno.akka" %% "akka-mock-scheduler" % "0.5.5" % "it,test"
     )

diff --git a/src/it/scala/io/iohk/ethereum/ledger/BlockImporterItSpec.scala b/src/it/scala/io/iohk/ethereum/ledger/BlockImporterItSpec.scala
@@ -267,7 +267,7 @@ class BlockImporterItSpec
 
       val msg = fetcherProbe
         .fishForMessage(Timeouts.longTimeout) {
-          case BlockFetcher.FetchStateNode(_) => true
+          case BlockFetcher.FetchStateNode(_, _) => true
           case _ => false
         }
         .asInstanceOf[BlockFetcher.FetchStateNode]

diff --git a/src/it/scala/io/iohk/ethereum/sync/RegularSyncItSpec.scala b/src/it/scala/io/iohk/ethereum/sync/RegularSyncItSpec.scala
@@ -28,24 +28,6 @@ class RegularSyncItSpec extends FreeSpecBase with Matchers with BeforeAndAfterAl
     testScheduler.awaitTermination(120.second)
   }
 
-  "a peer should reorganise when receives a checkpoint older than the current best from a peer" in customTestCaseResourceM(
-    FakePeer.start2FakePeersRes()
-  ) { case (peer1, peer2) =>
-    for {
-      _ <- peer1.importBlocksUntil(20)(IdentityUpdate)
-      _ <- peer2.importBlocksUntil(30)(IdentityUpdate)
-      _ <- peer1.startRegularSync()
-      _ <- peer2.startRegularSync()
-      _ <- peer1.addCheckpointedBlock(peer1.bl.getBestBlock().get)
-      _ <- peer1.waitForRegularSyncLoadLastBlock(21)
-      _ <- peer2.getCheckpointFromPeer(peer1.bl.getBestBlock().get, PeerId("Peer1"))
-      _ <- peer2.waitForRegularSyncLoadLastBlock(21)
-    } yield {
-      assert(peer1.bl.getBestBlock().get.hash == peer2.bl.getBestBlock().get.hash)
-      assert(peer1.bl.getLatestCheckpointBlockNumber() == peer2.bl.getLatestCheckpointBlockNumber())
-    }
-  }
-
   "peer 2 should sync to the top of peer1 blockchain" - {
     "given a previously imported blockchain" in customTestCaseResourceM(FakePeer.start2FakePeersRes()) {
       case (peer1, peer2) =>

diff --git a/src/it/scala/io/iohk/ethereum/sync/util/RegularSyncItSpecUtils.scala b/src/it/scala/io/iohk/ethereum/sync/util/RegularSyncItSpecUtils.scala
@@ -1,19 +1,13 @@
 package io.iohk.ethereum.sync.util
 
-import akka.actor.ActorRef
+import akka.actor.{ActorRef, typed}
 import akka.util.ByteString
 import cats.effect.Resource
 import io.iohk.ethereum.Mocks.MockValidatorsAlwaysSucceed
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcast.BlockToBroadcast
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor.BroadcastBlock
 import io.iohk.ethereum.blockchain.sync.regular.BlockImporter.Start
-import io.iohk.ethereum.blockchain.sync.regular.{
-  BlockBroadcast,
-  BlockBroadcasterActor,
-  BlockFetcher,
-  BlockImporter,
-  RegularSync
-}
+import io.iohk.ethereum.blockchain.sync.regular.{BlockBroadcast, BlockBroadcasterActor, BlockFetcher, BlockImporter, RegularSync}
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync.NewCheckpoint
 import io.iohk.ethereum.blockchain.sync.{PeersClient, SyncProtocol}
 import io.iohk.ethereum.checkpointing.CheckpointingTestHelpers
@@ -36,6 +30,8 @@ import io.iohk.ethereum.utils._
 import io.iohk.ethereum.vm.EvmConfig
 import monix.eval.Task
 import monix.execution.Scheduler
+import akka.actor.typed.scaladsl.adapter._
+import io.iohk.ethereum.blockchain.sync.regular.BlockFetcher.AdaptedMessageFromEventBus
 
 import scala.concurrent.duration._
 object RegularSyncItSpecUtils {
@@ -86,15 +82,15 @@ object RegularSyncItSpecUtils {
       "block-broadcaster"
     )
 
-    val fetcher: ActorRef =
-      system.actorOf(
-        BlockFetcher.props(peersClient, peerEventBus, regularSync, syncConfig, validators.blockValidator),
+    val fetcher: typed.ActorRef[BlockFetcher.FetchCommand] =
+      system.spawn(
+        BlockFetcher(peersClient, peerEventBus, regularSync, syncConfig, validators.blockValidator),
         "block-fetcher"
       )
 
     lazy val blockImporter = system.actorOf(
       BlockImporter.props(
-        fetcher,
+        fetcher.toClassic,
         ledger,
         bl,
         syncConfig,
@@ -181,7 +177,7 @@ object RegularSyncItSpecUtils {
 
     def getCheckpointFromPeer(checkpoint: Block, peerId: PeerId): Task[Unit] = Task {
       blockImporter ! Start
-      fetcher ! MessageFromPeer(NewBlock(checkpoint, checkpoint.header.difficulty), peerId)
+      fetcher ! AdaptedMessageFromEventBus(NewBlock(checkpoint, checkpoint.header.difficulty), peerId)
     }
 
     private def getMptForBlock(block: Block) = {