feat(net): test syncing from geth

crates/staged-sync/Cargo.toml

@@ -21,6 +21,7 @@ reth-network = {path = "../../crates/net/network", features = ["serde"] }
 reth-primitives = { path = "../../crates/primitives" }
 reth-provider = { path = "../../crates/storage/provider", features = ["test-utils"] }
 reth-net-nat = { path = "../../crates/net/nat" }
+reth-interfaces = { path = "../interfaces", optional = true }
 
 # io
 serde = "1.0"
@@ -32,3 +33,64 @@ walkdir = "2.3.2"
 eyre = "0.6.8"
 shellexpand = "3.0.0"
 tracing = "0.1.37"
+
+# crypto
+rand = { version = "0.8", optional = true }
+
+# errors
+thiserror = { version = "1", optional = true }
+
+# enr
+enr = { version = "0.7.0", features = ["serde", "rust-secp256k1"], optional = true }
+
+# ethers
+ethers-core = { git = "https://github.com/gakonst/ethers-rs", default-features = false, optional = true }
+ethers-providers = { git = "https://github.com/gakonst/ethers-rs", features = ["ws"], default-features = false, optional = true }
+ethers-middleware = { git = "https://github.com/gakonst/ethers-rs", default-features = false, optional = true }
+ethers-signers = { git = "https://github.com/gakonst/ethers-rs", default-features = false, optional = true }
+
+# async / futures
+async-trait = { version = "0.1", optional = true }
+tokio = { version = "1", features = ["io-util", "net", "macros", "rt-multi-thread", "time"], optional = true }
+
+# misc
+tempfile = { version = "3.3", optional = true }
+hex = { version = "0.4", optional = true }
+
+[dev-dependencies]
+# reth crates
+reth-tracing = { path = "../tracing" }
+reth-stages = { path = "../stages" }
+reth-downloaders = { path = "../net/downloaders" }
+reth-staged-sync = { path = ".", features = ["test-utils"] }
+
+# async/futures
+futures = "0.3"
+tokio = { version = "1", features = ["io-util", "net", "macros", "rt-multi-thread", "time"] }
+tokio-stream = "0.1"
+
+# crypto
+secp256k1 = { version = "0.24", features = [
+    "global-context",
+    "rand-std",
+    "recovery",
+] }
+
+[features]
+test-utils = [
+    "reth-network/test-utils",
+    "reth-interfaces/test-utils",
+    "reth-network/test-utils",
+    "reth-provider/test-utils",
+    "dep:enr",
+    "dep:ethers-core",
+    "dep:tempfile",
+    "dep:thiserror",
+    "dep:hex",
+    "dep:rand",
+    "dep:tokio",
+    "dep:ethers-signers",
+    "dep:ethers-providers",
+    "dep:ethers-middleware",
+    "dep:async-trait"
+]

crates/staged-sync/src/lib.rs

@@ -2,3 +2,7 @@ pub mod config;
 pub use config::Config;
 
 pub mod utils;
+
+#[cfg(any(test, feature = "test-utils"))]
+/// Common helpers for integration testing.
+pub mod test_utils;

crates/staged-sync/src/test_utils/clique.rs

