feature: implement libp2p node with customizable request-response protocols

Cargo.toml

@@ -5,12 +5,15 @@ members = [
     "crates/validator",
     "crates/shared",
     "crates/orchestrator",
+    "crates/p2p",
     "crates/dev-utils",
 ]
 resolver = "2"
 
 [workspace.dependencies]
 shared = { path = "crates/shared" }
+p2p = { path = "crates/p2p" }
+
 actix-web = "4.9.0"
 clap = { version = "4.5.27", features = ["derive"] }
 serde = { version = "1.0.219", features = ["derive"] }
@@ -42,6 +45,7 @@ rand_core_v6 = { package = "rand_core", version = "0.6.4", features = ["std"] }
 ipld-core = "0.4"
 rust-ipfs = "0.14"
 cid = "0.11"
+tracing = "0.1.41"
 
 [workspace.package]
 version = "0.3.11"

crates/p2p/Cargo.toml

@@ -0,0 +1,18 @@
+[package]
+name = "p2p"
+version.workspace = true
+edition.workspace = true
+
+[dependencies]
+libp2p = { version = "0.54", features = ["request-response", "identify", "ping", "mdns", "noise", "tcp", "autonat", "kad", "tokio", "cbor", "macros", "yamux"] }
+void = "1.0"
+
+anyhow = {workspace = true}
+nalgebra = {workspace = true}
+serde = {workspace = true}
+tokio = {workspace = true, features = ["sync"]}
+tokio-util = { workspace = true, features = ["rt"] }
+tracing = { workspace = true }
+
+[lints]
+workspace = true

crates/p2p/src/behaviour.rs

