Merge master

.github/workflows/ci.yml

@@ -46,7 +46,7 @@ jobs:
     container:
       image: rust
       env:
-        CC: clang-10
+        CC: clang-11
     steps:
 
     - name: Cancel Previous Runs
@@ -75,7 +75,7 @@ jobs:
         wget -O - https://apt.llvm.org/llvm-snapshot.gpg.key | apt-key add -
         echo "deb http://apt.llvm.org/bionic/ llvm-toolchain-bionic-10 main" >> /etc/apt/sources.list
         apt-get update
-        apt-get install -y clang-10
+        apt-get install -y clang-11
 
     - name: Install CMake
       run: apt-get install -y cmake

CHANGELOG.md

@@ -16,6 +16,7 @@
 - [`libp2p-ping` CHANGELOG](protocols/ping/CHANGELOG.md)
 - [`libp2p-relay` CHANGELOG](protocols/relay/CHANGELOG.md)
 - [`libp2p-request-response` CHANGELOG](protocols/request-response/CHANGELOG.md)
+- [`libp2p-rendezvous` CHANGELOG](protocols/rendezvous/CHANGELOG.md)
 
 ## Transport Protocols & Upgrades
 
@@ -36,6 +37,7 @@
 
 ## Utilities
 
+- [`libp2p-metrics` CHANGELOG](misc/metrics/CHANGELOG.md)
 - [`multistream-select` CHANGELOG](misc/multistream-select/CHANGELOG.md)
 
 # `libp2p` facade crate
@@ -66,7 +68,9 @@
 
 - Re-export the `wasm-bindgen` feature from `parking_lot`, so
   `libp2p` users can opt-in to that crate's Wasm support. See [PR 2180].
-
+
+- Add `libp2p-metrics`.
+
 [PR 2180]: https://github.com/libp2p/rust-libp2p/pull/2180/
 
 ## Version 0.39.1 [2021-07-12]

Cargo.toml

