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,11 @@
 ## 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).
+
+- Add `StreamMuxer::substream_id` and `SubstreamBox::transport_stream_id`, surfacing transport stream ids (QUIC) for datagram flows. `SubstreamBox::new` now takes the id.
+  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/either.rs

@@ -89,6 +89,13 @@ where
             future::Either::Right(inner) => inner.poll(cx).map_err(Either::Right),
         }
     }
+
+    fn substream_id(substream: &Self::Substream) -> Option<u64> {
+        match substream {
+            future::Either::Left(s) => A::substream_id(s),
+            future::Either::Right(s) => B::substream_id(s),
+        }
+    }
 }
 
 /// Implements `Future` and dispatches all method calls to either `First` or `Second`.

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,51 @@ 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
+    }
+
+    /// Transport-assigned stream id, where one exists (QUIC, WebTransport).
+    ///
+    /// Keys a datagram flow to its control stream, see [libp2p/specs#680].
+    ///
+    /// [libp2p/specs#680]: https://github.com/libp2p/specs/pull/680
+    fn substream_id(substream: &Self::Substream) -> Option<u64>
+    where
+        Self: Sized,
+    {
+        let _ = substream;
+        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 +203,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 {
@@ -26,7 +27,7 @@ impl fmt::Debug for StreamMuxerBox {
 ///
 /// A [`SubstreamBox`] erases the concrete type it is given and only retains its `AsyncRead`
 /// and `AsyncWrite` capabilities.
-pub struct SubstreamBox(Pin<Box<dyn AsyncReadWrite + Send>>);
+pub struct SubstreamBox(Pin<Box<dyn AsyncReadWrite + Send>>, Option<u64>);
 
 #[pin_project]
 struct Wrap<T>
@@ -53,7 +54,7 @@ where
         self.project()
             .inner
             .poll_inbound(cx)
-            .map_ok(SubstreamBox::new)
+            .map_ok(|s| SubstreamBox::new(T::substream_id(&s), s))
             .map_err(into_io_error)
     }
 
@@ -64,7 +65,7 @@ where
         self.project()
             .inner
             .poll_outbound(cx)
-            .map_ok(SubstreamBox::new)
+            .map_ok(|s| SubstreamBox::new(T::substream_id(&s), s))
             .map_err(into_io_error)
     }
 
@@ -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,13 +148,30 @@ 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()
+    }
+
+    fn substream_id(substream: &Self::Substream) -> Option<u64> {
+        substream.transport_stream_id()
+    }
 }
 
 impl SubstreamBox {
     /// Construct a new [`SubstreamBox`] from something
     /// that implements [`AsyncRead`] and [`AsyncWrite`].
-    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(stream: S) -> Self {
-        Self(Box::pin(stream))
+    pub fn new<S: AsyncRead + AsyncWrite + Send + 'static>(id: Option<u64>, stream: S) -> Self {
+        Self(Box::pin(stream), id)
+    }
+
+    /// Transport-assigned stream id, see [`StreamMuxer::substream_id`].
+    pub fn transport_stream_id(&self) -> Option<u64> {
+        self.1
     }
 }
 

misc/metrics/CHANGELOG.md

@@ -1,5 +1,8 @@
 ## 0.18.0
 
+- Forward `StreamMuxer::substream_id` through the bandwidth-metering muxer.
+  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).
 

misc/metrics/src/bandwidth.rs

@@ -224,6 +224,10 @@ where
         let this = self.project();
         this.inner.poll_close(cx)
     }
+
+    fn substream_id(substream: &Self::Substream) -> Option<u64> {
+        SMInner::substream_id(&substream.inner)
+    }
 }
 
 /// Wraps around an [`AsyncRead`] + [`AsyncWrite`] and logs the bandwidth that goes through it.

misc/multistream-select/CHANGELOG.md

@@ -1,5 +1,8 @@
 ## 0.14.0
 
+- Add `Negotiated::as_inner` to access the underlying stream after negotiation.
+  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).
 

misc/multistream-select/src/negotiated.rs

@@ -111,6 +111,14 @@ impl<TInner> Negotiated<TInner> {
         }
     }
 
+    /// Reference to the underlying I/O stream, once negotiation has completed.
+    pub fn as_inner(&self) -> Option<&TInner> {
+        match &self.state {
+            State::Completed { io } => Some(io),
+            _ => None,
+        }
+    }
+
     /// Polls the `Negotiated` for completion.
     fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), NegotiationError>>
     where

protocols/datagram/CHANGELOG.md

@@ -0,0 +1,9 @@
+## 0.1.0
+
+- Initial release: a `Behaviour` and `Control` for sending and receiving
+  unreliable datagrams over libp2p connections (QUIC only), implementing the
+  `/dg/1` control stream and framing from [libp2p/specs#680]. `Behaviour::new`
+  takes the application protocol the datagrams belong to.
+  See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489).
+
+[libp2p/specs#680]: https://github.com/libp2p/specs/pull/680

protocols/datagram/Cargo.toml

@@ -0,0 +1,29 @@
+[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 }
+unsigned-varint = { workspace = true, features = ["futures"] }
+
+[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,27 @@
+# libp2p-datagram
+
+Unreliable datagrams over libp2p connections, per [libp2p/specs#680].
+
+Datagrams ride a QUIC `/dg/1` control stream that binds them to one application
+protocol. They 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;
+use libp2p_swarm::StreamProtocol;
+
+let mut behaviour = datagram::Behaviour::new(StreamProtocol::new("/my/app/1.0.0"));
+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.
+
+[libp2p/specs#680]: https://github.com/libp2p/specs/pull/680