script/research/consensusd: daemon prototype implementation

script/research/consensusd/Cargo.toml

@@ -0,0 +1,32 @@
+[package]
+name = "consensusd"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies.darkfi]
+path = "../../darkfi"
+features = ["crypto", "rpc"]
+
+[dependencies]
+chrono = "0.4.19"
+rand = "0.8.5"
+
+# Async
+smol = "1.2.5"
+async-std = "1.10.0"
+async-trait = "0.1.52"
+async-channel = "1.6.1"
+async-executor = "1.4.1"
+easy-parallel = "3.2.0"
+
+# Misc
+clap = {version = "3.0.7", features = ["derive"]}
+log = "0.4.14"
+num_cpus = "1.13.1"
+simplelog = "0.11.2"
+
+# Encoding and parsing
+serde = {version = "1.0.133", features = ["derive"]}
+serde_json = "1.0.74"
+
+[workspace]

script/research/consensusd/consensusd_config.toml

@@ -0,0 +1,25 @@
+## chaind configuration file
+##
+## Please make sure you go through all the settings so you can configure
+## your daemon properly.
+
+# The endpoint where chaind will bind its RPC socket
+rpc_listen_address = "127.0.0.1:9000"
+
+# Whether to listen with TLS or plain TCP
+serve_tls = false
+
+# Path to DER-formatted PKCS#12 archive. (Unused if serve_tls=false)
+# This can be created using openssl:
+# openssl pkcs12 -export -out chaind_identity.pfx -inkey key.pem -in cert.pem -certfile chain_certs.pem
+tls_identity_path = "~/.config/darkfi/consensusd_identity.pfx"
+
+# Password for the created TLS identity. (Unused if serve_tls=false)
+tls_identity_password = "FOOBAR"
+
+# Path to the state file 
+state_path = "~/.config/darkfi/consensusd_state_0"
+
+# Node ID, used only for testing
+id = 0
+

script/research/consensusd/src/lib.rs

@@ -0,0 +1 @@
+pub mod service;

script/research/consensusd/src/main.rs

@@ -0,0 +1,106 @@
+use std::{net::SocketAddr, path::PathBuf};
+
+use async_executor::Executor;
+use async_std::sync::Arc;
+use clap::{IntoApp, Parser};
+use easy_parallel::Parallel;
+use log::debug;
+use serde::{Deserialize, Serialize};
+use simplelog::{ColorChoice, TermLogger, TerminalMode};
+
+use darkfi::{
+    rpc::rpcserver::{listen_and_serve, RpcServerConfig},
+    util::{
+        cli::{log_config, spawn_config, Config},
+        expand_path, join_config_path,
+    },
+    Result,
+};
+
+use consensusd::service::ConsensusService;
+
+/// This struct represent the configuration parameters used by the Consensus daemon.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct ConsensusdConfig {
+    /// The endpoint where chaind will bind its RPC socket
+    pub rpc_listen_address: SocketAddr,
+    /// Whether to listen with TLS or plain TCP
+    pub serve_tls: bool,
+    /// Path to DER-formatted PKCS#12 archive. (Unused if serve_tls=false)
+    pub tls_identity_path: String,
+    /// Password for the TLS identity. (Unused if serve_tls=false)
+    pub tls_identity_password: String,
+    /// Path to the state file
+    pub state_path: String,
+    /// Node ID, used only for testing
+    pub id: u64,
+}
+
+/// Chaind cli configuration.
+#[derive(Parser)]
+#[clap(name = "consensusd")]
+pub struct CliConsensusd {
+    /// Sets a custom config file
+    #[clap(short, long)]
+    pub config: Option<String>,
+    /// Increase verbosity
+    #[clap(short, parse(from_occurrences))]
+    pub verbose: u8,
+}
+
+/// Consensus service initialization.
+async fn start(executor: Arc<Executor<'_>>, config: ConsensusdConfig) -> Result<()> {
+    let server_config = RpcServerConfig {
+        socket_addr: config.rpc_listen_address,
+        use_tls: config.serve_tls,
+        identity_path: expand_path(&config.clone().tls_identity_path)?,
+        identity_pass: config.tls_identity_password.clone(),
+    };
+
+    let state_path = expand_path(&config.state_path)?;
+    let id = config.id;
+
+    let chain_service = ConsensusService::new(id, state_path)?;
+
+    listen_and_serve(server_config, chain_service, executor).await
+}
+
+const CONFIG_FILE_CONTENTS: &[u8] = include_bytes!("../consensusd_config.toml");
+
+/// Consensus daemon initialization.
+#[async_std::main]
+async fn main() -> Result<()> {
+    let args = CliConsensusd::parse();
+    let matches = CliConsensusd::command().get_matches();
+
+    let config_path = if args.config.is_some() {
+        expand_path(&args.config.unwrap())?
+    } else {
+        join_config_path(&PathBuf::from("consensusd.toml"))?
+    };
+
+    spawn_config(&config_path, CONFIG_FILE_CONTENTS)?;
+    let verbosity_level = matches.occurrences_of("verbose");
+    let (lvl, conf) = log_config(verbosity_level)?;
+    TermLogger::init(lvl, conf, TerminalMode::Mixed, ColorChoice::Auto)?;
+    let config: ConsensusdConfig = Config::<ConsensusdConfig>::load(config_path)?;
+
+    let ex = Arc::new(Executor::new());
+    let (signal, shutdown) = async_channel::unbounded::<()>();
+    let ex2 = ex.clone();
+
+    let nthreads = num_cpus::get();
+    debug!(target: "CONSENSUS DAEMON", "Run {} executor threads", nthreads);
+    let (_, result) = Parallel::new()
+        .each(0..nthreads, |_| smol::future::block_on(ex.run(shutdown.recv())))
+        // Run the main future on the current thread.
+        .finish(|| {
+            smol::future::block_on(async move {
+                start(ex2, config).await?;
+                drop(signal);
+                Ok::<(), darkfi::Error>(())
+            })
+        });
+
+    result
+}

