[EPROD-1061][EPROD-1083] Pool and batching for eth_sendRawTransaction calls

bin/reth/Cargo.toml

@@ -100,26 +100,27 @@ similar-asserts.workspace = true
 async-channel.workspace = true
 candid.workspace = true
 did.workspace = true
+revm-primitives.workspace = true
 evm-canister-client = { workspace = true, features = ["ic-agent-client"] }
 lightspeed_scheduler = { workspace = true, features = ["tracing"] }
-# rlp = { workspace = true }
+ethereum-json-rpc-client = { workspace = true, features = ["reqwest"] }
 
 [dev-dependencies]
 tempfile.workspace = true
 
 # bitfinity dev dependencies
 async-trait = { workspace = true }
 dirs.workspace = true
-ethereum-json-rpc-client = { workspace = true, features = ["reqwest"] }
 jsonrpsee = { workspace = true }
 parking_lot = { workspace = true }
 rand = { workspace = true }
 reth-primitives = { workspace = true, features = ["test-utils"] }
 reth-trie = { workspace = true, features = ["test-utils"] }
-revm-primitives.workspace = true
+reth-transaction-pool = { workspace = true, features = ["test-utils"] }
 reth-db-common.workspace = true
 reth-db = { workspace = true, features = ["mdbx", "test-utils"] }
 serial_test.workspace = true
+reth-discv5.workspace = true
 
 [features]
 default = ["jemalloc"]

bin/reth/src/bitfinity_tasks.rs

@@ -0,0 +1,3 @@
+//! This module contains tasks to run in parallel with reth node.
+
+pub mod send_txs;

bin/reth/src/bitfinity_tasks/send_txs.rs

@@ -0,0 +1,153 @@
+//! Utils for raw transaction batching.
+
+use std::time::Duration;
+
+use did::H256;
+use ethereum_json_rpc_client::reqwest::ReqwestClient;
+use ethereum_json_rpc_client::{EthJsonRpcClient, Id, Params};
+use eyre::eyre;
+use futures::future::join_all;
+use lightspeed_scheduler::job::Job;
+use lightspeed_scheduler::scheduler::Scheduler;
+use lightspeed_scheduler::JobExecutor;
+use reth_node_core::version::SHORT_VERSION;
+use reth_rpc_api::eth::helpers::bitfinity_tx_forwarder::SharedQueue;
+use revm_primitives::{hex, U256};
+use tracing::{info, trace, warn};
+
+/// Periodically sends transactions from priority queue.
+#[derive(Debug, Clone)]
+pub struct BitfinityTransactionSender {
+    queue: SharedQueue,
+    rpc_url: String,
+    period: Duration,
+    batch_size: usize,
+    txs_per_execution_threshold: usize,
+}
+
+impl BitfinityTransactionSender {
+    /// Creates new instance of the transaction sender.
+    pub const fn new(
+        queue: SharedQueue,
+        rpc_url: String,
+        period: Duration,
+        batch_size: usize,
+        txs_per_execution_threshold: usize,
+    ) -> Self {
+        Self { queue, rpc_url, period, batch_size, txs_per_execution_threshold }
+    }
+
+    /// Schedule the transaction sending job and return a handle to it.
+    pub async fn schedule_execution(
+        self,
+        job_executor: Option<JobExecutor>,
+    ) -> eyre::Result<(JobExecutor, tokio::task::JoinHandle<()>)> {
+        info!(target: "reth::cli - BitfinityTransactionSender", "reth {} starting", SHORT_VERSION);
+
+        let job_executor = job_executor.unwrap_or_else(JobExecutor::new_with_local_tz);
+
+        // Schedule the import job
+        {
+            let interval =
+                Scheduler::Interval { interval_duration: self.period, execute_at_startup: true };
+            job_executor
+                .add_job_with_scheduler(
+                    interval,
+                    Job::new("send transactions", "bitfinity tx sending", None, move || {
+                        let tx_sender = self.clone();
+                        Box::pin(async move {
+                            tx_sender.single_execution().await?;
+                            Ok(())
+                        })
+                    }),
+                )
+                .await;
+        }
+
+        let job_handle = job_executor.run().await?;
+        Ok((job_executor, job_handle))
+    }
+
+    /// Execute the transaction sending job.
+    pub async fn single_execution(&self) -> eyre::Result<()> {
+        let mut to_send = self.get_transactions_to_send().await;
+        let batch_size = self.batch_size.max(1);
+        let mut send_futures = vec![];
+
+        loop {
+            let last_idx = batch_size.min(to_send.len());
+            if last_idx == 0 {
+                break;
+            }
+
+            let to_send_batch: Vec<_> = to_send.drain(..last_idx).collect();
+
+            let send_future = async move {
+                let result = match to_send_batch.len() {
+                    0 => return,
+                    1 => self.send_single_tx(&to_send_batch[0].1).await,
+                    _ => self.send_txs_batch(&to_send_batch).await,
+                };
+
+                if let Err(e) = result {
+                    warn!("Failed to send transactions to EVM: {e}");
+                }
+            };
+            send_futures.push(send_future);
+        }
+
+        join_all(send_futures).await;
+
+        Ok(())
+    }
+
+    async fn get_transactions_to_send(&self) -> Vec<(U256, Vec<u8>)> {
+        let mut batch = Vec::with_capacity(self.txs_per_execution_threshold);
+        let mut queue = self.queue.lock().await;
+        let txs_to_pop = self.txs_per_execution_threshold.max(1); // if batch size is zero, take at least one tx.
+
+        for _ in 0..txs_to_pop {
+            let Some(entry) = queue.pop_tx_with_highest_price() else {
+                break;
+            };
+
+            batch.push(entry);
+        }
+
+        batch
+    }
+
+    async fn send_single_tx(&self, to_send: &[u8]) -> Result<(), eyre::Error> {
+        let client = self.get_client()?;
+        let hash = client
+            .send_raw_transaction_bytes(to_send)
+            .await
+            .map_err(|e| eyre!("failed to send single transaction: {e}"))?;
+
+        trace!("Single transaction with hash {hash} sent.");
+
+        Ok(())
+    }
+
+    async fn send_txs_batch(&self, to_send: &[(U256, Vec<u8>)]) -> Result<(), eyre::Error> {
+        let client = self.get_client()?;
+
+        let params =
+            to_send.iter().map(|(_, raw)| (Params::Array(vec![hex::encode(raw).into()]), Id::Null));
+        let max_batch_size = usize::MAX;
+        let hashes = client
+            .batch_request::<H256>("eth_sendRawTransaction".into(), params, max_batch_size)
+            .await
+            .map_err(|e| eyre!("failed to send single transaction: {e}"))?;
+
+        trace!("Raw transactions batch sent. Hashes: {hashes:?}");
+
+        Ok(())
+    }
+
+    fn get_client(&self) -> eyre::Result<EthJsonRpcClient<ReqwestClient>> {
+        let client = EthJsonRpcClient::new(ReqwestClient::new(self.rpc_url.clone()));
+
+        Ok(client)
+    }
+}

