diff --git a/crates/driver/example.toml b/crates/driver/example.toml index aeb4580ea1..2788842209 100644 --- a/crates/driver/example.toml +++ b/crates/driver/example.toml @@ -29,6 +29,24 @@ max-additional-tip = "5000000000" additional-tip-percentage = 0.05 mines-reverting-txs = true +# Optional list of block builders to submit to directly. When it is non-empty +# the driver signs the settlement itself and broadcasts the raw transaction to +# every builder, and cancellations take the same route; `url` above then only +# serves nonce and txpool queries. The submission succeeds as long as one +# builder accepts the transaction. Builders mine reverting transactions, so +# `mines-reverting-txs` keeps its usual default of `true`. Requests carry an +# `X-Flashbots-Signature` header signed by the settlement account, which +# builders that authenticate their callers (e.g. BuilderNet) require and the +# rest ignore. +# +# A settlement that may revert only races the best configured tier: the +# mempools with builders, else the ones with `mines-reverting-txs = false`, +# else the public ones. So configuring builders takes every other mempool out +# of the race for those settlements. +# [[submission.mempool.builders]] +# name = "titan" +# url = "https://rpc.titanbuilder.xyz/" + [contracts] # Optionally override the contract addresses, necessary on less popular blockchains gp-v2-settlement = "0x9008D19f58AAbD9eD0D60971565AA8510560ab41" weth = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2" diff --git a/crates/driver/src/domain/mempools.rs b/crates/driver/src/domain/mempools.rs index 239bdfdb65..36561a3b5f 100644 --- a/crates/driver/src/domain/mempools.rs +++ b/crates/driver/src/domain/mempools.rs @@ -97,28 +97,21 @@ impl Mempools { Ok(res?.tx_hash) } - /// A mempool is disabled if all of the following are true: - /// * the settlement may revert (see [`Settlement::may_revert`]) - /// * the pool has revert protection enabled (see - /// [`Self::revert_protection`]) - /// * reverts can get mined (see [`infra::Mempool::reverts_can_get_mined`]) + /// A settlement that may revert (see [`Settlement::may_revert`]) only + /// races the best configured [`Tier`], so a mempool of a worse tier is + /// disabled. Settlements that cannot revert race every mempool. fn is_disabled(&self, mempool: &infra::Mempool, settlement: &Settlement) -> bool { - settlement.may_revert() - && matches!(self.revert_protection(), RevertProtection::Enabled) - && mempool.reverts_can_get_mined() + settlement.may_revert() && Tier::of(mempool) > self.best_tier() } - /// Defines if the mempools are configured in a way that guarantees that - /// settled solution will not revert. - pub fn revert_protection(&self) -> RevertProtection { - match self - .mempools + /// The best tier among the configured mempools. + fn best_tier(&self) -> Tier { + self.mempools .iter() - .all(|mempool| mempool.reverts_can_get_mined()) - { - true => RevertProtection::Disabled, - false => RevertProtection::Enabled, - } + .map(Tier::of) + .min() + // `try_new` rejects an empty list of mempools. + .expect("no mempools configured") } async fn submit( @@ -600,12 +593,31 @@ impl SubmissionSuccess { #[error("no mempools configured, cannot execute settlements")] pub struct NoMempools; -/// Defines if the mempools are configured in a way that guarantees that -/// /settle'd solution will not revert. -#[derive(Debug, Clone, Copy)] -pub enum RevertProtection { - Enabled, - Disabled, +/// Where a settlement that may revert can go, best tier first. Only the best +/// configured tier races: the builders if there are any, else the revert +/// protected mempools, else the public ones. +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord)] +enum Tier { + /// Handed straight to block builders. They mine reverting txs but offer + /// the best inclusion. + Builders, + /// An RPC that drops reverting txs instead of mining them, e.g. MEV + /// Blocker. + RevertProtected, + /// A public mempool, which mines reverting txs. + Public, +} + +impl Tier { + fn of(mempool: &infra::Mempool) -> Self { + if mempool.submits_to_builders() { + Self::Builders + } else if mempool.reverts_can_get_mined() { + Self::Public + } else { + Self::RevertProtected + } + } } #[derive(Debug, thiserror::Error)] diff --git a/crates/driver/src/domain/mod.rs b/crates/driver/src/domain/mod.rs index f17cb1ec9f..ba13ba581e 100644 --- a/crates/driver/src/domain/mod.rs +++ b/crates/driver/src/domain/mod.rs @@ -12,5 +12,5 @@ pub use { flashloan::Flashloan, interaction::Interaction, liquidity::Liquidity, - mempools::{Mempools, RevertProtection}, + mempools::Mempools, }; diff --git a/crates/driver/src/infra/config/file/load.rs b/crates/driver/src/infra/config/file/load.rs index d91a0c42ea..cd0711d361 100644 --- a/crates/driver/src/infra/config/file/load.rs +++ b/crates/driver/src/infra/config/file/load.rs @@ -348,6 +348,14 @@ pub async fn load(chain: Chain, path: &Path) -> infra::Config { }, max_additional_tip: mempool.max_additional_tip, additional_tip_percentage: mempool.additional_tip_percentage, + builders: mempool + .builders + .iter() + .map(|builder| mempool::Builder { + name: builder.name.clone(), + url: builder.url.clone(), + }) + .collect(), }) .collect(), simulator: config.simulator, diff --git a/crates/driver/src/infra/config/file/mod.rs b/crates/driver/src/infra/config/file/mod.rs index c15de8de96..b6d68a04b7 100644 --- a/crates/driver/src/infra/config/file/mod.rs +++ b/crates/driver/src/infra/config/file/mod.rs @@ -153,7 +153,7 @@ impl From for BlockNumberOrTag { #[serde_as] #[derive(Debug, Deserialize)] -#[serde(rename_all = "kebab-case")] +#[serde(rename_all = "kebab-case", deny_unknown_fields)] struct Mempool { /// Name for better logging and metrics. name: Option, @@ -175,6 +175,22 @@ struct Mempool { /// assume reverting transactions will get mined eventually. #[serde(default = "default_mines_reverting_txs")] mines_reverting_txs: bool, + /// Block builders to send the settlement transaction to directly. When + /// this is non-empty settlements and cancellations both go to the builders + /// and `url` only serves nonce and txpool queries. + #[serde(default)] + builders: Vec, +} + +/// A block builder that accepts settlement transactions over +/// `eth_sendRawTransaction`. +#[derive(Debug, Deserialize)] +#[serde(rename_all = "kebab-case", deny_unknown_fields)] +struct Builder { + /// Name for better logging and metrics. + name: String, + /// The RPC URL to send the signed transaction to. + url: Url, } #[derive(Debug, Deserialize)] diff --git a/crates/driver/src/infra/mempool/builders.rs b/crates/driver/src/infra/mempool/builders.rs new file mode 100644 index 0000000000..b455fb7aba --- /dev/null +++ b/crates/driver/src/infra/mempool/builders.rs @@ -0,0 +1,272 @@ +//! Direct submission of settlements to block builders. +//! +//! Builders neither agree on the shape of a successful +//! `eth_sendRawTransaction` response nor on whether they authenticate their +//! callers, so the requests are built by hand instead of through an RPC client. + +use { + crate::infra::{observe::metrics, solver::Account}, + alloy::{ + consensus::TxEnvelope, + eips::eip2718::Encodable2718, + hex, + network::TxSigner, + primitives::{Address, eip191_hash_message, keccak256}, + }, + anyhow::{Context, anyhow}, + bytes::Bytes, + eth_domain_types as eth, + futures::future::join_all, + std::collections::HashMap, + url::Url, +}; + +/// A builder that stops answering must not hold up the whole broadcast: +/// submission is on the critical path of the block. +const REQUEST_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(3); + +/// A block builder that accepts settlement transactions directly. +#[derive(Debug, Clone)] +pub struct Builder { + pub name: String, + pub url: Url, +} + +/// Broadcasts settlements to all the block builders of one mempool. +#[derive(Debug, Clone)] +pub struct Builders { + /// Name of the mempool these builders belong to, for logs and metrics. + mempool: String, + builders: Vec, + http: reqwest::Client, + /// The accounts that sign the settlements, by address. Also used to sign + /// the requests for builders that authenticate their callers. + signers: HashMap, +} + +impl Builders { + pub fn new( + mempool: String, + builders: Vec, + signers: HashMap, + ) -> Self { + Self { + mempool, + builders, + http: reqwest::ClientBuilder::new() + .timeout(REQUEST_TIMEOUT) + .build() + .expect("failed to build the builder http client"), + signers, + } + } + + pub fn is_empty(&self) -> bool { + self.builders.is_empty() + } + + /// Broadcasts an already signed settlement to every builder. Succeeds as + /// long as one builder accepted it. + pub async fn broadcast( + &self, + envelope: &TxEnvelope, + signer: eth::Address, + ) -> anyhow::Result { + let hash = eth::TxId(*envelope.tx_hash()); + // Every builder receives the exact same bytes, so one body and one + // request signature are enough. + let body = build_body(envelope)?; + let signature = self.sign_request(signer, &body).await?; + + let accepted = join_all(self.builders.iter().map(|builder| { + let body = body.clone(); + let signature = signature.as_str(); + async move { + let result = self.post(builder, body, signature).await; + self.observe_result(builder, &hash, &result); + result.is_ok() + } + })) + .await + .into_iter() + .filter(|accepted| *accepted) + .count(); + + if accepted == 0 { + return Err(anyhow!( + "all {} builders rejected the tx", + self.builders.len() + )); + } + + tracing::debug!(?hash, accepted, total = self.builders.len(), "broadcast tx"); + Ok(hash) + } + + /// Builds the `X-Flashbots-Signature` value for the request body. It is + /// sent to every builder: the ones that don't authenticate their callers + /// ignore the header. Costs one signing call per submission, which is a + /// second KMS round trip for KMS backed accounts. + async fn sign_request(&self, signer: eth::Address, body: &[u8]) -> anyhow::Result { + let account = self + .signers + .get(&signer) + .with_context(|| format!("no account registered for {signer}"))?; + signature_header(account, body).await + } + + /// Posts an already encoded JSON-RPC body to a single builder and returns + /// the response body when the builder accepted the tx. + async fn post( + &self, + builder: &Builder, + body: Bytes, + signature: &str, + ) -> anyhow::Result { + let response = self + .http + .post(builder.url.clone()) + .header(reqwest::header::CONTENT_TYPE, "application/json") + .header("X-Flashbots-Signature", signature) + .body(body) + .send() + .await + .context("failed to reach the builder")?; + let status = response.status(); + let body = response + .text() + .await + .context("failed to read the builder response")?; + check_response(status, &body)?; + Ok(body) + } + + /// Logs and counts the outcome of a single builder submission. + fn observe_result(&self, builder: &Builder, hash: ð::TxId, result: &anyhow::Result) { + let label = match result { + Ok(body) => { + tracing::debug!( + builder = builder.name, + ?hash, + body, + "builder accepted the tx" + ); + "Success" + } + Err(err) => { + tracing::warn!( + builder = builder.name, + ?err, + ?hash, + "builder rejected the tx" + ); + "Rejected" + } + }; + metrics::get() + .builder_submission + .with_label_values(&[&self.mempool, &builder.name, label]) + .inc(); + } +} + +/// Encodes the `eth_sendRawTransaction` request body for a signed settlement. +fn build_body(envelope: &TxEnvelope) -> anyhow::Result { + Ok(Bytes::from(serde_json::to_vec(&serde_json::json!({ + "jsonrpc": "2.0", + "id": 1, + "method": "eth_sendRawTransaction", + "params": [hex::encode_prefixed(envelope.encoded_2718())], + }))?)) +} + +/// Builds the `X-Flashbots-Signature` header value. It authenticates the +/// caller, not the transaction, and is required by builders like BuilderNet, +/// which reject every unsigned request. +async fn signature_header(account: &Account, body: &[u8]) -> anyhow::Result { + // The signed message is the hex string of the body hash, not the hash + // itself, and it gets the EIP-191 prefix on top. + let message = hex::encode_prefixed(keccak256(body)); + let signature = account + .sign_hash(&eip191_hash_message(message)) + .await + .context("failed to sign the request body")?; + Ok(format!( + "{}:{}", + TxSigner::address(account), + hex::encode_prefixed(signature.as_bytes()) + )) +} + +/// Builders disagree on the body of an acceptance (a tx hash, `null`, the +/// number 200), so only a failing status or a JSON-RPC `error` object counts +/// as a rejection. +fn check_response(status: reqwest::StatusCode, body: &str) -> anyhow::Result<()> { + if !status.is_success() { + return Err(anyhow!("builder returned status {status}: {body}")); + } + let error = serde_json::from_str::(body) + .ok() + .and_then(|response| response.get("error").cloned()) + .filter(|error| !error.is_null()); + match error { + Some(error) => Err(anyhow!("builder returned error: {error}")), + None => Ok(()), + } +} + +#[cfg(test)] +mod tests { + use {super::*, alloy::signers::local::PrivateKeySigner, reqwest::StatusCode}; + + #[test] + fn acceptance_bodies_are_not_parsed() { + for body in [ + r#"{"jsonrpc":"2.0","id":1,"result":"0x1234"}"#, + r#"{"jsonrpc":"2.0","id":1,"result":null}"#, + r#"{"result":200,"error":null,"id":1}"#, + "", + ] { + check_response(StatusCode::OK, body).unwrap(); + } + } + + #[test] + fn rejections_are_detected() { + check_response( + StatusCode::OK, + r#"{"jsonrpc":"2.0","id":1,"error":{"code":-32000,"message":"insufficient funds"}}"#, + ) + .unwrap_err(); + check_response(StatusCode::BAD_GATEWAY, "bad gateway").unwrap_err(); + check_response(StatusCode::TOO_MANY_REQUESTS, "").unwrap_err(); + } + + #[tokio::test] + async fn signature_header_recovers_to_the_signer() { + let key = PrivateKeySigner::from_bytes( + &"0x59c6995e998f97a5a0044966f0945389dc9e86dae88c7a8412f4603b6b78690d" + .parse() + .unwrap(), + ) + .unwrap(); + let address = key.address(); + let body = + br#"{"jsonrpc":"2.0","id":1,"method":"eth_sendRawTransaction","params":["0x02"]}"#; + + let header = signature_header(&Account::PrivateKey(key), body) + .await + .unwrap(); + + let (signer, signature) = header.split_once(':').unwrap(); + assert_eq!(signer, address.to_string()); + let signature: alloy::primitives::Signature = signature.parse().unwrap(); + let message = hex::encode_prefixed(keccak256(body)); + assert_eq!( + signature + .recover_address_from_msg(message.as_bytes()) + .unwrap(), + address + ); + } +} diff --git a/crates/driver/src/infra/mempool/mod.rs b/crates/driver/src/infra/mempool/mod.rs index f2b00c9126..a8ea5b50f5 100644 --- a/crates/driver/src/infra/mempool/mod.rs +++ b/crates/driver/src/infra/mempool/mod.rs @@ -1,3 +1,6 @@ +mod builders; + +pub use builders::Builder; use { crate::{ boundary::{Web3, unbuffered_web3}, @@ -7,14 +10,16 @@ use { alloy::{ consensus::Transaction, eips::{BlockNumberOrTag, eip1559::Eip1559Estimation}, + network::{Ethereum, NetworkWallet, TransactionBuilder, TxSigner}, primitives::Address, providers::{Provider, ext::TxPoolApi}, rpc::types::TransactionRequest, }, anyhow::Context, + builders::Builders, dashmap::DashMap, eth_domain_types as eth, - std::sync::Arc, + std::{collections::HashMap, sync::Arc}, url::Url, }; @@ -32,6 +37,9 @@ pub struct Config { pub revert_protection: RevertProtection, pub max_additional_tip: eth::U256, pub additional_tip_percentage: f64, + /// Block builders to broadcast the signed transaction to. When empty the + /// transaction is sent to `url` instead. + pub builders: Vec, } #[cfg(test)] @@ -47,6 +55,7 @@ impl Config { additional_tip_percentage: 0., revert_protection: infra::mempool::RevertProtection::Disabled, nonce_block_number: None, + builders: Default::default(), url, } } @@ -66,6 +75,11 @@ pub enum RevertProtection { #[derive(Debug, Clone)] pub struct Mempool { transport: Web3, + /// The configured block builders. Empty when the settlement is submitted + /// to `config.url` instead. + builders: Builders, + /// Chain id of the transactions we sign ourselves for the builders. + chain_id: u64, config: Config, last_submissions: Arc>, } @@ -83,15 +97,42 @@ impl std::fmt::Display for Mempool { } impl Mempool { - pub fn new(config: Config, solver_accounts: Vec) -> Self { + pub fn new(config: Config, solver_accounts: Vec, chain_id: u64) -> Self { let transport = unbuffered_web3(&config.url); // Register the solver accounts into the wallet to submit txs on their // behalf + let mut signers = HashMap::new(); for account in solver_accounts { + signers.insert(TxSigner::address(&account), account.clone()); transport.wallet.register_signer(account); } + if !config.builders.is_empty() { + // Builders get txs we sign ourselves. An address-only account + // relies on the node to sign and would fail on every settlement. + let address_only: Vec<_> = signers + .iter() + .filter(|(_, account)| matches!(account, Account::Address(_))) + .map(|(address, _)| address) + .collect(); + assert!( + address_only.is_empty(), + "mempool {} submits to builders but accounts {address_only:?} cannot sign", + config.name + ); + } + let builders = Builders::new(config.name.clone(), config.builders.clone(), signers); + + tracing::info!( + mempool = config.name, + url = %config.url, + builders = ?config.builders, + "configured mempool" + ); + Self { transport, + builders, + chain_id, config, last_submissions: Default::default(), } @@ -141,17 +182,18 @@ impl Mempool { .gas_limit(gas_limit) .input(tx.input.into()) .value(tx.value.0) - .access_list(tx.access_list.into()); + .access_list(tx.access_list.into()) + // Must be explicit: signing the request ourselves for the builders + // silently falls back to mainnet when the chain id is missing. + .with_chain_id(self.chain_id); - let submission = self - .transport - .provider - .send_transaction(tx_request) - .await - .map_err(anyhow::Error::from); + let submission = match self.submits_to_builders() { + true => self.send_to_builders(tx_request, signer).await, + false => self.send_to_node(tx_request).await, + }; match submission { - Ok(tx) => { + Ok(hash) => { tracing::debug!( ?nonce, ?gas_price, @@ -161,7 +203,7 @@ impl Mempool { ); self.last_submissions .insert(signer, Submission { nonce, gas_price }); - Ok(eth::TxId(*tx.tx_hash())) + Ok(hash) } Err(err) => { // log pending tx in case we failed to replace a pending tx @@ -181,6 +223,25 @@ impl Mempool { } } + /// Sends the transaction to the configured node, which signs it with the + /// registered wallet and forwards it to its own mempool. + async fn send_to_node(&self, tx: TransactionRequest) -> anyhow::Result { + let pending = self.transport.provider.send_transaction(tx).await?; + Ok(eth::TxId(*pending.tx_hash())) + } + + /// Signs the transaction locally and hands the raw bytes to the builders. + async fn send_to_builders( + &self, + tx: TransactionRequest, + signer: eth::Address, + ) -> anyhow::Result { + let envelope = NetworkWallet::::sign_request(&self.transport.wallet, tx) + .await + .context("failed to sign tx for the builders")?; + self.builders.broadcast(&envelope, signer).await + } + /// Queries the mempool for a pending transaction of the given solver and /// nonce. pub async fn find_pending_tx_in_mempool( @@ -216,6 +277,12 @@ impl Mempool { &self.config } + /// Whether the settlement is signed here and handed to block builders + /// instead of being forwarded to the mempool of `config.url`. + pub fn submits_to_builders(&self) -> bool { + !self.builders.is_empty() + } + pub fn reverts_can_get_mined(&self) -> bool { matches!( self.config.revert_protection, diff --git a/crates/driver/src/infra/observe/metrics.rs b/crates/driver/src/infra/observe/metrics.rs index 0e556b2fe9..544432f65b 100644 --- a/crates/driver/src/infra/observe/metrics.rs +++ b/crates/driver/src/infra/observe/metrics.rs @@ -25,6 +25,9 @@ pub struct Metrics { /// atempted and the error detection. #[metric(labels("mempool", "result"))] pub mempool_submission_results_blocks_passed: prometheus::IntCounterVec, + /// The results of broadcasting a settlement tx to a single block builder. + #[metric(labels("mempool", "builder", "result"))] + pub builder_submission: prometheus::IntCounterVec, /// How many orders detected by specific solver and strategy. #[metric(labels("solver"))] pub bad_orders_detected: prometheus::IntCounterVec, diff --git a/crates/driver/src/run.rs b/crates/driver/src/run.rs index 83ce4c0f61..bf10fb2b81 100644 --- a/crates/driver/src/run.rs +++ b/crates/driver/src/run.rs @@ -128,6 +128,7 @@ async fn run_with(args: cli::Args, addr_sender: Option { + write!( + file, + r#"[[submission.mempool]] + url = "{}" + additional-tip-percentage = 0.0 + "#, + blockchain.web3_url, + ) + .unwrap(); + for (index, url) in urls.iter().enumerate() { + write!( + file, + r#"[[submission.mempool.builders]] + name = "builder_{index}" + url = "{}" + "#, + url.as_deref().unwrap_or(&blockchain.web3_url), + ) + .unwrap(); + } + } } } diff --git a/crates/driver/src/tests/setup/mod.rs b/crates/driver/src/tests/setup/mod.rs index 624ad3d5d2..1e15b952d5 100644 --- a/crates/driver/src/tests/setup/mod.rs +++ b/crates/driver/src/tests/setup/mod.rs @@ -512,6 +512,12 @@ pub enum Mempool { url: Option, mines_reverting_txs: bool, }, + /// Signs the tx locally and broadcasts it to the given block builders + /// instead of the node. + Builders { + /// Uses the ethrpc node for the entries that are None + urls: Vec>, + }, } /// Create a builder for the setup process.