Refactor txpool last commit state

Author: andr-dev

monad-eth-txpool/src/pool/mod.rs

@@ -27,7 +27,7 @@ use monad_chain_config::{
     ChainConfig, MockChainConfig,
 };
 use monad_consensus_types::{
-    block::{BlockPolicyError, ProposedExecutionInputs},
+    block::{BlockPolicyError, ConsensusBlockHeader, ProposedExecutionInputs},
     payload::RoundSignature,
 };
 use monad_crypto::certificate_signature::{
@@ -41,7 +41,7 @@ use monad_system_calls::{SystemTransactionGenerator, SYSTEM_SENDER_ETH_ADDRESS};
 use monad_types::{Epoch, NodeId, Round, SeqNum};
 use monad_validator::signature_collection::SignatureCollection;
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
-use tracing::{debug, info, warn};
+use tracing::{debug, error, info, warn};
 
 pub use self::transaction::max_eip2718_encoded_length;
 use self::{pending::PendingTxMap, tracked::TrackedTxMap, transaction::ValidEthTransaction};
@@ -70,6 +70,8 @@ where
     pending: PendingTxMap,
     tracked: TrackedTxMap<ST, SCT, SBT, CCT, CRT>,
 
+    last_commit: Option<ConsensusBlockHeader<ST, SCT, EthExecutionProtocol>>,
+
     chain_id: u64,
     chain_revision: CRT,
     execution_revision: MonadExecutionRevision,
@@ -98,6 +100,8 @@ where
             pending: PendingTxMap::default(),
             tracked: TrackedTxMap::new(soft_tx_expiry, hard_tx_expiry),
 
+            last_commit: None,
+
             chain_id,
             chain_revision,
             execution_revision,
@@ -136,7 +140,7 @@ where
             return;
         }
 
