From 10861b9d2eabba391d78930f0b024222dd0eee66 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 9 Sep 2022 00:10:21 +1000
Subject: [PATCH 01/16] Add tests

---
 core/src/muxing/boxed.rs | 126 +++++++++++++++++++++++++++++++++++++++
 1 file changed, 126 insertions(+)

diff --git a/core/src/muxing/boxed.rs b/core/src/muxing/boxed.rs
index 99f7a87c6a5..7f9d7e2f7f3 100644
--- a/core/src/muxing/boxed.rs
+++ b/core/src/muxing/boxed.rs
@@ -94,6 +94,18 @@ impl StreamMuxerBox {
         }
     }
 
+    /// Returns the number of active inbound streams i.e. streams that
+    /// have been returned from [`StreamMuxer::poll_inbound`] and have not been dropped since.
+    pub fn active_inbound_streams(&self) -> usize {
+        todo!()
+    }
+
+    /// Returns the number of active outbound streams i.e. streams that
+    /// have been returned from [`StreamMuxer::poll_outbound`] and have not been dropped since.
+    pub fn active_outbound_streams(&self) -> usize {
+        todo!()
+    }
+
     fn project(
         self: Pin<&mut Self>,
     ) -> Pin<&mut (dyn StreamMuxer<Substream = SubstreamBox, Error = io::Error> + Send)> {
@@ -205,3 +217,117 @@ impl AsyncWrite for SubstreamBox {
         self.0.as_mut().poll_close(cx)
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::muxing::StreamMuxerExt;
+
+    #[async_std::test]
+    async fn stream_muxer_box_tracks_alive_inbound_streams() {
+        let mut muxer = StreamMuxerBox::new(DummyStreamMuxer);
+
+        let _stream1 = muxer.next_inbound().await.unwrap();
+        let _stream2 = muxer.next_inbound().await.unwrap();
+
+        assert_eq!(muxer.active_inbound_streams(), 2);
+
+        drop(_stream1);
+
+        assert_eq!(muxer.active_inbound_streams(), 1);
+    }
+
+    #[async_std::test]
+    async fn stream_muxer_box_tracks_alive_outbound_streams() {
+        let mut muxer = StreamMuxerBox::new(DummyStreamMuxer);
+
+        let _stream1 = muxer.next_outbound().await.unwrap();
+        let _stream2 = muxer.next_outbound().await.unwrap();
+
+        assert_eq!(muxer.active_outbound_streams(), 2);
+
+        drop(_stream1);
+
+        assert_eq!(muxer.active_outbound_streams(), 1);
+    }
+
+    #[test]
+    fn stream_muxer_box_starts_with_zero_active_inbound_streams() {
+        let muxer = StreamMuxerBox::new(DummyStreamMuxer);
+
+        let num_active_inbound_streams = muxer.active_inbound_streams();
+
+        assert_eq!(num_active_inbound_streams, 0);
+    }
+
+    #[test]
+    fn stream_muxer_box_starts_with_zero_active_outbound_streams() {
+        let muxer = StreamMuxerBox::new(DummyStreamMuxer);
+
+        let num_active_outbound_streams = muxer.active_outbound_streams();
+
+        assert_eq!(num_active_outbound_streams, 0);
+    }
+
+    struct DummyStreamMuxer;
+
+    impl StreamMuxer for DummyStreamMuxer {
+        type Substream = PendingSubstream;
+        type Error = void::Void;
+
+        fn poll_inbound(
+            self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+        ) -> Poll<Result<Self::Substream, Self::Error>> {
+            Poll::Ready(Ok(PendingSubstream))
+        }
+
+        fn poll_outbound(
+            self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+        ) -> Poll<Result<Self::Substream, Self::Error>> {
+            Poll::Ready(Ok(PendingSubstream))
+        }
+
+        fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            Poll::Pending
+        }
+
+        fn poll(
+            self: Pin<&mut Self>,
+            _: &mut Context<'_>,
+        ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
+            Poll::Pending
+        }
+    }
+
+    struct PendingSubstream;
+
+    impl AsyncRead for PendingSubstream {
+        fn poll_read(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &mut [u8],
+        ) -> Poll<io::Result<usize>> {
+            Poll::Pending
+        }
+    }
+
+    impl AsyncWrite for PendingSubstream {
+        fn poll_write(
+            self: Pin<&mut Self>,
+            _cx: &mut Context<'_>,
+            _buf: &[u8],
+        ) -> Poll<io::Result<usize>> {
+            Poll::Pending
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Pending
+        }
+
+        fn poll_close(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Pending
+        }
+    }
+}

From dbd0e24446970dfa32f2ac26f6cb81756b7eb0bf Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 9 Sep 2022 00:27:37 +1000
Subject: [PATCH 02/16] Track number of active inbound and outbound streams

---
 core/CHANGELOG.md                       |   6 +
 core/Cargo.toml                         |   2 +-
 core/src/muxing/boxed.rs                | 151 ++++++++++++++++--------
 swarm/src/connection/handler_wrapper.rs |  13 +-
 4 files changed, 118 insertions(+), 54 deletions(-)

diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md
index 4a5e07f14f8..8f9ebbbb17c 100644
--- a/core/CHANGELOG.md
+++ b/core/CHANGELOG.md
@@ -1,3 +1,9 @@
+# 0.36.1
+
+- Add `StreamMuxerBox::active_inbound_streams` and `StreamMuxerBox::active_outbound_streams`. See [PR XXXX]. 
+
+[PR XXXX]: https://github.com/libp2p/rust-libp2p/pull/XXXX
+
 # 0.36.0
 
 - Make RSA keypair support optional. To enable RSA support, `rsa` feature should be enabled.
diff --git a/core/Cargo.toml b/core/Cargo.toml
index 3e5285ef6f4..2b5e0b1c2ab 100644
--- a/core/Cargo.toml
+++ b/core/Cargo.toml
@@ -3,7 +3,7 @@ name = "libp2p-core"
 edition = "2021"
 rust-version = "1.56.1"
 description = "Core traits and structs of libp2p"
-version = "0.36.0"
+version = "0.36.1"
 authors = ["Parity Technologies <admin@parity.io>"]
 license = "MIT"
 repository = "https://github.com/libp2p/rust-libp2p"
diff --git a/core/src/muxing/boxed.rs b/core/src/muxing/boxed.rs
index 7f9d7e2f7f3..1d40c26378c 100644
--- a/core/src/muxing/boxed.rs
+++ b/core/src/muxing/boxed.rs
@@ -6,18 +6,27 @@ use std::fmt;
 use std::io;
 use std::io::{IoSlice, IoSliceMut};
 use std::pin::Pin;
+use std::sync::{Arc, Weak};
 use std::task::{Context, Poll};
 
 /// Abstract `StreamMuxer`.
 pub struct StreamMuxerBox {
-    inner: Pin<Box<dyn StreamMuxer<Substream = SubstreamBox, Error = io::Error> + Send>>,
+    inner: Pin<
+        Box<
+            dyn private::InstrumentedStreamMuxer<Substream = SubstreamBox, Error = io::Error>
+                + Send,
+        >,
+    >,
 }
 
 /// Abstract type for asynchronous reading and writing.
 ///
 /// A [`SubstreamBox`] erases the concrete type it is given and only retains its `AsyncRead`
 /// and `AsyncWrite` capabilities.
-pub struct SubstreamBox(Pin<Box<dyn AsyncReadWrite + Send>>);
+pub struct SubstreamBox {
+    inner: Pin<Box<dyn AsyncReadWrite + Send>>,
+    _counter: Weak<()>,
+}
 
 #[pin_project]
 struct Wrap<T>
@@ -26,49 +35,79 @@ where
 {
     #[pin]
     inner: T,
+    inbound_counter: Arc<()>,
+    outbound_counter: Arc<()>,
 }
 
