Vendor patched libp2p-stream instead of forking libp2p

Cargo.toml

@@ -3,18 +3,13 @@ members = ["crates/*"]
 resolver = "2"
 
 [workspace.dependencies]
-libp2p = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
+libp2p = { version = "0.56.0" }
+libp2p-core = { version = "0.43.1" }
 libp2p-identity = { version = "0.2.12", features = ["peerid"] }
-libp2p-connection-limits = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
-libp2p-swarm-derive = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
-libp2p-stream = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
-libp2p-swarm-test = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
-
-# The fork builds its crates against its in-tree libp2p-identity (path = "identity"), so consumers
-# must redirect the crates.io libp2p-identity to the same source, otherwise two incompatible copies
-# of libp2p_identity are linked. See the fork's "Use libp2p-identity from the fork" commit.
-[patch.crates-io]
-libp2p-identity = { git = "https://github.com/kalabukdima/rust-libp2p.git", rev = "fac3b0c9" }
+libp2p-connection-limits = { version = "0.6.0" }
+libp2p-swarm = { version = "0.47.0" }
+libp2p-swarm-derive = { version = "0.35.1" }
+libp2p-swarm-test = { version = "0.6.0" }
 
 [workspace.lints.rust]
 unsafe_code = "deny"

crates/transport/Cargo.toml

@@ -15,7 +15,14 @@ futures-core = "0.3"
 lazy_static = { version = "1" }
 libp2p = { workspace = true, features = ["dns", "tokio", "identify", "kad", "ping", "request-response", "serde", "autonat", "quic"] }
 libp2p-connection-limits = { workspace = true }
-libp2p-stream = { workspace = true }
+# libp2p-core, libp2p-identity, libp2p-swarm, rand and tracing are used by the vendored
+# `libp2p_stream` module only — the rest of the crate goes through the `libp2p` facade. Bump
+# these three together with `libp2p` itself: pointing them at a version the facade does not use
+# links two copies of the swarm types and fails with "expected NetworkBehaviour, found
+# NetworkBehaviour".
+libp2p-core = { workspace = true }
+libp2p-identity = { workspace = true }
+libp2p-swarm = { workspace = true }
 libp2p-swarm-derive = { workspace = true }
 log = "0.4"
 lru = "0.12"
@@ -29,16 +36,17 @@ thiserror = "1"
 tokio = { version = "1", features = ["fs", "macros", "sync"] }
 tokio-util = { version = "0.7", features = ["time"] }
 tokio-stream = "0.1.17"
+tracing = { version = "0.1", optional = true }
 
 sqd-contract-client = { path = "../contract-client" }
 sqd-messages = { path = "../messages", features = ["signatures", "semver"] }
 
 [features]
 actors = ["behaviour"]
-behaviour = []
+behaviour = ["dep:rand", "dep:tracing"]
 proto = []
 pubsub = ["behaviour", "libp2p/gossipsub"]
-noise = ["rand"]
+noise = ["dep:rand"]
 request-server = []
 stream-server = ["behaviour"]
 portal = ["actors"]

crates/transport/src/behaviour/keep_alive.rs

@@ -9,7 +9,7 @@ use libp2p::{
 };
 use std::task::{Context, Poll};
 
-use crate::{protocol::KEEP_ALIVE_PROTOCOL, Multiaddr, PeerId};
+use crate::{libp2p_stream, protocol::KEEP_ALIVE_PROTOCOL, Multiaddr, PeerId};
 
 /// A behaviour that allows keeping connections alive forever.
 ///

crates/transport/src/behaviour/noise.rs

@@ -9,7 +9,7 @@ use libp2p::{
 };
 use std::task::{Context, Poll};
 
-use crate::{protocol::NOISE_PROTOCOL, Multiaddr, PeerId};
+use crate::{libp2p_stream, protocol::NOISE_PROTOCOL, Multiaddr, PeerId};
 
 const BUFFER_SIZE: usize = 10 * 1024 * 1024; // 10 MB
 

crates/transport/src/behaviour/stream_client.rs

@@ -2,9 +2,12 @@ use std::{collections::HashMap, sync::Arc, time::Duration};
 
 use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
 use libp2p::{PeerId, StreamProtocol};
-use libp2p_stream::OpenStreamError;
 
-use crate::{util::StreamWithPayload, BehaviourWrapper};
+use crate::{
+    libp2p_stream::{self, OpenStreamError},
+    util::StreamWithPayload,
+    BehaviourWrapper,
+};
 
 #[derive(Debug, Clone, Copy)]
 pub struct ClientConfig {
@@ -136,18 +139,13 @@ impl Clone for StreamClientHandle {
     }
 }
 
+// Derived rather than written out: now that `libp2p_stream` lives in this crate, clippy sees
+// that `Behaviour: Default` and flags the manual impl as derivable.
+#[derive(Default)]
 pub struct ClientBehaviour {
     inner: libp2p_stream::Behaviour,
 }
 
