Improve logs for substream limits

Author: teor2345

core/src/muxing.rs

@@ -56,7 +56,7 @@ use std::future::Future;
 use std::pin::Pin;
 
 pub use self::boxed::StreamMuxerBox;
-pub use self::boxed::SubstreamBox;
+pub use self::boxed::{AsyncReadWrite, SubstreamBox};
 
 mod boxed;
 

core/src/muxing/boxed.rs

@@ -152,7 +152,7 @@ impl fmt::Debug for SubstreamBox {
 }
 
 /// Workaround because Rust does not allow `Box<dyn AsyncRead + AsyncWrite>`.
-trait AsyncReadWrite: AsyncRead + AsyncWrite {
+pub trait AsyncReadWrite: AsyncRead + AsyncWrite {
     /// Helper function to capture the erased inner type.
     ///
     /// Used to make the [`Debug`] implementation of [`SubstreamBox`] more useful.

misc/multistream-select/src/length_delimited.rs

@@ -42,7 +42,7 @@ const DEFAULT_BUFFER_SIZE: usize = 64;
 pub(crate) struct LengthDelimited<R> {
     /// The inner I/O resource.
     #[pin]
-    inner: R,
+    pub(crate) inner: R,
     /// Read buffer for a single incoming unsigned-varint length-delimited frame.
     read_buffer: BytesMut,
     /// Write buffer for outgoing unsigned-varint length-delimited frames.
@@ -298,7 +298,7 @@ where
 #[derive(Debug)]
 pub(crate) struct LengthDelimitedReader<R> {
     #[pin]
-    inner: LengthDelimited<R>,
+    pub(crate) inner: LengthDelimited<R>,
 }
 
 impl<R> LengthDelimitedReader<R> {

misc/multistream-select/src/lib.rs

@@ -95,7 +95,7 @@ mod protocol;
 
 pub use self::dialer_select::{dialer_select_proto, DialerSelectFuture};
 pub use self::listener_select::{listener_select_proto, ListenerSelectFuture};
-pub use self::negotiated::{Negotiated, NegotiatedComplete, NegotiationError};
+pub use self::negotiated::{Negotiated, NegotiatedComplete, NegotiationError, State};
 pub use self::protocol::ProtocolError;
 
 /// Supported multistream-select versions.

misc/multistream-select/src/negotiated.rs

@@ -48,7 +48,7 @@ use std::{
 #[derive(Debug)]
 pub struct Negotiated<TInner> {
     #[pin]
-    state: State<TInner>,
+    pub state: State<TInner>,
 }
 
 /// A `Future` that waits on the completion of protocol negotiation.
@@ -197,7 +197,8 @@ impl<TInner> Negotiated<TInner> {
 /// The states of a `Negotiated` I/O stream.
 #[pin_project(project = StateProj)]
 #[derive(Debug)]
-enum State<R> {
+#[allow(private_interfaces)]
+pub enum State<R> {
     /// In this state, a `Negotiated` is still expecting to
     /// receive confirmation of the protocol it has optimistically
     /// settled on.
@@ -224,6 +225,26 @@ enum State<R> {
     Invalid,
 }
 
+impl<R> State<R> {
+    /// Returns the underlying protocol, if negotiation is in progress.
+    pub fn protocol(&self) -> Option<&str> {
+        match self {
+            State::Expecting { protocol, .. } => Some(protocol.as_ref()),
+            State::Completed { .. } => None,
+            State::Invalid => None,
+        }
+    }
+
+    /// Returns the underlying stream, if it is available.
+    pub fn inner(&self) -> Option<&R> {
+        match self {
+            State::Expecting { io, .. } => Some(&io.inner.inner.inner),
+            State::Completed { io } => Some(io),
+            State::Invalid => None,
+        }
+    }
+}
+
 impl<TInner> AsyncRead for Negotiated<TInner>
 where
     TInner: AsyncRead + AsyncWrite + Unpin,

misc/multistream-select/src/protocol.rs

@@ -321,7 +321,7 @@ where
 #[derive(Debug)]
 pub(crate) struct MessageReader<R> {
     #[pin]
-    inner: LengthDelimitedReader<R>,
+    pub(crate) inner: LengthDelimitedReader<R>,
 }
 
 impl<R> MessageReader<R> {

misc/quick-protobuf-codec/src/lib.rs

@@ -15,6 +15,7 @@ pub use generated::test as proto;
 ///
 /// to prefix messages with their length and uses [`quick_protobuf`] and a provided
 /// `struct` implementing [`MessageRead`] and [`MessageWrite`] to do the encoding.
+#[derive(Debug)]
 pub struct Codec<In, Out = In> {
     max_message_len_bytes: usize,
     phantom: PhantomData<(In, Out)>,

protocols/kad/src/handler.rs

@@ -35,7 +35,7 @@ use libp2p_swarm::{
     ConnectionHandler, ConnectionHandlerEvent, Stream, StreamUpgradeError, SubstreamProtocol,
     SupportedProtocols,
 };
-use std::collections::VecDeque;
+use std::collections::{HashMap, VecDeque};
 use std::task::Waker;
 use std::time::Duration;
 use std::{error, fmt, io, marker::PhantomData, pin::Pin, task::Context, task::Poll};
@@ -98,6 +98,7 @@ struct ProtocolStatus {
 }
 
 /// State of an active inbound substream.
+#[derive(Debug)]
 enum InboundSubstreamState {
     /// Waiting for a request from the remote.
     WaitingMessage {
@@ -173,6 +174,23 @@ impl InboundSubstreamState {
             InboundSubstreamState::Poisoned { .. } => unreachable!(),
         }
     }
+
+    /// Returns the `KadInStreamSink` associated with this state, or `None` if the state has no substream.
+    fn sink(&self) -> Option<&KadInStreamSink<Stream>> {
+        match self {
+            InboundSubstreamState::WaitingMessage { substream, .. }
+            | InboundSubstreamState::WaitingBehaviour(_, substream, _)
+            | InboundSubstreamState::PendingSend(_, substream, _)
+            | InboundSubstreamState::PendingFlush(_, substream)
+            | InboundSubstreamState::Closing(substream) => Some(substream),
+            InboundSubstreamState::Cancelled | InboundSubstreamState::Poisoned { .. } => None,
+        }
+    }
+
+    /// Returns the `Stream` associated with this state, or `None` if the state has no substream.
+    fn stream(&self) -> Option<&Stream> {
+        self.sink().map(|v| &**v)
+    }
 }
 
 /// Event produced by the Kademlia handler.
@@ -524,15 +542,23 @@ impl Handler {
                     InboundSubstreamState::WaitingMessage { first: false, .. }
                 )
             }) {
+                let prev_state = format!("{s:?}");
+
                 *s = InboundSubstreamState::Cancelled;
+
                 tracing::debug!(
                     peer=?self.remote_peer_id,
+                    ?prev_state,
+                    ?MAX_NUM_STREAMS,
+                    active_streams=?self.debug_inbound_substreams(),
                     "New inbound substream to peer exceeds inbound substream limit. \
                     Removed older substream waiting to be reused."
-                )
+                );
             } else {
                 tracing::warn!(
                     peer=?self.remote_peer_id,
+                    ?MAX_NUM_STREAMS,
+                    active_streams=?self.debug_inbound_substreams(),
                     "New inbound substream to peer exceeds inbound substream limit. \
                      No older substream waiting to be reused. Dropping new substream."
                 );
@@ -550,6 +576,30 @@ impl Handler {
             });
     }
 
+    /// Returns a summary of the protocols of the inbound substreams.
+    fn debug_inbound_substreams(&self) -> String {
+        use libp2p_core::muxing::AsyncReadWrite;
+
+        let mut protocols = HashMap::<_, usize>::new();
+        let mut stream_types = HashMap::<_, usize>::new();
+
+        for substream in &self.inbound_substreams {
+            let state = substream.stream().map(|s| &s.stream.state);
+            let protocol = state.and_then(|s| s.protocol());
+            let stream_type = state.and_then(|s| s.inner()).map(|i| i.type_name());
+
+            if let Some(protocol) = protocol {
+                *protocols.entry(protocol).or_default() += 1;
+            }
+
+            if let Some(stream_type) = stream_type {
+                *stream_types.entry(stream_type).or_default() += 1;
+            }
+        }
+
+        format!("protocols: {protocols:?}, stream types: {stream_types:?}")
+    }
+
     /// Takes the given [`KadRequestMsg`] and composes it into an outbound request-response protocol handshake using a [`oneshot::channel`].
     fn queue_new_stream(&mut self, id: QueryId, msg: KadRequestMsg) {
         let (sender, receiver) = oneshot::channel();

protocols/kad/src/protocol.rs

@@ -199,6 +199,7 @@ impl UpgradeInfo for ProtocolConfig {
 }
 
 /// Codec for Kademlia inbound and outbound message framing.
+#[derive(Debug)]
 pub struct Codec<A, B> {
     codec: quick_protobuf_codec::Codec<proto::Message>,
     __phantom: PhantomData<(A, B)>,

swarm/src/stream.rs

@@ -9,9 +9,17 @@ use std::{
 };
 
 /// Counter for the number of active streams on a connection.
-#[derive(Debug, Clone)]
+#[derive(Clone)]
 pub(crate) struct ActiveStreamCounter(Arc<()>);
 
+impl std::fmt::Debug for ActiveStreamCounter {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_tuple("ActiveStreamCounter")
+            .field(&self.num_alive_streams())
+            .finish()
+    }
+}
+
 impl ActiveStreamCounter {
     pub(crate) fn default() -> Self {
         Self(Arc::new(()))
@@ -28,7 +36,7 @@ impl ActiveStreamCounter {
 
 #[derive(Debug)]
 pub struct Stream {
-    stream: Negotiated<SubstreamBox>,
+    pub stream: Negotiated<SubstreamBox>,
     counter: Option<ActiveStreamCounter>,
 }