chore(engine): Remove ConsistentDbView

Author: mediocregopher

Cargo.lock

@@ -30,7 +30,6 @@ reth-prune.workspace = true
 reth-revm.workspace = true
 reth-stages-api.workspace = true
 reth-tasks.workspace = true
-reth-trie-db.workspace = true
 reth-trie-parallel.workspace = true
 reth-trie-sparse = { workspace = true, features = ["std", "metrics"] }
 reth-trie-sparse-parallel = { workspace = true, features = ["std"] }
@@ -133,7 +132,6 @@ test-utils = [
     "reth-trie/test-utils",
     "reth-trie-sparse/test-utils",
     "reth-prune-types?/test-utils",
-    "reth-trie-db/test-utils",
     "reth-trie-parallel/test-utils",
     "reth-ethereum-primitives/test-utils",
     "reth-node-ethereum/test-utils",

crates/engine/tree/Cargo.toml

@@ -20,11 +20,10 @@ use reth_evm::OnStateHook;
 use reth_evm_ethereum::EthEvmConfig;
 use reth_primitives_traits::{Account as RethAccount, Recovered, StorageEntry};
 use reth_provider::{
-    providers::{BlockchainProvider, ConsistentDbView},
+    providers::{BlockchainProvider, OverlayStateProviderFactory},
     test_utils::{create_test_provider_factory_with_chain_spec, MockNodeTypesWithDB},
     AccountReader, ChainSpecProvider, HashingWriter, ProviderFactory,
 };
-use reth_trie::TrieInput;
 use revm_primitives::{HashMap, U256};
 use revm_state::{Account as RevmAccount, AccountInfo, AccountStatus, EvmState, EvmStorageSlot};
 use std::{hint::black_box, sync::Arc};
@@ -238,8 +237,7 @@ fn bench_state_root(c: &mut Criterion) {
                                         >,
                                     >(),
                                     StateProviderBuilder::new(provider.clone(), genesis_hash, None),
-                                    ConsistentDbView::new_with_latest_tip(provider).unwrap(),
-                                    TrieInput::default(),
+                                    OverlayStateProviderFactory::new(provider),
                                     &TreeConfig::default(),
                                 )
                                 .map_err(|(err, ..)| err)

crates/engine/tree/benches/state_root_task.rs

@@ -29,9 +29,8 @@ use reth_payload_primitives::{
 };
 use reth_primitives_traits::{NodePrimitives, RecoveredBlock, SealedBlock, SealedHeader};
 use reth_provider::{
-    providers::ConsistentDbView, BlockReader, DatabaseProviderFactory, HashedPostStateProvider,
-    ProviderError, StateProviderBox, StateProviderFactory, StateReader, TransactionVariant,
-    TrieReader,
+    BlockReader, DatabaseProviderFactory, HashedPostStateProvider, ProviderError, StateProviderBox,
+    StateProviderFactory, StateReader, TransactionVariant, TrieReader,
 };
 use reth_revm::database::StateProviderDatabase;
 use reth_stages_api::ControlFlow;

crates/engine/tree/src/tree/mod.rs

@@ -25,12 +25,12 @@ use reth_evm::{
     ConfigureEvm, EvmEnvFor, OnStateHook, SpecFor, TxEnvFor,
 };
 use reth_primitives_traits::NodePrimitives;
-use reth_provider::{
-    providers::ConsistentDbView, BlockReader, DatabaseProviderFactory, StateProviderFactory,
-    StateReader,
-};
+use reth_provider::{BlockReader, DatabaseProviderROFactory, StateProviderFactory, StateReader};
 use reth_revm::{db::BundleState, state::EvmState};
-use reth_trie::TrieInput;
+use reth_trie::{
+    hashed_cursor::HashedCursorFactory, prefix_set::TriePrefixSetsMut,
+    trie_cursor::TrieCursorFactory,
+};
 use reth_trie_parallel::{
     proof_task::{ProofTaskCtx, ProofWorkerHandle},
     root::ParallelStateRootError,
@@ -94,8 +94,6 @@ where
     >,
     /// Whether to disable the parallel sparse trie.
     disable_parallel_sparse_trie: bool,
-    /// A cleared trie input, kept around to be reused so allocations can be minimized.
-    trie_input: Option<TrieInput>,
     /// Maximum concurrency for prewarm task.
     prewarm_max_concurrency: usize,
 }
@@ -122,7 +120,6 @@ where
             precompile_cache_disabled: config.precompile_cache_disabled(),
             precompile_cache_map,
             sparse_state_trie: Arc::default(),
-            trie_input: None,
             disable_parallel_sparse_trie: config.disable_parallel_sparse_trie(),
             prewarm_max_concurrency: config.prewarm_max_concurrency(),
         }
