feat(txpool): Batch insertions into the Tx Pool

crates/node/builder/src/rpc.rs

@@ -18,6 +18,7 @@ use reth_node_api::{
     NodeAddOns, NodeTypes, PayloadTypes, PayloadValidator, PrimitivesTy, TreeConfig,
 };
 use reth_node_core::{
+    cli::config::RethTransactionPoolConfig,
     node_config::NodeConfig,
     version::{version_metadata, CLIENT_CODE},
 };
@@ -863,7 +864,8 @@ where
             }),
         );
 
-        let ctx = EthApiCtx { components: &node, config: config.rpc.eth_config(), cache };
+        let eth_config = config.rpc.eth_config().max_batch_size(config.txpool.max_batch_size());
+        let ctx = EthApiCtx { components: &node, config: eth_config, cache };
         let eth_api = eth_api_builder.build_eth_api(ctx).await?;
 
         let auth_config = config.rpc.auth_server_config(jwt_secret)?;
@@ -1040,6 +1042,7 @@ impl<'a, N: FullNodeComponents<Types: NodeTypes<ChainSpec: EthereumHardforks>>>
             .fee_history_cache_config(self.config.fee_history_cache)
             .proof_permits(self.config.proof_permits)
             .gas_oracle_config(self.config.gas_oracle)
+            .max_batch_size(self.config.max_batch_size)
     }
 }
 

crates/node/core/src/args/txpool.rs

@@ -132,6 +132,10 @@ pub struct TxPoolArgs {
         conflicts_with = "transactions_backup_path"
     )]
     pub disable_transactions_backup: bool,
+
+    /// Max batch size for transaction pool insertions
+    #[arg(long = "txpool.max-batch-size", default_value_t = 1)]
+    pub max_batch_size: usize,
 }
 
 impl Default for TxPoolArgs {
@@ -165,6 +169,7 @@ impl Default for TxPoolArgs {
             max_queued_lifetime: MAX_QUEUED_TRANSACTION_LIFETIME,
             transactions_backup_path: None,
             disable_transactions_backup: false,
+            max_batch_size: 1,
         }
     }
 }
@@ -209,6 +214,11 @@ impl RethTransactionPoolConfig for TxPoolArgs {
             max_queued_lifetime: self.max_queued_lifetime,
         }
     }
+
+    /// Returns max batch size for transaction batch insertion.
+    fn max_batch_size(&self) -> usize {
+        self.max_batch_size
+    }
 }
 
 #[cfg(test)]

crates/node/core/src/cli/config.rs

@@ -87,4 +87,7 @@ impl<N: NetworkPrimitives> RethNetworkConfig for reth_network::NetworkManager<N>
 pub trait RethTransactionPoolConfig {
     /// Returns transaction pool configuration.
     fn pool_config(&self) -> PoolConfig;
+
+    /// Returns max batch size for transaction batch insertion.
+    fn max_batch_size(&self) -> usize;
 }

crates/optimism/rpc/src/eth/transaction.rs

@@ -46,11 +46,8 @@ where
                 })?;
 
             // Retain tx in local tx pool after forwarding, for local RPC usage.
-            let _ = self
-                .pool()
-                .add_transaction(TransactionOrigin::Local, pool_transaction)
-                .await.inspect_err(|err| {
-                    tracing::warn!(target: "rpc::eth", %err, %hash, "successfully sent tx to sequencer, but failed to persist in local tx pool");
+            let _ = self.inner.eth_api.add_pool_transaction(pool_transaction).await.inspect_err(|err| {
+                tracing::warn!(target: "rpc::eth", %err, %hash, "successfully sent tx to sequencer, but failed to persist in local tx pool");
             });
 
             return Ok(hash)

crates/rpc/rpc-eth-types/src/builder/config.rs

@@ -45,6 +45,8 @@ pub struct EthConfig {
     pub fee_history_cache: FeeHistoryCacheConfig,
     /// The maximum number of getproof calls that can be executed concurrently.
     pub proof_permits: usize,
+    /// Maximum batch size for transaction pool insertions.
+    pub max_batch_size: usize,
 }
 
 impl EthConfig {
@@ -72,6 +74,7 @@ impl Default for EthConfig {
             stale_filter_ttl: DEFAULT_STALE_FILTER_TTL,
             fee_history_cache: FeeHistoryCacheConfig::default(),
             proof_permits: DEFAULT_PROOF_PERMITS,
+            max_batch_size: 1,
         }
     }
 }
@@ -136,6 +139,12 @@ impl EthConfig {
         self.proof_permits = permits;
         self
     }
+
+    /// Configures the maximum batch size for transaction pool insertions
+    pub const fn max_batch_size(mut self, max_batch_size: usize) -> Self {
+        self.max_batch_size = max_batch_size;
+        self
+    }
 }
 
 /// Config for the filter

crates/rpc/rpc-eth-types/src/error/mod.rs

@@ -24,6 +24,7 @@ use revm::context_interface::result::{
 };
 use revm_inspectors::tracing::MuxError;
 use std::convert::Infallible;
