refactor: move init_db to reth_db and add DatabaseEnv (paradigm…

…xyz#3450)

bin/reth/src/chain/import.rs

@@ -11,14 +11,14 @@ use reth_provider::{ProviderFactory, StageCheckpointReader};
 
 use crate::args::utils::genesis_value_parser;
 use reth_config::Config;
-use reth_db::database::Database;
+use reth_db::{database::Database, init_db};
 use reth_downloaders::{
     bodies::bodies::BodiesDownloaderBuilder,
     headers::reverse_headers::ReverseHeadersDownloaderBuilder, test_utils::FileClient,
 };
 use reth_interfaces::consensus::Consensus;
 use reth_primitives::{stage::StageId, ChainSpec, H256};
-use reth_staged_sync::utils::init::{init_db, init_genesis};
+use reth_staged_sync::utils::init::init_genesis;
 use reth_stages::{
     prelude::*,
     stages::{

bin/reth/src/chain/init.rs

@@ -3,8 +3,9 @@ use crate::{
     dirs::{DataDirPath, MaybePlatformPath},
 };
 use clap::Parser;
+use reth_db::init_db;
 use reth_primitives::ChainSpec;
-use reth_staged_sync::utils::init::{init_db, init_genesis};
+use reth_staged_sync::utils::init::init_genesis;
 use std::sync::Arc;
 use tracing::info;
 

bin/reth/src/db/mod.rs

@@ -12,7 +12,7 @@ use reth_db::{
     database::Database,
     mdbx::{Env, NoWriteMap, WriteMap},
     version::{get_db_version, DatabaseVersionError, DB_VERSION},
-    Tables,
+    DatabaseEnv, Tables,
 };
 use reth_primitives::ChainSpec;
 use std::{path::Path, sync::Arc};
@@ -178,7 +178,7 @@ fn read_only_db(path: &Path) -> eyre::Result<Env<NoWriteMap>> {
         .with_context(|| format!("Could not open database at path: {}", path.display()))
 }
 
-fn read_write_db(path: &Path) -> eyre::Result<Env<WriteMap>> {
+fn read_write_db(path: &Path) -> eyre::Result<DatabaseEnv> {
     Env::<WriteMap>::open(path, reth_db::mdbx::EnvKind::RW)
         .with_context(|| format!("Could not open database at path: {}", path.display()))
 }

bin/reth/src/debug_cmd/execution.rs

@@ -10,10 +10,7 @@ use clap::Parser;
 use futures::{stream::select as stream_select, StreamExt};
 use reth_beacon_consensus::BeaconConsensus;
 use reth_config::Config;
-use reth_db::{
-    database::Database,
-    mdbx::{Env, WriteMap},
-};
+use reth_db::{database::Database, init_db, DatabaseEnv};
 use reth_discv4::DEFAULT_DISCOVERY_PORT;
 use reth_downloaders::{
     bodies::bodies::BodiesDownloaderBuilder,
@@ -27,7 +24,7 @@ use reth_network::NetworkHandle;
 use reth_network_api::NetworkInfo;
 use reth_primitives::{stage::StageId, BlockHashOrNumber, BlockNumber, ChainSpec, H256};
 use reth_provider::{BlockExecutionWriter, ProviderFactory, StageCheckpointReader};
-use reth_staged_sync::utils::init::{init_db, init_genesis};
+use reth_staged_sync::utils::init::init_genesis;
 use reth_stages::{
     sets::DefaultStages,
     stages::{
@@ -153,7 +150,7 @@ impl Command {
         &self,
         config: &Config,
         task_executor: TaskExecutor,
-        db: Arc<Env<WriteMap>>,
+        db: Arc<DatabaseEnv>,
         network_secret_path: PathBuf,
         default_peers_path: PathBuf,
     ) -> eyre::Result<NetworkHandle> {

bin/reth/src/debug_cmd/merkle.rs

@@ -4,13 +4,12 @@ use crate::{
     dirs::{DataDirPath, MaybePlatformPath},
 };
 use clap::Parser;
-use reth_db::{cursor::DbCursorRO, tables, transaction::DbTx};
+use reth_db::{cursor::DbCursorRO, init_db, tables, transaction::DbTx};
 use reth_primitives::{
     stage::{StageCheckpoint, StageId},
     ChainSpec,
 };
 use reth_provider::{ProviderFactory, StageCheckpointReader};
-use reth_staged_sync::utils::init::init_db;
 use reth_stages::{
     stages::{
         AccountHashingStage, ExecutionStage, ExecutionStageThresholds, MerkleStage,

bin/reth/src/node/mod.rs

@@ -20,10 +20,7 @@ use reth_blockchain_tree::{
     config::BlockchainTreeConfig, externals::TreeExternals, BlockchainTree, ShareableBlockchainTree,
 };
 use reth_config::Config;
-use reth_db::{
-    database::Database,
-    mdbx::{Env, WriteMap},
-};
+use reth_db::{database::Database, init_db, DatabaseEnv};
 use reth_discv4::DEFAULT_DISCOVERY_PORT;
 use reth_downloaders::{
     bodies::bodies::BodiesDownloaderBuilder,
@@ -47,7 +44,7 @@ use reth_provider::{
 use reth_revm::Factory;
 use reth_revm_inspectors::stack::Hook;
 use reth_rpc_engine_api::EngineApi;
-use reth_staged_sync::utils::init::{init_db, init_genesis};
+use reth_staged_sync::utils::init::init_genesis;
 use reth_stages::{
     prelude::*,
     stages::{
@@ -478,7 +475,7 @@ impl Command {
         }
     }
 
-    async fn start_metrics_endpoint(&self, db: Arc<Env<WriteMap>>) -> eyre::Result<()> {
+    async fn start_metrics_endpoint(&self, db: Arc<DatabaseEnv>) -> eyre::Result<()> {
         if let Some(listen_addr) = self.metrics {
             info!(target: "reth::cli", addr = %listen_addr, "Starting metrics endpoint");
             prometheus_exporter::initialize(listen_addr, db, metrics_process::Collector::default())
@@ -519,7 +516,7 @@ impl Command {
         Ok(handle)
     }
 
-    fn lookup_head(&self, db: Arc<Env<WriteMap>>) -> Result<Head, reth_interfaces::Error> {
+    fn lookup_head(&self, db: Arc<DatabaseEnv>) -> Result<Head, reth_interfaces::Error> {
         let factory = ProviderFactory::new(db, self.chain.clone());
         let provider = factory.provider()?;
 
@@ -604,12 +601,12 @@ impl Command {
     fn load_network_config(
         &self,
         config: &Config,
-        db: Arc<Env<WriteMap>>,
+        db: Arc<DatabaseEnv>,
         executor: TaskExecutor,
         head: Head,
         secret_key: SecretKey,
         default_peers_path: PathBuf,
-    ) -> NetworkConfig<ProviderFactory<Arc<Env<WriteMap>>>> {
+    ) -> NetworkConfig<ProviderFactory<Arc<DatabaseEnv>>> {
         self.network
             .network_config(config, self.chain.clone(), secret_key, default_peers_path)
             .with_task_executor(Box::new(executor))

bin/reth/src/prometheus_exporter.rs

@@ -6,11 +6,7 @@ use hyper::{
 };
 use metrics_exporter_prometheus::{PrometheusBuilder, PrometheusHandle};
 use metrics_util::layers::{PrefixLayer, Stack};
-use reth_db::{
-    database::Database,
-    mdbx::{Env, WriteMap},
-    tables,
-};
+use reth_db::{database::Database, tables, DatabaseEnv};
 use reth_metrics::metrics::{self, absolute_counter, describe_counter, Unit};
 use std::{convert::Infallible, net::SocketAddr, sync::Arc};
 
@@ -73,7 +69,7 @@ async fn start_endpoint<F: Hook + 'static>(
 /// metrics.
 pub(crate) async fn initialize(
     listen_addr: SocketAddr,
-    db: Arc<Env<WriteMap>>,
+    db: Arc<DatabaseEnv>,
     process: metrics_process::Collector,
 ) -> eyre::Result<()> {
     let db_stats = move || {

bin/reth/src/stage/drop.rs

@@ -10,6 +10,7 @@ use reth_db::{
     mdbx::{Env, WriteMap},
     tables,
     transaction::DbTxMut,
+    DatabaseEnv,
 };
 use reth_primitives::{stage::StageId, ChainSpec};
 use reth_staged_sync::utils::init::{insert_genesis_header, insert_genesis_state};
@@ -70,7 +71,7 @@ impl Command {
                     tx.clear::<tables::BlockOmmers>()?;
                     tx.clear::<tables::BlockWithdrawals>()?;
                     tx.put::<tables::SyncStage>(StageId::Bodies.to_string(), Default::default())?;
-                    insert_genesis_header::<Env<WriteMap>>(tx, self.chain)?;
+                    insert_genesis_header::<DatabaseEnv>(tx, self.chain)?;
                 }
                 StageEnum::Senders => {
                     tx.clear::<tables::TxSenders>()?;
@@ -90,7 +91,7 @@ impl Command {
                         StageId::Execution.to_string(),
                         Default::default(),
                     )?;
-                    insert_genesis_state::<Env<WriteMap>>(tx, self.chain.genesis())?;
+                    insert_genesis_state::<DatabaseEnv>(tx, self.chain.genesis())?;
                 }
                 StageEnum::AccountHashing => {
                     tx.clear::<tables::HashedAccount>()?;
@@ -155,7 +156,7 @@ impl Command {
                         StageId::TotalDifficulty.to_string(),
                         Default::default(),
                     )?;
-                    insert_genesis_header::<Env<WriteMap>>(tx, self.chain)?;
+                    insert_genesis_header::<DatabaseEnv>(tx, self.chain)?;
                 }
                 _ => {
                     info!("Nothing to do for stage {:?}", self.stage);

bin/reth/src/stage/dump/mod.rs

@@ -5,10 +5,10 @@ use crate::{
 };
 use clap::Parser;
 use reth_db::{
-    cursor::DbCursorRO, database::Database, table::TableImporter, tables, transaction::DbTx,
+    cursor::DbCursorRO, database::Database, init_db, table::TableImporter, tables,
+    transaction::DbTx,
 };
 use reth_primitives::ChainSpec;
-use reth_staged_sync::utils::init::init_db;
 use std::{path::PathBuf, sync::Arc};
 use tracing::info;
 

bin/reth/src/stage/run.rs

@@ -10,10 +10,10 @@ use crate::{
 use clap::Parser;
 use reth_beacon_consensus::BeaconConsensus;
 use reth_config::Config;
+use reth_db::init_db;
 use reth_downloaders::bodies::bodies::BodiesDownloaderBuilder;
 use reth_primitives::ChainSpec;
 use reth_provider::{ProviderFactory, StageCheckpointReader};
-use reth_staged_sync::utils::init::init_db;
 use reth_stages::{
     stages::{
         AccountHashingStage, BodyStage, ExecutionStage, ExecutionStageThresholds,

crates/blockchain-tree/src/blockchain_tree.rs

@@ -1085,10 +1085,7 @@ mod tests {
     use crate::block_buffer::BufferedBlocks;
     use assert_matches::assert_matches;
     use linked_hash_set::LinkedHashSet;
-    use reth_db::{
-        mdbx::{test_utils::create_test_rw_db, Env, WriteMap},
-        transaction::DbTxMut,
-    };
+    use reth_db::{mdbx::test_utils::create_test_rw_db, transaction::DbTxMut, DatabaseEnv};
     use reth_interfaces::test_utils::TestConsensus;
     use reth_primitives::{
         proofs::EMPTY_ROOT, stage::StageCheckpoint, ChainSpecBuilder, H256, MAINNET,
@@ -1102,7 +1099,7 @@ mod tests {
 
     fn setup_externals(
         exec_res: Vec<PostState>,
-    ) -> TreeExternals<Arc<Env<WriteMap>>, Arc<TestConsensus>, TestExecutorFactory> {
+    ) -> TreeExternals<Arc<DatabaseEnv>, Arc<TestConsensus>, TestExecutorFactory> {
         let db = create_test_rw_db();
         let consensus = Arc::new(TestConsensus::default());
         let chain_spec = Arc::new(

crates/consensus/beacon/src/engine/mod.rs

@@ -1361,7 +1361,7 @@ mod tests {
         config::BlockchainTreeConfig, externals::TreeExternals, post_state::PostState,
         BlockchainTree, ShareableBlockchainTree,
     };
-    use reth_db::mdbx::{test_utils::create_test_rw_db, Env, WriteMap};
+    use reth_db::{mdbx::test_utils::create_test_rw_db, DatabaseEnv};
     use reth_interfaces::{
         sync::NoopSyncStateUpdater,
         test_utils::{NoopFullBlockClient, TestConsensus},
@@ -1381,10 +1381,10 @@ mod tests {
     };
 
     type TestBeaconConsensusEngine = BeaconConsensusEngine<
-        Arc<Env<WriteMap>>,
+        Arc<DatabaseEnv>,
         BlockchainProvider<
-            Arc<Env<WriteMap>>,
-            ShareableBlockchainTree<Arc<Env<WriteMap>>, TestConsensus, TestExecutorFactory>,
+            Arc<DatabaseEnv>,
+            ShareableBlockchainTree<Arc<DatabaseEnv>, TestConsensus, TestExecutorFactory>,
         >,
         NoopFullBlockClient,
     >;
@@ -1498,7 +1498,7 @@ mod tests {
         }
 
         /// Builds the test consensus engine into a `TestConsensusEngine` and `TestEnv`.
-        fn build(self) -> (TestBeaconConsensusEngine, TestEnv<Arc<Env<WriteMap>>>) {
+        fn build(self) -> (TestBeaconConsensusEngine, TestEnv<Arc<DatabaseEnv>>) {
             reth_tracing::init_test_tracing();
             let db = create_test_rw_db();
             let consensus = TestConsensus::default();

crates/interfaces/src/blockchain_tree/error.rs

@@ -272,6 +272,7 @@ impl From<crate::Error> for InsertBlockErrorKind {
             Error::Database(err) => InsertBlockErrorKind::Internal(Box::new(err)),
             Error::Provider(err) => InsertBlockErrorKind::Internal(Box::new(err)),
             Error::Network(err) => InsertBlockErrorKind::Internal(Box::new(err)),
+            Error::Custom(err) => InsertBlockErrorKind::Internal(err.into()),
         }
     }
 }

crates/interfaces/src/error.rs

@@ -19,4 +19,7 @@ pub enum Error {
 
     #[error(transparent)]
     Network(#[from] reth_network_api::NetworkError),
+
+    #[error("{0}")]
+    Custom(std::string::String),
 }

crates/net/downloaders/src/bodies/test_utils.rs

@@ -5,6 +5,7 @@ use reth_db::{
     mdbx::{Env, WriteMap},
     tables,
     transaction::DbTxMut,
+    DatabaseEnv,
 };
 use reth_interfaces::{db, p2p::bodies::response::BlockResponse};
 use reth_primitives::{Block, BlockBody, SealedBlock, SealedHeader, H256};
@@ -46,7 +47,7 @@ pub(crate) fn create_raw_bodies<'a>(
 }
 
 #[inline]
-pub(crate) fn insert_headers(db: &Env<WriteMap>, headers: &[SealedHeader]) {
+pub(crate) fn insert_headers(db: &DatabaseEnv, headers: &[SealedHeader]) {
     db.update(|tx| -> Result<(), db::DatabaseError> {
         for header in headers {
             tx.put::<tables::CanonicalHeaders>(header.number, header.hash())?;