script/research/consensusd/src/service/block.rs

@@ -0,0 +1,57 @@
+use serde::{Deserialize, Serialize};
+use std::hash::{Hash, Hasher};
+
+use super::metadata::Metadata;
+
+// use darkfi::{tx::Transaction, util::serial::Encodable};
+use darkfi::util::serial::Encodable; // testing
+
+/// This struct represents a tuple of the form (st, sl, txs, metadata).
+/// Each blocks parent hash h may be computed simply as a hash of the parent block.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct Block {
+    /// Previous block hash
+    pub st: String,
+    /// Slot uid, generated by the beacon
+    pub sl: u64,
+    /// Transactions payload
+    pub txs: Vec<String>,
+    /// Additional block information
+    pub metadata: Metadata,
+}
+
+impl Block {
+    pub fn new(
+        st: String,
+        sl: u64,
+        txs: Vec<String>,
+        proof: String,
+        r: String,
+        s: String,
+    ) -> Block {
+        Block { st, sl, txs, metadata: Metadata::new(proof, r, s) }
+    }
+
+    pub fn signature_encode(&self) -> Vec<u8> {
+        let mut encoded_block = Vec::new();
+        let mut len = 0;
+        len += self.st.encode(&mut encoded_block).unwrap();
+        len += self.sl.encode(&mut encoded_block).unwrap();
+        // len += self.txs.encode(&mut encoded_block).unwrap();
+        len += String::from_iter(self.txs.clone()).encode(&mut encoded_block).unwrap(); // testing
+        assert_eq!(len, encoded_block.len());
+        encoded_block
+    }
+}
+
+impl PartialEq for Block {
+    fn eq(&self, other: &Self) -> bool {
+        self.st == other.st && self.sl == other.sl && self.txs == other.txs
+    }
+}
+
+impl Hash for Block {
+    fn hash<H: Hasher>(&self, hasher: &mut H) {
+        format!("{:?}{:?}{:?}", self.st, self.sl, self.txs).hash(hasher);
+    }
+}

script/research/consensusd/src/service/blockchain.rs

@@ -0,0 +1,56 @@
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+};
+
+use serde::{Deserialize, Serialize};
+
+use super::block::Block;
+
+/// This struct represents a sequence of blocks starting with the genesis block.
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct Blockchain {
+    pub blocks: Vec<Block>,
+}
+
+impl Blockchain {
+    pub fn new(intial_block: Block) -> Blockchain {
+        Blockchain { blocks: vec![intial_block] }
+    }
+
+    /// A block is considered valid when its parent hash is equal to the hash of the
+    /// previous block and their epochs are incremental, exluding genesis.
+    /// Additional validity rules can be applied.
+    pub fn check_block_validity(&self, block: &Block, previous_block: &Block) {
+        assert!(block.st != "⊥", "Genesis block provided.");
+        let mut hasher = DefaultHasher::new();
+        previous_block.hash(&mut hasher);
+        assert!(
+            block.st == hasher.finish().to_string() && block.sl > previous_block.sl,
+            "Provided block is invalid."
+        );
+    }
+
+    /// A blockchain is considered valid, when every block is valid, based on check_block_validity method.
+    pub fn check_chain_validity(&self) {
+        for (index, block) in self.blocks[1..].iter().enumerate() {
+            self.check_block_validity(&block, &self.blocks[index])
+        }
+    }
+
+    /// Insertion of a valid block.
+    pub fn add_block(&mut self, block: &Block) {
+        self.check_block_validity(&block, &self.blocks.last().unwrap());
+        self.blocks.push(block.clone());
+    }
+
+    /// Blockchain notarization check.
+    pub fn is_notarized(&self) -> bool {
+        for block in &self.blocks {
+            if !block.metadata.sm.notarized {
+                return false
+            }
+        }
+        true
+    }
+}

