diff --git a/Cargo.lock b/Cargo.lock index cba67ebeafc..d69562c5424 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2963,6 +2963,7 @@ dependencies = [ name = "libp2p-core" version = "0.44.0" dependencies = [ + "bytes", "either", "fnv", "futures", @@ -2985,6 +2986,22 @@ dependencies = [ "web-time", ] +[[package]] +name = "libp2p-datagram" +version = "0.1.0" +dependencies = [ + "bytes", + "futures", + "libp2p-core", + "libp2p-identity", + "libp2p-quic", + "libp2p-swarm", + "thiserror 2.0.18", + "tokio", + "tracing-subscriber", + "unsigned-varint", +] + [[package]] name = "libp2p-dcutr" version = "0.15.0" @@ -3376,6 +3393,7 @@ dependencies = [ name = "libp2p-quic" version = "0.14.0" dependencies = [ + "bytes", "futures", "futures-timer", "if-watch", @@ -3510,6 +3528,7 @@ dependencies = [ name = "libp2p-swarm" version = "0.48.0" dependencies = [ + "bytes", "criterion 0.5.1", "either", "fnv", @@ -6928,6 +6947,8 @@ checksum = "eb066959b24b5196ae73cb057f45598450d2c5f71460e98c49b738086eff9c06" dependencies = [ "asynchronous-codec", "bytes", + "futures-io", + "futures-util", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 8846990fb65..f778983c0e8 100644 --- a/Cargo.toml +++ b/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. diff --git a/core/CHANGELOG.md b/core/CHANGELOG.md index be980183c3c..5c9a871ffdc 100644 --- a/core/CHANGELOG.md +++ b/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). diff --git a/core/Cargo.toml b/core/Cargo.toml index fc047a15ca3..f2ef12411b2 100644 --- a/core/Cargo.toml +++ b/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"] } diff --git a/core/src/either.rs b/core/src/either.rs index 7dc9d92d875..482bf1196db 100644 --- a/core/src/either.rs +++ b/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 { + 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`. diff --git a/core/src/muxing.rs b/core/src/muxing.rs index c749436764b..783ece9c8ae 100644 --- a/core/src/muxing.rs +++ b/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,6 +112,30 @@ pub trait StreamMuxer { self: Pin<&mut Self>, cx: &mut Context<'_>, ) -> Poll>; + + /// 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 { + 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 + where + Self: Sized, + { + let _ = substream; + None + } } /// An event produced by a [`StreamMuxer`]. @@ -118,6 +143,20 @@ pub trait StreamMuxer { 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 { Close(self) diff --git a/core/src/muxing/boxed.rs b/core/src/muxing/boxed.rs index 12c3036cd98..24e713d7d54 100644 --- a/core/src/muxing/boxed.rs +++ b/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>); +pub struct SubstreamBox(Pin>, Option); #[pin_project] struct Wrap @@ -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> { 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 { + self.inner.max_datagram_size() + } } fn into_io_error(err: E) -> io::Error @@ -139,13 +148,30 @@ impl StreamMuxer for StreamMuxerBox { ) -> Poll> { 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 { + self.inner.as_ref().get_ref().max_datagram_size() + } + + fn substream_id(substream: &Self::Substream) -> Option { + substream.transport_stream_id() + } } impl SubstreamBox { /// Construct a new [`SubstreamBox`] from something /// that implements [`AsyncRead`] and [`AsyncWrite`]. - pub fn new(stream: S) -> Self { - Self(Box::pin(stream)) + pub fn new(id: Option, stream: S) -> Self { + Self(Box::pin(stream), id) + } + + /// Transport-assigned stream id, see [`StreamMuxer::substream_id`]. + pub fn transport_stream_id(&self) -> Option { + self.1 } } diff --git a/misc/metrics/CHANGELOG.md b/misc/metrics/CHANGELOG.md index 083ab5b448b..310b7675ba2 100644 --- a/misc/metrics/CHANGELOG.md +++ b/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). diff --git a/misc/metrics/src/bandwidth.rs b/misc/metrics/src/bandwidth.rs index edbfaf4b60f..b2d71792f15 100644 --- a/misc/metrics/src/bandwidth.rs +++ b/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 { + SMInner::substream_id(&substream.inner) + } } /// Wraps around an [`AsyncRead`] + [`AsyncWrite`] and logs the bandwidth that goes through it. diff --git a/misc/multistream-select/CHANGELOG.md b/misc/multistream-select/CHANGELOG.md index af5fc465a6e..4dc527d0454 100644 --- a/misc/multistream-select/CHANGELOG.md +++ b/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). diff --git a/misc/multistream-select/src/negotiated.rs b/misc/multistream-select/src/negotiated.rs index e503144a5fc..8b9318c4f62 100644 --- a/misc/multistream-select/src/negotiated.rs +++ b/misc/multistream-select/src/negotiated.rs @@ -111,6 +111,14 @@ impl Negotiated { } } + /// 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> where diff --git a/protocols/datagram/CHANGELOG.md b/protocols/datagram/CHANGELOG.md new file mode 100644 index 00000000000..ea22137aad4 --- /dev/null +++ b/protocols/datagram/CHANGELOG.md @@ -0,0 +1,13 @@ +## 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). + +- Add `Control::max_datagram_size`, the per-connection max outbound datagram size + learned from the send path, so senders can size fragments to the path budget. + See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489). + +[libp2p/specs#680]: https://github.com/libp2p/specs/pull/680 diff --git a/protocols/datagram/Cargo.toml b/protocols/datagram/Cargo.toml new file mode 100644 index 00000000000..928d6db460a --- /dev/null +++ b/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", "macros"] } +tokio = { workspace = true, features = ["full"] } +tracing-subscriber = { workspace = true, features = ["env-filter"] } + +[lints] +workspace = true diff --git a/protocols/datagram/README.md b/protocols/datagram/README.md new file mode 100644 index 00000000000..653a262062f --- /dev/null +++ b/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 diff --git a/protocols/datagram/src/behaviour.rs b/protocols/datagram/src/behaviour.rs new file mode 100644 index 00000000000..2b041fbede2 --- /dev/null +++ b/protocols/datagram/src/behaviour.rs @@ -0,0 +1,148 @@ +use std::{ + collections::HashMap, + pin::Pin, + sync::{Arc, Mutex}, + 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::{ + ConnectionClosed, ConnectionDenied, ConnectionId, FromSwarm, NetworkBehaviour, NotifyHandler, + StreamProtocol, THandler, THandlerInEvent, THandlerOutEvent, ToSwarm, +}; + +use crate::{Control, handler::Handler}; + +const CHANNEL_BUFFER: usize = 256; + +/// Per-connection max outbound datagram size, shared with every [`Control`]. +pub(crate) type DatagramSizes = Arc>>; + +/// Sends and receives unreliable datagrams for a single application protocol. +pub struct Behaviour { + protocol: StreamProtocol, + outbound_tx: mpsc::Sender, + outbound_rx: mpsc::Receiver, + inbound_tx: mpsc::Sender<(PeerId, Bytes)>, + incoming: Option>, + sizes: DatagramSizes, +} + +pub(crate) struct OutboundDatagram { + pub(crate) peer: PeerId, + pub(crate) connection: Option, + pub(crate) data: Bytes, +} + +impl Behaviour { + /// Datagrams negotiated under `protocol` on the `/dg/1` control stream. + pub fn new(protocol: StreamProtocol) -> Self { + let (outbound_tx, outbound_rx) = mpsc::channel(CHANNEL_BUFFER); + let (inbound_tx, incoming) = mpsc::channel(CHANNEL_BUFFER); + Self { + protocol, + outbound_tx, + outbound_rx, + inbound_tx, + incoming: Some(incoming), + sizes: Arc::new(Mutex::new(HashMap::new())), + } + } + + /// A new [`Control`] for sending datagrams. + pub fn new_control(&self) -> Control { + Control::new(self.outbound_tx.clone(), self.sizes.clone()) + } + + /// Stream of inbound `(sender, payload)`. `None` after the first call. + pub fn incoming_datagrams(&mut self) -> Option { + 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> { + 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, ConnectionDenied> { + Ok(Handler::new( + peer, + Endpoint::Listener, + self.protocol.clone(), + self.inbound_tx.clone(), + )) + } + + fn handle_established_outbound_connection( + &mut self, + _: ConnectionId, + peer: PeerId, + _: &Multiaddr, + _: Endpoint, + _: PortUse, + ) -> Result, ConnectionDenied> { + Ok(Handler::new( + peer, + Endpoint::Dialer, + self.protocol.clone(), + self.inbound_tx.clone(), + )) + } + + fn on_swarm_event(&mut self, event: FromSwarm) { + if let FromSwarm::ConnectionClosed(ConnectionClosed { + peer_id, + connection_id, + .. + }) = event + { + self.sizes.lock().unwrap().remove(&(peer_id, connection_id)); + } + } + + fn on_connection_handler_event( + &mut self, + peer: PeerId, + connection: ConnectionId, + max: THandlerOutEvent, + ) { + self.sizes.lock().unwrap().insert((peer, connection), max); + } + + fn poll( + &mut self, + cx: &mut Context<'_>, + ) -> Poll>> { + 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 + } +} diff --git a/protocols/datagram/src/control.rs b/protocols/datagram/src/control.rs new file mode 100644 index 00000000000..29872283298 --- /dev/null +++ b/protocols/datagram/src/control.rs @@ -0,0 +1,69 @@ +use bytes::Bytes; +use futures::channel::mpsc; +use libp2p_identity::PeerId; +use libp2p_swarm::ConnectionId; + +use crate::behaviour::{DatagramSizes, OutboundDatagram}; + +/// Cloneable handle for sending datagrams. +#[derive(Clone)] +pub struct Control { + sender: mpsc::Sender, + sizes: DatagramSizes, +} + +impl Control { + pub(crate) fn new(sender: mpsc::Sender, sizes: DatagramSizes) -> Self { + Self { sender, sizes } + } + + /// The connection's current max outbound datagram size, learned from sends + /// on it. `None` until the first datagram has been sent. + pub fn max_datagram_size(&self, peer: PeerId, connection: ConnectionId) -> Option { + self.sizes.lock().unwrap().get(&(peer, connection)).copied() + } + + /// Send to `peer` over whichever connection the swarm picks. + pub fn send_datagram(&mut self, peer: PeerId, data: Bytes) -> Result<(), SendError> { + self.enqueue(peer, None, data) + } + + /// Send pinned to a specific `connection`. + pub fn send_datagram_on_connection( + &mut self, + peer: PeerId, + connection: ConnectionId, + data: Bytes, + ) -> Result<(), SendError> { + self.enqueue(peer, Some(connection), data) + } + + fn enqueue( + &mut self, + peer: PeerId, + connection: Option, + data: Bytes, + ) -> Result<(), SendError> { + self.sender + .try_send(OutboundDatagram { + peer, + connection, + data, + }) + .map_err(|e| { + if e.is_full() { + SendError::Full + } else { + SendError::Closed + } + }) + } +} + +#[derive(Debug, thiserror::Error)] +pub enum SendError { + #[error("datagram queue is full; datagram dropped")] + Full, + #[error("datagram behaviour has shut down")] + Closed, +} diff --git a/protocols/datagram/src/framing.rs b/protocols/datagram/src/framing.rs new file mode 100644 index 00000000000..e8bb1819100 --- /dev/null +++ b/protocols/datagram/src/framing.rs @@ -0,0 +1,85 @@ +//! Datagram wire format ([libp2p/specs#680]): QUIC varint control-stream id, +//! then the payload. +//! +//! [libp2p/specs#680]: https://github.com/libp2p/specs/pull/680 + +use bytes::{BufMut, Bytes, BytesMut}; + +/// Frame `payload` for the flow keyed by `control_stream_id`. +pub(crate) fn frame(control_stream_id: u64, payload: &[u8]) -> Bytes { + let mut buf = BytesMut::with_capacity(8 + payload.len()); + put_varint(control_stream_id, &mut buf); + buf.put_slice(payload); + buf.freeze() +} + +/// Split a datagram into its control-stream id and payload. Test-only; inbound +/// decoding lives in the connection layer, this just validates `frame`. +#[cfg(test)] +pub(crate) fn parse(datagram: &[u8]) -> Option<(u64, Bytes)> { + let (id, len) = get_varint(datagram)?; + Some((id, Bytes::copy_from_slice(&datagram[len..]))) +} + +/// QUIC varint encoding (RFC 9000 section 16). +fn put_varint(value: u64, out: &mut BytesMut) { + if value < (1 << 6) { + out.put_u8(value as u8); + } else if value < (1 << 14) { + out.put_u16(value as u16 | (0b01 << 14)); + } else if value < (1 << 30) { + out.put_u32(value as u32 | (0b10 << 30)); + } else { + debug_assert!(value < (1 << 62), "stream ids fit in 62 bits"); + out.put_u64(value | (0b11 << 62)); + } +} + +/// Returns the decoded value and the number of bytes it consumed. +#[cfg(test)] +fn get_varint(buf: &[u8]) -> Option<(u64, usize)> { + let first = *buf.first()?; + let len = 1usize << (first >> 6); + if buf.len() < len { + return None; + } + let mut value = u64::from(first & 0x3f); + for &b in &buf[1..len] { + value = (value << 8) | u64::from(b); + } + Some((value, len)) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn varint_roundtrip_across_lengths() { + for value in [0, 63, 64, 16_383, 16_384, 1 << 29, 1 << 30, (1 << 62) - 1] { + let mut buf = BytesMut::new(); + put_varint(value, &mut buf); + assert_eq!(get_varint(&buf), Some((value, buf.len()))); + } + } + + #[test] + fn frame_then_parse_preserves_payload() { + let framed = frame(4611686018427387903, b"payload"); + assert_eq!( + parse(&framed), + Some((4611686018427387903, Bytes::from_static(b"payload"))) + ); + } + + #[test] + fn empty_payload_is_valid() { + assert_eq!(parse(&frame(7, b"")), Some((7, Bytes::new()))); + } + + #[test] + fn truncated_id_is_rejected() { + assert_eq!(parse(&[]), None); + assert_eq!(parse(&[0b1100_0000]), None); + } +} diff --git a/protocols/datagram/src/handler.rs b/protocols/datagram/src/handler.rs new file mode 100644 index 00000000000..0a6eab2c502 --- /dev/null +++ b/protocols/datagram/src/handler.rs @@ -0,0 +1,145 @@ +use std::{ + collections::VecDeque, + task::{Context, Poll}, +}; + +use bytes::Bytes; +use futures::channel::mpsc; +use libp2p_core::Endpoint; +use libp2p_identity::PeerId; +use libp2p_swarm::{ + ConnectionHandler, ConnectionHandlerEvent, Stream, StreamProtocol, SubstreamProtocol, + handler::{ConnectionEvent, FullyNegotiatedInbound, FullyNegotiatedOutbound}, +}; + +use crate::{ + framing, + protocol::{CONTROL_PROTOCOL, Upgrade}, +}; + +/// Pre-establish backlog cap; head-drop oldest past it (delivery is unreliable). +const MAX_OUTBOUND_BACKLOG: usize = 256; + +/// Drives one datagram flow: the dialer opens the `/dg/1` control stream, both +/// ends learn its stream id, and datagrams are framed and filtered by it. +pub struct Handler { + remote: PeerId, + role: Endpoint, + protocol: StreamProtocol, + inbound: mpsc::Sender<(PeerId, Bytes)>, + requested_control_stream: bool, + control_stream_id: Option, + // Held open for the life of the flow, per spec; never read or written again. + control_stream: Option, + outbound: VecDeque, + reported_max: Option, + pending_max: Option, +} + +impl Handler { + pub(crate) fn new( + remote: PeerId, + role: Endpoint, + protocol: StreamProtocol, + inbound: mpsc::Sender<(PeerId, Bytes)>, + ) -> Self { + Self { + remote, + role, + protocol, + inbound, + requested_control_stream: false, + control_stream_id: None, + control_stream: None, + outbound: VecDeque::new(), + reported_max: None, + pending_max: None, + } + } + + fn upgrade(&self) -> SubstreamProtocol { + SubstreamProtocol::new( + Upgrade { + application_protocol: self.protocol.clone(), + }, + (), + ) + } + + fn establish(&mut self, stream: Stream) { + self.control_stream_id = stream.transport_stream_id(); + self.control_stream = Some(stream); + } +} + +impl ConnectionHandler for Handler { + type FromBehaviour = Bytes; + type ToBehaviour = usize; + type InboundProtocol = Upgrade; + type OutboundProtocol = Upgrade; + type InboundOpenInfo = (); + type OutboundOpenInfo = (); + + fn listen_protocol(&self) -> SubstreamProtocol { + self.upgrade() + } + + fn supports_datagrams(&self, protocol: &StreamProtocol) -> bool { + *protocol == CONTROL_PROTOCOL + } + + fn poll(&mut self, _: &mut Context<'_>) -> Poll> { + if let Some(max) = self.pending_max.take() { + return Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(max)); + } + + if self.role == Endpoint::Dialer && !self.requested_control_stream { + self.requested_control_stream = true; + return Poll::Ready(ConnectionHandlerEvent::OutboundSubstreamRequest { + protocol: self.upgrade(), + }); + } + + if let Some(id) = self.control_stream_id + && let Some(payload) = self.outbound.pop_front() + { + return Poll::Ready(ConnectionHandlerEvent::SendDatagram(framing::frame( + id, &payload, + ))); + } + + Poll::Pending + } + + fn on_behaviour_event(&mut self, data: Bytes) { + if self.outbound.len() >= MAX_OUTBOUND_BACKLOG { + self.outbound.pop_front(); + } + self.outbound.push_back(data); + } + + fn on_connection_event( + &mut self, + event: ConnectionEvent, + ) { + match event { + ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { + protocol: stream, + .. + }) + | ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { + protocol: stream, + .. + }) => self.establish(stream), + ConnectionEvent::Datagram(datagram) => { + // Already parsed and routed to us by the connection, id stripped. + let _ = self.inbound.try_send((self.remote, datagram.data.clone())); + } + ConnectionEvent::DatagramMaxSize(max) if self.reported_max != Some(max) => { + self.reported_max = Some(max); + self.pending_max = Some(max); + } + _ => {} + } + } +} diff --git a/protocols/datagram/src/lib.rs b/protocols/datagram/src/lib.rs new file mode 100644 index 00000000000..e64a6d62898 --- /dev/null +++ b/protocols/datagram/src/lib.rs @@ -0,0 +1,16 @@ +//! Unreliable datagrams over libp2p connections, per [libp2p/specs#680]. +//! +//! Datagrams ride a QUIC `/dg/1` control stream that pins them to one +//! application protocol; they may be dropped, reordered, or duplicated, and the +//! caller owns reliability. Only QUIC carries them today; other transports drop. +//! +//! [libp2p/specs#680]: https://github.com/libp2p/specs/pull/680 + +mod behaviour; +mod control; +mod framing; +mod handler; +mod protocol; + +pub use behaviour::{Behaviour, IncomingDatagrams}; +pub use control::{Control, SendError}; diff --git a/protocols/datagram/src/protocol.rs b/protocols/datagram/src/protocol.rs new file mode 100644 index 00000000000..4fe1e94bd54 --- /dev/null +++ b/protocols/datagram/src/protocol.rs @@ -0,0 +1,78 @@ +//! The `/dg/1` control stream ([libp2p/specs#680]): the initiator sends the +//! length-prefixed application protocol id, then it stays open for the flow. +//! +//! [libp2p/specs#680]: https://github.com/libp2p/specs/pull/680 + +use std::{io, iter}; + +use futures::{AsyncReadExt, AsyncWriteExt, future::BoxFuture, prelude::*}; +use libp2p_core::upgrade::{InboundUpgrade, OutboundUpgrade, UpgradeInfo}; +use libp2p_swarm::{Stream, StreamProtocol}; + +pub(crate) const CONTROL_PROTOCOL: StreamProtocol = StreamProtocol::new("/dg/1"); + +const MAX_PROTOCOL_LEN: usize = 256; + +/// Control-stream upgrade. Public only to satisfy the connection-handler signature. +pub struct Upgrade { + pub(crate) application_protocol: StreamProtocol, +} + +impl UpgradeInfo for Upgrade { + type Info = StreamProtocol; + type InfoIter = iter::Once; + + fn protocol_info(&self) -> Self::InfoIter { + iter::once(CONTROL_PROTOCOL) + } +} + +impl OutboundUpgrade for Upgrade { + type Output = Stream; + type Error = io::Error; + type Future = BoxFuture<'static, Result>; + + fn upgrade_outbound(self, mut stream: Stream, _: StreamProtocol) -> Self::Future { + async move { + let id = self.application_protocol.as_ref().as_bytes(); + let mut len = unsigned_varint::encode::usize_buffer(); + stream + .write_all(unsigned_varint::encode::usize(id.len(), &mut len)) + .await?; + stream.write_all(id).await?; + stream.flush().await?; + Ok(stream) + } + .boxed() + } +} + +impl InboundUpgrade for Upgrade { + type Output = Stream; + type Error = io::Error; + type Future = BoxFuture<'static, Result>; + + fn upgrade_inbound(self, mut stream: Stream, _: StreamProtocol) -> Self::Future { + async move { + let len = unsigned_varint::aio::read_usize(&mut stream) + .await + .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; + if len > MAX_PROTOCOL_LEN { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "protocol id too long", + )); + } + let mut id = vec![0u8; len]; + stream.read_exact(&mut id).await?; + if id != self.application_protocol.as_ref().as_bytes() { + return Err(io::Error::new( + io::ErrorKind::InvalidData, + "unexpected application protocol", + )); + } + Ok(stream) + } + .boxed() + } +} diff --git a/protocols/datagram/tests/datagram.rs b/protocols/datagram/tests/datagram.rs new file mode 100644 index 00000000000..c7d8fc0b771 --- /dev/null +++ b/protocols/datagram/tests/datagram.rs @@ -0,0 +1,170 @@ +use std::time::Duration; + +use bytes::Bytes; +use futures::StreamExt as _; +use libp2p_core::{Transport as _, muxing::StreamMuxerBox}; +use libp2p_datagram as datagram; +use libp2p_identity::Keypair; +use libp2p_swarm::{Config, NetworkBehaviour, StreamProtocol, Swarm, SwarmEvent}; + +fn quic_swarm() -> Swarm { + let keypair = Keypair::generate_ed25519(); + let peer_id = keypair.public().to_peer_id(); + let transport = libp2p_quic::tokio::Transport::new(libp2p_quic::Config::new(&keypair)) + .map(|(p, c), _| (p, StreamMuxerBox::new(c))) + .boxed(); + Swarm::new( + transport, + datagram::Behaviour::new(StreamProtocol::new("/example/datagram/1.0.0")), + peer_id, + Config::with_tokio_executor().with_idle_connection_timeout(Duration::from_secs(10)), + ) +} + +#[tokio::test] +async fn datagram_roundtrip_over_quic() { + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); + + let mut listener = quic_swarm(); + let mut dialer = quic_swarm(); + + let listener_peer = *listener.local_peer_id(); + let dialer_peer = *dialer.local_peer_id(); + + let mut incoming = listener.behaviour_mut().incoming_datagrams().unwrap(); + let mut control = dialer.behaviour().new_control(); + + listener + .listen_on("/ip4/127.0.0.1/udp/0/quic-v1".parse().unwrap()) + .unwrap(); + let addr = loop { + if let SwarmEvent::NewListenAddr { address, .. } = listener.select_next_some().await { + break address; + } + }; + + dialer.dial(addr).unwrap(); + + let payload = Bytes::from_static(b"unreliable datagram"); + // Lossy: resend until one lands. + let mut resend = tokio::time::interval(Duration::from_millis(20)); + let mut connected = false; + let mut conn_id = None; + + let (from, data) = loop { + tokio::select! { + _ = listener.select_next_some() => {} + event = dialer.select_next_some() => { + if let SwarmEvent::ConnectionEstablished { connection_id, .. } = event { + conn_id = Some(connection_id); + connected = true; + } + } + _ = resend.tick(), if connected => { + let _ = control.send_datagram(listener_peer, payload.clone()); + } + Some(msg) = incoming.next() => break msg, + } + }; + + assert_eq!(from, dialer_peer); + assert_eq!(data, payload); + + // The connection's max datagram size is learned from the send path. + let conn_id = conn_id.unwrap(); + let max = loop { + if let Some(max) = control.max_datagram_size(listener_peer, conn_id) { + break max; + } + tokio::select! { + _ = listener.select_next_some() => {} + _ = dialer.select_next_some() => {} + _ = resend.tick() => { + let _ = control.send_datagram(listener_peer, payload.clone()); + } + } + }; + assert!(max > 0); +} + +/// Two datagram behaviours on one connection, combined through +/// `#[derive(NetworkBehaviour)]` (i.e. a nested `ConnectionHandlerSelect`, as a +/// real node composes them). A datagram sent on one must route only to that +/// behaviour, keyed on its control-stream id. +#[derive(NetworkBehaviour)] +#[behaviour(prelude = "libp2p_swarm::derive_prelude")] +struct TwoDatagrams { + a: datagram::Behaviour, + b: datagram::Behaviour, +} + +fn composed_quic_swarm() -> Swarm { + let keypair = Keypair::generate_ed25519(); + let peer_id = keypair.public().to_peer_id(); + let transport = libp2p_quic::tokio::Transport::new(libp2p_quic::Config::new(&keypair)) + .map(|(p, c), _| (p, StreamMuxerBox::new(c))) + .boxed(); + Swarm::new( + transport, + TwoDatagrams { + a: datagram::Behaviour::new(StreamProtocol::new("/example/dg-a/1.0.0")), + b: datagram::Behaviour::new(StreamProtocol::new("/example/dg-b/1.0.0")), + }, + peer_id, + Config::with_tokio_executor().with_idle_connection_timeout(Duration::from_secs(10)), + ) +} + +#[tokio::test] +async fn datagram_routes_to_owning_behaviour_when_composed() { + let _ = tracing_subscriber::fmt().with_test_writer().try_init(); + + let mut listener = composed_quic_swarm(); + let mut dialer = composed_quic_swarm(); + + let listener_peer = *listener.local_peer_id(); + let dialer_peer = *dialer.local_peer_id(); + + let mut a_in = listener.behaviour_mut().a.incoming_datagrams().unwrap(); + let mut b_in = listener.behaviour_mut().b.incoming_datagrams().unwrap(); + let mut a_control = dialer.behaviour().a.new_control(); + + listener + .listen_on("/ip4/127.0.0.1/udp/0/quic-v1".parse().unwrap()) + .unwrap(); + let addr = loop { + if let SwarmEvent::NewListenAddr { address, .. } = listener.select_next_some().await { + break address; + } + }; + + dialer.dial(addr).unwrap(); + + let payload = Bytes::from_static(b"for a only"); + let mut resend = tokio::time::interval(Duration::from_millis(20)); + let mut connected = false; + + let deadline = tokio::time::sleep(Duration::from_secs(30)); + tokio::pin!(deadline); + + // Sent on `a`, so it must surface on `a` and never leak into `b`. + let (from, data) = loop { + tokio::select! { + _ = &mut deadline => panic!("timed out waiting for the datagram on `a`"), + _ = listener.select_next_some() => {} + event = dialer.select_next_some() => { + if let SwarmEvent::ConnectionEstablished { .. } = event { + connected = true; + } + } + _ = resend.tick(), if connected => { + let _ = a_control.send_datagram(listener_peer, payload.clone()); + } + Some(msg) = a_in.next() => break msg, + Some(_) = b_in.next() => panic!("datagram for `a` leaked into `b`"), + } + }; + + assert_eq!(from, dialer_peer); + assert_eq!(data, payload); +} diff --git a/protocols/kad/src/handler.rs b/protocols/kad/src/handler.rs index 0f18d77e755..0ff6d134777 100644 --- a/protocols/kad/src/handler.rs +++ b/protocols/kad/src/handler.rs @@ -478,6 +478,7 @@ impl Handler { FullyNegotiatedOutbound { protocol: stream, info: (), + .. }: FullyNegotiatedOutbound<::OutboundProtocol>, ) { if let Some(sender) = self.pending_streams.pop_front() { diff --git a/protocols/perf/src/client/handler.rs b/protocols/perf/src/client/handler.rs index 619a391fd53..d5d697c6bf8 100644 --- a/protocols/perf/src/client/handler.rs +++ b/protocols/perf/src/client/handler.rs @@ -115,6 +115,7 @@ impl ConnectionHandler for Handler { ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol, info: (), + .. }) => { let Command { id, params } = self .requested_streams diff --git a/protocols/perf/src/server/handler.rs b/protocols/perf/src/server/handler.rs index b464eb910cc..32738b6ffb7 100644 --- a/protocols/perf/src/server/handler.rs +++ b/protocols/perf/src/server/handler.rs @@ -86,6 +86,7 @@ impl ConnectionHandler for Handler { ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { protocol, info: _, + .. }) => { #[allow(clippy::collapsible_match)] if self diff --git a/protocols/request-response/src/handler.rs b/protocols/request-response/src/handler.rs index 9bf632f1328..2dbdc7a8721 100644 --- a/protocols/request-response/src/handler.rs +++ b/protocols/request-response/src/handler.rs @@ -127,6 +127,7 @@ where FullyNegotiatedInbound { protocol: (mut stream, protocol), info: (), + .. }: FullyNegotiatedInbound<::InboundProtocol>, ) { let mut codec = self.codec.clone(); @@ -175,6 +176,7 @@ where FullyNegotiatedOutbound { protocol: (mut stream, protocol), info: (), + .. }: FullyNegotiatedOutbound<::OutboundProtocol>, ) { let message = self diff --git a/protocols/stream/src/handler.rs b/protocols/stream/src/handler.rs index 10747ebae6d..7fb9cef6d37 100644 --- a/protocols/stream/src/handler.rs +++ b/protocols/stream/src/handler.rs @@ -99,12 +99,14 @@ impl ConnectionHandler for Handler { ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { protocol: (stream, protocol), info: (), + .. }) => { Shared::lock(&self.shared).on_inbound_stream(self.remote, stream, protocol); } ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol: (stream, actual_protocol), info: (), + .. }) => { let Some((expected_protocol, sender)) = self.pending_upgrade.take() else { debug_assert!( diff --git a/swarm/CHANGELOG.md b/swarm/CHANGELOG.md index b5578f90a38..d2f45bcea09 100644 --- a/swarm/CHANGELOG.md +++ b/swarm/CHANGELOG.md @@ -1,5 +1,17 @@ ## 0.48.0 +- Add unreliable datagram support: `ConnectionEvent::Datagram` (inbound) and `ConnectionHandlerEvent::SendDatagram` (outbound). + See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489). + +- Add `ConnectionEvent::DatagramMaxSize`, reporting a connection's current max outbound datagram size to the handler after a send. + See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489). + +- Add `Stream::transport_stream_id`, exposing the transport stream id (QUIC) for datagram flows. + See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489). + +- Route inbound datagrams centrally: add `ConnectionHandler::supports_datagrams` and a `stream_id` field on `FullyNegotiatedInbound`/`FullyNegotiatedOutbound`. The connection parses the control-stream id once and routes each datagram to the owning handler instead of broadcasting. + See [PR 6489](https://github.com/libp2p/rust-libp2p/pull/6489). + - Remove `wasm-bindgen` feature and make `wasm` support implicit. See [PR 6102](https://github.com/libp2p/rust-libp2p/pull/6102) diff --git a/swarm/Cargo.toml b/swarm/Cargo.toml index 3a8544f3c52..31a31214f71 100644 --- a/swarm/Cargo.toml +++ b/swarm/Cargo.toml @@ -11,6 +11,7 @@ keywords = ["peer-to-peer", "libp2p", "networking"] categories = ["network-programming", "asynchronous"] [dependencies] +bytes = { workspace = true } either = "1.16.0" fnv = "1.0" futures = { workspace = true } diff --git a/swarm/src/behaviour/toggle.rs b/swarm/src/behaviour/toggle.rs index de86499b79a..46763ae8c96 100644 --- a/swarm/src/behaviour/toggle.rs +++ b/swarm/src/behaviour/toggle.rs @@ -201,6 +201,7 @@ where FullyNegotiatedInbound { protocol: out, info, + stream_id, }: FullyNegotiatedInbound< ::InboundProtocol, ::InboundOpenInfo, @@ -219,6 +220,7 @@ where FullyNegotiatedInbound { protocol: out, info, + stream_id, }, )); } else { @@ -294,6 +296,12 @@ where .unwrap_or(false) } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + self.inner + .as_ref() + .is_some_and(|h| h.supports_datagrams(protocol)) + } + fn poll( &mut self, cx: &mut Context<'_>, @@ -323,6 +331,7 @@ where ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol: out, info, + stream_id, }) => self .inner .as_mut() @@ -331,6 +340,7 @@ where FullyNegotiatedOutbound { protocol: out, info, + stream_id, }, )), ConnectionEvent::AddressChange(address_change) => { @@ -340,6 +350,16 @@ where })); } } + ConnectionEvent::Datagram(datagram) => { + if let Some(inner) = self.inner.as_mut() { + inner.on_connection_event(ConnectionEvent::Datagram(datagram)); + } + } + ConnectionEvent::DatagramMaxSize(max) => { + if let Some(inner) = self.inner.as_mut() { + inner.on_connection_event(ConnectionEvent::DatagramMaxSize(max)); + } + } ConnectionEvent::DialUpgradeError(DialUpgradeError { info, error: err }) => self .inner .as_mut() diff --git a/swarm/src/connection.rs b/swarm/src/connection.rs index 0a1a634d8b8..e9b6ce7929e 100644 --- a/swarm/src/connection.rs +++ b/swarm/src/connection.rs @@ -43,7 +43,7 @@ use libp2p_core::{ Endpoint, connection::ConnectedPoint, multiaddr::Multiaddr, - muxing::{StreamMuxerBox, StreamMuxerEvent, StreamMuxerExt, SubstreamBox}, + muxing::{StreamMuxer, StreamMuxerBox, StreamMuxerEvent, StreamMuxerExt, SubstreamBox}, transport::PortUse, upgrade, upgrade::{NegotiationError, ProtocolError}, @@ -55,7 +55,7 @@ use web_time::Instant; use crate::{ ConnectionHandlerEvent, Stream, StreamProtocol, StreamUpgradeError, SubstreamProtocol, handler::{ - AddressChange, ConnectionEvent, ConnectionHandler, DialUpgradeError, + AddressChange, ConnectionEvent, ConnectionHandler, Datagram, DialUpgradeError, FullyNegotiatedInbound, FullyNegotiatedOutbound, ListenUpgradeError, ProtocolSupport, ProtocolsChange, UpgradeInfoSend, }, @@ -299,6 +299,15 @@ where Poll::Ready(ConnectionHandlerEvent::NotifyBehaviour(event)) => { return Poll::Ready(Ok(Event::Handler(event))); } + Poll::Ready(ConnectionHandlerEvent::SendDatagram(data)) => { + if let Err(e) = muxing.send_datagram_unpin(data) { + tracing::error!("failed to send datagram: {e}"); + } + if let Some(max) = muxing.max_datagram_size() { + handler.on_connection_event(ConnectionEvent::DatagramMaxSize(max)); + } + continue; + } Poll::Ready(ConnectionHandlerEvent::ReportRemoteProtocols( ProtocolSupport::Added(protocols), )) => { @@ -329,9 +338,14 @@ where // negotiating outbound streams. match negotiating_out.poll_next_unpin(cx) { Poll::Pending | Poll::Ready(None) => {} - Poll::Ready(Some((info, Ok(protocol)))) => { + Poll::Ready(Some((info, Ok((protocol, negotiated, stream_id))))) => { + let stream_id = datagram_stream_id(handler, negotiated, stream_id); handler.on_connection_event(ConnectionEvent::FullyNegotiatedOutbound( - FullyNegotiatedOutbound { protocol, info }, + FullyNegotiatedOutbound { + protocol, + info, + stream_id, + }, )); continue; } @@ -347,9 +361,14 @@ where // make any more progress, poll the negotiating inbound streams. match negotiating_in.poll_next_unpin(cx) { Poll::Pending | Poll::Ready(None) => {} - Poll::Ready(Some((info, Ok(protocol)))) => { + Poll::Ready(Some((info, Ok((protocol, negotiated, stream_id))))) => { + let stream_id = datagram_stream_id(handler, negotiated, stream_id); handler.on_connection_event(ConnectionEvent::FullyNegotiatedInbound( - FullyNegotiatedInbound { protocol, info }, + FullyNegotiatedInbound { + protocol, + info, + stream_id, + }, )); continue; } @@ -409,6 +428,21 @@ where })); return Poll::Ready(Ok(Event::AddressChange(address))); } + Poll::Ready(StreamMuxerEvent::Datagram(data)) => { + match decode_stream_id(&data) { + Some((stream_id, prefix)) => { + let payload = data.slice(prefix..); + handler.on_connection_event(ConnectionEvent::Datagram(Datagram { + stream_id, + data: &payload, + })); + } + None => { + tracing::debug!("dropping datagram with truncated control-stream id") + } + } + continue; + } } if let Some(requested_substream) = requested_substreams.iter_mut().next() { @@ -542,10 +576,14 @@ impl IncomingInfo<'_> { } } +/// A successfully negotiated stream, its negotiated protocol, and its +/// transport-assigned id (where the transport has one, e.g. QUIC). +type UpgradeOk = (TOk, Option, Option); + struct StreamUpgrade { user_data: Option, timeout: Delay, - upgrade: BoxFuture<'static, Result>>, + upgrade: BoxFuture<'static, Result, StreamUpgradeError>>, } impl StreamUpgrade { @@ -578,6 +616,7 @@ impl StreamUpgrade { user_data: Some(user_data), timeout, upgrade: Box::pin(async move { + let stream_id = substream.transport_stream_id(); let (info, stream) = multistream_select::dialer_select_proto( substream, protocols, @@ -586,12 +625,13 @@ impl StreamUpgrade { .await .map_err(to_stream_upgrade_error)?; + let negotiated = StreamProtocol::try_from_owned(info.as_ref().to_owned()).ok(); let output = upgrade .upgrade_outbound(Stream::new(stream, counter), info) .await .map_err(StreamUpgradeError::Apply)?; - Ok(output) + Ok((output, negotiated, stream_id)) }), } } @@ -614,22 +654,50 @@ impl StreamUpgrade { user_data: Some(open_info), timeout: Delay::new(timeout), upgrade: Box::pin(async move { + let stream_id = substream.transport_stream_id(); let (info, stream) = multistream_select::listener_select_proto(substream, protocols) .await .map_err(to_stream_upgrade_error)?; + let negotiated = StreamProtocol::try_from_owned(info.as_ref().to_owned()).ok(); let output = upgrade .upgrade_inbound(Stream::new(stream, counter), info) .await .map_err(StreamUpgradeError::Apply)?; - Ok(output) + Ok((output, negotiated, stream_id)) }), } } } +/// Transport stream id to tag onto a fully-negotiated stream: `Some` only when +/// the handler owns datagrams for the negotiated protocol. +fn datagram_stream_id( + handler: &H, + negotiated: Option, + stream_id: Option, +) -> Option { + let id = stream_id?; + handler.supports_datagrams(&negotiated?).then_some(id) +} + +/// Decode a datagram's leading QUIC varint control-stream id (RFC 9000 s16) and +/// its byte length, or `None` if truncated. Mirrors the `libp2p-datagram` encoder. +fn decode_stream_id(datagram: &[u8]) -> Option<(u64, usize)> { + let first = *datagram.first()?; + let len = 1usize << (first >> 6); + if datagram.len() < len { + return None; + } + let mut value = u64::from(first & 0x3f); + for &b in &datagram[1..len] { + value = (value << 8) | u64::from(b); + } + Some((value, len)) +} + fn to_stream_upgrade_error(e: NegotiationError) -> StreamUpgradeError { match e { NegotiationError::Failed => StreamUpgradeError::NegotiationFailed, @@ -641,7 +709,7 @@ fn to_stream_upgrade_error(e: NegotiationError) -> StreamUpgradeError { impl Unpin for StreamUpgrade {} impl Future for StreamUpgrade { - type Output = (UserData, Result>); + type Output = (UserData, Result, StreamUpgradeError>); fn poll(mut self: Pin<&mut Self>, cx: &mut Context) -> Poll { match self.timeout.poll_unpin(cx) { @@ -1215,7 +1283,9 @@ mod tests { ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) | ConnectionEvent::LocalProtocolsChange(_) - | ConnectionEvent::RemoteProtocolsChange(_) => {} + | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::Datagram(_) + | ConnectionEvent::DatagramMaxSize(_) => {} } } diff --git a/swarm/src/dummy.rs b/swarm/src/dummy.rs index 2f4f1f2fb1b..42e0b175ac7 100644 --- a/swarm/src/dummy.rs +++ b/swarm/src/dummy.rs @@ -107,7 +107,9 @@ impl crate::handler::ConnectionHandler for ConnectionHandler { ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) | ConnectionEvent::LocalProtocolsChange(_) - | ConnectionEvent::RemoteProtocolsChange(_) => {} + | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::Datagram(_) + | ConnectionEvent::DatagramMaxSize(_) => {} } } } diff --git a/swarm/src/handler.rs b/swarm/src/handler.rs index eae30b4d46f..e51961b74da 100644 --- a/swarm/src/handler.rs +++ b/swarm/src/handler.rs @@ -54,6 +54,7 @@ use std::{ time::Duration, }; +use bytes::Bytes; use libp2p_core::Multiaddr; pub use map_in::MapInEvent; pub use map_out::MapOutEvent; @@ -154,6 +155,12 @@ pub trait ConnectionHandler: Send + 'static { false } + /// Whether `protocol` is a datagram control protocol this handler owns. + /// Combinators return the disjunction of their children. + fn supports_datagrams(&self, _protocol: &StreamProtocol) -> bool { + false + } + /// Should behave like `Stream::poll()`. fn poll( &mut self, @@ -239,6 +246,11 @@ pub enum ConnectionEvent<'a, IP: InboundUpgradeSend, OP: OutboundUpgradeSend, IO LocalProtocolsChange(ProtocolsChange<'a>), /// The remote [`ConnectionHandler`] now supports a different set of protocols. RemoteProtocolsChange(ProtocolsChange<'a>), + /// An unreliable datagram was received on the connection. + Datagram(Datagram<'a>), + /// The connection's current maximum outbound datagram size, reported after a + /// send so the sender can size fragments to the real path budget. + DatagramMaxSize(usize), } impl fmt::Debug for ConnectionEvent<'_, IP, OP, IOI, OOI> @@ -273,6 +285,10 @@ where ConnectionEvent::RemoteProtocolsChange(v) => { f.debug_tuple("RemoteProtocolsChange").field(v).finish() } + ConnectionEvent::Datagram(v) => f.debug_tuple("Datagram").field(v).finish(), + ConnectionEvent::DatagramMaxSize(v) => { + f.debug_tuple("DatagramMaxSize").field(v).finish() + } } } } @@ -283,13 +299,14 @@ impl /// Whether the event concerns an outbound stream. pub fn is_outbound(&self) -> bool { match self { - ConnectionEvent::DialUpgradeError(_) | ConnectionEvent::FullyNegotiatedOutbound(_) => { - true - } + ConnectionEvent::DialUpgradeError(_) + | ConnectionEvent::FullyNegotiatedOutbound(_) + | ConnectionEvent::DatagramMaxSize(_) => true, ConnectionEvent::FullyNegotiatedInbound(_) | ConnectionEvent::AddressChange(_) | ConnectionEvent::LocalProtocolsChange(_) | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::Datagram(_) | ConnectionEvent::ListenUpgradeError(_) => false, } } @@ -297,13 +314,14 @@ impl /// Whether the event concerns an inbound stream. pub fn is_inbound(&self) -> bool { match self { - ConnectionEvent::FullyNegotiatedInbound(_) | ConnectionEvent::ListenUpgradeError(_) => { - true - } + ConnectionEvent::FullyNegotiatedInbound(_) + | ConnectionEvent::Datagram(_) + | ConnectionEvent::ListenUpgradeError(_) => true, ConnectionEvent::FullyNegotiatedOutbound(_) | ConnectionEvent::AddressChange(_) | ConnectionEvent::LocalProtocolsChange(_) | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::DatagramMaxSize(_) | ConnectionEvent::DialUpgradeError(_) => false, } } @@ -321,6 +339,8 @@ impl pub struct FullyNegotiatedInbound { pub protocol: IP::Output, pub info: IOI, + /// Transport stream id; set only for datagram control streams. + pub stream_id: Option, } /// [`ConnectionEvent`] variant that informs the handler about successful upgrade on a new outbound @@ -332,6 +352,8 @@ pub struct FullyNegotiatedInbound { pub struct FullyNegotiatedOutbound { pub protocol: OP::Output, pub info: OOI, + /// Transport stream id; set only for datagram control streams. + pub stream_id: Option, } /// [`ConnectionEvent`] variant that informs the handler about a change in the address of the @@ -341,6 +363,14 @@ pub struct AddressChange<'a> { pub new_address: &'a Multiaddr, } +/// [`ConnectionEvent`] variant carrying an unreliable datagram, already routed +/// to the owning handler with its control-stream id parsed off. +#[derive(Debug)] +pub struct Datagram<'a> { + pub stream_id: u64, + pub data: &'a Bytes, +} + /// [`ConnectionEvent`] variant that informs the handler about a change in the protocols supported /// on the connection. #[derive(Debug, Clone)] @@ -608,6 +638,9 @@ pub enum ConnectionHandlerEvent /// Event that is sent to a [`NetworkBehaviour`](crate::behaviour::NetworkBehaviour). NotifyBehaviour(TCustom), + + /// Send an unreliable datagram over the connection. + SendDatagram(Bytes), } #[derive(Debug, Clone, PartialEq, Eq)] @@ -643,6 +676,9 @@ impl ConnectionHandlerEvent::ReportRemoteProtocols(support) => { ConnectionHandlerEvent::ReportRemoteProtocols(support) } + ConnectionHandlerEvent::SendDatagram(data) => { + ConnectionHandlerEvent::SendDatagram(data) + } } } @@ -664,6 +700,9 @@ impl ConnectionHandlerEvent::ReportRemoteProtocols(support) => { ConnectionHandlerEvent::ReportRemoteProtocols(support) } + ConnectionHandlerEvent::SendDatagram(data) => { + ConnectionHandlerEvent::SendDatagram(data) + } } } @@ -685,6 +724,9 @@ impl ConnectionHandlerEvent::ReportRemoteProtocols(support) => { ConnectionHandlerEvent::ReportRemoteProtocols(support) } + ConnectionHandlerEvent::SendDatagram(data) => { + ConnectionHandlerEvent::SendDatagram(data) + } } } } diff --git a/swarm/src/handler/either.rs b/swarm/src/handler/either.rs index 1dc62e0eb2a..bd18ae859fc 100644 --- a/swarm/src/handler/either.rs +++ b/swarm/src/handler/either.rs @@ -44,11 +44,21 @@ where FullyNegotiatedInbound { protocol: future::Either::Left(protocol), info: Either::Left(info), - } => Either::Left(FullyNegotiatedInbound { protocol, info }), + stream_id, + } => Either::Left(FullyNegotiatedInbound { + protocol, + info, + stream_id, + }), FullyNegotiatedInbound { protocol: future::Either::Right(protocol), info: Either::Right(info), - } => Either::Right(FullyNegotiatedInbound { protocol, info }), + stream_id, + } => Either::Right(FullyNegotiatedInbound { + protocol, + info, + stream_id, + }), _ => unreachable!(), } } @@ -118,6 +128,13 @@ where } } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + match self { + Either::Left(handler) => handler.supports_datagrams(protocol), + Either::Right(handler) => handler.supports_datagrams(protocol), + } + } + fn poll( &mut self, cx: &mut Context<'_>, @@ -213,6 +230,22 @@ where handler.on_connection_event(ConnectionEvent::AddressChange(address_change)) } }, + ConnectionEvent::Datagram(datagram) => match self { + Either::Left(handler) => { + handler.on_connection_event(ConnectionEvent::Datagram(datagram)) + } + Either::Right(handler) => { + handler.on_connection_event(ConnectionEvent::Datagram(datagram)) + } + }, + ConnectionEvent::DatagramMaxSize(max) => match self { + Either::Left(handler) => { + handler.on_connection_event(ConnectionEvent::DatagramMaxSize(max)) + } + Either::Right(handler) => { + handler.on_connection_event(ConnectionEvent::DatagramMaxSize(max)) + } + }, ConnectionEvent::LocalProtocolsChange(supported_protocols) => match self { Either::Left(handler) => handler.on_connection_event( ConnectionEvent::LocalProtocolsChange(supported_protocols), diff --git a/swarm/src/handler/map_in.rs b/swarm/src/handler/map_in.rs index 55885b351bb..9619858a564 100644 --- a/swarm/src/handler/map_in.rs +++ b/swarm/src/handler/map_in.rs @@ -76,6 +76,10 @@ where self.inner.connection_keep_alive() } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + self.inner.supports_datagrams(protocol) + } + fn poll( &mut self, cx: &mut Context<'_>, diff --git a/swarm/src/handler/map_out.rs b/swarm/src/handler/map_out.rs index 6d05551aeec..2ee7031b40c 100644 --- a/swarm/src/handler/map_out.rs +++ b/swarm/src/handler/map_out.rs @@ -69,6 +69,10 @@ where self.inner.connection_keep_alive() } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + self.inner.supports_datagrams(protocol) + } + fn poll( &mut self, cx: &mut Context<'_>, @@ -85,6 +89,9 @@ where ConnectionHandlerEvent::ReportRemoteProtocols(support) => { ConnectionHandlerEvent::ReportRemoteProtocols(support) } + ConnectionHandlerEvent::SendDatagram(data) => { + ConnectionHandlerEvent::SendDatagram(data) + } }) } diff --git a/swarm/src/handler/multi.rs b/swarm/src/handler/multi.rs index 9fe4afdf61a..3449316df80 100644 --- a/swarm/src/handler/multi.rs +++ b/swarm/src/handler/multi.rs @@ -49,6 +49,8 @@ use crate::{ #[derive(Clone)] pub struct MultiHandler { handlers: HashMap, + /// Control-stream id -> owning handler key, for routing inbound datagrams. + datagram_routes: HashMap, } impl fmt::Debug for MultiHandler @@ -77,6 +79,7 @@ where { let m = MultiHandler { handlers: HashMap::from_iter(iter), + datagram_routes: HashMap::new(), }; uniq_proto_names( m.handlers @@ -156,12 +159,17 @@ where ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol, info: (key, arg), + stream_id, }) => { if let Some(h) = self.handlers.get_mut(&key) { + if let Some(id) = stream_id { + self.datagram_routes.insert(id, key.clone()); + } h.on_connection_event(ConnectionEvent::FullyNegotiatedOutbound( FullyNegotiatedOutbound { protocol, info: arg, + stream_id, }, )); } else { @@ -171,13 +179,18 @@ where ConnectionEvent::FullyNegotiatedInbound(FullyNegotiatedInbound { protocol: (key, arg), mut info, + stream_id, }) => { if let Some(h) = self.handlers.get_mut(&key) { if let Some(i) = info.take(&key) { + if let Some(id) = stream_id { + self.datagram_routes.insert(id, key.clone()); + } h.on_connection_event(ConnectionEvent::FullyNegotiatedInbound( FullyNegotiatedInbound { protocol: arg, info: i, + stream_id, }, )); } @@ -192,6 +205,18 @@ where })); } } + ConnectionEvent::Datagram(datagram) => { + if let Some(key) = self.datagram_routes.get(&datagram.stream_id) + && let Some(h) = self.handlers.get_mut(key) + { + h.on_connection_event(ConnectionEvent::Datagram(datagram)); + } + } + ConnectionEvent::DatagramMaxSize(max) => { + for h in self.handlers.values_mut() { + h.on_connection_event(ConnectionEvent::DatagramMaxSize(max)); + } + } ConnectionEvent::DialUpgradeError(DialUpgradeError { info: (key, arg), error, @@ -241,6 +266,12 @@ where .unwrap_or(false) } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + self.handlers + .values() + .any(|h| h.supports_datagrams(protocol)) + } + fn poll( &mut self, cx: &mut Context<'_>, diff --git a/swarm/src/handler/one_shot.rs b/swarm/src/handler/one_shot.rs index eba0a3a25a3..775fd701f6c 100644 --- a/swarm/src/handler/one_shot.rs +++ b/swarm/src/handler/one_shot.rs @@ -189,7 +189,9 @@ where ConnectionEvent::AddressChange(_) | ConnectionEvent::ListenUpgradeError(_) | ConnectionEvent::LocalProtocolsChange(_) - | ConnectionEvent::RemoteProtocolsChange(_) => {} + | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::Datagram(_) + | ConnectionEvent::DatagramMaxSize(_) => {} } } } diff --git a/swarm/src/handler/pending.rs b/swarm/src/handler/pending.rs index 0d9191f2de5..ae75ba04130 100644 --- a/swarm/src/handler/pending.rs +++ b/swarm/src/handler/pending.rs @@ -77,6 +77,7 @@ impl ConnectionHandler for PendingConnectionHandler { ConnectionEvent::FullyNegotiatedOutbound(FullyNegotiatedOutbound { protocol, info: _info, + .. }) => { libp2p_core::util::unreachable(protocol); #[allow(unreachable_code, clippy::used_underscore_binding)] @@ -88,7 +89,9 @@ impl ConnectionHandler for PendingConnectionHandler { | ConnectionEvent::DialUpgradeError(_) | ConnectionEvent::ListenUpgradeError(_) | ConnectionEvent::LocalProtocolsChange(_) - | ConnectionEvent::RemoteProtocolsChange(_) => {} + | ConnectionEvent::RemoteProtocolsChange(_) + | ConnectionEvent::Datagram(_) + | ConnectionEvent::DatagramMaxSize(_) => {} } } } diff --git a/swarm/src/handler/select.rs b/swarm/src/handler/select.rs index 0f6dbe988ff..fbbf5b46e6b 100644 --- a/swarm/src/handler/select.rs +++ b/swarm/src/handler/select.rs @@ -20,6 +20,7 @@ use std::{ cmp, + collections::HashMap, task::{Context, Poll}, }; @@ -36,6 +37,13 @@ use crate::{ upgrade::SendWrapper, }; +/// Which child of a [`ConnectionHandlerSelect`] owns a datagram control stream. +#[derive(Debug, Clone, Copy)] +enum Side { + First, + Second, +} + /// Implementation of [`ConnectionHandler`] that combines two protocols into one. #[derive(Debug, Clone)] pub struct ConnectionHandlerSelect { @@ -43,12 +51,18 @@ pub struct ConnectionHandlerSelect { proto1: TProto1, /// The second protocol. proto2: TProto2, + /// Control-stream id -> owning child, for routing inbound datagrams. + datagram_routes: HashMap, } impl ConnectionHandlerSelect { /// Builds a [`ConnectionHandlerSelect`]. pub(crate) fn new(proto1: TProto1, proto2: TProto2) -> Self { - ConnectionHandlerSelect { proto1, proto2 } + ConnectionHandlerSelect { + proto1, + proto2, + datagram_routes: HashMap::new(), + } } pub fn into_inner(self) -> (TProto1, TProto2) { @@ -71,11 +85,21 @@ where FullyNegotiatedOutbound { protocol: future::Either::Left(protocol), info: Either::Left(info), - } => Either::Left(FullyNegotiatedOutbound { protocol, info }), + stream_id, + } => Either::Left(FullyNegotiatedOutbound { + protocol, + info, + stream_id, + }), FullyNegotiatedOutbound { protocol: future::Either::Right(protocol), info: Either::Right(info), - } => Either::Right(FullyNegotiatedOutbound { protocol, info }), + stream_id, + } => Either::Right(FullyNegotiatedOutbound { + protocol, + info, + stream_id, + }), _ => panic!("wrong API usage: the protocol doesn't match the upgrade info"), } } @@ -94,11 +118,21 @@ where FullyNegotiatedInbound { protocol: future::Either::Left(protocol), info: (i1, _i2), - } => Either::Left(FullyNegotiatedInbound { protocol, info: i1 }), + stream_id, + } => Either::Left(FullyNegotiatedInbound { + protocol, + info: i1, + stream_id, + }), FullyNegotiatedInbound { protocol: future::Either::Right(protocol), info: (_i1, i2), - } => Either::Right(FullyNegotiatedInbound { protocol, info: i2 }), + stream_id, + } => Either::Right(FullyNegotiatedInbound { + protocol, + info: i2, + stream_id, + }), } } } @@ -221,6 +255,10 @@ where ) } + fn supports_datagrams(&self, protocol: &crate::StreamProtocol) -> bool { + self.proto1.supports_datagrams(protocol) || self.proto2.supports_datagrams(protocol) + } + fn poll( &mut self, cx: &mut Context<'_>, @@ -241,6 +279,9 @@ where Poll::Ready(ConnectionHandlerEvent::ReportRemoteProtocols(support)) => { return Poll::Ready(ConnectionHandlerEvent::ReportRemoteProtocols(support)); } + Poll::Ready(ConnectionHandlerEvent::SendDatagram(data)) => { + return Poll::Ready(ConnectionHandlerEvent::SendDatagram(data)); + } Poll::Pending => (), }; @@ -260,6 +301,9 @@ where Poll::Ready(ConnectionHandlerEvent::ReportRemoteProtocols(support)) => { return Poll::Ready(ConnectionHandlerEvent::ReportRemoteProtocols(support)); } + Poll::Ready(ConnectionHandlerEvent::SendDatagram(data)) => { + return Poll::Ready(ConnectionHandlerEvent::SendDatagram(data)); + } Poll::Pending => (), }; @@ -290,22 +334,38 @@ where match event { ConnectionEvent::FullyNegotiatedOutbound(fully_negotiated_outbound) => { match fully_negotiated_outbound.transpose() { - Either::Left(f) => self - .proto1 - .on_connection_event(ConnectionEvent::FullyNegotiatedOutbound(f)), - Either::Right(f) => self - .proto2 - .on_connection_event(ConnectionEvent::FullyNegotiatedOutbound(f)), + Either::Left(f) => { + if let Some(id) = f.stream_id { + self.datagram_routes.insert(id, Side::First); + } + self.proto1 + .on_connection_event(ConnectionEvent::FullyNegotiatedOutbound(f)); + } + Either::Right(f) => { + if let Some(id) = f.stream_id { + self.datagram_routes.insert(id, Side::Second); + } + self.proto2 + .on_connection_event(ConnectionEvent::FullyNegotiatedOutbound(f)); + } } } ConnectionEvent::FullyNegotiatedInbound(fully_negotiated_inbound) => { match fully_negotiated_inbound.transpose() { - Either::Left(f) => self - .proto1 - .on_connection_event(ConnectionEvent::FullyNegotiatedInbound(f)), - Either::Right(f) => self - .proto2 - .on_connection_event(ConnectionEvent::FullyNegotiatedInbound(f)), + Either::Left(f) => { + if let Some(id) = f.stream_id { + self.datagram_routes.insert(id, Side::First); + } + self.proto1 + .on_connection_event(ConnectionEvent::FullyNegotiatedInbound(f)); + } + Either::Right(f) => { + if let Some(id) = f.stream_id { + self.datagram_routes.insert(id, Side::Second); + } + self.proto2 + .on_connection_event(ConnectionEvent::FullyNegotiatedInbound(f)); + } } } ConnectionEvent::AddressChange(address) => { @@ -319,6 +379,26 @@ where new_address: address.new_address, })); } + ConnectionEvent::Datagram(datagram) => { + match self.datagram_routes.get(&datagram.stream_id) { + Some(Side::First) => { + self.proto1 + .on_connection_event(ConnectionEvent::Datagram(datagram)); + } + Some(Side::Second) => { + self.proto2 + .on_connection_event(ConnectionEvent::Datagram(datagram)); + } + // Unknown control stream; drop. + None => {} + } + } + ConnectionEvent::DatagramMaxSize(max) => { + self.proto1 + .on_connection_event(ConnectionEvent::DatagramMaxSize(max)); + self.proto2 + .on_connection_event(ConnectionEvent::DatagramMaxSize(max)); + } ConnectionEvent::DialUpgradeError(dial_upgrade_error) => { match dial_upgrade_error.transpose() { Either::Left(err) => self diff --git a/swarm/src/stream.rs b/swarm/src/stream.rs index caa4e56bd27..ee4cd701562 100644 --- a/swarm/src/stream.rs +++ b/swarm/src/stream.rs @@ -51,6 +51,15 @@ impl Stream { pub fn ignore_for_keep_alive(&mut self) { self.counter.take(); } + + /// Transport-assigned stream id, where one exists (QUIC, WebTransport). + /// + /// See [libp2p/specs#680](https://github.com/libp2p/specs/pull/680). + pub fn transport_stream_id(&self) -> Option { + self.stream + .as_inner() + .and_then(SubstreamBox::transport_stream_id) + } } impl AsyncRead for Stream { diff --git a/transports/quic/CHANGELOG.md b/transports/quic/CHANGELOG.md index 00898299001..f522cc532e7 100644 --- a/transports/quic/CHANGELOG.md +++ b/transports/quic/CHANGELOG.md @@ -1,5 +1,8 @@ ## 0.14.0 +- Support unreliable QUIC datagrams, enabled by default. Tune via `Config::datagram_receive_buffer_size` and `datagram_send_buffer_size`. Substreams now report their QUIC stream id via `StreamMuxer::substream_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). diff --git a/transports/quic/Cargo.toml b/transports/quic/Cargo.toml index fd8ac3862d8..cde7339535d 100644 --- a/transports/quic/Cargo.toml +++ b/transports/quic/Cargo.toml @@ -9,6 +9,7 @@ repository = "https://github.com/libp2p/rust-libp2p" license = "MIT" [dependencies] +bytes = { workspace = true } futures = { workspace = true } futures-timer = { workspace = true } if-watch = { workspace = true } diff --git a/transports/quic/src/config.rs b/transports/quic/src/config.rs index 0d514db5751..181b58f1887 100644 --- a/transports/quic/src/config.rs +++ b/transports/quic/src/config.rs @@ -51,6 +51,13 @@ pub struct Config { /// of a connection. pub max_connection_data: u32, + /// Bytes to buffer for received unreliable datagrams. `None` disables + /// datagram support. + pub datagram_receive_buffer_size: Option, + + /// Bytes to buffer for outgoing unreliable datagrams. + pub datagram_send_buffer_size: usize, + /// Support QUIC version draft-29 for dialing and listening. /// /// Per default only QUIC Version 1 / [`libp2p_core::multiaddr::Protocol::QuicV1`] @@ -96,6 +103,8 @@ impl Config { // Ensure that one stream is not consuming the whole connection. max_stream_data: 10_000_000, + datagram_receive_buffer_size: Some(256 * 1024), + datagram_send_buffer_size: 256 * 1024, keypair: keypair.clone(), mtu_discovery_config: Some(Default::default()), } @@ -135,6 +144,8 @@ impl From for QuinnConfig { keep_alive_interval, max_connection_data, max_stream_data, + datagram_receive_buffer_size, + datagram_send_buffer_size, support_draft_29, handshake_timeout: _, keypair, @@ -144,8 +155,8 @@ impl From for QuinnConfig { // Disable uni-directional streams. transport.max_concurrent_uni_streams(0u32.into()); transport.max_concurrent_bidi_streams(max_concurrent_stream_limit.into()); - // Disable datagrams. - transport.datagram_receive_buffer_size(None); + transport.datagram_receive_buffer_size(datagram_receive_buffer_size); + transport.datagram_send_buffer_size(datagram_send_buffer_size); transport.keep_alive_interval(Some(keep_alive_interval)); transport.max_idle_timeout(Some(VarInt::from_u32(max_idle_timeout).into())); transport.allow_spin(false); diff --git a/transports/quic/src/connection.rs b/transports/quic/src/connection.rs index a822b56d9d4..b7f97fe4e6d 100644 --- a/transports/quic/src/connection.rs +++ b/transports/quic/src/connection.rs @@ -26,9 +26,10 @@ use std::{ task::{Context, Poll}, }; +use bytes::Bytes; pub use connecting::Connecting; use futures::{FutureExt, future::BoxFuture}; -use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; +use libp2p_core::muxing::{SendDatagramError, StreamMuxer, StreamMuxerEvent}; pub use stream::Stream; use crate::{ConnectionError, Error}; @@ -47,6 +48,8 @@ pub struct Connection { >, /// Future to wait for the connection to be closed. closing: Option>, + /// Future for the next inbound datagram. + datagrams: Option>>, } impl Connection { @@ -60,6 +63,7 @@ impl Connection { incoming: None, outgoing: None, closing: None, + datagrams: None, } } } @@ -104,11 +108,39 @@ impl StreamMuxer for Connection { fn poll( self: Pin<&mut Self>, - _cx: &mut Context<'_>, + cx: &mut Context<'_>, ) -> Poll> { - // TODO: If connection migration is enabled (currently disabled) address - // change on the connection needs to be handled. - Poll::Pending + let this = self.get_mut(); + let datagrams = this.datagrams.get_or_insert_with(|| { + let connection = this.connection.clone(); + async move { connection.read_datagram().await }.boxed() + }); + let bytes = futures::ready!(datagrams.poll_unpin(cx)).map_err(ConnectionError)?; + this.datagrams.take(); + Poll::Ready(Ok(StreamMuxerEvent::Datagram(bytes))) + } + + fn send_datagram(self: Pin<&mut Self>, data: Bytes) -> Result<(), SendDatagramError> { + let connection = &self.get_mut().connection; + let size = data.len(); + connection.send_datagram(data).map_err(|e| match e { + quinn::SendDatagramError::UnsupportedByPeer | quinn::SendDatagramError::Disabled => { + SendDatagramError::Unsupported + } + quinn::SendDatagramError::TooLarge => SendDatagramError::TooLarge { + size, + max: connection.max_datagram_size().unwrap_or(0), + }, + quinn::SendDatagramError::ConnectionLost(_) => SendDatagramError::ConnectionClosed, + }) + } + + fn max_datagram_size(&self) -> Option { + self.connection.max_datagram_size() + } + + fn substream_id(substream: &Self::Substream) -> Option { + Some(substream.id()) } fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { diff --git a/transports/quic/src/connection/stream.rs b/transports/quic/src/connection/stream.rs index 4bc7a28639c..6e3e3dfff29 100644 --- a/transports/quic/src/connection/stream.rs +++ b/transports/quic/src/connection/stream.rs @@ -44,6 +44,10 @@ impl Stream { close_result: None, } } + + pub(super) fn id(&self) -> u64 { + self.send.id().into() + } } impl AsyncRead for Stream { diff --git a/transports/quic/tests/smoke.rs b/transports/quic/tests/smoke.rs index 2cbbe7ecb7f..b60dea491a5 100644 --- a/transports/quic/tests/smoke.rs +++ b/transports/quic/tests/smoke.rs @@ -21,7 +21,7 @@ use futures_timer::Delay; use libp2p_core::{ Endpoint, Multiaddr, Transport, multiaddr::Protocol, - muxing::{StreamMuxerBox, StreamMuxerExt, SubstreamBox}, + muxing::{StreamMuxer, StreamMuxerBox, StreamMuxerEvent, StreamMuxerExt, SubstreamBox}, transport::{ Boxed, DialOpts, ListenerId, OrTransport, PortUse, TransportError, TransportEvent, }, @@ -42,6 +42,54 @@ async fn tokio_smoke() { smoke::().await } +#[cfg(feature = "tokio")] +#[tokio::test] +async fn tokio_datagram_roundtrip() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + let (_, mut a_transport) = create_default_transport::(); + let (_, mut b_transport) = create_default_transport::(); + + let a_addr = start_listening(&mut a_transport, "/ip4/127.0.0.1/udp/0/quic-v1").await; + let ((_, _, mut a_conn), (_, mut b_conn)) = + connect(&mut a_transport, &mut b_transport, a_addr).await; + + assert!(b_conn.max_datagram_size().is_some(), "datagrams enabled"); + + let payload = bytes::Bytes::from_static(b"unreliable datagram"); + Pin::new(&mut b_conn) + .send_datagram(payload.clone()) + .unwrap(); + + let received = poll_fn(|cx| { + let _ = b_conn.poll_unpin(cx); + match a_conn.poll_unpin(cx) { + Poll::Ready(Ok(StreamMuxerEvent::Datagram(bytes))) => Poll::Ready(bytes), + Poll::Ready(Err(e)) => panic!("muxer error: {e}"), + _ => Poll::Pending, + } + }) + .await; + + assert_eq!(received, payload); +} + +#[cfg(feature = "tokio")] +#[tokio::test] +async fn tokio_substream_id_surfaced() { + let _ = tracing_subscriber::fmt() + .with_env_filter(EnvFilter::from_default_env()) + .try_init(); + + let (stream_a, stream_b) = build_streams::().await; + + // Both ends of one bidi stream report the same QUIC stream id (libp2p/specs#680). + let id = stream_a.transport_stream_id(); + assert!(id.is_some(), "QUIC surfaces a stream id"); + assert_eq!(id, stream_b.transport_stream_id()); +} + #[cfg(feature = "tokio")] #[tokio::test] async fn endpoint_reuse() {