etcm-75 added support for saving latest checkpoint number in blockchain

Author: pslaski

src/it/scala/io/iohk/ethereum/txExecTest/util/DumpChainApp.scala

@@ -169,4 +169,6 @@ object DumpChainApp extends App with NodeKeyBuilder with SecureRandomBuilder wit
     override def save(block: Block, receipts: Seq[Receipt], totalDifficulty: BigInt, saveAsBestBlock: Boolean): Unit = ???
 
     override def getStateStorage: StateStorage = ???
+
+    override def getLatestCheckpointBlockNumber(): BigInt = ???
   }

src/main/scala/io/iohk/ethereum/db/storage/StateStorage.scala

@@ -18,8 +18,8 @@ trait StateStorage {
   def getBackingStorage(bn: BigInt): MptStorage
   def getReadOnlyStorage: MptStorage
 
-  def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit
-  def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit
+  def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit
+  def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit
 
   def saveNode(nodeHash: NodeHash, nodeEncoded: NodeEncoded, bn: BigInt)
   def getNode(nodeHash: NodeHash): Option[MptNode]
@@ -34,15 +34,15 @@ class ArchiveStateStorage(private val nodeStorage: NodeStorage,
     true
   }
 
-  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     if (cachedNodeStorage.persist()) {
-      updateBestBlock(None)
+      updateBestBlocksData
     }
   }
 
-  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     if (cachedNodeStorage.persist()) {
-      updateBestBlock(None)
+      updateBestBlocksData
     }
   }
 
@@ -71,21 +71,21 @@ class ReferenceCountedStateStorage(private val nodeStorage: NodeStorage,
     true
   }
 
-  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     val blockToPrune = bn - pruningHistory
 
     ReferenceCountNodeStorage.prune(blockToPrune, cachedNodeStorage, inMemory = blockToPrune > currentBestSavedBlock)
 
     if (cachedNodeStorage.persist()) {
-      updateBestBlock(None)
+      updateBestBlocksData
     }
   }
 
-  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     ReferenceCountNodeStorage.rollback(bn, cachedNodeStorage, inMemory = bn > currentBestSavedBlock)
 
     if (cachedNodeStorage.persist()) {
-      updateBestBlock(None)
+      updateBestBlocksData
     }
   }
 
@@ -120,19 +120,19 @@ class CachedReferenceCountedStateStorage(private val nodeStorage: NodeStorage,
     }
   }
 
-  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockSave(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     val blockToPrune = bn - pruningHistory
     changeLog.persistChangeLog(bn)
     changeLog.getDeathRowFromStorage(blockToPrune).foreach {deathRow =>
       CachedReferenceCountedStorage.prune(deathRow, lruCache, blockToPrune)
     }
     if (CachedReferenceCountedStorage.persistCache(lruCache, nodeStorage)) {
-      updateBestBlock(None)
+      updateBestBlocksData
     }
     changeLog.removeBlockMetaData(blockToPrune)
   }
 
