chore(multistream-select): replace futures_ringbuf with TCP

We seem to have a flaky test in `multistream-select`. I could reproduce it locally but not fully track down what the issue is. I think it is related to the "sockets" being dropped to early or something.

Once I swapped them out for a TCP listener, the problem did not occur again.

Pull-Request: #4693.

Author: thomaseizinger

misc/multistream-select/src/dialer_select.rs

@@ -201,3 +201,197 @@ where
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::listener_select_proto;
+    use async_std::future::timeout;
+    use async_std::net::{TcpListener, TcpStream};
+    use log::info;
+    use quickcheck::{Arbitrary, Gen, GenRange};
+    use std::time::Duration;
+
+    #[test]
+    fn select_proto_basic() {
+        async fn run(version: Version) {
+            let (client_connection, server_connection) = futures_ringbuf::Endpoint::pair(100, 100);
+
+            let server = async_std::task::spawn(async move {
+                let protos = vec!["/proto1", "/proto2"];
+                let (proto, mut io) = listener_select_proto(server_connection, protos)
+                    .await
+                    .unwrap();
+                assert_eq!(proto, "/proto2");
+
+                let mut out = vec![0; 32];
+                let n = io.read(&mut out).await.unwrap();
+                out.truncate(n);
+                assert_eq!(out, b"ping");
+
+                io.write_all(b"pong").await.unwrap();
+                io.flush().await.unwrap();
+            });
+
+            let client = async_std::task::spawn(async move {
+                let protos = vec!["/proto3", "/proto2"];
+                let (proto, mut io) = dialer_select_proto(client_connection, protos, version)
+                    .await
+                    .unwrap();
+                assert_eq!(proto, "/proto2");
+
+                io.write_all(b"ping").await.unwrap();
+                io.flush().await.unwrap();
+
+                let mut out = vec![0; 32];
+                let n = io.read(&mut out).await.unwrap();
+                out.truncate(n);
+                assert_eq!(out, b"pong");
+            });
+
+            server.await;
+            client.await;
+        }
+
+        async_std::task::block_on(run(Version::V1));
+        async_std::task::block_on(run(Version::V1Lazy));
+    }
+
+    /// Tests the expected behaviour of failed negotiations.
+    #[test]
+    fn negotiation_failed() {
+        fn prop(
+            version: Version,
+            DialerProtos(dial_protos): DialerProtos,
+            ListenerProtos(listen_protos): ListenerProtos,
+            DialPayload(dial_payload): DialPayload,
+        ) {
+            let _ = env_logger::try_init();
+
+            async_std::task::block_on(async move {
+                let listener = TcpListener::bind("0.0.0.0:0").await.unwrap();
+                let addr = listener.local_addr().unwrap();
+
+                let server = async_std::task::spawn(async move {
+                    let server_connection = listener.accept().await.unwrap().0;
+
+                    let io = match timeout(
+                        Duration::from_secs(2),
+                        listener_select_proto(server_connection, listen_protos),
+                    )
+                    .await
+                    .unwrap()
+                    {
+                        Ok((_, io)) => io,
+                        Err(NegotiationError::Failed) => return,
+                        Err(NegotiationError::ProtocolError(e)) => {
+                            panic!("Unexpected protocol error {e}")
+                        }
+                    };
+                    match io.complete().await {
+                        Err(NegotiationError::Failed) => {}
+                        _ => panic!(),
+                    }
+                });
+
+                let client = async_std::task::spawn(async move {
+                    let client_connection = TcpStream::connect(addr).await.unwrap();
+
+                    let mut io = match timeout(
+                        Duration::from_secs(2),
+                        dialer_select_proto(client_connection, dial_protos, version),
+                    )
+                    .await
+                    .unwrap()
+                    {
+                        Err(NegotiationError::Failed) => return,
+                        Ok((_, io)) => io,
+                        Err(_) => panic!(),
+                    };
+                    // The dialer may write a payload that is even sent before it
+                    // got confirmation of the last proposed protocol, when `V1Lazy`
+                    // is used.
+
+                    info!("Writing early data");
+
+                    io.write_all(&dial_payload).await.unwrap();
+                    match io.complete().await {
+                        Err(NegotiationError::Failed) => {}
+                        _ => panic!(),
+                    }
+                });
+
+                server.await;
+                client.await;
+
+                info!("---------------------------------------")
+            });
+        }
+
+        quickcheck::QuickCheck::new()
+            .tests(1000)
+            .quickcheck(prop as fn(_, _, _, _));
+    }
+
+    #[async_std::test]
+    async fn v1_lazy_do_not_wait_for_negotiation_on_poll_close() {
+        let (client_connection, _server_connection) =
+            futures_ringbuf::Endpoint::pair(1024 * 1024, 1);
+
+        let client = async_std::task::spawn(async move {
+            // Single protocol to allow for lazy (or optimistic) protocol negotiation.
+            let protos = vec!["/proto1"];
+            let (proto, mut io) = dialer_select_proto(client_connection, protos, Version::V1Lazy)
+                .await
+                .unwrap();
+            assert_eq!(proto, "/proto1");
+
+            // client can close the connection even though protocol negotiation is not yet done, i.e.
+            // `_server_connection` had been untouched.
+            io.close().await.unwrap();
+        });
+
+        async_std::future::timeout(Duration::from_secs(10), client)
+            .await
+            .unwrap();
+    }
+
+    #[derive(Clone, Debug)]
+    struct DialerProtos(Vec<&'static str>);
+
+    impl Arbitrary for DialerProtos {
+        fn arbitrary(g: &mut Gen) -> Self {
+            if bool::arbitrary(g) {
+                DialerProtos(vec!["/proto1"])
+            } else {
+                DialerProtos(vec!["/proto1", "/proto2"])
+            }
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    struct ListenerProtos(Vec<&'static str>);
+
+    impl Arbitrary for ListenerProtos {
+        fn arbitrary(g: &mut Gen) -> Self {
+            if bool::arbitrary(g) {
+                ListenerProtos(vec!["/proto3"])
+            } else {
+                ListenerProtos(vec!["/proto3", "/proto4"])
+            }
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    struct DialPayload(Vec<u8>);
+
+    impl Arbitrary for DialPayload {
+        fn arbitrary(g: &mut Gen) -> Self {
+            DialPayload(
+                (0..g.gen_range(0..2u8))
+                    .map(|_| g.gen_range(1..255)) // We can generate 0 as that will produce a different error.
+                    .collect(),
+            )
+        }
+    }
+}

misc/multistream-select/src/lib.rs

@@ -140,3 +140,11 @@ pub enum Version {
     // Draft: https://github.com/libp2p/specs/pull/95
     // V2,
 }
+
+#[cfg(test)]
+impl quickcheck::Arbitrary for Version {
+    fn arbitrary(g: &mut quickcheck::Gen) -> Self {
+        *g.choose(&[Version::V1, Version::V1Lazy])
+            .expect("slice not empty")
+    }
+}

misc/multistream-select/src/listener_select.rs

@@ -124,9 +124,9 @@ where
             match mem::replace(this.state, State::Done) {
                 State::RecvHeader { mut io } => {
                     match io.poll_next_unpin(cx) {
-                        Poll::Ready(Some(Ok(Message::Header(h)))) => match h {
-                            HeaderLine::V1 => *this.state = State::SendHeader { io },
-                        },
+                        Poll::Ready(Some(Ok(Message::Header(HeaderLine::V1)))) => {
+                            *this.state = State::SendHeader { io }
+                        }
                         Poll::Ready(Some(Ok(_))) => {
                             return Poll::Ready(Err(ProtocolError::InvalidMessage.into()))
                         }