@@ -26,6 +26,7 @@ default = [
     "quic",
     "relay",
     "request-response",
+    "rendezvous",
     "secp256k1",
     "tcp-async-io",
     "uds",
@@ -38,17 +39,19 @@ dns-async-std = ["libp2p-dns", "libp2p-dns/async-std"]
 dns-tokio = ["libp2p-dns", "libp2p-dns/tokio"]
 floodsub = ["libp2p-floodsub"]
 gossipsub = ["libp2p-gossipsub"]
-identify = ["libp2p-identify"]
-kad = ["libp2p-kad"]
+identify = ["libp2p-identify", "libp2p-metrics/identify"]
+kad = ["libp2p-kad", "libp2p-metrics/kad"]
+metrics = ["libp2p-metrics"]
 mdns = ["libp2p-mdns"]
 mplex = ["libp2p-mplex"]
 noise = ["libp2p-noise"]
-ping = ["libp2p-ping"]
+ping = ["libp2p-ping", "libp2p-metrics/ping"]
 plaintext = ["libp2p-plaintext"]
 pnet = ["libp2p-pnet"]
 quic = ["libp2p-quic"]
 relay = ["libp2p-relay"]
 request-response = ["libp2p-request-response"]
+rendezvous = ["libp2p-rendezvous"]
 tcp-async-io = ["libp2p-tcp", "libp2p-tcp/async-io"]
 tcp-tokio = ["libp2p-tcp", "libp2p-tcp/tokio"]
 uds = ["libp2p-uds"]
@@ -72,15 +75,17 @@ libp2p-floodsub = { version = "0.31.0", path = "protocols/floodsub", optional =
 libp2p-gossipsub = { version = "0.33.0", path = "./protocols/gossipsub", optional = true }
 libp2p-identify = { version = "0.31.0", path = "protocols/identify", optional = true }
 libp2p-kad = { version = "0.32.0", path = "protocols/kad", optional = true }
+libp2p-metrics = { version = "0.1.0", path = "misc/metrics", optional = true }
 libp2p-mplex = { version = "0.30.0", path = "muxers/mplex", optional = true }
 libp2p-noise = { version = "0.33.0", path = "transports/noise", optional = true }
 libp2p-ping = { version = "0.31.0", path = "protocols/ping", optional = true }
 libp2p-plaintext = { version = "0.30.0", path = "transports/plaintext", optional = true }
-libp2p-pnet = { version = "0.21.0", path = "transports/pnet", optional = true }
+libp2p-pnet = { version = "0.22.0", path = "transports/pnet", optional = true }
 libp2p-relay = { version = "0.4.0", path = "protocols/relay", optional = true }
+libp2p-rendezvous = { version = "0.1.0", path = "protocols/rendezvous", optional = true }
 libp2p-request-response = { version = "0.13.0", path = "protocols/request-response", optional = true }
 libp2p-swarm = { version = "0.31.0", path = "swarm" }
-libp2p-swarm-derive = { version = "0.24.0", path = "swarm-derive" }
+libp2p-swarm-derive = { version = "0.25.0", path = "swarm-derive" }
 libp2p-uds = { version = "0.30.0", path = "transports/uds", optional = true }
 libp2p-wasm-ext = { version = "0.30.0", path = "transports/wasm-ext", default-features = false, optional = true }
 libp2p-yamux = { version = "0.34.0", path = "muxers/yamux", optional = true }
@@ -100,19 +105,23 @@ libp2p-websocket = { version = "0.31.0", path = "transports/websocket", optional
 
 [dev-dependencies]
 async-std = { version = "1.6.2", features = ["attributes"] }
+async-trait = "0.1"
 env_logger = "0.9.0"
+structopt = "0.3.21"
 tokio = { version = "1.0.1", features = ["io-util", "io-std", "macros", "rt", "rt-multi-thread"] }
 
 [workspace]
 resolver = "2"
 members = [
     "core",
+    "misc/metrics",
     "misc/multistream-select",
     "misc/peer-id-generator",
     "muxers/mplex",
     "muxers/yamux",
     "protocols/floodsub",
     "protocols/gossipsub",
+    "protocols/rendezvous",
     "protocols/identify",
     "protocols/kad",
     "protocols/mdns",

README.md

@@ -12,7 +12,10 @@ This repository is the central place for Rust development of the [libp2p](https:
 - The **[examples](examples)** folder contains small binaries showcasing the
   many protocols in this repository.
 
-- To **report bugs, suggest improvements or request new features** please open
+- For **security related issues** please reach out to security@ipfs.io. Please
+  do not file a public issue on GitHub.
+
+- To **report bugs, suggest improvements or request new features** please open a
   GitHub issue on this repository.
 
 - For **rust-libp2p specific questions** please use the GitHub _Discussions_

core/CHANGELOG.md

@@ -20,10 +20,32 @@
 - Require `ConnectionHandler::{InEvent,OutEvent,Error}` to implement `Debug`
   (see [PR 2183]).
 
+- Remove `DisconnectedPeer::set_connected` and `Pool::add` (see [PR 2195]).
+
+- Report `ConnectionLimit` error through `ConnectionError` and thus through
+  `NetworkEvent::ConnectionClosed` instead of previously through
+  `PendingConnectionError` and thus `NetworkEvent::{IncomingConnectionError,
+  DialError}` (see [PR 2191]).
+
+- Report abortion of pending connection through `DialError`,
+  `UnknownPeerDialError` or `IncomingConnectionError` (see [PR 2191]).
+
+- Remove deprecated functions `upgrade::write_one`, `upgrade::write_with_len_prefix`
+  and `upgrade::read_one` (see [PR 2213]).
+
+- Add `SignedEnvelope` and `PeerRecord` according to [RFC0002] and [RFC0003]
+  (see [PR 2107]).
+
 [PR 2145]: https://github.com/libp2p/rust-libp2p/pull/2145
+[PR 2213]: https://github.com/libp2p/rust-libp2p/pull/2213
 [PR 2142]: https://github.com/libp2p/rust-libp2p/pull/2142
 [PR 2137]: https://github.com/libp2p/rust-libp2p/pull/2137
 [PR 2183]: https://github.com/libp2p/rust-libp2p/pull/2183
+[PR 2191]: https://github.com/libp2p/rust-libp2p/pull/2191
+[PR 2195]: https://github.com/libp2p/rust-libp2p/pull/2195
+[PR 2107]: https://github.com/libp2p/rust-libp2p/pull/2107
+[RFC0002]: https://github.com/libp2p/specs/blob/master/RFC/0002-signed-envelopes.md
+[RFC0003]: https://github.com/libp2p/specs/blob/master/RFC/0003-routing-records.md
 
 # 0.29.0 [2021-07-12]
 

core/build.rs

@@ -19,5 +19,13 @@
 // DEALINGS IN THE SOFTWARE.
 
 fn main() {
-    prost_build::compile_protos(&["src/keys.proto"], &["src"]).unwrap();
+    prost_build::compile_protos(
+        &[
+            "src/keys.proto",
+            "src/envelope.proto",
+            "src/peer_record.proto",
+        ],
+        &["src"],
+    )
+    .unwrap();
 }

core/src/connection.rs

@@ -229,10 +229,10 @@ where
         self.handler.inject_event(event);
     }
 
-    /// Begins an orderly shutdown of the connection, returning a
-    /// `Future` that resolves when connection shutdown is complete.
-    pub fn close(self) -> Close<TMuxer> {
-        self.muxing.close().0
+    /// Begins an orderly shutdown of the connection, returning the connection
+    /// handler and a `Future` that resolves when connection shutdown is complete.
+    pub fn close(self) -> (THandler, Close<TMuxer>) {
+        (self.handler, self.muxing.close().0)
     }
 
     /// Polls the connection for events produced by the associated handler

core/src/connection/error.rs

@@ -29,6 +29,10 @@ pub enum ConnectionError<THandlerErr> {
     // TODO: Eventually this should also be a custom error?
     IO(io::Error),
 
+    /// The connection was dropped because the connection limit
+    /// for a peer has been reached.
+    ConnectionLimit(ConnectionLimit),
+
     /// The connection handler produced an error.
     Handler(THandlerErr),
 }
@@ -41,6 +45,9 @@ where
         match self {
             ConnectionError::IO(err) => write!(f, "Connection error: I/O error: {}", err),
             ConnectionError::Handler(err) => write!(f, "Connection error: Handler error: {}", err),
+            ConnectionError::ConnectionLimit(l) => {
+                write!(f, "Connection error: Connection limit: {}.", l)
+            }
         }
     }
 }
@@ -53,6 +60,7 @@ where
         match self {
             ConnectionError::IO(err) => Some(err),
             ConnectionError::Handler(err) => Some(err),
+            ConnectionError::ConnectionLimit(..) => None,
         }
     }
 }
@@ -63,14 +71,13 @@ pub enum PendingConnectionError<TTransErr> {
     /// An error occurred while negotiating the transport protocol(s).
     Transport(TransportError<TTransErr>),
 
+    /// Pending connection attempt has been aborted.
+    Aborted,
+
     /// The peer identity obtained on the connection did not
     /// match the one that was expected or is otherwise invalid.
     InvalidPeerId,
 
-    /// The connection was dropped because the connection limit
-    /// for a peer has been reached.
-    ConnectionLimit(ConnectionLimit),
-
     /// An I/O error occurred on the connection.
     // TODO: Eventually this should also be a custom error?
     IO(io::Error),
@@ -83,15 +90,13 @@ where
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         match self {
             PendingConnectionError::IO(err) => write!(f, "Pending connection: I/O error: {}", err),
+            PendingConnectionError::Aborted => write!(f, "Pending connection: Aborted."),
             PendingConnectionError::Transport(err) => {
                 write!(f, "Pending connection: Transport error: {}", err)
             }
             PendingConnectionError::InvalidPeerId => {
                 write!(f, "Pending connection: Invalid peer ID.")
             }
-            PendingConnectionError::ConnectionLimit(l) => {
-                write!(f, "Connection error: Connection limit: {}.", l)
-            }
         }
     }
 }
