Merge pull request #61 from logos-co/aa/mempool-setup

Setup custom mempool

sz-poc-offsite-2025/Cargo.toml

@@ -1,5 +1,5 @@
 [workspace]
-members = ["node"]
+members = ["evm/node", "evm/sequencer-mempool"]
 resolver = "3"
 
 [workspace.package]
@@ -7,7 +7,15 @@ edition = "2024"
 
 [workspace.dependencies]
 # Internal
-node = { path = "node" }
+evm-node = { path = "evm/node" }
+evm-sequencer-mempool = { path = "evm/sequencer-mempool" }
 
 # External
+clap = { version = "4.5" }
+eyre = { version = "0.6" }
+futures-util = { version = "0.3" }
+jsonrpsee = { version = "0.24" }
+reth = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.8" }
+reth-cli-commands = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.8" }
+reth-node-ethereum = { git = "https://github.com/paradigmxyz/reth", tag = "v1.3.8" }
 tokio = { version = "1.0" }

sz-poc-offsite-2025/evm/node/Cargo.toml

@@ -0,0 +1,13 @@
+[package]
+name = "evm-node"
+edition = { workspace = true }
+
+[dependencies]
+evm-sequencer-mempool = { workspace = true }
+
+clap = { workspace = true }
+futures-util = { workspace = true }
+jsonrpsee = { workspace = true }
+reth = { workspace = true }
+reth-cli-commands = { workspace = true }
+reth-node-ethereum = { workspace = true }

sz-poc-offsite-2025/evm/node/src/main.rs

@@ -0,0 +1,26 @@
+use reth::cli::Cli;
+use reth_node_ethereum::node::{EthereumAddOns, EthereumNode};
+
+use evm_sequencer_mempool::EvmSequencerMempoolBuilder;
+
+fn main() {
+    Cli::parse_args()
+        .run(|builder, _| async move {
+            // launch the node
+            let handle = Box::pin(
+                builder
+                    // use the default ethereum node types
+                    .with_types::<EthereumNode>()
+                    // use default ethereum components but use our custom pool
+                    .with_components(
+                        EthereumNode::components().pool(EvmSequencerMempoolBuilder::default()),
+                    )
+                    .with_add_ons(EthereumAddOns::default())
+                    .launch(),
+            )
+            .await?;
+
+            handle.wait_for_node_exit().await
+        })
+        .unwrap();
+}

sz-poc-offsite-2025/evm/sequencer-mempool/Cargo.toml

@@ -0,0 +1,9 @@
+[package]
+name = "evm-sequencer-mempool"
+edition = { workspace = true }
+
+[dependencies]
+eyre = { workspace = true }
+reth = { workspace = true }
+reth-node-ethereum = { workspace = true }
+tokio = { workspace = true }

sz-poc-offsite-2025/evm/sequencer-mempool/src/lib.rs

