reth_transaction_pool/pool/
mod.rs

1//! Transaction Pool internals.
2//!
3//! Incoming transactions are validated before they enter the pool first. The validation outcome can
4//! have 3 states:
5//!
6//!  1. Transaction can _never_ be valid
7//!  2. Transaction is _currently_ valid
8//!  3. Transaction is _currently_ invalid, but could potentially become valid in the future
9//!
10//! However, (2.) and (3.) of a transaction can only be determined on the basis of the current
11//! state, whereas (1.) holds indefinitely. This means once the state changes (2.) and (3.) the
12//! state of a transaction needs to be reevaluated again.
13//!
14//! The transaction pool is responsible for storing new, valid transactions and providing the next
15//! best transactions sorted by their priority. Where priority is determined by the transaction's
16//! score ([`TransactionOrdering`]).
17//!
18//! Furthermore, the following characteristics fall under (3.):
19//!
20//!  a) Nonce of a transaction is higher than the expected nonce for the next transaction of its
21//! sender. A distinction is made here whether multiple transactions from the same sender have
22//! gapless nonce increments.
23//!
24//!  a)(1) If _no_ transaction is missing in a chain of multiple
25//! transactions from the same sender (all nonce in row), all of them can in principle be executed
26//! on the current state one after the other.
27//!
28//!  a)(2) If there's a nonce gap, then all
29//! transactions after the missing transaction are blocked until the missing transaction arrives.
30//!
31//!  b) Transaction does not meet the dynamic fee cap requirement introduced by EIP-1559: The
32//! fee cap of the transaction needs to be no less than the base fee of block.
33//!
34//!
35//! In essence the transaction pool is made of three separate sub-pools:
36//!
37//!  - Pending Pool: Contains all transactions that are valid on the current state and satisfy (3.
38//!    a)(1): _No_ nonce gaps. A _pending_ transaction is considered _ready_ when it has the lowest
39//!    nonce of all transactions from the same sender. Once a _ready_ transaction with nonce `n` has
40//!    been executed, the next highest transaction from the same sender `n + 1` becomes ready.
41//!
42//!  - Queued Pool: Contains all transactions that are currently blocked by missing transactions:
43//!    (3. a)(2): _With_ nonce gaps or due to lack of funds.
44//!
45//!  - Basefee Pool: To account for the dynamic base fee requirement (3. b) which could render an
46//!    EIP-1559 and all subsequent transactions of the sender currently invalid.
47//!
48//! The classification of transactions is always dependent on the current state that is changed as
49//! soon as a new block is mined. Once a new block is mined, the account changeset must be applied
50//! to the transaction pool.
51//!
52//!
53//! Depending on the use case, consumers of the [`TransactionPool`](crate::traits::TransactionPool)
54//! are interested in (2.) and/or (3.).
55
56//! A generic [`TransactionPool`](crate::traits::TransactionPool) that only handles transactions.
57//!
58//! This Pool maintains two separate sub-pools for (2.) and (3.)
59//!
60//! ## Terminology
61//!
62//!  - _Pending_: pending transactions are transactions that fall under (2.). These transactions can
63//!    currently be executed and are stored in the pending sub-pool
64//!  - _Queued_: queued transactions are transactions that fall under category (3.). Those
65//!    transactions are _currently_ waiting for state changes that eventually move them into
66//!    category (2.) and become pending.
67
68use crate::{
69    blobstore::BlobStore,
70    error::{PoolError, PoolErrorKind, PoolResult},
71    identifier::{SenderId, SenderIdentifiers, TransactionId},
72    metrics::BlobStoreMetrics,
73    pool::{
74        listener::{
75            BlobTransactionSidecarListener, PendingTransactionHashListener, PoolEventBroadcast,
76            TransactionListener,
77        },
78        state::SubPool,
79        txpool::{SenderInfo, TxPool},
80        update::UpdateOutcome,
81    },
82    traits::{
83        AllPoolTransactions, BestTransactionsAttributes, BlockInfo, GetPooledTransactionLimit,
84        NewBlobSidecar, PoolSize, PoolTransaction, PropagatedTransactions, TransactionOrigin,
85    },
86    validate::{TransactionValidationOutcome, ValidPoolTransaction, ValidTransaction},
87    CanonicalStateUpdate, EthPoolTransaction, PoolConfig, TransactionOrdering,
88    TransactionValidator,
89};
90
91use alloy_primitives::{Address, TxHash, B256};
92use best::BestTransactions;
93use parking_lot::{Mutex, RwLock, RwLockReadGuard, RwLockWriteGuard};
94use reth_eth_wire_types::HandleMempoolData;
95use reth_execution_types::ChangedAccount;
96
97use alloy_eips::{eip7594::BlobTransactionSidecarVariant, Typed2718};
98use reth_primitives_traits::Recovered;
99use rustc_hash::FxHashMap;
100use std::{collections::HashSet, fmt, sync::Arc, time::Instant};
101use tokio::sync::mpsc;
102use tracing::{debug, trace, warn};
103mod events;
104pub use best::{BestTransactionFilter, BestTransactionsWithPrioritizedSenders};
105pub use blob::{blob_tx_priority, fee_delta, BlobOrd, BlobTransactions};
106pub use events::{FullTransactionEvent, NewTransactionEvent, TransactionEvent};
107pub use listener::{AllTransactionsEvents, TransactionEvents, TransactionListenerKind};
108pub use parked::{BasefeeOrd, ParkedOrd, ParkedPool, QueuedOrd};
109pub use pending::PendingPool;
110use reth_primitives_traits::Block;
111
112mod best;
113mod blob;
114mod listener;
115mod parked;
116pub(crate) mod pending;
117pub(crate) mod size;
118pub(crate) mod state;
119pub mod txpool;
120mod update;
121
122/// Bound on number of pending transactions from `reth_network::TransactionsManager` to buffer.
123pub const PENDING_TX_LISTENER_BUFFER_SIZE: usize = 2048;
124/// Bound on number of new transactions from `reth_network::TransactionsManager` to buffer.
125pub const NEW_TX_LISTENER_BUFFER_SIZE: usize = 1024;
126
127const BLOB_SIDECAR_LISTENER_BUFFER_SIZE: usize = 512;
128
129/// Transaction pool internals.
130pub struct PoolInner<V, T, S>
131where
132    T: TransactionOrdering,
133{
134    /// Internal mapping of addresses to plain ints.
135    identifiers: RwLock<SenderIdentifiers>,
136    /// Transaction validator.
137    validator: V,
138    /// Storage for blob transactions
139    blob_store: S,
140    /// The internal pool that manages all transactions.
141    pool: RwLock<TxPool<T>>,
142    /// Pool settings.
143    config: PoolConfig,
144    /// Manages listeners for transaction state change events.
145    event_listener: RwLock<PoolEventBroadcast<T::Transaction>>,
146    /// Listeners for new _full_ pending transactions.
147    pending_transaction_listener: Mutex<Vec<PendingTransactionHashListener>>,
148    /// Listeners for new transactions added to the pool.
149    transaction_listener: Mutex<Vec<TransactionListener<T::Transaction>>>,
150    /// Listener for new blob transaction sidecars added to the pool.
151    blob_transaction_sidecar_listener: Mutex<Vec<BlobTransactionSidecarListener>>,
152    /// Metrics for the blob store
153    blob_store_metrics: BlobStoreMetrics,
154}
155
156// === impl PoolInner ===
157
158impl<V, T, S> PoolInner<V, T, S>
159where
160    V: TransactionValidator,
161    T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
162    S: BlobStore,
163{
164    /// Create a new transaction pool instance.
165    pub fn new(validator: V, ordering: T, blob_store: S, config: PoolConfig) -> Self {
166        Self {
167            identifiers: Default::default(),
168            validator,
169            event_listener: Default::default(),
170            pool: RwLock::new(TxPool::new(ordering, config.clone())),
171            pending_transaction_listener: Default::default(),
172            transaction_listener: Default::default(),
173            blob_transaction_sidecar_listener: Default::default(),
174            config,
175            blob_store,
176            blob_store_metrics: Default::default(),
177        }
178    }
179
180    /// Returns the configured blob store.
181    pub const fn blob_store(&self) -> &S {
182        &self.blob_store
183    }
184
185    /// Returns stats about the size of the pool.
186    pub fn size(&self) -> PoolSize {
187        self.get_pool_data().size()
188    }
189
190    /// Returns the currently tracked block
191    pub fn block_info(&self) -> BlockInfo {
192        self.get_pool_data().block_info()
193    }
194    /// Sets the currently tracked block
195    pub fn set_block_info(&self, info: BlockInfo) {
196        self.pool.write().set_block_info(info)
197    }
198
199    /// Returns the internal [`SenderId`] for this address
200    pub fn get_sender_id(&self, addr: Address) -> SenderId {
201        self.identifiers.write().sender_id_or_create(addr)
202    }
203
204    /// Returns the internal [`SenderId`]s for the given addresses.
205    pub fn get_sender_ids(&self, addrs: impl IntoIterator<Item = Address>) -> Vec<SenderId> {
206        self.identifiers.write().sender_ids_or_create(addrs)
207    }
208
209    /// Returns all senders in the pool
210    pub fn unique_senders(&self) -> HashSet<Address> {
211        self.get_pool_data().unique_senders()
212    }
213
214    /// Converts the changed accounts to a map of sender ids to sender info (internal identifier
215    /// used for accounts)
216    fn changed_senders(
217        &self,
218        accs: impl Iterator<Item = ChangedAccount>,
219    ) -> FxHashMap<SenderId, SenderInfo> {
220        let mut identifiers = self.identifiers.write();
221        accs.into_iter()
222            .map(|acc| {
223                let ChangedAccount { address, nonce, balance } = acc;
224                let sender_id = identifiers.sender_id_or_create(address);
225                (sender_id, SenderInfo { state_nonce: nonce, balance })
226            })
227            .collect()
228    }
229
230    /// Get the config the pool was configured with.
231    pub const fn config(&self) -> &PoolConfig {
232        &self.config
233    }
234
235    /// Get the validator reference.
236    pub const fn validator(&self) -> &V {
237        &self.validator
238    }
239
240    /// Adds a new transaction listener to the pool that gets notified about every new _pending_
241    /// transaction inserted into the pool
242    pub fn add_pending_listener(&self, kind: TransactionListenerKind) -> mpsc::Receiver<TxHash> {
243        let (sender, rx) = mpsc::channel(self.config.pending_tx_listener_buffer_size);
244        let listener = PendingTransactionHashListener { sender, kind };
245        self.pending_transaction_listener.lock().push(listener);
246        rx
247    }
248
249    /// Adds a new transaction listener to the pool that gets notified about every new transaction.
250    pub fn add_new_transaction_listener(
251        &self,
252        kind: TransactionListenerKind,
253    ) -> mpsc::Receiver<NewTransactionEvent<T::Transaction>> {
254        let (sender, rx) = mpsc::channel(self.config.new_tx_listener_buffer_size);
255        let listener = TransactionListener { sender, kind };
256        self.transaction_listener.lock().push(listener);
257        rx
258    }
259    /// Adds a new blob sidecar listener to the pool that gets notified about every new
260    /// eip4844 transaction's blob sidecar.
261    pub fn add_blob_sidecar_listener(&self) -> mpsc::Receiver<NewBlobSidecar> {
262        let (sender, rx) = mpsc::channel(BLOB_SIDECAR_LISTENER_BUFFER_SIZE);
263        let listener = BlobTransactionSidecarListener { sender };
264        self.blob_transaction_sidecar_listener.lock().push(listener);
265        rx
266    }
267
268    /// If the pool contains the transaction, this adds a new listener that gets notified about
269    /// transaction events.
270    pub fn add_transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents> {
271        self.get_pool_data()
272            .contains(&tx_hash)
273            .then(|| self.event_listener.write().subscribe(tx_hash))
274    }
275
276    /// Adds a listener for all transaction events.
277    pub fn add_all_transactions_event_listener(&self) -> AllTransactionsEvents<T::Transaction> {
278        self.event_listener.write().subscribe_all()
279    }
280
281    /// Returns a read lock to the pool's data.
282    pub fn get_pool_data(&self) -> RwLockReadGuard<'_, TxPool<T>> {
283        self.pool.read()
284    }
285
286    /// Returns hashes of _all_ transactions in the pool.
287    pub fn pooled_transactions_hashes(&self) -> Vec<TxHash> {
288        self.get_pool_data()
289            .all()
290            .transactions_iter()
291            .filter(|tx| tx.propagate)
292            .map(|tx| *tx.hash())
293            .collect()
294    }
295
296    /// Returns _all_ transactions in the pool.
297    pub fn pooled_transactions(&self) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
298        self.get_pool_data().all().transactions_iter().filter(|tx| tx.propagate).cloned().collect()
299    }
300
301    /// Returns only the first `max` transactions in the pool.
302    pub fn pooled_transactions_max(
303        &self,
304        max: usize,
305    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
306        self.get_pool_data()
307            .all()
308            .transactions_iter()
309            .filter(|tx| tx.propagate)
310            .take(max)
311            .cloned()
312            .collect()
313    }
314
315    /// Converts the internally tracked transaction to the pooled format.
316    ///
317    /// If the transaction is an EIP-4844 transaction, the blob sidecar is fetched from the blob
318    /// store and attached to the transaction.
319    fn to_pooled_transaction(
320        &self,
321        transaction: Arc<ValidPoolTransaction<T::Transaction>>,
322    ) -> Option<Recovered<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled>>
323    where
324        <V as TransactionValidator>::Transaction: EthPoolTransaction,
325    {
326        if transaction.is_eip4844() {
327            let sidecar = self.blob_store.get(*transaction.hash()).ok()??;
328            transaction.transaction.clone().try_into_pooled_eip4844(sidecar)
329        } else {
330            transaction
331                .transaction
332                .clone()
333                .try_into_pooled()
334                .inspect_err(|err| {
335                    debug!(
336                        target: "txpool", %err,
337                        "failed to convert transaction to pooled element; skipping",
338                    );
339                })
340                .ok()
341        }
342    }
343
344    /// Returns pooled transactions for the given transaction hashes.
345    pub fn get_pooled_transaction_elements(
346        &self,
347        tx_hashes: Vec<TxHash>,
348        limit: GetPooledTransactionLimit,
349    ) -> Vec<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled>
350    where
351        <V as TransactionValidator>::Transaction: EthPoolTransaction,
352    {
353        let transactions = self.get_all(tx_hashes);
354        let mut elements = Vec::with_capacity(transactions.len());
355        let mut size = 0;
356        for transaction in transactions {
357            let encoded_len = transaction.encoded_length();
358            let Some(pooled) = self.to_pooled_transaction(transaction) else {
359                continue;
360            };
361
362            size += encoded_len;
363            elements.push(pooled.into_inner());
364
365            if limit.exceeds(size) {
366                break
367            }
368        }
369
370        elements
371    }
372
373    /// Returns converted pooled transaction for the given transaction hash.
374    pub fn get_pooled_transaction_element(
375        &self,
376        tx_hash: TxHash,
377    ) -> Option<Recovered<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled>>
378    where
379        <V as TransactionValidator>::Transaction: EthPoolTransaction,
380    {
381        self.get(&tx_hash).and_then(|tx| self.to_pooled_transaction(tx))
382    }
383
384    /// Updates the entire pool after a new block was executed.
385    pub fn on_canonical_state_change<B>(&self, update: CanonicalStateUpdate<'_, B>)
386    where
387        B: Block,
388    {
389        trace!(target: "txpool", ?update, "updating pool on canonical state change");
390
391        let block_info = update.block_info();
392        let CanonicalStateUpdate {
393            new_tip, changed_accounts, mined_transactions, update_kind, ..
394        } = update;
395        self.validator.on_new_head_block(new_tip);
396
397        let changed_senders = self.changed_senders(changed_accounts.into_iter());
398
399        // update the pool
400        let outcome = self.pool.write().on_canonical_state_change(
401            block_info,
402            mined_transactions,
403            changed_senders,
404            update_kind,
405        );
406
407        // This will discard outdated transactions based on the account's nonce
408        self.delete_discarded_blobs(outcome.discarded.iter());
409
410        // notify listeners about updates
411        self.notify_on_new_state(outcome);
412    }
413
414    /// Performs account updates on the pool.
415    ///
416    /// This will either promote or discard transactions based on the new account state.
417    pub fn update_accounts(&self, accounts: Vec<ChangedAccount>) {
418        let changed_senders = self.changed_senders(accounts.into_iter());
419        let UpdateOutcome { promoted, discarded } =
420            self.pool.write().update_accounts(changed_senders);
421        let mut listener = self.event_listener.write();
422
423        promoted.iter().for_each(|tx| listener.pending(tx.hash(), None));
424        discarded.iter().for_each(|tx| listener.discarded(tx.hash()));
425
426        // This deletes outdated blob txs from the blob store, based on the account's nonce. This is
427        // called during txpool maintenance when the pool drifted.
428        self.delete_discarded_blobs(discarded.iter());
429    }
430
431    /// Add a single validated transaction into the pool.
432    ///
433    /// Note: this is only used internally by [`Self::add_transactions()`], all new transaction(s)
434    /// come in through that function, either as a batch or `std::iter::once`.
435    fn add_transaction(
436        &self,
437        pool: &mut RwLockWriteGuard<'_, TxPool<T>>,
438        origin: TransactionOrigin,
439        tx: TransactionValidationOutcome<T::Transaction>,
440    ) -> PoolResult<TxHash> {
441        match tx {
442            TransactionValidationOutcome::Valid {
443                balance,
444                state_nonce,
445                transaction,
446                propagate,
447                bytecode_hash,
448                authorities,
449            } => {
450                let sender_id = self.get_sender_id(transaction.sender());
451                let transaction_id = TransactionId::new(sender_id, transaction.nonce());
452
453                // split the valid transaction and the blob sidecar if it has any
454                let (transaction, maybe_sidecar) = match transaction {
455                    ValidTransaction::Valid(tx) => (tx, None),
456                    ValidTransaction::ValidWithSidecar { transaction, sidecar } => {
457                        debug_assert!(
458                            transaction.is_eip4844(),
459                            "validator returned sidecar for non EIP-4844 transaction"
460                        );
461                        (transaction, Some(sidecar))
462                    }
463                };
464
465                let tx = ValidPoolTransaction {
466                    transaction,
467                    transaction_id,
468                    propagate,
469                    timestamp: Instant::now(),
470                    origin,
471                    authority_ids: authorities.map(|auths| self.get_sender_ids(auths)),
472                };
473
474                let added = pool.add_transaction(tx, balance, state_nonce, bytecode_hash)?;
475                let hash = *added.hash();
476
477                // transaction was successfully inserted into the pool
478                if let Some(sidecar) = maybe_sidecar {
479                    // notify blob sidecar listeners
480                    self.on_new_blob_sidecar(&hash, &sidecar);
481                    // store the sidecar in the blob store
482                    self.insert_blob(hash, sidecar);
483                }
484
485                if let Some(replaced) = added.replaced_blob_transaction() {
486                    debug!(target: "txpool", "[{:?}] delete replaced blob sidecar", replaced);
487                    // delete the replaced transaction from the blob store
488                    self.delete_blob(replaced);
489                }
490
491                // Notify about new pending transactions
492                if let Some(pending) = added.as_pending() {
493                    self.on_new_pending_transaction(pending);
494                }
495
496                // Notify tx event listeners
497                self.notify_event_listeners(&added);
498
499                if let Some(discarded) = added.discarded_transactions() {
500                    self.delete_discarded_blobs(discarded.iter());
501                }
502
503                // Notify listeners for _all_ transactions
504                self.on_new_transaction(added.into_new_transaction_event());
505
506                Ok(hash)
507            }
508            TransactionValidationOutcome::Invalid(tx, err) => {
509                let mut listener = self.event_listener.write();
510                listener.invalid(tx.hash());
511                Err(PoolError::new(*tx.hash(), err))
512            }
513            TransactionValidationOutcome::Error(tx_hash, err) => {
514                let mut listener = self.event_listener.write();
515                listener.discarded(&tx_hash);
516                Err(PoolError::other(tx_hash, err))
517            }
518        }
519    }
520
521    /// Adds a transaction and returns the event stream.
522    pub fn add_transaction_and_subscribe(
523        &self,
524        origin: TransactionOrigin,
525        tx: TransactionValidationOutcome<T::Transaction>,
526    ) -> PoolResult<TransactionEvents> {
527        let listener = {
528            let mut listener = self.event_listener.write();
529            listener.subscribe(tx.tx_hash())
530        };
531        let mut results = self.add_transactions(origin, std::iter::once(tx));
532        results.pop().expect("result length is the same as the input")?;
533        Ok(listener)
534    }
535
536    /// Adds all transactions in the iterator to the pool, returning a list of results.
537    ///
538    /// Note: A large batch may lock the pool for a long time that blocks important operations
539    /// like updating the pool on canonical state changes. The caller should consider having
540    /// a max batch size to balance transaction insertions with other updates.
541    pub fn add_transactions(
542        &self,
543        origin: TransactionOrigin,
544        transactions: impl IntoIterator<Item = TransactionValidationOutcome<T::Transaction>>,
545    ) -> Vec<PoolResult<TxHash>> {
546        // Add the transactions and enforce the pool size limits in one write lock
547        let (mut added, discarded) = {
548            let mut pool = self.pool.write();
549            let added = transactions
550                .into_iter()
551                .map(|tx| self.add_transaction(&mut pool, origin, tx))
552                .collect::<Vec<_>>();
553
554            // Enforce the pool size limits if at least one transaction was added successfully
555            let discarded = if added.iter().any(Result::is_ok) {
556                pool.discard_worst()
557            } else {
558                Default::default()
559            };
560
561            (added, discarded)
562        };
563
564        if !discarded.is_empty() {
565            // Delete any blobs associated with discarded blob transactions
566            self.delete_discarded_blobs(discarded.iter());
567
568            let discarded_hashes =
569                discarded.into_iter().map(|tx| *tx.hash()).collect::<HashSet<_>>();
570
571            {
572                let mut listener = self.event_listener.write();
573                discarded_hashes.iter().for_each(|hash| listener.discarded(hash));
574            }
575
576            // A newly added transaction may be immediately discarded, so we need to
577            // adjust the result here
578            for res in &mut added {
579                if let Ok(hash) = res {
580                    if discarded_hashes.contains(hash) {
581                        *res = Err(PoolError::new(*hash, PoolErrorKind::DiscardedOnInsert))
582                    }
583                }
584            }
585        }
586
587        added
588    }
589
590    /// Notify all listeners about a new pending transaction.
591    fn on_new_pending_transaction(&self, pending: &AddedPendingTransaction<T::Transaction>) {
592        let propagate_allowed = pending.is_propagate_allowed();
593
594        let mut transaction_listeners = self.pending_transaction_listener.lock();
595        transaction_listeners.retain_mut(|listener| {
596            if listener.kind.is_propagate_only() && !propagate_allowed {
597                // only emit this hash to listeners that are only allowed to receive propagate only
598                // transactions, such as network
599                return !listener.sender.is_closed()
600            }
601
602            // broadcast all pending transactions to the listener
603            listener.send_all(pending.pending_transactions(listener.kind))
604        });
605    }
606
607    /// Notify all listeners about a newly inserted pending transaction.
608    fn on_new_transaction(&self, event: NewTransactionEvent<T::Transaction>) {
609        let mut transaction_listeners = self.transaction_listener.lock();
610        transaction_listeners.retain_mut(|listener| {
611            if listener.kind.is_propagate_only() && !event.transaction.propagate {
612                // only emit this hash to listeners that are only allowed to receive propagate only
613                // transactions, such as network
614                return !listener.sender.is_closed()
615            }
616
617            listener.send(event.clone())
618        });
619    }
620
621    /// Notify all listeners about a blob sidecar for a newly inserted blob (eip4844) transaction.
622    fn on_new_blob_sidecar(&self, tx_hash: &TxHash, sidecar: &BlobTransactionSidecarVariant) {
623        let mut sidecar_listeners = self.blob_transaction_sidecar_listener.lock();
624        if sidecar_listeners.is_empty() {
625            return
626        }
627        let sidecar = Arc::new(sidecar.clone());
628        sidecar_listeners.retain_mut(|listener| {
629            let new_blob_event = NewBlobSidecar { tx_hash: *tx_hash, sidecar: sidecar.clone() };
630            match listener.sender.try_send(new_blob_event) {
631                Ok(()) => true,
632                Err(err) => {
633                    if matches!(err, mpsc::error::TrySendError::Full(_)) {
634                        debug!(
635                            target: "txpool",
636                            "[{:?}] failed to send blob sidecar; channel full",
637                            sidecar,
638                        );
639                        true
640                    } else {
641                        false
642                    }
643                }
644            }
645        })
646    }
647
648    /// Notifies transaction listeners about changes once a block was processed.
649    fn notify_on_new_state(&self, outcome: OnNewCanonicalStateOutcome<T::Transaction>) {
650        trace!(target: "txpool", promoted=outcome.promoted.len(), discarded= outcome.discarded.len() ,"notifying listeners on state change");
651
652        // notify about promoted pending transactions
653        // emit hashes
654        self.pending_transaction_listener
655            .lock()
656            .retain_mut(|listener| listener.send_all(outcome.pending_transactions(listener.kind)));
657
658        // emit full transactions
659        self.transaction_listener.lock().retain_mut(|listener| {
660            listener.send_all(outcome.full_pending_transactions(listener.kind))
661        });
662
663        let OnNewCanonicalStateOutcome { mined, promoted, discarded, block_hash } = outcome;
664
665        // broadcast specific transaction events
666        let mut listener = self.event_listener.write();
667
668        mined.iter().for_each(|tx| listener.mined(tx, block_hash));
669        promoted.iter().for_each(|tx| listener.pending(tx.hash(), None));
670        discarded.iter().for_each(|tx| listener.discarded(tx.hash()));
671    }
672
673    /// Fire events for the newly added transaction if there are any.
674    fn notify_event_listeners(&self, tx: &AddedTransaction<T::Transaction>) {
675        let mut listener = self.event_listener.write();
676
677        match tx {
678            AddedTransaction::Pending(tx) => {
679                let AddedPendingTransaction { transaction, promoted, discarded, replaced } = tx;
680
681                listener.pending(transaction.hash(), replaced.clone());
682                promoted.iter().for_each(|tx| listener.pending(tx.hash(), None));
683                discarded.iter().for_each(|tx| listener.discarded(tx.hash()));
684            }
685            AddedTransaction::Parked { transaction, replaced, .. } => {
686                listener.queued(transaction.hash());
687                if let Some(replaced) = replaced {
688                    listener.replaced(replaced.clone(), *transaction.hash());
689                }
690            }
691        }
692    }
693
694    /// Returns an iterator that yields transactions that are ready to be included in the block.
695    pub fn best_transactions(&self) -> BestTransactions<T> {
696        self.get_pool_data().best_transactions()
697    }
698
699    /// Returns an iterator that yields transactions that are ready to be included in the block with
700    /// the given base fee and optional blob fee attributes.
701    pub fn best_transactions_with_attributes(
702        &self,
703        best_transactions_attributes: BestTransactionsAttributes,
704    ) -> Box<dyn crate::traits::BestTransactions<Item = Arc<ValidPoolTransaction<T::Transaction>>>>
705    {
706        self.get_pool_data().best_transactions_with_attributes(best_transactions_attributes)
707    }
708
709    /// Returns only the first `max` transactions in the pending pool.
710    pub fn pending_transactions_max(
711        &self,
712        max: usize,
713    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
714        self.get_pool_data().pending_transactions_iter().take(max).collect()
715    }
716
717    /// Returns all transactions from the pending sub-pool
718    pub fn pending_transactions(&self) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
719        self.get_pool_data().pending_transactions()
720    }
721
722    /// Returns all transactions from parked pools
723    pub fn queued_transactions(&self) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
724        self.get_pool_data().queued_transactions()
725    }
726
727    /// Returns all transactions in the pool
728    pub fn all_transactions(&self) -> AllPoolTransactions<T::Transaction> {
729        let pool = self.get_pool_data();
730        AllPoolTransactions {
731            pending: pool.pending_transactions(),
732            queued: pool.queued_transactions(),
733        }
734    }
735
736    /// Removes and returns all matching transactions from the pool.
737    ///
738    /// This behaves as if the transactions got discarded (_not_ mined), effectively introducing a
739    /// nonce gap for the given transactions.
740    pub fn remove_transactions(
741        &self,
742        hashes: Vec<TxHash>,
743    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
744        if hashes.is_empty() {
745            return Vec::new()
746        }
747        let removed = self.pool.write().remove_transactions(hashes);
748
749        let mut listener = self.event_listener.write();
750
751        removed.iter().for_each(|tx| listener.discarded(tx.hash()));
752
753        removed
754    }
755
756    /// Removes and returns all matching transactions and their dependent transactions from the
757    /// pool.
758    pub fn remove_transactions_and_descendants(
759        &self,
760        hashes: Vec<TxHash>,
761    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
762        if hashes.is_empty() {
763            return Vec::new()
764        }
765        let removed = self.pool.write().remove_transactions_and_descendants(hashes);
766
767        let mut listener = self.event_listener.write();
768
769        removed.iter().for_each(|tx| listener.discarded(tx.hash()));
770
771        removed
772    }
773
774    /// Removes and returns all transactions by the specified sender from the pool.
775    pub fn remove_transactions_by_sender(
776        &self,
777        sender: Address,
778    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
779        let sender_id = self.get_sender_id(sender);
780        let removed = self.pool.write().remove_transactions_by_sender(sender_id);
781
782        let mut listener = self.event_listener.write();
783
784        removed.iter().for_each(|tx| listener.discarded(tx.hash()));
785
786        removed
787    }
788
789    /// Removes and returns all transactions that are present in the pool.
790    pub fn retain_unknown<A>(&self, announcement: &mut A)
791    where
792        A: HandleMempoolData,
793    {
794        if announcement.is_empty() {
795            return
796        }
797        let pool = self.get_pool_data();
798        announcement.retain_by_hash(|tx| !pool.contains(tx))
799    }
800
801    /// Returns the transaction by hash.
802    pub fn get(&self, tx_hash: &TxHash) -> Option<Arc<ValidPoolTransaction<T::Transaction>>> {
803        self.get_pool_data().get(tx_hash)
804    }
805
806    /// Returns all transactions of the address
807    pub fn get_transactions_by_sender(
808        &self,
809        sender: Address,
810    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
811        let sender_id = self.get_sender_id(sender);
812        self.get_pool_data().get_transactions_by_sender(sender_id)
813    }
814
815    /// Returns all queued transactions of the address by sender
816    pub fn get_queued_transactions_by_sender(
817        &self,
818        sender: Address,
819    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
820        let sender_id = self.get_sender_id(sender);
821        self.get_pool_data().queued_txs_by_sender(sender_id)
822    }
823
824    /// Returns all pending transactions filtered by predicate
825    pub fn pending_transactions_with_predicate(
826        &self,
827        predicate: impl FnMut(&ValidPoolTransaction<T::Transaction>) -> bool,
828    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
829        self.get_pool_data().pending_transactions_with_predicate(predicate)
830    }
831
832    /// Returns all pending transactions of the address by sender
833    pub fn get_pending_transactions_by_sender(
834        &self,
835        sender: Address,
836    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
837        let sender_id = self.get_sender_id(sender);
838        self.get_pool_data().pending_txs_by_sender(sender_id)
839    }
840
841    /// Returns the highest transaction of the address
842    pub fn get_highest_transaction_by_sender(
843        &self,
844        sender: Address,
845    ) -> Option<Arc<ValidPoolTransaction<T::Transaction>>> {
846        let sender_id = self.get_sender_id(sender);
847        self.get_pool_data().get_highest_transaction_by_sender(sender_id)
848    }
849
850    /// Returns the transaction with the highest nonce that is executable given the on chain nonce.
851    pub fn get_highest_consecutive_transaction_by_sender(
852        &self,
853        sender: Address,
854        on_chain_nonce: u64,
855    ) -> Option<Arc<ValidPoolTransaction<T::Transaction>>> {
856        let sender_id = self.get_sender_id(sender);
857        self.get_pool_data().get_highest_consecutive_transaction_by_sender(
858            sender_id.into_transaction_id(on_chain_nonce),
859        )
860    }
861
862    /// Returns the transaction given a [`TransactionId`]
863    pub fn get_transaction_by_transaction_id(
864        &self,
865        transaction_id: &TransactionId,
866    ) -> Option<Arc<ValidPoolTransaction<T::Transaction>>> {
867        self.get_pool_data().all().get(transaction_id).map(|tx| tx.transaction.clone())
868    }
869
870    /// Returns all transactions that where submitted with the given [`TransactionOrigin`]
871    pub fn get_transactions_by_origin(
872        &self,
873        origin: TransactionOrigin,
874    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
875        self.get_pool_data()
876            .all()
877            .transactions_iter()
878            .filter(|tx| tx.origin == origin)
879            .cloned()
880            .collect()
881    }
882
883    /// Returns all pending transactions filted by [`TransactionOrigin`]
884    pub fn get_pending_transactions_by_origin(
885        &self,
886        origin: TransactionOrigin,
887    ) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
888        self.get_pool_data().pending_transactions_iter().filter(|tx| tx.origin == origin).collect()
889    }
890
891    /// Returns all the transactions belonging to the hashes.
892    ///
893    /// If no transaction exists, it is skipped.
894    pub fn get_all(&self, txs: Vec<TxHash>) -> Vec<Arc<ValidPoolTransaction<T::Transaction>>> {
895        if txs.is_empty() {
896            return Vec::new()
897        }
898        self.get_pool_data().get_all(txs).collect()
899    }
900
901    /// Notify about propagated transactions.
902    pub fn on_propagated(&self, txs: PropagatedTransactions) {
903        if txs.0.is_empty() {
904            return
905        }
906        let mut listener = self.event_listener.write();
907
908        txs.0.into_iter().for_each(|(hash, peers)| listener.propagated(&hash, peers))
909    }
910
911    /// Number of transactions in the entire pool
912    pub fn len(&self) -> usize {
913        self.get_pool_data().len()
914    }
915
916    /// Whether the pool is empty
917    pub fn is_empty(&self) -> bool {
918        self.get_pool_data().is_empty()
919    }
920
921    /// Returns whether or not the pool is over its configured size and transaction count limits.
922    pub fn is_exceeded(&self) -> bool {
923        self.pool.read().is_exceeded()
924    }
925
926    /// Inserts a blob transaction into the blob store
927    fn insert_blob(&self, hash: TxHash, blob: BlobTransactionSidecarVariant) {
928        debug!(target: "txpool", "[{:?}] storing blob sidecar", hash);
929        if let Err(err) = self.blob_store.insert(hash, blob) {
930            warn!(target: "txpool", %err, "[{:?}] failed to insert blob", hash);
931            self.blob_store_metrics.blobstore_failed_inserts.increment(1);
932        }
933        self.update_blob_store_metrics();
934    }
935
936    /// Delete a blob from the blob store
937    pub fn delete_blob(&self, blob: TxHash) {
938        let _ = self.blob_store.delete(blob);
939    }
940
941    /// Delete all blobs from the blob store
942    pub fn delete_blobs(&self, txs: Vec<TxHash>) {
943        let _ = self.blob_store.delete_all(txs);
944    }
945
946    /// Cleans up the blob store
947    pub fn cleanup_blobs(&self) {
948        let stat = self.blob_store.cleanup();
949        self.blob_store_metrics.blobstore_failed_deletes.increment(stat.delete_failed as u64);
950        self.update_blob_store_metrics();
951    }
952
953    fn update_blob_store_metrics(&self) {
954        if let Some(data_size) = self.blob_store.data_size_hint() {
955            self.blob_store_metrics.blobstore_byte_size.set(data_size as f64);
956        }
957        self.blob_store_metrics.blobstore_entries.set(self.blob_store.blobs_len() as f64);
958    }
959
960    /// Deletes all blob transactions that were discarded.
961    fn delete_discarded_blobs<'a>(
962        &'a self,
963        transactions: impl IntoIterator<Item = &'a Arc<ValidPoolTransaction<T::Transaction>>>,
964    ) {
965        let blob_txs = transactions
966            .into_iter()
967            .filter(|tx| tx.transaction.is_eip4844())
968            .map(|tx| *tx.hash())
969            .collect();
970        self.delete_blobs(blob_txs);
971    }
972}
973
974impl<V, T: TransactionOrdering, S> fmt::Debug for PoolInner<V, T, S> {
975    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
976        f.debug_struct("PoolInner").field("config", &self.config).finish_non_exhaustive()
977    }
978}
979
980/// Tracks an added transaction and all graph changes caused by adding it.
981#[derive(Debug, Clone)]
982pub struct AddedPendingTransaction<T: PoolTransaction> {
983    /// Inserted transaction.
984    transaction: Arc<ValidPoolTransaction<T>>,
985    /// Replaced transaction.
986    replaced: Option<Arc<ValidPoolTransaction<T>>>,
987    /// transactions promoted to the pending queue
988    promoted: Vec<Arc<ValidPoolTransaction<T>>>,
989    /// transactions that failed and became discarded
990    discarded: Vec<Arc<ValidPoolTransaction<T>>>,
991}
992
993impl<T: PoolTransaction> AddedPendingTransaction<T> {
994    /// Returns all transactions that were promoted to the pending pool and adhere to the given
995    /// [`TransactionListenerKind`].
996    ///
997    /// If the kind is [`TransactionListenerKind::PropagateOnly`], then only transactions that
998    /// are allowed to be propagated are returned.
999    pub(crate) fn pending_transactions(
1000        &self,
1001        kind: TransactionListenerKind,
1002    ) -> impl Iterator<Item = B256> + '_ {
1003        let iter = std::iter::once(&self.transaction).chain(self.promoted.iter());
1004        PendingTransactionIter { kind, iter }
1005    }
1006
1007    /// Returns if the transaction should be propagated.
1008    pub(crate) fn is_propagate_allowed(&self) -> bool {
1009        self.transaction.propagate
1010    }
1011}
1012
1013pub(crate) struct PendingTransactionIter<Iter> {
1014    kind: TransactionListenerKind,
1015    iter: Iter,
1016}
1017
1018impl<'a, Iter, T> Iterator for PendingTransactionIter<Iter>
1019where
1020    Iter: Iterator<Item = &'a Arc<ValidPoolTransaction<T>>>,
1021    T: PoolTransaction + 'a,
1022{
1023    type Item = B256;
1024
1025    fn next(&mut self) -> Option<Self::Item> {
1026        loop {
1027            let next = self.iter.next()?;
1028            if self.kind.is_propagate_only() && !next.propagate {
1029                continue
1030            }
1031            return Some(*next.hash())
1032        }
1033    }
1034}
1035
1036/// An iterator over full pending transactions
1037pub(crate) struct FullPendingTransactionIter<Iter> {
1038    kind: TransactionListenerKind,
1039    iter: Iter,
1040}
1041
1042impl<'a, Iter, T> Iterator for FullPendingTransactionIter<Iter>
1043where
1044    Iter: Iterator<Item = &'a Arc<ValidPoolTransaction<T>>>,
1045    T: PoolTransaction + 'a,
1046{
1047    type Item = NewTransactionEvent<T>;
1048
1049    fn next(&mut self) -> Option<Self::Item> {
1050        loop {
1051            let next = self.iter.next()?;
1052            if self.kind.is_propagate_only() && !next.propagate {
1053                continue
1054            }
1055            return Some(NewTransactionEvent {
1056                subpool: SubPool::Pending,
1057                transaction: next.clone(),
1058            })
1059        }
1060    }
1061}
1062
1063/// Represents a transaction that was added into the pool and its state
1064#[derive(Debug, Clone)]
1065pub enum AddedTransaction<T: PoolTransaction> {
1066    /// Transaction was successfully added and moved to the pending pool.
1067    Pending(AddedPendingTransaction<T>),
1068    /// Transaction was successfully added but not yet ready for processing and moved to a
1069    /// parked pool instead.
1070    Parked {
1071        /// Inserted transaction.
1072        transaction: Arc<ValidPoolTransaction<T>>,
1073        /// Replaced transaction.
1074        replaced: Option<Arc<ValidPoolTransaction<T>>>,
1075        /// The subpool it was moved to.
1076        subpool: SubPool,
1077    },
1078}
1079
1080impl<T: PoolTransaction> AddedTransaction<T> {
1081    /// Returns whether the transaction has been added to the pending pool.
1082    pub(crate) const fn as_pending(&self) -> Option<&AddedPendingTransaction<T>> {
1083        match self {
1084            Self::Pending(tx) => Some(tx),
1085            _ => None,
1086        }
1087    }
1088
1089    /// Returns the replaced transaction if there was one
1090    pub(crate) const fn replaced(&self) -> Option<&Arc<ValidPoolTransaction<T>>> {
1091        match self {
1092            Self::Pending(tx) => tx.replaced.as_ref(),
1093            Self::Parked { replaced, .. } => replaced.as_ref(),
1094        }
1095    }
1096
1097    /// Returns the discarded transactions if there were any
1098    pub(crate) fn discarded_transactions(&self) -> Option<&[Arc<ValidPoolTransaction<T>>]> {
1099        match self {
1100            Self::Pending(tx) => Some(&tx.discarded),
1101            Self::Parked { .. } => None,
1102        }
1103    }
1104
1105    /// Returns the hash of the replaced transaction if it is a blob transaction.
1106    pub(crate) fn replaced_blob_transaction(&self) -> Option<B256> {
1107        self.replaced().filter(|tx| tx.transaction.is_eip4844()).map(|tx| *tx.transaction.hash())
1108    }
1109
1110    /// Returns the hash of the transaction
1111    pub(crate) fn hash(&self) -> &TxHash {
1112        match self {
1113            Self::Pending(tx) => tx.transaction.hash(),
1114            Self::Parked { transaction, .. } => transaction.hash(),
1115        }
1116    }
1117
1118    /// Converts this type into the event type for listeners
1119    pub(crate) fn into_new_transaction_event(self) -> NewTransactionEvent<T> {
1120        match self {
1121            Self::Pending(tx) => {
1122                NewTransactionEvent { subpool: SubPool::Pending, transaction: tx.transaction }
1123            }
1124            Self::Parked { transaction, subpool, .. } => {
1125                NewTransactionEvent { transaction, subpool }
1126            }
1127        }
1128    }
1129
1130    /// Returns the subpool this transaction was added to
1131    #[cfg(test)]
1132    pub(crate) const fn subpool(&self) -> SubPool {
1133        match self {
1134            Self::Pending(_) => SubPool::Pending,
1135            Self::Parked { subpool, .. } => *subpool,
1136        }
1137    }
1138
1139    /// Returns the [`TransactionId`] of the added transaction
1140    #[cfg(test)]
1141    pub(crate) fn id(&self) -> &TransactionId {
1142        match self {
1143            Self::Pending(added) => added.transaction.id(),
1144            Self::Parked { transaction, .. } => transaction.id(),
1145        }
1146    }
1147}
1148
1149/// Contains all state changes after a [`CanonicalStateUpdate`] was processed
1150#[derive(Debug)]
1151pub(crate) struct OnNewCanonicalStateOutcome<T: PoolTransaction> {
1152    /// Hash of the block.
1153    pub(crate) block_hash: B256,
1154    /// All mined transactions.
1155    pub(crate) mined: Vec<TxHash>,
1156    /// Transactions promoted to the pending pool.
1157    pub(crate) promoted: Vec<Arc<ValidPoolTransaction<T>>>,
1158    /// transaction that were discarded during the update
1159    pub(crate) discarded: Vec<Arc<ValidPoolTransaction<T>>>,
1160}
1161
1162impl<T: PoolTransaction> OnNewCanonicalStateOutcome<T> {
1163    /// Returns all transactions that were promoted to the pending pool and adhere to the given
1164    /// [`TransactionListenerKind`].
1165    ///
1166    /// If the kind is [`TransactionListenerKind::PropagateOnly`], then only transactions that
1167    /// are allowed to be propagated are returned.
1168    pub(crate) fn pending_transactions(
1169        &self,
1170        kind: TransactionListenerKind,
1171    ) -> impl Iterator<Item = B256> + '_ {
1172        let iter = self.promoted.iter();
1173        PendingTransactionIter { kind, iter }
1174    }
1175
1176    /// Returns all FULL transactions that were promoted to the pending pool and adhere to the given
1177    /// [`TransactionListenerKind`].
1178    ///
1179    /// If the kind is [`TransactionListenerKind::PropagateOnly`], then only transactions that
1180    /// are allowed to be propagated are returned.
1181    pub(crate) fn full_pending_transactions(
1182        &self,
1183        kind: TransactionListenerKind,
1184    ) -> impl Iterator<Item = NewTransactionEvent<T>> + '_ {
1185        let iter = self.promoted.iter();
1186        FullPendingTransactionIter { kind, iter }
1187    }
1188}
1189
1190#[cfg(test)]
1191mod tests {
1192    use crate::{
1193        blobstore::{BlobStore, InMemoryBlobStore},
1194        test_utils::{MockTransaction, TestPoolBuilder},
1195        validate::ValidTransaction,
1196        BlockInfo, PoolConfig, SubPoolLimit, TransactionOrigin, TransactionValidationOutcome, U256,
1197    };
1198    use alloy_eips::{eip4844::BlobTransactionSidecar, eip7594::BlobTransactionSidecarVariant};
1199    use std::{fs, path::PathBuf};
1200
1201    #[test]
1202    fn test_discard_blobs_on_blob_tx_eviction() {
1203        let blobs = {
1204            // Read the contents of the JSON file into a string.
1205            let json_content = fs::read_to_string(
1206                PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("test_data/blob1.json"),
1207            )
1208            .expect("Failed to read the blob data file");
1209
1210            // Parse the JSON contents into a serde_json::Value.
1211            let json_value: serde_json::Value =
1212                serde_json::from_str(&json_content).expect("Failed to deserialize JSON");
1213
1214            // Extract blob data from JSON and convert it to Blob.
1215            vec![
1216                // Extract the "data" field from the JSON and parse it as a string.
1217                json_value
1218                    .get("data")
1219                    .unwrap()
1220                    .as_str()
1221                    .expect("Data is not a valid string")
1222                    .to_string(),
1223            ]
1224        };
1225
1226        // Generate a BlobTransactionSidecar from the blobs.
1227        let sidecar = BlobTransactionSidecarVariant::Eip4844(
1228            BlobTransactionSidecar::try_from_blobs_hex(blobs).unwrap(),
1229        );
1230
1231        // Define the maximum limit for blobs in the sub-pool.
1232        let blob_limit = SubPoolLimit::new(1000, usize::MAX);
1233
1234        // Create a test pool with default configuration and the specified blob limit.
1235        let test_pool = &TestPoolBuilder::default()
1236            .with_config(PoolConfig { blob_limit, ..Default::default() })
1237            .pool;
1238
1239        // Set the block info for the pool, including a pending blob fee.
1240        test_pool
1241            .set_block_info(BlockInfo { pending_blob_fee: Some(10_000_000), ..Default::default() });
1242
1243        // Create an in-memory blob store.
1244        let blob_store = InMemoryBlobStore::default();
1245
1246        // Loop to add transactions to the pool and test blob eviction.
1247        for n in 0..blob_limit.max_txs + 10 {
1248            // Create a mock transaction with the generated blob sidecar.
1249            let mut tx = MockTransaction::eip4844_with_sidecar(sidecar.clone());
1250
1251            // Set non zero size
1252            tx.set_size(1844674407370951);
1253
1254            // Insert the sidecar into the blob store if the current index is within the blob limit.
1255            if n < blob_limit.max_txs {
1256                blob_store.insert(*tx.get_hash(), sidecar.clone()).unwrap();
1257            }
1258
1259            // Add the transaction to the pool with external origin and valid outcome.
1260            test_pool.add_transactions(
1261                TransactionOrigin::External,
1262                [TransactionValidationOutcome::Valid {
1263                    balance: U256::from(1_000),
1264                    state_nonce: 0,
1265                    bytecode_hash: None,
1266                    transaction: ValidTransaction::ValidWithSidecar {
1267                        transaction: tx,
1268                        sidecar: sidecar.clone(),
1269                    },
1270                    propagate: true,
1271                    authorities: None,
1272                }],
1273            );
1274        }
1275
1276        // Assert that the size of the pool's blob component is equal to the maximum blob limit.
1277        assert_eq!(test_pool.size().blob, blob_limit.max_txs);
1278
1279        // Assert that the size of the pool's blob_size component matches the expected value.
1280        assert_eq!(test_pool.size().blob_size, 1844674407370951000);
1281
1282        // Assert that the pool's blob store matches the expected blob store.
1283        assert_eq!(*test_pool.blob_store(), blob_store);
1284    }
1285}