refactor(sync)!: Use signed transaction hashes to skip transaction si…

…gnature verification (#295)

* refactor(sync): Use signed transaction hashes to skip transaction signature verification.

* update muta-sdk

* update sdk

* cargo fmt

* fix e2e

core/api/src/schema/block.rs

@@ -25,31 +25,33 @@ pub struct BlockHeader {
     #[graphql(
         description = "Identifier of a chain in order to prevent replay attacks across channels "
     )]
-    pub chain_id:          Hash,
+    pub chain_id:                       Hash,
     #[graphql(description = "block height")]
-    pub height:            Uint64,
+    pub height:                         Uint64,
     #[graphql(description = "The height to which the block has been executed")]
-    pub exec_height:       Uint64,
+    pub exec_height:                    Uint64,
     #[graphql(description = "The hash of the serialized previous block")]
-    pub pre_hash:          Hash,
+    pub prev_hash:                      Hash,
     #[graphql(description = "A timestamp that records when the block was created")]
-    pub timestamp:         Uint64,
+    pub timestamp:                      Uint64,
     #[graphql(description = "The merkle root of ordered transactions")]
-    pub order_root:        MerkleRoot,
+    pub order_root:                     MerkleRoot,
+    #[graphql(description = "The hash of ordered signed transactions")]
+    pub order_signed_transactions_hash: Hash,
     #[graphql(description = "The merkle roots of all the confirms")]
-    pub confirm_root:      Vec<MerkleRoot>,
+    pub confirm_root:                   Vec<MerkleRoot>,
     #[graphql(description = "The merkle root of state root")]
-    pub state_root:        MerkleRoot,
+    pub state_root:                     MerkleRoot,
     #[graphql(description = "The merkle roots of receipts")]
-    pub receipt_root:      Vec<MerkleRoot>,
+    pub receipt_root:                   Vec<MerkleRoot>,
     #[graphql(description = "The sum of all transactions costs")]
-    pub cycles_used:       Vec<Uint64>,
+    pub cycles_used:                    Vec<Uint64>,
     #[graphql(description = "The address descirbed who packed the block")]
-    pub proposer:          Address,
-    pub proof:             Proof,
+    pub proposer:                       Address,
+    pub proof:                          Proof,
     #[graphql(description = "The version of validator is designed for cross chain")]
-    pub validator_version: Uint64,
-    pub validators:        Vec<Validator>,
+    pub validator_version:              Uint64,
+    pub validators:                     Vec<Validator>,
 }
 
 #[derive(juniper::GraphQLObject, Clone)]
