CORE-285 Storage refactoring

executor/evm/src/block_hash.rs

@@ -3,8 +3,7 @@
 use std::sync::Arc;
 
 use casper_storage::block_store::{
-    lmdb::IndexedLmdbBlockStore, types::BlockHeight, BlockStoreError, BlockStoreProvider,
-    DataReader,
+    lmdb::LmdbBlockStore, types::BlockHeight, BlockStoreError, BlockStoreProvider, DataReader,
 };
 use casper_types::{BlockHash, BlockHeader};
 
@@ -42,12 +41,12 @@ impl BlockHashProvider for NoBlockHashProvider {
 /// Block hash provider backed by Casper's indexed LMDB block store.
 #[derive(Clone, Debug)]
 pub struct IndexedLmdbBlockHashProvider {
-    block_store: Arc<IndexedLmdbBlockStore>,
+    block_store: Arc<LmdbBlockStore>,
 }
 
 impl IndexedLmdbBlockHashProvider {
     /// Creates a block hash provider backed by `block_store`.
-    pub fn new(block_store: Arc<IndexedLmdbBlockStore>) -> Self {
+    pub fn new(block_store: Arc<LmdbBlockStore>) -> Self {
         Self { block_store }
     }
 }

node/src/components/block_accumulator/tests.rs

@@ -187,10 +187,20 @@ impl Reactor for MockReactor {
         )
         .unwrap();
 
-        let storage = Storage::new(
+        let protocol_version = ProtocolVersion::from_parts(1, 0, 0);
+        let (storage_root, mut storage_block_store) =
+            storage::open_block_store(&storage_withdir, "test").unwrap();
+        storage::prune_block_store(
+            &mut storage_block_store,
+            chainspec.hard_reset_to_start_of_era(),
+            protocol_version,
+        )
+        .unwrap();
+        let mut storage = Storage::new(
             &storage_withdir,
-            None,
-            ProtocolVersion::from_parts(1, 0, 0),
+            storage_root,
+            storage_block_store,
+            protocol_version,
             EraId::default(),
             "test",
             chainspec.transaction_config.max_ttl.into(),
@@ -200,6 +210,7 @@ impl Reactor for MockReactor {
             TransactionConfig::default(),
         )
         .unwrap();
+        storage.initialize_for_test();
 
         let reactor = MockReactor {
             storage,

node/src/components/contract_runtime.rs

@@ -14,7 +14,6 @@ mod utils;
 
 use std::{
     cmp::Ordering,
-    collections::BTreeMap,
     convert::TryInto,
     fmt::{self, Debug, Formatter},
     path::Path,
@@ -33,7 +32,7 @@ use casper_storage::{
     data_access_layer::{
         AddressableEntityRequest, AddressableEntityResult, BlockStore, DataAccessLayer,
         EntryPointExistsRequest, ExecutionResultsChecksumRequest, FlushRequest, FlushResult,
-        GenesisRequest, GenesisResult, TrieRequest,
+        GenesisRequest, GenesisResult, ProtocolUpgradeRequest, ProtocolUpgradeResult, TrieRequest,
     },
     global_state::{
         state::{lmdb::LmdbGlobalState, CommitProvider, StateProvider},
@@ -44,13 +43,13 @@ use casper_storage::{
     tracking_copy::TrackingCopyError,
 };
 use casper_types::{
-    account::AccountHash, ActivationPoint, Chainspec, ChainspecRawBytes, ChainspecRegistry,
-    EntityAddr, EraId, Key, PublicKey,
+    account::AccountHash, ActivationPoint, Chainspec, ChainspecRawBytes, ChainspecRegistry, Digest,
+    EntityAddr, EraId, Key, ProtocolUpgradeConfig,
 };
 
 use crate::{
     components::{fetcher::FetchResponse, Component, ComponentState},
-    contract_runtime::{types::EraPrice, utils::handle_protocol_upgrade},
+    contract_runtime::types::EraPrice,
     effect::{
         announcements::{
             ContractRuntimeAnnouncement, FatalAnnouncement, MetaBlockAnnouncement,
@@ -62,10 +61,7 @@ use crate::{
     },
     fatal,
     protocol::Message,
-    types::{
-        BlockPayload, ExecutableBlock, FinalizedBlock, InternalEraReport, MetaBlockState,
-        TrieOrChunk, TrieOrChunkId,
-    },
+    types::{TrieOrChunk, TrieOrChunkId},
     NodeRng,
 };
 pub(crate) use config::Config;
@@ -307,6 +303,47 @@ impl ContractRuntime {
         result
     }
 
+    /// Commits a protocol upgrade against global state and flushes it to disk.
+    ///
+    /// The commit itself runs on the blocking thread-pool (via `run_intensive_task`), since it
+    /// can take a long time; this lets the caller bound the wait with a timeout instead of
+    /// stalling its task indefinitely.
+    pub(crate) async fn commit_protocol_upgrade(
+        &self,
+        upgrade_config: ProtocolUpgradeConfig,
+    ) -> Result<Digest, String> {
+        debug!(?upgrade_config, "upgrade");
+        let start = Instant::now();
+        let upgrade_request = ProtocolUpgradeRequest::new(upgrade_config);
+
+        let data_access_layer = Arc::clone(&self.data_access_layer);
+        let metrics = Arc::clone(&self.metrics);
+        run_intensive_task(move || {
+            let result = data_access_layer.protocol_upgrade(upgrade_request);
+            if result.is_success() {
+                info!("committed upgrade");
+                metrics
+                    .commit_upgrade
+                    .observe(start.elapsed().as_secs_f64());
+                let flush_req = FlushRequest::new();
+                if let FlushResult::Failure(err) = data_access_layer.flush(flush_req) {
+                    return Err(format!("{:?}", err));
+                }
+            }
+
+            match result {
+                ProtocolUpgradeResult::RootNotFound => {
+                    Err("Root not found for protocol upgrade".to_string())
+                }
+                ProtocolUpgradeResult::Failure(err) => Err(format!("{:?}", err)),
+                ProtocolUpgradeResult::Success {
+                    post_state_hash, ..
+                } => Ok(post_state_hash),
+            }
+        })
+        .await
+    }
+
     /// Handles a contract runtime request.
     fn handle_contract_runtime_request<REv>(
         &mut self,
@@ -546,91 +583,6 @@ impl ContractRuntime {
                 }
                 .ignore()
             }
-            ContractRuntimeRequest::UpdatePreState { new_pre_state } => {
-                let next_block_height = new_pre_state.next_block_height();
-                self.set_execution_pre_state(new_pre_state);
-                let current_price = self.current_gas_price.gas_price();
-                async move {
-                    let block_header = match effect_builder
-                        .get_highest_complete_block_header_from_storage()
-                        .await
-                    {
-                        Some(header)
-                            if header.is_switch_block()
-                                && (header.height() + 1 == next_block_height) =>
-                        {
-                            header
-                        }
-                        Some(_) => {
-                            return fatal!(
-                                effect_builder,
-                                "Latest complete block is not a switch block to update state"
-                            )
-                            .await;
-                        }
-                        None => {
-                            return fatal!(
-                                effect_builder,
-                                "No complete block header found to update post upgrade state"
-                            )
-                            .await;
-                        }
-                    };
-
-                    let payload = BlockPayload::new(
-                        BTreeMap::new(),
-                        vec![],
-                        Default::default(),
-                        false,
-                        current_price,
-                    );
-
-                    let finalized_block = FinalizedBlock::new(
-                        payload,
-                        Some(InternalEraReport::default()),
-                        block_header.timestamp(),
-                        block_header.next_block_era_id(),
-                        next_block_height,
-                        PublicKey::System,
-                    );
-
-                    info!("Enqueuing block for execution post state refresh");
-
-                    effect_builder
-                        .enqueue_block_for_execution(
-                            ExecutableBlock::from_finalized_block_and_transactions(
-                                finalized_block,
-                                vec![],
-                            ),
-                            MetaBlockState::new_not_to_be_gossiped(),
-                        )
-                        .await;
-                }
-                .ignore()
-            }
-            ContractRuntimeRequest::DoProtocolUpgrade {
-                protocol_upgrade_config,
-                next_block_height,
-                parent_hash,
-                parent_seed,
-            } => {
-                let mut effects = Effects::new();
-                let data_access_layer = Arc::clone(&self.data_access_layer);
-                let metrics = Arc::clone(&self.metrics);
-                effects.extend(
-                    handle_protocol_upgrade(
-                        effect_builder,
-                        data_access_layer,
-                        metrics,
-                        *protocol_upgrade_config,
-                        next_block_height,
-                        parent_hash,
-                        parent_seed,
-                    )
-                    .ignore(),
-                );
-                effects
-            }
             ContractRuntimeRequest::EnqueueBlockForExecution {
                 executable_block,
                 key_block_height_for_activation_point,
@@ -847,6 +799,10 @@ impl ContractRuntime {
     pub(crate) fn current_era_price(&self) -> EraPrice {
         self.current_gas_price
     }
+
+    pub(crate) fn current_gas_price(&self) -> u8 {
+        self.current_gas_price.gas_price()
+    }
 }
 
 impl<REv> Component<REv> for ContractRuntime

node/src/components/contract_runtime/tests.rs

@@ -124,9 +124,18 @@ impl reactor::Reactor for Reactor {
         }
 
         let storage_withdir = WithDir::new(storage_tempdir.path(), storage_config);
-        let storage = Storage::new(
+        let (storage_root, mut storage_block_store) =
+            storage::open_block_store(&storage_withdir, "test").unwrap();
+        storage::prune_block_store(
+            &mut storage_block_store,
+            chainspec.hard_reset_to_start_of_era(),
+            chainspec.protocol_version(),
+        )
+        .unwrap();
+        let mut storage = Storage::new(
             &storage_withdir,
-            None,
+            storage_root,
+            storage_block_store,
             chainspec.protocol_version(),
             EraId::default(),
             "test",
@@ -137,6 +146,7 @@ impl reactor::Reactor for Reactor {
             TransactionConfig::default(),
         )
         .unwrap();
+        storage.initialize_for_test();
 
         let contract_runtime =
             ContractRuntime::new(storage.root_path(), &config.config, chainspec, registry)?;

node/src/components/contract_runtime/utils.rs

@@ -10,7 +10,6 @@ use std::{
     fmt::Debug,
     ops::Range,
     sync::{Arc, Mutex},
-    time::Instant,
 };
 use tracing::{debug, error, info};
 
@@ -36,15 +35,10 @@ use crate::{
 use casper_binary_port::SpeculativeExecutionResult;
 use casper_execution_engine::engine_state::{ExecutionEngineV1, WasmV1Result};
 use casper_storage::{
-    data_access_layer::{
-        DataAccessLayer, FlushRequest, FlushResult, ProtocolUpgradeRequest, ProtocolUpgradeResult,
-        TransferResult,
-    },
-    global_state::state::{lmdb::LmdbGlobalState, CommitProvider, StateProvider},
-};
-use casper_types::{
-    BlockHash, Chainspec, Digest, EraId, Gas, Key, ProtocolUpgradeConfig, Transaction,
+    data_access_layer::{DataAccessLayer, TransferResult},
+    global_state::state::lmdb::LmdbGlobalState,
 };
+use casper_types::{BlockHash, Chainspec, EraId, Gas, Key, Transaction};
 
 /// Maximum number of resource intensive tasks that can be run in parallel.
 ///
@@ -493,75 +487,6 @@ pub(super) async fn exec_and_check_next<REv>(
     }
 }
 
-pub(super) async fn handle_protocol_upgrade<REv>(
-    effect_builder: EffectBuilder<REv>,
-    data_access_layer: Arc<DataAccessLayer<LmdbGlobalState>>,
-    metrics: Arc<Metrics>,
-    upgrade_config: ProtocolUpgradeConfig,
-    next_block_height: u64,
-    parent_hash: BlockHash,
-    parent_seed: Digest,
-) where
-    REv: From<ContractRuntimeRequest>
-        + From<ContractRuntimeAnnouncement>
-        + From<StorageRequest>
-        + From<MetaBlockAnnouncement>
-        + From<FatalAnnouncement>
-        + Send,
-{
-    debug!(?upgrade_config, "upgrade");
-    let start = Instant::now();
-    let upgrade_request = ProtocolUpgradeRequest::new(upgrade_config);
-
-    let result = run_intensive_task(move || {
-        let result = data_access_layer.protocol_upgrade(upgrade_request);
-        if result.is_success() {
-            info!("committed upgrade");
-            metrics
-                .commit_upgrade
-                .observe(start.elapsed().as_secs_f64());
-            let flush_req = FlushRequest::new();
-            if let FlushResult::Failure(err) = data_access_layer.flush(flush_req) {
-                return Err(format!("{:?}", err));
-            }
-        }
-
-        Ok(result)
-    })
-    .await;
-
-    match result {
-        Err(error_msg) => {
-            // The only way this happens is if there is a problem in the flushing.
-            error!(%error_msg, ":Error in post upgrade flush");
-            fatal!(effect_builder, "{}", error_msg).await;
-        }
-        Ok(result) => match result {
-            ProtocolUpgradeResult::RootNotFound => {
-                let error_msg = "Root not found for protocol upgrade";
-                fatal!(effect_builder, "{}", error_msg).await;
-            }
-            ProtocolUpgradeResult::Failure(err) => {
-                fatal!(effect_builder, "{:?}", err).await;
-            }
-            ProtocolUpgradeResult::Success {
-                post_state_hash, ..
-            } => {
-                let post_upgrade_state = ExecutionPreState::new(
-                    next_block_height,
-                    post_state_hash,
-                    parent_hash,
-                    parent_seed,
-                );
-
-                effect_builder
-                    .update_contract_runtime_state(post_upgrade_state)
-                    .await
-            }
-        },
-    }
-}
-
 fn generate_range_by_index(
     highest_era: u64,
     batch_size: u64,