script/research: Added streamlet rust folder.

Folder contains the rust implementation of the structures required for the consensus protocol.
Protocol simulation is presented in lib.rs unit test.

script/research/streamlet_rust/Cargo.toml

@@ -0,0 +1,7 @@
+[package]
+name = "streamlet_rust"
+version = "0.1.0"
+edition = "2021"
+
+[dependencies]
+chrono = "0.4"

script/research/streamlet_rust/src/lib.rs

@@ -0,0 +1,144 @@
+pub mod structures;
+
+#[cfg(test)]
+mod tests {
+    use chrono::Utc;
+    use std::{thread, time};
+
+    use super::structures::{block::Block, node::Node};
+
+    #[test]
+    fn protocol_execution() {
+        // Genesis block is generated.
+        let mut genesis_block = Block::new(String::from("⊥"), 0, vec![String::from("⊥")]);
+        genesis_block.notarized = true;
+        genesis_block.finalized = true;
+
+        let genesis_time = Utc::now().timestamp();
+
+        // We create some nodes to participate in the Protocol.
+        let mut node0 =
+            Node::new(0, genesis_time, String::from("node_password0"), genesis_block.clone());
+        let mut node1 =
+            Node::new(1, genesis_time, String::from("node_password1"), genesis_block.clone());
+        let mut node2 =
+            Node::new(2, genesis_time, String::from("node_password2"), genesis_block.clone());
+
+        // We simulate some epochs to test consistency.
+        node0.receive_transaction(String::from("tx0"));
+        node0.broadcast_transaction(vec![&mut node1, &mut node2], String::from("tx0"));
+        node1.receive_transaction(String::from("tx1"));
+        node1.broadcast_transaction(vec![&mut node0, &mut node2], String::from("tx1"));
+        node2.receive_transaction(String::from("tx2"));
+        node2.broadcast_transaction(vec![&mut node0, &mut node1], String::from("tx2"));
+
+        // Each node checks if they are the epoch leader. Leader will propose the block.
+        let proposed_block = if node0.check_if_epoch_leader(3) {
+            node0.propose_block()
+        } else if node1.check_if_epoch_leader(3) {
+            node1.propose_block()
+        } else {
+            node2.propose_block()
+        };
+
+        // Leader broadcasts the proposed_block to rest nodes and they vote on it.
+        let node0_vote = node0.receive_proposed_block(&proposed_block).unwrap();
+        let node1_vote = node1.receive_proposed_block(&proposed_block).unwrap();
+        let node2_vote = node2.receive_proposed_block(&proposed_block).unwrap();
+
+        // Each node broadcasts its vote to rest nodes.
+        node0.receive_vote(&node0_vote, 3);
+        node0.receive_vote(&node1_vote, 3);
+        node0.receive_vote(&node2_vote, 3);
+        node1.receive_vote(&node0_vote, 3);
+        node1.receive_vote(&node1_vote, 3);
+        node1.receive_vote(&node2_vote, 3);
+        node2.receive_vote(&node0_vote, 3);
+        node2.receive_vote(&node1_vote, 3);
+        node2.receive_vote(&node2_vote, 3);
+
+        // We verify that all nodes have the same blockchain on round end.
+        verify_outputs(&node0, &node1, &node2);
+
+        // We use thread sleep to simulate sinchronization period.
+        thread::sleep(time::Duration::from_millis(5000));
+
+        node0.receive_transaction(String::from("tx3"));
+        node0.broadcast_transaction(vec![&mut node1, &mut node2], String::from("tx3"));
+        node1.receive_transaction(String::from("tx4"));
+        node1.broadcast_transaction(vec![&mut node0, &mut node2], String::from("tx4"));
+        node2.receive_transaction(String::from("tx5"));
+        node2.broadcast_transaction(vec![&mut node0, &mut node1], String::from("tx5"));
+
+        // Each node checks if they are the epoch leader. Leader will propose the block.
+        let proposed_block = if node0.check_if_epoch_leader(3) {
+            node0.propose_block()
+        } else if node1.check_if_epoch_leader(3) {
+            node1.propose_block()
+        } else {
+            node2.propose_block()
+        };
+
+        // Leader broadcasts the proposed_block to rest nodes and they vote on it.
+        let node0_vote = node0.receive_proposed_block(&proposed_block).unwrap();
+        let node1_vote = node1.receive_proposed_block(&proposed_block).unwrap();
+        let node2_vote = node2.receive_proposed_block(&proposed_block).unwrap();
+
+        // Each node broadcasts its vote to rest nodes.
+        node0.receive_vote(&node0_vote, 3);
+        node0.receive_vote(&node1_vote, 3);
+        node0.receive_vote(&node2_vote, 3);
+        node1.receive_vote(&node0_vote, 3);
+        node1.receive_vote(&node1_vote, 3);
+        node1.receive_vote(&node2_vote, 3);
+        node2.receive_vote(&node0_vote, 3);
+        node2.receive_vote(&node1_vote, 3);
+        node2.receive_vote(&node2_vote, 3);
+
+        // We verify that all nodes have the same blockchain on round end.
+        verify_outputs(&node0, &node1, &node2);
+
+        // We use thread sleep to simulate sinchronization period.
+        thread::sleep(time::Duration::from_millis(5000));
+
+        node0.receive_transaction(String::from("tx6"));
+        node0.broadcast_transaction(vec![&mut node1, &mut node2], String::from("tx6"));
+        node1.receive_transaction(String::from("tx7"));
+        node1.broadcast_transaction(vec![&mut node0, &mut node2], String::from("tx7"));
+        node2.receive_transaction(String::from("tx8"));
+        node2.broadcast_transaction(vec![&mut node0, &mut node1], String::from("tx8"));
+
+        // Each node checks if they are the epoch leader. Leader will propose the block.
+        let proposed_block = if node0.check_if_epoch_leader(3) {
+            node0.propose_block()
+        } else if node1.check_if_epoch_leader(3) {
+            node1.propose_block()
+        } else {
+            node2.propose_block()
+        };
+
+        // Leader broadcasts the proposed_block to rest nodes and they vote on it.
+        let node0_vote = node0.receive_proposed_block(&proposed_block).unwrap();
+        let node1_vote = node1.receive_proposed_block(&proposed_block).unwrap();
+        let node2_vote = node2.receive_proposed_block(&proposed_block).unwrap();
+
+        // Each node broadcasts its vote to rest nodes.
+        node0.receive_vote(&node0_vote, 3);
+        node0.receive_vote(&node1_vote, 3);
+        node0.receive_vote(&node2_vote, 3);
+        node1.receive_vote(&node0_vote, 3);
+        node1.receive_vote(&node1_vote, 3);
+        node1.receive_vote(&node2_vote, 3);
+        node2.receive_vote(&node0_vote, 3);
+        node2.receive_vote(&node1_vote, 3);
+        node2.receive_vote(&node2_vote, 3);
+
+        // We verify that all nodes have the same blockchain on round end.
+        verify_outputs(&node0, &node1, &node2);
+    }
+
+    fn verify_outputs(node0: &Node, node1: &Node, node2: &Node) {
+        assert!(node0.output() == node1.output());
+        assert!(node1.output() == node2.output());
+    }
+}

