feat(mempool): integrate trust metric (#245)

* add trust metric in mempool * change(mempool): remove outdated FIXME * fix(mempool): no peer id in context for trust metric feedback * fix(mempool): rebase master * ci: force travis pull Co-authored-by: wancencen <[email protected]>
nervosnetwork · May 6, 2020 · 49474fd · 49474fd
1 parent 3dd6bc1
commit 49474fd
Show file tree

Hide file tree

Showing 6 changed files with 80 additions and 7 deletions.
diff --git a/core/mempool/Cargo.toml b/core/mempool/Cargo.toml
@@ -12,6 +12,7 @@ protocol = { path = "../../protocol", package = "muta-protocol" }
 common-crypto = { path = "../../common/crypto" }
 core-network = { path = "../network" }
 
+
 futures = { version = "0.3", features = [ "async-await" ] }
 crossbeam-queue = "0.2"
 derive_more = "0.99"

diff --git a/core/mempool/src/adapter/message.rs b/core/mempool/src/adapter/message.rs
@@ -65,7 +65,6 @@ where
             log::error!("[core_mempool] mempool batch insert error");
         }
 
-        // FIXME
         TrustFeedback::Neutral
     }
 }
@@ -122,7 +121,6 @@ where
             .unwrap_or_else(move |err| log::warn!("[core_mempool] push txs {}", err))
             .await;
 
-        // FIXME
         TrustFeedback::Neutral
     }
 }
