darkfi/src/consensus2/task/proposal.rs

use std::time::Duration;

use log::{debug, error, info};

use super::consensus_sync_task;
use crate::{
    consensus2::{state::ValidatorStatePtr, Participant},
    net,
    util::async_util::sleep,
};

/// async task used for participating in the consensus protocol
pub async fn proposal_task(p2p: net::P2pPtr, state: ValidatorStatePtr) {
    // Node waits just before the current or next epoch end,
    // so it can start syncing latest state.
    let mut seconds_until_next_epoch = state.read().await.next_epoch_start();
    let one_sec = Duration::new(1, 0);
    loop {
        if seconds_until_next_epoch > one_sec {
            seconds_until_next_epoch -= one_sec;
            break
        }
        info!("Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
        sleep(seconds_until_next_epoch.as_secs()).await;
        seconds_until_next_epoch = state.read().await.next_epoch_start();
    }
    info!("Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
    sleep(seconds_until_next_epoch.as_secs()).await;

    // Node syncs its consensus state
    match consensus_sync_task(p2p.clone(), state.clone()).await {
        Ok(()) => {}
        Err(e) => {
            error!("Failed syncing consensus state: {}. Quitting consensus.", e);
            return
        }
    }

    // Node signals the network that it will start participating
    let participant = Participant::new(state.read().await.id, state.read().await.current_epoch());
    state.write().await.append_self_participant(participant.clone());

    match p2p.broadcast(participant).await {
        Ok(()) => info!("Consensus participation message broadcasted successfully."),
        Err(e) => error!("Failed broadcasting consensus participation: {}", e),
    }

    // After initialization node waits for next epoch to start participating
    let seconds_until_next_epoch = state.read().await.next_epoch_start().as_secs();
    info!(target: "consensus", "Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
    sleep(seconds_until_next_epoch).await;

    // Note modifies its participating flag to true.
    state.write().await.participating = true;

    loop {
        // Node refreshes participants records
        match state.write().await.refresh_participants() {
            Ok(()) => debug!("Participants refreshed successfully."),
            Err(e) => error!("Failed refreshing participants: {}", e),
        }

        // Node checks if it's the epoch leader to generate a new proposal
        // for that epoch.
        let result = if state.write().await.is_epoch_leader() {
            state.read().await.propose()
        } else {
            Ok(None)
        };

        match result {
            Ok(proposal) => {
                if proposal.is_none() {
                    info!(target: "consensus", "Node is not the epoch leader. Sleeping till next epoch...");
                } else {
                    // Leader creates a vote for the proposal and broadcasts them both
                    let proposal = proposal.unwrap();
                    info!(target: "consensus", "Node is the epoch leader: Proposed block: {:?}", proposal);
                    let vote = state.write().await.receive_proposal(&proposal);
                    match vote {
                        Ok(v) => {
                            if v.is_none() {
                                debug!("proposal_task(): Node did not vote for the proposed block");
                            } else {
                                let vote = v.unwrap();
                                let result = state.write().await.receive_vote(&vote);
                                match result {
                                    Ok(_) => info!(target: "consensus", "Vote saved successfully."),
                                    Err(e) => {
                                        error!(target: "consensus", "Vote save failed: {}", e)
                                    }
                                }

                                // Broadcast block
                                let result = p2p.broadcast(proposal).await;
                                match result {
                                    Ok(()) => {
                                        info!(target: "consensus", "Proposal broadcasted successfully.")
                                    }
                                    Err(e) => {
                                        error!(target: "consensus", "Failed broadcasting proposal: {}", e)
                                    }
                                }

                                // Broadcast leader vote
                                let result = p2p.broadcast(vote).await;
                                match result {
                                    Ok(()) => {
                                        info!(target: "consensus", "Leader vote broadcasted successfully.")
                                    }
                                    Err(e) => {
                                        error!(target: "consensus", "Failed broadcasting leader vote: {}", e)
                                    }
                                }
                            }
                        }
                        Err(e) => error!(target: "consensus", "Failed processing proposal: {}", e),
                    }
                }
            }
            Err(e) => error!("Block proposal failed: {}", e),
        }

        let seconds_until_next_epoch = state.read().await.next_epoch_start().as_secs();
        info!(target: "consensus", "Waiting for next epoch ({:?} sec)...", seconds_until_next_epoch);
        sleep(seconds_until_next_epoch).await;
    }
}