script/research/streamlet_rust/src/structures/block.rs

@@ -0,0 +1,39 @@
+use std::hash::{Hash, Hasher};
+
+use super::vote::Vote;
+
+/// This struct represents a tuple of the form (h, e, txs).
+/// Each blocks parent hash h may be computed simply as a hash of the parent block.
+#[derive(Debug, Clone)]
+pub struct Block {
+    /// parent hash
+    pub h: String,
+    /// epoch number
+    pub e: i64,
+    /// transactions payload
+    pub txs: Vec<String>,
+    /// Epoch votes
+    pub votes: Vec<Vote>,
+    /// block notarization flag
+    pub notarized: bool,
+    /// block finalization flag
+    pub finalized: bool,
+}
+
+impl Block {
+    pub fn new(h: String, e: i64, txs: Vec<String>) -> Block {
+        Block { h, e, txs, votes: Vec::new(), notarized: false, finalized: false }
+    }
+}
+
+impl PartialEq for Block {
+    fn eq(&self, other: &Self) -> bool {
+        self.h == other.h && self.e == other.e && self.txs == other.txs
+    }
+}
+
+impl Hash for Block {
+    fn hash<H: Hasher>(&self, hasher: &mut H) {
+        (&self.h, &self.e, &self.txs).hash(hasher);
+    }
+}

script/research/streamlet_rust/src/structures/blockchain.rs

@@ -0,0 +1,60 @@
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+};
+
+use super::block::Block;
+
+/// This struct represents a sequence of blocks starting with the genesis block.
+#[derive(Debug, Clone)]
+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.h != "⊥", "Genesis block provided.");
+        let mut hasher = DefaultHasher::new();
+        previous_block.hash(&mut hasher);
+        assert!(
+            block.h == hasher.finish().to_string() && block.e > previous_block.e,
+            "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.notarized {
+                return false
+            }
+        }
+        true
+    }
+}
+
+impl PartialEq for Blockchain {
+    fn eq(&self, other: &Self) -> bool {
+        self.blocks == other.blocks
+    }
+}

script/research/streamlet_rust/src/structures/mod.rs

@@ -0,0 +1,13 @@
+//! # Structures
+//!
+//! A library for modeling consensus algorithm structures.
+
+pub mod block;
+pub mod blockchain;
+pub mod node;
+pub mod vote;
+
+pub use block::Block;
+pub use blockchain::Blockchain;
+pub use node::Node;
+pub use vote::Vote;

script/research/streamlet_rust/src/structures/node.rs

