diff --git a/README.md b/README.md index ab648f2a..031035ae 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,8 @@ The following environment variables are exposed to configure the Builder: HOST_CHAIN_ID="17000" # Holesky Testnet RU_CHAIN_ID="17001" HOST_RPC_URL="http://host.url.here" +# trailing slash is required +TX_BROADCAST_URLS="http://tx.broadcast.url.here/,https://additional.url.here/" ZENITH_ADDRESS="ZENITH_ADDRESS_HERE" QUINCEY_URL="http://signer.url.here" BUILDER_PORT="8080" diff --git a/src/config.rs b/src/config.rs index f89dd600..73acc322 100644 --- a/src/config.rs +++ b/src/config.rs @@ -14,6 +14,7 @@ use zenith_types::Zenith; const HOST_CHAIN_ID: &str = "HOST_CHAIN_ID"; const RU_CHAIN_ID: &str = "RU_CHAIN_ID"; const HOST_RPC_URL: &str = "HOST_RPC_URL"; +const TX_BROADCAST_URLS: &str = "TX_BROADCAST_URLS"; const ZENITH_ADDRESS: &str = "ZENITH_ADDRESS"; const QUINCEY_URL: &str = "QUINCEY_URL"; const BUILDER_PORT: &str = "BUILDER_PORT"; @@ -42,6 +43,8 @@ pub struct BuilderConfig { pub ru_chain_id: u64, /// URL for Host RPC node. pub host_rpc_url: Cow<'static, str>, + /// Additional RPC URLs to which to broadcast transactions. + pub tx_broadcast_urls: Vec>, /// address of the Zenith contract on Host. pub zenith_address: Address, /// URL for remote Quincey Sequencer server to sign blocks. @@ -133,6 +136,12 @@ impl BuilderConfig { host_chain_id: load_u64(HOST_CHAIN_ID)?, ru_chain_id: load_u64(RU_CHAIN_ID)?, host_rpc_url: load_url(HOST_RPC_URL)?, + tx_broadcast_urls: env::var(TX_BROADCAST_URLS) + .unwrap_or_default() + .split(',') + .map(ToOwned::to_owned) + .map(Into::into) + .collect(), zenith_address: load_address(ZENITH_ADDRESS)?, quincey_url: load_url(QUINCEY_URL)?, builder_port: load_u16(BUILDER_PORT)?, @@ -180,6 +189,20 @@ impl BuilderConfig { .map_err(Into::into) } + pub async fn connect_additional_broadcast( + &self, + ) -> Result>, ConfigError> { + let mut providers = Vec::with_capacity(self.tx_broadcast_urls.len()); + for url in self.tx_broadcast_urls.iter() { + let provider = ProviderBuilder::new() + .on_builtin(url) + .await + .map_err(Into::::into)?; + providers.push(provider); + } + Ok(providers) + } + pub fn connect_zenith(&self, provider: Provider) -> ZenithInstance { Zenith::new(self.zenith_address, provider) } diff --git a/src/tasks/submit.rs b/src/tasks/submit.rs index ea153b8c..04986e3e 100644 --- a/src/tasks/submit.rs +++ b/src/tasks/submit.rs @@ -3,13 +3,13 @@ use crate::{ signer::LocalOrAws, tasks::block::InProgressBlock, }; -use alloy::consensus::SimpleCoder; use alloy::network::{TransactionBuilder, TransactionBuilder4844}; use alloy::providers::{Provider as _, WalletProvider}; use alloy::rpc::types::eth::TransactionRequest; use alloy::signers::Signer; use alloy::sol_types::SolCall; use alloy::transports::TransportError; +use alloy::{consensus::SimpleCoder, providers::SendableTx}; use alloy_primitives::{FixedBytes, U256}; use eyre::bail; use oauth2::{ @@ -17,9 +17,22 @@ use oauth2::{ ClientSecret, EmptyExtraTokenFields, StandardTokenResponse, TokenResponse, TokenUrl, }; use tokio::{sync::mpsc, task::JoinHandle}; -use tracing::{debug, error, instrument, trace}; +use tracing::{debug, error, instrument, trace, warn}; use zenith_types::{SignRequest, SignResponse, Zenith}; +macro_rules! spawn_provider_send { + ($provider:expr, $tx:expr) => { + let p = $provider.clone(); + let t = $tx.clone(); + tokio::spawn(async move { + if let Err(e) = p.send_tx_envelope(t).await { + warn!(%e, "error in transaction broadcast"); + } + }); + + }; +} + /// OAuth Audience Claim Name, required param by IdP for client credential grant const OAUTH_AUDIENCE_CLAIM: &str = "audience"; @@ -161,23 +174,40 @@ impl SubmitTask { bail!("bailing transaction submission") } + self.send_transaction(resp, tx).await?; + + Ok(()) + } + + async fn send_transaction( + &self, + resp: &SignResponse, + tx: TransactionRequest, + ) -> Result<(), eyre::Error> { tracing::debug!( host_block_number = %resp.req.host_block_number, gas_limit = %resp.req.gas_limit, "sending transaction to network" ); - let result = self.provider.send_transaction(tx).await?; + let SendableTx::Envelope(tx) = self.provider.fill(tx).await? else { + bail!("failed to fill transaction") + }; - let tx_hash = result.tx_hash(); + // Send the tx via the primary provider + spawn_provider_send!(&self.provider, &tx); + + // Spawn send_tx futures for all additional broadcast providers + for provider in self.config.connect_additional_broadcast().await? { + spawn_provider_send!(&provider, &tx); + } tracing::info!( - %tx_hash, + tx_hash = %tx.tx_hash(), ru_chain_id = %resp.req.ru_chain_id, gas_limit = %resp.req.gas_limit, "dispatched to network" ); - Ok(()) } diff --git a/tests/tx_poller_test.rs b/tests/tx_poller_test.rs index f9378b9d..11e8c889 100644 --- a/tests/tx_poller_test.rs +++ b/tests/tx_poller_test.rs @@ -68,6 +68,7 @@ mod tests { host_chain_id: 17000, ru_chain_id: 17001, host_rpc_url: "http://rpc.holesky.signet.sh".into(), + tx_broadcast_urls: vec!["http://localhost:9000".into()], zenith_address: Address::default(), quincey_url: "http://localhost:8080".into(), builder_port: 8080,