feat: unreliable datagram support (QUIC + libp2p-datagram)

.github/workflows/ci.yml

@@ -307,6 +307,10 @@ jobs:
         with:
           shared-key: semver
       - uses: obi1kenobi/cargo-semver-checks-action@v2
+        with:
+          # `libp2p-datagram` is new and unpublished, so it has no crates.io
+          # baseline to diff against. Remove this once it is first released.
+          exclude: libp2p-datagram
 
   rustfmt:
     runs-on: ubuntu-latest

Cargo.toml

@@ -42,6 +42,7 @@ members = [
     "protocols/gossipsub",
     "protocols/identify",
     "protocols/kad",
+    "protocols/datagram",
     "protocols/mdns",
     "protocols/perf",
     "protocols/ping",
@@ -102,6 +103,7 @@ libp2p-relay = { version = "0.22.0", path = "protocols/relay" }
 libp2p-rendezvous = { version = "0.18.0", path = "protocols/rendezvous" }
 libp2p-request-response = { version = "0.30.0", path = "protocols/request-response" }
 libp2p-server = { version = "0.13.0", path = "misc/server" }
+libp2p-datagram = { version = "0.1.0", path = "protocols/datagram" }
 libp2p-stream = { version = "0.5.0-alpha", path = "protocols/stream" }
 libp2p-swarm = { version = "0.48.0", path = "swarm" }
 libp2p-swarm-derive = { version = "=0.36.0", path = "swarm-derive" } # `libp2p-swarm-derive` may not be compatible with different `libp2p-swarm` non-breaking releases. E.g. `libp2p-swarm` might introduce a new enum variant `FromSwarm` (which is `#[non-exhaustive]`) in a non-breaking release. Older versions of `libp2p-swarm-derive` would not forward this enum variant within the `NetworkBehaviour` hierarchy. Thus the version pinning is required.

core/CHANGELOG.md

@@ -1,5 +1,8 @@
 ## 0.44.0
 
+- Add unreliable datagram support to `StreamMuxer`: `send_datagram`, `max_datagram_size`, and `StreamMuxerEvent::Datagram`. Unsupported muxers return `SendDatagramError::Unsupported`.
+  See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489).
+
 - Raise MSRV to 1.88.0.
   See [PR 6273](https://github.com/libp2p/rust-libp2p/pull/6273).
 

core/Cargo.toml

@@ -11,6 +11,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"]
 categories = ["network-programming", "asynchronous"]
 
 [dependencies]
+bytes = { workspace = true }
 either = "1.16"
 fnv = "1.0"
 futures = { workspace = true, features = ["executor", "thread-pool"] }

core/src/muxing.rs

@@ -52,6 +52,7 @@
 
 use std::{future::Future, pin::Pin};
 
+use bytes::Bytes;
 use futures::{
     AsyncRead, AsyncWrite,
     task::{Context, Poll},
@@ -111,13 +112,38 @@ pub trait StreamMuxer {
         self: Pin<&mut Self>,
         cx: &mut Context<'_>,
     ) -> Poll<Result<StreamMuxerEvent, Self::Error>>;
+
+    /// Send an unreliable datagram. Inbound arrive via [`StreamMuxerEvent::Datagram`].
+    fn send_datagram(self: Pin<&mut Self>, data: Bytes) -> Result<(), SendDatagramError> {
+        let _ = data;
+        Err(SendDatagramError::Unsupported)
+    }
+
+    /// Largest [`StreamMuxer::send_datagram`] payload, or `None` if unsupported.
+    fn max_datagram_size(&self) -> Option<usize> {
+        None
+    }
 }
 
 /// An event produced by a [`StreamMuxer`].
 #[derive(Debug)]
 pub enum StreamMuxerEvent {
     /// The address of the remote has changed.
     AddressChange(Multiaddr),
+    /// An unreliable datagram was received from the remote.
+    Datagram(Bytes),
+}
+
+/// Error returned by [`StreamMuxer::send_datagram`].
+#[derive(Debug, thiserror::Error)]
+#[non_exhaustive]
+pub enum SendDatagramError {
+    #[error("datagrams are not supported by this muxer")]
+    Unsupported,
+    #[error("datagram of {size} bytes exceeds the {max} byte limit")]
+    TooLarge { size: usize, max: usize },
+    #[error("connection closed")]
+    ConnectionClosed,
 }
 
 /// Extension trait for [`StreamMuxer`].
@@ -164,6 +190,15 @@ pub trait StreamMuxerExt: StreamMuxer + Sized {
         Pin::new(self).poll_close(cx)
     }
 
+    /// Convenience function for calling [`StreamMuxer::send_datagram`]
+    /// for [`StreamMuxer`]s that are `Unpin`.
+    fn send_datagram_unpin(&mut self, data: Bytes) -> Result<(), SendDatagramError>
+    where
+        Self: Unpin,
+    {
+        Pin::new(self).send_datagram(data)
+    }
+
     /// Returns a future for closing this [`StreamMuxer`].
     fn close(self) -> Close<Self> {
         Close(self)

core/src/muxing/boxed.rs

@@ -6,10 +6,11 @@ use std::{
     task::{Context, Poll},
 };
 
+use bytes::Bytes;
 use futures::{AsyncRead, AsyncWrite};
 use pin_project::pin_project;
 
-use crate::muxing::{StreamMuxer, StreamMuxerEvent};
+use crate::muxing::{SendDatagramError, StreamMuxer, StreamMuxerEvent};
 
 /// Abstract `StreamMuxer`.
 pub struct StreamMuxerBox {
@@ -79,6 +80,14 @@ where
     ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
         self.project().inner.poll(cx).map_err(into_io_error)
     }
+
+    fn send_datagram(self: Pin<&mut Self>, data: Bytes) -> Result<(), SendDatagramError> {
+        self.project().inner.send_datagram(data)
+    }
+
+    fn max_datagram_size(&self) -> Option<usize> {
+        self.inner.max_datagram_size()
+    }
 }
 
 fn into_io_error<E>(err: E) -> io::Error
@@ -139,6 +148,14 @@ impl StreamMuxer for StreamMuxerBox {
     ) -> Poll<Result<StreamMuxerEvent, Self::Error>> {
         self.project().poll(cx)
     }
+
+    fn send_datagram(self: Pin<&mut Self>, data: Bytes) -> Result<(), SendDatagramError> {
+        self.project().send_datagram(data)
+    }
+
+    fn max_datagram_size(&self) -> Option<usize> {
+        self.inner.as_ref().get_ref().max_datagram_size()
+    }
 }
 
 impl SubstreamBox {

protocols/datagram/CHANGELOG.md

@@ -0,0 +1,5 @@
+## 0.1.0
+
+- Initial release: a `Behaviour` and `Control` for sending and receiving
+  unreliable datagrams over libp2p connections (QUIC only).
+  See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489).

