Merge pull request #4136 from ProvableHQ/feat/extended_commit_tracking

[Feat] Log the combined stake at our SHA (in validator mode)

Author: vicsn

Cargo.lock

@@ -155,6 +155,9 @@ default-features = false
 [workspace.dependencies.rand_distr]
 version = "0.4"
 
+[workspace.dependencies.smol_str]
+version = "=0.3.2" # Update after Rust >1.88
+
 [workspace.dependencies.tracing]
 version = "0.1"
 default-features = false
@@ -356,8 +359,7 @@ workspace = true
 version = "0.11.2"
 
 [build-dependencies.built]
-version = "0.8"
-features = [ "git2" ]
+workspace = true
 
 [build-dependencies.toml]
 version = "0.9"

Cargo.toml

@@ -102,6 +102,9 @@ workspace = true
 version = "0.10"
 default-features = false
 
+[dependencies.smol_str]
+workspace = true
+
 [dependencies.snarkos-account]
 workspace = true
 

node/bft/Cargo.toml

@@ -24,6 +24,7 @@ use crate::{
     helpers::{Cache, PrimarySender, Storage, SyncSender, WorkerSender, assign_to_worker},
     spawn_blocking,
 };
+use smol_str::SmolStr;
 use snarkos_account::Account;
 use snarkos_node_bft_events::{
     BlockRequest,
@@ -50,6 +51,7 @@ use snarkos_node_network::{
     bootstrap_peers,
     get_repo_commit_hash,
     log_repo_sha_comparison,
+    shorten_snarkos_sha,
 };
 use snarkos_node_sync::{InsertBlockResponseError, MAX_BLOCKS_BEHIND, communication_service::CommunicationService};
 use snarkos_node_tcp::{
@@ -475,6 +477,7 @@ impl<N: Network> Gateway<N> {
                 address,
                 NodeType::Validator,
                 0,
+                get_repo_commit_hash(),
                 ConnectionMode::Gateway,
             );
         }
@@ -886,7 +889,7 @@ impl<N: Network> Gateway<N> {
     /// Logs the connected validators.
     fn log_connected_validators(&self) {
         // Retrieve the connected validators and current committee.
-        let connected_validators = self.connected_peers();
+        let connected_validators = self.get_connected_peers();
         let committee = match self.ledger.current_committee() {
             Ok(c) => c,
             Err(err) => {
@@ -908,15 +911,32 @@ impl<N: Network> Gateway<N> {
 
         // Collect the connected validator addresses and stake.
         let mut connected_validator_addresses = HashSet::with_capacity(connected_validators.len());
+        let mut connected_validator_shas: HashMap<SmolStr, u64> = HashMap::with_capacity(connected_validators.len());
+        // Insert our sha.
+        let our_sha = shorten_snarkos_sha(&get_repo_commit_hash());
+        let our_stake = committee.get_stake(self.account.address());
+        connected_validator_shas.insert(our_sha.clone(), our_stake);
         // Include our own address.
         connected_validator_addresses.insert(self.account.address());
         // Include and log the connected validators.
-        for peer_ip in &connected_validators {
-            let address = self.resolve_to_aleo_addr(*peer_ip).map_or("Unknown".to_string(), |a| {
-                connected_validator_addresses.insert(a);
-                a.to_string()
-            });
-            debug!("{}", format!("  Connected to: {peer_ip} - {address}").dimmed());
+        for peer in &connected_validators {
+            let peer_ip = peer.listener_addr;
+            // Register the Aleo address.
+            let address = peer.aleo_addr;
+            connected_validator_addresses.insert(address);
+            // Register the snarkOS commit SHA and the associated stake.
+            let address_stake = committee.get_stake(address);
+            let short_peer_sha = shorten_snarkos_sha(&peer.snarkos_sha);
+            *connected_validator_shas.entry(short_peer_sha.clone()).or_default() += address_stake;
+            // Log the connected validator.
+            debug!("{}", format!("  Connected to: {peer_ip} - {address} @ {short_peer_sha}").dimmed());
+        }
+
+        if let Some(combined_stake) = connected_validator_shas.get(&our_sha) {
+            let percentage = *combined_stake as f64 / committee.total_stake() as f64 * 100.0;
+            debug!("{}", format!("  Combined stake @ {our_sha}: {percentage:.2}%").dimmed());
+            #[cfg(feature = "metrics")]
+            metrics::gauge(metrics::bft::CONNECTED_STAKE_WITH_MATCHING_SHA, percentage);
         }
 
         // Log the validators that are not connected.
@@ -1423,6 +1443,7 @@ impl<N: Network> Handshake for Gateway<N> {
                             cr.address,
                             node_type,
                             cr.version,
+                            cr.snarkos_sha,
                             ConnectionMode::Gateway,
                         );
                     }
@@ -1508,8 +1529,9 @@ impl<N: Network> Gateway<N> {
         // Determine the snarkOS SHA to send to the peer.
         let current_block_height = self.ledger.latest_block_height();
         let consensus_version = N::CONSENSUS_VERSION(current_block_height).unwrap();
-        let snarkos_sha = match (consensus_version >= ConsensusVersion::V12, get_repo_commit_hash()) {
-            (true, Some(sha)) => Some(sha),
+        let snarkos_sha = match (self.is_dev(), consensus_version >= ConsensusVersion::V12, get_repo_commit_hash()) {
+            (true, _, Some(sha)) => Some(sha),
+            (_, true, Some(sha)) => Some(sha),
             _ => None,
         };
         // Send a challenge request to the peer.
@@ -1615,8 +1637,9 @@ impl<N: Network> Gateway<N> {
         // Determine the snarkOS SHA to send to the peer.
         let current_block_height = self.ledger.latest_block_height();
         let consensus_version = N::CONSENSUS_VERSION(current_block_height).unwrap();
-        let snarkos_sha = match (consensus_version >= ConsensusVersion::V12, get_repo_commit_hash()) {
-            (true, Some(sha)) => Some(sha),
+        let snarkos_sha = match (self.is_dev(), consensus_version >= ConsensusVersion::V12, get_repo_commit_hash()) {
+            (true, _, Some(sha)) => Some(sha),
+            (_, true, Some(sha)) => Some(sha),
             _ => None,
         };
         // Send the challenge request.

node/bft/src/gateway.rs

@@ -16,9 +16,10 @@
 pub(super) const COUNTER_NAMES: [&str; 3] =
     [bft::LEADERS_ELECTED, consensus::STALE_UNCONFIRMED_TRANSACTIONS, consensus::STALE_UNCONFIRMED_SOLUTIONS];
 
-pub(super) const GAUGE_NAMES: [&str; 27] = [
+pub(super) const GAUGE_NAMES: [&str; 28] = [
     bft::CONNECTED,
     bft::CONNECTED_STAKE,
+    bft::CONNECTED_STAKE_WITH_MATCHING_SHA,
     bft::CONNECTING,
     bft::LAST_STORED_ROUND,
     bft::PROPOSAL_ROUND,
@@ -53,6 +54,7 @@ pub mod bft {
     pub const COMMIT_ROUNDS_LATENCY: &str = "snarkos_bft_commit_rounds_latency_secs"; // <-- This one doesn't even make sense.
     pub const CONNECTED: &str = "snarkos_bft_connected_total";
     pub const CONNECTED_STAKE: &str = "snarkos_bft_connected_stake_as_percentage";
+    pub const CONNECTED_STAKE_WITH_MATCHING_SHA: &str = "snarkos_bft_connected_stake_with_matching_sha_as_percentage";
     pub const CONNECTING: &str = "snarkos_bft_connecting_total";
     pub const LAST_STORED_ROUND: &str = "snarkos_bft_last_stored_round";
     pub const LEADERS_ELECTED: &str = "snarkos_bft_leaders_elected_total";

node/metrics/src/names.rs

@@ -37,6 +37,9 @@ workspace = true
 [dependencies.serde]
 workspace = true
 
+[dependencies.smol_str]
+workspace = true
+
 [dependencies.snarkos-node-tcp]
 workspace = true
 

node/network/Cargo.toml

@@ -29,6 +29,7 @@ pub use resolver::*;
 
 use snarkvm::prelude::Network;
 
+use smol_str::SmolStr;
 use std::{env::VarError, net::SocketAddr, str::FromStr};
 use tracing::*;
 
@@ -113,3 +114,16 @@ pub fn log_repo_sha_comparison(peer_addr: SocketAddr, peer_sha: &Option<[u8; 40]
 
     debug!("{ctx} Peer '{peer_addr}' uses snarkOS{sha_cmp}");
 }
+
+pub fn shorten_snarkos_sha(sha: &Option<[u8; 40]>) -> SmolStr {
+    if let Some(full_sha) = sha.as_ref().and_then(|s| str::from_utf8(s).ok()) {
+        let end_idx = full_sha.char_indices()
+            .nth(7) // GitHub commit SHA shorthand.
+            .map(|(i, _)| i)
+            .unwrap_or(full_sha.len()); // Can't really fail.
+
+        SmolStr::from(&full_sha[..end_idx])
+    } else {
+        "unknown snarkOS SHA".into()
+    }
+}

node/network/src/lib.rs

@@ -72,6 +72,8 @@ pub struct ConnectedPeer<N: Network> {
     pub node_type: NodeType,
     /// The message version of the peer.
     pub version: u32,
+    /// The snarkOS commit hash of the peer.
+    pub snarkos_sha: Option<[u8; 40]>,
     /// The latest block height known to be associated with the peer.
     pub last_height_seen: Option<u32>,
     /// The timestamp of the first message received from the peer.
@@ -105,13 +107,15 @@ impl<N: Network> Peer<N> {
     }
 
     /// Promote a connecting peer to a fully connected one.
+    #[allow(clippy::too_many_arguments)]
     pub fn upgrade_to_connected(
         &mut self,
         connected_addr: SocketAddr,
         listener_port: u16,
         aleo_address: Address<N>,
         node_type: NodeType,
         node_version: u32,
+        snarkos_sha: Option<[u8; 40]>,
         connection_mode: ConnectionMode,
     ) {
         let timestamp = Instant::now();
@@ -131,6 +135,7 @@ impl<N: Network> Peer<N> {
             node_type,
             trusted: self.is_trusted(),
             version: node_version,
+            snarkos_sha,
             last_height_seen: None,
             first_seen: timestamp,
             last_seen: timestamp,

node/network/src/peer.rs

@@ -143,6 +143,7 @@ impl<N: Network> Router<N> {
                             cr.address,
                             cr.node_type,
                             cr.version,
+                            cr.snarkos_sha,
                             ConnectionMode::Router,
                         );
                     }

node/router/src/handshake.rs

@@ -118,7 +118,7 @@ impl<N: Network> Handshake for BootstrapClient<N> {
 
         if let Some(addr) = listener_addr {
             match handshake_result {
-                Ok((peer_port, peer_aleo_addr, peer_node_type, peer_version, connection_mode)) => {
+                Ok((peer_port, peer_aleo_addr, peer_node_type, peer_version, peer_snarkos_sha, connection_mode)) => {
                     if let Some(peer) = self.peer_pool.write().get_mut(&addr) {
                         // Due to only having a single Resolver, the BootstrapClient only adds an Aleo
                         // address mapping for Gateway-mode connections, as it is only used there, and
@@ -133,6 +133,7 @@ impl<N: Network> Handshake for BootstrapClient<N> {
                             peer_aleo_addr,
                             peer_node_type,
                             peer_version,
+                            peer_snarkos_sha,
                             connection_mode,
                         );
                     }
@@ -160,24 +161,36 @@ impl<N: Network> BootstrapClient<N> {
         peer_addr: SocketAddr,
         listener_addr: &mut Option<SocketAddr>,
         stream: &'a mut TcpStream,
-    ) -> Result<(u16, Address<N>, NodeType, u32, ConnectionMode), ConnectError> {
+    ) -> Result<(u16, Address<N>, NodeType, u32, Option<[u8; 40]>, ConnectionMode), ConnectError> {
         // Construct the stream.
         let mut framed = Framed::new(stream, BootstrapClientCodec::<N>::handshake());
 
         /* Step 1: Receive the challenge request. */
 
         // Listen for the challenge request message, which can be either from a regular peer, or a validator.
         let peer_request = expect_handshake_msg!(HandshakeMessageKind::ChallengeRequest, framed, peer_addr);
-        let (peer_port, peer_nonce, peer_aleo_addr, peer_node_type, peer_version, connection_mode) = match peer_request
-        {
-            MessageOrEvent::Message(Message::ChallengeRequest(ref msg)) => {
-                (msg.listener_port, msg.nonce, msg.address, msg.node_type, msg.version, ConnectionMode::Router)
-            }
-            MessageOrEvent::Event(Event::ChallengeRequest(ref msg)) => {
-                (msg.listener_port, msg.nonce, msg.address, NodeType::Validator, msg.version, ConnectionMode::Gateway)
-            }
-            _ => unreachable!(),
-        };
+        let (peer_port, peer_nonce, peer_aleo_addr, peer_node_type, peer_version, peer_snarkos_sha, connection_mode) =
+            match peer_request {
+                MessageOrEvent::Message(Message::ChallengeRequest(ref msg)) => (
+                    msg.listener_port,
+                    msg.nonce,
+                    msg.address,
+                    msg.node_type,
+                    msg.version,
+                    msg.snarkos_sha,
+                    ConnectionMode::Router,
+                ),
+                MessageOrEvent::Event(Event::ChallengeRequest(ref msg)) => (
+                    msg.listener_port,
+                    msg.nonce,
+                    msg.address,
+                    NodeType::Validator,
+                    msg.version,
+                    msg.snarkos_sha,
+                    ConnectionMode::Gateway,
+                ),
+                _ => unreachable!(),
+            };
         debug!("Handshake mode: {connection_mode:?}");
 
         // Obtain the peer's listening address.
@@ -261,7 +274,7 @@ impl<N: Network> BootstrapClient<N> {
             return Err(ConnectError::application(DisconnectReason::InvalidChallengeResponse));
         }
 
-        Ok((peer_port, peer_aleo_addr, peer_node_type, peer_version, connection_mode))
+        Ok((peer_port, peer_aleo_addr, peer_node_type, peer_version, peer_snarkos_sha, connection_mode))
     }
 
     async fn verify_challenge_request(