-impl<T> StreamMuxer for Wrap<T>
-where
-    T: StreamMuxer,
-    T::Substream: Send + 'static,
-    T::Error: Send + Sync + 'static,
-{
-    type Substream = SubstreamBox;
-    type Error = io::Error;
+mod private {
+    use super::*;
 
-    fn poll_inbound(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Self::Substream, Self::Error>> {
-        self.project()
-            .inner
-            .poll_inbound(cx)
-            .map_ok(SubstreamBox::new)
-            .map_err(into_io_error)
+    pub trait InstrumentedStreamMuxer: StreamMuxer {
+        fn active_inbound_streams(&self) -> usize;
+        fn active_outbound_streams(&self) -> usize;
     }
 
-    fn poll_outbound(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<Self::Substream, Self::Error>> {
-        self.project()
-            .inner
-            .poll_outbound(cx)
-            .map_ok(SubstreamBox::new)
-            .map_err(into_io_error)
-    }
+    impl<T> StreamMuxer for Wrap<T>
+    where
+        T: StreamMuxer,
+        T::Substream: Send + 'static,
+        T::Error: Send + Sync + 'static,
+    {
+        type Substream = SubstreamBox;
+        type Error = io::Error;
 
-    #[inline]
-    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
-        self.project().inner.poll_close(cx).map_err(into_io_error)
+        fn poll_inbound(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<Self::Substream, Self::Error>> {
+            let counter = Arc::downgrade(&self.inbound_counter);
+
+            self.project()
+                .inner
+                .poll_inbound(cx)
+                .map_ok(|stream| SubstreamBox::new(stream, counter))
+                .map_err(into_io_error)
+        }
+
+        fn poll_outbound(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<Self::Substream, Self::Error>> {
+            let counter = Arc::downgrade(&self.outbound_counter);
+
+            self.project()
+                .inner
+                .poll_outbound(cx)
+                .map_ok(|stream| SubstreamBox::new(stream, counter))
+                .map_err(into_io_error)
+        }
+
+        #[inline]
+        fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+            self.project().inner.poll_close(cx).map_err(into_io_error)
+        }
+
+        fn poll(
+            self: Pin<&mut Self>,
+            cx: &mut Context<'_>,
+        ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
+            self.project().inner.poll(cx).map_err(into_io_error)
+        }
     }
 
-    fn poll(
-        self: Pin<&mut Self>,
-        cx: &mut Context<'_>,
-    ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
-        self.project().inner.poll(cx).map_err(into_io_error)
+    impl<T> InstrumentedStreamMuxer for Wrap<T>
+    where
+        T: StreamMuxer,
+        T::Substream: Send + 'static,
+        T::Error: Send + Sync + 'static,
+    {
+        fn active_inbound_streams(&self) -> usize {
+            Arc::weak_count(&self.inbound_counter)
+        }
+
+        fn active_outbound_streams(&self) -> usize {
+            Arc::weak_count(&self.outbound_counter)
+        }
     }
 }
 
@@ -87,7 +126,11 @@ impl StreamMuxerBox {
         T::Substream: Send + 'static,
         T::Error: Send + Sync + 'static,
     {
-        let wrap = Wrap { inner: muxer };
+        let wrap = Wrap {
+            inner: muxer,
+            inbound_counter: Arc::new(()),
+            outbound_counter: Arc::new(()),
+        };
 
         StreamMuxerBox {
             inner: Box::pin(wrap),
@@ -97,18 +140,21 @@ impl StreamMuxerBox {
     /// Returns the number of active inbound streams i.e. streams that
     /// have been returned from [`StreamMuxer::poll_inbound`] and have not been dropped since.
     pub fn active_inbound_streams(&self) -> usize {
-        todo!()
+        self.inner.active_inbound_streams()
     }
 
     /// Returns the number of active outbound streams i.e. streams that
     /// have been returned from [`StreamMuxer::poll_outbound`] and have not been dropped since.
     pub fn active_outbound_streams(&self) -> usize {
-        todo!()
+        self.inner.active_outbound_streams()
     }
 
     fn project(
         self: Pin<&mut Self>,
-    ) -> Pin<&mut (dyn StreamMuxer<Substream = SubstreamBox, Error = io::Error> + Send)> {
+    ) -> Pin<
+        &mut (dyn private::InstrumentedStreamMuxer<Substream = SubstreamBox, Error = io::Error>
+                  + Send),
+    > {
         self.get_mut().inner.as_mut()
     }
 }
@@ -146,14 +192,19 @@ impl StreamMuxer for StreamMuxerBox {
 
 impl SubstreamBox {
     /// Construct a new [`SubstreamBox`] from something that implements [`AsyncRead`] and [`AsyncWrite`].
-    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S) -> Self {
-        Self(Box::pin(stream))
+    ///
+    /// TOOD: Remove this constructor.
+    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S, counter: Weak<()>) -> Self {
+        Self {
+            inner: Box::pin(stream),
+            _counter: counter,
+        }
     }
 }
 
 impl fmt::Debug for SubstreamBox {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        write!(f, "SubstreamBox({})", self.0.type_name())
+        write!(f, "SubstreamBox({})", self.inner.type_name())
     }
 }
 
@@ -180,7 +231,7 @@ impl AsyncRead for SubstreamBox {
         cx: &mut Context<'_>,
         buf: &mut [u8],
     ) -> Poll<std::io::Result<usize>> {
-        self.0.as_mut().poll_read(cx, buf)
+        self.inner.as_mut().poll_read(cx, buf)
     }
 
     fn poll_read_vectored(
@@ -188,7 +239,7 @@ impl AsyncRead for SubstreamBox {
         cx: &mut Context<'_>,
         bufs: &mut [IoSliceMut<'_>],
     ) -> Poll<std::io::Result<usize>> {
-        self.0.as_mut().poll_read_vectored(cx, bufs)
+        self.inner.as_mut().poll_read_vectored(cx, bufs)
     }
 }
 
@@ -198,7 +249,7 @@ impl AsyncWrite for SubstreamBox {
         cx: &mut Context<'_>,
         buf: &[u8],
     ) -> Poll<std::io::Result<usize>> {
-        self.0.as_mut().poll_write(cx, buf)
+        self.inner.as_mut().poll_write(cx, buf)
     }
 
     fn poll_write_vectored(
@@ -206,15 +257,15 @@ impl AsyncWrite for SubstreamBox {
         cx: &mut Context<'_>,
         bufs: &[IoSlice<'_>],
     ) -> Poll<std::io::Result<usize>> {
-        self.0.as_mut().poll_write_vectored(cx, bufs)
+        self.inner.as_mut().poll_write_vectored(cx, bufs)
     }
 
     fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        self.0.as_mut().poll_flush(cx)
+        self.inner.as_mut().poll_flush(cx)
     }
 
     fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
-        self.0.as_mut().poll_close(cx)
+        self.inner.as_mut().poll_close(cx)
     }
 }
 
diff --git a/swarm/src/connection/handler_wrapper.rs b/swarm/src/connection/handler_wrapper.rs
index 03d09b3fbc1..bba940c3454 100644
--- a/swarm/src/connection/handler_wrapper.rs
+++ b/swarm/src/connection/handler_wrapper.rs
@@ -448,6 +448,8 @@ mod tests {
     use quickcheck::*;
     use std::sync::Arc;
 
+    // TODO: The implementation of this test is weird now.
+    // Merge https://github.com/libp2p/rust-libp2p/pull/2861 first which changes this test and we no longer rely on the `SubstreamBox::new` API.
     #[test]
     fn max_negotiating_inbound_streams() {
         fn prop(max_negotiating_inbound_streams: u8) {
@@ -465,8 +467,10 @@ mod tests {
             let alive_substreams_counter = Arc::new(());
 
             for _ in 0..max_negotiating_inbound_streams {
-                let substream =
-                    SubstreamBox::new(PendingSubstream(alive_substreams_counter.clone()));
+                let substream = SubstreamBox::new(
+                    PendingSubstream(alive_substreams_counter.clone()),
+                    Arc::downgrade(&Arc::new(())),
+                );
                 wrapper.inject_substream(substream, SubstreamEndpoint::Listener);
             }
 
@@ -476,7 +480,10 @@ mod tests {
                 "Expect none of the substreams up to the limit to be dropped."
             );
 
-            let substream = SubstreamBox::new(PendingSubstream(alive_substreams_counter.clone()));
+            let substream = SubstreamBox::new(
+                PendingSubstream(alive_substreams_counter.clone()),
+                Arc::downgrade(&Arc::new(())),
+            );
             wrapper.inject_substream(substream, SubstreamEndpoint::Listener);
 
             assert_eq!(

From 21ca44684fdfef78f3d88ad32a3fe0319771694b Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 9 Sep 2022 00:29:57 +1000
Subject: [PATCH 03/16] Update changelog with PR number

---
 core/CHANGELOG.md | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md
index 8f9ebbbb17c..1cba51aff51 100644
--- a/core/CHANGELOG.md
+++ b/core/CHANGELOG.md
@@ -1,8 +1,8 @@
 # 0.36.1
 
-- Add `StreamMuxerBox::active_inbound_streams` and `StreamMuxerBox::active_outbound_streams`. See [PR XXXX]. 
+- Add `StreamMuxerBox::active_inbound_streams` and `StreamMuxerBox::active_outbound_streams`. See [PR 2878]. 
 
-[PR XXXX]: https://github.com/libp2p/rust-libp2p/pull/XXXX
+[PR 2878]: https://github.com/libp2p/rust-libp2p/pull/2878
 
 # 0.36.0
 

From f3e17b1fcc53a1a5766bfd4bf8780fe32bedc23b Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Thu, 22 Sep 2022 17:36:00 +1000
Subject: [PATCH 04/16] Make ctor module private

---
 core/CHANGELOG.md        | 3 ++-
 core/src/muxing/boxed.rs | 5 +----
 2 files changed, 3 insertions(+), 5 deletions(-)

diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md
index 1cba51aff51..804d55c8588 100644
--- a/core/CHANGELOG.md
+++ b/core/CHANGELOG.md
@@ -1,6 +1,7 @@
-# 0.36.1
+# 0.37.0
 
 - Add `StreamMuxerBox::active_inbound_streams` and `StreamMuxerBox::active_outbound_streams`. See [PR 2878]. 
+- Make `SubstreamBox::new` constructor module private. See [PR 2878].
 
 [PR 2878]: https://github.com/libp2p/rust-libp2p/pull/2878
 
diff --git a/core/src/muxing/boxed.rs b/core/src/muxing/boxed.rs
index 1d40c26378c..eecbce1f9cd 100644
--- a/core/src/muxing/boxed.rs
+++ b/core/src/muxing/boxed.rs
@@ -191,10 +191,7 @@ impl StreamMuxer for StreamMuxerBox {
 }
 
 impl SubstreamBox {
-    /// Construct a new [`SubstreamBox`] from something that implements [`AsyncRead`] and [`AsyncWrite`].
-    ///
-    /// TOOD: Remove this constructor.
-    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S, counter: Weak<()>) -> Self {
+    fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S, counter: Weak<()>) -> Self {
         Self {
             inner: Box::pin(stream),
             _counter: counter,

From 63f16d1bf6a3a2d89ad6cc5c9a1e4225ac180f48 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Thu, 22 Sep 2022 17:40:06 +1000
Subject: [PATCH 05/16] Use total number of inbound streams for back-pressure

---
 swarm/src/connection.rs | 24 +++++++++---------------
 1 file changed, 9 insertions(+), 15 deletions(-)

diff --git a/swarm/src/connection.rs b/swarm/src/connection.rs
index dc9a2eb92e3..e918b984f73 100644
--- a/swarm/src/connection.rs
+++ b/swarm/src/connection.rs
@@ -299,7 +299,7 @@ where
                 }
             }
 
-            if negotiating_in.len() < *max_negotiating_inbound_streams {
+            if muxing.active_inbound_streams() < *max_negotiating_inbound_streams {
                 match muxing.poll_inbound_unpin(cx)? {
                     Poll::Pending => {}
                     Poll::Ready(substream) => {
@@ -558,7 +558,6 @@ mod tests {
     use libp2p_core::upgrade::DeniedUpgrade;
     use libp2p_core::StreamMuxer;
     use quickcheck::*;
-    use std::sync::{Arc, Weak};
     use void::Void;
 
     #[test]
@@ -566,12 +565,8 @@ mod tests {
         fn prop(max_negotiating_inbound_streams: u8) {
             let max_negotiating_inbound_streams: usize = max_negotiating_inbound_streams.into();
 
-            let alive_substream_counter = Arc::new(());
-
             let mut connection = Connection::new(
-                StreamMuxerBox::new(DummyStreamMuxer {
-                    counter: alive_substream_counter.clone(),
-                }),
+                StreamMuxerBox::new(InfiniteStreamMuxer),
                 DummyConnectionHandler {
                     keep_alive: KeepAlive::Yes,
                 },
@@ -584,7 +579,7 @@ mod tests {
 
             assert!(result.is_pending());
             assert_eq!(
-                Arc::weak_count(&alive_substream_counter),
+                connection.muxing.active_inbound_streams(),
                 max_negotiating_inbound_streams,
                 "Expect no more than the maximum number of allowed streams"
             );
@@ -618,11 +613,10 @@ mod tests {
         ))
     }
 
-    struct DummyStreamMuxer {
-        counter: Arc<()>,
-    }
+    /// A [`StreamMuxer`] which produces an infinite number of streams.
+    struct InfiniteStreamMuxer;
 
-    impl StreamMuxer for DummyStreamMuxer {
+    impl StreamMuxer for InfiniteStreamMuxer {
         type Substream = PendingSubstream;
         type Error = Void;
 
@@ -630,14 +624,14 @@ mod tests {
             self: Pin<&mut Self>,
             _: &mut Context<'_>,
         ) -> Poll<Result<Self::Substream, Self::Error>> {
-            Poll::Ready(Ok(PendingSubstream(Arc::downgrade(&self.counter))))
+            Poll::Ready(Ok(PendingSubstream))
         }
 
         fn poll_outbound(
             self: Pin<&mut Self>,
             _: &mut Context<'_>,
         ) -> Poll<Result<Self::Substream, Self::Error>> {
-            Poll::Pending
+            Poll::Ready(Ok(PendingSubstream))
         }
 
         fn poll_close(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
@@ -685,7 +679,7 @@ mod tests {
         }
     }
 
-    struct PendingSubstream(Weak<()>);
+    struct PendingSubstream;
 
     impl AsyncRead for PendingSubstream {
         fn poll_read(

From 9fba25f28211c5e199cac0dac0dae8b33e3c7c5f Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Thu, 22 Sep 2022 17:47:00 +1000
Subject: [PATCH 06/16] Add changelog entry to swarm

---
 swarm/CHANGELOG.md | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/swarm/CHANGELOG.md b/swarm/CHANGELOG.md
index 3937c265d31..611937d7964 100644
--- a/swarm/CHANGELOG.md
+++ b/swarm/CHANGELOG.md
@@ -1,3 +1,16 @@
+# 0.40.0 [unreleased]
+
+- Use the total number of alive inbound streams for back-pressure. This can have a BIG impact on your application
+  depending on how it uses `libp2p`. Previously, the limit for inbound streams per connection only applied to the
+  _upgrade_ phase, i.e. for the time `InboundUpgrade` was running. Any stream being returned from `InboundUpgrade` and
+  given to the `ConnectionHandler` did not count towards that limit, essentially mitigating the back-pressure mechanism.
+  With this release, substreams count towards that limit until they are dropped and thus we actually enforce, how many
+  inbound streams can be active at one time _per connection_. `libp2p` will not accept any more incoming streams once
+  that limit is hit. If you experience stalls or unaccepted streams in your application, consider upping the limit via
+  `SwarmBuilder::max_negotiating_inbound_streams`. See [PR 2878].
+
+[PR 2878]: https://github.com/libp2p/rust-libp2p/pull/2878
+
 # 0.39.0
 
 - Remove deprecated `NetworkBehaviourEventProcess`. See [libp2p-swarm v0.38.0 changelog entry] for

From 40ab076ac88a16ec7e9c647935725b92a764753a Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 30 Sep 2022 10:39:19 +1000
Subject: [PATCH 07/16] Introduce `ConnectionHandler::max_inbound_streams`

This effectively deprecates `SwarmBuilder::max_negotiating_inbound_streams`.
---
 core/src/connection.rs                    |  7 ++
 protocols/request-response/src/handler.rs | 10 ++-
 protocols/request-response/src/lib.rs     | 13 ++++
 swarm/CHANGELOG.md                        |  6 +-
 swarm/src/behaviour/toggle.rs             |  9 ++-
 swarm/src/connection.rs                   | 84 +++++++++++++----------
 swarm/src/connection/pool.rs              | 12 ++--
 swarm/src/connection/pool/task.rs         |  9 +++
 swarm/src/handler.rs                      |  9 +++
 swarm/src/handler/dummy.rs                |  6 +-
 swarm/src/handler/either.rs               | 24 ++++---
 swarm/src/handler/map_in.rs               |  4 ++
 swarm/src/handler/map_out.rs              |  4 ++
 swarm/src/handler/multi.rs                |  8 +++
 swarm/src/handler/select.rs               | 40 ++++++-----
 swarm/src/lib.rs                          |  4 ++
 16 files changed, 175 insertions(+), 74 deletions(-)

diff --git a/core/src/connection.rs b/core/src/connection.rs
index 91008408fe2..7b16e4b5bdc 100644
--- a/core/src/connection.rs
+++ b/core/src/connection.rs
@@ -19,6 +19,7 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::multiaddr::{Multiaddr, Protocol};
+use std::fmt;
 
 /// Connection identifier.
 #[derive(Debug, Copy, Clone, Hash, PartialEq, Eq, PartialOrd, Ord)]
@@ -35,6 +36,12 @@ impl ConnectionId {
     }
 }
 
+impl fmt::Display for ConnectionId {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "{}", self.0)
+    }
+}
+
 impl std::ops::Add<usize> for ConnectionId {
     type Output = Self;
 
diff --git a/protocols/request-response/src/handler.rs b/protocols/request-response/src/handler.rs
index fd843b7d473..42148ccd175 100644
--- a/protocols/request-response/src/handler.rs
+++ b/protocols/request-response/src/handler.rs
@@ -62,6 +62,8 @@ where
     substream_timeout: Duration,
     /// The current connection keep-alive.
     keep_alive: KeepAlive,
+    /// The maximum number of allowed inbound streams.
+    max_inbound_streams: usize,
     /// A pending fatal error that results in the connection being closed.
     pending_error: Option<ConnectionHandlerUpgrErr<io::Error>>,
     /// Queue of events to emit in `poll()`.
@@ -93,6 +95,7 @@ where
         codec: TCodec,
         keep_alive_timeout: Duration,
         substream_timeout: Duration,
+        max_inbound_streams: usize,
         inbound_request_id: Arc<AtomicU64>,
     ) -> Self {
         Self {
@@ -101,6 +104,7 @@ where
             keep_alive: KeepAlive::Yes,
             keep_alive_timeout,
             substream_timeout,
+            max_inbound_streams,
             outbound: VecDeque::new(),
             inbound: FuturesUnordered::new(),
             pending_events: VecDeque::new(),
@@ -199,8 +203,8 @@ where
     type Error = ConnectionHandlerUpgrErr<io::Error>;
     type InboundProtocol = ResponseProtocol<TCodec>;
     type OutboundProtocol = RequestProtocol<TCodec>;
-    type OutboundOpenInfo = RequestId;
     type InboundOpenInfo = RequestId;
+    type OutboundOpenInfo = RequestId;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         // A channel for notifying the handler when the inbound
@@ -236,6 +240,10 @@ where
         SubstreamProtocol::new(proto, request_id).with_timeout(self.substream_timeout)
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        self.max_inbound_streams
+    }
+
     fn inject_fully_negotiated_inbound(&mut self, sent: bool, request_id: RequestId) {
         if sent {
             self.pending_events
diff --git a/protocols/request-response/src/lib.rs b/protocols/request-response/src/lib.rs
index c4e18d894fb..7bd3e2201be 100644
--- a/protocols/request-response/src/lib.rs
+++ b/protocols/request-response/src/lib.rs
@@ -267,6 +267,7 @@ impl fmt::Display for RequestId {
 pub struct RequestResponseConfig {
     request_timeout: Duration,
     connection_keep_alive: Duration,
+    max_pending_inbound_requests: usize,
 }
 
 impl Default for RequestResponseConfig {
@@ -274,6 +275,7 @@ impl Default for RequestResponseConfig {
         Self {
             connection_keep_alive: Duration::from_secs(10),
             request_timeout: Duration::from_secs(10),
+            max_pending_inbound_requests: libp2p_swarm::handler::DEFAULT_MAX_INBOUND_STREAMS,
         }
     }
 }
@@ -290,6 +292,16 @@ impl RequestResponseConfig {
         self.request_timeout = v;
         self
     }
+
+    /// Sets the maximum number of pending inbound requests.
+    ///
+    /// Once this limit is hit, we will stop accepting requests from the remote until an
+    /// inbound substream is closed. Either because you send a response, a timeout is hit or
+    /// the remote closed the stream.
+    pub fn set_max_pending_inbound_requests(&mut self, n: usize) -> &mut Self {
+        self.max_pending_inbound_requests = n;
+        self
+    }
 }
 
 /// A request/response protocol for some message codec.
@@ -573,6 +585,7 @@ where
             self.codec.clone(),
             self.config.connection_keep_alive,
             self.config.request_timeout,
+            self.config.max_pending_inbound_requests,
             self.next_inbound_id.clone(),
         )
     }
diff --git a/swarm/CHANGELOG.md b/swarm/CHANGELOG.md
index 611937d7964..9f5b44760ba 100644
--- a/swarm/CHANGELOG.md
+++ b/swarm/CHANGELOG.md
@@ -4,10 +4,10 @@
   depending on how it uses `libp2p`. Previously, the limit for inbound streams per connection only applied to the
   _upgrade_ phase, i.e. for the time `InboundUpgrade` was running. Any stream being returned from `InboundUpgrade` and
   given to the `ConnectionHandler` did not count towards that limit, essentially mitigating the back-pressure mechanism.
-  With this release, substreams count towards that limit until they are dropped and thus we actually enforce, how many
+  With this release, substreams count towards that limit until they are dropped, so we actually enforce, how many
   inbound streams can be active at one time _per connection_. `libp2p` will not accept any more incoming streams once
-  that limit is hit. If you experience stalls or unaccepted streams in your application, consider upping the limit via
-  `SwarmBuilder::max_negotiating_inbound_streams`. See [PR 2878].
+  that limit is hit. You can configure the number of allowed substreams by overriding
+  `ConnectionHandler::max_inbound_streams`. See [PR 2878].
 
 [PR 2878]: https://github.com/libp2p/rust-libp2p/pull/2878
 
diff --git a/swarm/src/behaviour/toggle.rs b/swarm/src/behaviour/toggle.rs
index 07a15d56da9..30ff1f91f23 100644
--- a/swarm/src/behaviour/toggle.rs
+++ b/swarm/src/behaviour/toggle.rs
@@ -277,8 +277,8 @@ where
     type InboundProtocol =
         EitherUpgrade<SendWrapper<TInner::InboundProtocol>, SendWrapper<DeniedUpgrade>>;
     type OutboundProtocol = TInner::OutboundProtocol;
-    type OutboundOpenInfo = TInner::OutboundOpenInfo;
     type InboundOpenInfo = Either<TInner::InboundOpenInfo, ()>;
+    type OutboundOpenInfo = TInner::OutboundOpenInfo;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         if let Some(inner) = self.inner.as_ref() {
@@ -294,6 +294,13 @@ where
         }
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        self.inner
+            .as_ref()
+            .map(|h| h.max_inbound_streams())
+            .unwrap_or(0)
+    }
+
     fn inject_fully_negotiated_inbound(
         &mut self,
         out: <Self::InboundProtocol as InboundUpgradeSend>::Output,
diff --git a/swarm/src/connection.rs b/swarm/src/connection.rs
index e918b984f73..483669169f1 100644
--- a/swarm/src/connection.rs
+++ b/swarm/src/connection.rs
@@ -58,12 +58,14 @@ pub struct Connected {
 }
 
 /// Event generated by a [`Connection`].
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq, Eq)]
 pub enum Event<T> {
     /// Event generated by the [`ConnectionHandler`].
     Handler(T),
     /// Address of the remote has changed.
     AddressChange(Multiaddr),
+    /// This connection is operating at the maximum number of allowed substreams.
+    SubstreamLimitHit { num_streams: usize },
 }
 
 /// A multiplexed connection to a peer with an associated [`ConnectionHandler`].
@@ -93,16 +95,10 @@ where
     shutdown: Shutdown,
     /// The substream upgrade protocol override, if any.
     substream_upgrade_protocol_override: Option<upgrade::Version>,
-    /// The maximum number of inbound streams concurrently negotiating on a
-    /// connection. New inbound streams exceeding the limit are dropped and thus
-    /// reset.
+    /// The maximum number of inbound streams a connection.
     ///
-    /// Note: This only enforces a limit on the number of concurrently
-    /// negotiating inbound streams. The total number of inbound streams on a
-    /// connection is the sum of negotiating and negotiated streams. A limit on
-    /// the total number of streams can be enforced at the
-    /// [`StreamMuxerBox`](libp2p_core::muxing::StreamMuxerBox) level.
-    max_negotiating_inbound_streams: usize,
+    /// New inbound streams exceeding the limit are dropped and thus reset.
+    legacy_max_negotiating_inbound_streams: Option<usize>,
     /// Contains all upgrades that are waiting for a new outbound substream.
     ///
     /// The upgrade timeout is already ticking here so this may fail in case the remote is not quick
@@ -136,7 +132,7 @@ where
         muxer: StreamMuxerBox,
         handler: THandler,
         substream_upgrade_protocol_override: Option<upgrade::Version>,
-        max_negotiating_inbound_streams: usize,
+        legacy_max_negotiating_inbound_streams: Option<usize>,
     ) -> Self {
         Connection {
             muxing: muxer,
@@ -145,7 +141,7 @@ where
             negotiating_out: Default::default(),
             shutdown: Shutdown::None,
             substream_upgrade_protocol_override,
-            max_negotiating_inbound_streams,
+            legacy_max_negotiating_inbound_streams,
             requested_substreams: Default::default(),
         }
     }
@@ -174,7 +170,7 @@ where
             negotiating_out,
             negotiating_in,
             shutdown,
-            max_negotiating_inbound_streams,
+            legacy_max_negotiating_inbound_streams,
             substream_upgrade_protocol_override,
         } = self.get_mut();
 
@@ -299,16 +295,24 @@ where
                 }
             }
 
-            if muxing.active_inbound_streams() < *max_negotiating_inbound_streams {
-                match muxing.poll_inbound_unpin(cx)? {
-                    Poll::Pending => {}
-                    Poll::Ready(substream) => {
-                        let protocol = handler.listen_protocol();
+            // Compute effective limit whilst we are backwards compatible with the old config.
+            let effective_substream_limit =
+                legacy_max_negotiating_inbound_streams.unwrap_or(handler.max_inbound_streams());
 
-                        negotiating_in.push(SubstreamUpgrade::new_inbound(substream, protocol));
+            if muxing.active_inbound_streams() >= effective_substream_limit {
+                return Poll::Ready(Ok(Event::SubstreamLimitHit {
+                    num_streams: muxing.active_inbound_streams(),
+                }));
+            }
 
-                        continue; // Go back to the top, handler can potentially make progress again.
-                    }
+            match muxing.poll_inbound_unpin(cx)? {
+                Poll::Pending => {}
+                Poll::Ready(substream) => {
+                    let protocol = handler.listen_protocol();
+
+                    negotiating_in.push(SubstreamUpgrade::new_inbound(substream, protocol));
+
+                    continue; // Go back to the top, handler can potentially make progress again.
                 }
             }
 
@@ -552,9 +556,8 @@ enum Shutdown {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::handler::DummyConnectionHandler;
-    use futures::AsyncRead;
     use futures::AsyncWrite;
+    use futures::{future, AsyncRead};
     use libp2p_core::upgrade::DeniedUpgrade;
     use libp2p_core::StreamMuxer;
     use quickcheck::*;
@@ -567,21 +570,24 @@ mod tests {
 
             let mut connection = Connection::new(
                 StreamMuxerBox::new(InfiniteStreamMuxer),
-                DummyConnectionHandler {
-                    keep_alive: KeepAlive::Yes,
-                },
+                MockConnectionHandler::new(
+                    Duration::from_secs(10),
+                    max_negotiating_inbound_streams,
+                ),
+                None,
                 None,
-                max_negotiating_inbound_streams,
             );
 
-            let result = Pin::new(&mut connection)
-                .poll(&mut Context::from_waker(futures::task::noop_waker_ref()));
+            let event = future::poll_fn(|cx| Pin::new(&mut connection).poll(cx))
+                .now_or_never()
+                .unwrap()
+                .unwrap();
 
-            assert!(result.is_pending());
             assert_eq!(
-                connection.muxing.active_inbound_streams(),
-                max_negotiating_inbound_streams,
-                "Expect no more than the maximum number of allowed streams"
+                event,
+                Event::SubstreamLimitHit {
+                    num_streams: max_negotiating_inbound_streams
+                }
             );
         }
 
@@ -593,9 +599,9 @@ mod tests {
         let upgrade_timeout = Duration::from_secs(1);
         let mut connection = Connection::new(
             StreamMuxerBox::new(PendingStreamMuxer),
-            MockConnectionHandler::new(upgrade_timeout.clone()),
+            MockConnectionHandler::new(upgrade_timeout.clone(), 2),
+            None,
             None,
-            2,
         );
 
         connection.handler.open_new_outbound();
@@ -713,14 +719,16 @@ mod tests {
         outbound_requested: bool,
         error: Option<ConnectionHandlerUpgrErr<Void>>,
         upgrade_timeout: Duration,
+        max_streams: usize,
     }
 
     impl MockConnectionHandler {
-        fn new(upgrade_timeout: Duration) -> Self {
+        fn new(upgrade_timeout: Duration, max_streams: usize) -> Self {
             Self {
                 outbound_requested: false,
                 error: None,
                 upgrade_timeout,
+                max_streams,
             }
         }
 
@@ -744,6 +752,10 @@ mod tests {
             SubstreamProtocol::new(DeniedUpgrade, ()).with_timeout(self.upgrade_timeout)
         }
 
+        fn max_inbound_streams(&self) -> usize {
+            self.max_streams
+        }
+
         fn inject_fully_negotiated_inbound(
             &mut self,
             protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
diff --git a/swarm/src/connection/pool.rs b/swarm/src/connection/pool.rs
index c43d5efb61e..8be70e7662d 100644
--- a/swarm/src/connection/pool.rs
+++ b/swarm/src/connection/pool.rs
@@ -91,7 +91,7 @@ where
     /// The maximum number of inbound streams concurrently negotiating on a connection.
     ///
     /// See [`Connection::max_negotiating_inbound_streams`].
-    max_negotiating_inbound_streams: usize,
+    legacy_max_negotiating_inbound_streams: Option<usize>,
 
     /// The executor to use for running the background tasks. If `None`,
     /// the tasks are kept in `local_spawns` instead and polled on the
@@ -269,7 +269,7 @@ where
             task_command_buffer_size: config.task_command_buffer_size,
             dial_concurrency_factor: config.dial_concurrency_factor,
             substream_upgrade_protocol_override: config.substream_upgrade_protocol_override,
-            max_negotiating_inbound_streams: config.max_negotiating_inbound_streams,
+            legacy_max_negotiating_inbound_streams: config.legacy_max_negotiating_inbound_streams,
             executor: config.executor,
             local_spawns: FuturesUnordered::new(),
             pending_connection_events_tx,
@@ -751,7 +751,7 @@ where
                         muxer,
                         handler.into_handler(&obtained_peer_id, &endpoint),
                         self.substream_upgrade_protocol_override,
-                        self.max_negotiating_inbound_streams,
+                        self.legacy_max_negotiating_inbound_streams,
                     );
                     self.spawn(
                         task::new_for_established_connection(
@@ -1165,7 +1165,7 @@ pub struct PoolConfig {
     /// The maximum number of inbound streams concurrently negotiating on a connection.
     ///
     /// See [`Connection::max_negotiating_inbound_streams`].
-    max_negotiating_inbound_streams: usize,
+    legacy_max_negotiating_inbound_streams: Option<usize>,
 }
 
 impl Default for PoolConfig {
@@ -1177,7 +1177,7 @@ impl Default for PoolConfig {
             // Set to a default of 8 based on frequency of dialer connections
             dial_concurrency_factor: NonZeroU8::new(8).expect("8 > 0"),
             substream_upgrade_protocol_override: None,
-            max_negotiating_inbound_streams: 128,
+            legacy_max_negotiating_inbound_streams: None,
         }
     }
 }
@@ -1241,7 +1241,7 @@ impl PoolConfig {
     ///
     /// See [`Connection::max_negotiating_inbound_streams`].
     pub fn with_max_negotiating_inbound_streams(mut self, v: usize) -> Self {
-        self.max_negotiating_inbound_streams = v;
+        self.legacy_max_negotiating_inbound_streams = Some(v);
         self
     }
 }
diff --git a/swarm/src/connection/pool/task.rs b/swarm/src/connection/pool/task.rs
index 866049e50da..c504dd0fb1f 100644
--- a/swarm/src/connection/pool/task.rs
+++ b/swarm/src/connection/pool/task.rs
@@ -22,6 +22,7 @@
 //! Async functions driving pending and established connections in the form of a task.
 
 use super::concurrent_dial::ConcurrentDial;
+use crate::connection::Event;
 use crate::{
     connection::{
         self, ConnectionError, PendingInboundConnectionError, PendingOutboundConnectionError,
@@ -248,6 +249,14 @@ pub async fn new_for_established_connection<THandler>(
                             .await;
                         return;
                     }
+                    Ok(Event::SubstreamLimitHit { num_streams }) => {
+                        // A handler might completely deny inbound streams at which point this would spam the logs.
+                        // Very likely, this log is only useful if we configured a limit > 0.
+                        if num_streams > 0 {
+                            // This is not necessarily bad but an indication that we have a remote peer that is keeping us very busy.
+                            log::info!("Connection {connection_id} to peer {peer_id} has is operating at the limit of allowed substeams ({num_streams})");
+                        }
+                    }
                 }
             }
         }
diff --git a/swarm/src/handler.rs b/swarm/src/handler.rs
index c6125f277b1..13f833d2b1a 100644
--- a/swarm/src/handler.rs
+++ b/swarm/src/handler.rs
@@ -60,6 +60,8 @@ pub use one_shot::{OneShotHandler, OneShotHandlerConfig};
 pub use pending::PendingConnectionHandler;
 pub use select::{ConnectionHandlerSelect, IntoConnectionHandlerSelect};
 
+pub const DEFAULT_MAX_INBOUND_STREAMS: usize = 128;
+
 /// A handler for a set of protocols used on a connection with a remote.
 ///
 /// This trait should be implemented for a type that maintains the state for
@@ -117,6 +119,13 @@ pub trait ConnectionHandler: Send + 'static {
     /// >           This allows a remote to put the list of supported protocols in a cache.
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo>;
 
+    /// The maximum number of inbound substreams allowed on the underlying connection.
+    ///
+    /// Once this limit is hit, we will stop accepting new inbound streams from the remote.
+    fn max_inbound_streams(&self) -> usize {
+        DEFAULT_MAX_INBOUND_STREAMS
+    }
+
     /// Injects the output of a successful upgrade on a new inbound substream.
     ///
     /// Note that it is up to the [`ConnectionHandler`] implementation to manage the lifetime of the
diff --git a/swarm/src/handler/dummy.rs b/swarm/src/handler/dummy.rs
index 67594686674..2a5906b01f1 100644
--- a/swarm/src/handler/dummy.rs
+++ b/swarm/src/handler/dummy.rs
@@ -50,13 +50,17 @@ impl ConnectionHandler for DummyConnectionHandler {
     type Error = Void;
     type InboundProtocol = DeniedUpgrade;
     type OutboundProtocol = DeniedUpgrade;
-    type OutboundOpenInfo = Void;
     type InboundOpenInfo = ();
+    type OutboundOpenInfo = Void;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         SubstreamProtocol::new(DeniedUpgrade, ())
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        0
+    }
+
     fn inject_fully_negotiated_inbound(
         &mut self,
         _: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,
diff --git a/swarm/src/handler/either.rs b/swarm/src/handler/either.rs
index 21d386ec56a..30a2a0e57bb 100644
--- a/swarm/src/handler/either.rs
+++ b/swarm/src/handler/either.rs
@@ -94,33 +94,37 @@ where
         }
     }
 
-    fn inject_fully_negotiated_outbound(
+    fn max_inbound_streams(&self) -> usize {
+        either::for_both!(self, inner => inner.max_inbound_streams())
+    }
+
+    fn inject_fully_negotiated_inbound(
         &mut self,
-        output: <Self::OutboundProtocol as OutboundUpgradeSend>::Output,
-        info: Self::OutboundOpenInfo,
+        output: <Self::InboundProtocol as InboundUpgradeSend>::Output,
+        info: Self::InboundOpenInfo,
     ) {
         match (self, output, info) {
             (Either::Left(handler), EitherOutput::First(output), Either::Left(info)) => {
-                handler.inject_fully_negotiated_outbound(output, info)
+                handler.inject_fully_negotiated_inbound(output, info)
             }
             (Either::Right(handler), EitherOutput::Second(output), Either::Right(info)) => {
-                handler.inject_fully_negotiated_outbound(output, info)
+                handler.inject_fully_negotiated_inbound(output, info)
             }
             _ => unreachable!(),
         }
     }
 
-    fn inject_fully_negotiated_inbound(
+    fn inject_fully_negotiated_outbound(
         &mut self,
-        output: <Self::InboundProtocol as InboundUpgradeSend>::Output,
-        info: Self::InboundOpenInfo,
+        output: <Self::OutboundProtocol as OutboundUpgradeSend>::Output,
+        info: Self::OutboundOpenInfo,
     ) {
         match (self, output, info) {
             (Either::Left(handler), EitherOutput::First(output), Either::Left(info)) => {
-                handler.inject_fully_negotiated_inbound(output, info)
+                handler.inject_fully_negotiated_outbound(output, info)
             }
             (Either::Right(handler), EitherOutput::Second(output), Either::Right(info)) => {
-                handler.inject_fully_negotiated_inbound(output, info)
+                handler.inject_fully_negotiated_outbound(output, info)
             }
             _ => unreachable!(),
         }
diff --git a/swarm/src/handler/map_in.rs b/swarm/src/handler/map_in.rs
index a209225045e..39471d65d22 100644
--- a/swarm/src/handler/map_in.rs
+++ b/swarm/src/handler/map_in.rs
@@ -64,6 +64,10 @@ where
         self.inner.listen_protocol()
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        self.inner.max_inbound_streams()
+    }
+
     fn inject_fully_negotiated_inbound(
         &mut self,
         protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
diff --git a/swarm/src/handler/map_out.rs b/swarm/src/handler/map_out.rs
index 2eb0c2f9bdc..a750e807956 100644
--- a/swarm/src/handler/map_out.rs
+++ b/swarm/src/handler/map_out.rs
@@ -59,6 +59,10 @@ where
         self.inner.listen_protocol()
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        self.inner.max_inbound_streams()
+    }
+
     fn inject_fully_negotiated_inbound(
         &mut self,
         protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
diff --git a/swarm/src/handler/multi.rs b/swarm/src/handler/multi.rs
index f7e4d7ed1b4..381626c7d04 100644
--- a/swarm/src/handler/multi.rs
+++ b/swarm/src/handler/multi.rs
@@ -121,6 +121,14 @@ where
         SubstreamProtocol::new(upgrade, info).with_timeout(timeout)
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        self.handlers
+            .values()
+            .map(|h| h.max_inbound_streams())
+            .max()
+            .unwrap_or(0) // No handlers? No substreams.
+    }
+
     fn inject_fully_negotiated_outbound(
         &mut self,
         protocol: <Self::OutboundProtocol as OutboundUpgradeSend>::Output,
diff --git a/swarm/src/handler/select.rs b/swarm/src/handler/select.rs
index 70a6f3c26f6..36acf8b9850 100644
--- a/swarm/src/handler/select.rs
+++ b/swarm/src/handler/select.rs
@@ -29,6 +29,7 @@ use libp2p_core::{
     upgrade::{EitherUpgrade, NegotiationError, ProtocolError, SelectUpgrade, UpgradeError},
     ConnectedPoint, Multiaddr, PeerId,
 };
+use std::cmp::max;
 use std::{cmp, task::Context, task::Poll};
 
 /// Implementation of `IntoConnectionHandler` that combines two protocols into one.
@@ -113,8 +114,8 @@ where
         SendWrapper<TProto1::OutboundProtocol>,
         SendWrapper<TProto2::OutboundProtocol>,
     >;
-    type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
     type InboundOpenInfo = (TProto1::InboundOpenInfo, TProto2::InboundOpenInfo);
+    type OutboundOpenInfo = EitherOutput<TProto1::OutboundOpenInfo, TProto2::OutboundOpenInfo>;
 
     fn listen_protocol(&self) -> SubstreamProtocol<Self::InboundProtocol, Self::InboundOpenInfo> {
         let proto1 = self.proto1.listen_protocol();
@@ -126,6 +127,28 @@ where
         SubstreamProtocol::new(choice, (i1, i2)).with_timeout(timeout)
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        max(
+            self.proto1.max_inbound_streams(),
+            self.proto2.max_inbound_streams(),
+        )
+    }
+
+    fn inject_fully_negotiated_inbound(
+        &mut self,
+        protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
+        (i1, i2): Self::InboundOpenInfo,
+    ) {
+        match protocol {
+            EitherOutput::First(protocol) => {
+                self.proto1.inject_fully_negotiated_inbound(protocol, i1)
+            }
+            EitherOutput::Second(protocol) => {
+                self.proto2.inject_fully_negotiated_inbound(protocol, i2)
+            }
+        }
+    }
+
     fn inject_fully_negotiated_outbound(
         &mut self,
         protocol: <Self::OutboundProtocol as OutboundUpgradeSend>::Output,
@@ -147,21 +170,6 @@ where
         }
     }
 
-    fn inject_fully_negotiated_inbound(
-        &mut self,
-        protocol: <Self::InboundProtocol as InboundUpgradeSend>::Output,
-        (i1, i2): Self::InboundOpenInfo,
-    ) {
-        match protocol {
-            EitherOutput::First(protocol) => {
-                self.proto1.inject_fully_negotiated_inbound(protocol, i1)
-            }
-            EitherOutput::Second(protocol) => {
-                self.proto2.inject_fully_negotiated_inbound(protocol, i2)
-            }
-        }
-    }
-
     fn inject_event(&mut self, event: Self::InEvent) {
         match event {
             EitherOutput::First(event) => self.proto1.inject_event(event),
diff --git a/swarm/src/lib.rs b/swarm/src/lib.rs
index 7faaa43b44d..cec59ec68b1 100644
--- a/swarm/src/lib.rs
+++ b/swarm/src/lib.rs
@@ -1371,6 +1371,10 @@ where
     /// The maximum number of inbound streams concurrently negotiating on a connection.
     ///
     /// See [`PoolConfig::with_max_negotiating_inbound_streams`].
+    #[deprecated(
+        since = "0.40.0",
+        note = "Override `ConnectionHandler::max_inbound_streams instead.`"
+    )]
     pub fn max_negotiating_inbound_streams(mut self, v: usize) -> Self {
         self.pool_config = self.pool_config.with_max_negotiating_inbound_streams(v);
         self

From a008719f5f4fc52bb284c223af5f79bc993cf91f Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 30 Sep 2022 16:41:28 +1000
Subject: [PATCH 08/16] Don't override `max_inbound_streams` for
 `DummyConnectionHandler`

This will result in peers never being able to open a substream
and we currently depend on this behaviour, despite the upgrade
never being successful.
---
 swarm/src/handler/dummy.rs | 4 ----
 1 file changed, 4 deletions(-)

diff --git a/swarm/src/handler/dummy.rs b/swarm/src/handler/dummy.rs
index 2a5906b01f1..6397747b6b2 100644
--- a/swarm/src/handler/dummy.rs
+++ b/swarm/src/handler/dummy.rs
@@ -57,10 +57,6 @@ impl ConnectionHandler for DummyConnectionHandler {
         SubstreamProtocol::new(DeniedUpgrade, ())
     }
 
-    fn max_inbound_streams(&self) -> usize {
-        0
-    }
-
     fn inject_fully_negotiated_inbound(
         &mut self,
         _: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,

From 9cd40830e2ebe853397a0ee519e4e79fe0b351c1 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 30 Sep 2022 17:00:21 +1000
Subject: [PATCH 09/16] Remove outdated rustdoc links

These will eventually go away so we don't bother replacing the links.
---
 swarm/src/connection/pool.rs | 6 ------
 1 file changed, 6 deletions(-)

diff --git a/swarm/src/connection/pool.rs b/swarm/src/connection/pool.rs
index 8be70e7662d..526fa2ebce2 100644
--- a/swarm/src/connection/pool.rs
+++ b/swarm/src/connection/pool.rs
@@ -89,8 +89,6 @@ where
     substream_upgrade_protocol_override: Option<libp2p_core::upgrade::Version>,
 
     /// The maximum number of inbound streams concurrently negotiating on a connection.
-    ///
-    /// See [`Connection::max_negotiating_inbound_streams`].
     legacy_max_negotiating_inbound_streams: Option<usize>,
 
     /// The executor to use for running the background tasks. If `None`,
@@ -1163,8 +1161,6 @@ pub struct PoolConfig {
     substream_upgrade_protocol_override: Option<libp2p_core::upgrade::Version>,
 
     /// The maximum number of inbound streams concurrently negotiating on a connection.
-    ///
-    /// See [`Connection::max_negotiating_inbound_streams`].
     legacy_max_negotiating_inbound_streams: Option<usize>,
 }
 
@@ -1238,8 +1234,6 @@ impl PoolConfig {
     }
 
     /// The maximum number of inbound streams concurrently negotiating on a connection.
-    ///
-    /// See [`Connection::max_negotiating_inbound_streams`].
     pub fn with_max_negotiating_inbound_streams(mut self, v: usize) -> Self {
         self.legacy_max_negotiating_inbound_streams = Some(v);
         self

From e2b4969af186c8df14a8f6d9a123ad4ab850125c Mon Sep 17 00:00:00 2001
From: Ryan Plauche <ryan.plauche@gmail.com>
Date: Thu, 29 Sep 2022 22:58:35 -0500
Subject: [PATCH 10/16] protocols/{autonat,dcutr}: Fixing filename collision in
 examples (#2959)

Co-authored-by: Ryan Plauche <ryan@littlebearlabs.io>
---
 protocols/autonat/examples/{client.rs => autonat_client.rs} | 0
 protocols/autonat/examples/{server.rs => autonat_server.rs} | 0
 protocols/dcutr/examples/{client.rs => dcutr.rs}            | 0
 3 files changed, 0 insertions(+), 0 deletions(-)
 rename protocols/autonat/examples/{client.rs => autonat_client.rs} (100%)
 rename protocols/autonat/examples/{server.rs => autonat_server.rs} (100%)
 rename protocols/dcutr/examples/{client.rs => dcutr.rs} (100%)

diff --git a/protocols/autonat/examples/client.rs b/protocols/autonat/examples/autonat_client.rs
similarity index 100%
rename from protocols/autonat/examples/client.rs
rename to protocols/autonat/examples/autonat_client.rs
diff --git a/protocols/autonat/examples/server.rs b/protocols/autonat/examples/autonat_server.rs
similarity index 100%
rename from protocols/autonat/examples/server.rs
rename to protocols/autonat/examples/autonat_server.rs
diff --git a/protocols/dcutr/examples/client.rs b/protocols/dcutr/examples/dcutr.rs
similarity index 100%
rename from protocols/dcutr/examples/client.rs
rename to protocols/dcutr/examples/dcutr.rs

From 311e6741f2872b77a13cd83d05ecfdd80dd68fd0 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 7 Oct 2022 10:29:36 +1100
Subject: [PATCH 11/16] Fix typo in log

---
 swarm/src/connection/pool/task.rs | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/swarm/src/connection/pool/task.rs b/swarm/src/connection/pool/task.rs
index c504dd0fb1f..c1283d2ca4c 100644
--- a/swarm/src/connection/pool/task.rs
+++ b/swarm/src/connection/pool/task.rs
@@ -254,7 +254,7 @@ pub async fn new_for_established_connection<THandler>(
                         // Very likely, this log is only useful if we configured a limit > 0.
                         if num_streams > 0 {
                             // This is not necessarily bad but an indication that we have a remote peer that is keeping us very busy.
-                            log::info!("Connection {connection_id} to peer {peer_id} has is operating at the limit of allowed substeams ({num_streams})");
+                            log::info!("Connection {connection_id} to peer {peer_id} is operating at the limit of allowed substeams ({num_streams})");
                         }
                     }
                 }

From cbb6dd5b05093d45db6e7cd196bab207a553a4e3 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 7 Oct 2022 10:33:34 +1100
Subject: [PATCH 12/16] Use `max_inbound_streams` for kademlia handler

---
 protocols/kad/src/handler.rs | 28 ++++------------------------
 1 file changed, 4 insertions(+), 24 deletions(-)

diff --git a/protocols/kad/src/handler.rs b/protocols/kad/src/handler.rs
index 5be9ae17737..db4469099f1 100644
--- a/protocols/kad/src/handler.rs
+++ b/protocols/kad/src/handler.rs
@@ -523,6 +523,10 @@ where
         }
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        MAX_NUM_INBOUND_SUBSTREAMS
+    }
+
     fn inject_fully_negotiated_outbound(
         &mut self,
         protocol: <Self::OutboundProtocol as OutboundUpgrade<NegotiatedSubstream>>::Output,
@@ -559,30 +563,6 @@ where
             self.protocol_status = ProtocolStatus::Confirmed;
         }
 
-        if self.inbound_substreams.len() == MAX_NUM_INBOUND_SUBSTREAMS {
-            if let Some(position) = self.inbound_substreams.iter().position(|s| {
-                matches!(
-                    s,
-                    // An inbound substream waiting to be reused.
-                    InboundSubstreamState::WaitingMessage { first: false, .. }
-                )
-            }) {
-                self.inbound_substreams.remove(position);
-                log::warn!(
-                    "New inbound substream to {:?} exceeds inbound substream limit. \
-                    Removed older substream waiting to be reused.",
-                    self.remote_peer_id,
-                )
-            } else {
-                log::warn!(
-                    "New inbound substream to {:?} exceeds inbound substream limit. \
-                     No older substream waiting to be reused. Dropping new substream.",
-                    self.remote_peer_id,
-                );
-                return;
-            }
-        }
-
         debug_assert!(self.config.allow_listening);
         let connec_unique_id = self.next_connec_unique_id;
         self.next_connec_unique_id.0 += 1;

From 534c3f6ed9d8bc43f6fdc8a9313df811343fe7dd Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 7 Oct 2022 10:43:22 +1100
Subject: [PATCH 13/16] Use `max_inbound_streams` in GossipSub handler

---
 protocols/gossipsub/src/handler.rs | 6 ++++++
 1 file changed, 6 insertions(+)

diff --git a/protocols/gossipsub/src/handler.rs b/protocols/gossipsub/src/handler.rs
index 9ea2e6ea49f..9ceab441b29 100644
--- a/protocols/gossipsub/src/handler.rs
+++ b/protocols/gossipsub/src/handler.rs
@@ -195,6 +195,12 @@ impl ConnectionHandler for GossipsubHandler {
         self.listen_protocol.clone()
     }
 
+    // Gossipsub is supposed to have a single long-lived inbound substream.
+    // Don't allow the remote to create more than one at a time.
+    fn max_inbound_streams(&self) -> usize {
+        1
+    }
+
     fn inject_fully_negotiated_inbound(
         &mut self,
         protocol: <Self::InboundProtocol as InboundUpgrade<NegotiatedSubstream>>::Output,

From a6bd2e48e72dbb4267585403b184d4871523654a Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 7 Oct 2022 10:45:20 +1100
Subject: [PATCH 14/16] Use `max_inbound_streams` for ping handler

---
 protocols/ping/src/handler.rs | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/protocols/ping/src/handler.rs b/protocols/ping/src/handler.rs
index af1ef898981..0cac23fb064 100644
--- a/protocols/ping/src/handler.rs
+++ b/protocols/ping/src/handler.rs
@@ -239,6 +239,10 @@ impl ConnectionHandler for Handler {
         SubstreamProtocol::new(ReadyUpgrade::new(PROTOCOL_NAME), ())
     }
 
+    fn max_inbound_streams(&self) -> usize {
+        1 // There is only ever one active inbound stream.
+    }
+
     fn inject_fully_negotiated_inbound(&mut self, stream: NegotiatedSubstream, (): ()) {
         self.inbound = Some(protocol::recv_ping(stream).boxed());
     }

From 5531d9610dda029b6f34db81b3113a89effa3f30 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 14 Oct 2022 21:26:53 +1100
Subject: [PATCH 15/16] Remove use of deprecated functions

---
 core/src/muxing/boxed.rs | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/core/src/muxing/boxed.rs b/core/src/muxing/boxed.rs
index eecbce1f9cd..3b0a70a7aa2 100644
--- a/core/src/muxing/boxed.rs
+++ b/core/src/muxing/boxed.rs
@@ -270,13 +270,14 @@ impl AsyncWrite for SubstreamBox {
 mod tests {
     use super::*;
     use crate::muxing::StreamMuxerExt;
+    use futures::future::poll_fn;
 
     #[async_std::test]
     async fn stream_muxer_box_tracks_alive_inbound_streams() {
         let mut muxer = StreamMuxerBox::new(DummyStreamMuxer);
 
-        let _stream1 = muxer.next_inbound().await.unwrap();
-        let _stream2 = muxer.next_inbound().await.unwrap();
+        let _stream1 = poll_fn(|cx| muxer.poll_inbound_unpin(cx)).await.unwrap();
+        let _stream2 = poll_fn(|cx| muxer.poll_inbound_unpin(cx)).await.unwrap();
 
         assert_eq!(muxer.active_inbound_streams(), 2);
 
@@ -289,8 +290,8 @@ mod tests {
     async fn stream_muxer_box_tracks_alive_outbound_streams() {
         let mut muxer = StreamMuxerBox::new(DummyStreamMuxer);
 
-        let _stream1 = muxer.next_outbound().await.unwrap();
-        let _stream2 = muxer.next_outbound().await.unwrap();
+        let _stream1 = poll_fn(|cx| muxer.poll_outbound_unpin(cx)).await.unwrap();
+        let _stream2 = poll_fn(|cx| muxer.poll_outbound_unpin(cx)).await.unwrap();
 
         assert_eq!(muxer.active_outbound_streams(), 2);
 

From f3d36ebecb919391d9990433feb9e2d07dd248b8 Mon Sep 17 00:00:00 2001
From: Thomas Eizinger <thomas@eizinger.io>
Date: Fri, 14 Oct 2022 21:39:28 +1100
Subject: [PATCH 16/16] Rewrite changelog section

---
 swarm/CHANGELOG.md | 14 ++++++--------
 1 file changed, 6 insertions(+), 8 deletions(-)

diff --git a/swarm/CHANGELOG.md b/swarm/CHANGELOG.md
index 125692e246c..7c90ce650ce 100644
--- a/swarm/CHANGELOG.md
+++ b/swarm/CHANGELOG.md
@@ -10,14 +10,12 @@
 
 - Pass actual `PeerId` of dial to `NetworkBehaviour::inject_dial_failure` on `DialError::ConnectionLimit`. See [PR 2928].
 
-- Use the total number of alive inbound streams for back-pressure. This can have a BIG impact on your application
-  depending on how it uses `libp2p`. Previously, the limit for inbound streams per connection only applied to the
-  _upgrade_ phase, i.e. for the time `InboundUpgrade` was running. Any stream being returned from `InboundUpgrade` and
-  given to the `ConnectionHandler` did not count towards that limit, essentially mitigating the back-pressure mechanism.
-  With this release, substreams count towards that limit until they are dropped, so we actually enforce, how many
-  inbound streams can be active at one time _per connection_. `libp2p` will not accept any more incoming streams once
-  that limit is hit. You can configure the number of allowed substreams by overriding
-  `ConnectionHandler::max_inbound_streams`. See [PR 2878].
+- Deprecate `SwarmBuilder::max_negotiating_inbound_streams` and introduce `ConnectionHandler::max_inbound_streams`.
+  The limit returned from `ConnectionHandler::max_inbound_streams` defines, how many inbound substreams are allowed to
+  be _active_ (i.e. "floating around") at any single time on this connection. This differs from what
+  `max_negotiating_inbound_streams` used to do in that we previously only limited the number of concurrent _upgrades_,
+  i.e. the `InboundUpgrade` phase of a substream but did otherwise not enforce a limit on how many substreams a remote
+  can open. The default value for `ConnectionHandler::max_inbound_streams` is 128. See [PR 2878].
 
 [PR 2857]: https://github.com/libp2p/rust-libp2p/pull/2857
 [PR 2859]: https://github.com/libp2p/rust-libp2p/pull/2859/