protocols/datagram/Cargo.toml

@@ -0,0 +1,28 @@
+[package]
+name = "libp2p-datagram"
+version = "0.1.0"
+edition.workspace = true
+rust-version.workspace = true
+description = "Unreliable datagram protocol for libp2p"
+license = "MIT"
+repository = "https://github.com/libp2p/rust-libp2p"
+keywords = ["peer-to-peer", "libp2p", "networking"]
+categories = ["network-programming", "asynchronous"]
+
+[dependencies]
+bytes = { workspace = true }
+futures = { workspace = true }
+libp2p-core = { workspace = true }
+libp2p-identity = { workspace = true, features = ["peerid"] }
+libp2p-swarm = { workspace = true }
+thiserror = { workspace = true }
+
+[dev-dependencies]
+libp2p-identity = { workspace = true, features = ["ed25519", "rand"] }
+libp2p-quic = { workspace = true, features = ["tokio"] }
+libp2p-swarm = { workspace = true, features = ["tokio"] }
+tokio = { workspace = true, features = ["full"] }
+tracing-subscriber = { workspace = true, features = ["env-filter"] }
+
+[lints]
+workspace = true

protocols/datagram/README.md

@@ -0,0 +1,23 @@
+# libp2p-datagram
+
+Unreliable datagrams over libp2p connections.
+
+Datagrams may be dropped, reordered, or duplicated and carry no flow control;
+the caller owns reliability. Only QUIC carries them today; sends on other
+transports fail with `SendError`.
+
+```rust,no_run
+use libp2p_datagram as datagram;
+
+let mut behaviour = datagram::Behaviour::new();
+let mut control = behaviour.new_control();
+let mut incoming = behaviour.incoming_datagrams().unwrap();
+
+// add `behaviour` to your Swarm, then:
+// control.send_datagram(peer, bytes)?;
+// while let Some((from, bytes)) = incoming.next().await { ... }
+```
+
+## License
+
+Licensed under MIT.

