ETCM-963 Refactoring write layer: "storeBlock" from Blockchain class

Author: Leonor Boga

src/it/scala/io/iohk/ethereum/sync/util/CommonFakePeer.scala

@@ -5,20 +5,16 @@ import java.nio.file.Files
 import java.nio.file.Path
 import java.time.Clock
 import java.util.concurrent.atomic.AtomicReference
-
 import akka.actor.ActorRef
 import akka.actor.ActorSystem
 import akka.testkit.TestProbe
 import akka.util.ByteString
 import akka.util.Timeout
-
 import monix.eval.Task
 
 import scala.concurrent.duration.FiniteDuration
 import scala.concurrent.duration._
-
 import org.bouncycastle.crypto.AsymmetricCipherKeyPair
-
 import io.iohk.ethereum.Fixtures
 import io.iohk.ethereum.Timeouts
 import io.iohk.ethereum.blockchain.sync.BlockchainHostActor
@@ -36,11 +32,15 @@ import io.iohk.ethereum.db.storage.AppStateStorage
 import io.iohk.ethereum.db.storage.Namespaces
 import io.iohk.ethereum.db.storage.pruning.ArchivePruning
 import io.iohk.ethereum.db.storage.pruning.PruningMode
-import io.iohk.ethereum.domain.Block
-import io.iohk.ethereum.domain.Blockchain
-import io.iohk.ethereum.domain.BlockchainImpl
-import io.iohk.ethereum.domain.BlockchainReader
-import io.iohk.ethereum.domain.ChainWeight
+import io.iohk.ethereum.domain.{
+  Block,
+  Blockchain,
+  BlockchainImpl,
+  BlockchainMetadata,
+  BlockchainReader,
+  BlockchainWriter,
+  ChainWeight
+}
 import io.iohk.ethereum.ledger.InMemoryWorldStateProxy
 import io.iohk.ethereum.mpt.MerklePatriciaTrie
 import io.iohk.ethereum.network.EtcPeerManagerActor
@@ -135,8 +135,13 @@ abstract class CommonFakePeer(peerName: String, fakePeerCustomConfig: FakePeerCu
     )
   )
 
