chore(tx-pool): pool cli args (#10323)

Co-authored-by: Oliver <onbjerg@users.noreply.github.com>

book/cli/reth/node.md

@@ -442,6 +442,16 @@ TxPool:
 
           [default: 1]
 
+      --txpool.max-pending-txns <PENDING_TX_LISTENER_BUFFER_SIZE>
+          Maximum number of pending transactions from the network to buffer
+
+          [default: 2048]
+
+      --txpool.max-new-txns <NEW_TX_LISTENER_BUFFER_SIZE>
+          Maximum number of new transactions to buffer
+
+          [default: 1024]
+
 Builder:
       --builder.extradata <EXTRADATA>
           Block extra data set by the payload builder

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

@@ -4,7 +4,9 @@ use crate::cli::config::RethTransactionPoolConfig;
 use clap::Args;
 use reth_primitives::Address;
 use reth_transaction_pool::{
-    blobstore::disk::DEFAULT_MAX_CACHED_BLOBS, validate::DEFAULT_MAX_TX_INPUT_BYTES,
+    blobstore::disk::DEFAULT_MAX_CACHED_BLOBS,
+    pool::{NEW_TX_LISTENER_BUFFER_SIZE, PENDING_TX_LISTENER_BUFFER_SIZE},
+    validate::DEFAULT_MAX_TX_INPUT_BYTES,
     LocalTransactionConfig, PoolConfig, PriceBumpConfig, SubPoolLimit, DEFAULT_PRICE_BUMP,
     DEFAULT_TXPOOL_ADDITIONAL_VALIDATION_TASKS, REPLACE_BLOB_PRICE_BUMP,
     TXPOOL_MAX_ACCOUNT_SLOTS_PER_SENDER, TXPOOL_SUBPOOL_MAX_SIZE_MB_DEFAULT,
@@ -67,6 +69,14 @@ pub struct TxPoolArgs {
     /// Number of additional transaction validation tasks to spawn.
     #[arg(long = "txpool.additional-validation-tasks", alias = "txpool.additional_validation_tasks", default_value_t = DEFAULT_TXPOOL_ADDITIONAL_VALIDATION_TASKS)]
     pub additional_validation_tasks: usize,
+
+    /// Maximum number of pending transactions from the network to buffer
+    #[arg(long = "txpool.max-pending-txns", alias = "txpool.max_pending_txns", default_value_t = PENDING_TX_LISTENER_BUFFER_SIZE)]
+    pub pending_tx_listener_buffer_size: usize,
+
+    /// Maximum number of new transactions to buffer
+    #[arg(long = "txpool.max-new-txns", alias = "txpool.max_new_txns", default_value_t = NEW_TX_LISTENER_BUFFER_SIZE)]
+    pub new_tx_listener_buffer_size: usize,
 }
 
 impl Default for TxPoolArgs {
@@ -87,6 +97,8 @@ impl Default for TxPoolArgs {
             locals: Default::default(),
             no_local_transactions_propagation: false,
             additional_validation_tasks: DEFAULT_TXPOOL_ADDITIONAL_VALIDATION_TASKS,
+            pending_tx_listener_buffer_size: PENDING_TX_LISTENER_BUFFER_SIZE,
+            new_tx_listener_buffer_size: NEW_TX_LISTENER_BUFFER_SIZE,
         }
     }
 }
@@ -121,6 +133,8 @@ impl RethTransactionPoolConfig for TxPoolArgs {
                 default_price_bump: self.price_bump,
                 replace_blob_tx_price_bump: self.blob_transaction_price_bump,
             },
+            pending_tx_listener_buffer_size: self.pending_tx_listener_buffer_size,
+            new_tx_listener_buffer_size: self.new_tx_listener_buffer_size,
         }
     }
 }

crates/transaction-pool/src/config.rs

@@ -1,4 +1,7 @@
-use crate::{PoolSize, TransactionOrigin};
+use crate::{
+    pool::{NEW_TX_LISTENER_BUFFER_SIZE, PENDING_TX_LISTENER_BUFFER_SIZE},
+    PoolSize, TransactionOrigin,
+};
 use reth_primitives::{Address, EIP4844_TX_TYPE_ID};
 use std::collections::HashSet;
 /// Guarantees max transactions for one sender, compatible with geth/erigon
@@ -39,6 +42,10 @@ pub struct PoolConfig {
     /// How to handle locally received transactions:
     /// [`TransactionOrigin::Local`](crate::TransactionOrigin).
     pub local_transactions_config: LocalTransactionConfig,
+    /// Bound on number of pending transactions from `reth_network::TransactionsManager` to buffer.
+    pub pending_tx_listener_buffer_size: usize,
+    /// Bound on number of new transactions from `reth_network::TransactionsManager` to buffer.
+    pub new_tx_listener_buffer_size: usize,
 }
 
 impl PoolConfig {
@@ -62,6 +69,8 @@ impl Default for PoolConfig {
             max_account_slots: TXPOOL_MAX_ACCOUNT_SLOTS_PER_SENDER,
             price_bumps: Default::default(),
             local_transactions_config: Default::default(),
+            pending_tx_listener_buffer_size: PENDING_TX_LISTENER_BUFFER_SIZE,
+            new_tx_listener_buffer_size: NEW_TX_LISTENER_BUFFER_SIZE,
         }
     }
 }

crates/transaction-pool/src/pool/mod.rs

@@ -120,8 +120,11 @@ pub(crate) mod state;
 pub mod txpool;
 mod update;
 
-const PENDING_TX_LISTENER_BUFFER_SIZE: usize = 2048;
-const NEW_TX_LISTENER_BUFFER_SIZE: usize = 1024;
+/// Bound on number of pending transactions from `reth_network::TransactionsManager` to buffer.
+pub const PENDING_TX_LISTENER_BUFFER_SIZE: usize = 2048;
+/// Bound on number of new transactions from `reth_network::TransactionsManager` to buffer.
+pub const NEW_TX_LISTENER_BUFFER_SIZE: usize = 1024;
+
 const BLOB_SIDECAR_LISTENER_BUFFER_SIZE: usize = 512;
 
 /// Transaction pool internals.
@@ -233,7 +236,7 @@ where
     /// Adds a new transaction listener to the pool that gets notified about every new _pending_
     /// transaction inserted into the pool
     pub fn add_pending_listener(&self, kind: TransactionListenerKind) -> mpsc::Receiver<TxHash> {
-        let (sender, rx) = mpsc::channel(PENDING_TX_LISTENER_BUFFER_SIZE);
+        let (sender, rx) = mpsc::channel(self.config.pending_tx_listener_buffer_size);
         let listener = PendingTransactionHashListener { sender, kind };
         self.pending_transaction_listener.lock().push(listener);
         rx
@@ -244,7 +247,7 @@ where
         &self,
         kind: TransactionListenerKind,
     ) -> mpsc::Receiver<NewTransactionEvent<T::Transaction>> {
-        let (sender, rx) = mpsc::channel(NEW_TX_LISTENER_BUFFER_SIZE);
+        let (sender, rx) = mpsc::channel(self.config.new_tx_listener_buffer_size);
         let listener = TransactionListener { sender, kind };
         self.transaction_listener.lock().push(listener);
         rx