script/research/consensusd/src/service/consensus.rs

@@ -0,0 +1,241 @@
+use std::path::PathBuf;
+
+use async_executor::Executor;
+use async_std::sync::Arc;
+use async_trait::async_trait;
+use log::debug;
+use serde::Serialize;
+use serde_json::{json, Value};
+
+use darkfi::{
+    crypto::keypair::PublicKey,
+    rpc::{
+        jsonrpc,
+        jsonrpc::{response as jsonresp, ErrorCode::*, JsonRequest, JsonResult},
+        rpcserver::RequestHandler,
+    },
+    Result,
+};
+
+use super::{state::State, vote::Vote};
+
+/// This struct represent the Consensus service RPC daemon.
+#[derive(Serialize)]
+pub struct ConsensusService {
+    id: u64,
+    state_path: PathBuf,
+}
+
+impl ConsensusService {
+    pub fn new(id: u64, state_path: PathBuf) -> Result<Arc<ConsensusService>> {
+        match State::reset(id, &state_path) {
+            Err(e) => return Err(e),
+            _ => (),
+        }
+
+        Ok(Arc::new(ConsensusService { id, state_path }))
+    }
+
+    /// RPCAPI:
+    /// Hello world example.
+    /// --> {"jsonrpc": "2.0", "method": "say_hello", "params": [], "id": 0}
+    /// <-- {"jsonrpc": "2.0", "result": "hello world", "id": 0}
+    async fn say_hello(&self) -> JsonResult {
+        JsonResult::Resp(jsonresp(json!("hello world"), serde_json::to_value(self.id).unwrap()))
+    }
+
+    /// RPCAPI:
+    /// Node receives a transaction and stores it in its current state.
+    /// --> {"jsonrpc": "2.0", "method": "receive_tx", "params": ["tx"], "id": 0}
+    /// <-- {"jsonrpc": "2.0", "result": true, "id": 0}
+    async fn receive_tx(&self, params: Value) -> JsonResult {
+        let args = params.as_array().unwrap();
+
+        if args.len() != 1 {
+            return jsonrpc::error(InvalidParams, None, serde_json::to_value(self.id).unwrap())
+                .into()
+        }
+
+        let mut state = State::load_current_state(self.id, &self.state_path).unwrap();
+        let tx = String::from(args[0].as_str().unwrap());
+
+        let result = || -> Result<()> {
+            state.append_tx(tx);
+            state.save(&self.state_path)?;
+            Ok(())
+        };
+
+        match result() {
+            Ok(()) => {
+                JsonResult::Resp(jsonresp(json!(true), serde_json::to_value(self.id).unwrap()))
+            }
+            Err(e) => jsonrpc::error(
+                ServerError(-32603),
+                Some(e.to_string()),
+                serde_json::to_value(self.id).unwrap(),
+            )
+            .into(),
+        }
+    }
+
+    /// RPCAPI:
+    /// Node checks if its the current slot leader and generates the slot Block (represented as a Vote structure).
+    /// --> {"jsonrpc": "2.0", "method": "consensus_task", "params": [1], "id": 0}
+    /// <-- {"jsonrpc": "2.0", "result": [PublicKey, Vote], "id": 0}
+    /// TODO: 1, This should be an scheduled task.
+    ///       2. Nodes count not from request.
+    ///       3. Proposed block broadcast.
+    async fn consensus_task(&self, params: Value) -> JsonResult {
+        let args = params.as_array().unwrap();
+
+        if args.len() != 1 {
+            return jsonrpc::error(InvalidParams, None, serde_json::to_value(self.id).unwrap())
+                .into()
+        }
+
+        let state = State::load_current_state(self.id, &self.state_path).unwrap();
+        let nodes_count = args[0].as_u64().unwrap();
+
+        let result = || -> Result<_> {
+            let proposed_block =
+                if state.check_if_epoch_leader(nodes_count) { state.propose_block() } else { None };
+            Ok(proposed_block)
+        };
+
+        match result() {
+            Ok(x) => {
+                if x.is_none() {
+                    JsonResult::Resp(jsonresp(
+                        json!("Node is not the epoch leader"),
+                        serde_json::to_value(self.id).unwrap(),
+                    ))
+                } else {
+                    // TODO: Proposed block broadcast.
+                    JsonResult::Resp(jsonresp(
+                        json!((state.public_key, x)),
+                        serde_json::to_value(self.id).unwrap(),
+                    ))
+                }
+            }
+            Err(e) => jsonrpc::error(
+                ServerError(-32603),
+                Some(e.to_string()),
+                serde_json::to_value(self.id).unwrap(),
+            )
+            .into(),
+        }
+    }
+
+    /// RPCAPI:
+    /// Node receives a proposed block, verifies it and stores it in its current state.
+    /// --> {"jsonrpc": "2.0", "method": "receive_proposed_block", "params": [PublicKey, Vote, 1], "id": 0}
+    /// <-- {"jsonrpc": "2.0", "result": [PublicKey, Vote], "id": 0}
+    /// TODO: 1. Nodes count not from request.
+    ///       2. Vote broadcast.
+    async fn receive_proposed_block(&self, params: Value) -> JsonResult {
+        let args = params.as_array().unwrap();
+
+        if args.len() != 3 {
+            return jsonrpc::error(InvalidParams, None, serde_json::to_value(self.id).unwrap())
+                .into()
+        }
+
+        let mut state = State::load_current_state(self.id, &self.state_path).unwrap();
+        let proposer_public_key: PublicKey = serde_json::from_value(args[0].clone()).unwrap();
+        let proposed_block: Vote = serde_json::from_value(args[1].clone()).unwrap();
+        let nodes_count = args[2].as_u64().unwrap();
+
+        let mut result = || -> Result<_> {
+            let vote =
+                state.receive_proposed_block(&proposer_public_key, &proposed_block, nodes_count);
+            if vote.is_some() {
+                state.save(&self.state_path)?;
+            }
+            Ok(vote)
+        };
+
+        match result() {
+            Ok(x) => {
+                if x.is_none() {
+                    JsonResult::Resp(jsonresp(
+                        json!("Node did not vote for the proposed block."),
+                        serde_json::to_value(self.id).unwrap(),
+                    ))
+                } else {
+                    // TODO: Vote broadcast.
+                    JsonResult::Resp(jsonresp(
+                        json!((state.public_key, x)),
+                        serde_json::to_value(self.id).unwrap(),
+                    ))
+                }
+            }
+            Err(e) => jsonrpc::error(
+                ServerError(-32603),
+                Some(e.to_string()),
+                serde_json::to_value(self.id).unwrap(),
+            )
+            .into(),
+        }
+    }
+
+    /// RPCAPI:
+    /// Node receives a block vote and perform the consensus protocol corresponding functions, based on its current state.
+    /// --> {"jsonrpc": "2.0", "method": "receive_vote", "params": [PublicKey, Vote, 1], "id": 0}
+    /// <-- {"jsonrpc": "2.0", "result": true, "id": 0}
+    /// TODO: 1. Nodes count not from request.
+    async fn receive_vote(&self, params: Value) -> JsonResult {
+        let args = params.as_array().unwrap();
+
+        if args.len() != 3 {
+            return jsonrpc::error(InvalidParams, None, serde_json::to_value(self.id).unwrap())
+                .into()
+        }
+
+        let mut state = State::load_current_state(self.id, &self.state_path).unwrap();
+        let voter_public_key: PublicKey = serde_json::from_value(args[0].clone()).unwrap();
+        let vote: Vote = serde_json::from_value(args[1].clone()).unwrap();
+        let nodes_count = args[2].as_u64().unwrap() as usize;
+
+        let mut result = || -> Result<()> {
+            state.receive_vote(&voter_public_key, &vote, nodes_count);
+            state.save(&self.state_path)?;
+            Ok(())
+        };
+
+        match result() {
+            Ok(()) => {
+                JsonResult::Resp(jsonresp(json!(true), serde_json::to_value(self.id).unwrap()))
+            }
+            Err(e) => jsonrpc::error(
+                ServerError(-32603),
+                Some(e.to_string()),
+                serde_json::to_value(self.id).unwrap(),
+            )
+            .into(),
+        }
+    }
+}
+
+#[async_trait]
+impl RequestHandler for ConsensusService {
+    /// RPC methods configuration.
+    async fn handle_request(&self, req: JsonRequest, _executor: Arc<Executor<'_>>) -> JsonResult {
+        if req.id.as_u64().unwrap() != self.id || req.params.as_array().is_none() {
+            return jsonrpc::error(InvalidParams, None, serde_json::to_value(self.id).unwrap())
+                .into()
+        }
+
+        debug!(target: "RPC", "--> {}", serde_json::to_string(&req).unwrap());
+
+        return match req.method.as_str() {
+            Some("say_hello") => self.say_hello().await,
+            Some("receive_tx") => self.receive_tx(req.params).await,
+            Some("consensus_task") => self.consensus_task(req.params).await,
+            Some("receive_proposed_block") => self.receive_proposed_block(req.params).await,
+            Some("receive_vote") => self.receive_vote(req.params).await,
+            Some(_) | None => {
+                jsonrpc::error(MethodNotFound, None, serde_json::to_value(self.id).unwrap()).into()
+            }
+        }
+    }
+}

