Merge pull request #4511 from nervosnetwork/zhangsoledad/tx-pool

Modify the record scope of tx-pool reject record and fix rule for orphan tx.

shared/src/shared_builder.rs

@@ -416,10 +416,7 @@ fn register_tx_pool_callback(tx_pool_builder: &mut TxPoolServiceBuilder, notify:
         move |tx_pool: &mut TxPool, entry: &TxEntry, reject: Reject| {
             let tx_hash = entry.transaction().hash();
             // record recent reject
-            if matches!(
-                reject,
-                Reject::Resolve(..) | Reject::RBFRejected(..) | Reject::Invalidated(..)
-            ) {
+            if reject.should_recorded() {
                 if let Some(ref mut recent_reject) = tx_pool.recent_reject {
                     if let Err(e) = recent_reject.put(&tx_hash, reject.clone()) {
                         error!("record recent_reject failed {} {} {}", tx_hash, reject, e);

sync/src/relayer/mod.rs

@@ -752,6 +752,17 @@ impl Relayer {
                     TxVerificationResult::Reject { tx_hash } => {
                         self.shared.state().remove_from_known_txs(&tx_hash);
                     }
+                    TxVerificationResult::UnknownParents { peer, parents } => {
+                        let tx_hashes: Vec<_> = {
+                            let mut tx_filter = self.shared.state().tx_filter();
+                            tx_filter.remove_expired();
+                            parents
+                                .into_iter()
+                                .filter(|tx_hash| !tx_filter.contains(tx_hash))
+                                .collect()
+                        };
+                        self.shared.state().add_ask_for_txs(peer, tx_hashes);
+                    }
                 }
             }
         }

test/src/main.rs

@@ -431,6 +431,7 @@ fn all_specs() -> Vec<Box<dyn Spec>> {
         Box::new(TxPoolOrphanNormal),
         Box::new(TxPoolOrphanReverse),
         Box::new(TxPoolOrphanUnordered),
+        Box::new(TxPoolOrphanPartialInputUnknown),
         Box::new(TxPoolOrphanDoubleSpend),
         Box::new(OrphanTxRejected),
         Box::new(GetRawTxPool),

test/src/specs/tx_pool/orphan_tx.rs

@@ -1,10 +1,14 @@
-use crate::util::transaction::relay_tx;
+use crate::util::transaction::{relay_tx, send_tx};
 use crate::utils::wait_until;
 use crate::{Net, Node, Spec};
 use ckb_jsonrpc_types::Status;
 use ckb_network::SupportProtocols;
-use ckb_types::core::{capacity_bytes, Capacity, TransactionBuilder, TransactionView};
 use ckb_types::packed::CellOutputBuilder;
+use ckb_types::{
+    bytes::Bytes,
+    core::{capacity_bytes, Capacity, TransactionBuilder, TransactionView},
+    packed,
+};
 use ckb_types::{
     packed::{CellInput, OutPoint},
     prelude::*,
@@ -193,6 +197,30 @@ fn run_replay_tx(
     })
 }
 
+fn run_send_tx(
+    net: &Net,
+    node0: &Node,
+    tx: TransactionView,
+    orphan_tx_cnt: u64,
+    pending_cnt: u64,
+) -> bool {
+    send_tx(net, node0, tx, ALWAYS_SUCCESS_SCRIPT_CYCLE);
+
+    wait_until(5, || {
+        let tx_pool_info = node0.get_tip_tx_pool_info();
+        tx_pool_info.orphan.value() == orphan_tx_cnt && tx_pool_info.pending.value() == pending_cnt
+    })
+}
+
+fn should_receive_get_relay_transactions(net: &Net, node0: &Node, assert_message: &str) {
+    let ret = net.should_receive(node0, |data: &Bytes| {
+        packed::RelayMessage::from_slice(data)
+            .map(|message| message.to_enum().item_name() == packed::GetRelayTransactions::NAME)
+            .unwrap_or(false)
+    });
+    assert!(ret, "{}", assert_message);
+}
+
 pub struct TxPoolOrphanNormal;
 impl Spec for TxPoolOrphanNormal {
     fn run(&self, nodes: &mut Vec<Node>) {
@@ -236,23 +264,20 @@ impl Spec for TxPoolOrphanReverse {
             run_replay_tx(&net, node0, final_tx, 1, 0),
             "expect final_tx is in orphan pool"
         );
+        should_receive_get_relay_transactions(&net, node0, "node should ask for tx11 tx12 tx13");
+
+        assert!(run_send_tx(&net, node0, tx13, 2, 0), "tx13 in orphan pool");
+        should_receive_get_relay_transactions(&net, node0, "node should ask for tx1");
 
         assert!(
-            run_replay_tx(&net, node0, tx13, 2, 0),
-            "tx13 in orphan pool"
-        );
-        assert!(
-            run_replay_tx(&net, node0, tx12, 3, 0),
+            run_send_tx(&net, node0, tx12, 3, 0),
             "tx12 is in orphan pool"
         );
-        assert!(run_replay_tx(&net, node0, tx11, 4, 0), "tx11 is in orphan");
-
-        assert!(run_replay_tx(&net, node0, tx1, 5, 0), "tx1 is in orphan");
+        assert!(run_send_tx(&net, node0, tx11, 4, 0), "tx11 is in orphan");
+        assert!(run_send_tx(&net, node0, tx1, 5, 0), "tx1 is in orphan");
 
-        assert!(
-            run_replay_tx(&net, node0, parent, 0, 6),
-            "all is in pending"
-        );
+        should_receive_get_relay_transactions(&net, node0, "node should ask for parent");
+        assert!(run_send_tx(&net, node0, parent, 0, 6), "all is in pending");
     }
 }
 
@@ -267,13 +292,14 @@ impl Spec for TxPoolOrphanUnordered {
             "expect final_tx is in orphan pool"
         );
 
-        assert!(
-            run_replay_tx(&net, node0, tx11, 2, 0),
-            "tx11 in orphan pool"
-        );
+        should_receive_get_relay_transactions(&net, node0, "node should ask for tx11 tx12 tx13");
+
+        assert!(run_send_tx(&net, node0, tx11, 2, 0), "tx11 in orphan pool");
+        should_receive_get_relay_transactions(&net, node0, "node should ask for tx1");
+
         let tx12_clone = tx12.clone();
         assert!(
-            run_replay_tx(&net, node0, tx12, 3, 0),
+            run_send_tx(&net, node0, tx12, 3, 0),
             "tx12 is in orphan pool"
         );
 
@@ -292,12 +318,47 @@ impl Spec for TxPoolOrphanUnordered {
             "parent is sent, should be in pending without change orphan pool"
         );
         assert!(
-            run_replay_tx(&net, node0, tx1, 1, 4),
+            run_send_tx(&net, node0, tx1, 1, 4),
             "tx1 is sent, orphan pool only contains final_tx"
         );
 
         assert!(
-            run_replay_tx(&net, node0, tx13, 0, 6),
+            run_send_tx(&net, node0, tx13, 0, 6),
+            "tx13 is sent, orphan pool is empty"
+        );
+    }
+}
+
+pub struct TxPoolOrphanPartialInputUnknown;
+impl Spec for TxPoolOrphanPartialInputUnknown {
+    fn run(&self, nodes: &mut Vec<Node>) {
+        let node0 = &nodes[0];
+        let (net, (parent, tx1, tx11, tx12, tx13, final_tx)) = build_tx_chain(node0);
+
+        assert!(
+            run_replay_tx(&net, node0, parent, 0, 1),
+            "parent sended expect nothing in orphan pool"
+        );
+        assert!(
+            run_replay_tx(&net, node0, tx1, 0, 2),
+            "tx1 is sent expect nothing in orphan pool"
+        );
+        assert!(
+            run_replay_tx(&net, node0, tx11, 0, 3),
+            "tx11 is sent expect nothing in orphan pool"
+        );
+        assert!(
+            run_replay_tx(&net, node0, tx12, 0, 4),
+            "tx12 is sent expect nothing in orphan pool"
+        );
+        assert!(
+            run_replay_tx(&net, node0, final_tx, 1, 4),
+            "expect final_tx is in orphan pool"
+        );
+
+        should_receive_get_relay_transactions(&net, node0, "node should ask for tx13");
+        assert!(
+            run_send_tx(&net, node0, tx13, 0, 6),
             "tx13 is sent, orphan pool is empty"
         );
     }

test/src/util/transaction.rs

@@ -47,6 +47,26 @@ pub fn always_success_transactions_with_rand_data(
         .build()
 }
 
+pub fn send_tx(net: &Net, node: &Node, tx: TransactionView, cycles: u64) {
+    let relay_tx = packed::RelayTransaction::new_builder()
+        .cycles(cycles.pack())
+        .transaction(tx.data())
+        .build();
+
+    let tx_msg = packed::RelayMessage::new_builder()
+        .set(
+            packed::RelayTransactions::new_builder()
+                .transactions(
+                    packed::RelayTransactionVec::new_builder()
+                        .set(vec![relay_tx])
+                        .build(),
+                )
+                .build(),
+        )
+        .build();
+    net.send(node, SupportProtocols::RelayV3, tx_msg.as_bytes());
+}
+
 pub fn relay_tx(net: &Net, node: &Node, tx: TransactionView, cycles: u64) {
     let tx_hashes_msg = packed::RelayMessage::new_builder()
         .set(

tx-pool/src/component/orphan.rs

@@ -151,6 +151,7 @@ impl OrphanPool {
                 .insert(tx.proposal_short_id());
         }
 
+        // DoS prevention: do not allow OrphanPool to grow unbounded
         self.limit_size()
     }
 

tx-pool/src/process.rs

@@ -18,7 +18,6 @@ use ckb_logger::{debug, error, info, log_enabled_target, trace_target};
 use ckb_network::PeerIndex;
 use ckb_snapshot::Snapshot;
 use ckb_store::data_loader_wrapper::AsDataLoader;
-use ckb_store::ChainStore;
 use ckb_types::core::error::OutPointError;
 use ckb_types::{
     core::{cell::ResolvedTransaction, BlockView, Capacity, Cycle, HeaderView, TransactionView},
@@ -459,7 +458,10 @@ impl TxPoolService {
         match remote {
             Some((declared_cycle, peer)) => match ret {
                 Ok(_) => {
-                    debug!("after_process remote send_result_to_relayer {}", tx_hash);
+                    debug!(
+                        "after_process remote send_result_to_relayer {} {}",
+                        tx_hash, peer
+                    );
                     self.send_result_to_relayer(TxVerificationResult::Ok {
                         original_peer: Some(peer),
                         with_vm_2023,
@@ -468,8 +470,15 @@ impl TxPoolService {
                     self.process_orphan_tx(&tx).await;
                 }
                 Err(reject) => {
-                    debug!("after_process {} remote reject: {} ", tx_hash, reject);
-                    if is_missing_input(reject) && all_inputs_is_unknown(snapshot, &tx) {
+                    debug!(
+                        "after_process {} {} remote reject: {} ",
+                        tx_hash, peer, reject
+                    );
+                    if is_missing_input(reject) {
+                        self.send_result_to_relayer(TxVerificationResult::UnknownParents {
+                            peer,
+                            parents: tx.unique_parents(),
+                        });
                         self.add_orphan(tx, peer, declared_cycle).await;
                     } else {
                         if reject.is_malformed_tx() {
@@ -480,13 +489,7 @@ impl TxPoolService {
                                 tx_hash: tx_hash.clone(),
                             });
                         }
-
-                        if matches!(
-                            reject,
-                            Reject::Resolve(..)
-                                | Reject::Verification(..)
-                                | Reject::RBFRejected(..)
-                        ) {
+                        if reject.should_recorded() {
                             self.put_recent_reject(&tx_hash, reject).await;
                         }
                     }
@@ -514,12 +517,7 @@ impl TxPoolService {
                     }
                     Err(reject) => {
                         debug!("after_process {} reject: {} ", tx_hash, reject);
-                        if matches!(
-                            reject,
-                            Reject::Resolve(..)
-                                | Reject::Verification(..)
-                                | Reject::RBFRejected(..)
-                        ) {
+                        if reject.should_recorded() {
                             self.put_recent_reject(&tx_hash, reject).await;
                         }
                     }
@@ -628,12 +626,7 @@ impl TxPoolService {
                                         tx_hash: orphan.tx.hash(),
                                     });
                                 }
-                                if matches!(
-                                    reject,
-                                    Reject::Resolve(..)
-                                        | Reject::Verification(..)
-                                        | Reject::RBFRejected(..)
-                                ) {
+                                if reject.should_recorded() {
                                     self.put_recent_reject(&orphan.tx.hash(), &reject).await;
                                 }
                             }
@@ -1259,8 +1252,3 @@ fn _update_tx_pool_for_reorg(
     // Remove transactions from the pool until its size <= size_limit.
     let _ = tx_pool.limit_size(callbacks, None);
 }
-
-pub fn all_inputs_is_unknown(snapshot: &Snapshot, tx: &TransactionView) -> bool {
-    !tx.input_pts_iter()
-        .any(|pt| snapshot.transaction_exists(&pt.tx_hash()))
-}

tx-pool/src/service.rs

@@ -295,7 +295,8 @@ impl TxPoolController {
         send_message!(self, GetTransactionWithStatus, hash)
     }
 
-    /// Return txs for network
+    /// Mainly used for compact block reconstruction and block proposal pre-broadcasting
+    /// Orphan/conflicted/etc transactions that are returned for compact block reconstruction.
     pub fn fetch_txs(
         &self,
         short_ids: HashSet<ProposalShortId>,
@@ -304,6 +305,7 @@ impl TxPoolController {
     }
 
     /// Return txs with cycles
+    /// Mainly for relay transactions
     pub fn fetch_txs_with_cycles(
         &self,
         short_ids: HashSet<ProposalShortId>,
@@ -677,6 +679,13 @@ pub enum TxVerificationResult {
         /// transaction hash
         tx_hash: Byte32,
     },
+    /// tx parent is unknown
+    UnknownParents {
+        /// original peer
+        peer: PeerIndex,
+        /// parents hashes
+        parents: HashSet<Byte32>,
+    },
     /// tx is rejected
     Reject {
         /// transaction hash
@@ -852,11 +861,13 @@ async fn process(mut service: TxPoolService, message: Message) {
             arguments: short_ids,
         }) => {
             let tx_pool = service.tx_pool.read().await;
+            let orphan = service.orphan.read().await;
             let txs = short_ids
                 .into_iter()
                 .filter_map(|short_id| {
                     tx_pool
                         .get_tx_from_pool_or_store(&short_id)
+                        .or_else(|| orphan.get(&short_id).map(|entry| &entry.tx).cloned())
                         .map(|tx| (short_id, tx))
                 })
                 .collect();

util/types/src/core/tx_pool.rs

@@ -95,6 +95,11 @@ impl Reject {
         }
     }
 
+    /// Returns true if the reject should be recorded.
+    pub fn should_recorded(&self) -> bool {
+        !matches!(self, Reject::Duplicated(..))
+    }
+
     /// Returns true if tx can be resubmitted, allowing relay
     /// * Declared wrong cycles should allow relay with the correct cycles
     /// * Reject but is not malformed and the fee rate reached the threshold,

util/types/src/core/views.rs

@@ -422,6 +422,13 @@ impl TransactionView {
     pub fn proposal_short_id(&self) -> packed::ProposalShortId {
         packed::ProposalShortId::from_tx_hash(&self.hash())
     }
+
+    /// return deduplicate parent tx_hashes
+    pub fn unique_parents(&self) -> HashSet<packed::Byte32> {
+        self.input_pts_iter()
+            .map(|outpoint| outpoint.tx_hash())
+            .collect()
+    }
 }
 
 impl ExtraHashView {