-  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlock: Option[BigInt] => Unit): Unit = {
+  override def onBlockRollback(bn: BigInt, currentBestSavedBlock: BigInt)(updateBestBlocksData: => Unit): Unit = {
     changeLog.getChangeLogFromStorage(bn).foreach { changeLog =>
       CachedReferenceCountedStorage.rollback(lruCache, nodeStorage, changeLog, bn)
     }

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

@@ -25,6 +25,8 @@ case class Block(header: BlockHeader, body: BlockBody) {
 
   def hash: ByteString = header.hash
 
+  val hasCheckpoint: Boolean = header.hasCheckpoint
+
   def isParentOf(child: Block): Boolean = number + 1 == child.number && child.header.parentHash == hash
 }
 

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

@@ -10,10 +10,13 @@ import io.iohk.ethereum.db.storage.TransactionMappingStorage.TransactionLocation
 import io.iohk.ethereum.db.storage._
 import io.iohk.ethereum.db.storage.pruning.PruningMode
 import io.iohk.ethereum.domain
+import io.iohk.ethereum.domain.BlockchainImpl.BestBlockLatestCheckpointNumbers
 import io.iohk.ethereum.ledger.{InMemoryWorldStateProxy, InMemoryWorldStateProxyStorage}
 import io.iohk.ethereum.mpt.{MerklePatriciaTrie, MptNode}
 import io.iohk.ethereum.vm.{Storage, WorldStateProxy}
 
+import scala.annotation.tailrec
+
 /**
   * Entity to be used to persist and query  Blockchain related objects (blocks, transactions, ommers)
   */
@@ -123,6 +126,7 @@ trait Blockchain {
 
   def getBestBlock(): Block
 
+  def getLatestCheckpointBlockNumber(): BigInt
 
   /**
     * Persists full block along with receipts and total difficulty
@@ -209,7 +213,8 @@ class BlockchainImpl(
 
   // There is always only one writer thread (ensured by actor), but can by many readers (api calls)
   // to ensure visibility of writes, needs to be volatile or atomic ref
-  private val bestKnownBlock: AtomicReference[BigInt] = new AtomicReference(BigInt(0))
+  private val bestKnownBlockAndLatestCheckpoint: AtomicReference[BestBlockLatestCheckpointNumbers] =
+    new AtomicReference(BestBlockLatestCheckpointNumbers(BigInt(0), BigInt(0)))
 
   override def getBlockHeaderByHash(hash: ByteString): Option[BlockHeader] =
     blockHeadersStorage.get(hash)
@@ -225,12 +230,20 @@ class BlockchainImpl(
 
   override def getBestBlockNumber(): BigInt = {
     val bestBlockNum = appStateStorage.getBestBlockNumber()
-    if (bestKnownBlock.get() > bestBlockNum)
-      bestKnownBlock.get()
+    if (bestKnownBlockAndLatestCheckpoint.get().bestBlockNumber > bestBlockNum)
+      bestKnownBlockAndLatestCheckpoint.get().bestBlockNumber
     else
       bestBlockNum
   }
 
+  override def getLatestCheckpointBlockNumber(): BigInt = {
+    val latestCheckpointNumberInStorage = appStateStorage.getLatestCheckpointBlockNumber()
+    if (bestKnownBlockAndLatestCheckpoint.get().latestCheckpointNumber > latestCheckpointNumberInStorage)
+      bestKnownBlockAndLatestCheckpoint.get().latestCheckpointNumber
+    else
+      latestCheckpointNumberInStorage
+  }
+
   override def getBestBlock(): Block =
     getBlockByNumber(getBestBlockNumber()).get
 
@@ -252,8 +265,10 @@ class BlockchainImpl(
     ByteString(mpt.get(position).getOrElse(BigInt(0)).toByteArray)
   }
 
-  def saveBestBlock(bestBlock: Option[BigInt]): Unit = {
-    bestBlock.fold(appStateStorage.putBestBlockNumber(getBestBlockNumber()).commit())(best => appStateStorage.putBestBlockNumber(best).commit())
+  private def persistBestBlocksData(): Unit = {
+    appStateStorage.putBestBlockNumber(getBestBlockNumber())
+      .and(appStateStorage.putLatestCheckpointBlockNumber(getLatestCheckpointBlockNumber()))
+      .commit()
   }
 
   def save(block: Block, receipts: Seq[Receipt], totalDifficulty: BigInt, saveAsBestBlock: Boolean): Unit = {
@@ -263,8 +278,12 @@ class BlockchainImpl(
       .commit()
 
     // not transactional part
-    stateStorage.onBlockSave(block.header.number, appStateStorage.getBestBlockNumber())(saveBestBlock)
-    if (saveAsBestBlock) {
+    // the best blocks data will be persisted only when the cache will be persisted
+    stateStorage.onBlockSave(block.header.number, appStateStorage.getBestBlockNumber())(persistBestBlocksData())
+
+    if (saveAsBestBlock && block.hasCheckpoint) {
+      saveBestKnownBlockAndLatestCheckpointNumber(block.header.number, block.header.number)
+    } else if (saveAsBestBlock) {
       saveBestKnownBlock(block.header.number)
     }
   }
@@ -290,7 +309,15 @@ class BlockchainImpl(
     evmCodeStorage.put(hash, evmCode)
 
   override def saveBestKnownBlock(number: BigInt): Unit = {
-    bestKnownBlock.set(number)
+    bestKnownBlockAndLatestCheckpoint.updateAndGet(_.copy(bestBlockNumber = number))
+  }
+
+  def saveBestKnownBlockAndLatestCheckpointNumber(number: BigInt, latestCheckpointNumber: BigInt): Unit = {
+    bestKnownBlockAndLatestCheckpoint.set(BestBlockLatestCheckpointNumbers(number, latestCheckpointNumber))
+  }
+
+  def saveLatestCheckpointNumber(latestCheckpointNumber: BigInt): Unit = {
+    bestKnownBlockAndLatestCheckpoint.updateAndGet(_.copy(latestCheckpointNumber = latestCheckpointNumber))
   }
 
   def storeTotalDifficulty(blockhash: ByteString, td: BigInt): DataSourceBatchUpdate =
@@ -310,6 +337,7 @@ class BlockchainImpl(
     blockNumberMappingStorage.remove(number)
   }
 
+// scalastyle:off method.length
   override def removeBlock(blockHash: ByteString, withState: Boolean): Unit = {
     val maybeBlockHeader = getBlockHeaderByHash(blockHash)
     val maybeTxList = getBlockBodyByHash(blockHash).map(_.transactionList)
@@ -323,20 +351,64 @@ class BlockchainImpl(
       )
     }
 
+    val checkpointUpdates = maybeBlockHeader match {
+      case Some(header) =>
+        if (header.hasCheckpoint && header.number == getLatestCheckpointBlockNumber()) {
+          val prev = findPreviousCheckpointBlockNumber(header.number, header.number)
+          prev.map { num =>
+            // side effect
+            saveLatestCheckpointNumber(num)
+            appStateStorage.putLatestCheckpointBlockNumber(num)
+          }.getOrElse {
+            // side effect
+            saveLatestCheckpointNumber(0)
+            appStateStorage.removeLatestCheckpointBlockNumber()
+          }
+        } else appStateStorage.emptyBatchUpdate
+      case None =>
+        appStateStorage.emptyBatchUpdate
+    }
+
     blockHeadersStorage.remove(blockHash)
       .and(blockBodiesStorage.remove(blockHash))
       .and(totalDifficultyStorage.remove(blockHash))
       .and(receiptStorage.remove(blockHash))
       .and(maybeTxList.fold(transactionMappingStorage.emptyBatchUpdate)(removeTxsLocations))
       .and(blockNumberMappingUpdates)
+    // immediate last checkpoint save to not need to manage it from outside
+      .and(checkpointUpdates)
       .commit()
 
     // not transactional part
     maybeBlockHeader.foreach { h =>
       if (withState)
-        stateStorage.onBlockRollback(h.number, bestSavedBlock)(saveBestBlock)
+        stateStorage.onBlockRollback(h.number, bestSavedBlock)(persistBestBlocksData())
     }
   }
+  // scalastyle:on method.length
+
+  /**
+    * Recursive function which try to find the previous checkpoint by traversing blocks from top to the bottom.
+    * In case of finding the checkpoint block number, the function will finish the job and return result
+    */
+  @tailrec
+  private def findPreviousCheckpointBlockNumber(
+    blockNumberToCheck: BigInt,
+    latestCheckpointBlockNumber: BigInt
+  ): Option[BigInt] = {
+    if (blockNumberToCheck > 0) {
+      val maybePreviousCheckpointBlockNumber = for {
+        currentBlock <- getBlockByNumber(blockNumberToCheck)
+        if currentBlock.hasCheckpoint &&
+          currentBlock.number < latestCheckpointBlockNumber
+      } yield currentBlock.number
+
+      maybePreviousCheckpointBlockNumber match {
+        case Some(_) => maybePreviousCheckpointBlockNumber
+        case None => findPreviousCheckpointBlockNumber(blockNumberToCheck - 1, latestCheckpointBlockNumber)
+      }
+    } else None
+  }
 
   private def saveTxsLocations(blockHash: ByteString, blockBody: BlockBody): DataSourceBatchUpdate =
     blockBody.transactionList.zipWithIndex.foldLeft(transactionMappingStorage.emptyBatchUpdate) {
@@ -423,4 +495,6 @@ object BlockchainImpl {
       appStateStorage = storages.appStateStorage,
       stateStorage = storages.stateStorage
     )
+
+  private case class BestBlockLatestCheckpointNumbers(bestBlockNumber: BigInt, latestCheckpointNumber: BigInt)
 }

src/test/scala/io/iohk/ethereum/db/storage/StateStorageSpec.scala

@@ -70,7 +70,7 @@ class StateStorageSpec extends AnyFlatSpec with Matchers with ScalaCheckProperty
 
       if (testCache.shouldPersist) {
         val sizeBefore = ints.size
-        archiveStateStorage.onBlockSave(1, 0) { None =>
+        archiveStateStorage.onBlockSave(1, 0) {
           ints = 1 :: ints
         }
 
@@ -88,7 +88,7 @@ class StateStorageSpec extends AnyFlatSpec with Matchers with ScalaCheckProperty
 
       if (testCache.shouldPersist) {
         val sizeBefore = ints.size
-        archiveStateStorage.onBlockRollback(1, 0) { None =>
+        archiveStateStorage.onBlockRollback(1, 0) {
           ints = 1 :: ints
         }
 
@@ -118,7 +118,7 @@ class StateStorageSpec extends AnyFlatSpec with Matchers with ScalaCheckProperty
 
       if (testCache.shouldPersist) {
         val sizeBefore = ints.size
-        referenceCounteStateStorage.onBlockSave(1, 0) { None =>
+        referenceCounteStateStorage.onBlockSave(1, 0) {
           ints = 1 :: ints
         }
 
@@ -136,7 +136,7 @@ class StateStorageSpec extends AnyFlatSpec with Matchers with ScalaCheckProperty
 
       if (testCache.shouldPersist) {
         val sizeBefore = ints.size
-        referenceCounteStateStorage.onBlockRollback(1, 0) { None =>
+        referenceCounteStateStorage.onBlockRollback(1, 0) {
           ints = 1 :: ints
         }