deps(yamux): update yamux to v0.12

interop-tests/Cargo.toml

@@ -19,7 +19,7 @@ rand = "0.8.5"
 
 [target.'cfg(not(target_arch = "wasm32"))'.dependencies]
 axum = "0.6"
-libp2p = { path = "../libp2p", features = ["ping", "noise", "tls", "rsa", "macros", "websocket", "tokio", "yamux", "tcp", "dns"] }
+libp2p = { path = "../libp2p", features = ["ping", "noise", "tls", "rsa", "macros", "websocket", "tokio", "yamux", "tcp", "dns", "identify"] }
 libp2p-quic = { workspace = true, features = ["tokio"] }
 libp2p-webrtc = { workspace = true, features = ["tokio"] }
 libp2p-mplex = { path = "../muxers/mplex" }
@@ -34,7 +34,7 @@ tracing = "0.1"
 tracing-subscriber = { version = "0.3", features = ["env-filter"] }
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
-libp2p = { path = "../libp2p", features = ["ping", "macros", "webtransport-websys", "wasm-bindgen"] }
+libp2p = { path = "../libp2p", features = ["ping", "macros", "webtransport-websys", "wasm-bindgen", "identify"] }
 wasm-bindgen = { version = "0.2" }
 wasm-bindgen-futures = { version = "0.4" }
 wasm-logger = { version = "0.2.0" }

interop-tests/src/lib.rs

@@ -4,7 +4,7 @@ use std::time::Duration;
 use anyhow::{bail, Context, Result};
 use futures::{FutureExt, StreamExt};
 use libp2p::swarm::{keep_alive, NetworkBehaviour, SwarmEvent};
-use libp2p::{identity, ping, Multiaddr, PeerId};
+use libp2p::{identify, identity, ping, Multiaddr, PeerId};
 #[cfg(target_arch = "wasm32")]
 use wasm_bindgen::prelude::*;
 
@@ -29,12 +29,16 @@ pub async fn run_test(
     let redis_client = RedisClient::new(redis_addr).context("Could not connect to redis")?;
 
     // Build the transport from the passed ENV var.
-    let (boxed_transport, local_addr) = build_transport(local_key, ip, transport)?;
+    let (boxed_transport, local_addr) = build_transport(local_key.clone(), ip, transport)?;
     let mut swarm = swarm_builder(
         boxed_transport,
         Behaviour {
             ping: ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(1))),
             keep_alive: keep_alive::Behaviour,
+            identify: identify::Behaviour::new(identify::Config::new(
+                "/interop-tests".to_owned(),
+                local_key.public(),
+            )),
         },
         local_peer_id,
     )
@@ -237,6 +241,7 @@ impl FromStr for SecProtocol {
 struct Behaviour {
     ping: ping::Behaviour,
     keep_alive: keep_alive::Behaviour,
+    identify: identify::Behaviour,
 }
 
 /// Helper function to get a ENV variable into an test parameter like `Transport`.

muxers/yamux/Cargo.toml

@@ -14,7 +14,7 @@ categories = ["network-programming", "asynchronous"]
 futures = "0.3.28"
 libp2p-core = { workspace = true }
 thiserror = "1.0"
-yamux = "0.10.0"
+yamux = { git = "https://github.com/thomaseizinger/rust-yamux", branch = "feat/256-backlog" }
 log = "0.4"
 
 [dev-dependencies]

muxers/yamux/src/lib.rs

@@ -22,29 +22,27 @@
 
 #![cfg_attr(docsrs, feature(doc_cfg, doc_auto_cfg))]
 
-use futures::{future, prelude::*, ready, stream::BoxStream};
+use futures::{future, prelude::*, ready};
 use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent};
 use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo};
 use std::collections::VecDeque;
 use std::io::{IoSlice, IoSliceMut};
 use std::task::Waker;
 use std::{
-    fmt, io, iter, mem,
+    io, iter,
     pin::Pin,
     task::{Context, Poll},
 };
 use thiserror::Error;
 use yamux::ConnectionError;
 
 /// A Yamux connection.