script/research/consensusd/src/service/metadata.rs

@@ -0,0 +1,61 @@
+use serde::{Deserialize, Serialize};
+
+use super::{
+    util::{get_current_time, Timestamp},
+    vote::Vote,
+};
+
+/// This struct represents additional Block information used by the consensus protocol.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct Metadata {
+    /// Block information used by Ouroboros consensus
+    pub om: OuroborosMetadata,
+    /// Block information used by Streamlet consensus
+    pub sm: StreamletMetadata,
+    /// Block creation timestamp
+    pub timestamp: Timestamp,
+}
+
+impl Metadata {
+    pub fn new(proof: String, r: String, s: String) -> Metadata {
+        Metadata {
+            om: OuroborosMetadata::new(proof, r, s),
+            sm: StreamletMetadata::new(),
+            timestamp: get_current_time(),
+        }
+    }
+}
+
+/// This struct represents Block information used by Ouroboros consensus protocol.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct OuroborosMetadata {
+    /// Proof the stakeholder is the block owner
+    pub proof: String,
+    /// Random seed for VRF
+    pub r: String,
+    /// Block owner signature
+    pub s: String,
+}
+
+impl OuroborosMetadata {
+    pub fn new(proof: String, r: String, s: String) -> OuroborosMetadata {
+        OuroborosMetadata { proof, r, s }
+    }
+}
+
+/// This struct represents Block information used by Streamlet consensus protocol.
+#[derive(Debug, Clone, Deserialize, Serialize)]
+pub struct StreamletMetadata {
+    /// Epoch votes
+    pub votes: Vec<Vote>,
+    /// Block notarization flag
+    pub notarized: bool,
+    /// Block finalization flag
+    pub finalized: bool,
+}
+
+impl StreamletMetadata {
+    pub fn new() -> StreamletMetadata {
+        StreamletMetadata { votes: Vec::new(), notarized: false, finalized: false }
+    }
+}