@@ -0,0 +1,184 @@
+use anyhow::Context as _;
+use anyhow::Result;
+use libp2p::autonat;
+use libp2p::connection_limits;
+use libp2p::connection_limits::ConnectionLimits;
+use libp2p::identify;
+use libp2p::identity;
+use libp2p::kad;
+use libp2p::kad::store::MemoryStore;
+use libp2p::mdns;
+use libp2p::ping;
+use libp2p::request_response;
+use libp2p::swarm::NetworkBehaviour;
+use std::time::Duration;
+use tracing::debug;
+
+use crate::message::IncomingMessage;
+use crate::message::{Request, Response};
+use crate::Protocols;
+use crate::PRIME_STREAM_PROTOCOL;
+
+#[derive(NetworkBehaviour)]
+#[behaviour(to_swarm = "BehaviourEvent")]
+pub(crate) struct Behaviour {
+    // connection gating
+    connection_limits: connection_limits::Behaviour,
+
+    // discovery
+    mdns: mdns::tokio::Behaviour,
+    kademlia: kad::Behaviour<MemoryStore>,
+
+    // protocols
+    identify: identify::Behaviour,
+    ping: ping::Behaviour,
+    request_response: request_response::cbor::Behaviour<Request, Response>,
+
+    // nat traversal
+    autonat: autonat::Behaviour,
+}
+
+#[allow(clippy::large_enum_variant)]
+#[derive(Debug)]
+pub(crate) enum BehaviourEvent {
+    Autonat(autonat::Event),
+    Identify(identify::Event),
+    Kademlia(kad::Event),
+    Mdns(mdns::Event),
+    Ping(ping::Event),
+    RequestResponse(request_response::Event<Request, Response>),
+}
+
+impl From<void::Void> for BehaviourEvent {
+    fn from(_: void::Void) -> Self {
+        unreachable!("void::Void cannot be converted to BehaviourEvent")
+    }
+}
+
+impl From<autonat::Event> for BehaviourEvent {
+    fn from(event: autonat::Event) -> Self {
+        BehaviourEvent::Autonat(event)
+    }
+}
+
+impl From<kad::Event> for BehaviourEvent {
+    fn from(event: kad::Event) -> Self {
+        BehaviourEvent::Kademlia(event)
+    }
+}
+
+impl From<libp2p::mdns::Event> for BehaviourEvent {
+    fn from(event: libp2p::mdns::Event) -> Self {
+        BehaviourEvent::Mdns(event)
+    }
+}
+
+impl From<ping::Event> for BehaviourEvent {
+    fn from(event: ping::Event) -> Self {
+        BehaviourEvent::Ping(event)
+    }
+}
+
+impl From<identify::Event> for BehaviourEvent {
+    fn from(event: identify::Event) -> Self {
+        BehaviourEvent::Identify(event)
+    }
+}
+
+impl From<request_response::Event<Request, Response>> for BehaviourEvent {
+    fn from(event: request_response::Event<Request, Response>) -> Self {
+        BehaviourEvent::RequestResponse(event)
+    }
+}
+
+impl Behaviour {
+    pub(crate) fn new(
+        keypair: &identity::Keypair,
+        protocols: Protocols,
+        agent_version: String,
+    ) -> Result<Self> {
+        let peer_id = keypair.public().to_peer_id();
+
+        let protocols = protocols.into_iter().map(|protocol| {
+            (
+                protocol.as_stream_protocol(),
+                request_response::ProtocolSupport::Full, // TODO: configure inbound/outbound based on node role and protocol
+            )
+        });
+
+        let autonat = autonat::Behaviour::new(peer_id, autonat::Config::default());
+        let connection_limits = connection_limits::Behaviour::new(
+            ConnectionLimits::default().with_max_established(Some(100)),
+        );
+
+        let mdns = mdns::tokio::Behaviour::new(mdns::Config::default(), peer_id)
+            .context("failed to create mDNS behaviour")?;
+        let kademlia = kad::Behaviour::new(peer_id, MemoryStore::new(peer_id));
+
+        let identify = identify::Behaviour::new(
+            identify::Config::new(PRIME_STREAM_PROTOCOL.to_string(), keypair.public())
+                .with_agent_version(agent_version),
+        );
+        let ping = ping::Behaviour::new(ping::Config::new().with_interval(Duration::from_secs(10)));
+
+        Ok(Self {
+            autonat,
+            connection_limits,
+            kademlia,
+            mdns,
+            identify,
+            ping,
+            request_response: request_response::cbor::Behaviour::new(
+                protocols,
+                request_response::Config::default(),
+            ),
+        })
+    }
+
+    pub(crate) fn request_response(
+        &mut self,
+    ) -> &mut request_response::cbor::Behaviour<Request, Response> {
+        &mut self.request_response
+    }
+}
+
+impl BehaviourEvent {
+    pub(crate) async fn handle(self, message_tx: tokio::sync::mpsc::Sender<IncomingMessage>) {
+        match self {
+            BehaviourEvent::Autonat(_event) => {}
+            BehaviourEvent::Identify(_event) => {}
+            BehaviourEvent::Kademlia(_event) => { // TODO: potentially on outbound queries
+            }
+            BehaviourEvent::Mdns(_event) => {}
+            BehaviourEvent::Ping(_event) => {}
+            BehaviourEvent::RequestResponse(event) => match event {
+                request_response::Event::Message { peer, message } => {
+                    debug!("received message from peer {peer:?}: {message:?}");
+                    // if this errors, user dropped their incoming message channel
+                    let _ = message_tx.send(IncomingMessage { peer, message }).await;
+                }
+                request_response::Event::ResponseSent { peer, request_id } => {
+                    debug!("response sent to peer {peer:?} for request ID {request_id:?}");
+                }
+                request_response::Event::InboundFailure {
+                    peer,
+                    request_id,
+                    error,
+                } => {
+                    debug!(
+                        "inbound failure from peer {peer:?} for request ID {request_id:?}: {error}"
+                    );
+                }
+                request_response::Event::OutboundFailure {
+                    peer,
+                    request_id,
+                    error,
+                } => {
+                    debug!(
+                        "outbound failure to peer {peer:?} for request ID {request_id:?}: {error}"
+                    );
+                }
+            },
+        }
+    }
+}