refactor(l1): move mempool from Store to Blockchain

cmd/ethrex/ethrex.rs

@@ -261,7 +261,7 @@ async fn main() {
         peer_table.clone(),
         sync_mode,
         cancel_token.clone(),
-        blockchain,
+        blockchain.clone(),
     );
 
     // TODO: Check every module starts properly.
@@ -298,6 +298,7 @@ async fn main() {
                 http_socket_addr,
                 authrpc_socket_addr,
                 store.clone(),
+                blockchain.clone(),
                 jwt_secret_clone,
                 local_p2p_node,
                 local_node_record,
@@ -313,6 +314,7 @@ async fn main() {
                 http_socket_addr,
                 authrpc_socket_addr,
                 store.clone(),
+                blockchain.clone(),
                 jwt_secret_clone,
                 local_p2p_node,
                 local_node_record,
@@ -348,7 +350,7 @@ async fn main() {
                 error!("Cannot run with DEV_MODE if the `l2` feature is enabled.");
                 panic!("Run without the --dev argument.");
             }
-            let l2_proposer = ethrex_l2::start_proposer(store).into_future();
+            let l2_proposer = ethrex_l2::start_proposer(store.clone(), blockchain.clone()).into_future();
             tracker.spawn(l2_proposer);
         } else if #[cfg(feature = "dev")] {
             use ethrex_dev;
@@ -386,6 +388,7 @@ async fn main() {
                 signer,
                 peer_table.clone(),
                 store,
+                blockchain,
             )
             .await.expect("Network starts");
             tracker.spawn(ethrex_p2p::periodically_show_peer_stats(peer_table.clone()));

crates/blockchain/blockchain.rs

@@ -5,15 +5,21 @@ pub mod mempool;
 pub mod payload;
 mod smoke_test;
 
+use constants::MAX_INITCODE_SIZE;
+use error::MempoolError;
 use error::{ChainError, InvalidBlockError};
-use ethrex_common::constants::GAS_PER_BLOB;
+use ethrex_common::constants::{GAS_PER_BLOB, MIN_BASE_FEE_PER_BLOB_GAS};
 use ethrex_common::types::requests::{compute_requests_hash, EncodedRequests, Requests};
+use ethrex_common::types::BlobsBundle;
+use ethrex_common::types::MempoolTransaction;
 use ethrex_common::types::{
     compute_receipts_root, validate_block_header, validate_cancun_header_fields,
     validate_prague_header_fields, validate_pre_cancun_header_fields, Block, BlockHash,
     BlockHeader, BlockNumber, ChainConfig, EIP4844Transaction, Receipt, Transaction,
 };
-use ethrex_common::H256;
+
+use ethrex_common::{Address, H256};
+use mempool::Mempool;
 use std::{ops::Div, time::Instant};
 
 use ethrex_storage::error::StoreError;
@@ -31,20 +37,23 @@ use tracing::{error, info, warn};
 pub struct Blockchain {
     pub vm: EVM,
     pub storage: Store,
+    pub mempool: Mempool,
 }
 
 impl Blockchain {
     pub fn new(evm: EVM, store: Store) -> Self {
         Self {
             vm: evm,
             storage: store,
+            mempool: Mempool::new(),
         }
     }
 
     pub fn default_with_store(store: Store) -> Self {
         Self {
             vm: Default::default(),
             storage: store,
+            mempool: Mempool::new(),
         }
     }
 
@@ -150,6 +159,162 @@ impl Blockchain {
         }
         info!("Added {size} blocks to blockchain");
     }
+
+    /// Add a blob transaction and its blobs bundle to the mempool checking that the transaction is valid
+    #[cfg(feature = "c-kzg")]
+    pub fn add_blob_transaction_to_pool(
+        &self,
+        transaction: EIP4844Transaction,
+        blobs_bundle: BlobsBundle,
+    ) -> Result<H256, MempoolError> {
+        // Validate blobs bundle
+
+        blobs_bundle.validate(&transaction)?;
+
+        let transaction = Transaction::EIP4844Transaction(transaction);
+        let sender = transaction.sender();
+
+        // Validate transaction
+        self.validate_transaction(&transaction, sender)?;
+
+        // Add transaction and blobs bundle to storage
+        let hash = transaction.compute_hash();
+        self.mempool
+            .add_transaction(hash, MempoolTransaction::new(transaction, sender))?;
+        self.mempool.add_blobs_bundle(hash, blobs_bundle)?;
+        Ok(hash)
+    }
+
+    /// Add a transaction to the mempool checking that the transaction is valid
+    pub fn add_transaction_to_pool(&self, transaction: Transaction) -> Result<H256, MempoolError> {
+        // Blob transactions should be submitted via add_blob_transaction along with the corresponding blobs bundle
+        if matches!(transaction, Transaction::EIP4844Transaction(_)) {
+            return Err(MempoolError::BlobTxNoBlobsBundle);
+        }
+        let sender = transaction.sender();
+        // Validate transaction
+        self.validate_transaction(&transaction, sender)?;
+
+        let hash = transaction.compute_hash();
+
+        // Add transaction to storage
+        self.mempool
+            .add_transaction(hash, MempoolTransaction::new(transaction, sender))?;
+
+        Ok(hash)
+    }
+
+    /// Remove a transaction from the mempool
+    pub fn remove_transaction_from_pool(&self, hash: &H256) -> Result<(), StoreError> {
+        self.mempool.remove_transaction(hash)
+    }
+
+    /*
+
+    SOME VALIDATIONS THAT WE COULD INCLUDE
+    Stateless validations
+    1. This transaction is valid on current mempool
+        -> Depends on mempool transaction filtering logic
+    2. Ensure the maxPriorityFeePerGas is high enough to cover the requirement of the calling pool (the minimum to be included in)
+        -> Depends on mempool transaction filtering logic
+    3. Transaction's encoded size is smaller than maximum allowed
+        -> I think that this is not in the spec, but it may be a good idea
+    4. Make sure the transaction is signed properly
+    5. Ensure a Blob Transaction comes with its sidecar (Done! - All blob validations have been moved to `common/types/blobs_bundle.rs`):
+      1. Validate number of BlobHashes is positive (Done!)
+      2. Validate number of BlobHashes is less than the maximum allowed per block,
+         which may be computed as `maxBlobGasPerBlock / blobTxBlobGasPerBlob`
+      3. Ensure number of BlobHashes is equal to:
+        - The number of blobs (Done!)
+        - The number of commitments (Done!)
+        - The number of proofs (Done!)
+      4. Validate that the hashes matches with the commitments, performing a `kzg4844` hash. (Done!)
+      5. Verify the blob proofs with the `kzg4844` (Done!)
+    Stateful validations
+    1. Ensure transaction nonce is higher than the `from` address stored nonce
+    2. Certain pools do not allow for nonce gaps. Ensure a gap is not produced (that is, the transaction nonce is exactly the following of the stored one)
+    3. Ensure the transactor has enough funds to cover transaction cost:
+        - Transaction cost is calculated as `(gas * gasPrice) + (blobGas * blobGasPrice) + value`
+    4. In case of transaction reorg, ensure the transactor has enough funds to cover for transaction replacements without overdrafts.
+    - This is done by comparing the total spent gas of the transactor from all pooled transactions, and accounting for the necessary gas spenditure if any of those transactions is replaced.
+    5. Ensure the transactor is able to add a new transaction. The number of transactions sent by an account may be limited by a certain configured value
+
+    */
+
+    pub fn validate_transaction(
+        &self,
+        tx: &Transaction,
+        sender: Address,
+    ) -> Result<(), MempoolError> {
+        // TODO: Add validations here
+
+        let header_no = self.storage.get_latest_block_number()?;
+        let header = self
+            .storage
+            .get_block_header(header_no)?
+            .ok_or(MempoolError::NoBlockHeaderError)?;
+        let config = self.storage.get_chain_config()?;
+
+        // NOTE: We could add a tx size limit here, but it's not in the actual spec
+
+        // Check init code size
+        if config.is_shanghai_activated(header.timestamp)
+            && tx.is_contract_creation()
+            && tx.data().len() > MAX_INITCODE_SIZE
+        {
+            return Err(MempoolError::TxMaxInitCodeSizeError);
+        }
+
+        // Check gas limit is less than header's gas limit
+        if header.gas_limit < tx.gas_limit() {
+            return Err(MempoolError::TxGasLimitExceededError);
+        }
+
+        // Check priority fee is less or equal than gas fee gap
+        if tx.max_priority_fee().unwrap_or(0) > tx.max_fee_per_gas().unwrap_or(0) {
+            return Err(MempoolError::TxTipAboveFeeCapError);
+        }
+
+        // Check that the gas limit is covers the gas needs for transaction metadata.
+        if tx.gas_limit() < mempool::transaction_intrinsic_gas(tx, &header, &config)? {
+            return Err(MempoolError::TxIntrinsicGasCostAboveLimitError);
+        }
+
+        // Check that the specified blob gas fee is above the minimum value
+        if let Some(fee) = tx.max_fee_per_blob_gas() {
+            // Blob tx fee checks
+            if fee < MIN_BASE_FEE_PER_BLOB_GAS.into() {
+                return Err(MempoolError::TxBlobBaseFeeTooLowError);
+            }
+        };
+
+        let maybe_sender_acc_info = self.storage.get_account_info(header_no, sender)?;
+
+        if let Some(sender_acc_info) = maybe_sender_acc_info {
+            if tx.nonce() < sender_acc_info.nonce {
+                return Err(MempoolError::InvalidNonce);
+            }
+
+            let tx_cost = tx
+                .cost_without_base_fee()
+                .ok_or(MempoolError::InvalidTxGasvalues)?;
+
+            if tx_cost > sender_acc_info.balance {
+                return Err(MempoolError::NotEnoughBalance);
+            }
+        } else {
+            // An account that is not in the database cannot possibly have enough balance to cover the transaction cost
+            return Err(MempoolError::NotEnoughBalance);
+        }
+
+        if let Some(chain_id) = tx.chain_id() {
+            if chain_id != config.chain_id {
+                return Err(MempoolError::InvalidChainId(config.chain_id));
+            }
+        }
+
+        Ok(())
+    }
 }
 
 pub fn validate_requests_hash(