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::BlobTransactionSidecarVariant,
309};
310use alloy_primitives::{map::AddressSet, Address, TxHash, B128, 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_transaction_hashes(&self) -> Vec<TxHash> {
645 self.pool.all_transaction_hashes()
646 }
647
648 fn remove_transactions(
649 &self,
650 hashes: Vec<TxHash>,
651 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
652 self.pool.remove_transactions(hashes)
653 }
654
655 fn remove_transactions_and_descendants(
656 &self,
657 hashes: Vec<TxHash>,
658 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
659 self.pool.remove_transactions_and_descendants(hashes)
660 }
661
662 fn remove_transactions_by_sender(
663 &self,
664 sender: Address,
665 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
666 self.pool.remove_transactions_by_sender(sender)
667 }
668
669 fn prune_transactions(
670 &self,
671 hashes: Vec<TxHash>,
672 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
673 self.pool.prune_transactions(hashes)
674 }
675
676 fn retain_unknown<A>(&self, announcement: &mut A)
677 where
678 A: HandleMempoolData,
679 {
680 self.pool.retain_unknown(announcement)
681 }
682
683 fn retain_contains<A>(&self, announcement: &mut A)
684 where
685 A: HandleMempoolData,
686 {
687 self.pool.retain_contains(announcement)
688 }
689
690 fn get(&self, tx_hash: &TxHash) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
691 self.inner().get(tx_hash)
692 }
693
694 fn get_all(&self, txs: Vec<TxHash>) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
695 self.inner().get_all(txs)
696 }
697
698 fn on_propagated(&self, txs: PropagatedTransactions) {
699 self.inner().on_propagated(txs)
700 }
701
702 fn get_transactions_by_sender(
703 &self,
704 sender: Address,
705 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
706 self.pool.get_transactions_by_sender(sender)
707 }
708
709 fn get_pending_transactions_with_predicate(
710 &self,
711 predicate: impl FnMut(&ValidPoolTransaction<Self::Transaction>) -> bool,
712 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
713 self.pool.pending_transactions_with_predicate(predicate)
714 }
715
716 fn get_pending_transactions_by_sender(
717 &self,
718 sender: Address,
719 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
720 self.pool.get_pending_transactions_by_sender(sender)
721 }
722
723 fn get_queued_transactions_by_sender(
724 &self,
725 sender: Address,
726 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
727 self.pool.get_queued_transactions_by_sender(sender)
728 }
729
730 fn get_highest_transaction_by_sender(
731 &self,
732 sender: Address,
733 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
734 self.pool.get_highest_transaction_by_sender(sender)
735 }
736
737 fn get_highest_consecutive_transaction_by_sender(
738 &self,
739 sender: Address,
740 on_chain_nonce: u64,
741 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
742 self.pool.get_highest_consecutive_transaction_by_sender(sender, on_chain_nonce)
743 }
744
745 fn get_transaction_by_sender_and_nonce(
746 &self,
747 sender: Address,
748 nonce: u64,
749 ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
750 let sender_id = self.pool.sender_id(&sender)?;
751 let transaction_id = TransactionId::new(sender_id, nonce);
752
753 self.inner().get_pool_data().all().get(&transaction_id).map(|tx| tx.transaction.clone())
754 }
755
756 fn get_transactions_by_origin(
757 &self,
758 origin: TransactionOrigin,
759 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
760 self.pool.get_transactions_by_origin(origin)
761 }
762
763 fn get_pending_transactions_by_origin(
765 &self,
766 origin: TransactionOrigin,
767 ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
768 self.pool.get_pending_transactions_by_origin(origin)
769 }
770
771 fn unique_senders(&self) -> AddressSet {
772 self.pool.unique_senders()
773 }
774
775 fn get_blob(
776 &self,
777 tx_hash: TxHash,
778 ) -> Result<Option<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
779 self.pool.blob_store().get(tx_hash)
780 }
781
782 fn get_all_blobs(
783 &self,
784 tx_hashes: Vec<TxHash>,
785 ) -> Result<Vec<(TxHash, Arc<BlobTransactionSidecarVariant>)>, BlobStoreError> {
786 self.pool.blob_store().get_all(tx_hashes)
787 }
788
789 fn get_all_blobs_exact(
790 &self,
791 tx_hashes: Vec<TxHash>,
792 ) -> Result<Vec<Arc<BlobTransactionSidecarVariant>>, BlobStoreError> {
793 self.pool.blob_store().get_exact(tx_hashes)
794 }
795
796 fn get_blobs_for_versioned_hashes_v1(
797 &self,
798 versioned_hashes: &[B256],
799 ) -> Result<Vec<Option<BlobAndProofV1>>, BlobStoreError> {
800 self.pool.blob_store().get_by_versioned_hashes_v1(versioned_hashes)
801 }
802
803 fn get_blobs_for_versioned_hashes_v2(
804 &self,
805 versioned_hashes: &[B256],
806 ) -> Result<Option<Vec<BlobAndProofV2>>, BlobStoreError> {
807 self.pool.blob_store().get_by_versioned_hashes_v2(versioned_hashes)
808 }
809
810 fn get_blobs_for_versioned_hashes_v3(
811 &self,
812 versioned_hashes: &[B256],
813 ) -> Result<Vec<Option<BlobAndProofV2>>, BlobStoreError> {
814 self.pool.blob_store().get_by_versioned_hashes_v3(versioned_hashes)
815 }
816
817 fn get_blobs_for_versioned_hashes_v4(
818 &self,
819 versioned_hashes: &[B256],
820 indices_bitarray: B128,
821 ) -> Result<Vec<Option<BlobCellsAndProofsV1>>, BlobStoreError> {
822 self.pool.blob_store().get_by_versioned_hashes_v4(versioned_hashes, indices_bitarray)
823 }
824
825 fn has_blobs_for_versioned_hashes(
826 &self,
827 versioned_hashes: &[B256],
828 ) -> Result<Vec<bool>, BlobStoreError> {
829 self.pool.blob_store().has_versioned_hashes(versioned_hashes)
830 }
831
832 fn blob_store(&self) -> Box<dyn BlobStore> {
833 Box::new(self.pool.blob_store().clone())
834 }
835}
836
837impl<V, T, S> TransactionPoolExt for Pool<V, T, S>
838where
839 V: TransactionValidator,
840 <V as TransactionValidator>::Transaction: EthPoolTransaction,
841 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
842 S: BlobStore + Clone,
843{
844 type Block = V::Block;
845
846 #[instrument(skip(self), target = "txpool")]
847 fn set_block_info(&self, info: BlockInfo) {
848 trace!(target: "txpool", "updating pool block info");
849 self.pool.set_block_info(info)
850 }
851
852 fn on_canonical_state_change(&self, update: CanonicalStateUpdate<'_, Self::Block>) {
853 self.pool.on_canonical_state_change(update);
854 }
855
856 fn update_accounts(&self, accounts: Vec<ChangedAccount>) {
857 self.pool.update_accounts(accounts);
858 }
859
860 fn delete_blob(&self, tx: TxHash) {
861 self.pool.delete_blob(tx)
862 }
863
864 fn delete_blobs(&self, txs: Vec<TxHash>) {
865 self.pool.delete_blobs(txs)
866 }
867
868 fn cleanup_blobs(&self) {
869 self.pool.cleanup_blobs()
870 }
871}
872
873impl<V, T, S> ValidatingPool for Pool<V, T, S>
874where
875 V: TransactionValidator,
876 <V as TransactionValidator>::Transaction: EthPoolTransaction,
877 T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
878 S: BlobStore + Clone,
879{
880 type Validator = V;
881
882 fn validator(&self) -> &Self::Validator {
883 self.inner().validator()
884 }
885}
886
887impl<V, T: TransactionOrdering, S> Clone for Pool<V, T, S> {
888 fn clone(&self) -> Self {
889 Self { pool: Arc::clone(&self.pool) }
890 }
891}