Improve logging around subnet discoveries

Author: pawanjay176

beacon_node/eth2_libp2p/src/discovery/mod.rs

@@ -18,8 +18,9 @@ use futures::prelude::*;
 use futures::stream::FuturesUnordered;
 use libp2p::core::PeerId;
 use lru::LruCache;
-use slog::{crit, debug, error, info, warn};
+use slog::{crit, debug, error, info, trace, warn};
 use ssz::Encode;
+use std::fmt::Display;
 use std::{
     collections::{HashMap, VecDeque},
     net::{IpAddr, SocketAddr},
@@ -71,6 +72,19 @@ struct SubnetQuery {
     retries: usize,
 }
 
+impl Display for SubnetQuery {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let min_ttl_secs = self
+            .min_ttl
+            .map(|ttl| ttl.saturating_duration_since(Instant::now()).as_secs());
+        write!(
+            f,
+            "Subnet: {:?}, ttl_seconds: {:?}, attempt: {}",
+            self.subnet, min_ttl_secs, self.retries
+        )
+    }
+}
+
 #[derive(Debug, Clone, PartialEq)]
 enum QueryType {
     /// We are searching for subnet peers.
@@ -317,7 +331,7 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
         if !self.started {
             return;
         }
-        debug!(
+        trace!(
             self.log,
             "Making discovery query for subnets";
             "subnets" => ?subnets_to_discover.iter().map(|s| s.subnet).collect::<Vec<_>>()
@@ -601,7 +615,7 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
                 retries,
             });
             // update the metrics and insert into the queue.
-            debug!(self.log, "Queuing subnet query"; "subnet" => ?subnet, "retries" => retries);
+            trace!(self.log, "Queuing subnet query"; "subnet" => ?subnet, "retries" => retries);
             self.queued_queries.push_back(query);
             metrics::set_gauge(&metrics::DISCOVERY_QUEUE, self.queued_queries.len() as i64);
         }
@@ -655,11 +669,6 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
                         // This query is for searching for peers of a particular subnet
                         // Drain subnet_queries so we can re-use it as we continue to process the queue
                         let grouped_queries: Vec<SubnetQuery> = subnet_queries.drain(..).collect();
-                        debug!(
-                            self.log,
-                            "Starting grouped subnet query";
-                            "subnets" => ?grouped_queries.iter().map(|q| q.subnet).collect::<Vec<_>>(),
-                        );
                         self.start_subnet_query(grouped_queries);
                         processed = true;
                     }
@@ -704,13 +713,10 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
                 }
 
                 let target_peers = TARGET_SUBNET_PEERS - peers_on_subnet;
-                debug!(self.log, "Discovery query started for subnet";
-                    "subnet_id" => ?subnet_query.subnet,
+                trace!(self.log, "Discovery query started for subnet";
+                    "subnet_query" => %subnet_query,
                     "connected_peers_on_subnet" => peers_on_subnet,
-                    "target_subnet_peers" => TARGET_SUBNET_PEERS,
                     "peers_to_find" => target_peers,
-                    "attempt" => subnet_query.retries,
-                    "min_ttl" => ?subnet_query.min_ttl,
                 );
 
                 filtered_subnets.push(subnet_query.subnet);
@@ -723,6 +729,11 @@ impl<TSpec: EthSpec> Discovery<TSpec> {
             // build the subnet predicate as a combination of the eth2_fork_predicate and the subnet predicate
             let subnet_predicate = subnet_predicate::<TSpec>(filtered_subnets, &self.log);
 
+            debug!(
+                self.log,
+                "Starting grouped subnet query";
+                "subnets" => ?filtered_subnet_queries.iter().map(|q| q.to_string()).collect::<Vec<_>>(),
+            );
             self.start_query(
                 GroupedQueryType::Subnet(filtered_subnet_queries),
                 TARGET_PEERS_FOR_GROUPED_QUERY,

beacon_node/eth2_libp2p/src/peer_manager/mod.rs

@@ -957,6 +957,11 @@ impl<TSpec: EthSpec> PeerManager<TSpec> {
 
         // request the subnet query from discovery
         if !subnets_to_discover.is_empty() {
+            debug!(
+                self.log,
+                "Making subnet queries for maintaining sync committee peers";
+                "subnets" => ?subnets_to_discover.iter().map(|s| s.subnet).collect::<Vec<_>>()
+            );
             self.discovery.discover_subnet_peers(subnets_to_discover);
         }
     }

beacon_node/eth2_libp2p/src/rpc/codec/ssz_snappy.rs

@@ -161,8 +161,6 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyInboundCodec<TSpec> {
                 self.len = None;
                 let _read_bytes = src.split_to(n as usize);
 
-                // We need not check that decoded_buffer.len() is within bounds here
-                // since we have already checked `length` above.
                 match self.protocol.version {
                     Version::V1 => handle_v1_request(self.protocol.message_name, &decoded_buffer),
                     Version::V2 => handle_v2_request(self.protocol.message_name, &decoded_buffer),
@@ -298,8 +296,6 @@ impl<TSpec: EthSpec> Decoder for SSZSnappyOutboundCodec<TSpec> {
                 self.len = None;
                 let _read_bytes = src.split_to(n as usize);
 
-                // We need not check that decoded_buffer.len() is within bounds here
-                // since we have already checked `length` above.
                 match self.protocol.version {
                     Version::V1 => handle_v1_response(self.protocol.message_name, &decoded_buffer),
                     Version::V2 => handle_v2_response(
@@ -532,7 +528,7 @@ fn handle_v1_response<T: EthSpec>(
 // length = length-prefix received in the beginning of the stream.
 ///
 /// For BlocksByRange/BlocksByRoot reponses, decodes the appropriate response
-/// according to the received fork_name.
+/// according to the received `ForkName`.
 fn handle_v2_response<T: EthSpec>(
     protocol: Protocol,
     decoded_buffer: &[u8],
@@ -886,6 +882,7 @@ mod tests {
         );
     }
 
+    // Test RPCResponse encoding/decoding for V2 messages
     #[test]
     fn test_context_bytes_v2() {
         let fork_context = fork_context();
@@ -1001,8 +998,10 @@ mod tests {
         )
     }
 
+    /// Test a malicious snappy encoding for a V1 `Status` message where the attacker
+    /// sends a valid message filled with a stream of useless padding before the actual message.
     #[test]
-    fn test_decode_malicious_status_message() {
+    fn test_decode_malicious_v1_message() {
         // 10 byte snappy stream identifier
         let stream_identifier: &'static [u8] = b"\xFF\x06\x00\x00sNaPpY";
 
@@ -1055,6 +1054,8 @@ mod tests {
         );
     }
 
+    /// Test a malicious snappy encoding for a V2 `BlocksByRange` message where the attacker
+    /// sends a valid message filled with a stream of useless padding before the actual message.
     #[test]
     fn test_decode_malicious_v2_message() {
         let fork_context = Arc::new(fork_context());

beacon_node/eth2_libp2p/src/types/topics.rs

@@ -142,7 +142,7 @@ impl GossipTopic {
                 PROPOSER_SLASHING_TOPIC => GossipKind::ProposerSlashing,
                 ATTESTER_SLASHING_TOPIC => GossipKind::AttesterSlashing,
                 topic => match committee_topic_index(topic) {
-                    Some(subnet_id) => match subnet_id {
+                    Some(subnet) => match subnet {
                         Subnet::Attestation(s) => GossipKind::Attestation(s),
                         Subnet::SyncCommittee(s) => GossipKind::SyncCommitteeMessage(s),
                     },