validatord: proper p2p messages handling, n nodes simulation

script/research/validatord/simulation/sim10/simulation.sh

@@ -0,0 +1,53 @@
+#!/bin/bash
+
+# Simulation of the consensus network for n validator nodes.
+# Note: state files should be generated before executing.
+
+nodes=10
+
+# Copying the node state files with a blockchain containing only the genesis block.
+bound=$(($nodes - 1))
+for i in $(eval echo "{0..$bound}")
+do
+  cp validatord_state_$i ~/.config/darkfi/validatord_state_$i
+done
+
+# PIDs array
+pids=()
+
+# Starting node 0 (seed) in background
+cargo run -- --nodes $nodes &
+pids[${#pids[@]}]=$!
+
+# Waiting for seed to setup
+sleep 2
+
+# Starting nodes 1 till second to last node in background
+bound=$(($nodes-2))
+for i in $(eval echo "{1..$bound}")
+do
+  cargo run -- --accept 0.0.0.0:1100$i --seeds 127.0.0.1:11000 --rpc 127.0.0.1:666$i --external 127.0.0.1:1100$i --id $i --state ~/.config/darkfi/validatord_state_$i --nodes $nodes &
+  pids[${#pids[@]}]=$!
+  # waiting for node to setup
+  sleep 2
+done
+
+# Trap kill signal
+trap ctrl_c INT
+
+# On kill signal, terminate background node processes
+function ctrl_c() {
+    for pid in ${pids[@]}
+    do
+      kill $pid
+    done
+}
+
+bound=$(($nodes-1))
+# Starting last node
+cargo run -- --accept 0.0.0.0:1100$bound --seeds 127.0.0.1:11000 --rpc 127.0.0.1:666$bound --external 127.0.0.1:1100$bound --id 3 --state ~/.config/darkfi/validatord_state_$bound --nodes $nodes
+
+# Node states are flushed on each node state file at epoch end (every 2 minutes).
+# To sugmit a TX, telnet to a node and push the json as per following example:
+# telnet 127.0.0.1 6661
+# {"jsonrpc": "2.0", "method": "receive_tx", "params": ["tx"], "id": 42}

script/research/validatord/simulation/sim10/validatord_state_0

@@ -0,0 +1,31 @@
+{
+  "id": 0,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_1

@@ -0,0 +1,31 @@
+{
+  "id": 1,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_2

@@ -0,0 +1,31 @@
+{
+  "id": 2,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_3

@@ -0,0 +1,31 @@
+{
+  "id": 3,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_4

@@ -0,0 +1,31 @@
+{
+  "id": 4,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_5

@@ -0,0 +1,31 @@
+{
+  "id": 5,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_6

@@ -0,0 +1,31 @@
+{
+  "id": 6,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_7

@@ -0,0 +1,31 @@
+{
+  "id": 7,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_8

@@ -0,0 +1,31 @@
+{
+  "id": 8,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/sim10/validatord_state_9

@@ -0,0 +1,31 @@
+{
+  "id": 9,
+  "genesis_time": 1648383795,
+  "secret_key": "22c1a9eae232ec84096c90896e763087a36b3eb55f3de599d5d3f7bba280a11e",
+  "public_key": "ad50d94b531da6864b32bed5001fde813e745841202b45d9f892e05d57593d19",
+  "canonical_blockchain": {
+    "blocks": [
+      {
+        "st": "⊥",
+        "sl": 0,
+        "txs": [],
+        "metadata": {
+          "om": {
+            "proof": "proof",
+            "r": "r",
+            "s": "s"
+          },
+          "sm": {
+            "votes": [],
+            "notarized": true,
+            "finalized": true
+          },
+          "timestamp": 1648383795
+        }
+      }
+    ]
+  },
+  "node_blockchains": [],
+  "unconfirmed_txs": [],
+  "orphan_votes": []
+}

script/research/validatord/simulation/simulation.sh

@@ -1,46 +1,51 @@
 #!/bin/bash
 
-# Simulation of the consensus network for 4(hardcoded) validator nodes.
+# Simulation of the consensus network for n validator nodes.
+# Note: state files should be generated before executing.
+
+nodes=4
 
 # Copying the node state files with a blockchain containing only the genesis block.
-for i in {0..3}
+bound=$(($nodes - 1))
+for i in $(eval echo "{0..$bound}")
 do
   cp validatord_state_$i ~/.config/darkfi/validatord_state_$i
 done
 
+# PIDs array
+pids=()
+
 # Starting node 0 (seed) in background
-cargo run -- &
-NODE0=$!
+cargo run -- --nodes $nodes &
+pids[${#pids[@]}]=$!
 
 # Waiting for seed to setup
-sleep 10
-
-# Starting node 1 in background
-cargo run -- --accept 0.0.0.0:11001 --seeds 127.0.0.1:11000 --rpc 127.0.0.1:6661 --external 127.0.0.1:11001 --id 1 --state ~/.config/darkfi/validatord_state_1 &
-NODE1=$!
-
-# Waiting for node 1 to setup
-sleep 5
+sleep 2
 
-# Starting node 2 in background
-cargo run -- --accept 0.0.0.0:11002 --seeds 127.0.0.1:11000 --rpc 127.0.0.1:6662 --external 127.0.0.1:11002 --id 2 --state ~/.config/darkfi/validatord_state_2 &
-NODE2=$!
-
-# Waiting for node 2 to setup
-sleep 5
+# Starting nodes 1 till second to last node in background
+bound=$(($nodes-2))
+for i in $(eval echo "{1..$bound}")
+do
+  cargo run -- --accept 0.0.0.0:1100$i --seeds 127.0.0.1:11000 --rpc 127.0.0.1:666$i --external 127.0.0.1:1100$i --id $i --state ~/.config/darkfi/validatord_state_$i --nodes $nodes &
+  pids[${#pids[@]}]=$!
+  # waiting for node to setup
+  sleep 2
+done
 
 # Trap kill signal
 trap ctrl_c INT
 
 # On kill signal, terminate background node processes
 function ctrl_c() {
-    kill $NODE0
-    kill $NODE1
-    kill $NODE2
+    for pid in ${pids[@]}
+    do
+      kill $pid
+    done
 }
 
-# Starting node 3
-cargo run -- --accept 0.0.0.0:11003 --seeds 127.0.0.1:11000 --rpc 127.0.0.1:6663 --external 127.0.0.1:11003 --id 3 --state ~/.config/darkfi/validatord_state_3
+bound=$(($nodes-1))
+# Starting last node
+cargo run -- --accept 0.0.0.0:1100$bound --seeds 127.0.0.1:11000 --rpc 127.0.0.1:666$bound --external 127.0.0.1:1100$bound --id 3 --state ~/.config/darkfi/validatord_state_$bound --nodes $nodes
 
 # Node states are flushed on each node state file at epoch end (every 2 minutes).
 # To sugmit a TX, telnet to a node and push the json as per following example:

script/research/validatord/src/main.rs

@@ -81,8 +81,11 @@ struct Opt {
     /// Path to the state file
     state: String,
     #[structopt(long, default_value = "0")]
-    /// How many threads to utilize
+    /// Node ID, used only for testing
     id: u64,
+    #[structopt(long, default_value = "1")]
+    /// Nodes count, used only for testing
+    nodes: u64,
     #[structopt(short, long, default_value = "0")]
     /// How many threads to utilize
     threads: usize,
@@ -91,11 +94,8 @@ struct Opt {
     verbose: u8,
 }
 
-// TODO:
-//      1. Nodes count not hardcoded.
-//      2. Remove dummy delay.
-async fn proposal_task(p2p: net::P2pPtr, state: StatePtr, state_path: &PathBuf) {
-    let nodes_count = 4;
+// TODO: 1. Nodes count retrieval.
+async fn proposal_task(p2p: net::P2pPtr, state: StatePtr, state_path: &PathBuf, nodes_count: u64) {
     // After initialization node should wait for next epoch
     let seconds_until_next_epoch = state.read().unwrap().get_seconds_until_next_epoch_start();
     info!("Waiting for next epoch({:?} sec)...", seconds_until_next_epoch);
@@ -141,7 +141,7 @@ async fn proposal_task(p2p: net::P2pPtr, state: StatePtr, state_path: &PathBuf)
                             }
                         }
                         Err(e) => {
-                            debug!(target: "ircd", "ProtocolBlock::handle_receive_proposal() error prosessing proposal: {:?}", e)
+                            error!("Error prosessing proposal: {:?}", e)
                         }
                     }
                 }
@@ -156,7 +156,7 @@ async fn proposal_task(p2p: net::P2pPtr, state: StatePtr, state_path: &PathBuf)
         match result {
             Ok(()) => (),
             Err(e) => {
-                debug!(target: "ircd", "ProtocolVote::handle_receive_proposal() error saving state: {:?}", e)
+                error!("State could not be flushed: {:?}", e)
             }
         };
     }
@@ -182,6 +182,7 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
     // State setup
     let state_path = expand_path(&opts.state).unwrap();
     let id = opts.id.clone();
+    let nodes_count = opts.nodes.clone();
     let state = State::load_current_state(id, &state_path).unwrap();
 
     // P2P registry setup
@@ -191,27 +192,31 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
     // Adding ProtocolTx to the registry
     let state2 = state.clone();
     registry
-        .register(net::SESSION_ALL, move |channel, _p2p| {
+        .register(net::SESSION_ALL, move |channel, p2p| {
             let state = state2.clone();
-            async move { ProtocolTx::init(channel, state).await }
+            async move { ProtocolTx::init(channel, state, p2p).await }
         })
         .await;
 
     // Adding PropotolVote to the registry
     let state2 = state.clone();
+    let nodes_count2 = nodes_count.clone() as usize;
     registry
-        .register(net::SESSION_ALL, move |channel, _p2p| {
+        .register(net::SESSION_ALL, move |channel, p2p| {
             let state = state2.clone();
-            async move { ProtocolVote::init(channel, state).await }
+            let nodes_count = nodes_count2.clone();
+            async move { ProtocolVote::init(channel, state, p2p, nodes_count).await }
         })
         .await;
 
     // Adding ProtocolProposal to the registry
     let state2 = state.clone();
+    let nodes_count2 = nodes_count.clone();
     registry
         .register(net::SESSION_ALL, move |channel, p2p| {
             let state = state2.clone();
-            async move { ProtocolProposal::init(channel, state, p2p).await }
+            let nodes_count = nodes_count2.clone();
+            async move { ProtocolProposal::init(channel, state, p2p, nodes_count).await }
         })
         .await;
 
@@ -240,7 +245,7 @@ async fn start(executor: Arc<Executor<'_>>, opts: &Opt) -> Result<()> {
         .spawn(async move { listen_and_serve(rpc_server_config, rpc_interface, ex3).await })
         .detach();
 
-    proposal_task(p2p, state, &state_path).await;
+    proposal_task(p2p, state, &state_path, nodes_count).await;
 
     Ok(())
 }

script/research/validatord/src/protocols/protocol_proposal.rs

@@ -17,10 +17,16 @@ pub struct ProtocolProposal {
     jobsman: ProtocolJobsManagerPtr,
     state: StatePtr,
     p2p: P2pPtr,
+    nodes_count: u64,
 }
 
 impl ProtocolProposal {
-    pub async fn init(channel: ChannelPtr, state: StatePtr, p2p: P2pPtr) -> ProtocolBasePtr {
+    pub async fn init(
+        channel: ChannelPtr,
+        state: StatePtr,
+        p2p: P2pPtr,
+        nodes_count: u64,
+    ) -> ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<BlockProposal>().await;
 
@@ -32,12 +38,11 @@ impl ProtocolProposal {
             jobsman: ProtocolJobsManager::new("ProposalProtocol", channel),
             state,
             p2p,
+            nodes_count,
         })
     }
 
-    // TODO:
-    //      1. Nodes count not hardcoded.
-    //      2. Remove dummy delay.
+    // TODO: 1. Nodes count retrieval.
     async fn handle_receive_proposal(self: Arc<Self>) -> Result<()> {
         debug!(target: "ircd", "ProtocolBlock::handle_receive_proposal() [START]");
         loop {
@@ -49,10 +54,9 @@ impl ProtocolProposal {
                 proposal
             );
             let proposal_copy = (*proposal).clone();
-            let nodes_count = 4;
             let vote = self.state.write().unwrap().receive_proposed_block(
                 &proposal_copy,
-                nodes_count,
+                self.nodes_count,
                 false,
             );
             match vote {
@@ -61,7 +65,10 @@ impl ProtocolProposal {
                         debug!("Node did not vote for the proposed block.");
                     } else {
                         let vote = x.unwrap();
-                        self.state.write().unwrap().receive_vote(&vote, nodes_count as usize);
+                        self.state.write().unwrap().receive_vote(&vote, self.nodes_count as usize);
+                        // Broadcasting block to rest nodes
+                        self.p2p.broadcast(proposal_copy).await?;
+                        // Broadcasting vote
                         self.p2p.broadcast(vote).await?;
                     }
                 }

script/research/validatord/src/protocols/protocol_tx.rs

@@ -4,8 +4,8 @@ use async_trait::async_trait;
 use darkfi::{
     consensus::{state::StatePtr, tx::Tx},
     net::{
-        ChannelPtr, MessageSubscription, ProtocolBase, ProtocolBasePtr, ProtocolJobsManager,
-        ProtocolJobsManagerPtr,
+        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
     Result,
 };
@@ -16,16 +16,22 @@ pub struct ProtocolTx {
     tx_sub: MessageSubscription<Tx>,
     jobsman: ProtocolJobsManagerPtr,
     state: StatePtr,
+    p2p: P2pPtr,
 }
 
 impl ProtocolTx {
-    pub async fn init(channel: ChannelPtr, state: StatePtr) -> ProtocolBasePtr {
+    pub async fn init(channel: ChannelPtr, state: StatePtr, p2p: P2pPtr) -> ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<Tx>().await;
 
         let tx_sub = channel.subscribe_msg::<Tx>().await.expect("Missing Tx dispatcher!");
 
-        Arc::new(Self { tx_sub, jobsman: ProtocolJobsManager::new("TxProtocol", channel), state })
+        Arc::new(Self {
+            tx_sub,
+            jobsman: ProtocolJobsManager::new("TxProtocol", channel),
+            state,
+            p2p,
+        })
     }
 
     async fn handle_receive_tx(self: Arc<Self>) -> Result<()> {
@@ -39,7 +45,9 @@ impl ProtocolTx {
                 tx
             );
             let tx_copy = (*tx).clone();
-            self.state.write().unwrap().append_tx(tx_copy.clone());
+            if self.state.write().unwrap().append_tx(tx_copy.clone()) {
+                self.p2p.broadcast(tx_copy).await?;
+            }
         }
     }
 }

script/research/validatord/src/protocols/protocol_vote.rs

@@ -4,8 +4,8 @@ use async_trait::async_trait;
 use darkfi::{
     consensus::{state::StatePtr, vote::Vote},
     net::{
-        ChannelPtr, MessageSubscription, ProtocolBase, ProtocolBasePtr, ProtocolJobsManager,
-        ProtocolJobsManagerPtr,
+        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
     },
     Result,
 };
@@ -16,10 +16,17 @@ pub struct ProtocolVote {
     vote_sub: MessageSubscription<Vote>,
     jobsman: ProtocolJobsManagerPtr,
     state: StatePtr,
+    p2p: P2pPtr,
+    nodes_count: usize,
 }
 
 impl ProtocolVote {
-    pub async fn init(channel: ChannelPtr, state: StatePtr) -> ProtocolBasePtr {
+    pub async fn init(
+        channel: ChannelPtr,
+        state: StatePtr,
+        p2p: P2pPtr,
+        nodes_count: usize,
+    ) -> ProtocolBasePtr {
         let message_subsytem = channel.get_message_subsystem();
         message_subsytem.add_dispatch::<Vote>().await;
 
@@ -29,10 +36,12 @@ impl ProtocolVote {
             vote_sub,
             jobsman: ProtocolJobsManager::new("VoteProtocol", channel),
             state,
+            p2p,
+            nodes_count,
         })
     }
 
-    // TODO: 1. Nodes count not hardcoded.
+    // TODO: 1. Nodes count retrieval.
     async fn handle_receive_vote(self: Arc<Self>) -> Result<()> {
         debug!(target: "ircd", "ProtocolVote::handle_receive_vote() [START]");
         loop {
@@ -44,8 +53,9 @@ impl ProtocolVote {
                 vote
             );
             let vote_copy = (*vote).clone();
-            let nodes_count = 4;
-            self.state.write().unwrap().receive_vote(&vote_copy, nodes_count);
+            if self.state.write().unwrap().receive_vote(&vote_copy, self.nodes_count) {
+                self.p2p.broadcast(vote_copy).await?;
+            };
         }
     }
 }

script/research/validatord/validatord_config.toml

@@ -44,3 +44,6 @@ state = "~/.config/darkfi/validatord_state_0"
 # Node ID, used only for testing
 id = 0
 
+# Nodes count, used only for testing
+nodes = 1
+

src/consensus/state.rs

@@ -66,8 +66,12 @@ impl State {
 
     /// Node retreives a transaction and append it to the unconfirmed transactions list.
     /// Additional validity rules must be defined by the protocol for transactions.
-    pub fn append_tx(&mut self, tx: Tx) {
+    pub fn append_tx(&mut self, tx: Tx) -> bool {
+        if self.unconfirmed_txs.contains(&tx) {
+            return false
+        }
         self.unconfirmed_txs.push(tx);
+        true
     }
 
     /// Node calculates seconds until next epoch starting time.
@@ -269,14 +273,14 @@ impl State {
     /// nodes unconfirmed transactions list.
     /// Finally, we check if the notarization of the block can finalize parent blocks
     /// in its blockchain.
-    pub fn receive_vote(&mut self, vote: &Vote, nodes_count: usize) {
+    pub fn receive_vote(&mut self, vote: &Vote, nodes_count: usize) -> bool {
         let mut encoded_block = vec![];
         let result = vote.block.encode(&mut encoded_block);
         match result {
             Ok(_) => (),
             Err(e) => {
                 error!("Block encoding failed. Error: {:?}", e);
-                return
+                return false
             }
         };
         assert!(&vote.node_public_key.verify(&encoded_block[..], &vote.vote));
@@ -286,20 +290,23 @@ impl State {
             if !self.orphan_votes.contains(vote) {
                 self.orphan_votes.push(vote.clone());
             }
-            return
+            return false
         }
 
         let (unwrapped_vote_block, blockchain_index) = vote_block.unwrap();
         if !unwrapped_vote_block.metadata.sm.votes.contains(vote) {
             unwrapped_vote_block.metadata.sm.votes.push(vote.clone());
-        }
 
-        if !unwrapped_vote_block.metadata.sm.notarized &&
-            unwrapped_vote_block.metadata.sm.votes.len() > (2 * nodes_count / 3)
-        {
-            unwrapped_vote_block.metadata.sm.notarized = true;
-            self.check_blockchain_finalization(blockchain_index);
+            if !unwrapped_vote_block.metadata.sm.notarized &&
+                unwrapped_vote_block.metadata.sm.votes.len() > (2 * nodes_count / 3)
+            {
+                unwrapped_vote_block.metadata.sm.notarized = true;
+                self.check_blockchain_finalization(blockchain_index);
+            }
+
+            return true
         }
+        false
     }
 
     /// Node searches it the blockchains it holds for provided block.