From 06fd6a1ce98d99647f6f97864aed24bc738378e3 Mon Sep 17 00:00:00 2001 From: Lewis Date: Tue, 9 Jun 2026 22:10:33 +0300 Subject: [PATCH] ripple: transport from tcp to quic Lewis: May this revision serve well! --- Cargo.lock | 5 +- crates/tranquil-ripple/Cargo.toml | 5 +- crates/tranquil-ripple/src/engine.rs | 2 +- crates/tranquil-ripple/src/gossip.rs | 1 - crates/tranquil-ripple/src/transport.rs | 769 +++++++++++++----------- 5 files changed, 438 insertions(+), 344 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index d4e9cbf..c842fc3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7998,14 +7998,15 @@ dependencies = [ "async-trait", "backon", "bincode 2.0.1", - "bytes", "foca", "futures", "metrics", "parking_lot", + "quinn", "rand 0.9.2", + "rcgen", + "rustls 0.23.37", "serde", - "socket2 0.6.3", "thiserror 2.0.18", "tokio", "tokio-util", diff --git a/crates/tranquil-ripple/Cargo.toml b/crates/tranquil-ripple/Cargo.toml index 79601f6..7181d20 100644 --- a/crates/tranquil-ripple/Cargo.toml +++ b/crates/tranquil-ripple/Cargo.toml @@ -11,13 +11,14 @@ tranquil-infra = { workspace = true } async-trait = { workspace = true } backon = { workspace = true } bincode = { workspace = true } -bytes = { workspace = true } foca = { workspace = true } metrics = { workspace = true } parking_lot = { workspace = true } +quinn = { workspace = true } rand = "0.9" +rcgen = { workspace = true } +rustls = { workspace = true } serde = { workspace = true } -socket2 = "0.6.2" thiserror = { workspace = true } tokio = { workspace = true, features = ["net", "io-util", "sync", "time"] } tokio-util = { workspace = true } diff --git a/crates/tranquil-ripple/src/engine.rs b/crates/tranquil-ripple/src/engine.rs index ade193d..fd404e5 100644 --- a/crates/tranquil-ripple/src/engine.rs +++ b/crates/tranquil-ripple/src/engine.rs @@ -22,7 +22,7 @@ impl RippleEngine { let store = Arc::new(ShardedCrdtStore::new(config.machine_id)); let (transport, incoming_rx) = - Transport::bind(config.bind_addr, config.machine_id, shutdown.clone()) + Transport::bind(config.bind_addr, shutdown.clone()) .await .map_err(|e| RippleStartError::Bind(e.to_string()))?; diff --git a/crates/tranquil-ripple/src/gossip.rs b/crates/tranquil-ripple/src/gossip.rs index defb2a0..2149051 100644 --- a/crates/tranquil-ripple/src/gossip.rs +++ b/crates/tranquil-ripple/src/gossip.rs @@ -265,7 +265,6 @@ impl GossipEngine { } } } - _ => {} } } _ = gossip_tick.tick() => { diff --git a/crates/tranquil-ripple/src/transport.rs b/crates/tranquil-ripple/src/transport.rs index 938f490..db4eee0 100644 --- a/crates/tranquil-ripple/src/transport.rs +++ b/crates/tranquil-ripple/src/transport.rs @@ -1,27 +1,44 @@ use backon::{ExponentialBuilder, Retryable}; -use bytes::{Buf, BufMut, BytesMut}; +use quinn::crypto::rustls::{QuicClientConfig, QuicServerConfig}; +use quinn::{ + ClientConfig, Connection, Endpoint, IdleTimeout, RecvStream, ServerConfig, TransportConfig, + VarInt, +}; +use rustls::pki_types::{CertificateDer, PrivateKeyDer, PrivatePkcs8KeyDer}; use std::collections::HashMap; use std::net::SocketAddr; use std::sync::Arc; -use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering}; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; -use tokio::io::{AsyncReadExt, AsyncWriteExt}; -use tokio::net::{TcpListener, TcpStream}; -use tokio::sync::mpsc; +use tokio::sync::{Semaphore, mpsc}; use tokio_util::sync::CancellationToken; -pub(crate) const MAX_FRAME_SIZE: usize = 16 * 1024 * 1024; +pub(crate) const MAX_FRAME_SIZE: usize = 4 * 1024 * 1024; const MAX_INBOUND_CONNECTIONS: usize = 512; const MAX_OUTBOUND_CONNECTIONS: usize = 512; +const MAX_CONCURRENT_UNI_STREAMS: u32 = 64; +const READ_CHUNK_BYTES: usize = 256 * 1024; +const INCOMING_CHANNEL_DEPTH: usize = 1024; +const STREAM_RECEIVE_WINDOW: u32 = MAX_FRAME_SIZE as u32; +const CONNECTION_RECEIVE_WINDOW: u32 = 16 * 1024 * 1024; +const KEEPALIVE: Duration = Duration::from_secs(20); +const IDLE_TIMEOUT: Duration = Duration::from_secs(60); +const CONNECT_TIMEOUT: Duration = Duration::from_secs(5); const WRITE_TIMEOUT: Duration = Duration::from_secs(10); +const READ_TIMEOUT: Duration = Duration::from_secs(30); +const RIPPLE_ALPN: &[u8] = b"ripple/1"; +const RIPPLE_SERVER_NAME: &str = "ripple"; + +struct NodeIdentity { + cert: CertificateDer<'static>, + key: PrivateKeyDer<'static>, +} #[derive(Debug, Clone, Copy, PartialEq, Eq)] #[repr(u8)] pub enum ChannelTag { Gossip = 0x01, CrdtSync = 0x02, - Raft = 0x03, - Direct = 0x04, } impl ChannelTag { @@ -29,8 +46,6 @@ impl ChannelTag { match v { 0x01 => Some(Self::Gossip), 0x02 => Some(Self::CrdtSync), - 0x03 => Some(Self::Raft), - 0x04 => Some(Self::Direct), _ => None, } } @@ -43,20 +58,18 @@ pub struct IncomingFrame { pub data: Vec, } -struct ConnectionWriter { - tx: mpsc::Sender>, +struct PeerConn { + conn: Connection, generation: u64, } pub struct Transport { + endpoint: Endpoint, local_addr: SocketAddr, - _machine_id: u64, - connections: Arc>>, + connections: Arc>>, connecting: Arc>>, conn_generation: Arc, - #[allow(dead_code)] - inbound_count: Arc, - outbound_count: Arc, + outbound_permits: Arc, shutdown: CancellationToken, incoming_tx: mpsc::Sender, } @@ -64,67 +77,70 @@ pub struct Transport { impl Transport { pub async fn bind( addr: SocketAddr, - machine_id: u64, shutdown: CancellationToken, ) -> Result<(Self, mpsc::Receiver), std::io::Error> { - let listener = TcpListener::bind(addr).await?; - let local_addr = listener.local_addr()?; - let (incoming_tx, incoming_rx) = mpsc::channel(4096); - let inbound_count = Arc::new(AtomicUsize::new(0)); + let server_config = build_server_config() + .map_err(|e| std::io::Error::other(format!("ripple server config: {e}")))?; + let client_config = build_client_config() + .map_err(|e| std::io::Error::other(format!("ripple client config: {e}")))?; + + let mut endpoint = Endpoint::server(server_config, addr)?; + endpoint.set_default_client_config(client_config); + let local_addr = endpoint.local_addr()?; + let (incoming_tx, incoming_rx) = mpsc::channel(INCOMING_CHANNEL_DEPTH); let transport = Self { + endpoint: endpoint.clone(), local_addr, - _machine_id: machine_id, connections: Arc::new(parking_lot::Mutex::new(HashMap::new())), connecting: Arc::new(parking_lot::Mutex::new(std::collections::HashSet::new())), conn_generation: Arc::new(AtomicU64::new(0)), - inbound_count: inbound_count.clone(), - outbound_count: Arc::new(AtomicUsize::new(0)), + outbound_permits: Arc::new(Semaphore::new(MAX_OUTBOUND_CONNECTIONS)), shutdown: shutdown.clone(), incoming_tx: incoming_tx.clone(), }; + let inbound_permits = Arc::new(Semaphore::new(MAX_INBOUND_CONNECTIONS)); let cancel = shutdown.clone(); - let inbound_counter = inbound_count.clone(); tokio::spawn(async move { loop { tokio::select! { _ = cancel.cancelled() => break, - result = listener.accept() => { - match result { - Ok((stream, peer_addr)) => { - let current = inbound_counter.load(Ordering::Relaxed); - if current >= MAX_INBOUND_CONNECTIONS { - tracing::warn!( - peer = %peer_addr, - count = current, - max = MAX_INBOUND_CONNECTIONS, - "rejecting inbound connection: limit reached" - ); - drop(stream); - continue; - } - inbound_counter.fetch_add(1, Ordering::Relaxed); - configure_socket(&stream); - Self::spawn_reader( - stream, - peer_addr, - incoming_tx.clone(), - cancel.clone(), - inbound_counter.clone(), - ); - tracing::debug!(peer = %peer_addr, "accepted inbound connection"); - } - Err(e) => { - tracing::warn!(error = %e, "accept failed"); - } + incoming = endpoint.accept() => { + let Some(incoming) = incoming else { break }; + if !incoming.remote_address_validated() { + let _ = incoming.retry(); + continue; } + let Ok(permit) = inbound_permits.clone().try_acquire_owned() else { + tracing::warn!( + peer = %incoming.remote_address(), + max = MAX_INBOUND_CONNECTIONS, + "rejecting inbound connection: limit reached" + ); + incoming.refuse(); + continue; + }; + let tx = incoming_tx.clone(); + let conn_cancel = cancel.child_token(); + tokio::spawn(async move { + let _permit = permit; + match incoming.await { + Ok(conn) => { + let from = conn.remote_address(); + tracing::debug!(peer = %from, "accepted inbound connection"); + run_conn_reader(conn, from, tx, conn_cancel).await; + } + Err(e) => tracing::warn!(error = %e, "inbound handshake failed"), + } + }); } } } + endpoint.close(0u32.into(), b"shutdown"); }); - tracing::info!(addr = %local_addr, "ripple transport bound"); + tracing::info!(addr = %local_addr, "ripple quic transport bound"); Ok((transport, incoming_rx)) } @@ -133,45 +149,42 @@ impl Transport { } pub fn try_queue(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) -> bool { - let frame = match encode_frame(tag, data) { - Some(f) => f, - None => return false, - }; - let conns = self.connections.lock(); - match conns.get(&target) { - Some(writer) => writer.tx.try_send(frame).is_ok(), - None => false, - } + let conn = self.connections.lock().get(&target).map(|p| p.conn.clone()); + let Some(conn) = conn else { return false }; + let data = data.to_vec(); + tokio::spawn(async move { + if let Err(e) = write_frame(&conn, tag, &data).await { + tracing::debug!(error = %e, "queued write failed"); + } + }); + true } pub async fn send(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) { - let frame = match encode_frame(tag, data) { - Some(f) => f, - None => return, - }; - let writer = { - let conns = self.connections.lock(); - conns.get(&target).map(|w| (w.tx.clone(), w.generation)) - }; - match writer { - Some((tx, acquired_gen)) => { - if tx.send(frame).await.is_err() { + let existing = self + .connections + .lock() + .get(&target) + .map(|p| (p.conn.clone(), p.generation)); + if let Some((conn, generation)) = existing { + match write_frame(&conn, tag, data).await { + Ok(()) => return, + Err(e) if e.kind() == std::io::ErrorKind::TimedOut => { + tracing::debug!(peer = %target, "write timed out, keeping connection"); + return; + } + Err(_) => { + let mut conns = self.connections.lock(); + if conns + .get(&target) + .is_some_and(|p| p.generation == generation) { - let mut conns = self.connections.lock(); - let stale = conns - .get(&target) - .is_some_and(|w| w.generation == acquired_gen); - if stale { - conns.remove(&target); - } + conns.remove(&target); } - self.connect_and_send(target, tag, data).await; } } - None => { - self.connect_and_send(target, tag, data).await; - } } + self.connect_and_send(target, tag, data).await; } async fn connect_and_send(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) { @@ -184,17 +197,29 @@ impl Transport { connecting.insert(target); } - let result = self.connect_and_send_inner(target, tag, data).await; + self.connect_and_send_inner(target, tag, data).await; self.connecting.lock().remove(&target); - result } async fn connect_and_send_inner(&self, target: SocketAddr, tag: ChannelTag, data: &[u8]) { + let Ok(permit) = self.outbound_permits.clone().try_acquire_owned() else { + tracing::warn!( + peer = %target, + max = MAX_OUTBOUND_CONNECTIONS, + "outbound connection limit reached, dropping" + ); + return; + }; let shutdown = self.shutdown.clone(); - let stream = (|| async { - tokio::time::timeout(Duration::from_secs(5), TcpStream::connect(target)) + let endpoint = self.endpoint.clone(); + let conn = (|| async { + let connecting = endpoint + .connect(target, RIPPLE_SERVER_NAME) + .map_err(std::io::Error::other)?; + tokio::time::timeout(CONNECT_TIMEOUT, connecting) .await .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "connect timeout"))? + .map_err(std::io::Error::other) }) .retry( ExponentialBuilder::default() @@ -204,74 +229,44 @@ impl Transport { ) .when(|_| !shutdown.is_cancelled()) .await; - match stream { - Ok(stream) => { - if self.outbound_count.load(Ordering::Relaxed) >= MAX_OUTBOUND_CONNECTIONS { - tracing::warn!( - peer = %target, - max = MAX_OUTBOUND_CONNECTIONS, - "outbound connection limit reached, dropping" - ); - return; - } - self.outbound_count.fetch_add(1, Ordering::Relaxed); - configure_socket(&stream); - let (read_half, write_half) = stream.into_split(); - let (write_tx, mut write_rx) = mpsc::channel::>(1024); - let conn_gen = self.conn_generation.fetch_add(1, Ordering::Relaxed); + match conn { + Ok(conn) => { + let generation = self.conn_generation.fetch_add(1, Ordering::Relaxed); self.connections.lock().insert( target, - ConnectionWriter { - tx: write_tx.clone(), - generation: conn_gen, + PeerConn { + conn: conn.clone(), + generation, }, ); - if let Some(frame) = encode_frame(tag, data) { - let _ = write_tx.try_send(frame); + if let Err(e) = write_frame(&conn, tag, data).await { + tracing::warn!(peer = %target, error = %e, "initial write failed"); } - let conn_cancel = self.shutdown.child_token(); - let reader_cancel = conn_cancel.clone(); - let connections = self.connections.clone(); - let outbound_counter = self.outbound_count.clone(); - let peer = target; + let reader_cancel = self.shutdown.child_token(); + tokio::spawn(run_conn_reader( + conn.clone(), + target, + self.incoming_tx.clone(), + reader_cancel.clone(), + )); + let connections = self.connections.clone(); tokio::spawn(async move { - let mut writer = write_half; - loop { - tokio::select! { - _ = conn_cancel.cancelled() => break, - msg = write_rx.recv() => { - match msg { - Some(buf) => { - let write_result = tokio::time::timeout( - WRITE_TIMEOUT, - writer.write_all(&buf), - ).await; - match write_result { - Ok(Ok(())) => {} - Ok(Err(e)) => { - tracing::warn!(peer = %peer, error = %e, "write failed, closing connection"); - break; - } - Err(_) => { - tracing::warn!(peer = %peer, "write timed out, closing connection"); - break; - } - } - } - None => break, - } - } + let _permit = permit; + conn.closed().await; + { + let mut conns = connections.lock(); + if conns + .get(&target) + .is_some_and(|p| p.generation == generation) + { + conns.remove(&target); } } - connections.lock().remove(&peer); - outbound_counter.fetch_sub(1, Ordering::Relaxed); - conn_cancel.cancel(); + reader_cancel.cancel(); }); - - Self::spawn_reader_half(read_half, target, self.incoming_tx.clone(), reader_cancel); tracing::debug!(peer = %target, "established outbound connection"); } Err(e) => { @@ -279,215 +274,313 @@ impl Transport { } } } - - fn spawn_reader( - stream: TcpStream, - peer_addr: SocketAddr, - incoming_tx: mpsc::Sender, - cancel: CancellationToken, - inbound_counter: Arc, - ) { - tokio::spawn(async move { - let mut buf = BytesMut::with_capacity(8192); - let mut stream = stream; - loop { - if buf.len() > MAX_FRAME_SIZE * 2 { - tracing::warn!(peer = %peer_addr, buf_len = buf.len(), "read buffer exceeded limit, closing connection"); - break; - } - tokio::select! { - _ = cancel.cancelled() => break, - n = stream.read_buf(&mut buf) => { - match n { - Ok(0) | Err(_) => break, - Ok(_) => { - if !Self::process_frames(&mut buf, peer_addr, &incoming_tx) { - break; - } - } - } - } - } - } - inbound_counter.fetch_sub(1, Ordering::Relaxed); - }); - } - - fn spawn_reader_half( - read_half: tokio::net::tcp::OwnedReadHalf, - peer_addr: SocketAddr, - incoming_tx: mpsc::Sender, - cancel: CancellationToken, - ) { - tokio::spawn(async move { - let mut buf = BytesMut::with_capacity(8192); - let mut reader = read_half; - loop { - if buf.len() > MAX_FRAME_SIZE * 2 { - tracing::warn!(peer = %peer_addr, buf_len = buf.len(), "read buffer exceeded limit, closing connection"); - break; - } - tokio::select! { - _ = cancel.cancelled() => break, - n = reader.read_buf(&mut buf) => { - match n { - Ok(0) | Err(_) => break, - Ok(_) => { - if !Self::process_frames(&mut buf, peer_addr, &incoming_tx) { - break; - } - } - } - } - } - } - cancel.cancel(); - }); - } - - fn process_frames( - buf: &mut BytesMut, - peer_addr: SocketAddr, - incoming_tx: &mpsc::Sender, - ) -> bool { - loop { - match decode_frame(buf) { - DecodeResult::Frame(tag, data) => { - if let Err(e) = incoming_tx.try_send(IncomingFrame { - from: peer_addr, - tag, - data, - }) { - tracing::warn!(peer = %peer_addr, error = %e, "incoming frame channel full, dropping frame"); - } - } - DecodeResult::NeedMoreData => return true, - DecodeResult::Corrupt => return false, - } - } - } } -fn configure_socket(stream: &TcpStream) { - let sock_ref = socket2::SockRef::from(stream); - if let Err(e) = sock_ref.set_tcp_nodelay(true) { - tracing::warn!(error = %e, "failed to set TCP_NODELAY"); - } - let keepalive = socket2::TcpKeepalive::new().with_time(Duration::from_secs(30)); - #[cfg(any(target_os = "linux", target_os = "macos", target_os = "windows"))] - let keepalive = keepalive.with_interval(Duration::from_secs(10)); - let params = keepalive; - if let Err(e) = sock_ref.set_tcp_keepalive(¶ms) { - tracing::warn!(error = %e, "failed to set TCP keepalive"); - } -} - -fn encode_frame(tag: ChannelTag, data: &[u8]) -> Option> { - match data.len() > MAX_FRAME_SIZE { - true => { - tracing::warn!( - frame_len = data.len(), - max = MAX_FRAME_SIZE, - "refusing to encode oversized frame" - ); - None - } - false => { - let len = u32::try_from(data.len()).ok()?; - let mut buf = Vec::with_capacity(5 + data.len()); - buf.put_u32(len); - buf.put_u8(tag as u8); - buf.extend_from_slice(data); - Some(buf) - } - } -} - -enum DecodeResult { - Frame(ChannelTag, Vec), - NeedMoreData, - Corrupt, -} - -fn decode_frame(buf: &mut BytesMut) -> DecodeResult { +async fn run_conn_reader( + conn: Connection, + from: SocketAddr, + incoming_tx: mpsc::Sender, + cancel: CancellationToken, +) { loop { - if buf.len() < 5 { - return DecodeResult::NeedMoreData; - } - let len = u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize; - if len > MAX_FRAME_SIZE { - tracing::warn!( - frame_len = len, - max = MAX_FRAME_SIZE, - "oversized frame, closing connection" - ); - buf.clear(); - return DecodeResult::Corrupt; - } - if buf.len() < 5 + len { - return DecodeResult::NeedMoreData; - } - buf.advance(4); - let tag_byte = buf[0]; - buf.advance(1); - let data = buf.split_to(len).to_vec(); - match ChannelTag::from_u8(tag_byte) { - Some(tag) => return DecodeResult::Frame(tag, data), - None => { - tracing::debug!(tag = tag_byte, "skipping frame with unknown channel tag"); + tokio::select! { + _ = cancel.cancelled() => break, + accepted = conn.accept_uni() => match accepted { + Ok(recv) => { + tokio::spawn(read_frame( + recv, + from, + incoming_tx.clone(), + )); + } + Err(_) => break, } } } } +enum FrameReadError { + Oversize, + Stream(String), +} + +async fn read_frame(recv: RecvStream, from: SocketAddr, incoming_tx: mpsc::Sender) { + let read = async { + let mut recv = recv; + let mut tag_byte = [0u8; 1]; + recv.read_exact(&mut tag_byte) + .await + .map_err(|e| FrameReadError::Stream(e.to_string()))?; + let mut data = Vec::with_capacity(READ_CHUNK_BYTES); + loop { + match recv.read_chunk(READ_CHUNK_BYTES, true).await { + Ok(Some(chunk)) => { + let len = chunk.bytes.len(); + if data.len() + len > MAX_FRAME_SIZE { + return Err(FrameReadError::Oversize); + } + data.extend_from_slice(&chunk.bytes); + } + Ok(None) => break, + Err(e) => return Err(FrameReadError::Stream(e.to_string())), + } + } + Ok::<(u8, Vec), FrameReadError>((tag_byte[0], data)) + }; + + match tokio::time::timeout(READ_TIMEOUT, read).await { + Ok(Ok((tag_byte, data))) => match ChannelTag::from_u8(tag_byte) { + Some(tag) => { + let frame = IncomingFrame { from, tag, data }; + if let Err(e) = incoming_tx.try_send(frame) { + tracing::warn!(peer = %from, error = %e, "incoming frame channel full, dropping frame"); + } + } + None => tracing::debug!(tag = tag_byte, "unknown channel tag, dropping frame"), + }, + Ok(Err(FrameReadError::Oversize)) => { + tracing::debug!(peer = %from, max = MAX_FRAME_SIZE, "inbound frame exceeds max size, dropping"); + } + Ok(Err(FrameReadError::Stream(msg))) => { + tracing::debug!(peer = %from, error = %msg, "failed reading uni stream"); + } + Err(_) => { + tracing::debug!(peer = %from, "inbound frame read timed out, dropping"); + } + } +} + +async fn write_frame(conn: &Connection, tag: ChannelTag, data: &[u8]) -> std::io::Result<()> { + if data.len() > MAX_FRAME_SIZE { + tracing::warn!( + frame_len = data.len(), + max = MAX_FRAME_SIZE, + "refusing to send oversized frame" + ); + return Ok(()); + } + let timed_out = || std::io::Error::new(std::io::ErrorKind::TimedOut, "write timeout"); + let deadline = tokio::time::Instant::now() + WRITE_TIMEOUT; + match tokio::time::timeout_at(deadline, conn.open_uni()).await { + Ok(Ok(mut send)) => { + let write = async { + send.write_all(&[tag as u8]) + .await + .map_err(std::io::Error::other)?; + send.write_all(data).await.map_err(std::io::Error::other)?; + send.finish().map_err(std::io::Error::other) + }; + let outcome = tokio::time::timeout_at(deadline, write).await; + match outcome { + Ok(Ok(())) => Ok(()), + Ok(Err(e)) => { + let _ = send.reset(0u32.into()); + Err(e) + } + Err(_) => { + let _ = send.reset(0u32.into()); + Err(timed_out()) + } + } + } + Ok(Err(e)) => Err(std::io::Error::other(e)), + Err(_) => Err(timed_out()), + } +} + +fn transport_config() -> TransportConfig { + let mut tc = TransportConfig::default(); + tc.max_concurrent_uni_streams(VarInt::from(MAX_CONCURRENT_UNI_STREAMS)); + tc.stream_receive_window(VarInt::from_u32(STREAM_RECEIVE_WINDOW)); + tc.receive_window(VarInt::from_u32(CONNECTION_RECEIVE_WINDOW)); + tc.keep_alive_interval(Some(KEEPALIVE)); + tc.max_idle_timeout(Some( + IdleTimeout::try_from(IDLE_TIMEOUT).expect("idle timeout fits in varint"), + )); + tc +} + +type BoxError = Box; + +fn ephemeral_identity() -> Result { + let cert = rcgen::generate_simple_self_signed(vec![RIPPLE_SERVER_NAME.to_string()])?; + Ok(NodeIdentity { + cert: cert.cert.der().clone(), + key: PrivateKeyDer::Pkcs8(PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der())), + }) +} + +fn build_server_config() -> Result { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let builder = rustls::ServerConfig::builder_with_provider(provider) + .with_protocol_versions(&[&rustls::version::TLS13])?; + let ephemeral = ephemeral_identity()?; + let mut crypto = builder + .with_no_client_auth() + .with_single_cert(vec![ephemeral.cert], ephemeral.key)?; + crypto.alpn_protocols = vec![RIPPLE_ALPN.to_vec()]; + + let mut config = ServerConfig::with_crypto(Arc::new(QuicServerConfig::try_from(crypto)?)); + config.transport_config(Arc::new(transport_config())); + Ok(config) +} + +fn build_client_config() -> Result { + let provider = Arc::new(rustls::crypto::ring::default_provider()); + let verifier = Arc::new(SkipServerVerification::new(provider.clone())); + let mut crypto = rustls::ClientConfig::builder_with_provider(provider) + .with_protocol_versions(&[&rustls::version::TLS13])? + .dangerous() + .with_custom_certificate_verifier(verifier) + .with_no_client_auth(); + crypto.alpn_protocols = vec![RIPPLE_ALPN.to_vec()]; + + let mut config = ClientConfig::new(Arc::new(QuicClientConfig::try_from(crypto)?)); + config.transport_config(Arc::new(transport_config())); + Ok(config) +} + +#[derive(Debug)] +struct SkipServerVerification(Arc); + +impl SkipServerVerification { + fn new(provider: Arc) -> Self { + Self(provider) + } +} + +impl rustls::client::danger::ServerCertVerifier for SkipServerVerification { + fn verify_server_cert( + &self, + _end_entity: &CertificateDer<'_>, + _intermediates: &[CertificateDer<'_>], + _server_name: &rustls::pki_types::ServerName<'_>, + _ocsp_response: &[u8], + _now: rustls::pki_types::UnixTime, + ) -> Result { + Ok(rustls::client::danger::ServerCertVerified::assertion()) + } + + fn verify_tls12_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &rustls::DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls12_signature( + message, + cert, + dss, + &self.0.signature_verification_algorithms, + ) + } + + fn verify_tls13_signature( + &self, + message: &[u8], + cert: &CertificateDer<'_>, + dss: &rustls::DigitallySignedStruct, + ) -> Result { + rustls::crypto::verify_tls13_signature( + message, + cert, + dss, + &self.0.signature_verification_algorithms, + ) + } + + fn supported_verify_schemes(&self) -> Vec { + self.0.signature_verification_algorithms.supported_schemes() + } +} + #[cfg(test)] mod tests { use super::*; - #[test] - fn frame_roundtrip() { - let original = b"hello world"; - let encoded = encode_frame(ChannelTag::Gossip, original).expect("should encode"); - let mut buf = BytesMut::from(&encoded[..]); - match decode_frame(&mut buf) { - DecodeResult::Frame(tag, data) => { - assert_eq!(tag, ChannelTag::Gossip); - assert_eq!(data, original); - } - _ => panic!("expected frame"), - } - assert!(buf.is_empty()); + #[tokio::test] + async fn quic_frame_roundtrip() { + let shutdown = CancellationToken::new(); + let (sender, _rx_sender) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind sender"); + let (receiver, mut rx_receiver) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind receiver"); + let target = receiver.local_addr(); + + sender + .send(target, ChannelTag::Gossip, b"hello ripple") + .await; + + let frame = tokio::time::timeout(Duration::from_secs(5), rx_receiver.recv()) + .await + .expect("frame arrives before timeout") + .expect("channel open"); + assert_eq!(frame.tag, ChannelTag::Gossip); + assert_eq!(frame.data, b"hello ripple"); + + shutdown.cancel(); } - #[test] - fn partial_frame_returns_need_more() { - let encoded = encode_frame(ChannelTag::CrdtSync, b"test data").expect("should encode"); - let mut buf = BytesMut::from(&encoded[..3]); - assert!(matches!(decode_frame(&mut buf), DecodeResult::NeedMoreData)); + #[tokio::test] + async fn distinct_channels_roundtrip() { + let shutdown = CancellationToken::new(); + let (sender, _rx_sender) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind sender"); + let (receiver, mut rx_receiver) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind receiver"); + let target = receiver.local_addr(); + + sender.send(target, ChannelTag::Gossip, b"first").await; + sender.send(target, ChannelTag::CrdtSync, b"second").await; + + let mut seen = Vec::new(); + for _ in 0..2 { + let frame = tokio::time::timeout(Duration::from_secs(5), rx_receiver.recv()) + .await + .expect("frame arrives before timeout") + .expect("channel open"); + seen.push((frame.tag, frame.data)); + } + assert!(seen.contains(&(ChannelTag::Gossip, b"first".to_vec()))); + assert!(seen.contains(&(ChannelTag::CrdtSync, b"second".to_vec()))); + + shutdown.cancel(); } - #[test] - fn multiple_frames() { - let f1 = encode_frame(ChannelTag::Gossip, b"first").expect("should encode"); - let f2 = encode_frame(ChannelTag::Direct, b"second").expect("should encode"); - let mut buf = BytesMut::new(); - buf.extend_from_slice(&f1); - buf.extend_from_slice(&f2); + #[tokio::test] + async fn incoming_frame_from_matches_peer_listen_addr() { + let shutdown = CancellationToken::new(); + let (sender, _rx_sender) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind sender"); + let (receiver, mut rx_receiver) = + Transport::bind("127.0.0.1:0".parse().unwrap(), shutdown.clone()) + .await + .expect("bind receiver"); - match decode_frame(&mut buf) { - DecodeResult::Frame(tag1, data1) => { - assert_eq!(tag1, ChannelTag::Gossip); - assert_eq!(data1, b"first"); - } - _ => panic!("expected frame"), - } + sender + .send(receiver.local_addr(), ChannelTag::Gossip, b"addr check") + .await; - match decode_frame(&mut buf) { - DecodeResult::Frame(tag2, data2) => { - assert_eq!(tag2, ChannelTag::Direct); - assert_eq!(data2, b"second"); - } - _ => panic!("expected frame"), - } + let frame = tokio::time::timeout(Duration::from_secs(5), rx_receiver.recv()) + .await + .expect("frame arrives before timeout") + .expect("channel open"); + assert_eq!( + frame.from, + sender.local_addr(), + "inbound frames must report the peer's canonical listen address" + ); + + shutdown.cancel(); } }