+use tokio::sync::oneshot::error::RecvError;
 use tracing::error;
 
 /// A trait to convert an error to an RPC error.
@@ -166,6 +167,12 @@ pub enum EthApiError {
         /// Duration that was waited before timing out
         duration: Duration,
     },
+    /// Error thrown when batch tx response channel fails
+    #[error(transparent)]
+    BatchTxRecvError(#[from] RecvError),
+    /// Error thrown when batch tx send channel fails
+    #[error("Batch transaction sender channel closed")]
+    BatchTxSendError,
     /// Any other error
     #[error("{0}")]
     Other(Box<dyn ToRpcError>),
@@ -279,6 +286,10 @@ impl From<EthApiError> for jsonrpsee_types::error::ErrorObject<'static> {
             EthApiError::PrunedHistoryUnavailable => rpc_error_with_code(4444, error.to_string()),
             EthApiError::Other(err) => err.to_rpc_error(),
             EthApiError::MuxTracerError(msg) => internal_rpc_err(msg.to_string()),
+            EthApiError::BatchTxRecvError(err) => internal_rpc_err(err.to_string()),
+            EthApiError::BatchTxSendError => {
+                internal_rpc_err("Batch transaction sender channel closed".to_string())
+            }
         }
     }
 }

crates/rpc/rpc/src/eth/builder.rs

@@ -40,6 +40,7 @@ pub struct EthApiBuilder<N: RpcNodeCore, Rpc, NextEnv = ()> {
     blocking_task_pool: Option<BlockingTaskPool>,
     task_spawner: Box<dyn TaskSpawner + 'static>,
     next_env: NextEnv,
+    max_batch_size: usize,
 }
 
 impl<Provider, Pool, Network, EvmConfig, ChainSpec>
@@ -78,6 +79,7 @@ impl<N: RpcNodeCore, Rpc, NextEnv> EthApiBuilder<N, Rpc, NextEnv> {
             blocking_task_pool,
             task_spawner,
             next_env,
+            max_batch_size,
         } = self;
         EthApiBuilder {
             components,
@@ -94,6 +96,7 @@ impl<N: RpcNodeCore, Rpc, NextEnv> EthApiBuilder<N, Rpc, NextEnv> {
             blocking_task_pool,
             task_spawner,
             next_env,
+            max_batch_size,
         }
     }
 }
@@ -121,6 +124,7 @@ where
             gas_oracle_config: Default::default(),
             eth_state_cache_config: Default::default(),
             next_env: Default::default(),
+            max_batch_size: 1,
         }
     }
 }
@@ -155,6 +159,7 @@ where
             task_spawner,
             gas_oracle_config,
             next_env,
+            max_batch_size,
         } = self;
         EthApiBuilder {
             components,
@@ -171,6 +176,7 @@ where
             task_spawner,
             gas_oracle_config,
             next_env,
+            max_batch_size,
         }
     }
 
@@ -194,6 +200,7 @@ where
             task_spawner,
             gas_oracle_config,
             next_env: _,
+            max_batch_size,
         } = self;
         EthApiBuilder {
             components,
@@ -210,6 +217,7 @@ where
             task_spawner,
             gas_oracle_config,
             next_env,
+            max_batch_size,
         }
     }
 
@@ -281,6 +289,12 @@ where
         self
     }
 
+    /// Sets the max batch size for batching transaction insertions.
+    pub const fn max_batch_size(mut self, max_batch_size: usize) -> Self {
+        self.max_batch_size = max_batch_size;
+        self
+    }
+
     /// Builds the [`EthApiInner`] instance.
     ///
     /// If not configured, this will spawn the cache backend: [`EthStateCache::spawn`].
@@ -309,6 +323,7 @@ where
             proof_permits,
             task_spawner,
             next_env,
+            max_batch_size,
         } = self;
 
         let provider = components.provider().clone();
@@ -345,6 +360,7 @@ where
             proof_permits,
             rpc_converter,
             next_env,
+            max_batch_size,
         )
     }
 

crates/rpc/rpc/src/eth/core.rs

@@ -28,8 +28,11 @@ use reth_tasks::{
     pool::{BlockingTaskGuard, BlockingTaskPool},
     TaskSpawner, TokioTaskExecutor,
 };
-use reth_transaction_pool::noop::NoopTransactionPool;
-use tokio::sync::{broadcast, Mutex};
+use reth_transaction_pool::{
+    noop::NoopTransactionPool, AddedTransactionOutcome, BatchTxProcessor, BatchTxRequest,
+    TransactionPool,
+};
+use tokio::sync::{broadcast, mpsc, Mutex};
 
 const DEFAULT_BROADCAST_CAPACITY: usize = 2000;
 
@@ -147,6 +150,7 @@ where
         fee_history_cache: FeeHistoryCache<ProviderHeader<N::Provider>>,
         proof_permits: usize,
         rpc_converter: Rpc,