@@ -0,0 +1,365 @@
+use std::{collections::HashSet, sync::Arc};
+use tokio::sync::mpsc::Receiver;
+
+use reth::{
+    api::{Block, FullNodeTypes, NodeTypes},
+    builder::{BuilderContext, components::PoolBuilder},
+    chainspec::ChainSpec,
+    network::types::HandleMempoolData,
+    primitives::{EthPrimitives, Recovered},
+    providers::ChangedAccount,
+    revm::primitives::{Address, B256, alloy_primitives::TxHash},
+    rpc::types::{BlobTransactionSidecar, engine::BlobAndProofV1},
+    transaction_pool::{
+        AllPoolTransactions, AllTransactionsEvents, BestTransactions, BestTransactionsAttributes,
+        BlobStore, BlobStoreError, BlockInfo, CanonicalStateUpdate, CoinbaseTipOrdering,
+        EthPoolTransaction, EthPooledTransaction, EthTransactionValidator,
+        GetPooledTransactionLimit, NewBlobSidecar, NewTransactionEvent, Pool, PoolResult, PoolSize,
+        PoolTransaction, PropagatedTransactions, TransactionEvents, TransactionListenerKind,
+        TransactionOrdering, TransactionOrigin, TransactionPool, TransactionPoolExt,
+        TransactionValidationTaskExecutor, TransactionValidator, ValidPoolTransaction,
+        blobstore::DiskFileBlobStore,
+    },
+};
+use reth_node_ethereum::node::EthereumPoolBuilder;
+
+#[derive(Default)]
+pub struct EvmSequencerMempoolBuilder(EthereumPoolBuilder);
+
+impl<Types, Node> PoolBuilder<Node> for EvmSequencerMempoolBuilder
+where
+    Types: NodeTypes<ChainSpec = ChainSpec, Primitives = EthPrimitives>,
+    Node: FullNodeTypes<Types = Types>,
+{
+    type Pool = EvmSequencerMempool<Node::Provider, DiskFileBlobStore>;
+
+    async fn build_pool(self, ctx: &BuilderContext<Node>) -> eyre::Result<Self::Pool> {
+        let pool = self.0.build_pool(ctx).await?;
+        Ok(EvmSequencerPool(pool))
+    }
+}
+
+pub type EvmSequencerMempool<Client, S> = EvmSequencerPool<
+    TransactionValidationTaskExecutor<EthTransactionValidator<Client, EthPooledTransaction>>,
+    CoinbaseTipOrdering<EthPooledTransaction>,
+    S,
+>;
+
+#[derive(Debug)]
+pub struct EvmSequencerPool<V, T, S>(Pool<V, T, S>)
+where
+    T: TransactionOrdering;
+
+impl<V, T, S> TransactionPool for EvmSequencerPool<V, T, S>
+where
+    V: TransactionValidator,
+    <V as TransactionValidator>::Transaction: EthPoolTransaction,
+    T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
+    S: BlobStore,
+{
+    type Transaction = T::Transaction;
+
+    fn pool_size(&self) -> PoolSize {
+        self.0.pool_size()
+    }
+
+    fn block_info(&self) -> BlockInfo {
+        self.0.block_info()
+    }
+
+    fn add_transaction_and_subscribe(
+        &self,
+        origin: TransactionOrigin,
+        transaction: Self::Transaction,
+    ) -> impl Future<Output = PoolResult<TransactionEvents>> + Send {
+        self.0.add_transaction_and_subscribe(origin, transaction)
+    }
+
+    fn add_transaction(
+        &self,
+        origin: TransactionOrigin,
+        transaction: Self::Transaction,
+    ) -> impl Future<Output = PoolResult<TxHash>> + Send {
+        self.0.add_transaction(origin, transaction)
+    }
+
+    fn add_transactions(
+        &self,
+        origin: TransactionOrigin,
+        transactions: Vec<Self::Transaction>,
+    ) -> impl Future<Output = Vec<PoolResult<TxHash>>> + Send {
+        self.0.add_transactions(origin, transactions)
+    }
+
+    fn transaction_event_listener(&self, tx_hash: TxHash) -> Option<TransactionEvents> {
+        self.0.transaction_event_listener(tx_hash)
+    }
+
+    fn all_transactions_event_listener(&self) -> AllTransactionsEvents<Self::Transaction> {
+        self.0.all_transactions_event_listener()
+    }
+
+    fn pending_transactions_listener_for(&self, kind: TransactionListenerKind) -> Receiver<TxHash> {
+        self.0.pending_transactions_listener_for(kind)
+    }
+
+    fn blob_transaction_sidecars_listener(&self) -> Receiver<NewBlobSidecar> {
+        self.0.blob_transaction_sidecars_listener()
+    }
+
+    fn new_transactions_listener_for(
+        &self,
+        kind: TransactionListenerKind,
+    ) -> Receiver<NewTransactionEvent<Self::Transaction>> {
+        self.0.new_transactions_listener_for(kind)
+    }
+
+    fn pooled_transaction_hashes(&self) -> Vec<TxHash> {
+        self.0.pooled_transaction_hashes()
+    }
+
+    fn pooled_transaction_hashes_max(&self, max: usize) -> Vec<TxHash> {
+        self.0.pooled_transaction_hashes_max(max)
+    }
+
+    fn pooled_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.pooled_transactions()
+    }
+
+    fn pooled_transactions_max(
+        &self,
+        max: usize,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.pooled_transactions_max(max)
+    }
+
+    fn get_pooled_transaction_elements(
+        &self,
+        tx_hashes: Vec<TxHash>,
+        limit: GetPooledTransactionLimit,
+    ) -> Vec<<Self::Transaction as PoolTransaction>::Pooled> {
+        self.0.get_pooled_transaction_elements(tx_hashes, limit)
+    }
+
+    fn get_pooled_transaction_element(
+        &self,
+        tx_hash: TxHash,
+    ) -> Option<Recovered<<Self::Transaction as PoolTransaction>::Pooled>> {
+        self.0.get_pooled_transaction_element(tx_hash)
+    }
+
+    fn best_transactions(
+        &self,
+    ) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>> {
+        self.0.best_transactions()
+    }
+
+    fn best_transactions_with_attributes(
+        &self,
+        best_transactions_attributes: BestTransactionsAttributes,
+    ) -> Box<dyn BestTransactions<Item = Arc<ValidPoolTransaction<Self::Transaction>>>> {
+        self.0
+            .best_transactions_with_attributes(best_transactions_attributes)
+    }
+
+    fn pending_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.pending_transactions()
+    }
+
+    fn pending_transactions_max(
+        &self,
+        max: usize,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.pending_transactions_max(max)
+    }
+
+    fn queued_transactions(&self) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.queued_transactions()
+    }
+
+    fn all_transactions(&self) -> AllPoolTransactions<Self::Transaction> {
+        self.0.all_transactions()
+    }
+
+    fn remove_transactions(
+        &self,
+        hashes: Vec<TxHash>,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.remove_transactions(hashes)
+    }
+
+    fn remove_transactions_and_descendants(
+        &self,
+        hashes: Vec<TxHash>,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.remove_transactions_and_descendants(hashes)
+    }
+
+    fn remove_transactions_by_sender(
+        &self,
+        sender: Address,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.remove_transactions_by_sender(sender)
+    }
+
+    fn retain_unknown<A>(&self, announcement: &mut A)
+    where
+        A: HandleMempoolData,
+    {
+        self.0.retain_unknown(announcement);
+    }
+
+    fn get(&self, tx_hash: &TxHash) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get(tx_hash)
+    }
+
+    fn get_all(&self, txs: Vec<TxHash>) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_all(txs)
+    }
+
+    fn on_propagated(&self, txs: PropagatedTransactions) {
+        self.0.on_propagated(txs);
+    }
+
+    fn get_transactions_by_sender(
+        &self,
+        sender: Address,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_transactions_by_sender(sender)
+    }
+
+    fn get_pending_transactions_with_predicate(
+        &self,
+        predicate: impl FnMut(&ValidPoolTransaction<Self::Transaction>) -> bool,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_pending_transactions_with_predicate(predicate)
+    }
+
+    fn get_pending_transactions_by_sender(
+        &self,
+        sender: Address,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_pending_transactions_by_sender(sender)
+    }
+
+    fn get_queued_transactions_by_sender(
+        &self,
+        sender: Address,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_queued_transactions_by_sender(sender)
+    }
+
+    fn get_highest_transaction_by_sender(
+        &self,
+        sender: Address,
+    ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_highest_transaction_by_sender(sender)
+    }
+
+    fn get_highest_consecutive_transaction_by_sender(
+        &self,
+        sender: Address,
+        on_chain_nonce: u64,
+    ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0
+            .get_highest_consecutive_transaction_by_sender(sender, on_chain_nonce)
+    }
+
+    fn get_transaction_by_sender_and_nonce(
+        &self,
+        sender: Address,
+        nonce: u64,
+    ) -> Option<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_transaction_by_sender_and_nonce(sender, nonce)
+    }
+
+    fn get_transactions_by_origin(
+        &self,
+        origin: TransactionOrigin,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_transactions_by_origin(origin)
+    }
+
+    fn get_pending_transactions_by_origin(
+        &self,
+        origin: TransactionOrigin,
+    ) -> Vec<Arc<ValidPoolTransaction<Self::Transaction>>> {
+        self.0.get_pending_transactions_by_origin(origin)
+    }
+
+    fn unique_senders(&self) -> HashSet<Address> {
+        self.0.unique_senders()
+    }
+
+    fn get_blob(
+        &self,
+        tx_hash: TxHash,
+    ) -> Result<Option<Arc<BlobTransactionSidecar>>, BlobStoreError> {
+        self.0.get_blob(tx_hash)
+    }
+
+    fn get_all_blobs(
+        &self,
+        tx_hashes: Vec<TxHash>,
+    ) -> Result<Vec<(TxHash, Arc<BlobTransactionSidecar>)>, BlobStoreError> {
+        self.0.get_all_blobs(tx_hashes)
+    }
+
+    fn get_all_blobs_exact(
+        &self,
+        tx_hashes: Vec<TxHash>,
+    ) -> Result<Vec<Arc<BlobTransactionSidecar>>, BlobStoreError> {
+        self.0.get_all_blobs_exact(tx_hashes)
+    }
+
+    fn get_blobs_for_versioned_hashes(
+        &self,
+        versioned_hashes: &[B256],
+    ) -> Result<Vec<Option<BlobAndProofV1>>, BlobStoreError> {
+        self.0.get_blobs_for_versioned_hashes(versioned_hashes)
+    }
+}
+
+impl<V, T, S> TransactionPoolExt for EvmSequencerPool<V, T, S>
+where
+    V: TransactionValidator,
+    <V as TransactionValidator>::Transaction: EthPoolTransaction,
+    T: TransactionOrdering<Transaction = <V as TransactionValidator>::Transaction>,
+    S: BlobStore,
+{
+    fn set_block_info(&self, info: BlockInfo) {
+        self.0.set_block_info(info);
+    }
+
+    fn on_canonical_state_change<B>(&self, update: CanonicalStateUpdate<'_, B>)
+    where
+        B: Block,
+    {
+        self.0.on_canonical_state_change(update);
+    }
+
+    fn update_accounts(&self, accounts: Vec<ChangedAccount>) {
+        self.0.update_accounts(accounts);
+    }
+
+    fn delete_blob(&self, tx: TxHash) {
+        self.0.delete_blob(tx);
+    }
+
+    fn delete_blobs(&self, txs: Vec<TxHash>) {
+        self.0.delete_blobs(txs);
+    }
+
+    fn cleanup_blobs(&self) {
+        self.0.cleanup_blobs();
+    }
+}
+
+impl<V, T, S> Clone for EvmSequencerPool<V, T, S>
+where
+    T: TransactionOrdering,
+{
+    fn clone(&self) -> Self {
+        Self(self.0.clone())
+    }
+}

sz-poc-offsite-2025/node/Cargo.toml

@@ -1,6 +0,0 @@
-[package]
-name = "node"
-edition = { workspace = true }
-
-[dependencies]
-tokio = { workspace = true, features = ["full"] }

sz-poc-offsite-2025/node/src/main.rs

@@ -1,2 +0,0 @@
-#[tokio::main]
-async fn main() {}