-  val blockchainReader: BlockchainReader = BlockchainReader(storagesInstance.storages)
-  val bl: BlockchainImpl = BlockchainImpl(storagesInstance.storages, blockchainReader)
+  val blockchainMetadata = new BlockchainMetadata(
+    storagesInstance.storages.appStateStorage.getBestBlockNumber(),
+    storagesInstance.storages.appStateStorage.getLatestCheckpointBlockNumber()
+  )
+  val blockchainReader = BlockchainReader(storagesInstance.storages)
+  val blockchainWriter = BlockchainWriter(storagesInstance.storages, blockchainMetadata)
+  val bl: BlockchainImpl = BlockchainImpl(storagesInstance.storages, blockchainReader, blockchainMetadata)
   val evmCodeStorage = storagesInstance.storages.evmCodeStorage
 
   val genesis: Block = Block(

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

@@ -36,6 +36,7 @@ object FastSyncItSpecUtils {
         storagesInstance.storages.appStateStorage,
         bl,
         blockchainReader,
+        blockchainWriter,
         storagesInstance.storages.evmCodeStorage,
         storagesInstance.storages.nodeStorage,
         validators,

src/it/scala/io/iohk/ethereum/txExecTest/ECIP1017Test.scala

@@ -78,8 +78,12 @@ class ECIP1017Test extends AnyFlatSpec with Matchers {
 
     (startBlock to endBlock).foreach { blockToExecute =>
       val storages = FixtureProvider.prepareStorages(blockToExecute - 1, fixtures)
+      val blockchainMetadata = new BlockchainMetadata(
+        storages.appStateStorage.getBestBlockNumber(),
+        storages.appStateStorage.getLatestCheckpointBlockNumber()
+      )
       val blockchainReader = BlockchainReader(storages)
-      val blockchain = BlockchainImpl(storages, blockchainReader)
+      val blockchain = BlockchainImpl(storages, blockchainReader, blockchainMetadata)
       val blockValidation =
         new BlockValidation(consensus, blockchainReader, BlockQueue(blockchain, syncConfig))
       val blockExecution =

src/it/scala/io/iohk/ethereum/txExecTest/ForksTest.scala

@@ -72,7 +72,11 @@ class ForksTest extends AnyFlatSpec with Matchers {
     (startBlock to endBlock).foreach { blockToExecute =>
       val storages = FixtureProvider.prepareStorages(blockToExecute - 1, fixtures)
       val blockchainReader = BlockchainReader(storages)
-      val blockchain = BlockchainImpl(storages, blockchainReader)
+      val blockchainMetadata = new BlockchainMetadata(
+        storages.appStateStorage.getBestBlockNumber(),
+        storages.appStateStorage.getLatestCheckpointBlockNumber()
+      )
+      val blockchain = BlockchainImpl(storages, blockchainReader, blockchainMetadata)
       val blockValidation =
         new BlockValidation(consensus, blockchainReader, BlockQueue(blockchain, syncConfig))
       val blockExecution =

src/it/scala/io/iohk/ethereum/txExecTest/ScenarioSetup.scala

@@ -1,15 +1,18 @@
 package io.iohk.ethereum.txExecTest
 
 import io.iohk.ethereum.blockchain.sync.EphemBlockchainTestSetup
-import io.iohk.ethereum.domain.BlockchainImpl
-import io.iohk.ethereum.domain.BlockchainReader
-import io.iohk.ethereum.domain.BlockchainStorages
+import io.iohk.ethereum.domain.{BlockchainImpl, BlockchainMetadata, BlockchainReader, BlockchainStorages}
 import io.iohk.ethereum.ledger.VMImpl
 
 trait ScenarioSetup extends EphemBlockchainTestSetup {
   protected val testBlockchainStorages: BlockchainStorages
 
+  override lazy val blockchainMetadata = new BlockchainMetadata(
+    testBlockchainStorages.appStateStorage.getBestBlockNumber(),
+    testBlockchainStorages.appStateStorage.getLatestCheckpointBlockNumber()
+  )
   override lazy val blockchainReader: BlockchainReader = BlockchainReader(testBlockchainStorages)
-  override lazy val blockchain: BlockchainImpl = BlockchainImpl(testBlockchainStorages, blockchainReader)
+  override lazy val blockchain: BlockchainImpl =
+    BlockchainImpl(testBlockchainStorages, blockchainReader, blockchainMetadata)
   override lazy val vm: VMImpl = new VMImpl
 }

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

@@ -6,16 +6,14 @@ import akka.actor.ActorRef
 import akka.actor.PoisonPill
 import akka.actor.Props
 import akka.actor.Scheduler
-
 import io.iohk.ethereum.blockchain.sync.fast.FastSync
 import io.iohk.ethereum.blockchain.sync.regular.RegularSync
 import io.iohk.ethereum.consensus.validators.Validators
 import io.iohk.ethereum.db.storage.AppStateStorage
 import io.iohk.ethereum.db.storage.EvmCodeStorage
 import io.iohk.ethereum.db.storage.FastSyncStateStorage
 import io.iohk.ethereum.db.storage.NodeStorage
-import io.iohk.ethereum.domain.Blockchain
-import io.iohk.ethereum.domain.BlockchainReader
+import io.iohk.ethereum.domain.{Blockchain, BlockchainReader, BlockchainWriter}
 import io.iohk.ethereum.ledger.BlockImport
 import io.iohk.ethereum.ledger.BranchResolution
 import io.iohk.ethereum.utils.Config.SyncConfig
@@ -24,6 +22,7 @@ class SyncController(
     appStateStorage: AppStateStorage,
     blockchain: Blockchain,
     blockchainReader: BlockchainReader,
+    blockchainWriter: BlockchainWriter,
     evmCodeStorage: EvmCodeStorage,
     nodeStorage: NodeStorage,
     fastSyncStateStorage: FastSyncStateStorage,
@@ -92,6 +91,7 @@ class SyncController(
         appStateStorage,
         blockchain,
         blockchainReader,
+        blockchainWriter,
         evmCodeStorage,
         nodeStorage,
         validators,
@@ -139,6 +139,7 @@ object SyncController {
       appStateStorage: AppStateStorage,
       blockchain: Blockchain,
       blockchainReader: BlockchainReader,
+      blockchainWriter: BlockchainWriter,
       evmCodeStorage: EvmCodeStorage,
       nodeStorage: NodeStorage,
       syncStateStorage: FastSyncStateStorage,
@@ -156,6 +157,7 @@ object SyncController {
         appStateStorage,
         blockchain,
         blockchainReader,
+        blockchainWriter,
         evmCodeStorage,
         nodeStorage,
         syncStateStorage,

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

@@ -54,6 +54,7 @@ class FastSync(
     val appStateStorage: AppStateStorage,
     val blockchain: Blockchain,
     val blockchainReader: BlockchainReader,
+    blockchainWriter: BlockchainWriter,
     evmCodeStorage: EvmCodeStorage,
     nodeStorage: NodeStorage,
     val validators: Validators,
@@ -588,7 +589,7 @@ class FastSync(
     }
 
     private def updateSyncState(header: BlockHeader, parentWeight: ChainWeight): Unit = {
-      blockchain
+      blockchainWriter
         .storeBlockHeader(header)
         .and(blockchain.storeChainWeight(header.hash, parentWeight.increase(header)))
         .commit()
@@ -798,7 +799,7 @@ class FastSync(
       requestedHashes
         .zip(blockBodies)
         .map { case (hash, body) =>
-          blockchain.storeBlockBody(hash, body)
+          blockchainWriter.storeBlockBody(hash, body)
         }
         .reduce(_.and(_))
         .commit()
@@ -1147,6 +1148,7 @@ object FastSync {
       appStateStorage: AppStateStorage,
       blockchain: Blockchain,
       blockchainReader: BlockchainReader,
+      blockchainWriter: BlockchainWriter,
       evmCodeStorage: EvmCodeStorage,
       nodeStorage: NodeStorage,
       validators: Validators,
@@ -1162,6 +1164,7 @@ object FastSync {
         appStateStorage,
         blockchain,
         blockchainReader,
+        blockchainWriter,
         evmCodeStorage,
         nodeStorage,
         validators,

src/main/scala/io/iohk/ethereum/domain/Blockchain.scala

@@ -15,7 +15,6 @@ import io.iohk.ethereum.db.storage.NodeStorage.NodeHash
 import io.iohk.ethereum.db.storage.TransactionMappingStorage.TransactionLocation
 import io.iohk.ethereum.db.storage._
 import io.iohk.ethereum.domain
-import io.iohk.ethereum.domain.BlockchainImpl.BestBlockLatestCheckpointNumbers
 import io.iohk.ethereum.jsonrpc.ProofService.StorageProof
 import io.iohk.ethereum.ledger.InMemoryWorldStateProxy
 import io.iohk.ethereum.ledger.InMemoryWorldStateProxyStorage
@@ -139,7 +138,8 @@ class BlockchainImpl(
     protected val transactionMappingStorage: TransactionMappingStorage,
     protected val appStateStorage: AppStateStorage,
     protected val stateStorage: StateStorage,
-    blockchainReader: BlockchainReader
+    blockchainReader: BlockchainReader,
+    blockchainMetadata: BlockchainMetadata
 ) extends Blockchain
     with Logger {
 
@@ -164,7 +164,7 @@ class BlockchainImpl(
 
   override def getBestBlockNumber(): BigInt = {
     val bestSavedBlockNumber = appStateStorage.getBestBlockNumber()
-    val bestKnownBlockNumber = bestKnownBlockAndLatestCheckpoint.get().bestBlockNumber
+    val bestKnownBlockNumber = blockchainMetadata.bestKnownBlockAndLatestCheckpoint.get().bestBlockNumber
     log.debug(
       "Current best saved block number {}. Current best known block number {}",
       bestSavedBlockNumber,
@@ -177,7 +177,7 @@ class BlockchainImpl(
   }
 
   override def getLatestCheckpointBlockNumber(): BigInt =
-    bestKnownBlockAndLatestCheckpoint.get().latestCheckpointNumber
+    blockchainMetadata.bestKnownBlockAndLatestCheckpoint.get().latestCheckpointNumber
 
   //returns the best known block if it's available in the storage, otherwise the best stored block
   override def getBestBlock(): Option[Block] = {
@@ -311,10 +311,12 @@ class BlockchainImpl(
     }
 
   private def saveBestKnownBlock(bestBlockNumber: BigInt): Unit =
-    bestKnownBlockAndLatestCheckpoint.updateAndGet(_.copy(bestBlockNumber = bestBlockNumber))
+    blockchainMetadata.bestKnownBlockAndLatestCheckpoint.updateAndGet(_.copy(bestBlockNumber = bestBlockNumber))
 
   private def saveBestKnownBlockAndLatestCheckpointNumber(number: BigInt, latestCheckpointNumber: BigInt): Unit =
-    bestKnownBlockAndLatestCheckpoint.set(BestBlockLatestCheckpointNumbers(number, latestCheckpointNumber))
+    blockchainMetadata.bestKnownBlockAndLatestCheckpoint.set(
+      BestBlockLatestCheckpointNumbers(number, latestCheckpointNumber)
+    )
 
   def storeChainWeight(blockhash: ByteString, weight: ChainWeight): DataSourceBatchUpdate =
     chainWeightStorage.put(blockhash, weight)
@@ -461,7 +463,11 @@ trait BlockchainStorages {
 }
 
 object BlockchainImpl {
-  def apply(storages: BlockchainStorages, blockchainReader: BlockchainReader): BlockchainImpl =
+  def apply(
+      storages: BlockchainStorages,
+      blockchainReader: BlockchainReader,
+      metadata: BlockchainMetadata
+  ): BlockchainImpl =
     new BlockchainImpl(
       blockHeadersStorage = storages.blockHeadersStorage,
       blockBodiesStorage = storages.blockBodiesStorage,
@@ -471,8 +477,7 @@ object BlockchainImpl {
       transactionMappingStorage = storages.transactionMappingStorage,
       appStateStorage = storages.appStateStorage,
       stateStorage = storages.stateStorage,
-      blockchainReader = blockchainReader
+      blockchainReader = blockchainReader,
+      blockchainMetadata = metadata
     )
-
-  private case class BestBlockLatestCheckpointNumbers(bestBlockNumber: BigInt, latestCheckpointNumber: BigInt)
 }

src/main/scala/io/iohk/ethereum/domain/BlockchainMetadata.scala

@@ -0,0 +1,10 @@
+package io.iohk.ethereum.domain
+
+import java.util.concurrent.atomic.AtomicReference
+
+class BlockchainMetadata(bestBlockNumber: BigInt, latestCheckpointNumber: BigInt) {
+  lazy val bestKnownBlockAndLatestCheckpoint: AtomicReference[BestBlockLatestCheckpointNumbers] =
+    new AtomicReference(BestBlockLatestCheckpointNumbers(bestBlockNumber, latestCheckpointNumber))
+}
+
+case class BestBlockLatestCheckpointNumbers(bestBlockNumber: BigInt, latestCheckpointNumber: BigInt)

src/main/scala/io/iohk/ethereum/domain/BlockchainWriter.scala

@@ -0,0 +1,59 @@
+package io.iohk.ethereum.domain
+
+import akka.util.ByteString
+import io.iohk.ethereum.db.dataSource.DataSourceBatchUpdate
+import io.iohk.ethereum.db.storage.TransactionMappingStorage.TransactionLocation
+import io.iohk.ethereum.db.storage.{
+  BlockBodiesStorage,
+  BlockHeadersStorage,
+  BlockNumberMappingStorage,
+  TransactionMappingStorage
+}
+import io.iohk.ethereum.utils.Logger
+
+class BlockchainWriter(
+    blockHeadersStorage: BlockHeadersStorage,
+    blockBodiesStorage: BlockBodiesStorage,
+    blockNumberMappingStorage: BlockNumberMappingStorage,
+    transactionMappingStorage: TransactionMappingStorage
+) extends Logger {
+
+  /**
+    * Persists a block in the underlying Blockchain Database
+    * Note: all store* do not update the database immediately, rather they create
+    * a [[io.iohk.ethereum.db.dataSource.DataSourceBatchUpdate]] which then has to be committed (atomic operation)
+    *
+    * @param block Block to be saved
+    */
+  def storeBlock(block: Block): DataSourceBatchUpdate = {
+    storeBlockHeader(block.header).and(storeBlockBody(block.header.hash, block.body))
+  }
+
+  def storeBlockHeader(blockHeader: BlockHeader): DataSourceBatchUpdate = {
+    val hash = blockHeader.hash
+    blockHeadersStorage.put(hash, blockHeader).and(saveBlockNumberMapping(blockHeader.number, hash))
+  }
+
+  def storeBlockBody(blockHash: ByteString, blockBody: BlockBody): DataSourceBatchUpdate = {
+    blockBodiesStorage.put(blockHash, blockBody).and(saveTxsLocations(blockHash, blockBody))
+  }
+
+  private def saveBlockNumberMapping(number: BigInt, hash: ByteString): DataSourceBatchUpdate =
+    blockNumberMappingStorage.put(number, hash)
+
+  private def saveTxsLocations(blockHash: ByteString, blockBody: BlockBody): DataSourceBatchUpdate =
+    blockBody.transactionList.zipWithIndex.foldLeft(transactionMappingStorage.emptyBatchUpdate) {
+      case (updates, (tx, index)) =>
+        updates.and(transactionMappingStorage.put(tx.hash, TransactionLocation(blockHash, index)))
+    }
+}
+
+object BlockchainWriter {
+  def apply(storages: BlockchainStorages): BlockchainWriter =
+    new BlockchainWriter(
+      storages.blockHeadersStorage,
+      storages.blockBodiesStorage,
+      storages.blockNumberMappingStorage,
+      storages.transactionMappingStorage
+    )
+}