+        max_batch_size: usize,
     ) -> Self {
         let inner = EthApiInner::new(
             components,
@@ -161,6 +165,7 @@ where
             proof_permits,
             rpc_converter,
             (),
+            max_batch_size,
         );
 
         Self { inner: Arc::new(inner) }
@@ -290,6 +295,10 @@ pub struct EthApiInner<N: RpcNodeCore, Rpc: RpcConvert> {
 
     /// Builder for pending block environment.
     next_env_builder: Box<dyn PendingEnvBuilder<N::Evm>>,
+
+    /// Transaction batch sender for batching tx insertions
+    tx_batch_sender:
+        mpsc::UnboundedSender<BatchTxRequest<<N::Pool as TransactionPool>::Transaction>>,
 }
 
 impl<N, Rpc> EthApiInner<N, Rpc>
@@ -312,6 +321,7 @@ where
         proof_permits: usize,
         tx_resp_builder: Rpc,
         next_env: impl PendingEnvBuilder<N::Evm>,
+        max_batch_size: usize,
     ) -> Self {
         let signers = parking_lot::RwLock::new(Default::default());
         // get the block number of the latest block
@@ -327,6 +337,11 @@ where
 
         let (raw_tx_sender, _) = broadcast::channel(DEFAULT_BROADCAST_CAPACITY);
 
+        // Create tx pool insertion batcher
+        let (processor, tx_batch_sender) =
+            BatchTxProcessor::new(components.pool().clone(), max_batch_size);
+        task_spawner.spawn_critical("tx-batcher", Box::pin(processor));
+
         Self {
             components,
             signers,
@@ -344,6 +359,7 @@ where
             raw_tx_sender,
             tx_resp_builder,
             next_env_builder: Box::new(next_env),
+            tx_batch_sender,
         }
     }
 }
@@ -473,6 +489,30 @@ where
     pub fn broadcast_raw_transaction(&self, raw_tx: Bytes) {
         let _ = self.raw_tx_sender.send(raw_tx);
     }
+
+    /// Returns the transaction batch sender
+    #[inline]
+    const fn tx_batch_sender(
+        &self,
+    ) -> &mpsc::UnboundedSender<BatchTxRequest<<N::Pool as TransactionPool>::Transaction>> {
+        &self.tx_batch_sender
+    }
+
+    /// Adds an _unvalidated_ transaction into the pool via the transaction batch sender.
+    #[inline]
+    pub async fn add_pool_transaction(
+        &self,
+        transaction: <N::Pool as TransactionPool>::Transaction,
+    ) -> Result<AddedTransactionOutcome, EthApiError> {
+        let (response_tx, response_rx) = tokio::sync::oneshot::channel();
+        let request = reth_transaction_pool::BatchTxRequest::new(transaction, response_tx);
+
+        self.tx_batch_sender()
+            .send(request)
+            .map_err(|_| reth_rpc_eth_types::EthApiError::BatchTxSendError)?;
+
+        Ok(response_rx.await??)
+    }
 }
 
 #[cfg(test)]

crates/rpc/rpc/src/eth/helpers/transaction.rs

@@ -8,9 +8,7 @@ use reth_rpc_eth_api::{
     FromEvmError, RpcNodeCore,
 };
 use reth_rpc_eth_types::{utils::recover_raw_transaction, EthApiError};
-use reth_transaction_pool::{
-    AddedTransactionOutcome, PoolTransaction, TransactionOrigin, TransactionPool,
-};
+use reth_transaction_pool::{AddedTransactionOutcome, PoolTransaction, TransactionPool};
 
 impl<N, Rpc> EthTransactions for EthApi<N, Rpc>
 where
@@ -34,9 +32,8 @@ where
 
         let pool_transaction = <Self::Pool as TransactionPool>::Transaction::from_pooled(recovered);
 
-        // submit the transaction to the pool with a `Local` origin
         let AddedTransactionOutcome { hash, .. } =
-            self.pool().add_transaction(TransactionOrigin::Local, pool_transaction).await?;
+            self.inner.add_pool_transaction(pool_transaction).await?;
 
         Ok(hash)
     }

crates/transaction-pool/Cargo.toml

@@ -34,6 +34,7 @@ alloy-consensus = { workspace = true, features = ["kzg"] }
 # async/futures
 futures-util.workspace = true
 parking_lot.workspace = true
+pin-project.workspace = true
 tokio = { workspace = true, features = ["sync"] }
 tokio-stream.workspace = true
 
@@ -72,6 +73,7 @@ assert_matches.workspace = true
 tempfile.workspace = true
 serde_json.workspace = true
 tokio = { workspace = true, features = ["rt-multi-thread"] }
+futures.workspace = true
 
 [features]
 serde = [
@@ -133,6 +135,11 @@ name = "priority"
 required-features = ["arbitrary"]
 harness = false
 
+[[bench]]
+name = "insertion"
+required-features = ["test-utils", "arbitrary"]
+harness = false
+
 [[bench]]
 name = "canonical_state_change"
 required-features = ["test-utils", "arbitrary"]