Feature/etcm 967 db consistency check

src/main/scala/io/iohk/ethereum/nodebuilder/StdNode.scala

@@ -1,5 +1,7 @@
 package io.iohk.ethereum.nodebuilder
 
+import akka.actor.typed.ActorSystem
+
 import scala.concurrent.Await
 import scala.concurrent.ExecutionContext.Implicits.global
 import scala.util.Failure
@@ -13,6 +15,8 @@ import io.iohk.ethereum.metrics.MetricsConfig
 import io.iohk.ethereum.network.PeerManagerActor
 import io.iohk.ethereum.network.ServerActor
 import io.iohk.ethereum.network.discovery.PeerDiscoveryManager
+import io.iohk.ethereum.nodebuilder.tooling.PeriodicConsistencyCheck
+import io.iohk.ethereum.nodebuilder.tooling.StorageConsistencyChecker
 import io.iohk.ethereum.utils.Config
 
 /** A standard node is everything Ethereum prescribes except the mining algorithm,
@@ -24,28 +28,32 @@ import io.iohk.ethereum.utils.Config
   * @see [[io.iohk.ethereum.nodebuilder.Node Node]]
   */
 abstract class BaseNode extends Node {
-  private[this] def loadGenesisData(): Unit =
-    if (!Config.testmode) genesisDataLoader.loadGenesisData()
 
-  private[this] def startPeerManager(): Unit = peerManager ! PeerManagerActor.StartConnecting
+  def start(): Unit = {
+    startMetricsClient()
 
-  private[this] def startServer(): Unit = server ! ServerActor.StartServer(networkConfig.Server.listenAddress)
+    loadGenesisData()
 
-  private[this] def startSyncController(): Unit = syncController ! SyncProtocol.Start
+    runDBConsistencyCheck()
 
-  private[this] def startMining(): Unit = mining.startProtocol(this)
+    startPeerManager()
 
-  private[this] def startDiscoveryManager(): Unit = peerDiscoveryManager ! PeerDiscoveryManager.Start
+    startPortForwarding()
 
-  private[this] def startJsonRpcHttpServer(): Unit =
-    maybeJsonRpcHttpServer match {
-      case Right(jsonRpcServer) if jsonRpcConfig.httpServerConfig.enabled => jsonRpcServer.run()
-      case Left(error) if jsonRpcConfig.httpServerConfig.enabled          => log.error(error)
-      case _                                                              => //Nothing
-    }
+    startServer()
 
-  private[this] def startJsonRpcIpcServer(): Unit =
-    if (jsonRpcConfig.ipcServerConfig.enabled) jsonRpcIpcServer.run()
+    startSyncController()
+
+    startMining()
+
+    startDiscoveryManager()
+
+    startJsonRpcHttpServer()
+
+    startJsonRpcIpcServer()
+
+    startPeriodicDBConsistencyCheck()
+  }
 
   private[this] def startMetricsClient(): Unit = {
     val metricsConfig = MetricsConfig(Config.config)
@@ -56,25 +64,47 @@ abstract class BaseNode extends Node {
     }
   }
 
-  def start(): Unit = {
-    startMetricsClient()
+  private[this] def loadGenesisData(): Unit =
+    if (!Config.testmode) genesisDataLoader.loadGenesisData()
 
-    loadGenesisData()
+  private[this] def runDBConsistencyCheck(): Unit =
+    StorageConsistencyChecker.checkStorageConsistency(
+      storagesInstance.storages.appStateStorage.getBestBlockNumber(),
+      storagesInstance.storages.blockNumberMappingStorage,
+      storagesInstance.storages.blockHeadersStorage,
+      shutdown
+    )(log)
 
-    startPeerManager()
+  private[this] def startPeerManager(): Unit = peerManager ! PeerManagerActor.StartConnecting
 
-    startPortForwarding()
-    startServer()
+  private[this] def startServer(): Unit = server ! ServerActor.StartServer(networkConfig.Server.listenAddress)
 
-    startSyncController()
+  private[this] def startSyncController(): Unit = syncController ! SyncProtocol.Start
 
-    startMining()
+  private[this] def startMining(): Unit = mining.startProtocol(this)
 
-    startDiscoveryManager()
+  private[this] def startDiscoveryManager(): Unit = peerDiscoveryManager ! PeerDiscoveryManager.Start
 
-    startJsonRpcHttpServer()
-    startJsonRpcIpcServer()
-  }
+  private[this] def startJsonRpcHttpServer(): Unit =
+    maybeJsonRpcHttpServer match {
+      case Right(jsonRpcServer) if jsonRpcConfig.httpServerConfig.enabled => jsonRpcServer.run()
+      case Left(error) if jsonRpcConfig.httpServerConfig.enabled          => log.error(error)
+      case _                                                              => //Nothing
+    }
+
+  private[this] def startJsonRpcIpcServer(): Unit =
+    if (jsonRpcConfig.ipcServerConfig.enabled) jsonRpcIpcServer.run()
+
+  def startPeriodicDBConsistencyCheck(): Unit =
+    ActorSystem(
+      PeriodicConsistencyCheck.start(
+        storagesInstance.storages.appStateStorage,
+        storagesInstance.storages.blockNumberMappingStorage,
+        storagesInstance.storages.blockHeadersStorage,
+        shutdown
+      ),
+      "PeriodicDBConsistencyCheck"
+    )
 
   override def shutdown(): Unit = {
     def tryAndLogFailure(f: () => Any): Unit = Try(f()) match {

src/main/scala/io/iohk/ethereum/nodebuilder/tooling/PeriodicConsistencyCheck.scala

@@ -0,0 +1,55 @@
+package io.iohk.ethereum.nodebuilder.tooling
+
+import akka.actor.typed.Behavior
+import akka.actor.typed.scaladsl.Behaviors
+import akka.actor.typed.scaladsl.TimerScheduler
+
+import scala.concurrent.duration.DurationInt
+
+import io.iohk.ethereum.db.storage.AppStateStorage
+import io.iohk.ethereum.db.storage.BlockHeadersStorage
+import io.iohk.ethereum.db.storage.BlockNumberMappingStorage
+import io.iohk.ethereum.nodebuilder.tooling.PeriodicConsistencyCheck.ConsistencyCheck
+import io.iohk.ethereum.utils.Logger
+
+object PeriodicConsistencyCheck {
+  def start(
+      appStateStorage: AppStateStorage,
+      blockNumberMappingStorage: BlockNumberMappingStorage,
+      blockHeadersStorage: BlockHeadersStorage,
+      shutdown: () => Unit
+  ): Behavior[ConsistencyCheck] =
+    Behaviors.withTimers { timers =>
+      tick(timers)
+      PeriodicConsistencyCheck(timers, appStateStorage, blockNumberMappingStorage, blockHeadersStorage, shutdown)
+        .check()
+    }
+
+  sealed trait ConsistencyCheck extends Product with Serializable
+  case object Tick extends ConsistencyCheck
+
+  def tick(timers: TimerScheduler[ConsistencyCheck]): Unit =
+    timers.startSingleTimer(Tick, 10.minutes)
+}
+
+case class PeriodicConsistencyCheck(
+    timers: TimerScheduler[ConsistencyCheck],
+    appStateStorage: AppStateStorage,
+    blockNumberMappingStorage: BlockNumberMappingStorage,
+    blockHeadersStorage: BlockHeadersStorage,
+    shutdown: () => Unit
+) extends Logger {
+  import PeriodicConsistencyCheck._
+
+  def check(): Behavior[ConsistencyCheck] = Behaviors.receiveMessage { case Tick =>
+    log.debug("Running a storageConsistency check")
+    StorageConsistencyChecker.checkStorageConsistency(
+      appStateStorage.getBestBlockNumber(),
+      blockNumberMappingStorage,
+      blockHeadersStorage,
+      shutdown
+    )(log)
+    tick(timers)
+    Behaviors.same
+  }
+}

src/main/scala/io/iohk/ethereum/nodebuilder/tooling/StorageConsistencyChecker.scala

@@ -0,0 +1,29 @@
+package io.iohk.ethereum.nodebuilder.tooling
+
+import com.typesafe.scalalogging.Logger
+
+import io.iohk.ethereum.db.storage.BlockHeadersStorage
+import io.iohk.ethereum.db.storage.BlockNumberMappingStorage
+
+object StorageConsistencyChecker {
+  type ShutdownOp = () => Unit
+
+  val DefaultStep = 1000
+
+  def checkStorageConsistency(
+      bestBlockNumber: BigInt,
+      blockNumberMappingStorage: BlockNumberMappingStorage,
+      blockHeadersStorage: BlockHeadersStorage,
+      shutdown: ShutdownOp,
+      step: Int = DefaultStep
+  )(implicit log: Logger): Unit =
+    Range(0, bestBlockNumber.intValue, step).foreach { idx =>
+      (for {
+        hash <- blockNumberMappingStorage.get(idx)
+        _ <- blockHeadersStorage.get(hash)
+      } yield ()).fold {
+        log.error("Database seems to be in inconsistent state, shutting down")
+        shutdown()
+      }(_ => ())
+    }
+}