consensus: participants logic updated with keepalive

bin/darkfid/src/main.rs

@@ -10,7 +10,7 @@ use darkfi::{
     async_daemonize, cli_desc,
     consensus::{
         proto::{
-            ProtocolParticipant, ProtocolProposal, ProtocolSync, ProtocolSyncConsensus, ProtocolTx,
+            ProtocolKeepAlive, ProtocolParticipant, ProtocolProposal, ProtocolSync, ProtocolSyncConsensus, ProtocolTx,
         },
         state::ValidatorStatePtr,
         task::{block_sync_task, proposal_task},
@@ -375,7 +375,7 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
     let consensus_p2p = {
         if !args.consensus {
             None
-        } else {
+        } else {            
             info!("Registering consensus P2P protocols...");
             let consensus_network_settings = net::Settings {
                 inbound: args.consensus_p2p_accept,
@@ -401,6 +401,14 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
                 })
                 .await;
 
+            let _state = state.clone();
+            registry
+                .register(net::SESSION_ALL, move |channel, p2p| {
+                    let state = _state.clone();
+                    async move { ProtocolKeepAlive::init(channel, state, p2p).await.unwrap() }
+                })
+                .await;
+
             let _state = state.clone();
             registry
                 .register(net::SESSION_ALL, move |channel, p2p| {
@@ -465,7 +473,8 @@ async fn realmain(args: Args, ex: Arc<smol::Executor<'_>>) -> Result<()> {
         consensus_p2p.clone().unwrap().wait_for_outbound(ex.clone()).await?;
 
         info!("Starting consensus protocol task");
-        ex.spawn(proposal_task(consensus_p2p.unwrap(), sync_p2p.unwrap(), state)).detach();
+        let _ex = ex.clone();
+        ex.spawn(proposal_task(consensus_p2p.unwrap(), sync_p2p.unwrap(), state, _ex)).detach();
     } else {
         info!("Not starting consensus P2P network");
     }

src/consensus/mod.rs

@@ -10,7 +10,7 @@ pub use metadata::{
 
 /// Consensus participant
 pub mod participant;
-pub use participant::Participant;
+pub use participant::{KeepAlive, Participant};
 
 /// Consensus state
 pub mod state;

src/consensus/participant.rs

@@ -1,28 +1,26 @@
 use crate::{
-    crypto::{address::Address, keypair::PublicKey},
+    crypto::{address::Address, keypair::PublicKey, schnorr::Signature},
     net,
     serial::{SerialDecodable, SerialEncodable},
 };
 
 /// This struct represents a tuple of the form:
-/// (`node_address`, `slot_joined`, `last_slot_voted`, `slot_quarantined`)
+/// (`public_key`, `node_address`, `last_slot_seen`,`slot_quarantined`)
 #[derive(Debug, Clone, PartialEq, Eq, SerialEncodable, SerialDecodable)]
 pub struct Participant {
     /// Node public key
     pub public_key: PublicKey,
     /// Node wallet address
     pub address: Address,
-    /// Slot node joined the network
-    pub joined: u64,
-    /// Last slot node voted
-    pub voted: Option<u64>,
+    /// Last slot node send a keep alive message
+    pub seen: u64,
     /// Slot participant was quarantined by the node
     pub quarantined: Option<u64>,
 }
 
 impl Participant {
     pub fn new(public_key: PublicKey, address: Address, joined: u64) -> Self {
-        Self { public_key, address, joined, voted: None, quarantined: None }
+        Self { public_key, address, seen: joined, quarantined: None }
     }
 }
 
@@ -31,3 +29,20 @@ impl net::Message for Participant {
         "participant"
     }
 }
+
+/// Struct represending a keep alive message, containing signed slot for validation
+#[derive(Debug, Clone, SerialEncodable, SerialDecodable)]
+pub struct KeepAlive {
+    /// Node address
+    pub address: Address,
+    /// Slot message was send
+    pub slot: u64,
+    /// Slot signature
+    pub signature: Signature,
+}
+
+impl net::Message for KeepAlive {
+    fn name() -> &'static str {
+        "keepalive"
+    }
+}

src/consensus/proto/mod.rs

@@ -2,6 +2,10 @@
 mod protocol_participant;
 pub use protocol_participant::ProtocolParticipant;
 
+/// Participant keep alive protocol
+mod protocol_keep_alive;
+pub use protocol_keep_alive::ProtocolKeepAlive;
+
 /// Block proposal protocol
 mod protocol_proposal;
 pub use protocol_proposal::ProtocolProposal;

src/consensus/proto/protocol_keep_alive.rs

@@ -0,0 +1,93 @@
+use async_std::sync::Arc;
+
+use async_executor::Executor;
+use async_trait::async_trait;
+use log::{debug, error};
+use url::Url;
+
+use crate::{
+    consensus::{KeepAlive, ValidatorStatePtr},
+    net::{
+        ChannelPtr, MessageSubscription, P2pPtr, ProtocolBase, ProtocolBasePtr,
+        ProtocolJobsManager, ProtocolJobsManagerPtr,
+    },
+    Result,
+};
+
+pub struct ProtocolKeepAlive {
+    keep_alive_sub: MessageSubscription<KeepAlive>,
+    jobsman: ProtocolJobsManagerPtr,
+    state: ValidatorStatePtr,
+    p2p: P2pPtr,
+    channel_address: Url,
+}
+
+impl ProtocolKeepAlive {
+    pub async fn init(
+        channel: ChannelPtr,
+        state: ValidatorStatePtr,
+        p2p: P2pPtr,
+    ) -> Result<ProtocolBasePtr> {
+        debug!("Adding ProtocolKeepAlive to the protocol registry");
+        let msg_subsystem = channel.get_message_subsystem();
+        msg_subsystem.add_dispatch::<KeepAlive>().await;
+
+        let keep_alive_sub = channel.subscribe_msg::<KeepAlive>().await?;
+        let channel_address = channel.address();
+
+        Ok(Arc::new(Self {
+            keep_alive_sub,
+            jobsman: ProtocolJobsManager::new("ProtocolKeepAlive", channel),
+            state,
+            p2p,
+            channel_address,
+        }))
+    }
+
+    async fn handle_receive_keep_alive(self: Arc<Self>) -> Result<()> {
+        debug!("ProtocolKeepAlive::handle_receive_keep_alive() [START]");
+        let exclude_list = vec![self.channel_address.clone()];
+        loop {
+            let keep_alive = match self.keep_alive_sub.receive().await {
+                Ok(v) => v,
+                Err(e) => {
+                    error!("ProtocolKeepAlive::handle_receive_keep_alive(): recv error: {}", e);
+                    continue
+                }
+            };
+
+            debug!("ProtocolKeepAlive::handle_receive_keep_alive() recv: {:?}", keep_alive);
+
+            let keep_alive_copy = (*keep_alive).clone();
+
+            if self.state.write().await.participant_keep_alive(keep_alive_copy.clone()) {
+                if let Err(e) =
+                    self.p2p.broadcast_with_exclude(keep_alive_copy, &exclude_list).await
+                {
+                    error!(
+                        "ProtocolKeepAlive::handle_receive_keep_alive(): p2p broadcast failed: {}",
+                        e
+                    );
+                };
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl ProtocolBase for ProtocolKeepAlive {
+    async fn start(self: Arc<Self>, executor: Arc<Executor<'_>>) -> Result<()> {
+        debug!("ProtocolKeepAlive::start() [START]");
+        self.jobsman.clone().start(executor.clone());
+        self.jobsman
+            .clone()
+            .spawn(self.clone().handle_receive_keep_alive(), executor.clone())
+            .await;
+        debug!("ProtocolKeepAlive::start() [END]");
+        Ok(())
+    }
+
+    fn name(&self) -> &'static str {
+        "ProtocolKeepAlive"
+    }
+}

src/consensus/state.rs

@@ -13,7 +13,7 @@ use log::{debug, error, info, warn};
 use rand::rngs::OsRng;
 
 use super::{
-    Block, BlockInfo, BlockProposal, Header, OuroborosMetadata, Participant, ProposalChain,
+    Block, BlockInfo, BlockProposal, Header, KeepAlive, OuroborosMetadata, Participant, ProposalChain,
     StreamletMetadata,
 };
 
@@ -419,12 +419,9 @@ impl ValidatorState {
         // TODO: [PLACEHOLDER] Add crypsinous proof validation (to replace balance proof)
         // TODO: [PLACEHOLDER] Add rewards validation
         
-        // TODO: uncomment this after adding seen attribute to participant
-        /*
         if current > leader.seen {
             leader.seen = current;
         }
-        */
 
         // Invalidating quarantine
         leader.quarantined = None;
@@ -601,16 +598,59 @@ impl ValidatorState {
         if self.consensus.pending_participants.contains(&participant) {
             return false
         }
+        
+    // TODO: [PLACEHOLDER] Add balance proof validation
 
         self.consensus.pending_participants.push(participant);
         true
     }
+    
+    /// Update participant seen.
+    pub fn participant_keep_alive(&mut self, keep_alive: KeepAlive) -> bool {
+        match self.consensus.participants.get(&keep_alive.address) {
+            None => {
+                warn!(
+                    "Keep alive message from unknown participant: {}",
+                    keep_alive.address.to_string()
+                );
+                false
+            }
+            Some(participant) => {
+                let current = self.current_slot();
+                if current != keep_alive.slot {
+                    warn!("keep alive message slot is not current one for: {}", keep_alive.address);
+                    return false
+                }
+
+                let serialized = serialize(&current);
+                if !participant.public_key.verify(&serialized, &keep_alive.signature) {
+                    warn!(
+                        "Keep alive message signature could not be verified for: {}",
+                        keep_alive.address
+                    );
+                    return false
+                }
+
+                // TODO: [PLACEHOLDER] Add balance proof validation
+
+                // Updating participant last seen slot
+                let mut participant = participant.clone();
+                participant.seen = current;
+
+                // Invalidating quarantine
+                participant.quarantined = None;
+
+                self.consensus.participants.insert(participant.address, participant);
+
+                true
+            }
+        }
+    }
+
 
     /// Refresh the participants map, to retain only the active ones.
-    /// Active nodes are considered those that joined previous slot
-    /// or on the slot the last proposal was generated, either voted
-    /// or joined the previous of that slot. That ensures we cover
-    /// the case of a node joining while the chosen slot leader is inactive.
+    /// Active nodes are considered those that their last seen slot is
+    /// in range: [current_slot - QUARANTINE_DURATION, current_slot]
     /// Inactive nodes are marked as quarantined, so they can be removed if
     /// they are in quarantine more than the predifined quarantine period.
     pub fn refresh_participants(&mut self) -> Result<()> {
@@ -635,39 +675,23 @@ impl ValidatorState {
         self.consensus.pending_participants = vec![];
 
         let mut inactive = Vec::new();
-        let mut last_slot = self.last_slot()?;
-
-        // This check ensures that we don't chech the current slot,
-        // as a node might receive the proposal of current slot before
-        // starting refreshing participants, so the last_slot will be
-        // the current one.
-        if last_slot >= current {
-            last_slot = current - 1;
-        }
-
-        let previous_slot = current - 1;
-        // This check ensures that when restarting the network, previous
-        // from last slot is not u64::MAX
-        let previous_from_last_slot = match last_slot {
-            0 => 0,
-            _ => last_slot - 1,
-        };
+        let low_bound = current - QUARANTINE_DURATION;
 
         debug!(
-            "refresh_participants(): Node {:?} checking slots: previous - {:?}, last - {:?}, previous from last - {:?}",
-            self.address, previous_slot, last_slot, previous_from_last_slot
+            "refresh_participants(): Node {} checking slots range: {} -> {}",
+            self.address, low_bound, current
         );
 
         let leader = self.slot_leader();
         for (index, participant) in self.consensus.participants.iter_mut() {
             match participant.quarantined {
                 Some(slot) => {
-                    if (current - slot) > QUARANTINE_DURATION {
+                    if slot < low_bound {
                         warn!(
-                            "refresh_participants(): Removing participant: {:?} (joined {:?}, voted {:?})",
+                            "refresh_participants(): Removing participant: {} (seen {}, quarantined {})",
                             participant.address,
-                            participant.joined,
-                            participant.voted
+                            participant.seen,
+                            slot
                         );
                         inactive.push(*index);
                     }
@@ -677,40 +701,18 @@ impl ValidatorState {
                     // becoming the leader. This can be used for slashing in the future.
                     if participant.address == leader.address {
                         debug!(
-                            "refresh_participants(): Quaranteening leader: {:?} (joined {:?}, voted {:?})",
-                            participant.address,
-                            participant.joined,
-                            participant.voted
+                            "refresh_participants(): Quaranteening leader: {} (seen {})",
+                            participant.address, participant.seen
                         );
                         participant.quarantined = Some(current);
                         continue
                     }
-                    match participant.voted {
-                        Some(slot) => {
-                            if slot < last_slot {
-                                warn!(
-                                    "refresh_participants(): Quaranteening participant: {:?} (joined {:?}, voted {:?})",
-                                    participant.address,
-                                    participant.joined,
-                                    participant.voted
-                                );
-                                participant.quarantined = Some(current);
-                            }
-                        }
-                        None => {
-                            if (previous_slot == last_slot && participant.joined < previous_slot) ||
-                                (previous_slot != last_slot &&
-                                    participant.joined < previous_from_last_slot)
-                            {
-                                warn!(
-                                    "refresh_participants(): Quaranteening participant: {:?} (joined {:?}, voted {:?})",
-                                    participant.address,
-                                    participant.joined,
-                                    participant.voted
-                                );
-                                participant.quarantined = Some(current);
-                            }
-                        }
+                    if participant.seen < low_bound {
+                        warn!(
+                            "refresh_participants(): Quaranteening participant: {} (seen {})",
+                            participant.address, participant.seen
+                        );
+                        participant.quarantined = Some(current);
                     }
                 }
             }

src/consensus/task/keep_alive.rs

@@ -0,0 +1,51 @@
+use async_executor::Executor;
+use async_std::sync::Arc;
+use log::{debug, error};
+use rand::Rng;
+
+use crate::{
+    consensus::{state::QUARANTINE_DURATION, KeepAlive, ValidatorStatePtr},
+    crypto::schnorr::SchnorrSecret,
+    net::P2pPtr,
+    serial::serialize,
+    util::async_util::sleep,
+    Result,
+};
+
+/// async task used for sending keep alive messages in background.
+pub async fn keep_alive_task(
+    p2p: P2pPtr,
+    state: ValidatorStatePtr,
+    ex: Arc<Executor<'_>>,
+) -> Result<()> {
+    ex.spawn(async move {
+        loop {
+            // Pick a random slot in range: next slot + QUARANTINE_DURATION, exluding first and last
+            let slot = rand::thread_rng().gen_range(2..QUARANTINE_DURATION);
+            let seconds = state.read().await.next_n_slot_start(slot).as_secs();
+            debug!("keep_alive_task: Waiting for next {} slots ({} sec)", slot, seconds);
+
+            // Sleep until that slot
+            sleep(seconds).await;
+
+            // TODO: [PLACEHOLDER] Add balance proof creation
+
+            // Create keep alive message
+            let secret = state.read().await.secret;
+            let address = state.read().await.address;
+            let slot = state.read().await.current_slot();
+            let serialized = serialize(&slot);
+            let signature = secret.sign(&serialized);
+            let keep_alive = KeepAlive { address, slot, signature };
+
+            // Broadcast keep alive message
+            match p2p.broadcast(keep_alive).await {
+                Ok(()) => debug!("keep_alive_task: Keep alive message broadcasted successfully."),
+                Err(e) => error!("keep_alive_task: Failed broadcasting keep alive message: {}", e),
+            }
+        }
+    })
+    .detach();
+
+    Ok(())
+}

src/consensus/task/mod.rs

@@ -8,3 +8,6 @@ pub use consensus_sync::consensus_sync_task;
 
 mod proposal;
 pub use proposal::proposal_task;
+
+mod keep_alive;
+pub use keep_alive::keep_alive_task;

src/consensus/task/proposal.rs

@@ -1,8 +1,10 @@
 use std::time::Duration;
 
+use async_executor::Executor;
+use async_std::sync::Arc;
 use log::{debug, error, info};
 
-use super::consensus_sync_task;
+use super::{consensus_sync_task, keep_alive_task};
 use crate::{
     consensus::{Participant, ValidatorStatePtr},
     net::P2pPtr,
@@ -10,7 +12,14 @@ use crate::{
 };
 
 /// async task used for participating in the consensus protocol
-pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: ValidatorStatePtr) {
+pub async fn proposal_task(
+    consensus_p2p: P2pPtr,
+    sync_p2p: P2pPtr,
+    state: ValidatorStatePtr,
+    ex: Arc<Executor<'_>>,
+) {
+    // TODO: [PLACEHOLDER] Add balance proof creation
+
     // Node waits just before the current or next slot end, so it can
     // start syncing latest state.
     let mut seconds_until_next_slot = state.read().await.next_n_slot_start(1);
@@ -49,6 +58,12 @@ pub async fn proposal_task(consensus_p2p: P2pPtr, sync_p2p: P2pPtr, state: Valid
         Ok(()) => info!("consensus: Participation message broadcasted successfully."),
         Err(e) => error!("Failed broadcasting consensus participation: {}", e),
     }
+    
+    // Node initiates the background task to send keep alive messages
+    match keep_alive_task(consensus_p2p.clone(), state.clone(), ex).await {
+        Ok(()) => info!("consensus: Keep alive background task initiated successfully."),
+        Err(e) => error!("Failed to initiate keep alive background task: {}", e),
+    }
 
     // Node modifies its participating slot to next.
     match state.write().await.set_participating() {