1#![doc(
272 html_logo_url = "https://raw.githubusercontent.com/paradigmxyz/reth/main/assets/reth-docs.png",
273 html_favicon_url = "https://avatars0.githubusercontent.com/u/97369466?s=256",
274 issue_tracker_base_url = "https://github.com/paradigmxyz/reth/issues/"
275)]
276#![cfg_attr(docsrs, feature(doc_cfg))]
277#![cfg_attr(not(test), warn(unused_crate_dependencies))]
278
279pub use imbl::OrdMap;
280
281pub use crate::{
282 batcher::{BatchTxProcessor, BatchTxRequest},
283 blobstore::{BlobStore, BlobStoreError},
284 config::{
285 LocalTransactionConfig, PoolConfig, PriceBumpConfig, SubPoolLimit,
286 DEFAULT_MAX_INFLIGHT_DELEGATED_SLOTS, DEFAULT_PRICE_BUMP,
287 DEFAULT_TXPOOL_ADDITIONAL_VALIDATION_TASKS, MAX_NEW_PENDING_TXS_NOTIFICATIONS,
288 REPLACE_BLOB_PRICE_BUMP, TXPOOL_MAX_ACCOUNT_SLOTS_PER_SENDER,
289 TXPOOL_SUBPOOL_MAX_SIZE_MB_DEFAULT, TXPOOL_SUBPOOL_MAX_TXS_DEFAULT,
290 },
291 error::{PoolResult, RawPoolTransactionError},
292 ordering::{CoinbaseTipOrdering, Priority, TransactionOrdering},
293 pool::{
294 blob_tx_priority, fee_delta, state::SubPool, AddedTransactionOutcome,
295 AllTransactionsEvents, FullTransactionEvent, NewTransactionEvent, TransactionEvent,
296 TransactionEvents, TransactionListenerKind,
297 },
298 traits::*,
299 validate::{
300 EthTransactionValidator, StatefulValidationFn, StatelessValidationFn,
301 TransactionValidationOutcome, TransactionValidationTaskExecutor, TransactionValidator,
302 ValidPoolTransaction,
303 },
304};
305use crate::{identifier::TransactionId, pool::PoolInner};
306use alloy_eips::{
307 eip4844::{BlobAndProofV1, BlobAndProofV2, BlobCellsAndProofsV1},
308 eip7594::{BlobCellMask, BlobTransactionSidecarVariant},
309};
310use alloy_primitives::{map::AddressSet, Address, TxHash, B256, U256};
311use aquamarine as _;
312use reth_chainspec::{ChainSpecProvider, EthereumHardforks};
313use reth_eth_wire_types::HandleMempoolData;
314use reth_evm::ConfigureEvm;
315use reth_evm_ethereum::EthEvmConfig;
316use reth_execution_types::ChangedAccount;
317use reth_primitives_traits::{HeaderTy, Recovered};
318use reth_storage_api::{BlockReaderIdExt, StateProviderFactory};
319use std::sync::Arc;
320use tokio::sync::mpsc::Receiver;
321use tracing::{instrument, trace};
322
323pub mod error;
324pub mod maintain;
325pub mod metrics;
326pub mod noop;
327pub mod pool;
328pub mod validate;
329
330pub mod batcher;
331pub mod blobstore;
332mod config;
333pub mod identifier;
334mod ordering;
335mod traits;
336
337#[cfg(any(test, feature = "test-utils"))]
338pub mod test_utils;
340
341pub type EthTransactionPool<Client, S, Evm = EthEvmConfig, T = EthPooledTransaction> = Pool<
343 TransactionValidationTaskExecutor<EthTransactionValidator<Client, T, Evm>>,
344 CoinbaseTipOrdering<T>,
345 S,
346>;
347
348#[derive(Debug)]
350pub struct Pool<V, T: TransactionOrdering, S> {
351 pool: Arc<PoolInner<V, T, S>>,
353}
354
355impl<V, T, S> Pool<V, T, S>
358where
359 V: TransactionValidator,
360 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
361 S: BlobStore,
362{
363 pub fn new(validator: V, ordering: T, blob_store: S, config: PoolConfig) -> Self {
365 Self { pool: Arc::new(PoolInner::new(validator, ordering, blob_store, config)) }
366 }
367
368 pub fn inner(&self) -> &PoolInner<V, T, S> {
370 &self.pool
371 }
372
373 pub fn config(&self) -> &PoolConfig {
375 self.inner().config()
376 }
377
378 pub fn validator(&self) -> &V {
380 self.inner().validator()
381 }
382
383 async fn validate(
385 &self,
386 origin: TransactionOrigin,
387 transaction: V::Transaction,
388 ) -> TransactionValidationOutcome<V::Transaction> {
389 self.pool.validator().validate_transaction(origin, transaction).await
390 }
391
392 pub fn len(&self) -> usize {
394 self.pool.len()
395 }
396
397 pub fn is_empty(&self) -> bool {
399 self.pool.is_empty()
400 }
401
402 pub fn is_exceeded(&self) -> bool {
404 self.pool.is_exceeded()
405 }
406
407 pub fn blob_store(&self) -> &S {
409 self.pool.blob_store()
410 }
411}
412
413impl<Client, S, Evm> EthTransactionPool<Client, S, Evm>
414where
415 Client: ChainSpecProvider<ChainSpec: EthereumHardforks>
416 + StateProviderFactory
417 + Clone
418 + BlockReaderIdExt<Header = HeaderTy<Evm::Primitives>>
419 + 'static,
420 S: BlobStore,
421 Evm: ConfigureEvm + 'static,
422{
423 pub fn eth_pool(
458 validator: TransactionValidationTaskExecutor<
459 EthTransactionValidator<Client, EthPooledTransaction, Evm>,
460 >,
461 blob_store: S,
462 config: PoolConfig,
463 ) -> Self {
464 Self::new(validator, CoinbaseTipOrdering::default(), blob_store, config)
465 }
466}
467
468impl<V, T, S> TransactionPool for Pool<V, T, S>
470where
471 V: TransactionValidator,
472 <V as TransactionValidator>::Transaction: EthPoolTransaction,
473 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
474 S: BlobStore + Clone,
475{
476 type Transaction = T::Transaction;
477
478 fn pool_size(&self) -> PoolSize {
479 self.pool.size()
480 }
481
482 fn block_info(&self) -> BlockInfo {
483 self.pool.block_info()
484 }
485
486 async fn add_transaction_and_subscribe(
487 &self,
488 origin: TransactionOrigin,
489 transaction: Self::Transaction,
490 ) -> PoolResult<TransactionEvents> {
491 let tx = self.validate(origin, transaction).await;
492 self.pool.add_transaction_and_subscribe(origin, tx)
493 }
494
495 async fn add_transaction(
496 &self,
497 origin: TransactionOrigin,
498 transaction: Self::Transaction,
499 ) -> PoolResult<AddedTransactionOutcome> {
500 let tx = self.validate(origin, transaction).await;
501 let mut results = self.pool.add_transactions(origin, std::iter::once(tx));
502 results.pop().expect("result length is the same as the input")
503 }
504
505 async fn add_transactions(
506 &self,
507 origin: TransactionOrigin,
508 transactions: Vec<Self::Transaction>,
509 ) -> Vec<PoolResult<AddedTransactionOutcome>> {
510 if transactions.is_empty() {
511 return Vec::new()
512 }
513 let validated =
514 self.pool.validator().validate_transactions_with_origin(origin, transactions).await;
515 self.pool.add_transactions(origin, validated)
516 }
517
518 async fn add_transactions_with_origins(
519 &self,
520 transactions: Vec<(TransactionOrigin, Self::Transaction)>,
521 ) -> Vec<PoolResult<AddedTransactionOutcome>> {
522 if transactions.is_empty() {
523 return Vec::new()
524 }
525 let origins: Vec<_> = transactions.iter().map(|(origin, _)| *origin).collect();
526 let validated = self.pool.validator().validate_transactions(transactions).await;
527 self.pool.add_transactions_with_origins(origins.into_iter().zip(validated))
528 }
529
530 fn transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents> {
531 self.pool.add_transaction_event_listener(tx_hash)
532 }
533
534 fn all_transactions_event_listener(&self) -> AllTransactionsEvents<Self::Transaction> {
535 self.pool.add_all_transactions_event_listener()
536 }
537
538 fn pending_transactions_listener_for(&self, kind: TransactionListenerKind) -> Receiver<TxHash> {
539 self.pool.add_pending_listener(kind)
540 }
541
542 fn blob_transaction_sidecars_listener(&self) -> Receiver<NewBlobSidecar> {
543 self.pool.add_blob_sidecar_listener()
544 }
545
546 fn new_transactions_listener_for(
547 &self,
548 kind: TransactionListenerKind,
549 ) -> Receiver<NewTransactionEvent<Self::Transaction>> {
550 self.pool.add_new_transaction_listener(kind)
551 }
552
553 fn pooled_transaction_hashes(&self) -> Vec<TxHash> {
554 self.pool.pooled_transactions_hashes()
555 }
556
557 fn pooled_transaction_hashes_max(&self, max: usize) -> Vec<TxHash> {
558 self.pool.pooled_transactions_hashes_max(max)
559 }
560
561 fn pooled_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
562 self.pool.pooled_transactions()
563 }
564
565 fn pooled_transactions_max(
566 &self,
567 max: usize,
568 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
569 self.pool.pooled_transactions_max(max)
570 }
571
572 fn get_pooled_transaction_elements(
573 &self,
574 tx_hashes: Vec<TxHash>,
575 limit: GetPooledTransactionLimit,
576 ) -> Vec<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled> {
577 self.pool.get_pooled_transaction_elements(tx_hashes, limit)
578 }
579
580 fn append_pooled_transaction_elements(
581 &self,
582 tx_hashes: &[TxHash],
583 limit: GetPooledTransactionLimit,
584 out: &mut Vec<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled>,
585 ) {
586 self.pool.append_pooled_transaction_elements(tx_hashes, limit, out)
587 }
588
589 fn get_pooled_transaction_element(
590 &self,
591 tx_hash: TxHash,
592 ) -> Option<Recovered<<<V as TransactionValidator>::Transaction as PoolTransaction>::Pooled>>
593 {
594 self.pool.get_pooled_transaction_element(tx_hash)
595 }
596
597 fn best_transactions(
598 &self,
599 ) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>> {
600 Box::new(self.pool.best_transactions())
601 }
602
603 fn best_transactions_with_attributes(
604 &self,
605 best_transactions_attributes: BestTransactionsAttributes,
606 ) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>> {
607 self.pool.best_transactions_with_attributes(best_transactions_attributes)
608 }
609
610 fn pending_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
611 self.pool.pending_transactions()
612 }
613
614 fn get_pending_transaction_by_sender_and_nonce(
615 &self,
616 sender: Address,
617 nonce: u64,
618 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
619 self.pool.get_pending_transaction_by_sender_and_nonce(sender, nonce)
620 }
621
622 fn pending_transactions_max(
623 &self,
624 max: usize,
625 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
626 self.pool.pending_transactions_max(max)
627 }
628
629 fn queued_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
630 self.pool.queued_transactions()
631 }
632
633 fn pending_and_queued_txn_count(&self) -> (usize, usize) {
634 let data = self.pool.get_pool_data();
635 let pending = data.pending_transactions_count();
636 let queued = data.queued_transactions_count();
637 (pending, queued)
638 }
639
640 fn all_transactions(&self) -> AllPoolTransactions<Self::Transaction> {
641 self.pool.all_transactions()
642 }
643
644 fn all_transactions_by_sender(
645 &self,
646 sender: Address,
647 ) -> AllPoolTransactions<Self::Transaction> {
648 self.pool.all_transactions_by_sender(sender)
649 }
650
651 fn all_transaction_hashes(&self) -> Vec<TxHash> {
652 self.pool.all_transaction_hashes()
653 }
654
655 fn remove_transactions(
656 &self,
657 hashes: Vec<TxHash>,
658 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
659 self.pool.remove_transactions(hashes)
660 }
661
662 fn remove_transactions_and_descendants(
663 &self,
664 hashes: Vec<TxHash>,
665 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
666 self.pool.remove_transactions_and_descendants(hashes)
667 }
668
669 fn remove_transactions_by_sender(
670 &self,
671 sender: Address,
672 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
673 self.pool.remove_transactions_by_sender(sender)
674 }
675
676 fn prune_transactions(
677 &self,
678 hashes: Vec<TxHash>,
679 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
680 self.pool.prune_transactions(hashes)
681 }
682
683 fn retain_unknown<A>(&self, announcement: &mut A)
684 where
685 A: HandleMempoolData,
686 {
687 self.pool.retain_unknown(announcement)
688 }
689
690 fn retain_contains<A>(&self, announcement: &mut A)
691 where
692 A: HandleMempoolData,
693 {
694 self.pool.retain_contains(announcement)
695 }
696
697 fn get(&self, tx_hash: &TxHash) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
698 self.inner().get(tx_hash)
699 }
700
701 fn get_all(&self, txs: Vec<TxHash>) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
702 self.inner().get_all(txs)
703 }
704
705 fn on_propagated(&self, txs: PropagatedTransactions) {
706 self.inner().on_propagated(txs)
707 }
708
709 fn get_transactions_by_sender(
710 &self,
711 sender: Address,
712 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
713 self.pool.get_transactions_by_sender(sender)
714 }
715
716 fn get_pending_transactions_with_predicate(
717 &self,
718 predicate: impl FnMut(&ValidPoolTransaction<Self::Transaction>) -> bool,
719 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
720 self.pool.pending_transactions_with_predicate(predicate)
721 }
722
723 fn get_pending_transactions_by_sender(
724 &self,
725 sender: Address,
726 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
727 self.pool.get_pending_transactions_by_sender(sender)
728 }
729
730 fn get_queued_transactions_by_sender(
731 &self,
732 sender: Address,
733 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
734 self.pool.get_queued_transactions_by_sender(sender)
735 }
736
737 fn get_highest_transaction_by_sender(
738 &self,
739 sender: Address,
740 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
741 self.pool.get_highest_transaction_by_sender(sender)
742 }
743
744 fn get_highest_consecutive_transaction_by_sender(
745 &self,
746 sender: Address,
747 on_chain_nonce: u64,
748 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
749 self.pool.get_highest_consecutive_transaction_by_sender(sender, on_chain_nonce)
750 }
751
752 fn get_transaction_by_sender_and_nonce(
753 &self,
754 sender: Address,
755 nonce: u64,
756 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
757 let sender_id = self.pool.sender_id(&sender)?;
758 let transaction_id = TransactionId::new(sender_id, nonce);
759
760 self.inner().get_pool_data().all().get(&transaction_id).map(|tx| tx.transaction.clone())
761 }
762
763 fn get_transactions_by_origin(
764 &self,
765 origin: TransactionOrigin,
766 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
767 self.pool.get_transactions_by_origin(origin)
768 }
769
770 fn get_pending_transactions_by_origin(
772 &self,
773 origin: TransactionOrigin,
774 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
775 self.pool.get_pending_transactions_by_origin(origin)
776 }
777
778 fn unique_senders(&self) -> AddressSet {
779 self.pool.unique_senders()
780 }
781
782 fn get_blob(
783 &self,
784 tx_hash: TxHash,
785 ) -> Result<Option<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
786 self.pool.blob_store().get(tx_hash)
787 }
788
789 fn get_all_blobs(
790 &self,
791 tx_hashes: Vec<TxHash>,
792 ) -> Result<Vec<(TxHash, Arc<BlobTransactionSidecarVariant>)>, BlobStoreError> {
793 self.pool.blob_store().get_all(tx_hashes)
794 }
795
796 fn get_all_blobs_exact(
797 &self,
798 tx_hashes: Vec<TxHash>,
799 ) -> Result<Vec<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
800 self.pool.blob_store().get_exact(tx_hashes)
801 }
802
803 fn get_blobs_for_versioned_hashes_v1(
804 &self,
805 versioned_hashes: &[B256],
806 ) -> Result<Vec<Option<BlobAndProofV1>>, BlobStoreError> {
807 self.pool.blob_store().get_by_versioned_hashes_v1(versioned_hashes)
808 }
809
810 fn get_blobs_for_versioned_hashes_v2(
811 &self,
812 versioned_hashes: &[B256],
813 ) -> Result<Option<Vec<BlobAndProofV2>>, BlobStoreError> {
814 self.pool.blob_store().get_by_versioned_hashes_v2(versioned_hashes)
815 }
816
817 fn get_blobs_for_versioned_hashes_v3(
818 &self,
819 versioned_hashes: &[B256],
820 ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError> {
821 self.pool.blob_store().get_by_versioned_hashes_v3(versioned_hashes)
822 }
823
824 fn get_blobs_for_versioned_hashes_v4(
825 &self,
826 versioned_hashes: &[B256],
827 cell_mask: BlobCellMask,
828 ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
829 self.pool.blob_store().get_by_versioned_hashes_v4(versioned_hashes, cell_mask)
830 }
831
832 fn has_blobs_for_versioned_hashes(
833 &self,
834 versioned_hashes: &[B256],
835 ) -> Result<Vec<bool>, BlobStoreError> {
836 self.pool.blob_store().has_versioned_hashes(versioned_hashes)
837 }
838
839 fn blob_store(&self) -> Box<dyn BlobStore> {
840 Box::new(self.pool.blob_store().clone())
841 }
842}
843
844impl<V, T, S> TransactionPoolExt for Pool<V, T, S>
845where
846 V: TransactionValidator,
847 <V as TransactionValidator>::Transaction: EthPoolTransaction,
848 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
849 S: BlobStore + Clone,
850{
851 type Block = V::Block;
852
853 #[instrument(skip(self), target = "txpool")]
854 fn set_block_info(&self, info: BlockInfo) {
855 trace!(target: "txpool", "updating pool block info");
856 self.pool.set_block_info(info)
857 }
858
859 fn on_canonical_state_change(&self, update: CanonicalStateUpdate<'_, Self::Block>) {
860 self.pool.on_canonical_state_change(update);
861 }
862
863 fn update_accounts(&self, accounts: Vec<ChangedAccount>) {
864 self.pool.update_accounts(accounts);
865 }
866
867 fn delete_blob(&self, tx: TxHash) {
868 self.pool.delete_blob(tx)
869 }
870
871 fn delete_blobs(&self, txs: Vec<TxHash>) {
872 self.pool.delete_blobs(txs)
873 }
874
875 fn cleanup_blobs(&self) {
876 self.pool.cleanup_blobs()
877 }
878}
879
880impl<V, T, S> ValidatingPool for Pool<V, T, S>
881where
882 V: TransactionValidator,
883 <V as TransactionValidator>::Transaction: EthPoolTransaction,
884 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
885 S: BlobStore + Clone,
886{
887 type Validator = V;
888
889 fn validator(&self) -> &Self::Validator {
890 self.inner().validator()
891 }
892}
893
894impl<V, T: TransactionOrdering, S> Clone for Pool<V, T, S> {
895 fn clone(&self) -> Self {
896 Self { pool: Arc::clone(&self.pool) }
897 }
898}