feat: add blob transactions subpool (#4608)

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

@@ -0,0 +1,167 @@
+#![allow(dead_code, unused)]
+use crate::{
+    identifier::TransactionId, pool::size::SizeTracker, PoolTransaction, ValidPoolTransaction,
+};
+use std::{
+    cmp::Ordering,
+    collections::{BTreeMap, BTreeSet},
+    sync::Arc,
+};
+
+/// A set of __all__ validated blob transactions in the pool.
+///
+/// The purpose of this pool is keep track of blob transactions that are either pending or queued
+/// and to evict the worst blob transactions once the sub-pool is full.
+///
+/// This expects that certain constraints are met:
+///   - blob transactions are always gap less
+pub(crate) struct BlobTransactions<T: PoolTransaction> {
+    /// Keeps track of transactions inserted in the pool.
+    ///
+    /// This way we can determine when transactions were submitted to the pool.
+    submission_id: u64,
+    /// _All_ Transactions that are currently inside the pool grouped by their identifier.
+    by_id: BTreeMap<TransactionId, Arc<ValidPoolTransaction<T>>>,
+    /// _All_ transactions sorted by blob priority.
+    all: BTreeSet<BlobTransaction<T>>,
+    /// Keeps track of the size of this pool.
+    ///
+    /// See also [`PoolTransaction::size`](crate::traits::PoolTransaction::size).
+    size_of: SizeTracker,
+}
+
+// === impl BlobTransactions ===
+
+impl<T: PoolTransaction> BlobTransactions<T> {
+    /// Adds a new transactions to the pending queue.
+    ///
+    /// # Panics
+    ///
+    ///   - If the transaction is not a blob tx.
+    ///   - If the transaction is already included.
+    pub(crate) fn add_transaction(&mut self, tx: Arc<ValidPoolTransaction<T>>) {
+        assert!(tx.is_eip4844(), "transaction is not a blob tx");
+        let id = *tx.id();
+        assert!(
+            !self.by_id.contains_key(&id),
+            "transaction already included {:?}",
+            self.by_id.contains_key(&id)
+        );
+        let submission_id = self.next_id();
+
+        // keep track of size
+        self.size_of += tx.size();
+
+        self.by_id.insert(id, tx.clone());
+
+        let ord = BlobOrd { submission_id };
+        let transaction = BlobTransaction { ord, transaction: tx };
+        self.all.insert(transaction);
+    }
+
+    /// Removes the transaction from the pool
+    pub(crate) fn remove_transaction(
+        &mut self,
+        id: &TransactionId,
+    ) -> Option<Arc<ValidPoolTransaction<T>>> {
+        // remove from queues
+        let tx = self.by_id.remove(id)?;
+
+        // TODO: remove from ordered set
+        // self.best.remove(&tx);
+
+        // keep track of size
+        self.size_of -= tx.transaction.size();
+
+        Some(tx)
+    }
+
+    fn next_id(&mut self) -> u64 {
+        let id = self.submission_id;
+        self.submission_id = self.submission_id.wrapping_add(1);
+        id
+    }
+
+    /// The reported size of all transactions in this pool.
+    pub(crate) fn size(&self) -> usize {
+        self.size_of.into()
+    }
+
+    /// Number of transactions in the entire pool
+    pub(crate) fn len(&self) -> usize {
+        self.by_id.len()
+    }
+
+    /// Returns `true` if the transaction with the given id is already included in this pool.
+    #[cfg(test)]
+    #[allow(unused)]
+    pub(crate) fn contains(&self, id: &TransactionId) -> bool {
+        self.by_id.contains_key(id)
+    }
+}
+
+impl<T: PoolTransaction> Default for BlobTransactions<T> {
+    fn default() -> Self {
+        Self {
+            submission_id: 0,
+            by_id: Default::default(),
+            all: Default::default(),
+            size_of: Default::default(),
+        }
+    }
+}
+
+/// A transaction that is ready to be included in a block.
+struct BlobTransaction<T: PoolTransaction> {
+    /// Actual blob transaction.
+    transaction: Arc<ValidPoolTransaction<T>>,
+    /// The value that determines the order of this transaction.
+    ord: BlobOrd,
+}
+
+impl<T: PoolTransaction> Eq for BlobTransaction<T> {}
+
+impl<T: PoolTransaction> PartialEq<Self> for BlobTransaction<T> {
+    fn eq(&self, other: &Self) -> bool {
+        self.cmp(other) == Ordering::Equal
+    }
+}
+
+impl<T: PoolTransaction> PartialOrd<Self> for BlobTransaction<T> {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl<T: PoolTransaction> Ord for BlobTransaction<T> {
+    fn cmp(&self, other: &Self) -> Ordering {
+        self.ord.cmp(&other.ord)
+    }
+}
+
+#[derive(Debug)]
+struct BlobOrd {
+    /// Identifier that tags when transaction was submitted in the pool.
+    pub(crate) submission_id: u64,
+    // TODO(mattsse): add ord values
+}
+
+impl Eq for BlobOrd {}
+
+impl PartialEq<Self> for BlobOrd {
+    fn eq(&self, other: &Self) -> bool {
+        self.cmp(other) == Ordering::Equal
+    }
+}
+
+impl PartialOrd<Self> for BlobOrd {
+    fn partial_cmp(&self, other: &Self) -> Option<Ordering> {
+        Some(self.cmp(other))
+    }
+}
+
+impl Ord for BlobOrd {
+    fn cmp(&self, other: &Self) -> Ordering {
+        other.submission_id.cmp(&self.submission_id)
+    }
+}

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

@@ -110,6 +110,7 @@ pub use listener::{AllTransactionsEvents, TransactionEvents};
 use reth_rlp::Encodable;
 
 mod best;
+mod blob;
 mod parked;
 pub(crate) mod pending;
 pub(crate) mod size;

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

@@ -17,7 +17,7 @@ use std::{cmp::Ordering, collections::BTreeSet, ops::Deref, sync::Arc};
 pub(crate) struct ParkedPool<T: ParkedOrd> {
     /// Keeps track of transactions inserted in the pool.
     ///
-    /// This way we can determine when transactions where submitted to the pool.
+    /// This way we can determine when transactions were submitted to the pool.
     submission_id: u64,
     /// _All_ Transactions that are currently inside the pool grouped by their identifier.
     by_id: FnvHashMap<TransactionId, ParkedPoolTransaction<T>>,

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

@@ -28,7 +28,7 @@ pub(crate) struct PendingPool<T: TransactionOrdering> {
     ordering: T,
     /// Keeps track of transactions inserted in the pool.
     ///
-    /// This way we can determine when transactions where submitted to the pool.
+    /// This way we can determine when transactions were submitted to the pool.
     submission_id: u64,
     /// _All_ Transactions that are currently inside the pool grouped by their identifier.
     by_id: BTreeMap<TransactionId, PendingTransaction<T>>,

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

@@ -6,6 +6,7 @@ use crate::{
     metrics::TxPoolMetrics,
     pool::{
         best::BestTransactions,
+        blob::BlobTransactions,
         parked::{BasefeeOrd, ParkedPool, QueuedOrd},
         pending::PendingPool,
         state::{SubPool, TxState},
@@ -86,6 +87,9 @@ pub struct TxPool<T: TransactionOrdering> {
     /// Holds all parked transactions that currently violate the dynamic fee requirement but could
     /// be moved to pending if the base fee changes in their favor (decreases) in future blocks.
     basefee_pool: ParkedPool<BasefeeOrd<T::Transaction>>,
+    /// All blob transactions in the pool
+    #[allow(unused)]
+    blob_transactions: BlobTransactions<T::Transaction>,
     /// All transactions in the pool.
     all_transactions: AllTransactions<T::Transaction>,
     /// Transaction pool metrics
@@ -102,6 +106,7 @@ impl<T: TransactionOrdering> TxPool<T> {
             pending_pool: PendingPool::new(ordering),
             queued_pool: Default::default(),
             basefee_pool: Default::default(),
+            blob_transactions: Default::default(),
             all_transactions: AllTransactions::new(&config),
             config,
             metrics: Default::default(),