protocols/datagram/src/behaviour.rs

@@ -0,0 +1,125 @@
+use std::{
+    pin::Pin,
+    task::{Context, Poll},
+};
+
+use bytes::Bytes;
+use futures::{Stream, StreamExt, channel::mpsc};
+use libp2p_core::{Endpoint, Multiaddr, transport::PortUse};
+use libp2p_identity::PeerId;
+use libp2p_swarm::{
+    ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, NotifyHandler, THandler,
+    THandlerInEvent, THandlerOutEvent, ToSwarm,
+};
+
+use crate::{Control, handler::Handler};
+
+const CHANNEL_BUFFER: usize = 256;
+
+/// Sends and receives unreliable datagrams.
+pub struct Behaviour {
+    outbound_tx: mpsc::Sender<OutboundDatagram>,
+    outbound_rx: mpsc::Receiver<OutboundDatagram>,
+    inbound_tx: mpsc::Sender<(PeerId, Bytes)>,
+    incoming: Option<mpsc::Receiver<(PeerId, Bytes)>>,
+}
+
+pub(crate) struct OutboundDatagram {
+    pub(crate) peer: PeerId,
+    pub(crate) connection: Option<ConnectionId>,
+    pub(crate) data: Bytes,
+}
+
+impl Default for Behaviour {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl Behaviour {
+    pub fn new() -> Self {
+        let (outbound_tx, outbound_rx) = mpsc::channel(CHANNEL_BUFFER);
+        let (inbound_tx, incoming) = mpsc::channel(CHANNEL_BUFFER);
+        Self {
+            outbound_tx,
+            outbound_rx,
+            inbound_tx,
+            incoming: Some(incoming),
+        }
+    }
+
+    /// A new [`Control`] for sending datagrams.
+    pub fn new_control(&self) -> Control {
+        Control::new(self.outbound_tx.clone())
+    }
+
+    /// Stream of inbound `(sender, payload)`. `None` after the first call.
+    pub fn incoming_datagrams(&mut self) -> Option<IncomingDatagrams> {
+        self.incoming.take().map(IncomingDatagrams)
+    }
+}
+
+/// Stream of inbound datagrams tagged with the sending [`PeerId`].
+pub struct IncomingDatagrams(mpsc::Receiver<(PeerId, Bytes)>);
+
+impl Stream for IncomingDatagrams {
+    type Item = (PeerId, Bytes);
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        self.0.poll_next_unpin(cx)
+    }
+}
+
+impl NetworkBehaviour for Behaviour {
+    type ConnectionHandler = Handler;
+    type ToSwarm = ();
+
+    fn handle_established_inbound_connection(
+        &mut self,
+        _: ConnectionId,
+        peer: PeerId,
+        _: &Multiaddr,
+        _: &Multiaddr,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(peer, self.inbound_tx.clone()))
+    }
+
+    fn handle_established_outbound_connection(
+        &mut self,
+        _: ConnectionId,
+        peer: PeerId,
+        _: &Multiaddr,
+        _: Endpoint,
+        _: PortUse,
+    ) -> Result<THandler<Self>, ConnectionDenied> {
+        Ok(Handler::new(peer, self.inbound_tx.clone()))
+    }
+
+    fn on_swarm_event(&mut self, _event: FromSwarm) {}
+
+    fn on_connection_handler_event(
+        &mut self,
+        _: PeerId,
+        _: 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(out)) = self.outbound_rx.poll_next_unpin(cx) {
+            return Poll::Ready(ToSwarm::NotifyHandler {
+                peer_id: out.peer,
+                handler: match out.connection {
+                    Some(id) => NotifyHandler::One(id),
+                    None => NotifyHandler::Any,
+                },
+                event: out.data,
+            });
+        }
+        Poll::Pending
+    }
+}