consensus: moved state transitions/validations to state.rs

src/consensus/proto/protocol_proposal.rs

@@ -2,16 +2,15 @@ use async_std::sync::Arc;
 
 use async_executor::Executor;
 use async_trait::async_trait;
-use log::{debug, error, info, warn};
+use log::{debug, error, info};
 use url::Url;
 
 use crate::{
-    consensus::{BlockProposal, ValidatorState, ValidatorStatePtr},
+    consensus::{BlockProposal, ValidatorStatePtr},
     net::{
         ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
         ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
-    node::MemoryState,
     Result,
 };
 
@@ -64,26 +63,7 @@ impl ProtocolProposal {
 
             let proposal_copy = (*proposal).clone();
 
-            debug!(
-                "ProtocolProposal::handle_receive_proposal(): Starting state transition validation"
-            );
-            let canon_state_clone = self.state.read().await.state_machine.lock().await.clone();
-            let mem_state = MemoryState::new(canon_state_clone);
-
-            match ValidatorState::validate_state_transitions(mem_state, &proposal_copy.block.txs) {
-                Ok(_) => {
-                    debug!("ProtocolProposal::handle_receive_proposal(): State transition valid")
-                }
-                Err(e) => {
-                    warn!(
-                        "ProtocolProposal::handle_receive_proposal(): State transition fail: {}",
-                        e
-                    );
-                    continue
-                }
-            }
-
-            let vote = match self.state.write().await.receive_proposal(&proposal_copy) {
+            let vote = match self.state.write().await.receive_proposal(&proposal_copy).await {
                 Ok(v) => {
                     if v.is_none() {
                         debug!("ProtocolProposal::handle_receive_proposal(): Node didn't vote for proposed block.");

src/consensus/proto/protocol_sync.rs

@@ -1,18 +1,17 @@
 use async_executor::Executor;
-use async_std::sync::{Arc, Mutex};
+use async_std::sync::Arc;
 use async_trait::async_trait;
-use log::{debug, error, info, warn};
+use log::{debug, error, info};
 
 use crate::{
     consensus::{
         block::{BlockInfo, BlockOrder, BlockResponse},
-        ValidatorState, ValidatorStatePtr,
+        ValidatorStatePtr,
     },
     net::{
         ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
         ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
-    node::MemoryState,
     Result,
 };
 
@@ -27,7 +26,6 @@ pub struct ProtocolSync {
     state: ValidatorStatePtr,
     p2p: P2pPtr,
     consensus_mode: bool,
-    pending: Mutex<bool>,
 }
 
 impl ProtocolSync {
@@ -52,7 +50,6 @@ impl ProtocolSync {
             state,
             p2p,
             consensus_mode,
-            pending: Mutex::new(false),
         }))
     }
 
@@ -113,83 +110,26 @@ impl ProtocolSync {
                 info.header.headerhash()
             );
 
-            // We block here if there's a pending validation, otherwise we might
-            // apply the same block twice.
-            debug!("ProtocolSync::handle_receive_block(): Waiting for pending block to apply");
-            while *self.pending.lock().await {}
-            debug!("ProtocolSync::handle_receive_block(): Pending lock released");
-
-            // Node stores finalized block, if it doesn't exist (checking by slot),
-            // and removes its transactions from the unconfirmed_txs vector.
-            // Extra validations can be added here.
-            *self.pending.lock().await = true;
+            debug!("ProtocolSync::handle_receive_block(): Processing received block");
             let info_copy = (*info).clone();
-
-            let has_block = match self.state.read().await.blockchain.has_block(&info_copy) {
-                Ok(v) => v,
+            match self.state.write().await.receive_finalized_block(info_copy.clone()).await {
+                Ok(v) => {
+                    if v {
+                        debug!("ProtocolProposal::handle_receive_block(): block processed successfully, broadcasting...");
+                        if let Err(e) =
+                            self.p2p.broadcast_with_exclude(info_copy, &exclude_list).await
+                        {
+                            error!(
+                                "ProtocolSync::handle_receive_block(): p2p broadcast fail: {}",
+                                e
+                            );
+                        };
+                    }
+                }
                 Err(e) => {
-                    error!(
-                        "ProtocolSync::handle_receive_block(): failed checking for has_block(): {}",
-                        e
-                    );
-                    *self.pending.lock().await = false;
-                    continue
+                    debug!("ProtocolSync::handle_receive_block(): error processing finalized block: {}", e);
                 }
             };
-
-            if !has_block {
-                debug!(
-                    "ProtocolSync::handle_receive_block(): Starting state transition validation"
-                );
-                let canon_state_clone = self.state.read().await.state_machine.lock().await.clone();
-                let mem_state = MemoryState::new(canon_state_clone);
-                let state_updates =
-                    match ValidatorState::validate_state_transitions(mem_state, &info.txs) {
-                        Ok(v) => v,
-                        Err(e) => {
-                            warn!(
-                                "ProtocolSync::handle_receive_block(): State transition fail: {}",
-                                e
-                            );
-                            *self.pending.lock().await = false;
-                            continue
-                        }
-                    };
-                debug!("ProtocolSync::handle_receive_block(): All state transitions passed");
-
-                debug!("ProtocolSync::handle_receive_block(): Updating canon state machine");
-                if let Err(e) =
-                    self.state.write().await.update_canon_state(state_updates, None).await
-                {
-                    error!(
-                        "ProtocolSync::handle_receive_block(): Canon statemachine update fail: {}",
-                        e
-                    );
-                    *self.pending.lock().await = false;
-                    continue
-                };
-
-                debug!("ProtocolSync::handle_receive_block(): Appending block to ledger");
-                if let Err(e) = self.state.write().await.blockchain.add(&[info_copy.clone()]) {
-                    error!("ProtocolSync::handle_receive_block(): blockchain.add() fail: {}", e);
-                    *self.pending.lock().await = false;
-                    continue
-                };
-
-                if let Err(e) = self.state.write().await.remove_txs(info_copy.txs.clone()) {
-                    error!("ProtocolSync::handle_receive_block(): remove_txs() fail: {}", e);
-                    *self.pending.lock().await = false;
-                    continue
-                };
-
-                if let Err(e) = self.p2p.broadcast_with_exclude(info_copy, &exclude_list).await {
-                    error!("ProtocolSync::handle_receive_block(): p2p broadcast fail: {}", e);
-                    *self.pending.lock().await = false;
-                    continue
-                };
-            }
-
-            *self.pending.lock().await = false;
         }
     }
 }

src/consensus/proto/protocol_tx.rs

@@ -2,19 +2,17 @@ use async_std::sync::Arc;
 
 use async_executor::Executor;
 use async_trait::async_trait;
-use log::{debug, error, warn};
+use log::{debug, error};
 use url::Url;
 
 use crate::{
-    consensus::{ValidatorState, ValidatorStatePtr},
+    consensus::ValidatorStatePtr,
     net,
     net::{
         ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
         ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
-    node::MemoryState,
     tx::Transaction,
-    util::serial::serialize,
     Result,
 };
 
@@ -67,40 +65,12 @@ impl ProtocolTx {
             };
 
             let tx_copy = (*tx).clone();
-            let tx_hash = blake3::hash(&serialize(&tx_copy));
 
-            {
-                let state = &mut self.state.write().await;
-                let tx_in_txstore = match state.blockchain.transactions.contains(&tx_hash) {
-                    Ok(v) => v,
-                    Err(e) => {
-                        error!("handle_receive_tx(): Failed querying txstore: {}", e);
-                        continue
-                    }
+            // Nodes use unconfirmed_txs vector as seen_txs pool.
+            if self.state.write().await.append_tx(tx_copy.clone()).await {
+                if let Err(e) = self.p2p.broadcast_with_exclude(tx_copy, &exclude_list).await {
+                    error!("handle_receive_tx(): p2p broadcast fail: {}", e);
                 };
-
-                if state.unconfirmed_txs.contains(&tx_copy) || tx_in_txstore {
-                    debug!("ProtocolTx::handle_receive_tx(): We have already seen this tx.");
-                    continue
-                }
-
-                debug!("ProtocolTx::handle_receive_tx(): Starting state transition validation");
-                let canon_state_clone = state.state_machine.lock().await.clone();
-                let mem_state = MemoryState::new(canon_state_clone);
-                match ValidatorState::validate_state_transitions(mem_state, &[tx_copy.clone()]) {
-                    Ok(_) => debug!("ProtocolTx::handle_receive_tx(): State transition valid"),
-                    Err(e) => {
-                        warn!("ProtocolTx::handle_receive_tx(): State transition fail: {}", e);
-                        continue
-                    }
-                }
-
-                // Nodes use unconfirmed_txs vector as seen_txs pool.
-                if state.append_tx(tx_copy.clone()) {
-                    if let Err(e) = self.p2p.broadcast_with_exclude(tx_copy, &exclude_list).await {
-                        error!("handle_receive_tx(): p2p broadcast fail: {}", e);
-                    };
-                }
             }
         }
     }

src/consensus/state.rs

@@ -179,15 +179,34 @@ impl ValidatorState {
         Ok(state)
     }
 
-    /// The node retrieves a transaction and appends it to the unconfirmed
-    /// transactions list. Additional validity rules must be defined by the
-    /// protocol for transactions.
-    pub fn append_tx(&mut self, tx: Transaction) -> bool {
-        if self.unconfirmed_txs.contains(&tx) {
-            debug!("append_tx(): We already have this tx");
+    /// The node retrieves a transaction, validates its state transition,
+    /// and appends it to the unconfirmed transactions list.
+    pub async fn append_tx(&mut self, tx: Transaction) -> bool {
+        let tx_hash = blake3::hash(&serialize(&tx));
+        let tx_in_txstore = match self.blockchain.transactions.contains(&tx_hash) {
+            Ok(v) => v,
+            Err(e) => {
+                error!("append_tx(): Failed querying txstore: {}", e);
+                return false
+            }
+        };
+
+        if self.unconfirmed_txs.contains(&tx) || tx_in_txstore {
+            debug!("append_tx(): We have already seen this tx.");
             return false
         }
 
+        debug!("append_tx(): Starting state transition validation");
+        let canon_state_clone = self.state_machine.lock().await.clone();
+        let mem_state = MemoryState::new(canon_state_clone);
+        match self.validate_state_transitions(mem_state, &[tx.clone()]) {
+            Ok(_) => debug!("append_tx(): State transition valid"),
+            Err(e) => {
+                warn!("append_tx(): State transition fail: {}", e);
+                return false
+            }
+        }
+
         debug!("append_tx(): Appended tx to mempool");
         self.unconfirmed_txs.push(tx);
         true
@@ -349,7 +368,7 @@ impl ValidatorState {
 
     /// Receive the proposed block, verify its sender (slot leader),
     /// and proceed with voting on it.
-    pub fn receive_proposal(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
+    pub async fn receive_proposal(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
         // Node hasn't started participating
         match self.participating {
             Some(start) => {
@@ -380,14 +399,14 @@ impl ValidatorState {
             return Ok(None)
         }
 
-        self.vote(proposal)
+        self.vote(proposal).await
     }
 
     /// Given a proposal, the node finds which blockchain it extends.
     /// If the proposal extends the canonical blockchain, a new fork chain
     /// is created. The node votes on the proposal only if it extends the
-    /// longest notarized fork chain it has seen.
-    pub fn vote(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
+    /// longest notarized fork chain it has seen and its state transition is valid.
+    pub async fn vote(&mut self, proposal: &BlockProposal) -> Result<Option<Vote>> {
         let mut proposal = proposal.clone();
 
         // Generate proposal hash
@@ -429,6 +448,20 @@ impl ValidatorState {
             return Ok(None)
         }
 
+        debug!("vote(): Starting state transition validation");
+        let canon_state_clone = self.state_machine.lock().await.clone();
+        let mem_state = MemoryState::new(canon_state_clone);
+
+        match self.validate_state_transitions(mem_state, &proposal.block.txs) {
+            Ok(_) => {
+                debug!("vote(): State transition valid")
+            }
+            Err(e) => {
+                warn!("vote(): State transition fail: {}", e);
+                return Ok(None)
+            }
+        }
+
         let signed_hash = self.secret.sign(&serialize(&proposal_hash));
         Ok(Some(Vote::new(signed_hash, proposal_hash, proposal.block.header.slot, self.address)))
     }
@@ -691,7 +724,7 @@ impl ValidatorState {
             debug!(target: "consensus", "Applying state transition for finalized block");
             let canon_state_clone = self.state_machine.lock().await.clone();
             let mem_st = MemoryState::new(canon_state_clone);
-            let state_updates = ValidatorState::validate_state_transitions(mem_st, &proposal.txs)?;
+            let state_updates = self.validate_state_transitions(mem_st, &proposal.txs)?;
             self.update_canon_state(state_updates, None).await?;
             self.remove_txs(proposal.txs.clone())?;
         }
@@ -884,9 +917,63 @@ impl ValidatorState {
     // State transition functions
     // ==========================
 
+    /// Validate and append to canonical state received blocks.
+    pub async fn receive_blocks(&mut self, blocks: &[BlockInfo]) -> Result<()> {
+        // Verify state transitions for all blocks and their respective transactions.
+        debug!("receive_blocks(): Starting state transition validations");
+        let mut canon_updates = vec![];
+        let canon_state_clone = self.state_machine.lock().await.clone();
+        let mut mem_state = MemoryState::new(canon_state_clone);
+        for block in blocks {
+            let mut state_updates =
+                self.validate_state_transitions(mem_state.clone(), &block.txs)?;
+
+            for update in &state_updates {
+                mem_state.apply(update.clone());
+            }
+
+            canon_updates.append(&mut state_updates);
+        }
+        debug!("receive_blocks(): All state transitions passed");
+
+        debug!("receive_blocks(): Updating canon state");
+        self.update_canon_state(canon_updates, None).await?;
+
+        debug!("receive_blocks(): Appending blocks to ledger");
+        self.blockchain.add(blocks)?;
+
+        Ok(())
+    }
+
+    /// Validate and append to canonical state received finalized block.
+    /// Returns boolean flag indicating already existing block.
+    pub async fn receive_finalized_block(&mut self, block: BlockInfo) -> Result<bool> {
+        match self.blockchain.has_block(&block) {
+            Ok(v) => {
+                if v {
+                    debug!("receive_finalized_block(): Existing block received");
+                    return Ok(false)
+                }
+            }
+            Err(e) => {
+                error!("receive_finalized_block(): failed checking for has_block(): {}", e);
+                return Ok(false)
+            }
+        };
+
+        debug!("receive_finalized_block(): Executing state transitions");
+        self.receive_blocks(&[block.clone()]).await?;
+
+        debug!("receive_finalized_block(): Removing block transactions from unconfirmed_txs");
+        self.remove_txs(block.txs.clone())?;
+
+        Ok(true)
+    }
+
     /// Validate state transitions for given transactions and state and
     /// return a vector of [`StateUpdate`]
     pub fn validate_state_transitions(
+        &self,
         state: MemoryState,
         txs: &[Transaction],
     ) -> Result<Vec<StateUpdate>> {

src/consensus/task/block_sync.rs

@@ -1,11 +1,9 @@
 use crate::{
     consensus::{
         block::{BlockOrder, BlockResponse},
-        ValidatorState, ValidatorStatePtr,
+        ValidatorStatePtr,
     },
-    net,
-    node::MemoryState,
-    Result,
+    net, Result,
 };
 use log::{debug, info, warn};
 
@@ -40,28 +38,9 @@ pub async fn block_sync_task(p2p: net::P2pPtr, state: ValidatorStatePtr) -> Resu
             // Node stores response data.
             let resp = response_sub.receive().await?;
 
-            // Verify state transitions for all blocks and their respective transactions.
-            debug!("block_sync_task(): Starting state transition validations");
-            let mut canon_updates = vec![];
-            let canon_state_clone = state.read().await.state_machine.lock().await.clone();
-            let mut mem_state = MemoryState::new(canon_state_clone);
-            for block in &resp.blocks {
-                let mut state_updates =
-                    ValidatorState::validate_state_transitions(mem_state.clone(), &block.txs)?;
-
-                for update in &state_updates {
-                    mem_state.apply(update.clone());
-                }
-
-                canon_updates.append(&mut state_updates);
-            }
-            debug!("block_sync_task(): All state transitions passed");
-
-            debug!("block_sync_task(): Updating canon state");
-            state.write().await.update_canon_state(canon_updates, None).await?;
-
-            debug!("block_sync_task(): Appending blocks to ledger");
-            state.write().await.blockchain.add(&resp.blocks)?;
+            // Verify and store retrieved blocks
+            debug!("block_sync_task(): Processing received blocks");
+            state.write().await.receive_blocks(&resp.blocks).await?;
 
             let last_received = state.read().await.blockchain.last()?;
             info!("Last received block: {:?} - {:?}", last_received.0, last_received.1);

src/consensus/task/proposal.rs

@@ -91,7 +91,7 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
 
         info!("consensus: Node is the slot leader: Proposed block: {}", proposal);
         debug!("consensus: Full proposal: {:?}", proposal);
-        let vote = state.write().await.receive_proposal(&proposal);
+        let vote = state.write().await.receive_proposal(&proposal).await;
         let vote = match vote {
             Ok(v) => {
                 if v.is_none() {