@@ -0,0 +1,276 @@
+use chrono::Utc;
+use std::{
+    collections::hash_map::DefaultHasher,
+    hash::{Hash, Hasher},
+};
+
+use super::{block::Block, blockchain::Blockchain, vote::Vote};
+
+/// This struct represents a protocol 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(Debug)]
+pub struct Node {
+    pub id: u64,
+    pub genesis_time: i64,
+    pub password: String,
+    pub private_key: String,
+    pub public_key: String,
+    pub canonical_blockchain: Blockchain,
+    pub node_blockchains: Vec<Blockchain>,
+    pub unconfirmed_transactions: Vec<String>,
+}
+
+impl Node {
+    pub fn new(id: u64, genesis_time: i64, password: String, init_block: Block) -> Node {
+        // TODO: add keypair generation, clock sync
+        Node {
+            id,
+            genesis_time,
+            password,
+            private_key: String::from("private_key"),
+            public_key: String::from("public_key"),
+            canonical_blockchain: Blockchain::new(init_block),
+            node_blockchains: Vec::new(),
+            unconfirmed_transactions: Vec::new(),
+        }
+    }
+
+    /// A nodes output is the finalized (canonical) blockchain they hold.
+    pub fn output(&self) -> &Blockchain {
+        &self.canonical_blockchain
+    }
+
+    /// Node retreives a transaction and append it to the unconfirmed transactions list.
+    /// Additional validity rules must be defined by the protocol for its blockchain data structure.
+    pub fn receive_transaction(&mut self, transaction: String) {
+        self.unconfirmed_transactions.push(transaction);
+    }
+
+    /// Node broadcast a transaction to provided nodes list.
+    pub fn broadcast_transaction(&mut self, nodes: Vec<&mut Node>, transaction: String) {
+        for node in nodes {
+            node.receive_transaction(transaction.clone())
+        }
+    }
+
+    /// 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) -> i64 {
+        let delta = 2;
+        let current_time = Utc::now().timestamp();
+        ((current_time - self.genesis_time) % (2 * delta)) + 1
+    }
+
+    /// 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 for the current, containing all uncorfirmed transactions.
+    /// Block extends the longest notarized blockchain the node holds.
+    pub fn propose_block(&self) -> Block {
+        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 proposed_block =
+            Block::new(hasher.finish().to_string(), epoch, self.unconfirmed_transactions.clone());
+        // TODO: add block signing.
+        proposed_block
+    }
+
+    /// Node receives the proposed block, verifies its sender(epoch leader), and proceeds with voting on it.
+    pub fn receive_proposed_block(&mut self, proposed_block: &Block) -> Option<Vote> {
+        // TODO: verify leader keys.
+        self.vote_block(proposed_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) {
+            // TODO: add block signing.
+            return Some(Vote::new(String::from("signed_block"), block.clone(), 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.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() {
+            blockchain.blocks.last().unwrap().hash(&mut hasher);
+            if block.h == hasher.finish().to_string() &&
+                block.e > blockchain.blocks.last().unwrap().e
+            {
+                return index as i64
+            }
+        }
+
+        self.canonical_blockchain.blocks.last().unwrap().hash(&mut hasher);
+        if block.h != hasher.finish().to_string() ||
+            block.e <= self.canonical_blockchain.blocks.last().unwrap().e
+        {
+            panic!("Proposed block doesn't extend any known chains.");
+        }
+        -1
+    }
+
+    /// 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 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, vote: &Vote, nodes_count: usize) -> Option<Vote> {
+        // TODO: verify vote signature.
+        let vote_block = self.find_block(&vote.block);
+        if vote_block == None {
+            return self.vote_block(&vote.block)
+        }
+
+        let (unwrapped_vote_block, blockchain_index) = vote_block.unwrap();
+        if !unwrapped_vote_block.votes.contains(vote) {
+            unwrapped_vote_block.votes.push(vote.clone());
+        }
+
+        if !unwrapped_vote_block.notarized &&
+            unwrapped_vote_block.votes.len() > (2 * nodes_count / 3)
+        {
+            unwrapped_vote_block.notarized = true;
+
+            for transaction in unwrapped_vote_block.txs.clone() {
+                let txs_clone = transaction.clone();
+                if let Some(pos) =
+                    self.unconfirmed_transactions.iter().position(|txs| *txs == txs_clone)
+                {
+                    self.unconfirmed_transactions.remove(pos);
+                }
+            }
+
+            self.check_blockchain_finalization(blockchain_index);
+        }
+        None
+    }
+
+    /// 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.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.finalized = true;
+                    finalized_blocks.push(block.clone());
+                }
+                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.h != last_finalized_block_hash ||
+                        first_block.e <= last_finalized_block.e
+                    {
+                        dropped_blockchains.push(index);
+                    }
+                }
+                for index in dropped_blockchains {
+                    self.node_blockchains.remove(index);
+                }
+            }
+        }
+    }
+}

script/research/streamlet_rust/src/structures/vote.rs

@@ -0,0 +1,24 @@
+use super::block::Block;
+
+/// This struct represents a tuple of the form (vote, B, id).
+#[derive(Debug, Clone)]
+pub struct Vote {
+    /// signed block
+    pub vote: String,
+    /// block to vote on
+    pub block: Block,
+    /// node id
+    pub id: u64,
+}
+
+impl Vote {
+    pub fn new(vote: String, block: Block, id: u64) -> Vote {
+        Vote { vote, block, id }
+    }
+}
+
+impl PartialEq for Vote {
+    fn eq(&self, other: &Self) -> bool {
+        self.vote == other.vote && self.block == other.block && self.id == other.id
+    }
+}