bin/reth/src/commands/bitfinity_import.rs

@@ -99,10 +99,11 @@ impl BitfinityImportCommand {
     /// Schedule the import job and return a handle to it.
     pub async fn schedule_execution(
         self,
+        job_executor: Option<JobExecutor>,
     ) -> eyre::Result<(JobExecutor, tokio::task::JoinHandle<()>)> {
         info!(target: "reth::cli - BitfinityImportCommand", "reth {} starting", SHORT_VERSION);
 
-        let job_executor = JobExecutor::new_with_local_tz();
+        let job_executor = job_executor.unwrap_or_else(JobExecutor::new_with_local_tz);
 
         // Schedule the import job
         {

bin/reth/src/commands/mod.rs

@@ -1,5 +1,5 @@
 //! This contains all of the `reth` commands
 
-pub mod bitfinity_reset_evm_state;
 pub mod bitfinity_import;
+pub mod bitfinity_reset_evm_state;
 pub mod debug_cmd;

bin/reth/src/lib.rs

@@ -27,6 +27,7 @@
 #![cfg_attr(not(test), warn(unused_crate_dependencies))]
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
+pub mod bitfinity_tasks;
 pub mod cli;
 pub mod commands;
 

bin/reth/src/main.rs

@@ -3,7 +3,10 @@
 #[global_allocator]
 static ALLOC: reth_cli_util::allocator::Allocator = reth_cli_util::allocator::new_allocator();
 
+use std::{sync::Arc, time::Duration};
+
 use clap::{Args, Parser};
+use reth::bitfinity_tasks::send_txs::BitfinityTransactionSender;
 use reth_ethereum_cli::chainspec::EthereumChainSpecParser;
 use reth_node_builder::{
     engine_tree_config::{
@@ -13,7 +16,9 @@ use reth_node_builder::{
 };
 use reth_node_ethereum::node::EthereumAddOns;
 use reth_provider::providers::BlockchainProvider2;
+use reth_rpc_api::eth::helpers::bitfinity_tx_forwarder::{BitfinityTransactionsForwarder, TransactionsPriorityQueue};
 use reth_tracing::tracing::warn;
+use tokio::sync::Mutex;
 use tracing::info;
 
 /// Parameters for configuring the engine
@@ -102,19 +107,41 @@ fn main() {
                     let datadir = handle.node.data_dir.clone();
                     let (provider_factory, bitfinity) = handle.bitfinity_import.clone().expect("Bitfinity import not configured");
 
+
                     // Init bitfinity import
-                    {
+                    let executor = {
                         let import = BitfinityImportCommand::new(
                             config,
                             datadir,
                             chain,
-                            bitfinity,
+                            bitfinity.clone(),
                             provider_factory,
                             blockchain_provider,
                         );
-                        let _import_handle = import.schedule_execution().await?;
+                        let (executor, _import_handle) = import.schedule_execution(None).await?;
+                        executor
                     };
 
+                    if bitfinity.tx_queue {
+                        let queue = Arc::new(Mutex::new(TransactionsPriorityQueue::new(1000)));
+
+                        // Make EthApi handler move new txs to queue.
+                        let queue_clone = Arc::clone(&queue);
+                        handle.node.add_ons_handle.eth_api().set_bitfinity_tx_forwarder(BitfinityTransactionsForwarder::new(queue_clone));
+
+                        // Run batch transaction sender.
+                        let url = bitfinity.send_raw_transaction_rpc_url.unwrap_or(bitfinity.rpc_url);                        
+                        let period = Duration::from_secs(bitfinity.send_queued_txs_period_secs);
+                        let transaction_sending = BitfinityTransactionSender::new(
+                            queue,
+                            url,
+                            period,
+                            bitfinity.queued_txs_batch_size,
+                            bitfinity.queued_txs_per_execution_threshold,
+                        );
+                        let _sending_handle = transaction_sending.schedule_execution(Some(executor)).await?;
+                    }
+
                     handle.node_exit_future.await
                 }
             }

bin/reth/tests/bitfinity_it.rs

@@ -1,4 +1,4 @@
 //!
 //! bitfinity integration tests
-//! 
-pub mod commands;
+//!
+pub mod commands;