@@ -105,7 +110,7 @@ where
             PendingConnectionError::IO(err) => Some(err),
             PendingConnectionError::Transport(err) => Some(err),
             PendingConnectionError::InvalidPeerId => None,
-            PendingConnectionError::ConnectionLimit(..) => None,
+            PendingConnectionError::Aborted => None,
         }
     }
 }

core/src/connection/manager.rs

@@ -20,7 +20,7 @@
 
 use super::{
     handler::{THandlerError, THandlerInEvent, THandlerOutEvent},
-    Connected, ConnectedPoint, Connection, ConnectionError, ConnectionHandler,
+    Connected, ConnectedPoint, ConnectionError, ConnectionHandler, ConnectionLimit,
     IntoConnectionHandler, PendingConnectionError, Substream,
 };
 use crate::{muxing::StreamMuxer, Executor};
@@ -192,6 +192,7 @@ pub enum Event<'a, H: IntoConnectionHandler, TE> {
         /// The error that occurred, if any. If `None`, the connection
         /// has been actively closed.
         error: Option<ConnectionError<THandlerError<H>>>,
+        handler: H::Handler,
     },
 
     /// A connection has been established.
@@ -276,40 +277,6 @@ impl<H: IntoConnectionHandler, TE> Manager<H, TE> {
         ConnectionId(task_id)
     }
 
