[ETCM-213] Extract bloom loading to sparate class. More tests.

Author: KonradStaniec

src/it/scala/io/iohk/ethereum/sync/FastSyncItSpec.scala

@@ -129,6 +129,21 @@ class FastSyncItSpec extends FlatSpecBase with Matchers with BeforeAndAfter {
     }
   }
 
+  it should "sync state to peer from partially synced state" in customTestCaseResourceM(
+    FakePeer.start2FakePeersRes()
+  ) { case (peer1, peer2) =>
+    for {
+      _ <- peer2.importBlocksUntil(2000)(updateStateAtBlock(1500))
+      _ <- peer2.importBlocksUntil(3000)(updateStateAtBlock(2500, 1000, 2000))
+      _ <- peer1.importBlocksUntil(2000)(updateStateAtBlock(1500))
+      _ <- peer1.startWithState()
+      _ <- peer1.connectToPeers(Set(peer2.node))
+      _ <- peer1.startFastSync().delayExecution(50.milliseconds)
+      _ <- peer1.waitForFastSyncFinish()
+    } yield {
+      assert(peer1.bl.getBestBlockNumber() == peer2.bl.getBestBlockNumber() - peer2.testSyncConfig.pivotBlockOffset)
+    }
+  }
 }
 
 object FastSyncItSpec {
@@ -145,8 +160,12 @@ object FastSyncItSpec {
 
   val IdentityUpdate: (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy = (_, world) => world
 
-  def updateWorldWithNAccounts(n: Int, world: InMemoryWorldStateProxy): InMemoryWorldStateProxy = {
-    val resultWorld = (0 until n).foldLeft(world) { (world, num) =>
+  def updateWorldWithAccounts(
+      startAccount: Int,
+      endAccount: Int,
+      world: InMemoryWorldStateProxy
+  ): InMemoryWorldStateProxy = {
+    val resultWorld = (startAccount until endAccount).foldLeft(world) { (world, num) =>
       val randomBalance = num
       val randomAddress = Address(num)
       val codeBytes = BigInt(num).toByteArray
@@ -160,10 +179,14 @@ object FastSyncItSpec {
     InMemoryWorldStateProxy.persistState(resultWorld)
   }
 
-  def updateStateAtBlock(blockWithUpdate: BigInt): (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy = {
+  def updateStateAtBlock(
+      blockWithUpdate: BigInt,
+      startAccount: Int = 0,
+      endAccount: Int = 1000
+  ): (BigInt, InMemoryWorldStateProxy) => InMemoryWorldStateProxy = {
     (blockNr: BigInt, world: InMemoryWorldStateProxy) =>
       if (blockNr == blockWithUpdate) {
-        updateWorldWithNAccounts(1000, world)
+        updateWorldWithAccounts(startAccount, endAccount, world)
       } else {
         IdentityUpdate(blockNr, world)
       }

src/it/scala/io/iohk/ethereum/sync/FastSyncItSpecUtils.scala

@@ -9,6 +9,7 @@ import akka.testkit.TestProbe
 import akka.util.{ByteString, Timeout}
 import cats.effect.Resource
 import io.iohk.ethereum.Mocks.MockValidatorsAlwaysSucceed
+import io.iohk.ethereum.blockchain.sync.FastSync.SyncState
 import io.iohk.ethereum.{Fixtures, Timeouts}
 import io.iohk.ethereum.blockchain.sync.{BlockBroadcast, BlockchainHostActor, FastSync, TestSyncConfig}
 import io.iohk.ethereum.blockchain.sync.regular.BlockBroadcasterActor
@@ -363,6 +364,14 @@ object FastSyncItSpecUtils {
       }
     }
 
+    def startWithState(): Task[Unit] = {
+      val currentBest = bl.getBestBlock().header
+      val safeTarget = currentBest.number + syncConfig.fastSyncBlockValidationX
+      val nextToValidate = currentBest.number + 1
+      val syncState = SyncState(currentBest, safeTarget, Seq(), Seq(), 0, 0, currentBest.number, nextToValidate)
+      Task(storagesInstance.storages.fastSyncStateStorage.putSyncState(syncState)).map(_ => ())
+    }
+
     def startFastSync(): Task[Unit] = Task {
       fastSync ! FastSync.Start
     }

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

@@ -0,0 +1,41 @@
+package io.iohk.ethereum.blockchain.sync
+
+import com.google.common.hash.{BloomFilter, Funnel}
+import io.iohk.ethereum.blockchain.sync.LoadableBloomFilter.BloomFilterLoadingResult
+import io.iohk.ethereum.db.dataSource.RocksDbDataSource.IterationError
+import monix.eval.Task
+import monix.reactive.{Consumer, Observable}
+
+class LoadableBloomFilter[A](bloomFilter: BloomFilter[A], source: Observable[Either[IterationError, A]]) {
+  val loadFromSource: Task[BloomFilterLoadingResult] = {
+    source
+      .consumeWith(Consumer.foldLeftTask(BloomFilterLoadingResult()) { (s, e) =>
+        e match {
+          case Left(value) => Task.now(s.copy(error = Some(value)))
+          case Right(value) => Task(bloomFilter.put(value)).map(_ => s.copy(writtenElements = s.writtenElements + 1))
+        }
+      })
+      .memoizeOnSuccess
+  }
+
+  def put(elem: A): Boolean = bloomFilter.put(elem)
+
+  def mightContain(elem: A): Boolean = bloomFilter.mightContain(elem)
+
+  def approximateElementCount: Long = bloomFilter.approximateElementCount()
+}
+
+object LoadableBloomFilter {
+  def apply[A](expectedSize: Int, loadingSource: Observable[Either[IterationError, A]])(implicit
+      f: Funnel[A]
+  ): LoadableBloomFilter[A] = {
+    new LoadableBloomFilter[A](BloomFilter.create[A](f, expectedSize), loadingSource)
+  }
+
+  case class BloomFilterLoadingResult(writtenElements: Long, error: Option[IterationError])
+  object BloomFilterLoadingResult {
+    def apply(): BloomFilterLoadingResult = new BloomFilterLoadingResult(0, None)
+
+    def apply(ex: Throwable): BloomFilterLoadingResult = new BloomFilterLoadingResult(0, Some(IterationError(ex)))
+  }
+}

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

@@ -4,14 +4,13 @@ import java.util.Comparator
 
 import akka.util.ByteString
 import com.google.common.hash.{BloomFilter, Funnel, PrimitiveSink}
+import io.iohk.ethereum.blockchain.sync.LoadableBloomFilter.BloomFilterLoadingResult
 import io.iohk.ethereum.blockchain.sync.SyncStateScheduler._
-import io.iohk.ethereum.db.dataSource.RocksDbDataSource.IterationError
 import io.iohk.ethereum.domain.{Account, Blockchain}
 import io.iohk.ethereum.mpt.{BranchNode, ExtensionNode, HashNode, LeafNode, MerklePatriciaTrie, MptNode}
 import io.vavr.collection.PriorityQueue
 import io.iohk.ethereum.network.p2p.messages.PV63.MptNodeEncoders._
 import monix.eval.Task
-import monix.reactive.Consumer
 
 import scala.annotation.tailrec
 import scala.util.Try
@@ -43,18 +42,9 @@ import scala.util.Try
   *
   * Important part is that nodes retrieved by getMissingNodes, must eventually be provided for scheduler to make progress
   */
-class SyncStateScheduler(blockchain: Blockchain, bloomFilter: BloomFilter[ByteString]) {
-
-  def loadFilterFromBlockchain: Task[BloomFilterLoadingResult] = {
-    blockchain
-      .mptStateSavedKeys()
-      .consumeWith(Consumer.foldLeftTask(BloomFilterLoadingResult()) { (s, e) =>
-        e match {
-          case Left(value) => Task.now(s.copy(error = Some(value)))
-          case Right(value) => Task(bloomFilter.put(value)).map(_ => s.copy(writtenElements = s.writtenElements + 1))
-        }
-      })
-  }
+class SyncStateScheduler(blockchain: Blockchain, bloomFilter: LoadableBloomFilter[ByteString]) {
+
+  val loadFilterFromBlockchain: Task[BloomFilterLoadingResult] = bloomFilter.loadFromSource
 
   def initState(targetRootHash: ByteString): Option[SchedulerState] = {
     if (targetRootHash == emptyStateRootHash) {
@@ -282,7 +272,7 @@ object SyncStateScheduler {
 
   case object StorageNode extends NodeRequest
 
-  object ByteStringFunnel extends Funnel[ByteString] {
+  implicit object ByteStringFunnel extends Funnel[ByteString] {
     override def funnel(from: ByteString, into: PrimitiveSink): Unit = {
       into.putBytes(from.toArray)
     }
@@ -291,10 +281,14 @@ object SyncStateScheduler {
   def getEmptyFilter(expectedFilterSize: Int): BloomFilter[ByteString] = {
     BloomFilter.create[ByteString](ByteStringFunnel, expectedFilterSize)
   }
-  // TODO [ETCM-213] add method to load bloom filter after node restart. Perfect way to do it would be to expose Observable
-  // in RocksDBDataSource which underneath would use RockDbIterator which would traverse whole namespace.
+
   def apply(blockchain: Blockchain, expectedBloomFilterSize: Int): SyncStateScheduler = {
-    new SyncStateScheduler(blockchain, getEmptyFilter(expectedBloomFilterSize))
+    // provided source i.e mptStateSavedKeys() is guaranteed to finish on first `Left` element which means that returned
+    // error is the reason why loading has stopped
+    new SyncStateScheduler(
+      blockchain,
+      LoadableBloomFilter[ByteString](expectedBloomFilterSize, blockchain.mptStateSavedKeys())
+    )
   }
 
   final case class StateNodeRequest(
@@ -477,12 +471,4 @@ object SyncStateScheduler {
   object ProcessingStatistics {
     def apply(): ProcessingStatistics = new ProcessingStatistics(0, 0, 0)
   }
-
-  case class BloomFilterLoadingResult(writtenElements: Long, error: Option[IterationError])
-  object BloomFilterLoadingResult {
-    def apply(): BloomFilterLoadingResult = new BloomFilterLoadingResult(0, None)
-
-    def apply(ex: Throwable): BloomFilterLoadingResult = new BloomFilterLoadingResult(0, Some(IterationError(ex)))
-  }
-
 }

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

@@ -2,13 +2,9 @@ package io.iohk.ethereum.blockchain.sync
 
 import akka.actor.{Actor, ActorLogging, ActorRef, Props, Timers}
 import akka.util.ByteString
+import io.iohk.ethereum.blockchain.sync.LoadableBloomFilter.BloomFilterLoadingResult
 import io.iohk.ethereum.blockchain.sync.SyncStateDownloaderActor.{CancelDownload, RegisterScheduler}
-import io.iohk.ethereum.blockchain.sync.SyncStateScheduler.{
-  BloomFilterLoadingResult,
-  ProcessingStatistics,
-  SchedulerState,
-  SyncResponse
-}
+import io.iohk.ethereum.blockchain.sync.SyncStateScheduler.{ProcessingStatistics, SchedulerState, SyncResponse}
 import io.iohk.ethereum.blockchain.sync.SyncStateSchedulerActor.{
   BloomFilterResult,
   GetMissingNodes,
@@ -48,6 +44,10 @@ class SyncStateSchedulerActor(downloader: ActorRef, sync: SyncStateScheduler, sy
 
   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)

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

@@ -0,0 +1,55 @@
+package io.iohk.ethereum.blockchain.sync
+
+import com.google.common.hash.{Funnel, Funnels, PrimitiveSink}
+import io.iohk.ethereum.FlatSpecBase
+import io.iohk.ethereum.db.dataSource.RocksDbDataSource.IterationError
+import monix.eval.Task
+import monix.reactive.Observable
+
+class LoadableBloomFilterSpec extends FlatSpecBase {
+  implicit object LongFun extends Funnel[Long] {
+    override def funnel(from: Long, into: PrimitiveSink): Unit = {
+      Funnels.longFunnel().funnel(from, into)
+    }
+  }
+
+  "LoadableBloomFilter" should "load all correct elements " in testCaseM[Task] {
+    for {
+      source <- Task(Observable.fromIterable(Seq(Right(1L), Right(2L), Right(3L))))
+      filter <- Task.now(LoadableBloomFilter[Long](1000, source))
+      result <- filter.loadFromSource
+    } yield {
+      assert(result.writtenElements == 3)
+      assert(result.error.isEmpty)
+      assert(filter.approximateElementCount == 3)
+    }
+  }
+
+  it should "load filter only once" in testCaseM[Task] {
+    for {
+      source <- Task(Observable.fromIterable(Seq(Right(1L), Right(2L), Right(3L))))
+      filter <- Task.now(LoadableBloomFilter[Long](1000, source))
+      result <- filter.loadFromSource
+      result1 <- filter.loadFromSource
+    } yield {
+      assert(result.writtenElements == 3)
+      assert(result.error.isEmpty)
+      assert(filter.approximateElementCount == 3)
+      assert(result1 == result)
+    }
+  }
+
+  it should "report last error if encountered" in testCaseM[Task] {
+    for {
+      error <- Task(IterationError(new RuntimeException("test")))
+      source <- Task(Observable.fromIterable(Seq(Right(1L), Right(2L), Right(3L), Left(error))))
+      filter <- Task.now(LoadableBloomFilter[Long](1000, source))
+      result <- filter.loadFromSource
+    } yield {
+      assert(result.writtenElements == 3)
+      assert(result.error.contains(error))
+      assert(filter.approximateElementCount == 3)
+    }
+  }
+
+}

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

@@ -6,14 +6,15 @@ import java.util.concurrent.ThreadLocalRandom
 import akka.actor.{ActorRef, ActorSystem}
 import akka.testkit.TestActor.AutoPilot
 import akka.testkit.{TestKit, TestProbe}
-import io.iohk.ethereum.blockchain.sync.StateSyncUtils.TrieProvider
+import akka.util.ByteString
+import io.iohk.ethereum.blockchain.sync.StateSyncUtils.{MptNodeData, TrieProvider}
 import io.iohk.ethereum.blockchain.sync.SyncStateSchedulerActor.{
   RestartRequested,
   StartSyncingTo,
   StateSyncFinished,
   WaitingForNewTargetBlock
 }
-import io.iohk.ethereum.domain.BlockchainImpl
+import io.iohk.ethereum.domain.{Address, BlockchainImpl}
 import io.iohk.ethereum.network.EtcPeerManagerActor.{GetHandshakedPeers, HandshakedPeers, PeerInfo, SendMessage}
 import io.iohk.ethereum.network.PeerEventBusActor.PeerEvent.MessageFromPeer
 import io.iohk.ethereum.network.p2p.messages.CommonMessages.Status
@@ -30,6 +31,7 @@ import org.scalatest.matchers.should.Matchers
 import org.scalatestplus.scalacheck.ScalaCheckPropertyChecks
 
 import scala.concurrent.duration._
+import scala.util.Random
 
 class StateSyncSpec
     extends TestKit(ActorSystem("MySpec"))
@@ -89,6 +91,24 @@ class StateSyncSpec
     }
   }
 
+  it should "start state sync when receiving start signal while bloom filter is loading" in new TestSetup() {
+    override def buildBlockChain(): BlockchainImpl = {
+      val storages = getNewStorages
+      //iterating 1M key and values should force scheduler actor o enqueue last received command i.e StartSyncing
+      (0 until 1000000).foreach { i =>
+        storages.storages.nodeStorage.update(Seq(), Seq(genRandomByteString() -> genRandomArray()))
+      }
+      BlockchainImpl(storages.storages)
+    }
+    val nodeData = (0 until 1000).map(i => MptNodeData(Address(i), None, Seq(), i))
+    val initiator = TestProbe()
+    val trieProvider1 = TrieProvider()
+    val target = trieProvider1.buildWorld(nodeData)
+    setAutoPilotWithProvider(trieProvider1)
+    initiator.send(scheduler, StartSyncingTo(target, 1))
+    initiator.expectMsg(20.seconds, StateSyncFinished)
+  }
+
   class TestSetup extends EphemBlockchainTestSetup with TestSyncConfig {
     override implicit lazy val system = actorSystem
     type PeerConfig = Map[PeerId, PeerAction]
@@ -195,12 +215,22 @@ class StateSyncSpec
       BlockchainImpl(getNewStorages.storages)
     }
 
+    def genRandomArray(): Array[Byte] = {
+      val arr = new Array[Byte](32)
+      Random.nextBytes(arr)
+      arr
+    }
+
+    def genRandomByteString(): ByteString = {
+      ByteString.fromArrayUnsafe(genRandomArray())
+    }
+
     lazy val scheduler = system.actorOf(
       SyncStateSchedulerActor.props(
         downloader,
-        new SyncStateScheduler(
+        SyncStateScheduler(
           buildBlockChain(),
-          SyncStateScheduler.getEmptyFilter(syncConfig.stateSyncBloomFilterSize)
+          syncConfig.stateSyncBloomFilterSize
         ),
         syncConfig
       )