use std::time::Duration; use log::{debug, error, info}; use super::consensus_sync_task; use crate::{ consensus::{Participant, ValidatorStatePtr}, net::P2pPtr, util::async_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_n_epoch_start(1); 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_n_epoch_start(1); } 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!("consensus: Failed syncing consensus state: {}. Quitting consensus.", e); // TODO: Perhaps notify over a channel in order to // stop consensus p2p protocols. return }; // Node modifies its participating slot to next. match state.write().await.set_participating() { Ok(()) => info!("consensus: Node will start participating in the next slot"), Err(e) => error!("consensus: Failed to set participation slot: {}", e), } loop { let seconds_next_slot = state.read().await.next_n_slot_start(1).as_secs(); info!("consensus: Waiting for next slot ({} sec)", seconds_next_slot); sleep(seconds_next_slot).await; // Node checks if epoch has changed, to broadcast a new participation message match state.write().await.epoch_changed().await { Ok(changed) => { if changed { let lock = state.read().await; info!("consensus: New epoch started: {}", lock.current_epoch()); let public = lock.public; let address = lock.address; let mut coins = vec![]; for slot_coins in &lock.consensus.coins { let mut slot_coins_inputs = vec![]; for slot_coin in slot_coins { slot_coins_inputs.push(slot_coin.public_inputs()); } coins.push(slot_coins_inputs); } let participant = Participant::new(public, address, coins); state.write().await.append_participant(participant.clone()); match consensus_p2p.broadcast(participant).await { Ok(()) => { info!("consensus: Participation message broadcasted successfully.") } Err(e) => { error!("consensus: Failed broadcasting consensus participation: {}", e) } } // Node sleeps 2 seconds so all nodes can have the new epoch participants //TODO: optimize this sleep(2).await; } } Err(e) => { error!("consensus: Epoch check failed: {}", e); continue } }; // Node checks if it's the slot leader to generate a new proposal // for that slot. let lock = state.read().await; let (won, idx) = lock.is_slot_leader(); let result = if won { lock.propose(idx) } else { Ok(None) }; let proposal = match result { Ok(prop) => { if prop.is_none() { info!("consensus: Node is not the slot lead"); continue } prop.unwrap() } Err(e) => { error!("consensus: Block proposal failed: {}", e); continue } }; info!("consensus: Node is the slot leader: Proposed block: {}", proposal); debug!("consensus: Full proposal: {:?}", proposal); match state.write().await.receive_proposal(&proposal).await { Ok(to_broadcast) => { info!("consensus: Block proposal saved successfully"); // 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 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: Block proposal save failed: {}", e); } } } }