feat: add broadcaster middleware (#54)

This is a new middleware which will send a bundle to a list of builders

examples/broadcast.rs

@@ -0,0 +1,86 @@
+use ethers::core::{rand::thread_rng, types::transaction::eip2718::TypedTransaction};
+use ethers::prelude::*;
+use ethers_flashbots::*;
+use eyre::Result;
+use std::convert::TryFrom;
+use url::Url;
+
+// See https://www.mev.to/builders for up to date builder URLs
+static BUILDER_URLS: &[&str] = &[
+    "https://builder0x69.io",
+    "https://rpc.beaverbuild.org",
+    "https://relay.flashbots.net",
+    "https://rsync-builder.xyz",
+    "https://rpc.titanbuilder.xyz",
+    "https://api.blocknative.com/v1/auction",
+    "https://mev.api.blxrbdn.com",
+    "https://eth-builder.com",
+    "https://builder.gmbit.co/rpc",
+    "https://buildai.net",
+    "https://rpc.payload.de",
+    "https://rpc.lightspeedbuilder.info",
+    "https://rpc.nfactorial.xyz",
+];
+
+#[tokio::main]
+async fn main() -> Result<()> {
+    // Connect to the network
+    let provider = Provider::<Http>::try_from("https://mainnet.eth.aragon.network")?;
+
+    // This is your searcher identity
+    let bundle_signer = LocalWallet::new(&mut thread_rng());
+    // This signs transactions
+    let wallet = LocalWallet::new(&mut thread_rng());
+
+    // Add signer and Flashbots middleware
+    let client = SignerMiddleware::new(
+        BroadcasterMiddleware::new(
+            provider,
+            BUILDER_URLS
+                .iter()
+                .map(|url| Url::parse(url).unwrap())
+                .collect(),
+            Url::parse("https://relay.flashbots.net")?,
+            bundle_signer,
+        ),
+        wallet,
+    );
+
+    // get last block number
+    let block_number = client.get_block_number().await?;
+
+    // Build a custom bundle that pays 0x0000000000000000000000000000000000000000
+    let tx = {
+        let mut inner: TypedTransaction = TransactionRequest::pay(Address::zero(), 100).into();
+        client.fill_transaction(&mut inner, None).await?;
+        inner
+    };
+    let signature = client.signer().sign_transaction(&tx).await?;
+    let bundle = BundleRequest::new()
+        .push_transaction(tx.rlp_signed(&signature))
+        .set_block(block_number + 1)
+        .set_simulation_block(block_number)
+        .set_simulation_timestamp(0);
+
+    // Send it
+    let results = client.inner().send_bundle(&bundle).await?;
+
+    // You can also optionally wait to see if the bundle was included
+    for result in results {
+        match result {
+            Ok(pending_bundle) => match pending_bundle.await {
+                Ok(bundle_hash) => println!(
+                    "Bundle with hash {:?} was included in target block",
+                    bundle_hash
+                ),
+                Err(PendingBundleError::BundleNotIncluded) => {
+                    println!("Bundle was not included in target block.")
+                }
+                Err(e) => println!("An error occured: {}", e),
+            },
+            Err(e) => println!("An error occured: {}", e),
+        }
+    }
+
+    Ok(())
+}

src/lib.rs

@@ -18,7 +18,7 @@ mod user;
 pub use user::UserStats;
 
 mod middleware;
-pub use middleware::{FlashbotsMiddleware, FlashbotsMiddlewareError};
+pub use middleware::{BroadcasterMiddleware, FlashbotsMiddleware, FlashbotsMiddlewareError};
 
 mod jsonrpc;
 mod relay;

src/middleware.rs

@@ -13,6 +13,7 @@ use ethers::{
     providers::{Middleware, MiddlewareError, PendingTransaction},
     signers::Signer,
 };
+use futures_util::future;
 use thiserror::Error;
 use url::Url;
 
@@ -290,3 +291,211 @@ where
             .interval(self.provider().get_interval()))
     }
 }