@@ -167,43 +164,39 @@ where
     /// This returns a handle to await the final state root and to interact with the tasks (e.g.
     /// canceling)
     #[allow(clippy::type_complexity)]
-    pub fn spawn<P, I: ExecutableTxIterator<Evm>>(
+    pub fn spawn<P, F, I: ExecutableTxIterator<Evm>>(
         &mut self,
         env: ExecutionEnv<Evm>,
         transactions: I,
         provider_builder: StateProviderBuilder<N, P>,
-        consistent_view: ConsistentDbView<P>,
-        trie_input: TrieInput,
+        multiproof_provider_factory: F,
         config: &TreeConfig,
     ) -> Result<
         PayloadHandle<WithTxEnv<TxEnvFor<Evm>, I::Tx>, I::Error>,
         (ParallelStateRootError, I, ExecutionEnv<Evm>, StateProviderBuilder<N, P>),
     >
     where
-        P: DatabaseProviderFactory<Provider: BlockReader>
-            + BlockReader
-            + StateProviderFactory
-            + StateReader
+        P: BlockReader + StateProviderFactory + StateReader + Clone + 'static,
+        F: DatabaseProviderROFactory<Provider: TrieCursorFactory + HashedCursorFactory>
             + Clone
+            + Send
             + 'static,
     {
         let (to_sparse_trie, sparse_trie_rx) = channel();
-        // spawn multiproof task, save the trie input
-        let (trie_input, state_root_config) = MultiProofConfig::from_input(trie_input);
-        self.trie_input = Some(trie_input);
+
+        // We rely on the cursor factory to provide whatever DB overlay is necessary to see a
+        // consistent view of the database, including the trie tables. Because of this there is no
+        // need for an overarching prefix set to invalidate any section of the trie tables, and so
+        // we use an empty prefix set.
+        let prefix_sets = Arc::new(TriePrefixSetsMut::default());
 
         // Create and spawn the storage proof task
-        let task_ctx = ProofTaskCtx::new(
-            state_root_config.nodes_sorted.clone(),
-            state_root_config.state_sorted.clone(),
-            state_root_config.prefix_sets.clone(),
-        );
+        let task_ctx = ProofTaskCtx::new(multiproof_provider_factory, prefix_sets);
         let storage_worker_count = config.storage_worker_count();
         let account_worker_count = config.account_worker_count();
         let max_proof_task_concurrency = config.max_proof_task_concurrency() as usize;
         let proof_handle = ProofWorkerHandle::new(
             self.executor.handle().clone(),
-            consistent_view,
             task_ctx,
             storage_worker_count,
             account_worker_count,
@@ -213,7 +206,6 @@ where
         // spawn one Tokio task for the account proof, and one Tokio task for the storage proof.
         let max_multi_proof_task_concurrency = max_proof_task_concurrency / 2;
         let multi_proof_task = MultiProofTask::new(
-            state_root_config,
             self.executor.clone(),
             proof_handle.clone(),
             to_sparse_trie,
@@ -361,11 +353,6 @@ where
         CacheTaskHandle { cache, to_prewarm_task: Some(to_prewarm_task), cache_metrics }
     }
 
-    /// Takes the trie input from the inner payload processor, if it exists.
-    pub const fn take_trie_input(&mut self) -> Option<TrieInput> {
-        self.trie_input.take()
-    }
-
     /// Returns the cache for the given parent hash.
     ///
     /// If the given hash is different then what is recently cached, then this will create a new
@@ -663,12 +650,12 @@ mod tests {
     use reth_evm_ethereum::EthEvmConfig;
     use reth_primitives_traits::{Account, Recovered, StorageEntry};
     use reth_provider::{
-        providers::{BlockchainProvider, ConsistentDbView},
+        providers::{BlockchainProvider, OverlayStateProviderFactory},
         test_utils::create_test_provider_factory_with_chain_spec,
         ChainSpecProvider, HashingWriter,
     };
     use reth_testing_utils::generators;
-    use reth_trie::{test_utils::state_root, HashedPostState, TrieInput};
+    use reth_trie::{test_utils::state_root, HashedPostState};
     use revm_primitives::{Address, HashMap, B256, KECCAK_EMPTY, U256};
     use revm_state::{AccountInfo, AccountStatus, EvmState, EvmStorageSlot};
     use std::sync::Arc;
@@ -850,17 +837,18 @@ mod tests {
             &TreeConfig::default(),
             PrecompileCacheMap::default(),
         );
-        let provider = BlockchainProvider::new(factory).unwrap();
+
+        let provider_factory = BlockchainProvider::new(factory).unwrap();
+
         let mut handle =
             payload_processor
                 .spawn(
                     Default::default(),
                     core::iter::empty::<
                         Result<Recovered<TransactionSigned>, core::convert::Infallible>,
                     >(),
-                    StateProviderBuilder::new(provider.clone(), genesis_hash, None),
-                    ConsistentDbView::new_with_latest_tip(provider).unwrap(),
-                    TrieInput::from_state(hashed_state),
+                    StateProviderBuilder::new(provider_factory.clone(), genesis_hash, None),
+                    OverlayStateProviderFactory::new(provider_factory),
                     &TreeConfig::default(),
                 )
                 .map_err(|(err, ..)| err)

crates/engine/tree/src/tree/payload_processor/mod.rs

@@ -64,8 +64,8 @@ impl SparseTrieUpdate {
 }
 
 /// Common configuration for multi proof tasks
-#[derive(Debug, Clone)]
-pub(super) struct MultiProofConfig {
+#[derive(Debug, Clone, Default)]
+pub(crate) struct MultiProofConfig {
     /// The sorted collection of cached in-memory intermediate trie nodes that
     /// can be reused for computation.
     pub nodes_sorted: Arc<TrieUpdatesSorted>,
@@ -82,7 +82,7 @@ impl MultiProofConfig {
     ///
     /// This returns a cleared [`TrieInput`] so that we can reuse any allocated space in the
     /// [`TrieInput`].
-    pub(super) fn from_input(mut input: TrieInput) -> (TrieInput, Self) {
+    pub(crate) fn from_input(mut input: TrieInput) -> (TrieInput, Self) {
         let config = Self {
             nodes_sorted: Arc::new(input.nodes.drain_into_sorted()),
             state_sorted: Arc::new(input.state.drain_into_sorted()),
@@ -290,7 +290,6 @@ impl From<MultiproofInput> for PendingMultiproofTask {
 /// Input parameters for spawning a dedicated storage multiproof calculation.
 #[derive(Debug)]
 struct StorageMultiproofInput {
-    config: MultiProofConfig,
     source: Option<StateChangeSource>,
     hashed_state_update: HashedPostState,
     hashed_address: B256,
@@ -313,7 +312,6 @@ impl StorageMultiproofInput {
 /// Input parameters for spawning a multiproof calculation.
 #[derive(Debug)]
 struct MultiproofInput {
-    config: MultiProofConfig,
     source: Option<StateChangeSource>,
     hashed_state_update: HashedPostState,
     proof_targets: MultiProofTargets,
@@ -437,7 +435,6 @@ impl MultiproofManager {
     /// Spawns a single storage proof calculation task.
     fn spawn_storage_proof(&mut self, storage_multiproof_input: StorageMultiproofInput) {
         let StorageMultiproofInput {
-            config,
             source,
             hashed_state_update,
             hashed_address,
@@ -462,9 +459,7 @@ impl MultiproofManager {
             );
             let start = Instant::now();
             let proof_result = ParallelProof::new(
-                config.nodes_sorted,
-                config.state_sorted,
-                config.prefix_sets,
+                Default::default(), // prefix sets
                 missed_leaves_storage_roots,
                 storage_proof_worker_handle,
             )
@@ -511,7 +506,6 @@ impl MultiproofManager {
     /// Spawns a single multiproof calculation task.
     fn spawn_multiproof(&mut self, multiproof_input: MultiproofInput) {
         let MultiproofInput {
-            config,
             source,
             hashed_state_update,
             proof_targets,
@@ -540,7 +534,7 @@ impl MultiproofManager {
 
             // Extend prefix sets with targets
             let frozen_prefix_sets =
-                ParallelProof::extend_prefix_sets_with_targets(&config.prefix_sets, &proof_targets);
+                ParallelProof::extend_prefix_sets_with_targets(&Default::default(), &proof_targets);
 
             // Queue account multiproof to worker pool
             let input = AccountMultiproofInput {
@@ -658,8 +652,6 @@ pub(super) struct MultiProofTask {
     ///
     /// If [`None`], then chunking is disabled.
     chunk_size: Option<usize>,
-    /// Task configuration.
-    config: MultiProofConfig,
     /// Receiver for state root related messages.
     rx: Receiver<MultiProofMessage>,
     /// Sender for state root related messages.
@@ -681,7 +673,6 @@ pub(super) struct MultiProofTask {
 impl MultiProofTask {
     /// Creates a new multi proof task with the unified message channel
     pub(super) fn new(
-        config: MultiProofConfig,
         executor: WorkloadExecutor,
         proof_worker_handle: ProofWorkerHandle,
         to_sparse_trie: Sender<SparseTrieUpdate>,
@@ -693,7 +684,6 @@ impl MultiProofTask {
 
         Self {
             chunk_size,
-            config,
             rx,
             tx,
             to_sparse_trie,
@@ -742,7 +732,6 @@ impl MultiProofTask {
         let mut spawn = |proof_targets| {
             self.multiproof_manager.spawn_or_queue(
                 MultiproofInput {
-                    config: self.config.clone(),
                     source: None,
                     hashed_state_update: Default::default(),
                     proof_targets,
@@ -885,7 +874,6 @@ impl MultiProofTask {
 
             self.multiproof_manager.spawn_or_queue(
                 MultiproofInput {
-                    config: self.config.clone(),
                     source: Some(source),
                     hashed_state_update,
                     proof_targets,
@@ -1208,30 +1196,27 @@ mod tests {
     use super::*;
     use alloy_primitives::map::B256Set;
     use reth_provider::{
-        providers::ConsistentDbView, test_utils::create_test_provider_factory, BlockReader,
-        DatabaseProviderFactory,
+        providers::OverlayStateProviderFactory, test_utils::create_test_provider_factory,
+        BlockReader, DatabaseProviderFactory, StageCheckpointReader, TrieReader,
     };
-    use reth_trie::{MultiProof, TrieInput};
+    use reth_trie::MultiProof;
     use reth_trie_parallel::proof_task::{ProofTaskCtx, ProofWorkerHandle};
     use revm_primitives::{B256, U256};
 
     fn create_test_state_root_task<F>(factory: F) -> MultiProofTask
     where
-        F: DatabaseProviderFactory<Provider: BlockReader> + Clone + 'static,
+        F: DatabaseProviderFactory<Provider: BlockReader + TrieReader + StageCheckpointReader>
+            + Clone
+            + Send
+            + 'static,
     {
         let executor = WorkloadExecutor::default();
-        let (_trie_input, config) = MultiProofConfig::from_input(TrieInput::default());
-        let task_ctx = ProofTaskCtx::new(
-            config.nodes_sorted.clone(),
-            config.state_sorted.clone(),
-            config.prefix_sets.clone(),
-        );
-        let consistent_view = ConsistentDbView::new(factory, None);
-        let proof_handle =
-            ProofWorkerHandle::new(executor.handle().clone(), consistent_view, task_ctx, 1, 1);
+        let overlay_factory = OverlayStateProviderFactory::new(factory);
+        let task_ctx = ProofTaskCtx::new(overlay_factory, Default::default());
+        let proof_handle = ProofWorkerHandle::new(executor.handle().clone(), task_ctx, 1, 1);
         let channel = channel();
 
-        MultiProofTask::new(config, executor, proof_handle, channel.0, 1, None)
+        MultiProofTask::new(executor, proof_handle, channel.0, 1, None)
     }
 
     #[test]