feat(tx-pool): make metrics, listener structs, and fields public (#20087)

Author: datnguyencse

crates/transaction-pool/src/metrics.rs

@@ -10,72 +10,72 @@ use reth_metrics::{
 #[metrics(scope = "transaction_pool")]
 pub struct TxPoolMetrics {
     /// Number of transactions inserted in the pool
-    pub(crate) inserted_transactions: Counter,
+    pub inserted_transactions: Counter,
     /// Number of invalid transactions
-    pub(crate) invalid_transactions: Counter,
+    pub invalid_transactions: Counter,
     /// Number of removed transactions from the pool
-    pub(crate) removed_transactions: Counter,
+    pub removed_transactions: Counter,
 
     /// Number of transactions in the pending sub-pool
-    pub(crate) pending_pool_transactions: Gauge,
+    pub pending_pool_transactions: Gauge,
     /// Total amount of memory used by the transactions in the pending sub-pool in bytes
-    pub(crate) pending_pool_size_bytes: Gauge,
+    pub pending_pool_size_bytes: Gauge,
 
     /// Number of transactions in the basefee sub-pool
-    pub(crate) basefee_pool_transactions: Gauge,
+    pub basefee_pool_transactions: Gauge,
     /// Total amount of memory used by the transactions in the basefee sub-pool in bytes
-    pub(crate) basefee_pool_size_bytes: Gauge,
+    pub basefee_pool_size_bytes: Gauge,
 
     /// Number of transactions in the queued sub-pool
-    pub(crate) queued_pool_transactions: Gauge,
+    pub queued_pool_transactions: Gauge,
     /// Total amount of memory used by the transactions in the queued sub-pool in bytes
-    pub(crate) queued_pool_size_bytes: Gauge,
+    pub queued_pool_size_bytes: Gauge,
 
     /// Number of transactions in the blob sub-pool
-    pub(crate) blob_pool_transactions: Gauge,
+    pub blob_pool_transactions: Gauge,
     /// Total amount of memory used by the transactions in the blob sub-pool in bytes
-    pub(crate) blob_pool_size_bytes: Gauge,
+    pub blob_pool_size_bytes: Gauge,
 
     /// Number of all transactions of all sub-pools: pending + basefee + queued + blob
-    pub(crate) total_transactions: Gauge,
+    pub total_transactions: Gauge,
     /// Number of all legacy transactions in the pool
-    pub(crate) total_legacy_transactions: Gauge,
+    pub total_legacy_transactions: Gauge,
     /// Number of all EIP-2930 transactions in the pool
-    pub(crate) total_eip2930_transactions: Gauge,
+    pub total_eip2930_transactions: Gauge,
     /// Number of all EIP-1559 transactions in the pool
-    pub(crate) total_eip1559_transactions: Gauge,
+    pub total_eip1559_transactions: Gauge,
     /// Number of all EIP-4844 transactions in the pool
-    pub(crate) total_eip4844_transactions: Gauge,
+    pub total_eip4844_transactions: Gauge,
     /// Number of all EIP-7702 transactions in the pool
-    pub(crate) total_eip7702_transactions: Gauge,
+    pub total_eip7702_transactions: Gauge,
     /// Number of all other transactions in the pool
-    pub(crate) total_other_transactions: Gauge,
+    pub total_other_transactions: Gauge,
 
     /// How often the pool was updated after the canonical state changed
-    pub(crate) performed_state_updates: Counter,
+    pub performed_state_updates: Counter,
 
     /// Counter for the number of pending transactions evicted
-    pub(crate) pending_transactions_evicted: Counter,
+    pub pending_transactions_evicted: Counter,
     /// Counter for the number of basefee transactions evicted
-    pub(crate) basefee_transactions_evicted: Counter,
+    pub basefee_transactions_evicted: Counter,
     /// Counter for the number of blob transactions evicted
-    pub(crate) blob_transactions_evicted: Counter,
+    pub blob_transactions_evicted: Counter,
     /// Counter for the number of queued transactions evicted
-    pub(crate) queued_transactions_evicted: Counter,
+    pub queued_transactions_evicted: Counter,
 }
 
 /// Transaction pool blobstore metrics
 #[derive(Metrics)]
 #[metrics(scope = "transaction_pool")]
 pub struct BlobStoreMetrics {
     /// Number of failed inserts into the blobstore
-    pub(crate) blobstore_failed_inserts: Counter,
+    pub blobstore_failed_inserts: Counter,
     /// Number of failed deletes into the blobstore
-    pub(crate) blobstore_failed_deletes: Counter,
+    pub blobstore_failed_deletes: Counter,
     /// The number of bytes the blobs in the blobstore take up
-    pub(crate) blobstore_byte_size: Gauge,
+    pub blobstore_byte_size: Gauge,
     /// How many blobs are currently in the blobstore
-    pub(crate) blobstore_entries: Gauge,
+    pub blobstore_entries: Gauge,
 }
 
 /// Transaction pool maintenance metrics
@@ -84,35 +84,39 @@ pub struct BlobStoreMetrics {
 pub struct MaintainPoolMetrics {
     /// Gauge indicating the number of addresses with pending updates in the pool,
     /// requiring their account information to be fetched.
-    pub(crate) dirty_accounts: Gauge,
+    pub dirty_accounts: Gauge,
     /// Counter for the number of times the pool state diverged from the canonical blockchain
     /// state.
-    pub(crate) drift_count: Counter,
+    pub drift_count: Counter,
     /// Counter for the number of transactions reinserted into the pool following a blockchain
     /// reorganization (reorg).
-    pub(crate) reinserted_transactions: Counter,
+    pub reinserted_transactions: Counter,
     /// Counter for the number of finalized blob transactions that have been removed from tracking.
-    pub(crate) deleted_tracked_finalized_blobs: Counter,
+    pub deleted_tracked_finalized_blobs: Counter,
 }
 
 impl MaintainPoolMetrics {
+    /// Sets the number of dirty accounts in the pool.
     #[inline]
-    pub(crate) fn set_dirty_accounts_len(&self, count: usize) {
+    pub fn set_dirty_accounts_len(&self, count: usize) {
         self.dirty_accounts.set(count as f64);
     }
 
+    /// Increments the count of reinserted transactions.
     #[inline]
-    pub(crate) fn inc_reinserted_transactions(&self, count: usize) {
+    pub fn inc_reinserted_transactions(&self, count: usize) {
         self.reinserted_transactions.increment(count as u64);
     }
 
+    /// Increments the count of deleted tracked finalized blobs.
     #[inline]
-    pub(crate) fn inc_deleted_tracked_blobs(&self, count: usize) {
+    pub fn inc_deleted_tracked_blobs(&self, count: usize) {
         self.deleted_tracked_finalized_blobs.increment(count as u64);
     }
 
+    /// Increments the drift count by one.
     #[inline]
-    pub(crate) fn inc_drift(&self) {
+    pub fn inc_drift(&self) {
         self.drift_count.increment(1);
     }
 }
@@ -122,31 +126,31 @@ impl MaintainPoolMetrics {
 #[metrics(scope = "transaction_pool")]
 pub struct AllTransactionsMetrics {
     /// Number of all transactions by hash in the pool
-    pub(crate) all_transactions_by_hash: Gauge,
+    pub all_transactions_by_hash: Gauge,
     /// Number of all transactions by id in the pool
-    pub(crate) all_transactions_by_id: Gauge,
+    pub all_transactions_by_id: Gauge,
     /// Number of all transactions by all senders in the pool
-    pub(crate) all_transactions_by_all_senders: Gauge,
+    pub all_transactions_by_all_senders: Gauge,
     /// Number of blob transactions nonce gaps.
-    pub(crate) blob_transactions_nonce_gaps: Counter,
+    pub blob_transactions_nonce_gaps: Counter,
     /// The current blob base fee
-    pub(crate) blob_base_fee: Gauge,
+    pub blob_base_fee: Gauge,
     /// The current base fee
-    pub(crate) base_fee: Gauge,
+    pub base_fee: Gauge,
 }
 
 /// Transaction pool validation metrics
 #[derive(Metrics)]
 #[metrics(scope = "transaction_pool")]
 pub struct TxPoolValidationMetrics {
     /// How long to successfully validate a blob
-    pub(crate) blob_validation_duration: Histogram,
+    pub blob_validation_duration: Histogram,
 }
 
 /// Transaction pool validator task metrics
 #[derive(Metrics)]
 #[metrics(scope = "transaction_pool")]
 pub struct TxPoolValidatorMetrics {
     /// Number of in-flight validation job sends waiting for channel capacity
-    pub(crate) inflight_validation_jobs: Gauge,
+    pub inflight_validation_jobs: Gauge,
 }

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

@@ -82,7 +82,7 @@ impl<T: PoolTransaction> Stream for AllTransactionsEvents<T> {
 /// This is essentially a multi-producer, multi-consumer channel where each event is broadcast to
 /// all active receivers.
 #[derive(Debug)]
-pub(crate) struct PoolEventBroadcast<T: PoolTransaction> {
+pub struct PoolEventBroadcast<T: PoolTransaction> {
     /// All listeners for all transaction events.
     all_events_broadcaster: AllPoolEventsBroadcaster<T>,
     /// All listeners for events for a certain transaction hash.
@@ -121,12 +121,12 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
 
     /// Returns true if no listeners are installed
     #[inline]
-    pub(crate) fn is_empty(&self) -> bool {
+    pub fn is_empty(&self) -> bool {
         self.all_events_broadcaster.is_empty() && self.broadcasters_by_hash.is_empty()
     }
 
     /// Create a new subscription for the given transaction hash.
-    pub(crate) fn subscribe(&mut self, tx_hash: TxHash) -> TransactionEvents {
+    pub fn subscribe(&mut self, tx_hash: TxHash) -> TransactionEvents {
         let (tx, rx) = tokio::sync::mpsc::unbounded_channel();
 
         match self.broadcasters_by_hash.entry(tx_hash) {
@@ -141,14 +141,14 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
     }
 
     /// Create a new subscription for all transactions.
-    pub(crate) fn subscribe_all(&mut self) -> AllTransactionsEvents<T> {
+    pub fn subscribe_all(&mut self) -> AllTransactionsEvents<T> {
         let (tx, rx) = tokio::sync::mpsc::channel(TX_POOL_EVENT_CHANNEL_SIZE);
         self.all_events_broadcaster.senders.push(tx);
         AllTransactionsEvents::new(rx)
     }
 
     /// Notify listeners about a transaction that was added to the pending queue.
-    pub(crate) fn pending(&mut self, tx: &TxHash, replaced: Option<Arc<ValidPoolTransaction<T>>>) {
+    pub fn pending(&mut self, tx: &TxHash, replaced: Option<Arc<ValidPoolTransaction<T>>>) {
         self.broadcast_event(tx, TransactionEvent::Pending, FullTransactionEvent::Pending(*tx));
 
         if let Some(replaced) = replaced {
@@ -158,7 +158,7 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
     }
 
     /// Notify listeners about a transaction that was replaced.
-    pub(crate) fn replaced(&mut self, tx: Arc<ValidPoolTransaction<T>>, replaced_by: TxHash) {
+    pub fn replaced(&mut self, tx: Arc<ValidPoolTransaction<T>>, replaced_by: TxHash) {
         let transaction = Arc::clone(&tx);
         self.broadcast_event(
             tx.hash(),
@@ -168,7 +168,7 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
     }
 
     /// Notify listeners about a transaction that was added to the queued pool.
-    pub(crate) fn queued(&mut self, tx: &TxHash, reason: Option<QueuedReason>) {
+    pub fn queued(&mut self, tx: &TxHash, reason: Option<QueuedReason>) {
         self.broadcast_event(
             tx,
             TransactionEvent::Queued,
@@ -177,7 +177,7 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
     }
 
     /// Notify listeners about a transaction that was propagated.
-    pub(crate) fn propagated(&mut self, tx: &TxHash, peers: Vec<PropagateKind>) {
+    pub fn propagated(&mut self, tx: &TxHash, peers: Vec<PropagateKind>) {
         let peers = Arc::new(peers);
         self.broadcast_event(
             tx,
@@ -188,7 +188,7 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
 
     /// Notify listeners about all discarded transactions.
     #[inline]
-    pub(crate) fn discarded_many(&mut self, discarded: &[Arc<ValidPoolTransaction<T>>]) {
+    pub fn discarded_many(&mut self, discarded: &[Arc<ValidPoolTransaction<T>>]) {
         if self.is_empty() {
             return
         }
@@ -198,17 +198,17 @@ impl<T: PoolTransaction> PoolEventBroadcast<T> {
     }
 
     /// Notify listeners about a transaction that was discarded.
-    pub(crate) fn discarded(&mut self, tx: &TxHash) {
+    pub fn discarded(&mut self, tx: &TxHash) {
         self.broadcast_event(tx, TransactionEvent::Discarded, FullTransactionEvent::Discarded(*tx));
     }
 
     /// Notify listeners about a transaction that was invalid.
-    pub(crate) fn invalid(&mut self, tx: &TxHash) {
+    pub fn invalid(&mut self, tx: &TxHash) {
         self.broadcast_event(tx, TransactionEvent::Invalid, FullTransactionEvent::Invalid(*tx));
     }
 
     /// Notify listeners that the transaction was mined
-    pub(crate) fn mined(&mut self, tx: &TxHash, block_hash: B256) {
+    pub fn mined(&mut self, tx: &TxHash, block_hash: B256) {
         self.broadcast_event(
             tx,
             TransactionEvent::Mined(block_hash),
@@ -271,17 +271,18 @@ impl PoolEventBroadcaster {
 
 /// An active listener for new pending transactions.
 #[derive(Debug)]
-pub(crate) struct PendingTransactionHashListener {
-    pub(crate) sender: mpsc::Sender<TxHash>,
+pub struct PendingTransactionHashListener {
+    /// The sender of the channel to send transaction hashes to.
+    pub sender: mpsc::Sender<TxHash>,
     /// Whether to include transactions that should not be propagated over the network.
-    pub(crate) kind: TransactionListenerKind,
+    pub kind: TransactionListenerKind,
 }
 
 impl PendingTransactionHashListener {
     /// Attempts to send all hashes to the listener.
     ///
     /// Returns false if the channel is closed (receiver dropped)
-    pub(crate) fn send_all(&self, hashes: impl IntoIterator<Item = TxHash>) -> bool {
+    pub fn send_all(&self, hashes: impl IntoIterator<Item = TxHash>) -> bool {
         for tx_hash in hashes {
             match self.sender.try_send(tx_hash) {
                 Ok(()) => {}
@@ -305,27 +306,25 @@ impl PendingTransactionHashListener {
 
 /// An active listener for new pending transactions.
 #[derive(Debug)]
-pub(crate) struct TransactionListener<T: PoolTransaction> {
-    pub(crate) sender: mpsc::Sender<NewTransactionEvent<T>>,
+pub struct TransactionListener<T: PoolTransaction> {
+    /// The sender of the channel to send new transaction events to.
+    pub sender: mpsc::Sender<NewTransactionEvent<T>>,
     /// Whether to include transactions that should not be propagated over the network.
-    pub(crate) kind: TransactionListenerKind,
+    pub kind: TransactionListenerKind,
 }
 
 impl<T: PoolTransaction> TransactionListener<T> {
     /// Attempts to send the event to the listener.
     ///
     /// Returns false if the channel is closed (receiver dropped)
-    pub(crate) fn send(&self, event: NewTransactionEvent<T>) -> bool {
+    pub fn send(&self, event: NewTransactionEvent<T>) -> bool {
         self.send_all(std::iter::once(event))
     }
 
     /// Attempts to send all events to the listener.
     ///
     /// Returns false if the channel is closed (receiver dropped)
-    pub(crate) fn send_all(
-        &self,
-        events: impl IntoIterator<Item = NewTransactionEvent<T>>,
-    ) -> bool {
+    pub fn send_all(&self, events: impl IntoIterator<Item = NewTransactionEvent<T>>) -> bool {
         for event in events {
             match self.sender.try_send(event) {
                 Ok(()) => {}

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

@@ -112,10 +112,10 @@ mod best;
 pub use best::BestTransactions;
 
 mod blob;
-mod listener;
+pub mod listener;
 mod parked;
 pub mod pending;
-pub(crate) mod size;
+pub mod size;
 pub(crate) mod state;
 pub mod txpool;
 mod update;