-    /// Adds an existing connection to the manager.
-    pub fn add<M>(&mut self, conn: Connection<M, H::Handler>, info: Connected) -> ConnectionId
-    where
-        H: IntoConnectionHandler + Send + 'static,
-        H::Handler: ConnectionHandler<Substream = Substream<M>> + Send + 'static,
-        <H::Handler as ConnectionHandler>::OutboundOpenInfo: Send + 'static,
-        TE: error::Error + Send + 'static,
-        M: StreamMuxer + Send + Sync + 'static,
-        M::OutboundSubstream: Send + 'static,
-    {
-        let task_id = self.next_task_id;
-        self.next_task_id.0 += 1;
-
-        let (tx, rx) = mpsc::channel(self.task_command_buffer_size);
-        self.tasks.insert(
-            task_id,
-            TaskInfo {
-                sender: tx,
-                state: TaskState::Established(info),
-            },
-        );
-
-        let task: Pin<Box<Task<Pin<Box<future::Pending<_>>>, _, _, _>>> =
-            Box::pin(Task::established(task_id, self.events_tx.clone(), rx, conn));
-
-        if let Some(executor) = &mut self.executor {
-            executor.exec(task);
-        } else {
-            self.local_spawns.push(task);
-        }
-
-        ConnectionId(task_id)
-    }
-
     /// Gets an entry for a managed connection, if it exists.
     pub fn entry(&mut self, id: ConnectionId) -> Option<Entry<'_, THandlerInEvent<H>>> {
         if let hash_map::Entry::Occupied(task) = self.tasks.entry(id.0) {
@@ -384,14 +351,15 @@ impl<H: IntoConnectionHandler, TE> Manager<H, TE> {
                         new_endpoint: new,
                     }
                 }
-                task::Event::Closed { id, error } => {
+                task::Event::Closed { id, error, handler } => {
                     let id = ConnectionId(id);
                     let task = task.remove();
                     match task.state {
                         TaskState::Established(connected) => Event::ConnectionClosed {
                             id,
                             connected,
                             error,
+                            handler,
                         },
                         TaskState::Pending => unreachable!(
                             "`Event::Closed` implies (2) occurred on that task and thus (3)."
@@ -471,15 +439,15 @@ impl<'a, I> EstablishedEntry<'a, I> {
     ///
     /// When the connection is ultimately closed, [`Event::ConnectionClosed`]
     /// is emitted by [`Manager::poll`].
-    pub fn start_close(mut self) {
+    pub fn start_close(mut self, error: Option<ConnectionLimit>) {
         // Clone the sender so that we are guaranteed to have
         // capacity for the close command (every sender gets a slot).
         match self
             .task
             .get_mut()
             .sender
             .clone()
-            .try_send(task::Command::Close)
+            .try_send(task::Command::Close(error))
         {
             Ok(()) => {}
             Err(e) => assert!(e.is_disconnected(), "No capacity for close command."),
@@ -494,17 +462,6 @@ impl<'a, I> EstablishedEntry<'a, I> {
         }
     }
 
-    /// Instantly removes the entry from the manager, dropping
-    /// the command channel to the background task of the connection,
-    /// which will thus drop the connection asap without an orderly
-    /// close or emitting another event.
-    pub fn remove(self) -> Connected {
-        match self.task.remove().state {
-            TaskState::Established(c) => c,
-            TaskState::Pending => unreachable!("By Entry::new()"),
-        }
-    }
-
     /// Returns the connection ID.
     pub fn id(&self) -> ConnectionId {
         ConnectionId(*self.task.key())