Merge #456

456: Fix swarmapi tests and add more cases r=koivunej a=koivunej

This PR fixes the swarm tests and yet another swarmapi related panic reachable through only the swarmapi "unit tests". It also adds a two new cases which would had been quite ugly add to under `tests/`. 

First of all, the test had been broken long time ago in a rust-libp2p update when the listening address adding became asynchronous.

The additional panic exposed by now working test case comes from the `inject_connected` never arriving when only using SwarmApi as the only behaviour. It uses the DummyProtocolHandler which opens no streams and demands keepalive, the connection is immediatedly closed.

The additional test cases cover connecting to wrong peerid, and dialing multiple addresses when only first one succeeds, the second errors.

Co-authored-by: Joonas Koivunen <joonas.koivunen@gmail.com>
Co-authored-by: Joonas Koivunen <joonas@equilibrium.co>

Author: 3 people

src/p2p/swarm.rs

@@ -292,8 +292,33 @@ impl NetworkBehaviour for SwarmApi {
         if let ConnectedPoint::Dialer { .. } = cp {
             let addr = MultiaddrWithPeerId::from((closed_addr, peer_id.to_owned()));
 
-            self.connect_registry
-                .finish_subscription(addr.into(), Err("Connection reset by peer".to_owned()));
+            match self.pending_connections.entry(*peer_id) {
+                Entry::Occupied(mut oe) => {
+                    let connections = oe.get_mut();
+                    let pos = connections.iter().position(|x| addr.multiaddr == *x);
+
+                    if let Some(pos) = pos {
+                        connections.swap_remove(pos);
+
+                        // this needs to be guarded, so that the connect test case doesn't cause a
+                        // panic following inject_connection_established, inject_connection_closed
+                        // if there's only the DummyProtocolsHandler, which doesn't open a
+                        // substream and closes up immediatedly.
+                        self.connect_registry.finish_subscription(
+                            addr.into(),
+                            Err("Connection reset by peer".to_owned()),
+                        );
+                    }
+
+                    if connections.is_empty() {
+                        oe.remove();
+                    }
+                }
+                Entry::Vacant(_) => {}
+            }
+        } else {
+            // we were not dialing to the peer, thus we cannot have a pending subscription to
+            // finish.
         }
     }
 
@@ -338,8 +363,8 @@ impl NetworkBehaviour for SwarmApi {
                 });
         }
 
