Add additional logs for P2P debugging (#434)

* feat: make logs on network manager more granular

* feat: log error on session disconnect

* temp: manual peer

* temp: manual peer

* feat(linear-dl): make BadResponse error more descriptive

* feat(linear-dl): log retries on future

* chore: add target

* chore: add target

* chore: remove unused var

* chore: remove manual peer override

Co-authored-by: Oliver Nordbjerg <hi@notbjerg.me>

Cargo.lock

@@ -3293,6 +3293,7 @@ dependencies = [
  "reth-primitives",
  "reth-rpc-types",
  "tokio",
+ "tracing",
 ]
 
 [[package]]

bin/reth/src/node/mod.rs

@@ -20,7 +20,11 @@ use reth_network::{
 use reth_primitives::{hex_literal::hex, Bytes, Header, H256, U256};
 use reth_provider::{db_provider::ProviderImpl, BlockProvider, HeaderProvider};
 use reth_stages::stages::{bodies::BodyStage, headers::HeaderStage, senders::SendersStage};
-use std::{path::Path, str::FromStr, sync::Arc};
+use std::{
+    path::Path,
+    str::FromStr,
+    sync::Arc,
+};
 use tracing::{debug, info};
 
 /// Start the client

crates/interfaces/src/p2p/error.rs

@@ -1,4 +1,4 @@
-use reth_primitives::WithPeerId;
+use reth_primitives::{WithPeerId, H256};
 use tokio::sync::{mpsc, oneshot};
 
 /// Result alias for result of a request.
@@ -23,6 +23,10 @@ pub enum RequestError {
     Timeout,
     #[error("Received bad response.")]
     BadResponse,
+    #[error("Received empty headers.")]
+    EmptyHeaders,
+    #[error("Parent hash {0:?} does not match forkchoice head hash {1:?}")]
+    MismatchedParent(H256, H256),
 }
 
 // === impl RequestError ===

crates/net/downloaders/Cargo.toml

@@ -21,6 +21,7 @@ futures-util = "0.3.25"
 
 # misc
 backon = "0.2.0"
+tracing = "0.1.37"
 
 [dev-dependencies]
 reth-interfaces = { path = "../../interfaces", features = ["test-utils"] }

crates/net/downloaders/src/headers/linear.rs

@@ -208,6 +208,12 @@ where
         if !fut.inc_err() {
             return Err(())
         }
+        tracing::trace!(
+            target : "downloaders::headers",
+            "retrying future attempt: {}/{}",
+            fut.retries,
+            fut.max_retries
+        );
         let req = self.headers_request();
         fut.request = req.clone();
         let client = Arc::clone(&self.client);
@@ -235,7 +241,7 @@ where
                 headers.sort_unstable_by_key(|h| h.number);
 
                 if headers.is_empty() {
-                    return Err(RequestError::BadResponse.into())
+                    return Err(RequestError::EmptyHeaders.into())
                 }
 
                 // Iterate headers in reverse
@@ -255,7 +261,11 @@ where
                     } else if parent.hash() != self.forkchoice.head_block_hash {
                         // The buffer is empty and the first header does not match the
                         // tip, requeue the future
-                        return Err(RequestError::BadResponse.into())
+                        return Err(RequestError::MismatchedParent(
+                            parent.hash(),
+                            self.forkchoice.head_block_hash,
+                        )
+                        .into())
                     }
 
                     // Record new parent
@@ -328,6 +338,10 @@ where
             if let Poll::Ready(result) = fut.poll_unpin(cx) {
                 if let Err(err) = this.process_header_response(result) {
                     if this.try_fuse_request_fut(&mut fut).is_err() {
+                        tracing::trace!(
+                            target: "downloaders::headers",
+                            "ran out of retries terminating stream"
+                        );
                         // We exhausted all of the retries. Stream must terminate
                         this.done = true;
                         this.buffered.clear();

crates/net/network/src/manager.rs

@@ -50,7 +50,7 @@ use std::{
 };
 use tokio::sync::mpsc;
 use tokio_stream::wrappers::UnboundedReceiverStream;
-use tracing::{error, trace, warn};
+use tracing::{error, info, trace, warn};
 
 /// Manages the _entire_ state of the network.
 ///
@@ -518,7 +518,7 @@ where
                     direction,
                 } => {
                     let total_active = this.num_active_peers.fetch_add(1, Ordering::Relaxed) + 1;
-                    trace!(
+                    info!(
                         target : "net",
                         ?remote_addr,
                         ?peer_id,
@@ -601,6 +601,11 @@ where
                         .apply_reputation_change(&peer_id, ReputationChangeKind::FailedToConnect);
                 }
                 SwarmEvent::StatusUpdate(status) => {
+                    trace!(
+                        target : "net",
+                        ?status,
+                        "Status Update received"
+                    );
                     this.swarm.sessions_mut().on_status_update(status.clone())
                 }
             }

crates/net/network/src/state.rs

@@ -24,7 +24,7 @@ use std::{
     task::{Context, Poll},
 };
 use tokio::sync::oneshot;
-use tracing::trace;
+use tracing::error;
 
 /// Cache limit of blocks to keep track of for a single peer.
 const PEER_BLOCK_CACHE_LIMIT: usize = 512;
@@ -355,10 +355,11 @@ where
             for (id, peer) in self.active_peers.iter_mut() {
                 if let Some(mut response) = peer.pending_response.take() {
                     match response.poll(cx) {
-                        Poll::Ready(Err(_)) => {
-                            trace!(
+                        Poll::Ready(Err(err)) => {
+                            error!(
                                 target : "net",
                                 ?id,
+                                ?err,
                                 "Request canceled, response channel closed."
                             );
                             disconnect_sessions.push(*id);