-impl Default for ClientBehaviour {
-    fn default() -> Self {
-        Self {
-            inner: libp2p_stream::Behaviour::new(),
-        }
-    }
-}
-
 impl ClientBehaviour {
     pub fn new_handle(&self, protocol: &'static str, config: ClientConfig) -> StreamClientHandle {
         let control = self.inner.new_control();
@@ -174,7 +172,6 @@ impl From<OpenStreamError> for RequestError {
         match e {
             OpenStreamError::UnsupportedProtocol(_) => Self::UnsupportedProtocol,
             OpenStreamError::Io(e) => Self::Io(e),
-            _ => unreachable!(),
         }
     }
 }

crates/transport/src/behaviour/stream_server.rs

@@ -5,7 +5,10 @@ use futures::{AsyncReadExt, AsyncWriteExt, StreamExt};
 use libp2p::{swarm::ToSwarm, PeerId, Stream, StreamProtocol};
 use tokio::sync::mpsc;
 
-use crate::behaviour::wrapped::{BehaviourWrapper, TToSwarm};
+use crate::{
+    behaviour::wrapped::{BehaviourWrapper, TToSwarm},
+    libp2p_stream,
+};
 
 #[derive(Debug, Clone, Copy)]
 pub struct ServerConfig {

crates/transport/src/lib.rs

@@ -40,6 +40,8 @@ mod builder;
 mod cli;
 #[cfg(feature = "proto")]
 mod codec;
+#[cfg(feature = "behaviour")]
+pub mod libp2p_stream;
 #[cfg(feature = "metrics")]
 pub mod metrics;
 pub mod protocol;

crates/transport/src/libp2p_stream/LICENSE

@@ -0,0 +1,18 @@
+Copyright 2017-2020 Parity Technologies (UK) Ltd.
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of
+this software and associated documentation files (the "Software"), to deal in
+the Software without restriction, including without limitation the rights to
+use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
+the Software, and to permit persons to whom the Software is furnished to do so,
+subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
+FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS
+OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
+WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
+CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

crates/transport/src/libp2p_stream/behaviour.rs

@@ -0,0 +1,146 @@
+use core::fmt;
+use std::{
+    sync::{Arc, Mutex},
+    task::{Context, Poll},
+};
+
+use futures::{channel::mpsc, StreamExt};
+use libp2p_core::{transport::PortUse, Endpoint, Multiaddr};
+use libp2p_identity::PeerId;
+use libp2p_swarm::{
+    self as swarm, dial_opts::DialOpts, ConnectionDenied, ConnectionId, FromSwarm,
+    NetworkBehaviour, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm,
+};
+use swarm::{
+    behaviour::ConnectionEstablished, dial_opts::PeerCondition, ConnectionClosed, DialError,
+    DialFailure,
+};
+
+use super::{handler::Handler, shared::Shared, Control};
+
+/// A generic behaviour for stream-oriented protocols.
+pub struct Behaviour {
+    shared: Arc<Mutex<Shared>>,
+    dial_receiver: mpsc::UnboundedReceiver<PeerId>,
+}
+
+impl Default for Behaviour {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Behaviour {
+    pub fn new() -> Self {
+        let (dial_sender, dial_receiver) = mpsc::unbounded();
+
+        Self {
+            shared: Arc::new(Mutex::new(Shared::new(dial_sender))),
+            dial_receiver,
+        }
+    }
+
+    /// Obtain a new [`Control`].
+    pub fn new_control(&self) -> Control {
+        Control::new(self.shared.clone())
+    }
+}
+
+/// The protocol is already registered.
+#[derive(Debug)]
+pub struct AlreadyRegistered;
+
+impl fmt::Display for AlreadyRegistered {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        write!(f, "The protocol is already registered")
+    }
+}
+
+impl std::error::Error for AlreadyRegistered {}
+
+impl NetworkBehaviour for Behaviour {
+    type ConnectionHandler = Handler;
+    type ToSwarm = ();
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(
+            peer,
+            self.shared.clone(),
+            Shared::lock(&self.shared).receiver(peer, connection_id),
+        ))
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        connection_id: ConnectionId,
+        peer: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+        _: PortUse,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(
+            peer,
+            self.shared.clone(),
+            Shared::lock(&self.shared).receiver(peer, connection_id),
+        ))
+    }
+
+    fn on_swarm_event(&mut self, event: FromSwarm) {
+        match event {
+            FromSwarm::ConnectionEstablished(ConnectionEstablished {
+                peer_id,
+                connection_id,
+                ..
+            }) => Shared::lock(&self.shared).on_connection_established(connection_id, peer_id),
+            FromSwarm::ConnectionClosed(ConnectionClosed { connection_id, .. }) => {
+                Shared::lock(&self.shared).on_connection_closed(connection_id)
+            }
+            FromSwarm::DialFailure(DialFailure {
+                peer_id: Some(peer_id),
+                error:
+                    error @ (DialError::LocalPeerId { .. }
+                    | DialError::Transport(_)
+                    | DialError::Denied { .. }
+                    | DialError::NoAddresses
+                    | DialError::Aborted
+                    | DialError::WrongPeerId { .. }),
+                ..
+            }) => {
+                let reason = error.to_string(); // We can only forward the string repr but it is better than nothing.
+
+                Shared::lock(&self.shared).on_dial_failure(peer_id, reason)
+            }
+            _ => {}
+        }
+    }
+
+    fn on_connection_handler_event(
+        &mut self,
+        _peer_id: PeerId,
+        _connection_id: ConnectionId,
+        event: THandlerOutEvent<Self>,
+    ) {
+        libp2p_core::util::unreachable(event);
+    }
+
+    fn poll(
+        &mut self,
+        cx: &mut Context<'_>,
+    ) -> Poll<ToSwarm<Self::ToSwarm, THandlerInEvent<Self>>> {
+        if let Poll::Ready(Some(peer)) = self.dial_receiver.poll_next_unpin(cx) {
+            return Poll::Ready(ToSwarm::Dial {
+                opts: DialOpts::peer_id(peer)
+                    .condition(PeerCondition::DisconnectedAndNotDialing)
+                    .build(),
+            });
+        }
+
+        Poll::Pending
+    }
+}