feat(bin): simple reth node command (#158)

* feat(bin): simple reth node command

* use silent flag

* small chore

crates/db/src/kv/mod.rs

@@ -110,6 +110,7 @@ impl<E: EnvironmentKind> Deref for Env<E> {
 #[cfg(any(test, feature = "test-utils"))]
 pub mod test_utils {
     use super::{Env, EnvKind, EnvironmentKind, Path};
+    use std::sync::Arc;
 
     /// Error during database creation
     pub const ERROR_DB_CREATION: &str = "Not able to create the mdbx file.";
@@ -119,8 +120,11 @@ pub mod test_utils {
     pub const ERROR_TEMPDIR: &str = "Not able to create a temporary directory.";
 
     /// Create database for testing
-    pub fn create_test_db<E: EnvironmentKind>(kind: EnvKind) -> Env<E> {
-        create_test_db_with_path(kind, &tempfile::TempDir::new().expect(ERROR_TEMPDIR).into_path())
+    pub fn create_test_db<E: EnvironmentKind>(kind: EnvKind) -> Arc<Env<E>> {
+        Arc::new(create_test_db_with_path(
+            kind,
+            &tempfile::TempDir::new().expect(ERROR_TEMPDIR).into_path(),
+        ))
     }
 
     /// Create database for testing with specified path
@@ -141,7 +145,7 @@ mod tests {
     };
     use reth_libmdbx::{NoWriteMap, WriteMap};
     use reth_primitives::{Account, Address, Header, IntegerList, StorageEntry, H256, U256};
-    use std::str::FromStr;
+    use std::{str::FromStr, sync::Arc};
     use tempfile::TempDir;
 
     const ERROR_DB_CREATION: &str = "Not able to create the mdbx file.";
@@ -282,7 +286,7 @@ mod tests {
 
     #[test]
     fn db_sharded_key() {
-        let db: Env<WriteMap> = test_utils::create_test_db(EnvKind::RW);
+        let db: Arc<Env<WriteMap>> = test_utils::create_test_db(EnvKind::RW);
         let real_key = Address::from_str("0xa2c122be93b0074270ebee7f6b7292c7deb45047").unwrap();
 
         for i in 1..5 {

crates/stages/src/pipeline.rs

@@ -4,7 +4,10 @@ use crate::{
 };
 use reth_interfaces::db::{DBContainer, Database, DbTx};
 use reth_primitives::BlockNumber;
-use std::fmt::{Debug, Formatter};
+use std::{
+    fmt::{Debug, Formatter},
+    sync::Arc,
+};
 use tokio::sync::mpsc::Sender;
 use tracing::*;
 
@@ -139,7 +142,7 @@ impl<DB: Database> Pipeline<DB> {
 
     /// Run the pipeline in an infinite loop. Will terminate early if the user has specified
     /// a `max_block` in the pipeline.
-    pub async fn run(&mut self, db: &DB) -> Result<(), PipelineError> {
+    pub async fn run(&mut self, db: Arc<DB>) -> Result<(), PipelineError> {
         loop {
             let mut state = PipelineState {
                 events_sender: self.events_sender.clone(),
@@ -148,7 +151,7 @@ impl<DB: Database> Pipeline<DB> {
                 minimum_progress: None,
                 reached_tip: true,
             };
-            let next_action = self.run_loop(&mut state, db).await?;
+            let next_action = self.run_loop(&mut state, db.as_ref()).await?;
 
             // Terminate the loop early if it's reached the maximum user
             // configured block.
@@ -396,7 +399,7 @@ mod tests {
                     false,
                 )
                 .set_max_block(Some(10))
-                .run(&db)
+                .run(db)
                 .await
         });
 
@@ -450,7 +453,7 @@ mod tests {
                 .set_max_block(Some(10));
 
             // Sync first
-            pipeline.run(&db).await.expect("Could not run pipeline");
+            pipeline.run(db.clone()).await.expect("Could not run pipeline");
 
             // Unwind
             pipeline.set_channel(tx).unwind(&db, 1, None).await.expect("Could not unwind pipeline");
@@ -528,7 +531,7 @@ mod tests {
                 )
                 .set_max_block(Some(10))
                 .set_channel(tx)
-                .run(&db)
+                .run(db)
                 .await
                 .expect("Could not run pipeline");
         });
@@ -620,7 +623,7 @@ mod tests {
                 .set_max_block(Some(10));
 
             // Sync first
-            pipeline.run(&db).await.expect("Could not run pipeline");
+            pipeline.run(db.clone()).await.expect("Could not run pipeline");
 
             // Unwind
             pipeline.set_channel(tx).unwind(&db, 1, None).await.expect("Could not unwind pipeline");
@@ -690,7 +693,7 @@ mod tests {
                     true,
                 )
                 .set_max_block(Some(10))
-                .run(&db)
+                .run(db)
                 .await
         });
 

crates/stages/src/util.rs

@@ -145,7 +145,7 @@ pub(crate) mod test_utils {
     impl Default for StageTestDB {
         /// Create a new instance of [StageTestDB]
         fn default() -> Self {
-            Self { db: Arc::new(create_test_db::<WriteMap>(EnvKind::RW)) }
+            Self { db: create_test_db::<WriteMap>(EnvKind::RW) }
         }
     }