Feat/multiple l1 rpcs (#567)

* Propagating tx to many RPC nodes

Signed-off-by: Maciej Skrzypkowski <mskr@gmx.com>

* tests

* v1.1.0

* fixed error logs

* updated env var  L1_RPC_URLS

* updated .env.sample

---------

Signed-off-by: Maciej Skrzypkowski <mskr@gmx.com>

Author: mskrzypkows

Cargo.lock

@@ -10,7 +10,7 @@ resolver = "2"
 default-members = ["node"]
 
 [workspace.package]
-version = "1.0.1"
+version = "1.1.0"
 edition = "2024"
 repository = "https://github.com/NethermindEth/Catalyst"
 license = "MIT"

Cargo.toml

@@ -4,10 +4,10 @@ PRECONF_ROUTER_ADDRESS=0x8dDEA87FfA296B951881aDacbAe8011090C54cAC
 TAIKO_INBOX_ADDRESS=0xc2DD6e8DC8d0558F00Cc1FA6A16FFF1A62Cc436B
 TAIKO_ANCHOR_ADDRESS=0x1670100000000000000000000000000000010001
 VALIDATOR_INDEX=1
-TAIKO_GETH_WS_RPC_URL=ws://127.0.0.1:8546
+TAIKO_GETH_RPC_URL=ws://127.0.0.1:8546
 TAIKO_GETH_AUTH_RPC_URL=http://127.0.0.1:8551
 TAIKO_DRIVER_URL=http://127.0.0.1:1235
-L1_WS_RPC_URL=ws://127.0.0.1:32003
+L1_RPC_URLS=ws://127.0.0.1:32003,wss://123.123.123.123:32001
 L1_BEACON_URL=http://127.0.0.1:33001
 RUST_LOG=debug,reqwest=info,hyper=info,alloy_transport=info,alloy_rpc_client=info,alloy_provider=info
 JWT_SECRET_FILE_PATH=/some/path/jwtsecret.hex

node/.env.sample

@@ -36,7 +36,7 @@ impl TryFrom<L1ContractAddresses> for ContractAddresses {
 }
 
 pub struct EthereumL1Config {
-    pub execution_rpc_url: String,
+    pub execution_rpc_urls: Vec<String>,
     pub contract_addresses: ContractAddresses,
     pub consensus_rpc_url: String,
     pub min_priority_fee_per_gas_wei: u64,

node/src/ethereum_l1/config.rs

@@ -51,7 +51,10 @@ impl ExecutionLayer {
     ) -> Result<Self, Error> {
         let (provider, preconfer_address) = alloy_tools::construct_alloy_provider(
             &config.signer,
-            &config.execution_rpc_url,
+            config
+                .execution_rpc_urls
+                .first()
+                .ok_or_else(|| anyhow!("L1 RPC URL is required"))?,
             config.preconfer_address,
         )
         .await?;
@@ -508,7 +511,7 @@ impl ExecutionLayer {
         let metrics = Arc::new(Metrics::new());
 
         let ethereum_l1_config = EthereumL1Config {
-            execution_rpc_url: ws_rpc_url,
+            execution_rpc_urls: vec![ws_rpc_url],
             contract_addresses: ContractAddresses {
                 taiko_inbox: Address::ZERO,
                 taiko_token: OnceCell::new(),

node/src/ethereum_l1/execution_layer.rs

@@ -1,9 +1,12 @@
 use super::{config::EthereumL1Config, tools, transaction_error::TransactionError};
-use crate::metrics::Metrics;
+use crate::{
+    metrics::Metrics,
+    shared::{alloy_tools, signer::Signer},
+};
 use alloy::{
     consensus::TxType,
     network::{Network, ReceiptResponse, TransactionBuilder, TransactionBuilder4844},
-    primitives::B256,
+    primitives::{Address, B256},
     providers::{
         DynProvider, PendingTransactionBuilder, PendingTransactionError, Provider, RootProvider,
         WatchTxError,
@@ -34,6 +37,9 @@ pub struct TransactionMonitorConfig {
     max_attempts_to_send_tx: u64,
     max_attempts_to_wait_tx: u64,
     delay_between_tx_attempts: Duration,
+    execution_rpc_urls: Vec<String>,
+    preconfer_address: Option<Address>,
+    signer: Arc<Signer>,
 }
 
 pub struct TransactionMonitorThread {
@@ -74,6 +80,9 @@ impl TransactionMonitor {
                 delay_between_tx_attempts: Duration::from_secs(
                     config.delay_between_tx_attempts_sec,
                 ),
+                execution_rpc_urls: config.execution_rpc_urls.clone(),
+                preconfer_address: config.preconfer_address,
+                signer: config.signer.clone(),
             },
             join_handle: Mutex::new(None),
             error_notification_channel,
@@ -229,7 +238,7 @@ impl TransactionMonitorThread {
                 root_provider = Some(pending_tx.provider().clone());
             }
 
-            debug!(
+            info!(
                 "{} tx nonce: {}, attempt: {}, l1_block: {}, hash: {},  max_fee_per_gas: {}, max_priority_fee_per_gas: {}, max_fee_per_blob_gas: {:?}",
                 if sending_attempt == 0 {
                     "🟢 Send"
@@ -379,8 +388,11 @@ impl TransactionMonitorThread {
         previous_tx_hashes: &Vec<B256>,
         sending_attempt: u64,
     ) -> Option<PendingTransactionBuilder<alloy::network::Ethereum>> {
-        match self.provider.send_transaction(tx).await {
-            Ok(tx) => Some(tx),
+        match self.provider.send_transaction(tx.clone()).await {
+            Ok(pending_tx) => {
+                self.propagate_transaction_to_other_backup_nodes(tx).await;
+                Some(pending_tx)
+            }
             Err(e) => {
                 self.handle_rpc_error(e, previous_tx_hashes, sending_attempt)
                     .await;
@@ -389,6 +401,41 @@ impl TransactionMonitorThread {
         }
     }
 
+    /// Recreates each backup node every time to avoid connection issues
+    async fn propagate_transaction_to_other_backup_nodes(&self, tx: TransactionRequest) {
+        // Skip the first RPC URL since it is the main one
+        for url in self.config.execution_rpc_urls.iter().skip(1) {
+            let provider = alloy_tools::construct_alloy_provider(
+                &self.config.signer,
+                url,
+                self.config.preconfer_address,
+            )
+            .await;
+            match provider {
+                Ok(provider) => {
+                    let tx = provider.0.send_transaction(tx.clone()).await;
+                    if let Err(e) = tx {
+                        if e.to_string().contains("AlreadyKnown")
+                            || e.to_string().to_lowercase().contains("already known")
+                        {
+                            debug!("Transaction already known to backup node {}", url);
+                        } else {
+                            warn!("Failed to send transaction to backup node {}: {}", url, e);
+                        }
+                    } else {
+                        info!("Transaction sent to backup node {}", url);
+                    }
+                }
+                Err(e) => {
+                    warn!(
+                        "Failed to construct alloy provider for backup node {}: {}",
+                        url, e
+                    );
+                }
+            }
+        }
+    }
+
     async fn handle_rpc_error(
         &self,
         e: RpcError<TransportErrorKind>,

node/src/ethereum_l1/monitor_transaction.rs

@@ -72,7 +72,7 @@ async fn main() -> Result<(), Error> {
 
     let ethereum_l1 = ethereum_l1::EthereumL1::new(
         ethereum_l1::config::EthereumL1Config {
-            execution_rpc_url: config.l1_rpc_url.clone(),
+            execution_rpc_urls: config.l1_rpc_urls.clone(),
             contract_addresses: config.contract_addresses.clone().try_into()?,
             consensus_rpc_url: config.l1_beacon_url,
             slot_duration_sec: config.l1_slot_duration_sec,
@@ -172,7 +172,11 @@ async fn main() -> Result<(), Error> {
 
     let chain_monitor = Arc::new(
         chain_monitor::ChainMonitor::new(
-            config.l1_rpc_url,
+            config
+                .l1_rpc_urls
+                .first()
+                .expect("L1 RPC URL is required")
+                .clone(),
             config.taiko_geth_rpc_url,
             config.contract_addresses.taiko_inbox,
             cancel_token.clone(),

node/src/main.rs

@@ -1,6 +1,7 @@
 use super::web3signer::Web3Signer;
 use std::sync::Arc;
 
+#[derive(Debug)]
 pub enum Signer {
     Web3signer(Arc<Web3Signer>),
     PrivateKey(String),

node/src/shared/signer.rs

@@ -10,7 +10,7 @@ pub struct Config {
     pub taiko_driver_url: String,
     pub catalyst_node_ecdsa_private_key: Option<String>,
     pub mev_boost_url: String,
-    pub l1_rpc_url: String,
+    pub l1_rpc_urls: Vec<String>,
     pub l1_beacon_url: String,
     pub web3signer_l1_url: Option<String>,
     pub web3signer_l2_url: Option<String>,
@@ -366,7 +366,11 @@ impl Config {
             catalyst_node_ecdsa_private_key,
             mev_boost_url: std::env::var("MEV_BOOST_URL")
                 .unwrap_or("http://127.0.0.1:8080".to_string()),
-            l1_rpc_url: std::env::var("L1_RPC_URL").unwrap_or("wss://127.0.0.1".to_string()),
+            l1_rpc_urls: std::env::var("L1_RPC_URLS")
+                .unwrap_or("wss://127.0.0.1".to_string())
+                .split(",")
+                .map(|s| s.to_string())
+                .collect(),
             l1_beacon_url: std::env::var("L1_BEACON_URL")
                 .unwrap_or("http://127.0.0.1:4000".to_string()),
             web3signer_l1_url,
@@ -459,7 +463,14 @@ propose_forced_inclusion: {}
             config.taiko_geth_auth_rpc_url,
             config.taiko_driver_url,
             config.mev_boost_url,
-            config.l1_rpc_url,
+            match config.l1_rpc_urls.split_first() {
+                Some((first, rest)) => {
+                    let mut urls = vec![format!("{} (main)", first)];
+                    urls.extend(rest.iter().cloned());
+                    urls.join(", ")
+                }
+                None => String::new(),
+            },
             config.l1_beacon_url,
             config.web3signer_l1_url.as_deref().unwrap_or("not set"),
             config.web3signer_l2_url.as_deref().unwrap_or("not set"),