-        // this should not be executed once, but probably will be in case unsupported addresses
-        // happen
+        // this should not be executed once, but probably will be in case unsupported addresses or something
+        // surprising happens.
         for failed in self.pending_connections.remove(peer_id).unwrap_or_default() {
             let addr = MultiaddrWithoutPeerId::try_from(failed)
                 .expect("peerid has been stripped earlier")
@@ -405,36 +430,188 @@ fn connection_point_addr(cp: &ConnectedPoint) -> &Multiaddr {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::p2p::transport::{build_transport, TTransport};
+    use crate::p2p::transport::build_transport;
+    use futures::{
+        stream::{StreamExt, TryStreamExt},
+        TryFutureExt,
+    };
     use libp2p::identity::Keypair;
-    use libp2p::{multiaddr::Protocol, multihash::Multihash, swarm::Swarm};
+    use libp2p::swarm::SwarmEvent;
+    use libp2p::{multiaddr::Protocol, multihash::Multihash, swarm::Swarm, swarm::SwarmBuilder};
     use std::convert::TryInto;
 
     #[tokio::test]
     async fn swarm_api() {
-        let (peer1_id, trans) = mk_transport();
-        let mut swarm1 = Swarm::new(trans, SwarmApi::default(), peer1_id);
-
-        let (peer2_id, trans) = mk_transport();
-        let mut swarm2 = Swarm::new(trans, SwarmApi::default(), peer2_id);
+        let (peer1_id, mut swarm1) = build_swarm();
+        let (peer2_id, mut swarm2) = build_swarm();
 
         Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
 
-        for l in Swarm::listeners(&swarm1) {
-            let mut addr = l.to_owned();
+        loop {
+            if let SwarmEvent::NewListenAddr(_) = swarm1.next_event().await {
+                break;
+            }
+        }
+
+        let listeners = Swarm::listeners(&swarm1).cloned().collect::<Vec<_>>();
+
+        for mut addr in listeners {
             addr.push(Protocol::P2p(
                 Multihash::from_bytes(&peer1_id.to_bytes()).unwrap(),
             ));
-            if let Some(fut) = swarm2.connect(addr.try_into().unwrap()) {
-                fut.await.unwrap();
+
+            let mut sub = swarm2.connect(addr.try_into().unwrap()).unwrap();
+
+            loop {
+                tokio::select! {
+                    _ = (&mut swarm1).next_event() => {},
+                    _ = (&mut swarm2).next_event() => {},
+                    res = (&mut sub) => {
+                        // this is currently a success even though the connection is never really
+                        // established, the DummyProtocolsHandler doesn't do anything nor want the
+                        // connection to be kept alive and thats it.
+                        //
+                        // it could be argued that this should be `Err("keepalive disconnected")`
+                        // or something and I'd agree, but I also agree this can be an `Ok(())`;
+                        // it's the sort of difficulty with the cli functionality in general: what
+                        // does it mean to connect to a peer? one way to look at it would be to
+                        // make the peer a "pinned peer" or "friend" and to keep the connection
+                        // alive at all costs. perhaps that is something for the next round.
+                        // another aspect would be to fail this future because there was no
+                        // `inject_connected`, only `inject_connection_established`. taking that
+                        // route would be good; it does however leave the special case of adding
+                        // another connection, which does add even more complexity than it exists
+                        // at the present.
+                        res.unwrap();
+
+                        // just to confirm that there are no connections.
+                        assert_eq!(Vec::<Multiaddr>::new(), swarm1.connections_to(&peer2_id));
+                        break;
+                    }
+                }
             }
         }
     }
 
-    fn mk_transport() -> (PeerId, TTransport) {
+    #[tokio::test]
+    async fn wrong_peerid() {
+        let (_, mut swarm1) = build_swarm();
+        let (_, mut swarm2) = build_swarm();
+
+        let peer3_id = Keypair::generate_ed25519().public().into_peer_id();
+
+        Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
+
+        let address;
+
+        loop {
+            if let SwarmEvent::NewListenAddr(addr) = swarm1.next_event().await {
+                // wonder if there should be a timeout?
+                address = addr;
+                break;
+            }
+        }
+
+        let mut fut = swarm2
+            .connect(
+                MultiaddrWithoutPeerId::try_from(address)
+                    .unwrap()
+                    .with(peer3_id),
+            )
+            .unwrap()
+            // remove the private type wrapper
+            .map_err(|e| e.into_inner());
+
+        loop {
+            tokio::select! {
+                _ = swarm1.next_event() => {},
+                _ = swarm2.next_event() => {},
+                res = &mut fut => {
+                    assert_eq!(res.unwrap_err(), Some("Pending connection: Invalid peer ID.".into()));
+                    return;
+                }
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn racy_connecting_attempts() {
+        let (peer1_id, mut swarm1) = build_swarm();
+        let (_, mut swarm2) = build_swarm();
+
+        Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
+        Swarm::listen_on(&mut swarm1, "/ip4/127.0.0.1/tcp/0".parse().unwrap()).unwrap();
+
+        let mut addresses = Vec::with_capacity(2);
+
+        while addresses.len() < 2 {
+            if let SwarmEvent::NewListenAddr(addr) = swarm1.next_event().await {
+                addresses.push(addr);
+            }
+        }
+
+        let targets = (
+            MultiaddrWithoutPeerId::try_from(addresses[0].clone())
+                .unwrap()
+                .with(peer1_id),
+            MultiaddrWithoutPeerId::try_from(addresses[1].clone())
+                .unwrap()
+                .with(peer1_id),
+        );
+
+        let mut connections = futures::stream::FuturesOrdered::new();
+        // these two should be attempted in parallel. since we know both of them work, and they are
+        // given in this order, we know that in libp2p 0.34 only the first should win, however
+        // both should always be finished.
+        connections.push(swarm2.connect(targets.0).unwrap());
+        connections.push(swarm2.connect(targets.1).unwrap());
+        let ready = connections
+            // turn the private error type into Option
+            .map_err(|e| e.into_inner())
+            .collect::<Vec<_>>();
+
+        tokio::pin!(ready);
+
+        loop {
+            tokio::select! {
+                _ = swarm1.next_event() => {}
+                _ = swarm2.next_event() => {}
+                res = &mut ready => {
+
+                    assert_eq!(
+                        res,
+                        vec![
+                            Ok(()),
+                            Err(Some("finished connecting to another address".into()))
+                        ]);
+
+                    break;
+                }
+            }
+        }
+    }
+
+    fn build_swarm() -> (PeerId, libp2p::swarm::Swarm<SwarmApi>) {
         let key = Keypair::generate_ed25519();
         let peer_id = key.public().into_peer_id();
         let transport = build_transport(key).unwrap();
-        (peer_id, transport)
+
+        let swarm = SwarmBuilder::new(transport, SwarmApi::default(), peer_id)
+            .executor(Box::new(ThreadLocalTokio))
+            .build();
+        (peer_id, swarm)
+    }
+
+    use std::future::Future;
+    use std::pin::Pin;
+
+    // can only be used from within tokio context. this is required since otherwise libp2p-tcp will
+    // use tokio, but from a futures-executor threadpool, which is outside of tokio context.
+    struct ThreadLocalTokio;
+
+    impl libp2p::core::Executor for ThreadLocalTokio {
+        fn exec(&self, future: Pin<Box<dyn Future<Output = ()> + Send + 'static>>) {
+            tokio::task::spawn(future);
+        }
     }
 }

src/subscription.rs

@@ -144,6 +144,9 @@ impl<T: Debug + Clone + PartialEq, E: Debug + Clone> SubscriptionRegistry<T, E>
 
             #[cfg(debug_assertions)]
             if awoken == 0 {
+                // this assertion was originally added because we had quite a lot of trouble with
+                // the subscriptions and it still isn't perfect. if you hit these assertion please
+                // do report a bug.
                 let msg = format!(
                     "no subscriptions to be awoken! subs: {:?}; req_kind: {:?}",
                     related_subs, req_kind
@@ -216,6 +219,16 @@ impl<E: Debug + PartialEq> fmt::Display for SubscriptionErr<E> {
 
 impl<E: Debug + PartialEq> std::error::Error for SubscriptionErr<E> {}
 
+impl<E: Debug + PartialEq> SubscriptionErr<E> {
+    pub fn into_inner(self) -> Option<E> {
+        use SubscriptionErr::*;
+        match self {
+            Cancelled => None,
+            Failed(e) => Some(e),
+        }
+    }
+}
+
 /// Represents a request for a resource at different stages of its lifetime.
 pub enum Subscription<T, E> {
     /// A finished `Subscription` containing the desired `TRes` value.