+
+/// A middleware used to broadcast bundles to multiple builders.
+///
+/// **NOTE**: This middleware does **NOT** sign your transactions. Use
+/// another method to sign your transactions, and then forward the signed
+/// transactions to the middleware.
+///
+/// You can either send custom bundles (see [`BundleRequest`]) or send
+/// transactions as you normally would (see [`Middleware::send_transaction`]) from
+/// another middleware.
+///
+/// If you use [`Middleware::send_transaction`] then a bundle will be constructed
+/// for you with the following assumptions:
+///
+/// - You do not want to allow the transaction to revert
+/// - You do not care to set a minimum or maximum timestamp for the bundle
+/// - The block you are targetting with your bundle is the next block
+/// - You do not want to simulate the bundle before sending to the builder
+///
+/// # Example
+/// ```
+/// use ethers::prelude::*;
+/// use std::convert::TryFrom;
+/// use ethers_flashbots::BroadcasterMiddleware;
+/// use url::Url;
+///
+/// # async fn foo() -> Result<(), Box<dyn std::error::Error>> {
+/// let provider = Provider::<Http>::try_from("http://localhost:8545")
+///     .expect("Could not instantiate HTTP provider");
+///
+/// // Used to sign Flashbots relay requests - this is your searcher identity
+/// let signer: LocalWallet = "380eb0f3d505f087e438eca80bc4df9a7faa24f868e69fc0440261a0fc0567dc"
+///     .parse()?;
+///
+/// // Used to sign transactions
+/// let wallet: LocalWallet = "380eb0f3d505f087e438eca80bc4df9a7faa24f868e69fc0440261a0fc0567dc"
+///     .parse()?;
+///
+/// // Note: The order is important! You want the signer
+/// // middleware to sign your transactions *before* they
+/// // are sent to your Flashbots middleware.
+/// let mut client = SignerMiddleware::new(
+///     BroadcasterMiddleware::new(
+///         provider,
+///         vec![Url::parse("https://rpc.titanbuilder.xyz")?, Url::parse("https://relay.flashbots.net")?],
+///         Url::parse("https://relay.flashbots.net")?,
+///         signer
+///     ),
+///     wallet
+/// );
+///
+/// // This transaction will now be sent as a Flashbots bundle!
+/// let tx = TransactionRequest::pay("vitalik.eth", 100);
+/// let pending_tx = client.send_transaction(tx, None).await?;
+/// # Ok(())
+/// # }
+/// ```
+#[derive(Debug)]
+pub struct BroadcasterMiddleware<M, S> {
+    inner: M,
+    relays: Vec<Relay<S>>,
+    simulation_relay: Relay<S>,
+}
+
+impl<M: Middleware, S: Signer> BroadcasterMiddleware<M, S> {
+    /// Initialize a new Flashbots middleware.
+    ///
+    /// The signer is used to sign requests to the relay.
+    pub fn new(
+        inner: M,
+        relay_urls: Vec<Url>,
+        simulation_relay: impl Into<Url>,
+        relay_signer: S,
+    ) -> Self
+    where
+        S: Clone,
+    {
+        Self {
+            inner,
+            relays: relay_urls
+                .into_iter()
+                .map(|r| Relay::new(r, Some(relay_signer.clone())))
+                .collect(),
+            simulation_relay: Relay::new(simulation_relay, None),
+        }
+    }
+
+    /// Get the relay client used by the middleware.
+    pub fn relay(&self) -> &Vec<Relay<S>> {
+        &self.relays
+    }
+
+    /// Get the relay client used by the middleware to simulate
+    /// bundles.
+    pub fn simulation_relay(&self) -> &Relay<S> {
+        &self.simulation_relay
+    }
+
+    /// Simulate a bundle.
+    ///
+    /// See [`eth_callBundle`][fb_callBundle] for more information.
+    ///
+    /// [fb_callBundle]: https://docs.flashbots.net/flashbots-auction/searchers/advanced/rpc-endpoint#eth_callbundle
+    pub async fn simulate_bundle(
+        &self,
+        bundle: &BundleRequest,
+    ) -> Result<SimulatedBundle, FlashbotsMiddlewareError<M, S>> {
+        bundle
+            .block()
+            .and(bundle.simulation_block())
+            .and(bundle.simulation_timestamp())
+            .ok_or(FlashbotsMiddlewareError::MissingParameters)?;
+
+        self.simulation_relay
+            .request("eth_callBundle", [bundle])
+            .await
+            .map_err(FlashbotsMiddlewareError::RelayError)
+    }
+
+    /// Broadcast a bundle to the builders.
+    ///
+    /// See [`eth_sendBundle`][fb_sendBundle] for more information.
+    ///
+    /// [fb_sendBundle]: https://docs.flashbots.net/flashbots-auction/searchers/advanced/rpc-endpoint#eth_sendbundle
+    pub async fn send_bundle(
+        &self,
+        bundle: &BundleRequest,
+    ) -> Result<
+        Vec<
+            Result<
+                PendingBundle<'_, <Self as Middleware>::Provider>,
+                FlashbotsMiddlewareError<M, S>,
+            >,
+        >,
+        FlashbotsMiddlewareError<M, S>,
+    > {
+        // The target block must be set
+        bundle
+            .block()
+            .ok_or(FlashbotsMiddlewareError::MissingParameters)?;
+
+        let futures = self
+            .relays
+            .iter()
+            .map(|relay| async move {
+                let response = relay.request("eth_sendBundle", [bundle]).await;
+                response
+                    .map(|r: SendBundleResponse| {
+                        PendingBundle::new(
+                            r.bundle_hash,
+                            bundle.block().unwrap(),
+                            bundle.transaction_hashes(),
+                            self.provider(),
+                        )
+                    })
+                    .map_err(FlashbotsMiddlewareError::RelayError)
+            })
+            .collect::<Vec<_>>();
+
+        let responses = future::join_all(futures).await;
+
+        Ok(responses)
+    }
+}
+
+#[async_trait]
+impl<M, S> Middleware for BroadcasterMiddleware<M, S>
+where
+    M: Middleware,
+    S: Signer,
+{
+    type Error = FlashbotsMiddlewareError<M, S>;
+    type Provider = M::Provider;
+    type Inner = M;
+
+    fn inner(&self) -> &M {
+        &self.inner
+    }
+
+    async fn send_raw_transaction<'a>(
+        &'a self,
+        tx: Bytes,
+    ) -> Result<PendingTransaction<'a, Self::Provider>, Self::Error> {
+        let tx_hash = keccak256(&tx);
+
+        // Get the latest block
+        let latest_block = self
+            .inner
+            .get_block(BlockNumber::Latest)
+            .await
+            .map_err(FlashbotsMiddlewareError::MiddlewareError)?
+            .expect("The latest block is pending (this should not happen)");
+
+        // Construct the bundle, assuming that the target block is the
+        // next block.
+        let bundle = BundleRequest::new().push_transaction(tx.clone()).set_block(
+            latest_block
+                .number
+                .expect("The latest block is pending (this should not happen)")
+                + 1,
+        );
+
+        self.send_bundle(&bundle).await?;
+
+        Ok(PendingTransaction::new(tx_hash.into(), self.provider())
+            .interval(self.provider().get_interval()))
+    }
+}