validator: Port serialization to async functions

src/serial/src/lib.rs

@@ -29,7 +29,8 @@ mod async_lib;
 #[cfg(feature = "async")]
 pub use async_lib::{
     async_trait, deserialize_async, deserialize_async_partial, serialize_async, AsyncDecodable,
-    AsyncEncodable, AsyncRead, AsyncWrite, FutAsyncReadExt, FutAsyncWriteExt,
+    AsyncEncodable, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, FutAsyncReadExt,
+    FutAsyncWriteExt,
 };
 
 mod endian;

src/validator/consensus.rs

@@ -99,7 +99,7 @@ impl Consensus {
 
         // If no forks exist, create a new one as a basis to extend
         if forks.is_empty() {
-            forks.push(Fork::new(&self.blockchain, self.module.read().await.clone())?);
+            forks.push(Fork::new(&self.blockchain, self.module.read().await.clone()).await?);
         }
 
         for fork in forks.iter_mut() {
@@ -122,7 +122,7 @@ impl Consensus {
 
         // If no forks exist, create a new one as a basis to extend
         if forks.is_empty() {
-            forks.push(Fork::new(&self.blockchain, self.module.read().await.clone())?);
+            forks.push(Fork::new(&self.blockchain, self.module.read().await.clone()).await?);
         }
 
         // Grab previous slot information
@@ -214,7 +214,7 @@ impl Consensus {
         let (mut fork, index) = verify_proposal(self, proposal).await?;
 
         // Append proposal to the fork
-        fork.append_proposal(proposal.hash, self.pos_testing_mode)?;
+        fork.append_proposal(proposal.hash, self.pos_testing_mode).await?;
 
         // Update fork slots based on proposal version
         match proposal.block.header.version {
@@ -275,7 +275,7 @@ impl Consensus {
             }
 
             // Generate a new fork extending canonical
-            let mut fork = Fork::new(&self.blockchain, self.module.read().await.clone())?;
+            let mut fork = Fork::new(&self.blockchain, self.module.read().await.clone()).await?;
             if proposal.block.header.height < POS_START {
                 fork.generate_pow_slot()?;
             } else {
@@ -296,7 +296,7 @@ impl Consensus {
         }
 
         // Rebuild fork
-        let mut fork = Fork::new(&self.blockchain, self.module.read().await.clone())?;
+        let mut fork = Fork::new(&self.blockchain, self.module.read().await.clone()).await?;
         fork.proposals = original_fork.proposals[..p_index + 1].to_vec();
 
         // Retrieve proposals blocks from original fork
@@ -443,7 +443,7 @@ pub struct Fork {
 }
 
 impl Fork {
-    pub fn new(blockchain: &Blockchain, module: PoWModule) -> Result<Self> {
+    pub async fn new(blockchain: &Blockchain, module: PoWModule) -> Result<Self> {
         let mempool =
             blockchain.get_pending_txs()?.iter().map(|tx| blake3::hash(&serialize(tx))).collect();
         let overlay = BlockchainOverlay::new(blockchain)?;
@@ -451,13 +451,13 @@ impl Fork {
     }
 
     /// Auxiliary function to append a proposal and recalculate current fork rank
-    pub fn append_proposal(
+    pub async fn append_proposal(
         &mut self,
         proposal: blake3::Hash,
         pos_testing_mode: bool,
     ) -> Result<()> {
         self.proposals.push(proposal);
-        self.rank = self.rank(pos_testing_mode)?;
+        self.rank = self.rank(pos_testing_mode).await?;
 
         Ok(())
     }
@@ -607,7 +607,7 @@ impl Fork {
     }
 
     /// Auxiliarry function to compute fork's rank, assuming all proposals are valid.
-    pub fn rank(&self, pos_testing_mode: bool) -> Result<u64> {
+    pub async fn rank(&self, pos_testing_mode: bool) -> Result<u64> {
         // If the fork is empty its rank is 0
         if self.proposals.is_empty() {
             return Ok(0)
@@ -629,7 +629,7 @@ impl Fork {
             } else {
                 proposal.clone()
             };
-            sum += block_rank(proposal, &previous_previous, pos_testing_mode)?;
+            sum += block_rank(proposal, &previous_previous, pos_testing_mode).await?;
         }
 
         // Use fork(proposals) length as a multiplier to compute the actual fork rank

src/validator/mod.rs

@@ -22,7 +22,7 @@ use darkfi_sdk::{
     blockchain::{expected_reward, Slot},
     crypto::PublicKey,
 };
-use darkfi_serial::serialize;
+use darkfi_serial::serialize_async;
 use log::{debug, error, info, warn};
 use num_bigint::BigUint;
 use smol::lock::RwLock;
@@ -143,7 +143,7 @@ impl Validator {
         let overlay = BlockchainOverlay::new(&blockchain)?;
 
         // Deploy native wasm contracts
-        deploy_native_contracts(&overlay, &config.time_keeper, &config.faucet_pubkeys)?;
+        deploy_native_contracts(&overlay, &config.time_keeper, &config.faucet_pubkeys).await?;
 
         // Add genesis block if blockchain is empty
         if blockchain.genesis().is_err() {
@@ -182,7 +182,7 @@ impl Validator {
     /// The node retrieves a transaction, validates its state transition,
     /// and appends it to the pending txs store.
     pub async fn append_tx(&self, tx: &Transaction) -> Result<()> {
-        let tx_hash = blake3::hash(&serialize(tx));
+        let tx_hash = blake3::hash(&serialize_async(tx).await);
 
         // Check if we have already seen this tx
         let tx_in_txstore = self.blockchain.transactions.contains(&tx_hash)?;
@@ -262,7 +262,7 @@ impl Validator {
 
         let mut removed_txs = vec![];
         for tx in pending_txs {
-            let tx_hash = &blake3::hash(&serialize(&tx));
+            let tx_hash = &blake3::hash(&serialize_async(&tx).await);
             let tx_vec = [tx.clone()];
             let mut valid = false;
 
@@ -584,7 +584,7 @@ impl Validator {
             PoWModule::new(blockchain.clone(), pow_threads, pow_target, pow_fixed_difficulty)?;
 
         // Deploy native wasm contracts
-        deploy_native_contracts(&overlay, &time_keeper, &faucet_pubkeys)?;
+        deploy_native_contracts(&overlay, &time_keeper, &faucet_pubkeys).await?;
 
         // Validate genesis block
         verify_genesis_block(&overlay, &time_keeper, previous, genesis_txs_total).await?;

src/validator/utils.rs

@@ -16,8 +16,6 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::io::Cursor;
-
 use darkfi_sdk::{
     crypto::{
         ecvrf::VrfProof, pasta_prelude::PrimeField, PublicKey, CONSENSUS_CONTRACT_ID,
@@ -25,8 +23,9 @@ use darkfi_sdk::{
     },
     pasta::{group::ff::FromUniformBytes, pallas},
 };
-use darkfi_serial::{serialize, Decodable};
+use darkfi_serial::{serialize_async, AsyncDecodable};
 use log::info;
+use smol::io::Cursor;
 
 use crate::{
     blockchain::{BlockInfo, BlockchainOverlayPtr},
@@ -48,7 +47,7 @@ use crate::{
 /// touch anything, or just potentially update the db schemas or whatever
 /// is necessary. This logic should be handled in the init function of
 /// the actual contract, so make sure the native contracts handle this well.
-pub fn deploy_native_contracts(
+pub async fn deploy_native_contracts(
     overlay: &BlockchainOverlayPtr,
     time_keeper: &TimeKeeper,
     faucet_pubkeys: &Vec<PublicKey>,
@@ -57,7 +56,7 @@ pub fn deploy_native_contracts(
 
     // The faucet pubkeys are pubkeys which are allowed to create clear inputs
     // in the Money contract.
-    let money_contract_deploy_payload = serialize(faucet_pubkeys);
+    let money_contract_deploy_payload = serialize_async(faucet_pubkeys).await;
 
     // The DAO contract uses an empty payload to deploy itself.
     let dao_contract_deploy_payload = vec![];
@@ -105,7 +104,7 @@ pub fn deploy_native_contracts(
 /// Genesis block has rank 0.
 /// First 2 blocks rank is equal to their nonce, since their previous
 /// previous block producer doesn't exist or have a VRF.
-pub fn block_rank(
+pub async fn block_rank(
     block: &BlockInfo,
     previous_previous: &BlockInfo,
     pos_testing_mode: bool,
@@ -137,7 +136,7 @@ pub fn block_rank(
     };
     let mut decoder = Cursor::new(&data);
     decoder.set_position(position);
-    let vrf_proof: VrfProof = Decodable::decode(&mut decoder)?;
+    let vrf_proof: VrfProof = AsyncDecodable::decode_async(&mut decoder).await?;
 
     // Compute VRF u64
     let mut vrf = [0u8; 64];
@@ -179,7 +178,7 @@ pub fn median(mut v: Vec<u64>) -> u64 {
 /// genesis transactions set. This includes both staked and normal tokens.
 /// If a non-genesis transaction is found, execution fails.
 /// Set must also include the genesis transaction(empty) at last position.
-pub fn genesis_txs_total(txs: &[Transaction]) -> Result<u64> {
+pub async fn genesis_txs_total(txs: &[Transaction]) -> Result<u64> {
     let mut total = 0;
 
     if txs.is_empty() {
@@ -211,7 +210,7 @@ pub fn genesis_txs_total(txs: &[Transaction]) -> Result<u64> {
         let position = 1;
         let mut decoder = Cursor::new(&data);
         decoder.set_position(position);
-        let value: u64 = Decodable::decode(&mut decoder)?;
+        let value: u64 = AsyncDecodable::decode_async(&mut decoder).await?;
 
         total += value;
     }

src/validator/verification.rs

@@ -16,7 +16,7 @@
  * along with this program.  If not, see <https://www.gnu.org/licenses/>.
  */
 
-use std::{collections::HashMap, io::Cursor};
+use std::collections::HashMap;
 
 use darkfi_sdk::{
     blockchain::{block_version, expected_reward},
@@ -24,8 +24,9 @@ use darkfi_sdk::{
     dark_tree::dark_leaf_vec_integrity_check,
     pasta::pallas,
 };
-use darkfi_serial::{deserialize_async, Decodable, Encodable, WriteExt};
+use darkfi_serial::{deserialize_async, AsyncDecodable, AsyncEncodable, AsyncWriteExt, WriteExt};
 use log::{debug, error, warn};
+use smol::io::Cursor;
 
 use crate::{
     blockchain::{BlockInfo, BlockchainOverlayPtr},
@@ -244,8 +245,8 @@ pub async fn verify_producer_transaction(
 
     // Write the actual payload data
     let mut payload = vec![];
-    payload.write_u32(0)?; // Call index
-    tx.calls.encode(&mut payload)?; // Actual call data
+    payload.write_u32_async(0).await?; // Call index
+    tx.calls.encode_async(&mut payload).await?; // Actual call data
 
     debug!(target: "validator::verification::verify_producer_transaction", "Instantiating WASM runtime");
     let wasm = overlay.lock().unwrap().wasm_bincode.get(call.data.contract_id)?;
@@ -260,8 +261,9 @@ pub async fn verify_producer_transaction(
     let mut decoder = Cursor::new(&metadata);
 
     // The tuple is (zkas_ns, public_inputs)
-    let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
-    let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
+    let zkp_pub: Vec<(String, Vec<pallas::Base>)> =
+        AsyncDecodable::decode_async(&mut decoder).await?;
+    let sig_pub: Vec<PublicKey> = AsyncDecodable::decode_async(&mut decoder).await?;
 
     // Check that only one ZK proof and signature public key exist
     if zkp_pub.len() != 1 || sig_pub.len() != 1 {
@@ -345,15 +347,19 @@ pub async fn verify_transaction(
     let mut gas_used = 0;
 
     // Verify calls indexes integrity
-    dark_leaf_vec_integrity_check(&tx.calls, Some(MIN_TX_CALLS), Some(MAX_TX_CALLS))?;
+    if verify_fee {
+        dark_leaf_vec_integrity_check(&tx.calls, Some(MIN_TX_CALLS + 1), Some(MAX_TX_CALLS))?;
+    } else {
+        dark_leaf_vec_integrity_check(&tx.calls, Some(MIN_TX_CALLS), Some(MAX_TX_CALLS))?;
+    }
 
     // Table of public inputs used for ZK proof verification
     let mut zkp_table = vec![];
     // Table of public keys used for signature verification
     let mut sig_table = vec![];
 
-    // Verify that the first call is the transaction fee and that it has no parents or children.
     if verify_fee {
+        // Verify that the first call is the transaction fee and that it has no parents or children.
         if tx.calls[0].data.contract_id != *MONEY_CONTRACT_ID || tx.calls[0].data.data[0] != 0x00 {
             error!(
                 target: "validator::verification::verify_transaction",
@@ -386,7 +392,7 @@ pub async fn verify_transaction(
         // Write the actual payload data
         let mut payload = vec![];
         payload.write_u32(idx as u32)?; // Call index
-        tx.calls.encode(&mut payload)?; // Actual call data
+        tx.calls.encode_async(&mut payload).await?; // Actual call data
 
         debug!(target: "validator::verification::verify_transaction", "Instantiating WASM runtime");
         let wasm = overlay.lock().unwrap().wasm_bincode.get(call.data.contract_id)?;
@@ -401,8 +407,9 @@ pub async fn verify_transaction(
         let mut decoder = Cursor::new(&metadata);
 
         // The tuple is (zkas_ns, public_inputs)
-        let zkp_pub: Vec<(String, Vec<pallas::Base>)> = Decodable::decode(&mut decoder)?;
-        let sig_pub: Vec<PublicKey> = Decodable::decode(&mut decoder)?;
+        let zkp_pub: Vec<(String, Vec<pallas::Base>)> =
+            AsyncDecodable::decode_async(&mut decoder).await?;
+        let sig_pub: Vec<PublicKey> = AsyncDecodable::decode_async(&mut decoder).await?;
         // TODO: Make sure we've read all the bytes above.
         debug!(target: "validator::verification::verify_transaction", "Successfully executed \"metadata\" call");