+#[derive(Debug)]
 pub struct Muxer<C> {
-    /// The [`futures::stream::Stream`] of incoming substreams.
-    incoming: BoxStream<'static, Result<yamux::Stream, yamux::ConnectionError>>,
-    /// Handle to control the connection.
-    control: yamux::Control,
+    connection: yamux::Connection<C>,
     /// Temporarily buffers inbound streams in case our node is performing backpressure on the remote.
     ///
-    /// The only way how yamux can make progress is by driving the stream. However, the
+    /// The only way how yamux can make progress is by calling [`yamux::Connection::poll_next_inbound`]. However, the
     /// [`StreamMuxer`] interface is designed to allow a caller to selectively make progress via
     /// [`StreamMuxer::poll_inbound`] and [`StreamMuxer::poll_outbound`] whilst the more general
     /// [`StreamMuxer::poll`] is designed to make progress on existing streams etc.
@@ -54,40 +52,31 @@ pub struct Muxer<C> {
     inbound_stream_buffer: VecDeque<Stream>,
     /// Waker to be called when new inbound streams are available.
     inbound_stream_waker: Option<Waker>,
-
-    _phantom: std::marker::PhantomData<C>,
 }
 
-const MAX_BUFFERED_INBOUND_STREAMS: usize = 25;
-
-impl<S> fmt::Debug for Muxer<S> {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.write_str("Yamux")
-    }
-}
+/// How many streams to buffer before we start resetting them.
+///
+/// This is equal to the ACK BACKLOG in `rust-yamux`.
+/// Thus, for peers running on a recent version of `rust-libp2p`, we should never need to reset streams because they'll voluntarily stop opening them once they hit the ACK backlog.
+const MAX_BUFFERED_INBOUND_STREAMS: usize = 256;
 
 impl<C> Muxer<C>
 where
     C: AsyncRead + AsyncWrite + Send + Unpin + 'static,
 {
     /// Create a new Yamux connection.
     fn new(io: C, cfg: yamux::Config, mode: yamux::Mode) -> Self {
-        let conn = yamux::Connection::new(io, cfg, mode);
-        let ctrl = conn.control();
-
-        Self {
-            incoming: yamux::into_stream(conn).err_into().boxed(),
-            control: ctrl,
+        Muxer {
+            connection: yamux::Connection::new(io, cfg, mode),
             inbound_stream_buffer: VecDeque::default(),
             inbound_stream_waker: None,
-            _phantom: Default::default(),
         }
     }
 }
 
 impl<C> StreamMuxer for Muxer<C>
 where
-    C: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    C: AsyncRead + AsyncWrite + Unpin + 'static,
 {
     type Substream = Stream;
     type Error = Error;
@@ -112,10 +101,15 @@ where
         mut self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<Self::Substream, Self::Error>> {
-        Pin::new(&mut self.control)
-            .poll_open_stream(cx)
-            .map_ok(Stream)
-            .map_err(Error)
+        let stream = ready!(self.connection.poll_new_outbound(cx).map_err(Error)?);
+
+        Poll::Ready(Ok(Stream(stream)))
+    }
+
+    fn poll_close(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
+        ready!(self.connection.poll_close(cx).map_err(Error)?);
+
+        Poll::Ready(Ok(()))
     }
 
     fn poll(
@@ -141,23 +135,6 @@ where
         cx.waker().wake_by_ref();
         Poll::Pending
     }
-
-    fn poll_close(mut self: Pin<&mut Self>, c: &mut Context<'_>) -> Poll<Result<(), Error>> {
-        if let Poll::Ready(()) = Pin::new(&mut self.control).poll_close(c).map_err(Error)? {
-            return Poll::Ready(Ok(()));
-        }
-
-        while let Poll::Ready(maybe_inbound_stream) =
-            self.incoming.poll_next_unpin(c).map_err(Error)?
-        {
-            match maybe_inbound_stream {
-                Some(inbound_stream) => mem::drop(inbound_stream),
-                None => return Poll::Ready(Ok(())),
-            }
-        }
-
-        Poll::Pending
-    }
 }
 
 /// A stream produced by the yamux multiplexer.
@@ -210,18 +187,16 @@ impl AsyncWrite for Stream {
 
 impl<C> Muxer<C>
 where
-    C: AsyncRead + AsyncWrite + Send + Unpin + 'static,
+    C: AsyncRead + AsyncWrite + Unpin + 'static,
 {
     fn poll_inner(&mut self, cx: &mut Context<'_>) -> Poll<Result<Stream, Error>> {
-        self.incoming.poll_next_unpin(cx).map(|maybe_stream| {
-            let stream = maybe_stream
-                .transpose()
-                .map_err(Error)?
-                .map(Stream)
-                .ok_or(Error(ConnectionError::Closed))?;
-
-            Ok(stream)
-        })
+        let stream = ready!(self.connection.poll_next_inbound(cx))
+            .transpose()
+            .map_err(Error)?
+            .map(Stream)
+            .ok_or(Error(ConnectionError::Closed))?;
+
+        Poll::Ready(Ok(stream))
     }
 }