Txpool will evict tx when resolve error in block assembler

test/src/main.rs

@@ -462,6 +462,7 @@ fn all_specs() -> Vec<Box<dyn Spec>> {
         Box::new(RelayWithWrongTx::new()),
         Box::new(TxsRelayOrder),
         Box::new(SendTxChain),
+        Box::new(SendTxChainRevOrder),
         Box::new(DifferentTxsWithSameInputWithOutRBF),
         Box::new(RbfEnable),
         Box::new(RbfBasic),

test/src/specs/tx_pool/send_tx_chain.rs

@@ -1,4 +1,8 @@
+use crate::specs::tx_pool::utils::prepare_tx_family;
+use crate::utils::blank;
+use crate::utils::propose;
 use crate::{Node, Spec};
+use ckb_jsonrpc_types::TxStatus;
 use ckb_logger::info;
 use ckb_types::core::FeeRate;
 use ckb_types::{
@@ -93,3 +97,41 @@ impl Spec for SendTxChain {
         config.tx_pool.max_ancestors_count = MAX_ANCESTORS_COUNT;
     }
 }
+
+pub struct SendTxChainRevOrder;
+
+impl Spec for SendTxChainRevOrder {
+    crate::setup!(num_nodes: 1);
+
+    // Case: Check txpool will evict tx when tx check failed during block_assembler
+    //       avoid to stay for a long time in the pool
+    fn run(&self, nodes: &mut Vec<Node>) {
+        let node_a = &nodes[0];
+        let window = node_a.consensus().tx_proposal_window();
+
+        node_a.mine_until_out_bootstrap_period();
+        let family = prepare_tx_family(node_a);
+
+        node_a.submit_transaction(family.a());
+        node_a.submit_transaction(family.b());
+
+        let tx_pool_info = node_a.rpc_client().tx_pool_info();
+        assert_eq!(tx_pool_info.pending.value(), 2);
+
+        // send the child tx firstly
+        node_a.submit_block(&propose(node_a, &[family.b()]));
+
+        assert!(node_a.get_transaction(family.a().hash()) == TxStatus::pending());
+        assert!(node_a.get_transaction(family.b().hash()) == TxStatus::pending());
+        (0..window.closest()).for_each(|_| {
+            node_a.submit_block(&blank(node_a));
+        });
+
+        assert!(node_a.get_transaction(family.a().hash()) == TxStatus::pending());
+
+        // tx_b is removed by txpool, don't stay in the pool
+        assert!(node_a.get_transaction(family.b().hash()) == TxStatus::unknown());
+        let tx_pool_info = node_a.rpc_client().tx_pool_info();
+        assert_eq!(tx_pool_info.pending.value(), 1);
+    }
+}

tx-pool/src/block_assembler/mod.rs

@@ -122,7 +122,7 @@ impl BlockAssembler {
         let basic_block_size =
             Self::basic_block_size(cellbase.data(), &[], iter::empty(), extension.clone());
 
-        let (dao, _checked_txs) =
+        let (dao, _checked_txs, _failed_txs) =
             Self::calc_dao(&snapshot, &current_epoch, cellbase.clone(), vec![])
                 .expect("calc_dao for BlockAssembler initial");
 
@@ -197,12 +197,18 @@ impl BlockAssembler {
         };
 
         let proposals_size = proposals.len() * ProposalShortId::serialized_size();
-        let (dao, checked_txs) = Self::calc_dao(
+        let (dao, checked_txs, failed_txs) = Self::calc_dao(
             &current.snapshot,
             &current.epoch,
             current_template.cellbase.clone(),
             txs,
         )?;
+        if !failed_txs.is_empty() {
+            let mut tx_pool_writer = tx_pool.write().await;
+            for id in failed_txs {
+                tx_pool_writer.remove_tx(&id);
+            }
+        }
 
         let txs_size = checked_txs.iter().map(|tx| tx.size).sum();
         let total_size = basic_size + txs_size;
@@ -251,7 +257,7 @@ impl BlockAssembler {
         let basic_block_size =
             Self::basic_block_size(cellbase.data(), &uncles, iter::empty(), extension.clone());
 
-        let (dao, _checked_txs) =
+        let (dao, _checked_txs, _failed_txs) =
             Self::calc_dao(&snapshot, &current_epoch, cellbase.clone(), vec![])?;
 
         builder
@@ -407,7 +413,7 @@ impl BlockAssembler {
             txs
         };
 
-        if let Ok((dao, checked_txs)) = Self::calc_dao(
+        if let Ok((dao, checked_txs, _failed_txs)) = Self::calc_dao(
             &current.snapshot,
             &current.epoch,
             current_template.cellbase.clone(),
@@ -577,12 +583,13 @@ impl BlockAssembler {
         current_epoch: &EpochExt,
         cellbase: TransactionView,
         entries: Vec<TxEntry>,
-    ) -> Result<(Byte32, Vec<TxEntry>), AnyError> {
+    ) -> Result<(Byte32, Vec<TxEntry>, Vec<ProposalShortId>), AnyError> {
         let tip_header = snapshot.tip_header();
         let consensus = snapshot.consensus();
         let mut seen_inputs = HashSet::new();
         let mut transactions_checker = TransactionsChecker::new(iter::once(&cellbase));
 
+        let mut checked_failed_txs = vec![];
         let checked_entries: Vec<_> = block_in_place(|| {
             entries
                 .into_iter()
@@ -602,6 +609,7 @@ impl BlockAssembler {
                             entry.transaction().hash(),
                             err
                         );
+                        checked_failed_txs.push(entry.proposal_short_id());
                         None
                     } else {
                         transactions_checker.insert(entry.transaction());
@@ -620,7 +628,7 @@ impl BlockAssembler {
         let dao = DaoCalculator::new(consensus, &snapshot.borrow_as_data_loader())
             .dao_field_with_current_epoch(entries_iter, tip_header, current_epoch)?;
 
-        Ok((dao, checked_entries))
+        Ok((dao, checked_entries, checked_failed_txs))
     }
 
     pub(crate) async fn notify(&self) {

tx-pool/src/process.rs

@@ -106,8 +106,7 @@ impl TxPoolService {
     ) -> (Result<(), Reject>, Arc<Snapshot>) {
         let (ret, snapshot) = self
             .with_tx_pool_write_lock(move |tx_pool, snapshot| {
-                // if snapshot changed by context switch
-                // we need redo time_relative verify
+                // if snapshot changed by context switch we need redo time_relative verify
                 let tip_hash = snapshot.tip_hash();
                 if pre_resolve_tip != tip_hash {
                     debug!(