[Consensus 2.0] Core recovery from storage (#16038)

## Description 

Recover Core from storage to make sure that we can make progress from
the latest possible quorum.

## Test Plan 

CI

---
If your changes are not user-facing and do not break anything, you can
skip the following section. Otherwise, please briefly describe what has
changed under the Release Notes section.

### Type of Change (Check all that apply)

- [ ] protocol change
- [ ] user-visible impact
- [ ] breaking change for a client SDKs
- [ ] breaking change for FNs (FN binary must upgrade)
- [ ] breaking change for validators or node operators (must upgrade
binaries)
- [ ] breaking change for on-chain data layout
- [ ] necessitate either a data wipe or data migration

### Release notes

consensus/config/Cargo.toml

@@ -12,7 +12,6 @@ mysten-network.workspace = true
 rand.workspace = true
 serde.workspace = true
 shared-crypto.workspace = true
-
 workspace-hack.workspace = true
 
 [dev-dependencies]

consensus/config/src/parameters.rs

@@ -1,6 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use std::path::PathBuf;
 use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
@@ -16,18 +17,32 @@ pub struct Parameters {
     /// Time to wait for parent round leader before sealing a block.
     #[serde(default = "Parameters::default_leader_timeout")]
     pub leader_timeout: Duration,
+
+    /// The database path. The path should be provided in order for the node to be able to boot
+    pub db_path: Option<PathBuf>,
 }
 
 impl Parameters {
     pub fn default_leader_timeout() -> Duration {
         Duration::from_millis(250)
     }
+
+    pub fn db_path_str_unsafe(&self) -> String {
+        self.db_path
+            .clone()
+            .expect("DB path is not set")
+            .as_path()
+            .to_str()
+            .unwrap()
+            .to_string()
+    }
 }
 
 impl Default for Parameters {
     fn default() -> Self {
         Self {
             leader_timeout: Parameters::default_leader_timeout(),
+            db_path: None,
         }
     }
 }

consensus/config/tests/snapshots/parameters_test__parameters.snap

@@ -5,4 +5,5 @@ expression: parameters
 leader_timeout:
   secs: 0
   nanos: 250000000
+db_path: ~
 

consensus/core/src/authority_node.rs

@@ -16,6 +16,7 @@ use crate::core::{Core, CoreSignals};
 use crate::core_thread::CoreThreadDispatcher;
 use crate::leader_timeout::{LeaderTimeoutTask, LeaderTimeoutTaskHandle};
 use crate::metrics::initialise_metrics;
+use crate::storage::rocksdb_store::RocksDBStore;
 use crate::transactions_client::{TransactionsClient, TransactionsConsumer};
 
 pub struct AuthorityNode {
@@ -55,12 +56,14 @@ impl AuthorityNode {
         // Construct Core
         let (core_signals, signals_receivers) = CoreSignals::new();
         let block_manager = BlockManager::new();
+        let store = Arc::new(RocksDBStore::new(&context.parameters.db_path_str_unsafe()));
         let core = Core::new(
             context.clone(),
             tx_consumer,
             block_manager,
             core_signals,
             block_signer,
+            store,
         );
 
         let (core_dispatcher, core_dispatcher_handle) =
@@ -104,6 +107,7 @@ mod tests {
     use fastcrypto::traits::ToFromBytes;
     use prometheus::Registry;
     use sui_protocol_config::ProtocolConfig;
+    use tempfile::TempDir;
 
     use crate::authority_node::AuthorityNode;
     use crate::block_verifier::TestBlockVerifier;
@@ -112,7 +116,11 @@ mod tests {
     async fn start_and_stop() {
         let (committee, keypairs) = local_committee_and_keys(0, vec![1]);
         let registry = Registry::new();
-        let parameters = Parameters::default();
+        let temp_dir = TempDir::new().unwrap();
+        let parameters = Parameters {
+            db_path: Some(temp_dir.into_path()),
+            ..Default::default()
+        };
         let block_verifier = TestBlockVerifier {};
 
         let (own_index, _) = committee.authorities().last().unwrap();

consensus/core/src/block.rs

@@ -71,10 +71,10 @@ pub enum Block {
 }
 
 impl Block {
-    /// Generate the genesis blocks for the latest Block version. The tuple contains (my_genesis_block, others_genesis_blocks).
+    /// Generate the genesis blocks for the latest Block version. The tuple contains (my_genesis_block, all_genesis_blocks).
     /// The blocks are returned in authority index order.
     pub(crate) fn genesis(context: Arc<Context>) -> (VerifiedBlock, Vec<VerifiedBlock>) {
-        let (my_block, others_block): (Vec<_>, Vec<_>) = context
+        let blocks = context
             .committee
             .authorities()
             .map(|(authority_index, _)| {
@@ -85,8 +85,15 @@ impl Block {
                 VerifiedBlock::new_verified_unserialized(signed)
                     .expect("Shouldn't fail when creating verified block for genesis")
             })
-            .partition(|block| block.author() == context.own_index);
-        (my_block[0].clone(), others_block)
+            .collect::<Vec<VerifiedBlock>>();
+        (
+            blocks
+                .iter()
+                .find(|b| b.author() == context.own_index)
+                .cloned()
+                .expect("We should have found our own genesis block"),
+            blocks,
+        )
     }
 }
 

consensus/core/src/context.rs

@@ -6,6 +6,9 @@ use std::sync::Arc;
 use consensus_config::{AuthorityIndex, Committee, Parameters};
 use sui_protocol_config::ProtocolConfig;
 
+#[cfg(test)]
+use tempfile::TempDir;
+
 use crate::metrics::Metrics;
 
 #[cfg(test)]
@@ -56,11 +59,15 @@ impl Context {
         let (committee, keypairs) =
             consensus_config::local_committee_and_keys(0, vec![1; committee_size]);
         let metrics = test_metrics();
+        let temp_dir = TempDir::new().unwrap();
 
         let context = Context::new(
             AuthorityIndex::new_for_test(0),
             committee,
-            Parameters::default(),
+            Parameters {
+                db_path: Some(temp_dir.into_path()),
+                ..Default::default()
+            },
             ProtocolConfig::get_for_max_version_UNSAFE(),
             metrics,
         );