feat(gossipsub): log when sending and receiving gossipsub messages

To help with the upcoming [partial messages](sigp#577)

Pull-Request: #6234.

Author: jxs

protocols/gossipsub/CHANGELOG.md

@@ -1,4 +1,7 @@
 ## 0.50.0
+- Log when sending and receiving messages.
+  See [PR 6234](https://github.com/libp2p/rust-libp2p/pull/6234)
+
 - Prevent mesh exceeding mesh_n_high.
   See [PR 6184](https://github.com/libp2p/rust-libp2p/pull/6184)
 

protocols/gossipsub/src/behaviour.rs

@@ -3140,6 +3140,7 @@ where
         // This clones a reference to the Queue so any new handlers reference the same underlying
         // queue. No data is actually cloned here.
         Ok(Handler::new(
+            peer_id,
             self.config.protocol_config(),
             connected_peer.messages.clone(),
         ))
@@ -3169,6 +3170,7 @@ where
         // This clones a reference to the Queue so any new handlers reference the same underlying
         // queue. No data is actually cloned here.
         Ok(Handler::new(
+            peer_id,
             self.config.protocol_config(),
             connected_peer.messages.clone(),
         ))
@@ -3228,6 +3230,7 @@ where
                 rpc,
                 invalid_messages,
             } => {
+                tracing::debug!(peer=%propagation_source, message=?rpc, "Received gossipsub message");
                 // Handle the gossipsub RPC
 
                 // Handle subscriptions

protocols/gossipsub/src/handler.rs

@@ -25,7 +25,7 @@ use std::{
 
 use asynchronous_codec::Framed;
 use futures::{future::Either, prelude::*, StreamExt};
-use libp2p_core::upgrade::DeniedUpgrade;
+use libp2p_core::{upgrade::DeniedUpgrade, PeerId};
 use libp2p_swarm::{
     handler::{
         ConnectionEvent, ConnectionHandler, ConnectionHandlerEvent, DialUpgradeError,
@@ -89,6 +89,9 @@ pub enum Handler {
 
 /// Protocol Handler that manages a single long-lived substream with a peer.
 pub struct EnabledHandler {
+    /// Remote `PeerId` for this `ConnectionHandler`.
+    peer_id: PeerId,
+
     /// Upgrade configuration for the gossipsub protocol.
     listen_protocol: ProtocolConfig,
 
@@ -162,8 +165,13 @@ enum OutboundSubstreamState {
 
 impl Handler {
     /// Builds a new [`Handler`].
-    pub(crate) fn new(protocol_config: ProtocolConfig, message_queue: Queue) -> Self {
+    pub(crate) fn new(
+        peer_id: PeerId,
+        protocol_config: ProtocolConfig,
+        message_queue: Queue,
+    ) -> Self {
         Handler::Enabled(EnabledHandler {
+            peer_id,
             listen_protocol: protocol_config,
             inbound_substream: None,
             outbound_substream: None,
@@ -254,6 +262,7 @@ impl EnabledHandler {
                 Some(OutboundSubstreamState::WaitingOutput(substream)) => {
                     if let Poll::Ready(mut message) = Pin::new(&mut self.message_queue).poll_pop(cx)
                     {
+                        tracing::debug!(peer=%self.peer_id, ?message, "Sending gossipsub message");
                         match message {
                             RpcOut::Publish {
                                 message: _,