script/research/consensusd/src/service/mod.rs

@@ -0,0 +1,14 @@
+pub mod block;
+pub mod blockchain;
+pub mod consensus;
+pub mod metadata;
+pub mod state;
+pub mod util;
+pub mod vote;
+
+pub use block::Block;
+pub use blockchain::Blockchain;
+pub use consensus::ConsensusService;
+pub use metadata::Metadata;
+pub use state::State;
+pub use vote::Vote;

script/research/consensusd/src/service/state.rs

@@ -0,0 +1,342 @@
+use serde::{Deserialize, Serialize};
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+    path::PathBuf,
+};
+
+use super::{
+    block::Block,
+    blockchain::Blockchain,
+    util::{get_current_time, load, save, Timestamp},
+    vote::Vote,
+};
+
+use darkfi::{
+    crypto::{
+        keypair::{PublicKey, SecretKey},
+        schnorr::{SchnorrPublic, SchnorrSecret},
+    },
+    Result,
+};
+use rand::rngs::OsRng;
+
+/// This struct represents the state of a consensus node.
+/// Each node is numbered and has a secret-public keys pair, to sign messages.
+/// Nodes hold a set of Blockchains(some of which are not notarized)
+/// and a set of unconfirmed pending transactions.
+#[derive(Deserialize, Serialize)]
+pub struct State {
+    pub id: u64,
+    pub genesis_time: Timestamp,
+    pub secret_key: SecretKey,
+    pub public_key: PublicKey,
+    pub canonical_blockchain: Blockchain,
+    pub node_blockchains: Vec<Blockchain>,
+    pub unconfirmed_txs: Vec<String>,
+}
+
+impl State {
+    pub fn new(id: u64, genesis_time: Timestamp, init_block: Block) -> State {
+        // TODO: clock sync
+        let secret = SecretKey::random(&mut OsRng);
+        State {
+            id,
+            genesis_time,
+            secret_key: secret,
+            public_key: PublicKey::from_secret(secret),
+            canonical_blockchain: Blockchain::new(init_block),
+            node_blockchains: Vec::new(),
+            unconfirmed_txs: Vec::new(),
+        }
+    }
+
+    /// 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: String) {
+        self.unconfirmed_txs.push(tx);
+    }
+
+    /// Node calculates current epoch, based on elapsed time from the genesis block.
+    /// Epochs duration is configured using the delta value.
+    pub fn get_current_epoch(&self) -> u64 {
+        let delta = 10;
+        self.genesis_time.clone().elapsed() / (2 * delta)
+    }
+
+    /// Node finds epochs leader, using a simple hash method.
+    /// Leader calculation is based on how many nodes are participating in the network.
+    pub fn get_epoch_leader(&self, nodes_count: u64) -> u64 {
+        let epoch = self.get_current_epoch();
+        let mut hasher = DefaultHasher::new();
+        epoch.hash(&mut hasher);
+        hasher.finish() % nodes_count
+    }
+
+    /// Node checks if they are the current epoch leader.
+    pub fn check_if_epoch_leader(&self, nodes_count: u64) -> bool {
+        let leader = self.get_epoch_leader(nodes_count);
+        self.id == leader
+    }
+
+    /// Node generates a block proposal(mapped as Vote) for the current epoch,
+    /// containing all uncorfirmed transactions.
+    /// Block extends the longest notarized blockchain the node holds.
+    pub fn propose_block(&self) -> Option<Vote> {
+        let epoch = self.get_current_epoch();
+        let longest_notarized_chain = self.find_longest_notarized_chain();
+        let mut hasher = DefaultHasher::new();
+        longest_notarized_chain.blocks.last().unwrap().hash(&mut hasher);
+        let unproposed_txs = self.get_unproposed_txs();
+        let proposed_block = Block::new(
+            hasher.finish().to_string(),
+            epoch,
+            unproposed_txs,
+            String::from("proof"),
+            String::from("r"),
+            String::from("s"),
+        );
+        let signed_block = self.secret_key.sign(&proposed_block.signature_encode());
+        Some(Vote::new(signed_block, proposed_block, self.id))
+    }
+
+    /// Node retrieves all unconfiremd transactions not proposed in previous blocks.
+    pub fn get_unproposed_txs(&self) -> Vec<String> {
+        let mut unproposed_txs = self.unconfirmed_txs.clone();
+        for blockchain in &self.node_blockchains {
+            for block in &blockchain.blocks {
+                for tx in &block.txs {
+                    if let Some(pos) = unproposed_txs.iter().position(|txs| *txs == *tx) {
+                        unproposed_txs.remove(pos);
+                    }
+                }
+            }
+        }
+        unproposed_txs
+    }
+
+    /// Finds the longest fully notarized blockchain the node holds.
+    pub fn find_longest_notarized_chain(&self) -> &Blockchain {
+        let mut longest_notarized_chain = &self.canonical_blockchain;
+        let mut length = 0;
+        for blockchain in &self.node_blockchains {
+            if blockchain.is_notarized() && blockchain.blocks.len() > length {
+                length = blockchain.blocks.len();
+                longest_notarized_chain = &blockchain;
+            }
+        }
+        &longest_notarized_chain
+    }
+
+    /// Node receives the proposed block(mapped as Vote), verifies its sender(epoch leader),
+    /// and proceeds with voting on it.
+    pub fn receive_proposed_block(
+        &mut self,
+        leader_public_key: &PublicKey,
+        proposed_block_vote: &Vote,
+        nodes_count: u64,
+    ) -> Option<Vote> {
+        assert!(self.get_epoch_leader(nodes_count) == proposed_block_vote.id);
+        assert!(leader_public_key
+            .verify(&proposed_block_vote.block.signature_encode(), &proposed_block_vote.vote));
+        self.vote_block(&proposed_block_vote.block)
+    }
+
+    /// Given a block, node finds which blockchain it extends.
+    /// If block extends the canonical blockchain, a new fork blockchain is created.
+    /// Node votes on the block, only if it extends the longest notarized chain it has seen.
+    pub fn vote_block(&mut self, block: &Block) -> Option<Vote> {
+        let index = self.find_extended_blockchain_index(block);
+
+        let blockchain = if index == -1 {
+            let blockchain = Blockchain::new(block.clone());
+            self.node_blockchains.push(blockchain);
+            self.node_blockchains.last().unwrap()
+        } else {
+            self.node_blockchains[index as usize].add_block(&block);
+            &self.node_blockchains[index as usize]
+        };
+
+        if self.extends_notarized_blockchain(blockchain) {
+            let block_copy = block.clone();
+            let signed_block = self.secret_key.sign(&block_copy.signature_encode());
+            return Some(Vote::new(signed_block, block_copy, self.id))
+        }
+        None
+    }
+
+    /// Node verifies if provided blockchain is notarized excluding the last block.
+    pub fn extends_notarized_blockchain(&self, blockchain: &Blockchain) -> bool {
+        for block in &blockchain.blocks[..(blockchain.blocks.len() - 1)] {
+            if !block.metadata.sm.notarized {
+                return false
+            }
+        }
+        true
+    }
+
+    /// Given a block, node finds the index of the blockchain it extends.
+    pub fn find_extended_blockchain_index(&self, block: &Block) -> i64 {
+        let mut hasher = DefaultHasher::new();
+        for (index, blockchain) in self.node_blockchains.iter().enumerate() {
+            let last_block = blockchain.blocks.last().unwrap();
+            last_block.hash(&mut hasher);
+            if block.st == hasher.finish().to_string() && block.sl > last_block.sl {
+                return index as i64
+            }
+        }
+
+        let last_block = self.canonical_blockchain.blocks.last().unwrap();
+        last_block.hash(&mut hasher);
+        if block.st != hasher.finish().to_string() || block.sl <= last_block.sl {
+            panic!("Proposed block doesn't extend any known chains.");
+        }
+        -1
+    }
+
+    /// Node receives a vote for a block.
+    /// First, sender is verified using their public key.
+    /// Block is searched in nodes blockchains.
+    /// If the vote wasn't received before, it is appended to block votes list.
+    /// When a node sees 2n/3 votes for a block it notarizes it.
+    /// When a block gets notarized, the transactions it contains are removed from
+    /// 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, node_public_key: &PublicKey, vote: &Vote, nodes_count: usize) {
+        assert!(node_public_key.verify(&vote.block.signature_encode(), &vote.vote));
+        let vote_block = self.find_block(&vote.block);
+        if vote_block == None {
+            panic!("Received vote for unknown block.");
+        }
+
+        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);
+        }
+    }
+
+    /// Node searches it the blockchains it holds for provided block.
+    pub fn find_block(&mut self, vote_block: &Block) -> Option<(&mut Block, i64)> {
+        for (index, blockchain) in &mut self.node_blockchains.iter_mut().enumerate() {
+            for block in blockchain.blocks.iter_mut().rev() {
+                if vote_block == block {
+                    return Some((block, index as i64))
+                }
+            }
+        }
+
+        for block in &mut self.canonical_blockchain.blocks.iter_mut().rev() {
+            if vote_block == block {
+                return Some((block, -1))
+            }
+        }
+        None
+    }
+
+    /// Node checks if the index blockchain can be finalized.
+    /// Consensus finalization logic: If node has observed the notarization of 3 consecutive
+    /// blocks in a fork chain, it finalizes (appends to canonical blockchain) all blocks up to the middle block.
+    /// When fork chain blocks are finalized, rest fork chains not starting by those blocks are removed.
+    pub fn check_blockchain_finalization(&mut self, blockchain_index: i64) {
+        let blockchain = if blockchain_index == -1 {
+            &mut self.canonical_blockchain
+        } else {
+            &mut self.node_blockchains[blockchain_index as usize]
+        };
+
+        let blockchain_len = blockchain.blocks.len();
+        if blockchain_len > 2 {
+            let mut consecutive_notarized = 0;
+            for block in &blockchain.blocks {
+                if block.metadata.sm.notarized {
+                    consecutive_notarized = consecutive_notarized + 1;
+                } else {
+                    break
+                }
+            }
+
+            if consecutive_notarized > 2 {
+                let mut finalized_blocks = Vec::new();
+                for block in &mut blockchain.blocks[..(consecutive_notarized - 1)] {
+                    block.metadata.sm.finalized = true;
+                    finalized_blocks.push(block.clone());
+                    for tx in block.txs.clone() {
+                        if let Some(pos) = self.unconfirmed_txs.iter().position(|txs| *txs == tx) {
+                            self.unconfirmed_txs.remove(pos);
+                        }
+                    }
+                }
+                blockchain.blocks.drain(0..(consecutive_notarized - 1));
+                for block in &finalized_blocks {
+                    self.canonical_blockchain.blocks.push(block.clone());
+                }
+
+                let mut hasher = DefaultHasher::new();
+                let last_finalized_block = self.canonical_blockchain.blocks.last().unwrap();
+                last_finalized_block.hash(&mut hasher);
+                let last_finalized_block_hash = hasher.finish().to_string();
+                let mut dropped_blockchains = Vec::new();
+                for (index, blockchain) in self.node_blockchains.iter().enumerate() {
+                    let first_block = blockchain.blocks.first().unwrap();
+                    if first_block.st != last_finalized_block_hash ||
+                        first_block.sl <= last_finalized_block.sl
+                    {
+                        dropped_blockchains.push(index);
+                    }
+                }
+                for index in dropped_blockchains {
+                    self.node_blockchains.remove(index);
+                }
+            }
+        }
+    }
+
+    /// Util function to save the current node state to provided file path.
+    pub fn save(&self, path: &PathBuf) -> Result<()> {
+        save::<Self>(path, self)
+    }
+
+    /// Util function to load current node state by the provided file path.
+    //  If file is not found, node state is reset.
+    pub fn load_or_create(id: u64, path: &PathBuf) -> Result<Self> {
+        match load::<Self>(path) {
+            Ok(state) => Ok(state),
+            Err(_) => return Self::reset(id, path),
+        }
+    }
+
+    /// Util function to load the current node state by the provided file path.
+    pub fn load_current_state(id: u64, path: &PathBuf) -> Result<State> {
+        let state = Self::load_or_create(id, path)?;
+        Ok(state)
+    }
+
+    /// Util function to reset node state.
+    pub fn reset(id: u64, path: &PathBuf) -> Result<State> {
+        // Genesis block is generated.
+        let mut genesis_block = Block::new(
+            String::from("⊥"),
+            0,
+            vec![],
+            String::from("proof"),
+            String::from("r"),
+            String::from("s"),
+        );
+        genesis_block.metadata.sm.notarized = true;
+        genesis_block.metadata.sm.finalized = true;
+
+        let genesis_time = get_current_time();
+
+        let state = Self::new(id, genesis_time, genesis_block.clone());
+        state.save(path)?;
+        return Ok(state)
+    }
+}

