Merge pull request #220 from Vid201/feat/blocks

Listen for new blocks and remove user operations

Cargo.toml

@@ -14,19 +14,21 @@ default-members = ["bin/silius"]
 
 [workspace.package]
 authors = ["Vid Kersic <[email protected]>"]
-version = "0.2.0-alpha"
+version = "0.3.0-alpha"
 edition = "2021"
 license = "MIT OR Apache-2.0"
 repository = "https://github.com/Vid201/silius"
 rust-version = "1.71.1"
 
 [workspace.dependencies]
+async-stream = "0.3.5"
 async-trait = "0.1"
 ethers = { git = "https://github.com/gakonst/ethers-rs", rev = "fa3017715a298728d9fb341933818a5d0d84c2dc", features = [
     "ws",
 ] }
 expanded-pathbuf = "0.1"
 eyre = "0.6.8"
+futures-util = "0.3.28"
 parking_lot = "0.12"
 serde_json = "1"
 tokio = { version = "1.18", features = ["full"] }

bin/silius/src/bundler.rs

@@ -8,7 +8,8 @@ use silius_grpc::{
     uopool_service_run,
 };
 use silius_primitives::{
-    bundler::SendBundleMode, consts::flashbots_relay_endpoints, Chain, Wallet,
+    bundler::SendBundleMode, consts::flashbots_relay_endpoints, provider::BlockStream, Chain,
+    Wallet,
 };
 use silius_rpc::{
     debug_api::{DebugApiServer, DebugApiServerImpl},
@@ -25,13 +26,15 @@ pub async fn launch_bundler<M>(
     common_args: BundlerAndUoPoolArgs,
     rpc_args: RpcArgs,
     eth_client: Arc<M>,
+    block_streams: Vec<BlockStream>,
 ) -> eyre::Result<()>
 where
     M: Middleware + Clone + 'static,
 {
     launch_uopool(
         uopool_args.clone(),
         eth_client.clone(),
+        block_streams,
         common_args.chain.clone(),
         common_args.entry_points.clone(),
     )
@@ -131,6 +134,7 @@ where
 pub async fn launch_uopool<M>(
     args: UoPoolArgs,
     eth_client: Arc<M>,
+    block_streams: Vec<BlockStream>,
     chain: Option<String>,
     entry_points: Vec<Address>,
 ) -> eyre::Result<()>
@@ -154,6 +158,7 @@ where
         datadir,
         entry_points,
         eth_client,
+        block_streams,
         chain_id,
         args.max_verification_gas,
         args.min_stake,

bin/silius/src/cli/args.rs

@@ -102,7 +102,7 @@ pub struct BundlerAndUoPoolArgs {
     pub eth_client_address: String,
 
     /// Chain information.
-    #[clap(long, default_value= "dev", value_parser = SUPPORTED_CHAINS)]
+    #[clap(long, value_parser = SUPPORTED_CHAINS)]
     pub chain: Option<String>,
 
     /// Entry point addresses.

bin/silius/src/cli/commands.rs

@@ -1,9 +1,9 @@
 use super::args::{BundlerAndUoPoolArgs, BundlerArgs, CreateWalletArgs, RpcArgs, UoPoolArgs};
-use crate::{
-    bundler::{create_wallet, launch_bundler, launch_bundling, launch_rpc, launch_uopool},
-    utils::{create_http_provider, create_ws_provider},
-};
+use crate::bundler::{create_wallet, launch_bundler, launch_bundling, launch_rpc, launch_uopool};
 use clap::Parser;
+use silius_primitives::provider::{
+    create_http_block_streams, create_http_provider, create_ws_block_streams, create_ws_provider,
+};
 use std::{future::pending, sync::Arc};
 
 /// Start the bundler with all components (bundling component, user operation mempool, RPC server)
@@ -30,11 +30,31 @@ impl BundlerCommand {
     /// Execute the command
     pub async fn execute(self) -> eyre::Result<()> {
         if self.common.eth_client_address.clone().starts_with("http") {
-            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address)?);
-            launch_bundler(self.bundler, self.uopool, self.common, self.rpc, eth_client).await?;
+            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address).await?);
+            let block_streams =
+                create_http_block_streams(eth_client.clone(), self.common.entry_points.len()).await;
+            launch_bundler(
+                self.bundler,
+                self.uopool,
+                self.common,
+                self.rpc,
+                eth_client,
+                block_streams,
+            )
+            .await?;
         } else {
             let eth_client = Arc::new(create_ws_provider(&self.common.eth_client_address).await?);
-            launch_bundler(self.bundler, self.uopool, self.common, self.rpc, eth_client).await?;
+            let block_streams =
+                create_ws_block_streams(eth_client.clone(), self.common.entry_points.len()).await;
+            launch_bundler(
+                self.bundler,
+                self.uopool,
+                self.common,
+                self.rpc,
+                eth_client,
+                block_streams,
+            )
+            .await?;
         }
 
         pending().await
@@ -61,7 +81,7 @@ impl BundlingCommand {
     /// Execute the command
     pub async fn execute(self) -> eyre::Result<()> {
         if self.common.eth_client_address.clone().starts_with("http") {
-            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address)?);
+            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address).await?);
             launch_bundling(
                 self.bundler,
                 eth_client,
@@ -102,19 +122,25 @@ impl UoPoolCommand {
     /// Execute the command
     pub async fn execute(self) -> eyre::Result<()> {
         if self.common.eth_client_address.clone().starts_with("http") {
-            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address)?);
+            let eth_client = Arc::new(create_http_provider(&self.common.eth_client_address).await?);
+            let block_streams =
+                create_http_block_streams(eth_client.clone(), self.common.entry_points.len()).await;
             launch_uopool(
                 self.uopool,
                 eth_client,
+                block_streams,
                 self.common.chain,
                 self.common.entry_points,
             )
             .await?;
         } else {
             let eth_client = Arc::new(create_ws_provider(&self.common.eth_client_address).await?);
+            let block_streams =
+                create_ws_block_streams(eth_client.clone(), self.common.entry_points.len()).await;
             launch_uopool(
                 self.uopool,
                 eth_client,
+                block_streams,
                 self.common.chain,
                 self.common.entry_points,
             )

bin/silius/src/utils.rs

@@ -1,8 +1,5 @@
 use dirs::home_dir;
-use ethers::{
-    providers::{Http, Provider, Ws},
-    types::{Address, U256},
-};
+use ethers::types::{Address, U256};
 use expanded_pathbuf::ExpandedPathBuf;
 use pin_utils::pin_mut;
 use silius_primitives::{bundler::SendBundleMode, UoPoolMode};
@@ -67,15 +64,3 @@ where
 
     Ok(())
 }
-
-/// Creates ethers provider with HTTP connection
-pub fn create_http_provider(addr: &str) -> eyre::Result<Provider<Http>> {
-    let provider = Provider::<Http>::try_from(addr)?;
-    Ok(provider)
-}
-
-/// Creates ethers provider with WebSockets connection
-pub async fn create_ws_provider(addr: &str) -> eyre::Result<Provider<Ws>> {
-    let provider = Provider::<Ws>::connect_with_reconnects(addr, usize::MAX).await?;
-    Ok(provider)
-}

crates/bundler/src/bundler.rs

@@ -435,11 +435,11 @@ pub(crate) fn generate_flashbots_middleware<M: Middleware + 'static>(
 
     let mut flashbots_middleware = FlashbotsMiddleware::new(
         eth_client,
-        Url::parse(relay_endpoint.clone())?,
+        Url::parse(relay_endpoint)?,
         bundle_signer.clone(),
     );
     flashbots_middleware.set_simulation_relay(
-        Url::parse(relay_endpoint.clone()).expect("Failed to parse simulation relay URL"),
+        Url::parse(relay_endpoint).expect("Failed to parse simulation relay URL"),
         bundle_signer.clone(),
     );
 

crates/contracts/src/entry_point.rs

@@ -391,7 +391,6 @@ mod tests {
     fn deserialize_error_msg() -> eyre::Result<()> {
         let err_msg = Bytes::from_str("0x0000000000000000000000000000000000000000000000000000000000000020000000000000000000000000000000000000000000000000000000000000001841413934206761732076616c756573206f766572666c6f770000000000000000")?;
         let res = EntryPointAPIErrors::decode(err_msg)?;
-        println!("res: {:?}", res);
         match res {
             EntryPointAPIErrors::RevertString(s) => {
                 assert_eq!(s, "AA94 gas values overflow")

crates/grpc/Cargo.toml

@@ -17,6 +17,7 @@ dashmap = "5.4.0"
 ethers = { workspace = true }
 expanded-pathbuf = { workspace = true }
 eyre = { workspace = true }
+futures-util = { workspace = true }
 parking_lot = { workspace = true }
 prost = "0.11"
 serde_json = { workspace = true }