diff --git a/core/mempool/src/adapter/mod.rs b/core/mempool/src/adapter/mod.rs
@@ -1,3 +1,5 @@
+use super::TxContext;
+
 pub mod message;
 
 use std::{
@@ -24,7 +26,7 @@ use log::{debug, error};
 use common_crypto::Crypto;
 use protocol::{
     fixed_codec::FixedCodec,
-    traits::{Context, Gossip, MemPoolAdapter, Priority, Rpc, Storage},
+    traits::{Context, Gossip, MemPoolAdapter, PeerTrust, Priority, Rpc, Storage, TrustFeedback},
     types::{Hash, SignedTransaction},
     ProtocolError, ProtocolErrorKind, ProtocolResult,
 };
@@ -147,7 +149,7 @@ pub struct DefaultMemPoolAdapter<C, N, S> {
 impl<C, N, S> DefaultMemPoolAdapter<C, N, S>
 where
     C: Crypto,
-    N: Rpc + Gossip + Clone + Unpin + 'static,
+    N: Rpc + PeerTrust + Gossip + Clone + Unpin + 'static,
     S: Storage,
 {
     pub fn new(
@@ -193,7 +195,7 @@ where
 impl<C, N, S> MemPoolAdapter for DefaultMemPoolAdapter<C, N, S>
 where
     C: Crypto + Send + Sync + 'static,
-    N: Rpc + Gossip + Clone + Unpin + 'static,
+    N: Rpc + PeerTrust + Gossip + Clone + Unpin + 'static,
     S: Storage + 'static,
 {
     async fn pull_txs(
@@ -227,13 +229,25 @@ where
         Ok(())
     }
 
-    async fn check_signature(&self, _ctx: Context, tx: SignedTransaction) -> ProtocolResult<()> {
+    async fn check_signature(&self, ctx: Context, tx: SignedTransaction) -> ProtocolResult<()> {
+        let network = self.network.clone();
+
         let blocking_res = tokio::task::spawn_blocking(move || {
             // Verify transaction hash
             let fixed_bytes = tx.raw.encode_fixed()?;
             let tx_hash = Hash::digest(fixed_bytes);
 
             if tx_hash != tx.tx_hash {
+                if ctx.is_network_origin_txs() {
+                    network.report(
+                        ctx,
+                        TrustFeedback::Worse(format!(
+                            "Mempool wrong tx_hash of tx {:?}",
+                            tx.tx_hash
+                        )),
+                    );
+                }
+
                 let wrong_hash = MemPoolError::CheckHash {
                     expect: tx.tx_hash,
                     actual: tx_hash,
@@ -247,6 +261,16 @@ where
             let sig = tx.signature.as_ref();
 
             C::verify_signature(hash.as_ref(), sig, pub_key).map_err(|_| {
+                if ctx.is_network_origin_txs() {
+                    network.report(
+                        ctx,
+                        TrustFeedback::Worse(format!(
+                            "Mempool wrong signature of tx {:?}",
+                            tx.tx_hash
+                        )),
+                    );
+                }
+
                 MemPoolError::CheckSig {
                     tx_hash: tx.tx_hash,
                 }
@@ -267,14 +291,23 @@ where
     // TODO: Verify Fee?
     // TODO: Verify Nonce?
     // TODO: Cycle limit?
-    async fn check_transaction(&self, _ctx: Context, stx: SignedTransaction) -> ProtocolResult<()> {
+    async fn check_transaction(&self, ctx: Context, stx: SignedTransaction) -> ProtocolResult<()> {
         let fixed_bytes = stx.raw.encode_fixed()?;
         let size = fixed_bytes.len() as u64;
         let tx_hash = stx.tx_hash.clone();
 
         // check tx size
         let max_tx_size = self.max_tx_size.load(Ordering::SeqCst);
         if size > max_tx_size {
+            if ctx.is_network_origin_txs() {
+                self.network.report(
+                    ctx,
+                    TrustFeedback::Bad(format!(
+                        "Mempool exceed size limit of tx {:?}",
+                        stx.tx_hash
+                    )),
+                );
+            }
             return Err(MemPoolError::ExceedSizeLimit {
                 tx_hash,
                 max_tx_size,
@@ -287,6 +320,15 @@ where
         let cycles_limit_config = self.cycles_limit.load(Ordering::SeqCst);
         let cycles_limit_tx = stx.raw.cycles_limit;
         if cycles_limit_tx > cycles_limit_config {
+            if ctx.is_network_origin_txs() {
+                self.network.report(
+                    ctx,
+                    TrustFeedback::Bad(format!(
+                        "Mempool exceed cycle limit of tx {:?}",
+                        stx.tx_hash
+                    )),
+                );
+            }
             return Err(MemPoolError::ExceedCyclesLimit {
                 tx_hash,
                 cycles_limit_tx,
@@ -298,6 +340,12 @@ where
         // Verify chain id
         let latest_block = self.storage.get_latest_block().await?;
         if latest_block.header.chain_id != stx.raw.chain_id {
+            if ctx.is_network_origin_txs() {
+                self.network.report(
+                    ctx,
+                    TrustFeedback::Worse(format!("Mempool wrong chain of tx {:?}", stx.tx_hash)),
+                );
+            }
             let wrong_chain_id = MemPoolError::WrongChain {
                 tx_hash: stx.tx_hash,
             };
@@ -310,6 +358,12 @@ where
         let timeout_gap = self.timeout_gap.load(Ordering::SeqCst);
 
         if stx.raw.timeout > latest_height + timeout_gap {
+            if ctx.is_network_origin_txs() {
+                self.network.report(
+                    ctx,
+                    TrustFeedback::Bad(format!("Mempool invalid timeout of tx {:?}", stx.tx_hash)),
+                );
+            }
             let invalid_timeout = MemPoolError::InvalidTimeout {
                 tx_hash: stx.tx_hash,
             };
@@ -318,6 +372,12 @@ where
         }
 
         if stx.raw.timeout < latest_height {
+            if ctx.is_network_origin_txs() {
+                self.network.report(
+                    ctx,
+                    TrustFeedback::Bad(format!("Mempool timeout of tx {:?}", stx.tx_hash)),
+                );
+            }
             let timeout = MemPoolError::Timeout {
                 tx_hash: stx.tx_hash,
                 timeout: stx.raw.timeout,
@@ -348,6 +408,12 @@ where
         Ok(height)
     }
 
+    fn report_good(&self, ctx: Context) {
+        if ctx.is_network_origin_txs() {
+            self.network.report(ctx, TrustFeedback::Good);
+        }
+    }
+
     fn set_args(&self, timeout_gap: u64, cycles_limit: u64, max_tx_size: u64) {
         self.timeout_gap.store(timeout_gap, Ordering::Relaxed);
         self.cycles_limit.store(cycles_limit, Ordering::Relaxed);

diff --git a/core/mempool/src/lib.rs b/core/mempool/src/lib.rs
@@ -117,6 +117,8 @@ where
 
         if !ctx.is_network_origin_txs() {
             self.adapter.broadcast_tx(ctx, tx).await?;
+        } else {
+            self.adapter.report_good(ctx);
         }
 
         Ok(())
@@ -265,6 +267,8 @@ where
                     .insert(signed_tx.tx_hash.clone(), signed_tx)
                     .await;
             }
+
+            self.adapter.report_good(ctx);
         }
 
         Ok(())

diff --git a/core/mempool/src/tests/mod.rs b/core/mempool/src/tests/mod.rs
@@ -81,6 +81,8 @@ impl MemPoolAdapter for HashMemPoolAdapter {
         Ok(CURRENT_HEIGHT)
     }
 
+    fn report_good(&self, _ctx: Context) {}
+
     fn set_args(&self, _timeout_gap: u64, _cycles_limit: u64, _max_tx_size: u64) {}
 }
 

diff --git a/protocol/src/traits/mempool.rs b/protocol/src/traits/mempool.rs
@@ -74,5 +74,7 @@ pub trait MemPoolAdapter: Send + Sync {
 
     async fn get_latest_height(&self, ctx: Context) -> ProtocolResult<u64>;
 
+    fn report_good(&self, ctx: Context);
+
     fn set_args(&self, timeout_gap: u64, cycles_limit: u64, max_tx_size: u64);
 }
-Original file line number
+Diff line change
@@ Expand Up / @@ -81,6 +81,8 @@ impl MemPoolAdapter for HashMemPoolAdapter { @@
             Ok(CURRENT_HEIGHT)
         }
+        fn report_good(&self, _ctx: Context) {}
         fn set_args(&self, _timeout_gap: u64, _cycles_limit: u64, _max_tx_size: u64) {}
     }
@@ Expand Down @@