darkfi/src/consensus/task/proposal.rs

use std::time::Duration;

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

use super::consensus_sync_task;
use crate::{
    consensus::{Participant, ValidatorStatePtr},
    net::P2pPtr,
    util::sleep,
};

/// async task used for participating in the consensus protocol
pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: 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!("consensus: 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!("consensus: Waiting for next epoch ({:?} sec)", seconds_until_next_epoch);
    sleep(seconds_until_next_epoch.as_secs()).await;

    // Node syncs its consensus state
    if let Err(e) = consensus_sync_task(consensus_p2p.clone(), state.clone()).await {
        error!("Failed syncing consensus state: {}. Quitting consensus.", e);
        // TODO: Perhaps notify over a channel in order to
        // stop consensus p2p protocols.
        return
    };

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

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

    // Node modifies its participating epoch to next.
    match state.write().await.set_participating() {
        Ok(()) => info!("consensus: Node will start participating in the next epoch"),
        Err(e) => error!("Failed to set participation epoch: {}", e),
    }

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

        // Node refreshes participants records
        match state.write().await.refresh_participants() {
            Ok(()) => debug!("Participants refreshed successfully."),
            Err(e) => error!("Failed refreshing consensus 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)
        };

        let proposal = match result {
            Ok(prop) => {
                if prop.is_none() {
                    info!("consensus: Node is not the epoch lead");
                    continue
                }
                prop.unwrap()
            }
            Err(e) => {
                error!("consensus: Block proposal failed: {}", e);
                continue
            }
        };

        info!("consensus: Node is the epoch leader: Proposed block: {:?}", proposal);
        let vote = state.write().await.receive_proposal(&proposal);
        let vote = match vote {
            Ok(v) => {
                if v.is_none() {
                    debug!("proposal_task(): Node did not vote for the proposed block");
                    continue
                }
                v.unwrap()
            }
            Err(e) => {
                error!("consensus: Failed processing proposal: {}", e);
                continue
            }
        };

        let result = state.write().await.receive_vote(&vote).await;
        match result {
            Ok((_, to_broadcast)) => {
                info!("consensus: Vote saved successfully");
                // Broadcast finalized blocks info, if any:
                if let Some(blocks) = to_broadcast {
                    info!("consensus: Broadcasting finalized blocks");
                    for info in blocks {
                        match sync_p2p.broadcast(info).await {
                            Ok(()) => info!("consensus: Broadcasted block"),
                            Err(e) => error!("consensus: Failed broadcasting block: {}", e),
                        }
                    }
                } else {
                    info!("consensus: No finalized blocks to broadcast");
                }
            }
            Err(e) => {
                error!("consensus: Vote save failed: {}", e);
                // TODO: Is this fallthrough ok?
            }
        }

        // Broadcast block to other consensus nodes
        match consensus_p2p.broadcast(proposal).await {
            Ok(()) => info!("consensus: Proposal broadcasted successfully"),
            Err(e) => error!("consensus: Failed broadcasting proposal: {}", e),
        }

        // Broadcast leader vote
        match consensus_p2p.broadcast(vote).await {
            Ok(()) => info!("consensus: Leader vote broadcasted successfully"),
            Err(e) => error!("consensus: Failed broadcasting leader vote: {}", e),
        }
    }
}