@@ -73,32 +75,33 @@ pub struct Validator {
 impl From<protocol::types::BlockHeader> for BlockHeader {
     fn from(block_header: protocol::types::BlockHeader) -> Self {
         BlockHeader {
-            chain_id:          Hash::from(block_header.chain_id),
-            height:            Uint64::from(block_header.height),
-            exec_height:       Uint64::from(block_header.exec_height),
-            pre_hash:          Hash::from(block_header.pre_hash),
-            timestamp:         Uint64::from(block_header.timestamp),
-            order_root:        MerkleRoot::from(block_header.order_root),
-            state_root:        MerkleRoot::from(block_header.state_root),
-            confirm_root:      block_header
+            chain_id:                       Hash::from(block_header.chain_id),
+            height:                         Uint64::from(block_header.height),
+            exec_height:                    Uint64::from(block_header.exec_height),
+            prev_hash:                      Hash::from(block_header.prev_hash),
+            timestamp:                      Uint64::from(block_header.timestamp),
+            order_root:                     MerkleRoot::from(block_header.order_root),
+            order_signed_transactions_hash: Hash::from(block_header.order_signed_transactions_hash),
+            state_root:                     MerkleRoot::from(block_header.state_root),
+            confirm_root:                   block_header
                 .confirm_root
                 .into_iter()
                 .map(MerkleRoot::from)
                 .collect(),
-            receipt_root:      block_header
+            receipt_root:                   block_header
                 .receipt_root
                 .into_iter()
                 .map(MerkleRoot::from)
                 .collect(),
-            cycles_used:       block_header
+            cycles_used:                    block_header
                 .cycles_used
                 .into_iter()
                 .map(Uint64::from)
                 .collect(),
-            proposer:          Address::from(block_header.proposer),
-            proof:             Proof::from(block_header.proof),
-            validator_version: Uint64::from(block_header.validator_version),
-            validators:        block_header
+            proposer:                       Address::from(block_header.proposer),
+            proof:                          Proof::from(block_header.proof),
+            validator_version:              Uint64::from(block_header.validator_version),
+            validators:                     block_header
                 .validators
                 .into_iter()
                 .map(Validator::from)

core/consensus/Cargo.toml

@@ -25,6 +25,7 @@ rlp = "0.4"
 serde = { version = "1.0", features = ["derive"] }
 serde_json = "1.0"
 tokio = { version = "0.2", features = ["macros", "sync", "rt-core"] }
+bytes = { version = "0.5", features = ["serde"] }
 
 common-apm = { path = "../../common/apm" }
 common-crypto = { path = "../../common/crypto" }

core/consensus/src/adapter.rs

@@ -346,21 +346,6 @@ where
             .await?;
         Ok(ret.inner)
     }
-
-    #[muta_apm::derive::tracing_span(kind = "consensus.adapter", logs = "{'txs_len': 'txs.len()'}")]
-    async fn verify_txs_sync(
-        &self,
-        ctx: Context,
-        height: u64,
-        txs: Vec<SignedTransaction>,
-    ) -> ProtocolResult<()> {
-        if let Err(e) = self.mempool.ensure_order_txs_sync(ctx.clone(), txs).await {
-            log::error!("verify_txs error {:?}", e);
-            return Err(ConsensusError::VerifyTransaction(height).into());
-        }
-
-        Ok(())
-    }
 }
 
 #[async_trait]
@@ -534,19 +519,19 @@ where
 
         let previous_block_hash = Hash::digest(previous_block.header.encode_fixed()?);
 
-        if previous_block_hash != block.header.pre_hash {
+        if previous_block_hash != block.header.prev_hash {
             log::error!(
-                "[consensus] verify_block_header, previous_block_hash: {:?}, block.header.pre_hash: {:?}",
+                "[consensus] verify_block_header, previous_block_hash: {:?}, block.header.prev_hash: {:?}",
                 previous_block_hash,
-                block.header.pre_hash
+                block.header.prev_hash
             );
             return Err(
                 ConsensusError::VerifyBlockHeader(block.header.height, PreviousBlockHash).into(),
             );
         }
 
         // the block 0 and 1 's proof is consensus-ed by community
-        if block.header.height > 1u64 && block.header.pre_hash != block.header.proof.block_hash {
+        if block.header.height > 1u64 && block.header.prev_hash != block.header.proof.block_hash {
             log::error!(
                 "[consensus] verify_block_header, verifying_block header : {:?}",
                 block.header

core/consensus/src/engine.rs

@@ -34,7 +34,7 @@ use crate::message::{
     END_GOSSIP_SIGNED_VOTE,
 };
 use crate::status::StatusAgent;
-use crate::util::{check_list_roots, OverlordCrypto};
+use crate::util::{check_list_roots, digest_signed_transactions, OverlordCrypto};
 use crate::wal::SignedTxsWAL;
 use crate::ConsensusError;
 
@@ -78,13 +78,18 @@ impl<Adapter: ConsensusAdapter + 'static> Engine<FixedPill> for ConsensusEngine<
         let (ordered_tx_hashes, propose_hashes) = self
             .adapter
             .get_txs_from_mempool(
-                ctx,
+                ctx.clone(),
                 next_height,
                 current_consensus_status.cycles_limit,
                 current_consensus_status.tx_num_limit,
             )
             .await?
             .clap();
+        let signed_txs = self
+            .adapter
+            .get_full_txs(ctx.clone(), ordered_tx_hashes.clone())
+            .await?;
+        let order_signed_transactions_hash = digest_signed_transactions(&signed_txs)?;
 
         if current_consensus_status.latest_committed_height != next_height - 1 {
             return Err(ProtocolError::from(ConsensusError::MissingBlockHeader(
@@ -94,17 +99,17 @@ impl<Adapter: ConsensusAdapter + 'static> Engine<FixedPill> for ConsensusEngine<
         }
 
         let order_root = Merkle::from_hashes(ordered_tx_hashes.clone()).get_root_hash();
-
         let state_root = current_consensus_status.get_latest_state_root();
 
         let header = BlockHeader {
             chain_id: self.node_info.chain_id.clone(),
-            pre_hash: current_consensus_status.current_hash,
+            prev_hash: current_consensus_status.current_hash,
             height: next_height,
             exec_height: current_consensus_status.exec_height,
             timestamp: time_now(),
             logs_bloom: current_consensus_status.list_logs_bloom,
             order_root: order_root.unwrap_or_else(Hash::from_empty),
+            order_signed_transactions_hash,
             confirm_root: current_consensus_status.list_confirm_root,
             state_root,
             receipt_root: current_consensus_status.list_receipt_root.clone(),
@@ -218,6 +223,12 @@ impl<Adapter: ConsensusAdapter + 'static> Engine<FixedPill> for ConsensusEngine<
                     e
                 })?;
 
+            let signed_txs = self
+                .adapter
+                .get_full_txs(ctx.clone(), block.inner.block.ordered_tx_hashes.clone())
+                .await?;
+            self.check_order_transactions(ctx.clone(), &block.inner.block, &signed_txs)?;
+
             let adapter = Arc::clone(&self.adapter);
             let ctx_clone = ctx.clone();
             tokio::spawn(async move {
@@ -571,17 +582,17 @@ impl<Adapter: ConsensusAdapter + 'static> ConsensusEngine<Adapter> {
         let status = self.status_agent.to_inner();
 
         // check previous hash
-        if status.current_hash != block.pre_hash {
+        if status.current_hash != block.prev_hash {
             trace::error(
                 "check_block_prev_hash_diff".to_string(),
                 Some(json!({
-                    "next block prev_hash": block.pre_hash.as_hex(),
+                    "next block prev_hash": block.prev_hash.as_hex(),
                     "status current hash": status.current_hash.as_hex(),
                 })),
             );
             return Err(ConsensusError::InvalidPrevhash {
                 expect: status.current_hash,
-                actual: block.pre_hash.clone(),
+                actual: block.prev_hash.clone(),
             }
             .into());
         }
@@ -680,6 +691,39 @@ impl<Adapter: ConsensusAdapter + 'static> ConsensusEngine<Adapter> {
         Ok(())
     }
 
+    #[muta_apm::derive::tracing_span(
+        kind = "consensus.engine",
+        logs = "{'txs_len': 'signed_txs.len()'}"
+    )]
+    fn check_order_transactions(
+        &self,
+        ctx: Context,
+        block: &Block,
+        signed_txs: &[SignedTransaction],
+    ) -> ProtocolResult<()> {
+        let order_root = Merkle::from_hashes(block.ordered_tx_hashes.clone())
+            .get_root_hash()
+            .unwrap_or_else(Hash::from_empty);
+        if order_root != block.header.order_root {
+            return Err(ConsensusError::InvalidOrderRoot {
+                expect: order_root,
+                actual: block.header.order_root.clone(),
+            }
+            .into());
+        }
+
+        let order_signed_transactions_hash = digest_signed_transactions(signed_txs)?;
+        if order_signed_transactions_hash != block.header.order_signed_transactions_hash {
+            return Err(ConsensusError::InvalidOrderSignedTransactionsHash {
+                expect: order_signed_transactions_hash,
+                actual: block.header.order_signed_transactions_hash.clone(),
+            }
+            .into());
+        }
+
+        Ok(())
+    }
+
     /// After get the signed transactions:
     /// 1. Execute the signed transactions.
     /// 2. Save the signed transactions.
@@ -789,7 +833,7 @@ pub fn trace_block(block: &Block) {
         "commit_block".to_string(),
         Some(json!({
             "height": block.header.height,
-            "pre_hash": block.header.pre_hash.as_hex(),
+            "prev_hash": block.header.prev_hash.as_hex(),
             "order_root": block.header.order_root.as_hex(),
             "state_root": block.header.state_root.as_hex(),
             "proposer": block.header.proposer.as_hex(),

core/consensus/src/fixed_types.rs

@@ -196,10 +196,11 @@ mod test {
             chain_id: nonce.clone(),
             height,
             exec_height: height - 1,
-            pre_hash: nonce.clone(),
+            prev_hash: nonce.clone(),
             timestamp: 1000,
             logs_bloom: Default::default(),
             order_root: nonce.clone(),
+            order_signed_transactions_hash: nonce.clone(),
             confirm_root: Vec::new(),
             state_root: nonce,
             receipt_root: Vec::new(),

core/consensus/src/lib.rs

@@ -21,7 +21,7 @@ use derive_more::Display;
 
 use common_crypto::Error as CryptoError;
 
-use protocol::types::Hash;
+use protocol::types::{Hash, MerkleRoot};
 use protocol::{ProtocolError, ProtocolErrorKind};
 
 pub use crate::adapter::OverlordConsensusAdapter;
@@ -60,14 +60,23 @@ pub enum ConsensusType {
 /// Consensus errors defines here.
 #[derive(Debug, Display)]
 pub enum ConsensusError {
-    /// Send consensus message error.
-    #[display(fmt = "Send {:?} message failed", _0)]
-    SendMsgErr(ConsensusType),
-
     /// Check block error.
     #[display(fmt = "Check invalid prev_hash, expect {:?} get {:?}", expect, actual)]
     InvalidPrevhash { expect: Hash, actual: Hash },
 
+    #[display(fmt = "Check invalid order root, expect {:?} get {:?}", expect, actual)]
+    InvalidOrderRoot {
+        expect: MerkleRoot,
+        actual: MerkleRoot,
+    },
+
+    #[display(
+        fmt = "Check invalid order signed transactions hash, expect {:?} get {:?}",
+        expect,
+        actual
+    )]
+    InvalidOrderSignedTransactionsHash { expect: Hash, actual: Hash },
+
     #[display(fmt = "Check invalid status vec")]
     InvalidStatusVec,
 
@@ -108,14 +117,6 @@ pub enum ConsensusError {
     #[display(fmt = "Synchronization/Consensus {} block error : {}", _0, _1)]
     VerifyProof(u64, BlockProofField),
 
-    /// The Rpc response mismatch the request.
-    #[display(fmt = "Synchronization Rpc {:?} message mismatch", _0)]
-    RpcErr(ConsensusType),
-
-    ///
-    #[display(fmt = "Get merkle root failed {:?}", _0)]
-    MerkleErr(String),
-
     ///
     #[display(fmt = "Execute transactions error {:?}", _0)]
     ExecuteErr(String),
@@ -133,10 +134,10 @@ pub enum ConsensusError {
 
 #[derive(Debug, Display)]
 pub enum BlockHeaderField {
-    #[display(fmt = "The pre_hash mismatch the previous block")]
+    #[display(fmt = "The prev_hash mismatch the previous block")]
     PreviousBlockHash,
 
-    #[display(fmt = "The pre_hash mismatch the hash in the proof field")]
+    #[display(fmt = "The prev_hash mismatch the hash in the proof field")]
     ProofHash,
 
     #[display(fmt = "The proposer is not in the committee")]