script/research/consensusd/src/service/util.rs

@@ -0,0 +1,40 @@
+use chrono::{NaiveDateTime, Utc};
+use serde::{de::DeserializeOwned, Deserialize, Serialize};
+use std::{fs::File, io::BufReader, path::PathBuf};
+
+use darkfi::Result;
+
+/// Util function to load a structure saved as a JSON in the provided path file, using serde crate.
+pub fn load<T: DeserializeOwned>(path: &PathBuf) -> Result<T> {
+    let file = File::open(path)?;
+    let reader = BufReader::new(file);
+
+    let value: T = serde_json::from_reader(reader)?;
+    Ok(value)
+}
+
+/// Util function to save a structure as a JSON in the provided path file, using serde crate.
+pub fn save<T: Serialize>(path: &PathBuf, value: &T) -> Result<()> {
+    let file = File::create(path)?;
+    serde_json::to_writer_pretty(file, value)?;
+    Ok(())
+}
+
+/// Util structure to represend chrono UTC timestamps.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Timestamp(pub i64);
+
+impl Timestamp {
+    /// Calculates elapsed time of a Timestamp.
+    pub fn elapsed(self) -> u64 {
+        let start_time = NaiveDateTime::from_timestamp(self.0, 0);
+        let end_time = NaiveDateTime::from_timestamp(Utc::now().timestamp(), 0);
+        let diff = end_time - start_time;
+        diff.num_seconds().try_into().unwrap()
+    }
+}
+
+/// Util function to generate a Timestamp of current time.
+pub fn get_current_time() -> Timestamp {
+    Timestamp(Utc::now().timestamp())
+}

script/research/consensusd/src/service/vote.rs

@@ -0,0 +1,21 @@
+use serde::{Deserialize, Serialize};
+
+use super::block::Block;
+use darkfi::crypto::schnorr::Signature;
+
+/// This struct represents a tuple of the form (vote, B, id).
+#[derive(Debug, Clone, PartialEq, Deserialize, Serialize)]
+pub struct Vote {
+    /// signed block
+    pub vote: Signature,
+    /// block to vote on
+    pub block: Block,
+    /// node id
+    pub id: u64,
+}
+
+impl Vote {
+    pub fn new(vote: Signature, block: Block, id: u64) -> Vote {
+        Vote { vote, block, id }
+    }
+}