-        let Some(last_commit) = self.tracked.last_commit() else {
+        let Some(last_commit) = self.last_commit.as_ref() else {
             event_tracker.drop_all(txs.into_iter(), EthTxPoolDropReason::PoolNotReady);
             return;
         };
@@ -202,7 +206,7 @@ where
 
             let Some(tx) = self
                 .tracked
-                .try_insert_tx(event_tracker, tx)
+                .try_insert_tx(event_tracker, last_commit, tx)
                 .unwrap_or_else(|tx| {
                     self.pending
                         .try_insert_tx(event_tracker, tx, last_commit_base_fee)
@@ -216,6 +220,7 @@ where
 
         if !self.tracked.try_promote_pending(
             event_tracker,
+            last_commit,
             block_policy,
             state_backend,
             &mut self.pending,
@@ -235,8 +240,14 @@ where
         block_policy: &EthBlockPolicy<ST, SCT, CCT, CRT>,
         state_backend: &SBT,
     ) {
+        let Some(last_commit) = self.last_commit.as_ref() else {
+            warn!("txpool promote_pending called before last committed block set");
+            return;
+        };
+
         if !self.tracked.try_promote_pending(
             event_tracker,
+            last_commit,
             block_policy,
             state_backend,
             &mut self.pending,
@@ -323,21 +334,42 @@ where
             .map(|tx| tx.length() as u64)
             .sum();
 
-        let user_transactions = self.tracked.create_proposal(
-            event_tracker,
-            self.chain_id,
-            proposed_seq_num,
-            base_fee,
-            tx_limit - system_transactions.len(),
-            proposal_gas_limit,
-            proposal_byte_limit - system_txs_size,
-            block_policy,
-            extending_blocks.iter().collect(),
-            state_backend,
-            chain_config,
-            &self.chain_revision,
-            &self.execution_revision,
-        )?;
+        let user_transactions = if let Some(last_commit) = self.last_commit.as_ref() {
+            let last_commit_seq_num = last_commit.seq_num;
+
+            assert!(
+                block_policy.get_last_commit().ge(&last_commit_seq_num),
+                "txpool received block policy with lower committed seq num"
+            );
+
+            if last_commit_seq_num == block_policy.get_last_commit() {
+                self.tracked.create_proposal(
+                    event_tracker,
+                    self.chain_id,
+                    proposed_seq_num,
+                    base_fee,
+                    tx_limit - system_transactions.len(),
+                    proposal_gas_limit,
+                    proposal_byte_limit - system_txs_size,
+                    block_policy,
+                    extending_blocks.iter().collect(),
+                    state_backend,
+                    chain_config,
+                    &self.chain_revision,
+                    &self.execution_revision,
+                )?
+            } else {
+                error!(
+                    block_policy_last_commit = block_policy.get_last_commit().0,
+                    txpool_last_commit = last_commit_seq_num.0,
+                    "last commit update does not match block policy last commit"
+                );
+                Vec::default()
+            }
+        } else {
+            error!("txpool create_proposal called before last committed block set");
+            Vec::default()
+        };
 
         let body = EthBlockBody {
             transactions: system_transactions
@@ -423,13 +455,22 @@ where
         chain_config: &impl ChainConfig<CRT>,
         committed_block: EthValidatedBlock<ST, SCT>,
     ) {
-        let execution_revision = chain_config
-            .get_execution_chain_revision(committed_block.header().execution_inputs.timestamp);
+        {
+            let seqnum = committed_block.get_seq_num();
+            debug!(?seqnum, "txpool updating committed block");
+        }
 
-        self.tracked
-            .update_committed_block(event_tracker, committed_block, &mut self.pending);
+        if let Some(last_commit) = self.last_commit.as_ref() {
+            assert_eq!(
+                committed_block.get_seq_num(),
+                last_commit.seq_num + SeqNum(1),
+                "txpool received out of order committed block"
+            );
+        }
+        self.last_commit = Some(committed_block.header().clone());
 
-        self.tracked.evict_expired_txs(event_tracker);
+        let execution_revision = chain_config
+            .get_execution_chain_revision(committed_block.header().execution_inputs.timestamp);
 
         if self.execution_revision != execution_revision {
             self.execution_revision = execution_revision;
@@ -438,6 +479,14 @@ where
             self.static_validate_all_txs(event_tracker);
         }
 
+        self.tracked.update_committed_nonce_usages(
+            event_tracker,
+            committed_block.nonce_usages,
+            &mut self.pending,
+        );
+
+        self.tracked.evict_expired_txs(event_tracker);
+
         self.update_aggregate_metrics(event_tracker);
     }
 
@@ -447,6 +496,10 @@ where
         chain_config: &impl ChainConfig<CRT>,
         last_delay_committed_blocks: Vec<EthValidatedBlock<ST, SCT>>,
     ) {
+        self.last_commit = last_delay_committed_blocks
+            .last()
+            .map(|block| block.header().clone());
+
         let execution_revision = chain_config.get_execution_chain_revision(
             last_delay_committed_blocks
                 .last()
@@ -455,15 +508,15 @@ where
                 }),
         );
 
-        self.tracked.reset(last_delay_committed_blocks);
-
         if self.execution_revision != execution_revision {
             self.execution_revision = execution_revision;
             info!(execution_revision =? self.execution_revision, "updating execution revision");
 
             self.static_validate_all_txs(event_tracker);
         }
 
+        self.tracked.reset();
+
         self.update_aggregate_metrics(event_tracker);
     }
 
@@ -485,7 +538,7 @@ where
     pub fn get_forwardable_txs<const MIN_SEQNUM_DIFF: u64, const MAX_RETRIES: usize>(
         &mut self,
     ) -> Option<impl Iterator<Item = &TxEnvelope>> {
-        let last_commit = self.tracked.last_commit()?;
+        let last_commit = self.last_commit.as_ref()?;
 
         let last_commit_seq_num = last_commit.seq_num;
         let last_commit_base_fee = last_commit.execution_inputs.base_fee_per_gas;

monad-eth-txpool/src/pool/tracked/mod.rs

@@ -29,7 +29,7 @@ use monad_crypto::certificate_signature::{
     CertificateSignaturePubKey, CertificateSignatureRecoverable,
 };
 use monad_eth_block_policy::{
-    nonce_usage::{NonceUsage, NonceUsageRetrievable},
+    nonce_usage::{NonceUsage, NonceUsageMap},
     EthBlockPolicy, EthBlockPolicyBlockValidator, EthValidatedBlock,
 };
 use monad_eth_txpool_types::{EthTxPoolDropReason, EthTxPoolInternalDropReason};
@@ -70,15 +70,14 @@ where
     SCT: SignatureCollection<NodeIdPubKey = CertificateSignaturePubKey<ST>>,
     SBT: StateBackend<ST, SCT>,
 {
-    last_commit: Option<ConsensusBlockHeader<ST, SCT, EthExecutionProtocol>>,
-    soft_tx_expiry: Duration,
-    hard_tx_expiry: Duration,
-
     // By using IndexMap, we can iterate through the map with Vec-like performance and are able to
     // evict expired txs through the entry API.
     txs: IndexMap<Address, TrackedTxList>,
 
-    _phantom: PhantomData<(SBT, CCT, CRT)>,
+    soft_tx_expiry: Duration,
+    hard_tx_expiry: Duration,
+
+    _phantom: PhantomData<(ST, SCT, SBT, CCT, CRT)>,
 }
 
 impl<ST, SCT, SBT, CCT, CRT> TrackedTxMap<ST, SCT, SBT, CCT, CRT>
@@ -91,20 +90,15 @@ where
 {
     pub fn new(soft_tx_expiry: Duration, hard_tx_expiry: Duration) -> Self {
         Self {
-            last_commit: None,
+            txs: IndexMap::with_capacity(MAX_ADDRESSES),
+
             soft_tx_expiry,
             hard_tx_expiry,
 
-            txs: IndexMap::with_capacity(MAX_ADDRESSES),
-
             _phantom: PhantomData,
         }
     }
 
-    pub fn last_commit(&self) -> Option<&ConsensusBlockHeader<ST, SCT, EthExecutionProtocol>> {
-        self.last_commit.as_ref()
-    }
-
     pub fn is_empty(&self) -> bool {
         self.txs.is_empty()
     }
@@ -131,12 +125,9 @@ where
     pub fn try_insert_tx(
         &mut self,
         event_tracker: &mut EthTxPoolEventTracker<'_>,
+        last_commit: &ConsensusBlockHeader<ST, SCT, EthExecutionProtocol>,
         tx: ValidEthTransaction,
     ) -> Result<Option<&ValidEthTransaction>, ValidEthTransaction> {
-        let Some(last_commit) = self.last_commit.as_ref() else {
-            return Err(tx);
-        };
-
         let Some(tx_list) = self.txs.get_mut(tx.signer_ref()) else {
             return Err(tx);
         };
@@ -165,26 +156,6 @@ where
         chain_revision: &CRT,
         execution_revision: &MonadExecutionRevision,
     ) -> Result<Vec<Recovered<TxEnvelope>>, BlockPolicyError> {
-        let Some(last_commit) = &self.last_commit else {
-            return Ok(Vec::new());
-        };
-        let last_commit_seq_num = last_commit.seq_num;
-
-        assert!(
-            block_policy.get_last_commit().ge(&last_commit_seq_num),
-            "txpool received block policy with lower committed seq num"
-        );
-
-        if last_commit_seq_num != block_policy.get_last_commit() {
-            error!(
-                block_policy_last_commit = block_policy.get_last_commit().0,
-                txpool_last_commit = last_commit_seq_num.0,
-                "last commit update does not match block policy last commit"
-            );
-
-            return Ok(Vec::new());
-        }
-
         let _timer = DropTimer::start(Duration::ZERO, |elapsed| {
             debug!(?elapsed, "txpool create_proposal");
         });
@@ -196,7 +167,7 @@ where
         let sequencer = ProposalSequencer::new(&self.txs, &extending_blocks, base_fee, tx_limit);
         let sequencer_len = sequencer.len();
 
-        let (mut account_balances, state_backend_lookups) = {
+        let (account_balances, state_backend_lookups) = {
             let _timer = DropTimer::start(Duration::ZERO, |elapsed| {
                 debug!(
                     ?elapsed,
@@ -267,6 +238,7 @@ where
     pub fn try_promote_pending(
         &mut self,
         event_tracker: &mut EthTxPoolEventTracker<'_>,
+        last_commit: &ConsensusBlockHeader<ST, SCT, EthExecutionProtocol>,
         block_policy: &EthBlockPolicy<ST, SCT, CCT, CRT>,
         state_backend: &SBT,
         pending: &mut PendingTxMap,
@@ -293,18 +265,6 @@ where
             return true;
         }
 
-        let Some(last_commit) = &self.last_commit else {
-            warn!("txpool attempted to promote pending before first committed block");
-            event_tracker.drop_all(
-                to_insert
-                    .into_values()
-                    .map(PendingTxList::into_map)
-                    .flat_map(BTreeMap::into_values)
-                    .map(ValidEthTransaction::into_raw),
-                EthTxPoolDropReason::Internal(EthTxPoolInternalDropReason::NotReady),
-            );
-            return false;
-        };
         let last_commit_seq_num = last_commit.seq_num;
 
         let addresses = to_insert.len();
@@ -373,29 +333,15 @@ where
         true
     }
 
-    pub fn update_committed_block(
+    pub fn update_committed_nonce_usages(
         &mut self,
         event_tracker: &mut EthTxPoolEventTracker<'_>,
-        committed_block: EthValidatedBlock<ST, SCT>,
+        nonce_usages: NonceUsageMap,
         pending: &mut PendingTxMap,
     ) {
-        {
-            let seqnum = committed_block.get_seq_num();
-            debug!(?seqnum, "txpool updating committed block");
-        }
-
-        if let Some(last_commit) = &self.last_commit {
-            assert_eq!(
-                committed_block.get_seq_num(),
-                last_commit.seq_num + SeqNum(1),
-                "txpool received out of order committed block"
-            );
-        }
-        self.last_commit = Some(committed_block.header().clone());
-
         let mut insertable = MAX_ADDRESSES.saturating_sub(self.txs.len());
 
-        for (address, nonce_usage) in committed_block.get_nonce_usages().into_map() {
+        for (address, nonce_usage) in nonce_usages.into_map() {
             match self.txs.entry(address) {
                 IndexMapEntry::Occupied(tx_list) => {
                     TrackedTxList::update_committed_nonce_usage(event_tracker, tx_list, nonce_usage)
@@ -461,11 +407,8 @@ where
         }
     }
 
-    pub fn reset(&mut self, last_delay_committed_blocks: Vec<EthValidatedBlock<ST, SCT>>) {
+    pub fn reset(&mut self) {
         self.txs.clear();
-        self.last_commit = last_delay_committed_blocks
-            .last()
-            .map(|block| block.header().clone())
     }
 
     pub fn static_validate_all_txs(