Merge pull request #4232 from chenyukang/yukang-backport-112-rbf-rule-6

Backport: RBF will also replace tx not in Pending

test/src/main.rs

@@ -473,6 +473,7 @@ fn all_specs() -> Vec<Box<dyn Spec>> {
         Box::new(RbfContainInvalidInput),
         Box::new(RbfContainInvalidCells),
         Box::new(RbfRejectReplaceProposed),
+        Box::new(RbfReplaceProposedSuccess),
         Box::new(CompactBlockEmpty),
         Box::new(CompactBlockEmptyParentUnknown),
         Box::new(CompactBlockPrefilled),

test/src/specs/tx_pool/replace.rs

@@ -505,6 +505,7 @@ impl Spec for RbfContainInvalidCells {
 pub struct RbfRejectReplaceProposed;
 
 // RBF Rule #6
+// We removed rule #6, even tx in `Gap` and `Proposed` status can be replaced.
 impl Spec for RbfRejectReplaceProposed {
     fn run(&self, nodes: &mut Vec<Node>) {
         let node0 = &nodes[0];
@@ -557,29 +558,42 @@ impl Spec for RbfRejectReplaceProposed {
             .capacity(capacity_bytes!(70).pack())
             .build();
 
+        let tx1_hash = txs[2].hash();
         let tx2 = clone_tx
             .as_advanced_builder()
             .set_outputs(vec![output2])
             .build();
 
+        // begin to RBF
         let res = node0
             .rpc_client()
             .send_transaction_result(tx2.data().into());
-        assert!(res.is_err(), "tx2 should be rejected");
-        assert!(res
-            .err()
-            .unwrap()
-            .to_string()
-            .contains("all conflict Txs should be in Pending status"));
+        assert!(res.is_ok());
+
+        let old_tx_status = node0.rpc_client().get_transaction(tx1_hash).tx_status;
+        assert_eq!(old_tx_status.status, Status::Rejected);
+        assert!(old_tx_status.reason.unwrap().contains("RBFRejected"));
+
+        let tx2_status = node0.rpc_client().get_transaction(tx2.hash()).tx_status;
+        assert_eq!(tx2_status.status, Status::Pending);
 
-        // when tx1 was confirmed, tx2 should be rejected
         let window_count = node0.consensus().tx_proposal_window().closest();
         node0.mine(window_count);
+        // since old tx is already in BlockAssembler,
+        // tx1 will be committed, even it is not in tx_pool and with `Rejected` status now
         let ret = wait_until(20, || {
             let res = rpc_client0.get_transaction(txs[2].hash());
             res.tx_status.status == Status::Committed
         });
         assert!(ret, "tx1 should be committed");
+        let tx1_status = node0.rpc_client().get_transaction(txs[2].hash()).tx_status;
+        assert_eq!(tx1_status.status, Status::Committed);
+
+        // tx2 will be marked as `Rejected` because callback of `remove_committed_txs` from tx1
+        let tx2_status = node0.rpc_client().get_transaction(tx2.hash()).tx_status;
+        assert_eq!(tx2_status.status, Status::Rejected);
+
+        // the same tx2 can not be sent again
         let res = node0
             .rpc_client()
             .send_transaction_result(tx2.data().into());
@@ -597,3 +611,122 @@ impl Spec for RbfRejectReplaceProposed {
         config.tx_pool.min_rbf_rate = ckb_types::core::FeeRate(1500);
     }
 }
+
+pub struct RbfReplaceProposedSuccess;
+
+// RBF Rule #6
+// We removed rule #6, this spec testing that we can replace tx in `Gap` and `Proposed` successfully.
+impl Spec for RbfReplaceProposedSuccess {
+    fn run(&self, nodes: &mut Vec<Node>) {
+        let node0 = &nodes[0];
+
+        node0.mine_until_out_bootstrap_period();
+
+        // build txs chain
+        let tx0 = node0.new_transaction_spend_tip_cellbase();
+        let mut txs = vec![tx0];
+        let max_count = 5;
+        while txs.len() <= max_count {
+            let parent = txs.last().unwrap();
+            let child = parent
+                .as_advanced_builder()
+                .set_inputs(vec![{
+                    CellInput::new_builder()
+                        .previous_output(OutPoint::new(parent.hash(), 0))
+                        .build()
+                }])
+                .set_outputs(vec![parent.output(0).unwrap()])
+                .build();
+            txs.push(child);
+        }
+        assert_eq!(txs.len(), max_count + 1);
+        // send Tx chain
+        for tx in txs[..=max_count - 1].iter() {
+            let ret = node0.rpc_client().send_transaction_result(tx.data().into());
+            assert!(ret.is_ok());
+        }
+
+        let proposed = node0.mine_with_blocking(|template| template.proposals.len() != max_count);
+        let ret = node0.rpc_client().get_transaction(txs[2].hash());
+        assert!(
+            matches!(ret.tx_status.status, Status::Pending),
+            "tx1 should be pending"
+        );
+
+        node0.mine_with_blocking(|template| template.number.value() != (proposed + 1));
+
+        let rpc_client0 = node0.rpc_client();
+        let ret = wait_until(20, || {
+            let res = rpc_client0.get_transaction(txs[2].hash());
+            res.tx_status.status == Status::Proposed
+        });
+        assert!(ret, "tx1 should be proposed");
+
+        let clone_tx = txs[2].clone();
+        // Set tx2 fee to a higher value
+        let output2 = CellOutputBuilder::default()
+            .capacity(capacity_bytes!(70).pack())
+            .build();
+
+        let tx1_hash = txs[2].hash();
+        let tx2 = clone_tx
+            .as_advanced_builder()
+            .set_outputs(vec![output2])
+            .build();
+
+        // begin to RBF
+        let res = node0
+            .rpc_client()
+            .send_transaction_result(tx2.data().into());
+        assert!(res.is_ok());
+
+        let old_tx_status = node0.rpc_client().get_transaction(tx1_hash).tx_status;
+        assert_eq!(old_tx_status.status, Status::Rejected);
+        assert!(old_tx_status.reason.unwrap().contains("RBFRejected"));
+
+        let tx2_status = node0.rpc_client().get_transaction(tx2.hash()).tx_status;
+        assert_eq!(tx2_status.status, Status::Pending);
+
+        // submit a black block
+        let example = node0.new_block(None, None, None);
+        let blank_block = example
+            .as_advanced_builder()
+            .set_proposals(vec![])
+            .set_transactions(vec![example.transaction(0).unwrap()])
+            .build();
+        node0.submit_block(&blank_block);
+
+        wait_until(10, move || node0.get_tip_block() == blank_block);
+
+        let window_count = node0.consensus().tx_proposal_window().closest();
+        node0.mine(window_count);
+
+        let ret = wait_until(20, || {
+            let res = rpc_client0.get_transaction(tx2.hash());
+            res.tx_status.status == Status::Proposed
+        });
+        assert!(ret, "tx2 should be proposed");
+        let tx1_status = node0.rpc_client().get_transaction(txs[2].hash()).tx_status;
+        assert_eq!(tx1_status.status, Status::Rejected);
+
+        let window_count = node0.consensus().tx_proposal_window().closest();
+        node0.mine(window_count);
+        // since old tx is already in BlockAssembler,
+        // tx1 will be committed, even it is not in tx_pool and with `Rejected` status now
+        let ret = wait_until(20, || {
+            let res = rpc_client0.get_transaction(tx2.hash());
+            res.tx_status.status == Status::Committed
+        });
+        assert!(ret, "tx2 should be committed");
+
+        // the same tx2 can not be sent again
+        let res = node0
+            .rpc_client()
+            .send_transaction_result(tx2.data().into());
+        assert!(res.is_err(), "tx2 should be rejected");
+    }
+
+    fn modify_app_config(&self, config: &mut ckb_app_config::CKBAppConfig) {
+        config.tx_pool.min_rbf_rate = ckb_types::core::FeeRate(1500);
+    }
+}

tx-pool/src/pool.rs

@@ -234,6 +234,11 @@ impl TxPool {
         {
             let conflicts = self.pool_map.resolve_conflict(tx);
             for (entry, reject) in conflicts {
+                debug!(
+                    "removed {} for commited: {}",
+                    entry.transaction().hash(),
+                    tx.hash()
+                );
                 callbacks.call_reject(self, &entry, reject);
             }
         }
@@ -525,6 +530,7 @@ impl TxPool {
         fee: Capacity,
         tx_size: usize,
     ) -> Result<(), Reject> {
+        // Rule #1, the node has enabled RBF, which is checked by caller
         assert!(self.enable_rbf());
         assert!(!conflict_ids.is_empty());
 
@@ -615,19 +621,6 @@ impl TxPool {
                     ));
                 }
             }
-
-            let mut entries_status = entries.iter().map(|e| e.status).collect::<Vec<_>>();
-            entries_status.push(conflict.status);
-            // Rule #6, all conflict Txs should be in `Pending` or `Gap` status
-            if entries_status
-                .iter()
-                .any(|s| ![Status::Pending, Status::Gap].contains(s))
-            {
-                // Here we only refer to `Pending` status, since `Gap` is an internal status
-                return Err(Reject::RBFRejected(
-                    "all conflict Txs should be in Pending status".to_string(),
-                ));
-            }
         }
 
         Ok(())