/* This file is part of DarkFi (https://dark.fi)
*
* Copyright (C) 2020-2023 Dyne.org foundation
*
* This program is free software: you can redistribute it and/or modify
* it under the terms of the GNU Affero General Public License as
* published by the Free Software Foundation, either version 3 of the
* License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
* GNU Affero General Public License for more details.
*
* You should have received a copy of the GNU Affero General Public License
* along with this program. If not, see .
*/
use darkfi_sdk::{
blockchain::{expected_reward, PidOutput, PreviousSlot, Slot},
crypto::{schnorr::SchnorrSecret, MerkleNode, MerkleTree, SecretKey},
pasta::{group::ff::PrimeField, pallas},
};
use darkfi_serial::{async_trait, serialize, SerialDecodable, SerialEncodable};
use log::{error, info, warn};
use rand::rngs::OsRng;
use crate::{
blockchain::{BlockInfo, Blockchain, BlockchainOverlay, BlockchainOverlayPtr, Header},
tx::Transaction,
util::time::{TimeKeeper, Timestamp},
validator::{pid::slot_pid_output, verify_block, verify_transactions},
Error, Result,
};
/// Consensus configuration
const TXS_CAP: usize = 50;
/// This struct represents the information required by the consensus algorithm
pub struct Consensus {
/// Canonical (finalized) blockchain
pub blockchain: Blockchain,
/// Helper structure to calculate time related operations
pub time_keeper: TimeKeeper,
/// Node is participating to consensus
pub participating: bool,
/// Last slot node check for finalization
pub checked_finalization: u64,
/// Fork chains containing block proposals
pub forks: Vec,
/// Flag to enable testing mode
pub testing_mode: bool,
}
impl Consensus {
/// Generate a new Consensus state.
pub fn new(blockchain: Blockchain, time_keeper: TimeKeeper, testing_mode: bool) -> Self {
Self {
blockchain,
time_keeper,
participating: false,
checked_finalization: 0,
forks: vec![],
testing_mode,
}
}
/// Generate current hot/live slot for all current forks.
pub fn generate_slot(&mut self) -> Result<()> {
// Grab current slot id
let id = self.time_keeper.current_slot();
// If no forks exist, create a new one as a basis to extend
if self.forks.is_empty() {
self.forks.push(Fork::new(&self.blockchain)?);
}
// Grab previous slot information
let (producers, last_hashes, second_to_last_hashes) = self.previous_slot_info(id - 1)?;
for fork in self.forks.iter_mut() {
fork.generate_slot(id, producers, &last_hashes, &second_to_last_hashes)?;
}
Ok(())
}
/// Retrieve previous slot producers, last proposal hashes,
/// and their second to last hashes, from all current forks.
fn previous_slot_info(&self, slot: u64) -> Result<(u64, Vec, Vec)> {
let mut producers = 0;
let mut last_hashes = vec![];
let mut second_to_last_hashes = vec![];
for fork in &self.forks {
let last_proposal = fork.last_proposal()?;
if last_proposal.block.header.slot == slot {
producers += 1;
}
last_hashes.push(last_proposal.hash);
second_to_last_hashes.push(last_proposal.block.header.previous);
}
Ok((producers, last_hashes, second_to_last_hashes))
}
/// Generate a block proposal for the current hot/live(last) slot,
/// containing all pending transactions. Proposal extends the longest fork
/// chain the node is holding. This should only be called after
/// generate_slot(). Proposal is signed using provided secret key, which
/// must also have signed the provided proposal transaction.
pub async fn generate_proposal(
&self,
secret_key: SecretKey,
proposal_tx: Transaction,
) -> Result {
// Generate a time keeper for current slot
let time_keeper = self.time_keeper.current();
// Retrieve longest known fork
let mut fork_index = 0;
let mut max_fork_length = 0;
for (index, fork) in self.forks.iter().enumerate() {
if fork.proposals.len() > max_fork_length {
fork_index = index;
max_fork_length = fork.proposals.len();
}
}
let fork = &self.forks[fork_index];
// Grab forks' unproposed transactions
let mut unproposed_txs = fork.unproposed_txs(&self.blockchain, &time_keeper).await?;
unproposed_txs.push(proposal_tx);
// Calculate transactions tree root
let mut tree = MerkleTree::new(100);
// The following is pretty weird, so something better should be done.
for tx in &unproposed_txs {
let mut hash = [0_u8; 32];
hash[0..31].copy_from_slice(&blake3::hash(&serialize(tx)).as_bytes()[0..31]);
tree.append(MerkleNode::from(pallas::Base::from_repr(hash).unwrap()));
}
let root = tree.root(0).unwrap();
// Grab forks' last block proposal(previous)
let previous = fork.last_proposal()?;
// Generate the new header
let slot = fork.slots.last().unwrap();
// TODO: verify if header timestamp should be blockchain or system timestamp
let header = Header::new(
previous.block.blockhash(),
time_keeper.slot_epoch(slot.id),
slot.id,
Timestamp::current_time(),
root,
);
// TODO: sign more stuff?
// Sign block header using provided secret key
let signature = secret_key.sign(&mut OsRng, &header.headerhash()?.as_bytes()[..]);
// Generate the block and its proposal
let block =
BlockInfo::new(header, unproposed_txs, signature, slot.last_eta, fork.slots.clone());
let proposal = Proposal::new(block);
Ok(proposal)
}
/// Given a proposal, the node verifys it and finds which fork it extends.
/// If the proposal extends the canonical blockchain, a new fork chain is created.
/// A proposal is considered valid when the following rules apply:
/// 1. Node has not started current slot finalization
/// 2. Proposal refers to current slot
/// 3. Proposal hash matches the actual block one
/// 4. Block transactions don't exceed set limit
/// 5. If proposal extends a known fork, verify block slots
/// correspond to the fork hot/live ones
/// 6. Block is valid
/// Additional validity rules can be applied.
pub async fn append_proposal(&mut self, proposal: &Proposal) -> Result<()> {
// Generate a time keeper for current slot
let time_keeper = self.time_keeper.current();
// Node have already checked for finalization in this slot (1)
if time_keeper.verifying_slot <= self.checked_finalization {
warn!(target: "validator::consensus::append_proposal", "Proposal received after finalization sync period.");
return Err(Error::ProposalAfterFinalizationError)
}
// Proposal validations
let hdr = &proposal.block.header;
// Ignore proposal if not for current slot (2)
if hdr.slot != time_keeper.verifying_slot {
return Err(Error::ProposalNotForCurrentSlotError)
}
// Check if proposal hash matches actual one (3)
let proposal_hash = proposal.block.blockhash();
if proposal.hash != proposal_hash {
warn!(
target: "validator::consensus::append_proposal", "Received proposal contains mismatched hashes: {} - {}",
proposal.hash, proposal_hash
);
return Err(Error::ProposalHashesMissmatchError)
}
// TODO: verify if this should happen here or not.
// Check that proposal transactions don't exceed limit (4)
if proposal.block.txs.len() > TXS_CAP {
warn!(
target: "validator::consensus::append_proposal", "Received proposal transactions exceed configured cap: {} - {}",
proposal.block.txs.len(),
TXS_CAP
);
return Err(Error::ProposalTxsExceedCapError)
}
// Check if proposal extends any existing forks
let (mut fork, index) = self.find_extended_fork(proposal).await?;
// Verify block slots correspond to the forks' hot/live ones (5)
if !fork.slots.is_empty() && fork.slots != proposal.block.slots {
return Err(Error::ProposalContainsUnknownSlots)
}
// Insert last block slot so transactions can be validated against.
// Rest (empty) slots will be inserted along with the block.
// Since this fork uses an overlay clone, original overlay is not affected.
fork.overlay.lock().unwrap().slots.insert(&[proposal
.block
.slots
.last()
.unwrap()
.clone()])?;
// Grab overlay last block
let previous = fork.overlay.lock().unwrap().last_block()?;
// Retrieve expected reward
let expected_reward = expected_reward(time_keeper.verifying_slot);
// Verify proposal block (6)
if verify_block(
&fork.overlay,
&time_keeper,
&proposal.block,
&previous,
expected_reward,
self.testing_mode,
)
.await
.is_err()
{
error!(target: "validator::consensus::append_proposal", "Erroneous proposal block found");
fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
return Err(Error::BlockIsInvalid(proposal.hash.to_string()))
};
// If a fork index was found, replace forks with the mutated one,
// otherwise push the new fork.
fork.proposals.push(proposal.hash);
fork.slots = vec![];
match index {
Some(i) => {
self.forks[i] = fork;
}
None => {
self.forks.push(fork);
}
}
Ok(())
}
/// Given a proposal, find the index of the fork chain it extends, along with the specific
/// extended proposal index.
fn find_extended_fork_index(&self, proposal: &Proposal) -> Result<(usize, usize)> {
for (f_index, fork) in self.forks.iter().enumerate() {
// Traverse fork proposals sequence in reverse
for (p_index, p_hash) in fork.proposals.iter().enumerate().rev() {
if &proposal.block.header.previous == p_hash {
return Ok((f_index, p_index))
}
}
}
Err(Error::ExtendedChainIndexNotFound)
}
/// Given a proposal, find the fork chain it extends, and return its full clone.
/// If the proposal extends the fork not on its tail, a new fork is created and
/// we re-apply the proposals up to the extending one. If proposal extends canonical,
/// a new fork is created. Additionally, we return the fork index if a new fork
/// was not created, so caller can replace the fork.
async fn find_extended_fork(&self, proposal: &Proposal) -> Result<(Fork, Option)> {
// Check if proposal extends any fork
let found = self.find_extended_fork_index(proposal);
if found.is_err() {
// Check if we extend canonical
let (last_slot, last_block) = self.blockchain.last()?;
if proposal.block.header.previous != last_block ||
proposal.block.header.slot <= last_slot
{
return Err(Error::ExtendedChainIndexNotFound)
}
return Ok((Fork::new(&self.blockchain)?, None))
}
let (f_index, p_index) = found.unwrap();
let original_fork = &self.forks[f_index];
// Check if proposal extends fork at last proposal
if p_index == (original_fork.proposals.len() - 1) {
return Ok((original_fork.full_clone()?, Some(f_index)))
}
// Rebuild fork
let mut fork = Fork::new(&self.blockchain)?;
fork.proposals = original_fork.proposals[..p_index + 1].to_vec();
// Retrieve proposals blocks from original fork
let blocks = &original_fork.overlay.lock().unwrap().get_blocks_by_hash(&fork.proposals)?;
// Retrieve last block
let mut previous = &fork.overlay.lock().unwrap().last_block()?;
// Create a time keeper to validate each proposal block
let mut time_keeper = self.time_keeper.clone();
// Validate and insert each block
for block in blocks {
// Use block slot in time keeper
time_keeper.verifying_slot = block.header.slot;
// Retrieve expected reward
let expected_reward = expected_reward(time_keeper.verifying_slot);
// Verify block
if verify_block(
&fork.overlay,
&time_keeper,
block,
previous,
expected_reward,
self.testing_mode,
)
.await
.is_err()
{
error!(target: "validator::consensus::find_extended_fork_overlay", "Erroneous block found in set");
fork.overlay.lock().unwrap().overlay.lock().unwrap().purge_new_trees()?;
return Err(Error::BlockIsInvalid(block.blockhash().to_string()))
};
// Use last inserted block as next iteration previous
previous = block;
}
Ok((fork, None))
}
/// Node checks if any of the forks can be finalized.
/// Consensus finalization logic:
/// - If the node has observed the creation of a fork and no other forks exists at same or greater height,
/// all proposals in that fork can be finalized (append to canonical blockchain).
/// When a fork can be finalized, blocks(proposals) should be appended to canonical,
/// and forks should be removed.
pub async fn forks_finalization(&mut self) -> Result> {
let slot = self.time_keeper.current_slot();
info!(target: "validator::consensus::forks_finalization", "Started finalization check for slot: {}", slot);
// Set last slot finalization check occured to current slot
self.checked_finalization = slot;
// First we find longest fork without any other forks at same height
let mut fork_index = -1;
let mut max_length = 0;
for (index, fork) in self.forks.iter().enumerate() {
let length = fork.proposals.len();
// Check if less than max
if length < max_length {
continue
}
// Check if same length as max
if length == max_length {
// Setting fork_index so we know we have multiple
// forks at same length.
fork_index = -2;
continue
}
// Set fork as max
fork_index = index as i64;
max_length = length;
}
// Check if we found any fork to finalize
match fork_index {
-2 => {
info!(target: "validator::consensus::forks_finalization", "Eligible forks with same height exist, nothing to finalize.");
return Ok(vec![])
}
-1 => {
info!(target: "validator::consensus::forks_finalization", "Nothing to finalize.");
}
_ => {
info!(target: "validator::consensus::forks_finalization", "Fork {} can be finalized!", fork_index)
}
}
if max_length == 0 {
return Ok(vec![])
}
// Starting finalization
let fork = &self.forks[fork_index as usize];
let finalized = fork.overlay.lock().unwrap().get_blocks_by_hash(&fork.proposals)?;
info!(target: "validator::consensus::forks_finalization", "Finalized blocks: {}", finalized.len());
Ok(finalized)
}
}
/// This struct represents a block proposal, used for consensus.
#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
pub struct Proposal {
/// Block hash
pub hash: blake3::Hash,
/// Block data
pub block: BlockInfo,
}
impl Proposal {
pub fn new(block: BlockInfo) -> Self {
let hash = block.blockhash();
Self { hash, block }
}
}
impl From for BlockInfo {
fn from(proposal: Proposal) -> BlockInfo {
proposal.block
}
}
/// This struct represents a forked blockchain state, using an overlay over original
/// blockchain, containing all pending to-write records. Additionally, each fork
/// keeps a vector of valid pending transactions hashes, in order of receival, and
/// the proposals hashes sequence, for validations.
#[derive(Clone)]
pub struct Fork {
/// Overlay cache over canonical Blockchain
pub overlay: BlockchainOverlayPtr,
/// Fork proposal hashes sequence
pub proposals: Vec,
/// Hot/live slots
pub slots: Vec,
/// Valid pending transaction hashes
pub mempool: Vec,
}
impl Fork {
pub fn new(blockchain: &Blockchain) -> Result {
let mempool =
blockchain.get_pending_txs()?.iter().map(|tx| blake3::hash(&serialize(tx))).collect();
let overlay = BlockchainOverlay::new(blockchain)?;
Ok(Self { overlay, proposals: vec![], slots: vec![], mempool })
}
/// Auxiliary function to retrieve last proposal
pub fn last_proposal(&self) -> Result {
let block = if self.proposals.is_empty() {
self.overlay.lock().unwrap().last_block()?
} else {
self.overlay.lock().unwrap().get_blocks_by_hash(&[*self.proposals.last().unwrap()])?[0]
.clone()
};
Ok(Proposal::new(block))
}
/// Utility function to extract leader selection lottery randomness(eta),
/// defined as the hash of the last block, converted to pallas base.
fn get_last_eta(&self) -> Result {
// Retrieve last block(or proposal) hash
let hash = if self.proposals.is_empty() {
self.overlay.lock().unwrap().last_block()?.blockhash()
} else {
*self.proposals.last().unwrap()
};
// Read first 240 bits
let mut bytes: [u8; 32] = *hash.as_bytes();
bytes[30] = 0;
bytes[31] = 0;
Ok(pallas::Base::from_repr(bytes).unwrap())
}
/// Auxiliary function to retrieve unproposed valid transactions.
pub async fn unproposed_txs(
&self,
blockchain: &Blockchain,
time_keeper: &TimeKeeper,
) -> Result> {
// Retrieve all mempool transactions
let mut unproposed_txs: Vec = blockchain
.pending_txs
.get(&self.mempool, true)?
.iter()
.map(|x| x.clone().unwrap())
.collect();
// Iterate over fork proposals to find already proposed transactions
// and remove them from the unproposed_txs vector.
let proposals = self.overlay.lock().unwrap().get_blocks_by_hash(&self.proposals)?;
for proposal in proposals {
for tx in &proposal.txs {
unproposed_txs.retain(|x| x != tx);
}
}
// Check if transactions exceed configured cap
if unproposed_txs.len() > TXS_CAP {
unproposed_txs = unproposed_txs[0..TXS_CAP].to_vec()
}
// Clone forks' overlay
let overlay = self.overlay.lock().unwrap().full_clone()?;
// Verify transactions
let erroneous_txs = verify_transactions(&overlay, time_keeper, &unproposed_txs).await?;
if !erroneous_txs.is_empty() {
unproposed_txs.retain(|x| !erroneous_txs.contains(x));
}
Ok(unproposed_txs)
}
/// Generate current hot/live slot
pub fn generate_slot(
&mut self,
id: u64,
producers: u64,
last_hashes: &[blake3::Hash],
second_to_last_hashes: &[blake3::Hash],
) -> Result<()> {
// Grab last known fork slot
let previous_slot = if self.slots.is_empty() {
self.overlay.lock().unwrap().slots.get_last()?
} else {
self.slots.last().unwrap().clone()
};
// Generate previous slot information
let previous = PreviousSlot::new(
producers,
last_hashes.to_vec(),
second_to_last_hashes.to_vec(),
previous_slot.pid.error,
);
// Generate PID controller output
let (f, error, sigma1, sigma2) = slot_pid_output(&previous_slot, producers);
let pid = PidOutput::new(f, error, sigma1, sigma2);
// Each slot starts as an empty slot(not reward) when generated, carrying
// last eta
let last_eta = self.get_last_eta()?;
let total_tokens = previous_slot.total_tokens + previous_slot.reward;
let reward = 0;
let slot = Slot::new(id, previous, pid, last_eta, total_tokens, reward);
self.slots.push(slot);
Ok(())
}
/// Auxiliary function to create a full clone using BlockchainOverlay::full_clone.
/// Changes to this copy don't affect original fork overlay records, since underlying
/// overlay pointer have been updated to the cloned one.
pub fn full_clone(&self) -> Result {
let overlay = self.overlay.lock().unwrap().full_clone()?;
let proposals = self.proposals.clone();
let slots = self.slots.clone();
let mempool = self.mempool.clone();
Ok(Self { overlay, proposals, slots, mempool })
}
}