From c7d37a8dc047259fa35084946e1790487387ef9e Mon Sep 17 00:00:00 2001 From: yexiyue Date: Fri, 24 Jul 2026 23:18:59 +0800 Subject: [PATCH 1/3] feat(webrtc): negotiate data channel message limits --- misc/webrtc-utils/src/lib.rs | 4 +- misc/webrtc-utils/src/noise.rs | 129 +++++++++++++++++- misc/webrtc-utils/src/sdp.rs | 4 +- misc/webrtc-utils/src/stream.rs | 94 +++++++++++-- misc/webrtc-utils/src/stream/framed_dc.rs | 12 +- transports/webrtc-websys/src/connection.rs | 12 +- transports/webrtc-websys/src/stream.rs | 9 +- .../src/stream/poll_data_channel.rs | 12 +- transports/webrtc-websys/src/transport.rs | 19 ++- transports/webrtc-websys/src/upgrade.rs | 27 +++- transports/webrtc/src/tokio/connection.rs | 9 +- transports/webrtc/src/tokio/sdp.rs | 2 +- transports/webrtc/src/tokio/stream.rs | 13 +- transports/webrtc/src/tokio/transport.rs | 12 ++ transports/webrtc/src/tokio/upgrade.rs | 22 ++- 15 files changed, 316 insertions(+), 64 deletions(-) diff --git a/misc/webrtc-utils/src/lib.rs b/misc/webrtc-utils/src/lib.rs index 330cebee3b1..43c3cbbbc76 100644 --- a/misc/webrtc-utils/src/lib.rs +++ b/misc/webrtc-utils/src/lib.rs @@ -11,5 +11,7 @@ mod stream; mod transport; pub use fingerprint::{Fingerprint, SHA256}; -pub use stream::{DropListener, MAX_MSG_LEN, Stream}; +pub use stream::{ + DEFAULT_MAX_MESSAGE_SIZE, DropListener, MAX_MSG_LEN, MIN_MESSAGE_SIZE, Stream, StreamConfig, +}; pub use transport::parse_webrtc_dial_addr; diff --git a/misc/webrtc-utils/src/noise.rs b/misc/webrtc-utils/src/noise.rs index bcb5102120f..249e1da70e9 100644 --- a/misc/webrtc-utils/src/noise.rs +++ b/misc/webrtc-utils/src/noise.rs @@ -18,7 +18,9 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. -use futures::{AsyncRead, AsyncWrite, AsyncWriteExt}; +use std::num::NonZeroUsize; + +use futures::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt}; use libp2p_core::{ UpgradeInfo, upgrade::{InboundConnectionUpgrade, OutboundConnectionUpgrade}, @@ -28,7 +30,10 @@ use libp2p_identity::PeerId; use libp2p_noise as noise; pub use noise::Error; -use crate::fingerprint::Fingerprint; +use crate::{ + fingerprint::Fingerprint, + stream::{DEFAULT_MAX_MESSAGE_SIZE, MIN_MESSAGE_SIZE, StreamConfig}, +}; pub async fn inbound( id_keys: identity::Keypair, @@ -36,6 +41,28 @@ pub async fn inbound( client_fingerprint: Fingerprint, server_fingerprint: Fingerprint, ) -> Result +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + inbound_with_message_size( + id_keys, + stream, + client_fingerprint, + server_fingerprint, + StreamConfig::default(), + ) + .await + .map(|(peer_id, _)| peer_id) +} + +/// Authenticates the connection and negotiates its encoded message-size limit. +pub async fn inbound_with_message_size( + id_keys: identity::Keypair, + stream: T, + client_fingerprint: Fingerprint, + server_fingerprint: Fingerprint, + stream_config: StreamConfig, +) -> Result<(PeerId, StreamConfig), Error> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -47,9 +74,10 @@ where // send application data 0.5 RTT earlier. let (peer_id, mut channel) = noise.upgrade_outbound(stream, info).await?; - channel.close().await?; + let stream_config = negotiate_message_size(&mut channel, stream_config).await; + let _ = channel.close().await; - Ok(peer_id) + Ok((peer_id, stream_config)) } pub async fn outbound( @@ -58,6 +86,28 @@ pub async fn outbound( server_fingerprint: Fingerprint, client_fingerprint: Fingerprint, ) -> Result +where + T: AsyncRead + AsyncWrite + Unpin + Send + 'static, +{ + outbound_with_message_size( + id_keys, + stream, + server_fingerprint, + client_fingerprint, + StreamConfig::default(), + ) + .await + .map(|(peer_id, _)| peer_id) +} + +/// Authenticates the connection and negotiates its encoded message-size limit. +pub async fn outbound_with_message_size( + id_keys: identity::Keypair, + stream: T, + server_fingerprint: Fingerprint, + client_fingerprint: Fingerprint, + stream_config: StreamConfig, +) -> Result<(PeerId, StreamConfig), Error> where T: AsyncRead + AsyncWrite + Unpin + Send + 'static, { @@ -69,9 +119,52 @@ where // send application data 0.5 RTT earlier. let (peer_id, mut channel) = noise.upgrade_inbound(stream, info).await?; - channel.close().await?; + let stream_config = negotiate_message_size(&mut channel, stream_config).await; + let _ = channel.close().await; - Ok(peer_id) + Ok((peer_id, stream_config)) +} + +/// Exchanges the local limit after the authenticated Noise handshake. +/// +/// Older peers close the reserved data channel immediately after Noise. Treat that as their +/// historical 16 KiB limit, preserving compatibility while newer peers use the smaller limit. +async fn negotiate_message_size(channel: &mut T, local: StreamConfig) -> StreamConfig +where + T: AsyncRead + AsyncWrite + Unpin, +{ + let fallback = local.limited_by(DEFAULT_MAX_MESSAGE_SIZE); + let advertised = local.max_message_size() as u64; + + if channel.write_all(&advertised.to_be_bytes()).await.is_err() || channel.flush().await.is_err() + { + return fallback; + } + + let mut remote = [0; std::mem::size_of::()]; + if channel.read_exact(&mut remote).await.is_err() { + return fallback; + } + + effective_message_size(local, Some(u64::from_be_bytes(remote))) +} + +fn effective_message_size(local: StreamConfig, remote: Option) -> StreamConfig { + let fallback = local.limited_by(DEFAULT_MAX_MESSAGE_SIZE); + let Some(remote) = remote else { + return fallback; + }; + let Ok(remote) = usize::try_from(remote) else { + return fallback; + }; + let Some(remote) = NonZeroUsize::new(remote) else { + return fallback; + }; + if remote < MIN_MESSAGE_SIZE { + return fallback; + } + + local.limited_by(remote) } pub(crate) fn noise_prologue( @@ -115,4 +208,28 @@ mod tests { "6c69627032702d7765627274632d6e6f6973653a122030fc9f469c207419dfdd0aab5f27a86c973c94e40548db9375cca2e915973b9912203e79af40d6059617a0d83b83a52ce73b0c1f37a72c6043ad2969e2351bdca870" ); } + + #[test] + fn message_size_negotiation_uses_the_smaller_valid_limit() { + let local = StreamConfig::new(NonZeroUsize::new(16 * 1024).unwrap()); + + assert_eq!( + effective_message_size(local, Some(8 * 1024)).max_message_size(), + 8 * 1024 + ); + } + + #[test] + fn message_size_negotiation_falls_back_for_legacy_or_invalid_peers() { + let local = StreamConfig::new(NonZeroUsize::new(8 * 1024).unwrap()); + + assert_eq!( + effective_message_size(local, None).max_message_size(), + 8 * 1024 + ); + assert_eq!( + effective_message_size(local, Some(0)).max_message_size(), + 8 * 1024 + ); + } } diff --git a/misc/webrtc-utils/src/sdp.rs b/misc/webrtc-utils/src/sdp.rs index 2be0ed76f29..5eb7dac5e42 100644 --- a/misc/webrtc-utils/src/sdp.rs +++ b/misc/webrtc-utils/src/sdp.rs @@ -91,7 +91,7 @@ a=ice-pwd:{pwd} a=fingerprint:{fingerprint_algorithm} {fingerprint_value} a=setup:passive a=sctp-port:5000 -a=max-message-size:16384 +a=max-message-size:{max_message_size} a=candidate:1467250027 1 UDP 1467250027 {target_ip} {target_port} typ host a=end-of-candidates "; @@ -114,6 +114,7 @@ struct DescriptionContext { pub(crate) fingerprint_value: String, pub(crate) ufrag: String, pub(crate) pwd: String, + pub(crate) max_message_size: usize, } /// Renders a [`TinyTemplate`] description using the provided arguments. @@ -141,6 +142,7 @@ pub fn render_description( // NOTE: ufrag is equal to pwd. ufrag: ufrag.to_owned(), pwd: ufrag.to_owned(), + max_message_size: 16 * 1024, }; tt.render("description", &context).unwrap() } diff --git a/misc/webrtc-utils/src/stream.rs b/misc/webrtc-utils/src/stream.rs index 96ad82564c1..fad21372b48 100644 --- a/misc/webrtc-utils/src/stream.rs +++ b/misc/webrtc-utils/src/stream.rs @@ -21,6 +21,7 @@ use std::{ io, + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -41,18 +42,67 @@ mod drop_listener; mod framed_dc; mod state; -/// Maximum length of a message. -/// -/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message -/// size to 16 KB to avoid monopolization." -/// Source: -pub const MAX_MSG_LEN: usize = 16 * 1024; /// Length of varint, in bytes. const VARINT_LEN: usize = 2; /// Overhead of the protobuf encoding, in bytes. const PROTO_OVERHEAD: usize = 5; -/// Maximum length of data, in bytes. -const MAX_DATA_LEN: usize = MAX_MSG_LEN - VARINT_LEN - PROTO_OVERHEAD; + +/// Default maximum length of a WebRTC data-channel message. +/// +/// "As long as message interleaving is not supported, the sender SHOULD limit the maximum message +/// size to 16 KB to avoid monopolization." +/// Source: +pub const DEFAULT_MAX_MESSAGE_SIZE: NonZeroUsize = + NonZeroUsize::new(16 * 1024).expect("constant is non-zero"); +/// Smallest encoded message that can carry one byte of application data. +pub const MIN_MESSAGE_SIZE: NonZeroUsize = + NonZeroUsize::new(VARINT_LEN + PROTO_OVERHEAD + 1).expect("constant is non-zero"); +/// Backwards-compatible default message-size value. +pub const MAX_MSG_LEN: usize = DEFAULT_MAX_MESSAGE_SIZE.get(); + +/// Per-connection WebRTC message-size policy. +/// +/// The same value must be used by the framing codec, its write high-water mark and the +/// transport's data-channel backpressure accounting. Keeping it in one value prevents those +/// layers from silently disagreeing about a valid frame size. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub struct StreamConfig { + max_message_size: NonZeroUsize, +} + +impl StreamConfig { + /// Creates a stream configuration with the provided maximum encoded message size. + pub const fn new(max_message_size: NonZeroUsize) -> Self { + assert!(max_message_size.get() >= MIN_MESSAGE_SIZE.get()); + Self { max_message_size } + } + + /// Returns the smaller of two independently advertised limits. + pub const fn limited_by(self, remote_max_message_size: NonZeroUsize) -> Self { + Self::new( + if self.max_message_size.get() <= remote_max_message_size.get() { + self.max_message_size + } else { + remote_max_message_size + }, + ) + } + + /// Returns the maximum encoded data-channel message size. + pub const fn max_message_size(self) -> usize { + self.max_message_size.get() + } + + pub(crate) const fn max_data_size(self) -> usize { + self.max_message_size.get() - VARINT_LEN - PROTO_OVERHEAD + } +} + +impl Default for StreamConfig { + fn default() -> Self { + Self::new(DEFAULT_MAX_MESSAGE_SIZE) + } +} pub use drop_listener::DropListener; /// A stream backed by a WebRTC data channel. @@ -63,6 +113,7 @@ pub struct Stream { io: FramedDc, state: State, read_buffer: Bytes, + config: StreamConfig, /// Dropping this will close the oneshot and notify the receiver by emitting `Canceled`. drop_notifier: Option>, } @@ -74,15 +125,21 @@ where /// Returns a new [`Stream`] and a [`DropListener`], /// which will notify the receiver when/if the stream is dropped. pub fn new(data_channel: T) -> (Self, DropListener) { + Self::with_config(data_channel, StreamConfig::default()) + } + + /// Returns a new stream and drop listener using the supplied message-size policy. + pub fn with_config(data_channel: T, config: StreamConfig) -> (Self, DropListener) { let (sender, receiver) = oneshot::channel(); let stream = Self { - io: framed_dc::new(data_channel.clone()), + io: framed_dc::new(data_channel.clone(), config), state: State::Open, read_buffer: Bytes::default(), + config, drop_notifier: Some(sender), }; - let listener = DropListener::new(framed_dc::new(data_channel), receiver); + let listener = DropListener::new(framed_dc::new(data_channel, config), receiver); (stream, listener) } @@ -205,7 +262,7 @@ where ready!(self.io.poll_ready_unpin(cx))?; - let n = usize::min(buf.len(), MAX_DATA_LEN); + let n = usize::min(buf.len(), self.config.max_data_size()); Pin::new(&mut self.io).start_send(Message { flag: None, @@ -284,21 +341,30 @@ mod tests { #[test] fn max_data_len() { + let config = StreamConfig::default(); // Largest possible message. - let message = [0; MAX_DATA_LEN]; + let message = vec![0; config.max_data_size()]; let protobuf = Message { flag: Some(Flag::Fin as i32), message: Some(message.to_vec()), }; - let mut codec = codec(); + let mut codec = codec(config); let mut dst = BytesMut::new(); codec.encode(protobuf, &mut dst).unwrap(); // Ensure the varint prefixed and protobuf encoded largest message is no longer than the // maximum limit specified in the libp2p WebRTC specification. - assert_eq!(dst.len(), MAX_MSG_LEN); + assert_eq!(dst.len(), config.max_message_size()); + } + + #[test] + fn effective_limit_is_the_smaller_advertised_limit() { + let local = StreamConfig::new(NonZeroUsize::new(16 * 1024).unwrap()); + let remote = NonZeroUsize::new(8 * 1024).unwrap(); + + assert_eq!(local.limited_by(remote).max_message_size(), remote.get()); } } diff --git a/misc/webrtc-utils/src/stream/framed_dc.rs b/misc/webrtc-utils/src/stream/framed_dc.rs index f0bbda84ca8..5e040416623 100644 --- a/misc/webrtc-utils/src/stream/framed_dc.rs +++ b/misc/webrtc-utils/src/stream/framed_dc.rs @@ -23,21 +23,21 @@ use futures::{AsyncRead, AsyncWrite}; use crate::{ proto::Message, - stream::{MAX_DATA_LEN, MAX_MSG_LEN, VARINT_LEN}, + stream::{StreamConfig, VARINT_LEN}, }; pub(crate) type FramedDc = Framed>; -pub(crate) fn new(inner: T) -> FramedDc +pub(crate) fn new(inner: T, config: StreamConfig) -> FramedDc where T: AsyncRead + AsyncWrite, { - let mut framed = Framed::new(inner, codec()); + let mut framed = Framed::new(inner, codec(config)); // If not set, `Framed` buffers up to 131kB of data before sending, which leads to "outbound // packet larger than maximum message size" error in webrtc-rs. - framed.set_send_high_water_mark(MAX_DATA_LEN); + framed.set_send_high_water_mark(config.max_data_size()); framed } -pub(crate) fn codec() -> prost_codec::Codec { - prost_codec::Codec::new(MAX_MSG_LEN - VARINT_LEN) +pub(crate) fn codec(config: StreamConfig) -> prost_codec::Codec { + prost_codec::Codec::new(config.max_message_size() - VARINT_LEN) } diff --git a/transports/webrtc-websys/src/connection.rs b/transports/webrtc-websys/src/connection.rs index acd84cfe313..794016eb347 100644 --- a/transports/webrtc-websys/src/connection.rs +++ b/transports/webrtc-websys/src/connection.rs @@ -8,7 +8,7 @@ use std::{ use futures::{StreamExt, channel::mpsc, stream::FuturesUnordered}; use js_sys::{Object, Reflect}; use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; -use libp2p_webrtc_utils::Fingerprint; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig}; use send_wrapper::SendWrapper; use wasm_bindgen::prelude::*; use wasm_bindgen_futures::JsFuture; @@ -38,13 +38,14 @@ pub struct Connection { /// A list of futures, which, once completed, signal that a [`Stream`] has been dropped. drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, + stream_config: StreamConfig, _ondatachannel_closure: SendWrapper>, } impl Connection { /// Create a new inner WebRTC Connection - pub(crate) fn new(peer_connection: RtcPeerConnection) -> Self { + pub(crate) fn new(peer_connection: RtcPeerConnection, stream_config: StreamConfig) -> Self { // An ondatachannel Future enables us to poll for incoming data channel events in // poll_incoming let (mut tx_ondatachannel, rx_ondatachannel) = mpsc::channel(4); // we may get more than one data channel opened on a single peer connection @@ -72,13 +73,14 @@ impl Connection { closed: false, drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, + stream_config, inbound_data_channels: SendWrapper::new(rx_ondatachannel), _ondatachannel_closure: SendWrapper::new(ondatachannel_closure), } } fn new_stream_from_data_channel(&mut self, data_channel: RtcDataChannel) -> Stream { - let (stream, drop_listener) = Stream::new(data_channel); + let (stream, drop_listener) = Stream::new(data_channel, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { @@ -204,8 +206,8 @@ impl RtcPeerConnection { /// Creates the stream for the initial noise handshake. /// /// The underlying data channel MUST have `negotiated` set to `true` and carry the ID 0. - pub(crate) fn new_handshake_stream(&self) -> (Stream, DropListener) { - Stream::new(self.new_data_channel(true)) + pub(crate) fn new_handshake_stream(&self, config: StreamConfig) -> (Stream, DropListener) { + Stream::new(self.new_data_channel(true), config) } /// Creates a regular data channel for when the connection is already established. diff --git a/transports/webrtc-websys/src/stream.rs b/transports/webrtc-websys/src/stream.rs index ee0183b07f0..ca444570fd5 100644 --- a/transports/webrtc-websys/src/stream.rs +++ b/transports/webrtc-websys/src/stream.rs @@ -5,6 +5,7 @@ use std::{ }; use futures::{AsyncRead, AsyncWrite}; +use libp2p_webrtc_utils::StreamConfig; use send_wrapper::SendWrapper; use web_sys::RtcDataChannel; @@ -23,9 +24,11 @@ pub struct Stream { pub(crate) type DropListener = SendWrapper>; impl Stream { - pub(crate) fn new(data_channel: RtcDataChannel) -> (Self, DropListener) { - let (inner, drop_listener) = - libp2p_webrtc_utils::Stream::new(PollDataChannel::new(data_channel)); + pub(crate) fn new(data_channel: RtcDataChannel, config: StreamConfig) -> (Self, DropListener) { + let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config( + PollDataChannel::new(data_channel, config), + config, + ); ( Self { diff --git a/transports/webrtc-websys/src/stream/poll_data_channel.rs b/transports/webrtc-websys/src/stream/poll_data_channel.rs index eaeec4d6789..07a14d6e135 100644 --- a/transports/webrtc-websys/src/stream/poll_data_channel.rs +++ b/transports/webrtc-websys/src/stream/poll_data_channel.rs @@ -12,7 +12,7 @@ use std::{ use bytes::BytesMut; use futures::{AsyncRead, AsyncWrite, task::AtomicWaker}; -use libp2p_webrtc_utils::MAX_MSG_LEN; +use libp2p_webrtc_utils::StreamConfig; use wasm_bindgen::prelude::*; use web_sys::{Event, MessageEvent, RtcDataChannel, RtcDataChannelEvent, RtcDataChannelState}; @@ -44,6 +44,7 @@ pub(crate) struct PollDataChannel { /// Failing these will (very likely), /// cause the application developer to drop the stream which resets it. overloaded: Rc, + max_message_size: usize, // Store the closures for proper garbage collection. // These are wrapped in an [`Rc`] so we can implement [`Clone`]. @@ -54,7 +55,7 @@ pub(crate) struct PollDataChannel { } impl PollDataChannel { - pub(crate) fn new(inner: RtcDataChannel) -> Self { + pub(crate) fn new(inner: RtcDataChannel, config: StreamConfig) -> Self { let open_waker = Rc::new(AtomicWaker::new()); let on_open_closure = Closure::new({ let open_waker = open_waker.clone(); @@ -105,7 +106,7 @@ impl PollDataChannel { let mut read_buffer = read_buffer.lock().unwrap(); - if read_buffer.len() + data.length() as usize > MAX_MSG_LEN { + if read_buffer.len() + data.length() as usize > config.max_message_size() { overloaded.store(true, Ordering::SeqCst); tracing::warn!("Remote is overloading us with messages, resetting stream",); return; @@ -125,6 +126,7 @@ impl PollDataChannel { write_waker, close_waker, overloaded, + max_message_size: config.max_message_size(), _on_open_closure: Rc::new(on_open_closure), _on_write_closure: Rc::new(on_write_closure), _on_close_closure: Rc::new(on_close_closure), @@ -207,8 +209,8 @@ impl AsyncWrite for PollDataChannel { futures::ready!(this.poll_ready(cx))?; - debug_assert!(this.buffered_amount() <= MAX_MSG_LEN); - let remaining_space = MAX_MSG_LEN - this.buffered_amount(); + debug_assert!(this.buffered_amount() <= this.max_message_size); + let remaining_space = this.max_message_size - this.buffered_amount(); if remaining_space == 0 { this.write_waker.register(cx.waker()); diff --git a/transports/webrtc-websys/src/transport.rs b/transports/webrtc-websys/src/transport.rs index 365216f35c2..8052467d7b6 100644 --- a/transports/webrtc-websys/src/transport.rs +++ b/transports/webrtc-websys/src/transport.rs @@ -1,5 +1,7 @@ +use libp2p_webrtc_utils::StreamConfig; use std::{ future::Future, + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -18,6 +20,7 @@ use super::{Connection, Error, upgrade}; #[derive(Clone)] pub struct Config { keypair: Keypair, + stream_config: StreamConfig, } /// A WebTransport [`Transport`](libp2p_core::Transport) that works with `web-sys`. @@ -30,8 +33,15 @@ impl Config { pub fn new(keypair: &Keypair) -> Self { Config { keypair: keypair.to_owned(), + stream_config: StreamConfig::default(), } } + + /// Limits encoded WebRTC data-channel messages for connections created by this transport. + pub fn with_max_message_size(mut self, max_message_size: NonZeroUsize) -> Self { + self.stream_config = StreamConfig::new(max_message_size); + self + } } impl Transport { @@ -93,8 +103,13 @@ impl libp2p_core::Transport for Transport { let config = self.config.clone(); Ok(async move { - let (peer_id, connection) = - upgrade::outbound(sock_addr, server_fingerprint, config.keypair.clone()).await?; + let (peer_id, connection) = upgrade::outbound( + sock_addr, + server_fingerprint, + config.keypair.clone(), + config.stream_config, + ) + .await?; Ok((peer_id, connection)) } diff --git a/transports/webrtc-websys/src/upgrade.rs b/transports/webrtc-websys/src/upgrade.rs index d1e322a0838..1702e2d037d 100644 --- a/transports/webrtc-websys/src/upgrade.rs +++ b/transports/webrtc-websys/src/upgrade.rs @@ -1,7 +1,7 @@ use std::net::SocketAddr; use libp2p_identity::{Keypair, PeerId}; -use libp2p_webrtc_utils::{Fingerprint, noise}; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; use send_wrapper::SendWrapper; use super::Error; @@ -13,8 +13,14 @@ pub(crate) async fn outbound( sock_addr: SocketAddr, remote_fingerprint: Fingerprint, id_keys: Keypair, + stream_config: StreamConfig, ) -> Result<(PeerId, Connection), Error> { - let fut = SendWrapper::new(outbound_inner(sock_addr, remote_fingerprint, id_keys)); + let fut = SendWrapper::new(outbound_inner( + sock_addr, + remote_fingerprint, + id_keys, + stream_config, + )); fut.await } @@ -23,12 +29,13 @@ async fn outbound_inner( sock_addr: SocketAddr, remote_fingerprint: Fingerprint, id_keys: Keypair, + stream_config: StreamConfig, ) -> Result<(PeerId, Connection), Error> { let rtc_peer_connection = RtcPeerConnection::new(remote_fingerprint.algorithm()).await?; // Create stream for Noise handshake // Must create data channel before Offer is created for it to be included in the SDP - let (channel, listener) = rtc_peer_connection.new_handshake_stream(); + let (channel, listener) = rtc_peer_connection.new_handshake_stream(stream_config); drop(listener); let ufrag = libp2p_webrtc_utils::sdp::random_ufrag(); @@ -47,11 +54,17 @@ async fn outbound_inner( tracing::trace!(?local_fingerprint); tracing::trace!(?remote_fingerprint); - let peer_id = noise::outbound(id_keys, channel, remote_fingerprint, local_fingerprint) - .await - .map_err(AuthenticationError)?; + let (peer_id, stream_config) = noise::outbound_with_message_size( + id_keys, + channel, + remote_fingerprint, + local_fingerprint, + stream_config, + ) + .await + .map_err(AuthenticationError)?; tracing::debug!(peer=%peer_id, "Remote peer identified"); - Ok((peer_id, Connection::new(rtc_peer_connection))) + Ok((peer_id, Connection::new(rtc_peer_connection, stream_config))) } diff --git a/transports/webrtc/src/tokio/connection.rs b/transports/webrtc/src/tokio/connection.rs index 3009e7ed33c..03062a6ad7e 100644 --- a/transports/webrtc/src/tokio/connection.rs +++ b/transports/webrtc/src/tokio/connection.rs @@ -36,6 +36,7 @@ use futures::{ stream::FuturesUnordered, }; use libp2p_core::muxing::{StreamMuxer, StreamMuxerEvent}; +use libp2p_webrtc_utils::StreamConfig; use webrtc::{ data::data_channel::DataChannel as DetachedDataChannel, data_channel::RTCDataChannel, peer_connection::RTCPeerConnection, @@ -66,13 +67,14 @@ pub struct Connection { /// A list of futures, which, once completed, signal that a [`Stream`] has been dropped. drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, + stream_config: StreamConfig, } impl Unpin for Connection {} impl Connection { /// Creates a new connection. - pub(crate) async fn new(rtc_conn: RTCPeerConnection) -> Self { + pub(crate) async fn new(rtc_conn: RTCPeerConnection, stream_config: StreamConfig) -> Self { let (data_channel_tx, data_channel_rx) = mpsc::channel(MAX_DATA_CHANNELS_IN_FLIGHT); Connection::register_incoming_data_channels_handler( @@ -88,6 +90,7 @@ impl Connection { close_fut: None, drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, + stream_config, } } @@ -159,7 +162,7 @@ impl StreamMuxer for Connection { Some(detached) => { tracing::trace!(stream=%detached.stream_identifier(), "Incoming stream"); - let (stream, drop_listener) = Stream::new(detached); + let (stream, drop_listener) = Stream::new(detached, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { waker.wake() @@ -233,7 +236,7 @@ impl StreamMuxer for Connection { tracing::trace!(stream=%detached.stream_identifier(), "Outbound stream"); - let (stream, drop_listener) = Stream::new(detached); + let (stream, drop_listener) = Stream::new(detached, self.stream_config); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { waker.wake() diff --git a/transports/webrtc/src/tokio/sdp.rs b/transports/webrtc/src/tokio/sdp.rs index f28c5c33105..8d233b34ef5 100644 --- a/transports/webrtc/src/tokio/sdp.rs +++ b/transports/webrtc/src/tokio/sdp.rs @@ -140,5 +140,5 @@ a=ice-pwd:{pwd} a=fingerprint:{fingerprint_algorithm} {fingerprint_value} a=setup:actpass a=sctp-port:5000 -a=max-message-size:16384 +a=max-message-size:{max_message_size} "; diff --git a/transports/webrtc/src/tokio/stream.rs b/transports/webrtc/src/tokio/stream.rs index 9d5a9faf440..3b98ab79bac 100644 --- a/transports/webrtc/src/tokio/stream.rs +++ b/transports/webrtc/src/tokio/stream.rs @@ -25,7 +25,7 @@ use std::{ }; use futures::prelude::*; -use libp2p_webrtc_utils::MAX_MSG_LEN; +use libp2p_webrtc_utils::StreamConfig; use tokio_util::compat::{Compat, TokioAsyncReadCompatExt}; use webrtc::data::data_channel::{DataChannel, PollDataChannel}; @@ -42,11 +42,16 @@ pub(crate) type DropListener = libp2p_webrtc_utils::DropListener) -> (Self, DropListener) { + pub(crate) fn new( + data_channel: Arc, + config: StreamConfig, + ) -> (Self, DropListener) { let mut data_channel = PollDataChannel::new(data_channel).compat(); - data_channel.get_mut().set_read_buf_capacity(MAX_MSG_LEN); + data_channel + .get_mut() + .set_read_buf_capacity(config.max_message_size()); - let (inner, drop_listener) = libp2p_webrtc_utils::Stream::new(data_channel); + let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config(data_channel, config); (Self { inner }, drop_listener) } diff --git a/transports/webrtc/src/tokio/transport.rs b/transports/webrtc/src/tokio/transport.rs index d5857a42c11..f6eca7ca2ce 100644 --- a/transports/webrtc/src/tokio/transport.rs +++ b/transports/webrtc/src/tokio/transport.rs @@ -18,9 +18,11 @@ // FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER // DEALINGS IN THE SOFTWARE. +use libp2p_webrtc_utils::StreamConfig; use std::{ io, net::{IpAddr, SocketAddr}, + num::NonZeroUsize, pin::Pin, task::{Context, Poll, Waker}, }; @@ -71,6 +73,12 @@ impl Transport { listeners: SelectAll::new(), } } + + /// Limits encoded WebRTC data-channel messages for connections created by this transport. + pub fn with_max_message_size(mut self, max_message_size: NonZeroUsize) -> Self { + self.config.stream_config = StreamConfig::new(max_message_size); + self + } } impl libp2p_core::Transport for Transport { @@ -150,6 +158,7 @@ impl libp2p_core::Transport for Transport { let (peer_id, connection) = upgrade::outbound( sock_addr, config.inner, + config.stream_config, udp_mux, client_fingerprint.into_inner(), server_fingerprint, @@ -331,6 +340,7 @@ impl Stream for ListenStream { let upgrade = upgrade::inbound( new_addr.addr, self.config.inner.clone(), + self.config.stream_config, self.udp_mux.udp_mux_handle(), self.config.fingerprint.into_inner(), new_addr.ufrag, @@ -365,6 +375,7 @@ struct Config { inner: RTCConfiguration, fingerprint: Fingerprint, id_keys: identity::Keypair, + stream_config: StreamConfig, } impl Config { @@ -379,6 +390,7 @@ impl Config { ..RTCConfiguration::default() }, fingerprint, + stream_config: StreamConfig::default(), } } } diff --git a/transports/webrtc/src/tokio/upgrade.rs b/transports/webrtc/src/tokio/upgrade.rs index 5333dda5c94..f2cb1b73e00 100644 --- a/transports/webrtc/src/tokio/upgrade.rs +++ b/transports/webrtc/src/tokio/upgrade.rs @@ -31,7 +31,7 @@ use futures::{channel::oneshot, future::Either}; use futures_timer::Delay; use libp2p_identity as identity; use libp2p_identity::PeerId; -use libp2p_webrtc_utils::{Fingerprint, noise}; +use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; use webrtc::{ api::{APIBuilder, setting_engine::SettingEngine}, data::data_channel::DataChannel, @@ -48,6 +48,7 @@ use crate::tokio::{Connection, error::Error, sdp, sdp::random_ufrag, stream::Str pub(crate) async fn outbound( addr: SocketAddr, config: RTCConfiguration, + stream_config: StreamConfig, udp_mux: Arc, client_fingerprint: Fingerprint, server_fingerprint: Fingerprint, @@ -67,15 +68,19 @@ pub(crate) async fn outbound( peer_connection.set_remote_description(answer).await?; // This will start the gathering of ICE candidates. let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; - let peer_id = noise::outbound( + let (peer_id, stream_config) = noise::outbound_with_message_size( id_keys, data_channel, server_fingerprint, client_fingerprint, + stream_config, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok(( + peer_id, + Connection::new(peer_connection, stream_config).await, + )) } /// Creates a new inbound WebRTC connection. @@ -83,6 +88,7 @@ pub(crate) async fn outbound( pub(crate) async fn inbound( addr: SocketAddr, config: RTCConfiguration, + stream_config: StreamConfig, udp_mux: Arc, server_fingerprint: Fingerprint, remote_ufrag: String, @@ -103,15 +109,19 @@ pub(crate) async fn inbound( let data_channel = await_noise_data_channel_open(noise_channel_open_rx).await?; let client_fingerprint = get_remote_fingerprint(&peer_connection).await; - let peer_id = noise::inbound( + let (peer_id, stream_config) = noise::inbound_with_message_size( id_keys, data_channel, client_fingerprint, server_fingerprint, + stream_config, ) .await?; - Ok((peer_id, Connection::new(peer_connection).await)) + Ok(( + peer_id, + Connection::new(peer_connection, stream_config).await, + )) } #[allow(clippy::result_large_err)] @@ -248,7 +258,7 @@ async fn await_noise_data_channel_open( } }; - let (substream, drop_listener) = Stream::new(channel); + let (substream, drop_listener) = Stream::new(channel, StreamConfig::default()); drop(drop_listener); // Don't care about cancelled substreams during initial handshake. Ok(substream) From 9e3bcd9b2b30ebc33a1d35c10bd64e3ff14dffd5 Mon Sep 17 00:00:00 2001 From: yexiyue Date: Fri, 24 Jul 2026 23:25:53 +0800 Subject: [PATCH 2/3] docs: add WebRTC message limit changelogs --- misc/webrtc-utils/CHANGELOG.md | 2 ++ transports/webrtc-websys/CHANGELOG.md | 2 ++ transports/webrtc/CHANGELOG.md | 2 ++ 3 files changed, 6 insertions(+) diff --git a/misc/webrtc-utils/CHANGELOG.md b/misc/webrtc-utils/CHANGELOG.md index 600101a55e7..12ca527e4c0 100644 --- a/misc/webrtc-utils/CHANGELOG.md +++ b/misc/webrtc-utils/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.5.0 +- Negotiate WebRTC data-channel message limits after Noise authentication. + - Revert migration to `quick-protobuf`, migrate back to `prost`. See [PR 6363](https://github.com/libp2p/rust-libp2p/pull/6363). diff --git a/transports/webrtc-websys/CHANGELOG.md b/transports/webrtc-websys/CHANGELOG.md index 4e9f05ae892..ca9bbf39d99 100644 --- a/transports/webrtc-websys/CHANGELOG.md +++ b/transports/webrtc-websys/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.5.0 +- Add `Config::with_max_message_size` to configure the WebRTC data-channel framing limit. + - Require `getrandom/js` feature only under `wasm` target. See [PR 6102](https://github.com/libp2p/rust-libp2p/pull/6102) diff --git a/transports/webrtc/CHANGELOG.md b/transports/webrtc/CHANGELOG.md index b090ebd3efe..d22446f89be 100644 --- a/transports/webrtc/CHANGELOG.md +++ b/transports/webrtc/CHANGELOG.md @@ -1,5 +1,7 @@ ## 0.10.0-alpha +- Add `Transport::with_max_message_size` to configure the WebRTC data-channel framing limit. + - Update webrtc-rs to `v0.17` and fix libp2p noise data channel negotiation. See [PR 6429](https://github.com/libp2p/rust-libp2p/pull/6429) From 5984c716b5cd0d81c2ca4dc71e222d7b700b888d Mon Sep 17 00:00:00 2001 From: yexiyue Date: Sat, 25 Jul 2026 11:27:49 +0800 Subject: [PATCH 3/3] fix(webrtc-websys): configure receive buffer limit --- transports/webrtc-websys/CHANGELOG.md | 1 + transports/webrtc-websys/src/connection.rs | 20 ++++- transports/webrtc-websys/src/stream.rs | 9 ++- .../src/stream/poll_data_channel.rs | 73 ++++++++++++++++++- transports/webrtc-websys/src/transport.rs | 16 ++++ transports/webrtc-websys/src/upgrade.rs | 13 +++- 6 files changed, 120 insertions(+), 12 deletions(-) diff --git a/transports/webrtc-websys/CHANGELOG.md b/transports/webrtc-websys/CHANGELOG.md index ca9bbf39d99..c2c4b6dbfa4 100644 --- a/transports/webrtc-websys/CHANGELOG.md +++ b/transports/webrtc-websys/CHANGELOG.md @@ -1,6 +1,7 @@ ## 0.5.0 - Add `Config::with_max_message_size` to configure the WebRTC data-channel framing limit. +- Add `Config::with_max_read_buffer_size` to configure the browser callback receive buffer. - Require `getrandom/js` feature only under `wasm` target. See [PR 6102](https://github.com/libp2p/rust-libp2p/pull/6102) diff --git a/transports/webrtc-websys/src/connection.rs b/transports/webrtc-websys/src/connection.rs index 794016eb347..fa0854bf90c 100644 --- a/transports/webrtc-websys/src/connection.rs +++ b/transports/webrtc-websys/src/connection.rs @@ -1,6 +1,7 @@ //! A libp2p connection backed by an [RtcPeerConnection](https://developer.mozilla.org/en-US/docs/Web/API/RTCPeerConnection). use std::{ + num::NonZeroUsize, pin::Pin, task::{Context, Poll, Waker, ready}, }; @@ -39,13 +40,18 @@ pub struct Connection { drop_listeners: FuturesUnordered, no_drop_listeners_waker: Option, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, _ondatachannel_closure: SendWrapper>, } impl Connection { /// Create a new inner WebRTC Connection - pub(crate) fn new(peer_connection: RtcPeerConnection, stream_config: StreamConfig) -> Self { + pub(crate) fn new( + peer_connection: RtcPeerConnection, + stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> Self { // An ondatachannel Future enables us to poll for incoming data channel events in // poll_incoming let (mut tx_ondatachannel, rx_ondatachannel) = mpsc::channel(4); // we may get more than one data channel opened on a single peer connection @@ -74,13 +80,15 @@ impl Connection { drop_listeners: FuturesUnordered::default(), no_drop_listeners_waker: None, stream_config, + max_read_buffer_size, inbound_data_channels: SendWrapper::new(rx_ondatachannel), _ondatachannel_closure: SendWrapper::new(ondatachannel_closure), } } fn new_stream_from_data_channel(&mut self, data_channel: RtcDataChannel) -> Stream { - let (stream, drop_listener) = Stream::new(data_channel, self.stream_config); + let (stream, drop_listener) = + Stream::new(data_channel, self.stream_config, self.max_read_buffer_size); self.drop_listeners.push(drop_listener); if let Some(waker) = self.no_drop_listeners_waker.take() { @@ -206,8 +214,12 @@ impl RtcPeerConnection { /// Creates the stream for the initial noise handshake. /// /// The underlying data channel MUST have `negotiated` set to `true` and carry the ID 0. - pub(crate) fn new_handshake_stream(&self, config: StreamConfig) -> (Stream, DropListener) { - Stream::new(self.new_data_channel(true), config) + pub(crate) fn new_handshake_stream( + &self, + config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> (Stream, DropListener) { + Stream::new(self.new_data_channel(true), config, max_read_buffer_size) } /// Creates a regular data channel for when the connection is already established. diff --git a/transports/webrtc-websys/src/stream.rs b/transports/webrtc-websys/src/stream.rs index ca444570fd5..dc0c28bb0a1 100644 --- a/transports/webrtc-websys/src/stream.rs +++ b/transports/webrtc-websys/src/stream.rs @@ -1,5 +1,6 @@ //! The WebRTC [Stream] over the Connection use std::{ + num::NonZeroUsize, pin::Pin, task::{Context, Poll}, }; @@ -24,9 +25,13 @@ pub struct Stream { pub(crate) type DropListener = SendWrapper>; impl Stream { - pub(crate) fn new(data_channel: RtcDataChannel, config: StreamConfig) -> (Self, DropListener) { + pub(crate) fn new( + data_channel: RtcDataChannel, + config: StreamConfig, + max_read_buffer_size: NonZeroUsize, + ) -> (Self, DropListener) { let (inner, drop_listener) = libp2p_webrtc_utils::Stream::with_config( - PollDataChannel::new(data_channel, config), + PollDataChannel::new(data_channel, config, max_read_buffer_size), config, ); diff --git a/transports/webrtc-websys/src/stream/poll_data_channel.rs b/transports/webrtc-websys/src/stream/poll_data_channel.rs index 07a14d6e135..80de9212735 100644 --- a/transports/webrtc-websys/src/stream/poll_data_channel.rs +++ b/transports/webrtc-websys/src/stream/poll_data_channel.rs @@ -1,6 +1,7 @@ use std::{ - cmp::min, + cmp::{max, min}, io, + num::NonZeroUsize, pin::Pin, rc::Rc, sync::{ @@ -16,6 +17,14 @@ use libp2p_webrtc_utils::StreamConfig; use wasm_bindgen::prelude::*; use web_sys::{Event, MessageEvent, RtcDataChannel, RtcDataChannelEvent, RtcDataChannelState}; +fn effective_max_read_buffer_size(config: StreamConfig, configured: NonZeroUsize) -> usize { + max(configured.get(), config.max_message_size()) +} + +fn read_buffer_overflows(buffered: usize, incoming: usize, max_read_buffer_size: usize) -> bool { + buffered.saturating_add(incoming) > max_read_buffer_size +} + /// [`PollDataChannel`] is a wrapper around [`RtcDataChannel`] which implements [`AsyncRead`] and /// [`AsyncWrite`]. #[derive(Debug, Clone)] @@ -55,7 +64,13 @@ pub(crate) struct PollDataChannel { } impl PollDataChannel { - pub(crate) fn new(inner: RtcDataChannel, config: StreamConfig) -> Self { + pub(crate) fn new( + inner: RtcDataChannel, + config: StreamConfig, + configured_max_read_buffer_size: NonZeroUsize, + ) -> Self { + let max_read_buffer_size = + effective_max_read_buffer_size(config, configured_max_read_buffer_size); let open_waker = Rc::new(AtomicWaker::new()); let on_open_closure = Closure::new({ let open_waker = open_waker.clone(); @@ -106,7 +121,11 @@ impl PollDataChannel { let mut read_buffer = read_buffer.lock().unwrap(); - if read_buffer.len() + data.length() as usize > config.max_message_size() { + if read_buffer_overflows( + read_buffer.len(), + data.length() as usize, + max_read_buffer_size, + ) { overloaded.store(true, Ordering::SeqCst); tracing::warn!("Remote is overloading us with messages, resetting stream",); return; @@ -169,6 +188,54 @@ impl PollDataChannel { } } +#[cfg(test)] +mod tests { + use std::num::NonZeroUsize; + + use libp2p_webrtc_utils::StreamConfig; + + use super::{effective_max_read_buffer_size, read_buffer_overflows}; + + #[test] + fn read_buffer_accepts_multiple_valid_messages() { + let message_size = NonZeroUsize::new(8 * 1024).unwrap(); + let max_read_buffer_size = effective_max_read_buffer_size( + StreamConfig::new(message_size), + NonZeroUsize::new(256 * 1024).unwrap(), + ); + + assert!(!read_buffer_overflows( + message_size.get(), + message_size.get(), + max_read_buffer_size, + )); + } + + #[test] + fn configured_read_buffer_remains_bounded() { + let max_read_buffer_size = 16 * 1024; + + assert!(read_buffer_overflows( + max_read_buffer_size, + 1, + max_read_buffer_size, + )); + } + + #[test] + fn read_buffer_can_hold_one_larger_valid_message() { + let message_size = NonZeroUsize::new(512 * 1024).unwrap(); + + assert_eq!( + effective_max_read_buffer_size( + StreamConfig::new(message_size), + NonZeroUsize::new(256 * 1024).unwrap(), + ), + message_size.get(), + ); + } +} + impl AsyncRead for PollDataChannel { fn poll_read( self: Pin<&mut Self>, diff --git a/transports/webrtc-websys/src/transport.rs b/transports/webrtc-websys/src/transport.rs index 8052467d7b6..ec409d0469a 100644 --- a/transports/webrtc-websys/src/transport.rs +++ b/transports/webrtc-websys/src/transport.rs @@ -16,11 +16,16 @@ use libp2p_identity::{Keypair, PeerId}; use super::{Connection, Error, upgrade}; +/// Default maximum number of bytes browser callbacks may queue before the Rust task polls them. +const DEFAULT_MAX_READ_BUFFER_SIZE: NonZeroUsize = + NonZeroUsize::new(256 * 1024).expect("constant is non-zero"); + /// Config for the [`Transport`]. #[derive(Clone)] pub struct Config { keypair: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, } /// A WebTransport [`Transport`](libp2p_core::Transport) that works with `web-sys`. @@ -34,6 +39,7 @@ impl Config { Config { keypair: keypair.to_owned(), stream_config: StreamConfig::default(), + max_read_buffer_size: DEFAULT_MAX_READ_BUFFER_SIZE, } } @@ -42,6 +48,15 @@ impl Config { self.stream_config = StreamConfig::new(max_message_size); self } + + /// Limits bytes queued by browser data-channel callbacks before the stream is polled. + /// + /// The effective limit is raised as needed to hold one valid encoded message. This limit is + /// local to the browser transport and is not negotiated with the remote peer. + pub fn with_max_read_buffer_size(mut self, max_read_buffer_size: NonZeroUsize) -> Self { + self.max_read_buffer_size = max_read_buffer_size; + self + } } impl Transport { @@ -108,6 +123,7 @@ impl libp2p_core::Transport for Transport { server_fingerprint, config.keypair.clone(), config.stream_config, + config.max_read_buffer_size, ) .await?; diff --git a/transports/webrtc-websys/src/upgrade.rs b/transports/webrtc-websys/src/upgrade.rs index 1702e2d037d..abf5165b128 100644 --- a/transports/webrtc-websys/src/upgrade.rs +++ b/transports/webrtc-websys/src/upgrade.rs @@ -1,4 +1,4 @@ -use std::net::SocketAddr; +use std::{net::SocketAddr, num::NonZeroUsize}; use libp2p_identity::{Keypair, PeerId}; use libp2p_webrtc_utils::{Fingerprint, StreamConfig, noise}; @@ -14,12 +14,14 @@ pub(crate) async fn outbound( remote_fingerprint: Fingerprint, id_keys: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, ) -> Result<(PeerId, Connection), Error> { let fut = SendWrapper::new(outbound_inner( sock_addr, remote_fingerprint, id_keys, stream_config, + max_read_buffer_size, )); fut.await } @@ -30,12 +32,14 @@ async fn outbound_inner( remote_fingerprint: Fingerprint, id_keys: Keypair, stream_config: StreamConfig, + max_read_buffer_size: NonZeroUsize, ) -> Result<(PeerId, Connection), Error> { let rtc_peer_connection = RtcPeerConnection::new(remote_fingerprint.algorithm()).await?; // Create stream for Noise handshake // Must create data channel before Offer is created for it to be included in the SDP - let (channel, listener) = rtc_peer_connection.new_handshake_stream(stream_config); + let (channel, listener) = + rtc_peer_connection.new_handshake_stream(stream_config, max_read_buffer_size); drop(listener); let ufrag = libp2p_webrtc_utils::sdp::random_ufrag(); @@ -66,5 +70,8 @@ async fn outbound_inner( tracing::debug!(peer=%peer_id, "Remote peer identified"); - Ok((peer_id, Connection::new(rtc_peer_connection, stream_config))) + Ok(( + peer_id, + Connection::new(rtc_peer_connection, stream_config, max_read_buffer_size), + )) }