@@ -0,0 +1,120 @@
+//! Helper struct for working with a clique geth instance.
+
+use enr::k256::ecdsa::SigningKey;
+use ethers_core::utils::{Geth, GethInstance};
+use ethers_middleware::SignerMiddleware;
+use ethers_providers::{Provider, Ws};
+use ethers_signers::{LocalWallet, Wallet};
+use std::{
+    io::{BufRead, BufReader},
+    net::SocketAddr,
+};
+
+/// A [`Geth`](ethers_core::utils::Geth) instance configured with Clique and a custom
+/// [`Genesis`](ethers_core::utils::Genesis).
+///
+/// This holds a [`SignerMiddleware`](ethers_middleware::signer_middleware::SignerMiddleware) for
+/// enabling block production and creating transactions.
+///
+/// # Example
+/// ```
+/// # use ethers_core::utils::Geth;
+/// # use reth_staged_sync::test_utils::CliqueGethInstance;
+///
+/// // this creates a funded geth
+/// let clique_geth = Geth::new()
+///     .chain_id(chain_id);
+///
+/// // build the funded geth, generating a random signing key and enabling clique
+/// let (mut clique, provider) = CliqueGethInstance::new(clique_geth, None).await;
+///
+/// // don't print logs, but drain the stderr
+/// clique.prevent_blocking().await;
+/// ```
+pub struct CliqueGethInstance(
+    /// The spawned [`GethInstance`](ethers_core::utils::GethInstance).
+    pub GethInstance,
+);
+
+impl CliqueGethInstance {
+    /// Sets up a new [`SignerMiddleware`](ethers_middleware::signer_middleware::SignerMiddleware)
+    /// for the [`Geth`](ethers_core::utils::Geth) instance and returns the
+    /// [`CliqueGethInstance`].
+    ///
+    /// The signer is assumed to be the clique signer and the signer for any transactions sent for
+    /// block production.
+    ///
+    /// This also spawns the geth instance.
+    pub async fn new(
+        geth: Geth,
+        signer: Option<SigningKey>,
+    ) -> (Self, SignerMiddleware<Provider<Ws>, Wallet<SigningKey>>) {
+        let signer = signer.unwrap_or_else(|| SigningKey::random(&mut rand::thread_rng()));
+
+        let geth = geth.set_clique_private_key(signer.clone());
+
+        // spawn the geth instance
+        let instance = geth.spawn();
+
+        // create the signer
+        let wallet: LocalWallet = signer.clone().into();
+
+        // set up ethers provider
+        let geth_endpoint = SocketAddr::new([127, 0, 0, 1].into(), instance.port()).to_string();
+        let provider = Provider::<Ws>::connect(format!("ws://{geth_endpoint}")).await.unwrap();
+        let provider =
+            SignerMiddleware::new_with_provider_chain(provider, wallet.clone()).await.unwrap();
+
+        (Self(instance), provider)
+    }
+
+    /// Prints the logs of the [`Geth`](ethers_core::utils::Geth) instance in a new
+    /// [`task`](tokio::task).
+    #[allow(dead_code)]
+    pub async fn print_logs(&mut self) {
+        // take the stderr of the geth instance and print it
+        let stderr = self.0.stderr().unwrap();
+
+        // print logs in a new task
+        let mut err_reader = BufReader::new(stderr);
+
+        tokio::spawn(async move {
+            loop {
+                if let (Ok(line), line_str) = {
+                    let mut buf = String::new();
+                    (err_reader.read_line(&mut buf), buf.clone())
+                } {
+                    if line == 0 {
+                        break
+                    }
+                    if !line_str.is_empty() {
+                        dbg!(line_str);
+                    }
+                }
+            }
+        });
+    }
+
+    /// Prevents the [`Geth`](ethers_core::utils::Geth) instance from blocking due to the `stderr`
+    /// filling up.
+    pub async fn prevent_blocking(&mut self) {
+        // take the stderr of the geth instance and print it
+        let stderr = self.0.stderr().unwrap();
+
+        // print logs in a new task
+        let mut err_reader = BufReader::new(stderr);
+
+        tokio::spawn(async move {
+            loop {
+                if let (Ok(line), _line_str) = {
+                    let mut buf = String::new();
+                    (err_reader.read_line(&mut buf), buf.clone())
+                } {
+                    if line == 0 {
+                        break
+                    }
+                }
+            }
+        });
+    }
+}

crates/staged-sync/src/test_utils/clique_middleware.rs

