use crate::config::MpcConfig; use crate::metrics::networking_metrics::{ self, INCOMING_CONNECTION, MPC_P2P_TCP_WRITE_SIZE_BYTES, OUTGOING_CONNECTION, }; use crate::network::conn::{ AllNodeConnectivities, ConnectionVersion, NodeConnectivity, NodeConnectivityInterface, OptionSenderConnectionId, SenderConnectionId, }; use crate::network::constants::{MAX_MESSAGE_SIZE_BYTES, MESSAGE_READ_TIMEOUT_DURATION}; use crate::network::handshake::{ DialerData, HandshakeOutcome, ListenerData, MIN_EXPECTED_CONNECTION_ID, p2p_handshake_dialer, p2p_handshake_listener, }; use crate::network::{MeshNetworkTransportReceiver, MeshNetworkTransportSender}; use crate::primitives::{ IndexerHeightMessage, MpcMessage, MpcMessageKind, MpcPeerMessage, ParticipantId, PeerIndexerHeightMessage, PeerMessage, }; use crate::protocol_version::CURRENT_PROTOCOL_VERSION; use crate::tracking::{self, AutoAbortTask, AutoAbortTaskCollection}; use anyhow::{Context, anyhow}; use async_trait::async_trait; use borsh::{BorshDeserialize, BorshSerialize}; use bytes::Bytes; use ed25519_dalek::VerifyingKey; use futures::{SinkExt, StreamExt}; use rustls::{ClientConfig, CommonState}; use std::collections::HashMap; use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4}; use std::sync::Arc; use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt}; use tokio::net::{TcpListener, TcpStream}; use tokio::sync::mpsc::{self, UnboundedReceiver, UnboundedSender}; use tokio::time::timeout; use tokio_rustls::TlsAcceptor; use tokio_util::codec::{Decoder, Framed, LengthDelimitedCodec}; use tokio_util::sync::CancellationToken; use tokio_util::time::FutureExt; use tracing::info; /// Disables Nagle's algorithm, by setting TCP_NODELAY to true. /// This will send small packets immediately, reducing latency for node messages at /// the cost of higher packet overhead. const TCP_NODELAY: bool = true; const TCP_CONNECTION_KEEPALIVE_DELAY: std::time::Duration = std::time::Duration::from_secs(3); const TCP_CONNECTION_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); const TCP_CONNECTION_RETRIES: u32 = 3; const PING_KEEPALIVE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(5); type FrameHeader = u32; const FRAME_HEADER_SIZE_BYTES: usize = size_of::(); /// TCP_USER_TIMEOUT: Maximum time that transmitted data may remain /// unacknowledged before TCP will forcibly close the connection. /// This is critical for detecting half-open connections during active writes. /// Set to 30 seconds to detect when peer stops ACKing our writes. /// Supported on Linux #[cfg(target_os = "linux")] const TCP_USER_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30); /// Timeout for individual write operations to detect if writes are hanging. const WRITE_OPERATION_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); /// Implements MeshNetworkTransportSender for sending messages over a TLS-based /// mesh network. pub struct TlsMeshSender { my_id: ParticipantId, participants: Vec, connections: HashMap>, connectivities: Arc>, } /// Implements MeshNetworkTransportReceiver. pub struct TlsMeshReceiver { receiver: UnboundedReceiver, _incoming_connections_task: AutoAbortTask<()>, } /// Maps public keys to participant IDs. Used to identify incoming connections. #[derive(Default)] struct ParticipantIdentities { key_to_participant_id: HashMap, } /// A retrying connection that will automatically reconnect if the TCP /// connection is broken. struct PersistentConnection { target_participant_id: ParticipantId, connectivity: Arc>, // The task that loops to connect to the target. When `PersistentConnection` // is dropped, this task is aborted. The task owns any active connection, // so dropping it also frees any connection currently alive. _task: AutoAbortTask<()>, } /// State for a single TLS/TCP connection to one participant. We only ever send /// messages through this connection, so there is nothing to handle receiving. /// Dropping this struct will automatically close the connection. struct OutgoingConnection { /// Used to send messages via the connection. sender: UnboundedSender, /// Task that reads messages from the channel (other side of `sender`) and /// sends it over the TLS connection. This task owns the connection, so /// dropping it closes the connection. _sender_task: AutoAbortTask<()>, /// Task that periodically sends a Ping message to the other side. It does /// not expect a Pong, it simply keeps the connection alive (so we can /// quickly detect if the connection is broken). _keepalive_task: AutoAbortTask<()>, /// This is cancelled when the connection is closed. Used to wait for the /// connection to close. closed: CancellationToken, } /// Simple structure to cancel the CancellationToken when dropped. struct DropToCancel(CancellationToken); impl Drop for DropToCancel { fn drop(&mut self) { self.0.cancel(); } } #[derive(BorshSerialize, BorshDeserialize)] enum Packet { Ping, MpcMessage(MpcMessage), IndexerHeight(IndexerHeightMessage), } impl Packet { fn message_type_label(&self) -> &'static str { match &self { Packet::Ping => networking_metrics::PING_MESSAGE, Packet::IndexerHeight(_) => networking_metrics::INDEXER_HEIGHT_MESSAGE, Packet::MpcMessage(MpcMessage { kind, .. }) => match kind { MpcMessageKind::Start(_) => networking_metrics::MPC_START_MESSAGE, MpcMessageKind::Computation(_) => networking_metrics::MPC_COMPUTATION_MESSAGE, MpcMessageKind::Abort(_) => networking_metrics::MPC_ABORT_MESSAGE, MpcMessageKind::Success => networking_metrics::MPC_SUCCESS_MESSAGE, }, } } } impl OutgoingConnection { /// Both sides of the connection must complete handshake within this time, or else /// the connection is considered not successful. const HANDSHAKE_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(2); /// Maximum time to establish the TCP connection and complete the TLS /// handshake. Without this bound, a middlebox that accepts the TCP /// connection but never sends a byte back would hang the dial — and the /// persistent-connection retry loop awaiting it — forever. const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(10); /// Makes a TLS/TCP connection to the given address, authenticating the /// other side as the given participant. async fn new( client_config: Arc, target_address: &str, target_participant_id: ParticipantId, participant_identities: &ParticipantIdentities, sender_connection_id: u32, ) -> anyhow::Result { let mut tls_stream = timeout(Self::CONNECT_TIMEOUT, async { let tcp_stream = TcpStream::connect(target_address) .await .context("failed to establish tcp stream")?; let tcp_stream = configure_tcp_stream(tcp_stream).context("failed to configure tcp stream")?; tokio_rustls::TlsConnector::from(client_config) .connect("dummy".try_into().unwrap(), tcp_stream) .await .context("failed to establish tls stream") }) .await .context("timed out establishing tls connection")??; let peer_id = verify_peer_identity(tls_stream.get_ref().1, participant_identities) .context("verify server identity")?; if peer_id != target_participant_id { anyhow::bail!( "incorrect peer identity, expected {}, authenticated {}", target_participant_id, peer_id ); } info!("performing P2P handshake with: {:?}", target_address); let connection_info = timeout( Self::HANDSHAKE_TIMEOUT, p2p_handshake_dialer( DialerData { sender_connection_id, }, &mut tls_stream, ), ) .await??; match connection_info { HandshakeOutcome::Unsupported(peer_version) => { if let Err(err) = tls_stream.shutdown().await { tracing::error!(err = %err, "TLS shutdown failed"); } anyhow::bail!( "peer runs unsupported protocol version. Their version {}, our version {:?}", peer_version, CURRENT_PROTOCOL_VERSION ); } HandshakeOutcome::Dec2025(_) => { // a dialer concluding a successful handshake with a listening legacy node is // always assumed to be accepted } HandshakeOutcome::Jan2026(handshake_data) => { if !handshake_data.is_accepted() { tracing::warn!( "peer is not accepting this connection attempt {:?}", handshake_data ); if let Err(err) = tls_stream.shutdown().await { tracing::error!(err = %err, "TLS shutdown failed"); } anyhow::bail!("connection not accepted: {:?}", handshake_data); } } } let mut framed_tls_stream = configure_framed_stream(tls_stream); let (sender, mut receiver) = mpsc::unbounded_channel::(); let closed = CancellationToken::new(); let closed_clone = closed.clone(); let sender_task = tracking::spawn_checked( &format!("TLS connection to {}", target_participant_id), async move { let _drop_to_cancel = DropToCancel(closed_clone); let mut sent_bytes: u64 = 0; let peer_id_string = peer_id.to_string(); let result: anyhow::Result<()> = async { loop { tokio::select! { data = receiver.recv() => { let Some(data) = data else { break; }; let serialized = borsh::to_vec(&data)?; let bytes = Bytes::from(serialized); let payload_size = bytes.len(); // Add timeout to write operations to detect if writes are hanging // (e.g., due to half-open connection where peer stopped ACKing) match framed_tls_stream.send(bytes).timeout(WRITE_OPERATION_TIMEOUT).await { Ok(Ok(_)) => {}, Ok(Err(e)) => return Err(e.into()), Err(_) => { // Write timed out - connection is likely stuck/half-open return Err(anyhow::anyhow!( "write operation timed out after {}s (connection may be half-open)", WRITE_OPERATION_TIMEOUT.as_secs() )); } } let total_message_size_bytes: u64 = match FRAME_HEADER_SIZE_BYTES.checked_add(payload_size) { Some(size) => match size.try_into() { Ok(size) => size, Err(_) => { tracing::error!("total_message_size_bytes usize->u64 overflow: {size}"); u64::MAX } }, None => { tracing::error!("total_message_size_bytes overflow: FRAME_HEADER_SIZE_BYTES({FRAME_HEADER_SIZE_BYTES}) + payload_size({payload_size})"); u64::MAX } }; let metric_labels = [ peer_id_string.as_str(), OUTGOING_CONNECTION, data.message_type_label(), ]; MPC_P2P_TCP_WRITE_SIZE_BYTES .with_label_values(&metric_labels) .observe(total_message_size_bytes as f64); sent_bytes = match sent_bytes.checked_add(total_message_size_bytes) { Some(size) => size, None => { tracing::error!("sent_bytes overflow: {sent_bytes} + {total_message_size_bytes}"); u64::MAX } }; tracking::set_progress(&format!("sent {} bytes", sent_bytes)); } _ = futures::StreamExt::next(&mut framed_tls_stream) => { // We do not expect any data from the other side. However, // selecting on it will quickly return error if the connection // is broken before we have data to send. That way we can // immediately quit the loop as soon as the connection is broken // (so we can reconnect). break; } } } anyhow::Ok(()) }.await; // Peer closing without close_notify is normal in P2P networks (task aborts, // reconnections, process restarts). Treat it as a clean close, not an error. if let Err(err) = &result && is_tls_close_notify_error(err) { tracing::debug!( err = %err, peer_id = %peer_id, "peer closed connection without TLS close_notify" ); return Ok(()); } // Send TLS close_notify before dropping the connection so // the peer does not see an unexpected EOF. let mut tls_stream = framed_tls_stream.into_inner(); if let Err(err) = tls_stream.shutdown().await { tracing::debug!(err = %err, "TLS shutdown failed on outgoing connection"); } result }, ); let sender_clone = sender.clone(); let keepalive_task = tracking::spawn( &format!("ping keepalive task for {}", target_participant_id), async move { loop { tokio::time::sleep(PING_KEEPALIVE_INTERVAL).await; if sender_clone.send(Packet::Ping).is_err() { // The receiver side will be dropped when the sender task is // dropped (i.e. connection is closed). break; } } }, ); Ok(OutgoingConnection { sender, _sender_task: sender_task, _keepalive_task: keepalive_task, closed, }) } async fn wait_for_close(&self) { self.closed.cancelled().await; } fn send_mpc_message(&self, msg: MpcMessage) -> anyhow::Result<()> { self.sender.send(Packet::MpcMessage(msg))?; Ok(()) } fn send_indexer_height(&self, msg: IndexerHeightMessage) -> anyhow::Result<()> { self.sender.send(Packet::IndexerHeight(msg))?; Ok(()) } } impl PersistentConnection { const CONNECTION_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1); /// Sends a message over the connection. If the connection was reset, fail. fn send_mpc_message( &self, expected_version: ConnectionVersion, msg: MpcMessage, ) -> anyhow::Result<()> { self.connectivity .outgoing_connection_asserting(expected_version) .with_context(|| format!("Cannot send MPC message to {}", self.target_participant_id))? .send_mpc_message(msg) .with_context(|| { format!("Cannot send MPC message to {}", self.target_participant_id) })?; Ok(()) } /// Sends a message over the connection. This is done on a best-effort basis. fn send_indexer_height(&self, height: IndexerHeightMessage) { if let Some(conn) = self.connectivity.any_outgoing_connection() { let _ = conn.send_indexer_height(height); } } const MIN_CONNECTION_ID: u32 = 0; pub fn new( client_config: Arc, my_id: ParticipantId, target_address: String, target_participant_id: ParticipantId, participant_identities: Arc, connectivity: Arc>, ) -> anyhow::Result { let connectivity_clone = connectivity.clone(); let task = tracking::spawn( &format!("Persistent connection to {}", target_participant_id), async move { let mut connection_attempt = Self::MIN_CONNECTION_ID; loop { let new_conn = match OutgoingConnection::new( client_config.clone(), &target_address, target_participant_id, &participant_identities, connection_attempt, ) .await { Ok(new_conn) => { tracing::info!( my_id = %my_id, target_participant_id = %target_participant_id, "outgoing connection established" ); new_conn } Err(e) => { tracing::info!( my_id = %my_id, target_participant_id = %target_participant_id, error = %format_args!("{e:#}"), "could not connect, retrying" ); // Don't immediately retry, to avoid spamming the network with // connection attempts. tokio::time::sleep(Self::CONNECTION_RETRY_DELAY).await; continue; } }; let new_conn = Arc::new(new_conn); connectivity.set_outgoing_connection(&new_conn); new_conn.wait_for_close().await; connection_attempt = connection_attempt.wrapping_add(1); // TODO(#1829): check if we can add a timeout } }, ); Ok(PersistentConnection { target_participant_id, connectivity: connectivity_clone, _task: task, }) } } #[derive(Default)] pub struct IncomingConnection { sender_connection_id: u32, } impl SenderConnectionId for IncomingConnection { fn sender_connection_id(&self) -> u32 { self.sender_connection_id } } /// Creates a mesh network using TLS over TCP for communication. pub async fn new_tls_mesh_network( config: &MpcConfig, p2p_private_key: &ed25519_dalek::SigningKey, ) -> anyhow::Result<( impl MeshNetworkTransportSender + use<>, impl MeshNetworkTransportReceiver + use<>, )> { let (server_config, client_config) = mpc_tls::tls::configure_tls(p2p_private_key)?; let my_port = config .participants .participants .iter() .find(|participant| participant.id == config.my_participant_id) .map(|participant| participant.port) .ok_or_else(|| anyhow!("My ID not found in participants"))?; info!("Preparing participant data."); // Prepare participant data. let mut participant_identities = ParticipantIdentities::default(); let mut connections = HashMap::new(); let connectivities = Arc::new(AllNodeConnectivities::< OutgoingConnection, IncomingConnection, >::new( config.my_participant_id, &config .participants .participants .iter() .map(|p| p.id) .collect::>(), )); for participant in &config.participants.participants { if participant.id == config.my_participant_id { continue; } participant_identities .key_to_participant_id .insert(participant.p2p_public_key, participant.id); } let participant_identities = Arc::new(participant_identities); let client_config = Arc::new(client_config); for participant in &config.participants.participants { if participant.id == config.my_participant_id { continue; } connections.insert( participant.id, Arc::new(PersistentConnection::new( client_config.clone(), config.my_participant_id, format!("{}:{}", participant.address, participant.port), participant.id, participant_identities.clone(), connectivities.get(participant.id)?, )?), ); } let tls_acceptor = TlsAcceptor::from(Arc::new(server_config)); let (message_sender, message_receiver) = mpsc::unbounded_channel(); let tcp_listener = TcpListener::bind(SocketAddr::V4(SocketAddrV4::new( Ipv4Addr::new(0, 0, 0, 0), my_port, ))) .await .context("TCP bind")?; let connectivities_clone = connectivities.clone(); let my_id = config.my_participant_id; info!("Spawning incoming connections handler."); let incoming_connections_task = tracking::spawn("incoming connection handler tls mesh network", async move { let mut tasks = AutoAbortTaskCollection::new(); run_accept_loop(listener_accept_stream(tcp_listener), |tcp_stream| { tasks.spawn_checked::<_, ()>( "new connection handler tls mesh network", incoming_connection_handler( message_sender.clone(), connectivities_clone.clone(), tcp_stream, tls_acceptor.clone(), participant_identities.clone(), my_id, ), ); }) .await; }); let sender = TlsMeshSender { my_id: config.my_participant_id, participants: config .participants .participants .iter() .map(|p| p.id) .collect(), connections, connectivities, }; let receiver = TlsMeshReceiver { receiver: message_receiver, _incoming_connections_task: incoming_connections_task, }; Ok((sender, receiver)) } async fn incoming_connection_handler( message_sender: tokio::sync::mpsc::UnboundedSender, connectivities: Arc>, tcp_stream: TcpStream, tls_acceptor: TlsAcceptor, participant_identities: Arc, my_id: ParticipantId, ) -> anyhow::Result<()> { let tcp_stream = configure_tcp_stream(tcp_stream)?; let mut tls_stream = tls_acceptor.accept(tcp_stream).await?; let peer_id = verify_peer_identity(tls_stream.get_ref().1, &participant_identities)?; tracking::set_progress(&format!("Authenticated as {}", peer_id)); let peer = connectivities.get(peer_id)?; // If we have an existing connection, we require this connection attempt to // be newer than the existing connection (to avoid race conditions, c.f. github issue #1759). let min_expected_connection_id: u32 = match peer.sender_connection_id() { Some(existing_connection_id) => existing_connection_id.wrapping_add(1), None => MIN_EXPECTED_CONNECTION_ID, }; let connection_info = timeout( OutgoingConnection::HANDSHAKE_TIMEOUT, p2p_handshake_listener( ListenerData { min_expected_connection_id, }, &mut tls_stream, ), ) .await??; let sender_connection_id = match connection_info { HandshakeOutcome::Unsupported(peer_version) => { if let Err(err) = tls_stream.shutdown().await { tracing::error!(err = %err, "TLS shutdown failed"); } anyhow::bail!( "peer runs unsupported protocol version theirs={}, ours={}", peer_version, CURRENT_PROTOCOL_VERSION ); } HandshakeOutcome::Dec2025(accepted) => { if !accepted { tracing::info!("Connection attempt {} <-- {} ignored", my_id, peer_id,); tls_stream.shutdown().await?; return Ok(()); } 1 } HandshakeOutcome::Jan2026(connection_info) => { if !connection_info.is_accepted() { if let Err(err) = tls_stream.shutdown().await { tracing::error!(err = %err, "TLS shutdown failed"); } anyhow::bail!("Connection not accepted: {:?}", connection_info); } else { connection_info.sender_connection_id } } }; tracing::info!("Incoming {} <-- {} handshake succeeded", my_id, peer_id); let incoming_conn = Arc::new(IncomingConnection { sender_connection_id, }); if let Err(err) = connectivities .get(peer_id)? .set_incoming_connection(&incoming_conn) { tracing::info!("can't accept incoming connection: {}", err); tls_stream.shutdown().await?; return Ok(()); } let mut received_bytes: u64 = 0; let mut framed_tls_stream_reader = configure_framed_stream(tls_stream); let peer_id_string = peer_id.to_string(); let result: anyhow::Result<()> = async { loop { let payload_bytes = match framed_tls_stream_reader .next() .timeout(MESSAGE_READ_TIMEOUT_DURATION) .await? { Some(result) => result?, None => { // Stream ended cleanly (peer closed connection). return Ok(()); } }; let total_message_size_bytes: u64 = match FRAME_HEADER_SIZE_BYTES.checked_add(payload_bytes.len()) { Some(size) => match size.try_into() { Ok(size) => size, Err(_) => { tracing::error!("total_message_size_bytes usize->u64 overflow: {size}"); u64::MAX } }, None => { tracing::error!("total_message_size_bytes overflow: FRAME_HEADER_SIZE_BYTES({FRAME_HEADER_SIZE_BYTES}) + payload_size({})", payload_bytes.len()); u64::MAX } }; let packet = Packet::try_from_slice(&payload_bytes).context("Failed to deserialize packet")?; let message_label = packet.message_type_label(); match packet { Packet::Ping => { // Do nothing. Pings are just for TCP keepalive. } Packet::MpcMessage(mpc_message) => { message_sender.send(PeerMessage::Mpc(MpcPeerMessage { from: peer_id, message: mpc_message, }))?; } Packet::IndexerHeight(message) => { message_sender.send(PeerMessage::IndexerHeight(PeerIndexerHeightMessage { from: peer_id, message, }))?; } } let metric_labels = [peer_id_string.as_str(), INCOMING_CONNECTION, message_label]; MPC_P2P_TCP_WRITE_SIZE_BYTES .with_label_values(&metric_labels) .observe(total_message_size_bytes as f64); received_bytes = match received_bytes.checked_add(total_message_size_bytes) { Some(size) => size, None => { tracing::error!("received_bytes overflow: {received_bytes} + {total_message_size_bytes}"); u64::MAX } }; tracking::set_progress(&format!( "Received {} bytes from {}", received_bytes, peer_id )); } } .await; // Peer closing without close_notify is normal in P2P networks (task aborts, // reconnections, process restarts). Treat it as a clean close, not an error. if let Err(err) = &result && is_tls_close_notify_error(err) { tracing::debug!( err = %err, peer_id = %peer_id, "peer closed connection without TLS close_notify" ); return Ok(()); } // Send TLS close_notify before dropping the connection so // the peer does not see an unexpected EOF. let mut tls_stream = framed_tls_stream_reader.into_inner(); if let Err(err) = tls_stream.shutdown().await { tracing::debug!(err = %err, "TLS shutdown failed on incoming connection"); } result } /// Adapts a `TcpListener` into a `Stream` of accepted connections, so that the /// accept loop can be tested by substituting a fake stream of results. fn listener_accept_stream( listener: TcpListener, ) -> impl futures::Stream> { futures::stream::poll_fn(move |cx| { listener .poll_accept(cx) .map(|res| Some(res.map(|(stream, _)| stream))) }) } /// Drives `accept_stream` to completion, dispatching each successful item to /// `on_connection`. Errors are logged and the loop keeps running so the listener /// task does not silently die on a transient accept failure. async fn run_accept_loop( accept_stream: impl futures::Stream>, mut on_connection: impl FnMut(S), ) { let mut accept_stream = std::pin::pin!(accept_stream); while let Some(result) = accept_stream.next().await { match result { Ok(stream) => on_connection(stream), Err(err) => tracing::error!("error accepting tcp stream: {}", err), } } } /// Checks whether an error is caused by the peer closing the TLS connection /// without sending a close_notify alert. This is expected in P2P networks /// where connections are frequently torn down (task aborts, reconnections, /// process restarts) and the peer may not get a chance to send close_notify. /// /// TLS close_notify exists to prevent truncation attacks, but the /// length-delimited framing layer ensures partial messages are never delivered /// to the application — either a complete frame arrives or the read fails. /// A missing close_notify is therefore indistinguishable from a normal /// connection drop that the protocol already tolerates. /// /// We check for both `UnexpectedEof` kind and the rustls-specific error /// message to avoid accidentally suppressing unrelated EOF errors from /// other I/O layers (e.g. LengthDelimitedCodec truncated frame headers). fn is_tls_close_notify_error(err: &anyhow::Error) -> bool { for cause in err.chain() { if let Some(io_err) = cause.downcast_ref::() && io_err.kind() == std::io::ErrorKind::UnexpectedEof && io_err.to_string().contains("close_notify") { return true; } } false } fn configure_tcp_stream(tcp_stream: TcpStream) -> anyhow::Result { let socket = socket2::Socket::from(tcp_stream.into_std()?); let keepalive = socket2::TcpKeepalive::new() .with_time(TCP_CONNECTION_KEEPALIVE_DELAY) .with_interval(TCP_CONNECTION_RETRY_DELAY) .with_retries(TCP_CONNECTION_RETRIES); socket .set_tcp_keepalive(&keepalive) .context("failed to set TCP keepalive")?; socket .set_tcp_nodelay(TCP_NODELAY) .context("failed to enable `TCP_NODELAY`")?; // Set TCP_USER_TIMEOUT to detect half-open connections during active writes. // This ensures that if the peer stops ACKing our writes, the connection // will be closed after TCP_USER_TIMEOUT milliseconds. // socket2 handles platform detection - this is supported on Linux, Android, Fuchsia, etc. #[cfg(target_os = "linux")] { socket .set_tcp_user_timeout(Some(TCP_USER_TIMEOUT)) .context("failed to set TCP_USER_TIMEOUT")?; } Ok(TcpStream::from_std(socket.into())?) } fn configure_framed_stream(tls_stream: T) -> Framed where T: AsyncRead + AsyncWrite + Unpin, { tokio_util::codec::LengthDelimitedCodec::builder() .max_frame_length(MAX_MESSAGE_SIZE_BYTES) .length_field_length(FRAME_HEADER_SIZE_BYTES) .new_codec() .framed(tls_stream) } fn verify_peer_identity( conn: &CommonState, participant_identities: &ParticipantIdentities, ) -> anyhow::Result { let public_key = mpc_tls::tls::extract_public_key(conn)?; let Some(peer_id) = participant_identities .key_to_participant_id .get(&public_key) else { anyhow::bail!("Connection with unknown public key"); }; Ok(*peer_id) } #[async_trait::async_trait] impl MeshNetworkTransportSender for TlsMeshSender { fn my_participant_id(&self) -> ParticipantId { self.my_id } fn all_participant_ids(&self) -> Vec { self.participants.clone() } fn connectivity(&self, participant_id: ParticipantId) -> Arc { self.connectivities.get(participant_id).unwrap().clone() } fn send( &self, recipient_id: ParticipantId, message: MpcMessage, connection_version: ConnectionVersion, ) -> anyhow::Result<()> { self.connections .get(&recipient_id) .ok_or_else(|| anyhow!("Recipient not found"))? .send_mpc_message(connection_version, message) .with_context(|| format!("Cannot send MPC message to recipient {}", recipient_id))?; Ok(()) } fn send_indexer_height(&self, height: IndexerHeightMessage) { for conn in self.connections.values() { conn.send_indexer_height(height.clone()); } } async fn wait_for_ready( &self, threshold: usize, peers_to_consider: &[ParticipantId], ) -> anyhow::Result<()> { self.connectivities .wait_for_ready(threshold, peers_to_consider) .await; Ok(()) } } #[async_trait] impl MeshNetworkTransportReceiver for TlsMeshReceiver { async fn receive(&mut self) -> anyhow::Result { self.receiver .recv() .await .ok_or_else(|| anyhow!("Channel closed")) } } #[cfg(any(test, feature = "test-utils"))] pub mod testing { use crate::config::{MpcConfig, ParticipantInfo, ParticipantsConfig}; use crate::primitives::ParticipantId; use ed25519_dalek::SigningKey; use near_account_id::AccountId; use rand::rngs::OsRng; /// A unique seed for each integration test to avoid port conflicts during testing. #[derive(Copy, Clone)] pub struct PortSeed { port_number: u16, case: u16, } impl PortSeed { // The base port number used, hoping the OS is not using ports in this range pub const BASE_PORT: u16 = 10000; // This constant must be equal to the total number of ports defined below pub const TOTAL_DEFINED_PORTS: u16 = 23; // Maximum number of nodes that can be handled without port collisions pub const MAX_NODES: u16 = 10; // Maximum number of cases that can be handled without port collisions pub const MAX_CASES: u16 = 4; // Each function below corresponds to a port per node. Each defines an offset, // and all offsets must be different pub const TOTAL_PORTS_PER_NODE: u16 = 4; pub const fn new(port_number: u16) -> Self { Self { port_number, case: 0, } } pub fn with_case(&self, case: u16) -> Self { Self { port_number: self.port_number, case, } } fn compute_port(&self, node_index: u16, offset: u16) -> u16 { Self::BASE_PORT + self.port_number * Self::MAX_NODES * Self::MAX_CASES * Self::TOTAL_PORTS_PER_NODE + node_index * Self::MAX_CASES * Self::TOTAL_PORTS_PER_NODE + self.case * Self::TOTAL_PORTS_PER_NODE + offset } pub fn p2p_port(&self, node_index: usize) -> u16 { self.compute_port(node_index as u16, 0) } pub fn web_port(&self, node_index: usize) -> u16 { self.compute_port(node_index as u16, 1) } pub fn migration_web_port(&self, node_index: usize) -> u16 { self.compute_port(node_index as u16, 2) } pub fn pprof_web_port(&self, node_index: usize) -> u16 { self.compute_port(node_index as u16, 3) } pub const CLI_FOR_PYTEST: Self = Self::new(0); } impl PortSeed { // Each place that passes a PortSeed in should define a unique one here. pub const P2P_BASIC_TEST: Self = Self::new(1); pub const P2P_WAIT_FOR_READY_TEST: Self = Self::new(2); pub const BASIC_CLUSTER_TEST: Self = Self::new(3); pub const FAULTY_CLUSTER_TEST: Self = Self::new(4); pub const KEY_RESHARING_SIMPLE_TEST: Self = Self::new(5); pub const KEY_RESHARING_MULTISTAGE_TEST: Self = Self::new(6); pub const KEY_RESHARING_SIGNATURE_BUFFERING_TEST: Self = Self::new(7); pub const BASIC_MULTIDOMAIN_TEST: Self = Self::new(8); pub const FAULTY_STUCK_INDEXER_TEST: Self = Self::new(9); pub const RECOVERY_TEST: Self = Self::new(10); pub const ONBOARDING_TEST: Self = Self::new(11); pub const MIGRATION_WEBSERVER_SUCCESS_TEST: Self = Self::new(12); pub const MIGRATION_WEBSERVER_FAILURE_TEST: Self = Self::new(13); pub const MIGRATION_WEBSERVER_SUCCESS_TEST_GET_KEYSHARES: Self = Self::new(14); pub const MIGRATION_WEBSERVER_SUCCESS_TEST_SET_KEYSHARES: Self = Self::new(15); pub const MIGRATION_WEBSERVER_CHANGE_MIGRATION_INFO: Self = Self::new(16); pub const BACKUP_CLI_WEBSERVER_GET_KEYSHARES: Self = Self::new(17); pub const BACKUP_CLI_WEBSERVER_PUT_KEYSHARES: Self = Self::new(18); pub const RECONNECTION_TEST: Self = Self::new(19); pub const FOREIGN_CHAIN_POLICY_TEST: Self = Self::new(20); pub const BACKUP_CLI_WEBSERVER_PUT_KEYSHARES_HOSTNAME: Self = Self::new(21); pub const ASSET_GENERATION_SIGNING_CONTENTION_TEST: Self = Self::new(22); } pub fn generate_test_p2p_configs( participant_accounts: &[AccountId], threshold: usize, // this is a hack to make sure that when tests run in parallel, they don't // collide on the same port. port_seed: PortSeed, // Supply `Some` value here if you want to use pre-existing p2p key pairs p2p_keypairs: Option>, ) -> anyhow::Result> { let p2p_keypairs = if let Some(p2p_keypairs) = p2p_keypairs { p2p_keypairs } else { participant_accounts .iter() .map(|_account_id| SigningKey::generate(&mut OsRng)) .collect::>() }; let mut participants = Vec::new(); for (i, (participant_account, p2p_signing_key)) in participant_accounts .iter() .zip(p2p_keypairs.iter()) .enumerate() { participants.push(ParticipantInfo { id: ParticipantId::from_raw(rand::random()), address: "127.0.0.1".to_string(), port: port_seed.p2p_port(i), p2p_public_key: p2p_signing_key.verifying_key(), near_account_id: participant_account.clone(), }); } let mut configs = Vec::new(); for (i, singing_key) in p2p_keypairs.into_iter().enumerate() { let participants = ParticipantsConfig { threshold: threshold as u64, participants: participants.clone(), }; let mpc_config = MpcConfig { my_participant_id: participants.participants[i].id, participants, }; configs.push((mpc_config, singing_key)); } Ok(configs) } } #[cfg(test)] #[expect(non_snake_case)] mod tests { use super::{ IncomingConnection, OutgoingConnection, ParticipantIdentities, PersistentConnection, }; use crate::config::MpcConfig; use crate::network::conn::{AllNodeConnectivities, ConnectionVersion}; use crate::network::{MeshNetworkTransportReceiver, MeshNetworkTransportSender}; use crate::p2p::testing::{PortSeed, generate_test_p2p_configs}; use crate::primitives::{ ChannelId, MpcMessage, MpcStartMessage, MpcTaskId, ParticipantId, PeerMessage, UniqueId, }; use crate::providers::EcdsaTaskId; use crate::tracking::testing::start_root_task_with_periodic_dump; use mpc_primitives::{AttemptId, EpochId, KeyEventId, domain::DomainId}; use rand::{Rng, SeedableRng}; use std::sync::Arc; use std::time::Duration; use tokio::time::timeout; #[tokio::test] #[test_log::test] async fn test_basic_tls_mesh_network() { let configs = generate_test_p2p_configs( &["test0".parse().unwrap(), "test1".parse().unwrap()], 2, PortSeed::P2P_BASIC_TEST, None, ) .unwrap(); let participant0 = configs[0].0.my_participant_id; let participant1 = configs[1].0.my_participant_id; let all_participants = [participant0, participant1]; start_root_task_with_periodic_dump(async move { let (sender0, mut receiver0) = super::new_tls_mesh_network(&configs[0].0, &configs[0].1) .await .unwrap(); let (sender1, mut receiver1) = super::new_tls_mesh_network(&configs[1].0, &configs[1].1) .await .unwrap(); sender0.wait_for_ready(2, &all_participants).await.unwrap(); sender1.wait_for_ready(2, &all_participants).await.unwrap(); for _ in 0..100 { // TODO: adjust test? let domain_id = rand::thread_rng().r#gen(); let epoch_id = rand::thread_rng().r#gen(); let n_attempts = rand::thread_rng().r#gen::() % 100; let mut attempt_id = AttemptId::new(); for _ in 0..n_attempts { attempt_id = attempt_id.next(); } let channel_id = ChannelId(UniqueId::generate(participant0)); let key_id = KeyEventId::new(EpochId::new(epoch_id), DomainId(domain_id), attempt_id); let msg0to1 = MpcMessage { channel_id, kind: crate::primitives::MpcMessageKind::Start(MpcStartMessage { task_id: MpcTaskId::EcdsaTaskId(EcdsaTaskId::KeyResharing { key_event: key_id, }), participants: vec![participant0, participant1], }), }; sender0 .send( participant1, msg0to1.clone(), sender0.connectivity(participant1).connection_version(), ) .unwrap(); let PeerMessage::Mpc(msg) = receiver1.receive().await.unwrap() else { panic!("Expected MPC message"); }; assert_eq!(msg.from, participant0); assert_eq!(msg.message, msg0to1); let msg1to0 = MpcMessage { channel_id, kind: crate::primitives::MpcMessageKind::Abort("test".to_owned()), }; sender1 .send( participant0, msg1to0.clone(), sender1.connectivity(participant0).connection_version(), ) .unwrap(); let PeerMessage::Mpc(msg) = receiver0.receive().await.unwrap() else { panic!("Expected MPC message"); }; assert_eq!(msg.from, participant1); assert_eq!(msg.message, msg1to0); } }) .await; } fn all_alive_participant_ids(sender: &impl MeshNetworkTransportSender) -> Vec { let mut result = Vec::new(); for participant in sender.all_participant_ids() { if participant == sender.my_participant_id() { continue; } if sender .connectivity(participant) .is_bidirectionally_connected() { result.push(participant); } } result.push(sender.my_participant_id()); result.sort(); result } #[tokio::test] #[test_log::test] async fn test_wait_for_ready() { let mut configs = generate_test_p2p_configs( &[ "test0".parse().unwrap(), "test1".parse().unwrap(), "test2".parse().unwrap(), "test3".parse().unwrap(), ], 4, PortSeed::P2P_WAIT_FOR_READY_TEST, None, ) .unwrap(); let all_participants = |mpc_config: &MpcConfig| { mpc_config .participants .participants .iter() .map(|p| p.id) .collect::>() }; // Make node 3 use the wrong address for the 0th node. All connections should work // except from 3 to 0. configs[3].0.participants.participants[0].address = "169.254.1.1".to_owned(); start_root_task_with_periodic_dump(async move { let (sender0, _receiver0) = super::new_tls_mesh_network(&configs[0].0, &configs[0].1) .await .unwrap(); let (sender1, receiver1) = super::new_tls_mesh_network(&configs[1].0, &configs[1].1) .await .unwrap(); let (sender2, _receiver2) = super::new_tls_mesh_network(&configs[2].0, &configs[2].1) .await .unwrap(); let (sender3, _receiver3) = super::new_tls_mesh_network(&configs[3].0, &configs[3].1) .await .unwrap(); let all_participants0 = &all_participants(&configs[0].0); let all_participants1 = &all_participants(&configs[1].0); let all_participants2 = &all_participants(&configs[2].0); let all_participants3 = &all_participants(&configs[3].0); sender1.wait_for_ready(4, all_participants1).await.unwrap(); sender2.wait_for_ready(4, all_participants2).await.unwrap(); // Node 3 should not be able to connect to node 0, so if we wait for 4, // it should fail. This goes both ways (3 to 0 and 0 to 3). let _ = timeout( Duration::from_secs(1), sender0.wait_for_ready(4, all_participants0), ) .await .unwrap_err(); let _ = timeout( Duration::from_secs(1), sender3.wait_for_ready(4, all_participants3), ) .await .unwrap_err(); // But if we wait for 3, it should succeed. sender0.wait_for_ready(3, all_participants0).await.unwrap(); sender3.wait_for_ready(3, all_participants3).await.unwrap(); let ids: Vec<_> = configs[0] .0 .participants .participants .iter() .map(|p| p.id) .collect(); assert_eq!( all_alive_participant_ids(&sender0), sorted(&[ids[0], ids[1], ids[2]]), ); assert_eq!(all_alive_participant_ids(&sender1), sorted(&ids)); assert_eq!(all_alive_participant_ids(&sender2), sorted(&ids)); assert_eq!( all_alive_participant_ids(&sender3), sorted(&[ids[1], ids[2], ids[3]]), ); // Disconnect node 1. Other nodes should notice the change. drop((sender1, receiver1)); tokio::time::sleep(Duration::from_secs(2)).await; assert_eq!( all_alive_participant_ids(&sender0), sorted(&[ids[0], ids[2]]) ); assert_eq!( all_alive_participant_ids(&sender2), sorted(&[ids[0], ids[2], ids[3]]) ); assert_eq!( all_alive_participant_ids(&sender3), sorted(&[ids[2], ids[3]]) ); // Reconnect node 1. Other nodes should re-establish the connections. let (sender1, _receiver1) = super::new_tls_mesh_network(&configs[1].0, &configs[1].1) .await .unwrap(); sender0.wait_for_ready(3, all_participants0).await.unwrap(); sender1.wait_for_ready(4, all_participants1).await.unwrap(); sender2.wait_for_ready(4, all_participants2).await.unwrap(); sender3.wait_for_ready(3, all_participants3).await.unwrap(); assert_eq!( all_alive_participant_ids(&sender0), sorted(&[ids[0], ids[1], ids[2]]), ); assert_eq!(all_alive_participant_ids(&sender1), sorted(&ids)); assert_eq!(all_alive_participant_ids(&sender2), sorted(&ids)); assert_eq!( all_alive_participant_ids(&sender3), sorted(&[ids[1], ids[2], ids[3]]), ); }) .await; } fn sorted(ids: &[ParticipantId]) -> Vec { let mut ids = ids.to_vec(); ids.sort(); ids } #[tokio::test] #[test_log::test] async fn test_receiver_does_not_accept_new_connection_if_connected() { let mut configs = generate_test_p2p_configs( &["test0".parse().unwrap(), "test1".parse().unwrap()], 2, PortSeed::RECONNECTION_TEST, None, ) .unwrap(); let all_participants = |mpc_config: &MpcConfig| { mpc_config .participants .participants .iter() .map(|p| p.id) .collect::>() }; start_root_task_with_periodic_dump(async move { let (alice, _alice_receiver) = super::new_tls_mesh_network(&configs[0].0, &configs[0].1) .await .unwrap(); let (bob_initial, _bob_receiver) = super::new_tls_mesh_network(&configs[1].0, &configs[1].1) .await .unwrap(); let participants = all_participants(&configs[0].0); assert_eq!(&participants, &all_participants(&configs[1].0)); let alice_id = participants[0]; let bob_id = participants[1]; alice.wait_for_ready(2, &participants).await.unwrap(); bob_initial.wait_for_ready(2, &participants).await.unwrap(); assert!(alice.connectivity(bob_id).is_bidirectionally_connected()); assert!( bob_initial .connectivity(alice_id) .is_bidirectionally_connected() ); let alice_connection_versions = alice.connectivity(bob_id).connection_version(); assert_eq!( alice_connection_versions, ConnectionVersion { outgoing: 1, incoming: 1 } ); let bob_initial_connection_versions = bob_initial.connectivity(alice_id).connection_version(); assert_eq!( bob_initial_connection_versions, ConnectionVersion { outgoing: 1, incoming: 1 } ); configs[1].0.participants.participants[1].port = PortSeed::RECONNECTION_TEST.p2p_port(2); let (bob_new, _bob_new_receiver) = super::new_tls_mesh_network(&configs[1].0, &configs[1].1) .await .unwrap(); tokio::time::sleep(std::time::Duration::from_millis(500)).await; // bob_new will never connect bidirectionally with Alice assert!( !bob_new .connectivity(alice_id) .is_bidirectionally_connected() ); let bob_new_version = bob_new.connectivity(alice_id).connection_version(); assert_eq!( bob_new_version, ConnectionVersion { outgoing: 1, incoming: 1 } ); assert!( bob_initial .connectivity(alice_id) .is_bidirectionally_connected() ); }) .await; } /// Regression test to ensure that /// the incoming-connections accept loop keeps running after an /// `accept()` error, including across a burst of consecutive errors as can /// happen under FD exhaustion or other transient conditions. #[tokio::test] async fn run_accept_loop__should_continue_after_accept_error() { // Given a stream that interleaves Errs and Oks, including a burst of // consecutive Errs before any Ok. fn err() -> std::io::Result<()> { Err(std::io::Error::other("simulated accept error")) } let accept_stream = futures::stream::iter([err(), err(), Ok(()), err(), Ok(())]); let (on_connection_tx, mut on_connection_rx) = tokio::sync::mpsc::unbounded_channel::<()>(); // When run_accept_loop is driven against that stream. super::run_accept_loop(accept_stream, move |_| { on_connection_tx.send(()).unwrap(); }) .await; // Then `on_connection` is invoked for every Ok in the stream, proving // the loop survived both the leading burst of Errs and the Err between // the two Oks. A final `None` confirms the loop also terminated cleanly // on stream end (i.e. dropped the sender) rather than getting stuck. assert_eq!(on_connection_rx.recv().await, Some(())); assert_eq!(on_connection_rx.recv().await, Some(())); assert_eq!(on_connection_rx.recv().await, None); } /// Spawns a listener that accepts TCP connections, reports each accept on /// the returned channel, and never sends a byte back (holding the streams /// open). This mimics a middlebox that ACKs the ClientHello but never /// responds, hanging the dialer's TLS handshake. async fn must_spawn_silent_listener() -> ( String, tokio::sync::mpsc::UnboundedReceiver<()>, tokio::task::JoinHandle<()>, ) { let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap(); let target_address = listener.local_addr().unwrap().to_string(); let (accept_tx, accept_rx) = tokio::sync::mpsc::unbounded_channel(); let listener_task = tokio::spawn(async move { let mut held_streams = Vec::new(); loop { let (stream, _) = listener.accept().await.unwrap(); accept_tx.send(()).unwrap(); held_streams.push(stream); } }); (target_address, accept_rx, listener_task) } fn must_make_client_config() -> Arc { let signing_key = ed25519_dalek::SigningKey::generate(&mut rand::rngs::StdRng::seed_from_u64(42)); let (_server_config, client_config) = mpc_tls::tls::configure_tls(&signing_key).unwrap(); Arc::new(client_config) } // Paused tokio time makes CONNECT_TIMEOUT elapse instantly once the dial // hangs, so the test runs in milliseconds of real time. #[tokio::test(start_paused = true)] #[test_log::test] async fn outgoing_connection__should_fail_dial_when_peer_accepts_tcp_but_never_responds() { // Given a listener that accepts TCP connections but never responds. let (target_address, _accept_rx, listener_task) = must_spawn_silent_listener().await; // When dialing it (under an outer timeout so a regression fails the // test instead of hanging it). let result = timeout( Duration::from_secs(60), OutgoingConnection::new( must_make_client_config(), &target_address, ParticipantId::from_raw(1), &ParticipantIdentities::default(), 0, ), ) .await .expect("dial did not terminate within bounded time"); // Then the dial fails with a timeout instead of hanging. let Err(error) = result else { panic!("dial unexpectedly succeeded"); }; assert!( format!("{error:#}").contains("timed out"), "unexpected error: {error:#}" ); listener_task.abort(); } #[tokio::test(start_paused = true)] #[test_log::test] async fn persistent_connection__should_keep_retrying_when_dial_attempt_hangs() { // Given a listener that accepts TCP connections but never responds. let (target_address, mut accept_rx, listener_task) = must_spawn_silent_listener().await; let client_config = must_make_client_config(); let my_id = ParticipantId::from_raw(0); let target_id = ParticipantId::from_raw(1); start_root_task_with_periodic_dump(async move { let connectivities = AllNodeConnectivities::::new( my_id, &[my_id, target_id], ); // When a persistent connection keeps dialing that listener. let _connection = PersistentConnection::new( client_config, my_id, target_address, target_id, Arc::new(ParticipantIdentities::default()), connectivities.get(target_id).unwrap(), ) .unwrap(); // Then each hung attempt times out and is retried, so the listener // keeps receiving fresh dial attempts. for _ in 0..3 { timeout(Duration::from_secs(120), accept_rx.recv()) .await .expect("retry loop stopped dialing after a hung attempt") .unwrap(); } }) .await; listener_task.abort(); } }