@@ -0,0 +1,126 @@
+//! Helper extension traits for working with clique providers.
+
+use async_trait::async_trait;
+use enr::k256::ecdsa::SigningKey;
+use ethers_core::{
+    types::{transaction::eip2718::TypedTransaction, Address, Block, BlockNumber, H256},
+    utils::secret_key_to_address,
+};
+use ethers_middleware::SignerMiddleware;
+use ethers_providers::Middleware;
+use ethers_signers::Signer;
+use reth_network::test_utils::enr_to_peer_id;
+use reth_primitives::PeerId;
+use thiserror::Error;
+use tracing::trace;
+
+/// An error that can occur when using the [`CliqueMiddleware`].
+#[derive(Error, Debug)]
+pub enum CliqueError<E> {
+    /// Error encountered when using the provider
+    #[error(transparent)]
+    ProviderError(#[from] E),
+
+    /// No genesis block returned from the provider
+    #[error("no genesis block returned from the provider")]
+    NoGenesis,
+
+    /// No tip block returned from the provider
+    #[error("no tip block returned from the provider")]
+    NoTip,
+
+    /// Account was not successfully unlocked on the provider
+    #[error("account was not successfully unlocked on the provider")]
+    AccountNotUnlocked,
+
+    /// Mining was not successfully enabled on the provider
+    #[error("mining was not successfully enabled on the provider")]
+    MiningNotEnabled,
+
+    /// Mismatch between locally computed address and address returned from the provider
+    #[error("local address {local} does not match remote address {remote}")]
+    AddressMismatch {
+        /// The locally computed address
+        local: Address,
+
+        /// The address returned from the provider
+        remote: Address,
+    },
+}
+
+/// Error type for [`CliqueMiddleware`].
+pub type CliqueMiddlewareError<M> = CliqueError<<M as Middleware>::Error>;
+
+/// Extension trait for [`Middleware`] to provide clique specific functionality.
+#[async_trait(?Send)]
+pub trait CliqueMiddleware: Send + Sync + Middleware {
+    /// Enable mining on the clique geth instance by importing and unlocking the signer account
+    /// derived from given private key and password.
+    async fn enable_mining(
+        &self,
+        signer: SigningKey,
+        password: String,
+    ) -> Result<(), CliqueMiddlewareError<Self>> {
+        let our_address = secret_key_to_address(&signer);
+
+        // send the private key to geth and unlock it
+        let key_bytes = signer.to_bytes().to_vec().into();
+        trace!(
+            private_key=%hex::encode(&key_bytes),
+            "Importing private key"
+        );
+
+        let unlocked_addr = self.import_raw_key(key_bytes, password.to_string()).await?;
+        if unlocked_addr != our_address {
+            return Err(CliqueError::AddressMismatch { local: our_address, remote: unlocked_addr })
+        }
+
+        let unlock_success = self.unlock_account(our_address, password.to_string(), None).await?;
+
+        if !unlock_success {
+            return Err(CliqueError::AccountNotUnlocked)
+        }
+
+        // start mining?
+        self.start_mining(None).await?;
+
+        // check that we are mining
+        let mining = self.mining().await?;
+        if !mining {
+            return Err(CliqueError::MiningNotEnabled)
+        }
+        Ok(())
+    }
+
+    /// Returns the chain tip of the [`Geth`](ethers_core::utils::Geth) instance by calling
+    /// geth's `eth_getBlock`.
+    async fn remote_tip_block(&self) -> Result<Block<H256>, CliqueMiddlewareError<Self>> {
+        self.get_block(BlockNumber::Latest).await?.ok_or(CliqueError::NoTip)
+    }
+
+    /// Returns the genesis block of the [`Geth`](ethers_core::utils::Geth) instance by calling
+    /// geth's `eth_getBlock`.
+    async fn remote_genesis_block(&self) -> Result<Block<H256>, CliqueMiddlewareError<Self>> {
+        self.get_block(BlockNumber::Earliest).await?.ok_or(CliqueError::NoGenesis)
+    }
+
+    /// Signs and sends the given unsigned transactions sequentially, signing with the private key
+    /// used to configure the [`CliqueGethInstance`].
+    async fn send_requests<T: IntoIterator<Item = TypedTransaction>>(
+        &self,
+        txs: T,
+    ) -> Result<(), CliqueMiddlewareError<Self>> {
+        for tx in txs {
+            self.send_transaction(tx, None).await?;
+        }
+        Ok(())
+    }
+
+    /// Returns the [`Geth`](ethers_core::utils::Geth) instance [`PeerId`](reth_primitives::PeerId)
+    /// by calling geth's `admin_nodeInfo`.
+    async fn peer_id(&self) -> Result<PeerId, CliqueMiddlewareError<Self>> {
+        Ok(enr_to_peer_id(self.node_info().await?.enr))
+    }
+}
+
+impl<M: Middleware, S: Signer> CliqueMiddleware for SignerMiddleware<M, S> {}

crates/staged-sync/src/test_utils/mod.rs

@@ -0,0 +1,9 @@
+#![warn(missing_docs, unreachable_pub)]
+
+//! Common helpers for staged sync integration testing.
+
+pub mod clique;
+pub mod clique_middleware;
+
+pub use clique::CliqueGethInstance;
+pub use clique_middleware::{CliqueError, CliqueMiddleware, CliqueMiddlewareError};

crates/staged-sync/tests/lib.rs

@@ -0,0 +1,10 @@
+//! Integration tests and test helpers for reth.
+#![cfg(test)]
+#![warn(missing_docs, unreachable_pub)]
+#![deny(unused_must_use, rust_2018_idioms)]
+#![doc(test(
+    no_crate_inject,
+    attr(deny(warnings, rust_2018_idioms), allow(dead_code, unused_variables))
+))]
+
+pub mod sync;