diff --git a/crates/node-config/src/lib.rs b/crates/node-config/src/lib.rs index 3972d695d..6209f2b7f 100644 --- a/crates/node-config/src/lib.rs +++ b/crates/node-config/src/lib.rs @@ -17,6 +17,7 @@ use std::{ fs, net::{Ipv4Addr, SocketAddr, ToSocketAddrs}, path::Path, + time::Duration, }; const DEFAULT_PPROF_PORT: u16 = 34001; @@ -26,6 +27,10 @@ const DEFAULT_PPROF_PORT: u16 = 34001; /// participants that are too far behind in the indexer height. pub const MAX_INDEXER_HEIGHT_DIFF: u64 = 50; +/// How long a follower will wait, at the start of computation, for a +/// connection to each other participant in the round to be established before giving up. +pub const PARTICIPANT_CONNECTION_WAIT_TIMEOUT: Duration = Duration::from_secs(2); + #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] pub struct TripleConfig { pub concurrency: usize, diff --git a/crates/node/src/network.rs b/crates/node/src/network.rs index 000fcc946..4e7c26541 100644 --- a/crates/node/src/network.rs +++ b/crates/node/src/network.rs @@ -403,7 +403,7 @@ impl MeshNetworkClient { } struct IncompleteNetworkTaskChannel { - receiver: tokio::sync::mpsc::UnboundedReceiver, + receiver: mpsc::UnboundedReceiver, } enum SenderOrNewChannel { @@ -521,7 +521,7 @@ pub fn run_network_client( pub struct NetworkTaskChannel { sender: Arc, /// Used for calling receive(&mut self). - receiver: tokio::sync::mpsc::UnboundedReceiver, + receiver: mpsc::UnboundedReceiver, /// The set of participants who sent us a Success message; for leader only. successful_participants: HashSet, /// Function to clean up relevant data structures in the network transport implementation. @@ -621,6 +621,12 @@ impl NetworkTaskChannelSender { /// original connection), in which case the sending would then fail (due to outdated connection /// version) immediately. /// + /// The wait for each participant is bounded by `PARTICIPANT_CONNECTION_WAIT_TIMEOUT`, which + /// is ishort and independent of the overall computation timeout. This is to prevent further + /// waiting on the full computation when say the leader is connected to participants, + /// but some participants are not connected to each other. This will produce a specific + /// error earlier on in the process. + /// /// This should be called at the beginning of the computation if: /// - This is a leader-centric computation, and we are a follower. /// (Rationale: the leader already determined that the participants are online. We wait @@ -645,10 +651,28 @@ impl NetworkTaskChannelSender { continue; } tracking::set_progress(&format!("Waiting for connection to {}", participant)); - self.transport_sender - .connectivity(participant) - .wait_for_connection(self.connection_versions[&participant]) - .await?; + let connectivity = self.transport_sender.connectivity(participant); + tokio::time::timeout( + mpc_node_config::PARTICIPANT_CONNECTION_WAIT_TIMEOUT, + connectivity.wait_for_connection(self.connection_versions[&participant]), + ) + .await + .map_err(|_| { + tracing::warn!( + target: "network", + "[{}] [Task {:?}] Not connected to participant {} \ + after {:?}, cannot continue", + self.my_participant_id, + self.task_id, + participant, + mpc_node_config::PARTICIPANT_CONNECTION_WAIT_TIMEOUT, + ); + anyhow::anyhow!( + "Not connected to participant {} after {:?}, cannot continue", + participant, + mpc_node_config::PARTICIPANT_CONNECTION_WAIT_TIMEOUT, + ) + })??; } tracking::set_progress("All participants connected"); Ok(()) @@ -877,7 +901,15 @@ pub mod testing { pub struct TestMeshTransport { participant_ids: Vec, - senders: HashMap>, + senders: HashMap>, + // Used to simulate a network partition + blocked_pairs: HashSet<(ParticipantId, ParticipantId)>, + } + + impl TestMeshTransport { + fn is_blocked(&self, a: ParticipantId, b: ParticipantId) -> bool { + self.blocked_pairs.contains(&(a, b)) || self.blocked_pairs.contains(&(b, a)) + } } pub struct TestMeshTransportSender { @@ -886,30 +918,36 @@ pub mod testing { } pub struct TestMeshTransportReceiver { - receiver: tokio::sync::mpsc::UnboundedReceiver, + receiver: mpsc::UnboundedReceiver, } - pub struct TestConnectivityInterface; + pub struct TestConnectivityInterface { + connected: bool, + } #[async_trait::async_trait] impl NodeConnectivityInterface for TestConnectivityInterface { - fn is_bidirectionally_connected(&self) -> bool { - true + fn connection_version(&self) -> ConnectionVersion { + ConnectionVersion::default() + } + + fn was_connection_interrupted(&self, _connection_version: ConnectionVersion) -> bool { + false } async fn wait_for_connection( &self, _connection_version: ConnectionVersion, ) -> anyhow::Result<()> { - Ok(()) - } - - fn was_connection_interrupted(&self, _connection_version: ConnectionVersion) -> bool { - false + if self.connected { + Ok(()) + } else { + std::future::pending().await + } } - fn connection_version(&self) -> ConnectionVersion { - ConnectionVersion::default() + fn is_bidirectionally_connected(&self) -> bool { + self.connected } } @@ -925,9 +963,13 @@ pub mod testing { fn connectivity( &self, - _participant_id: ParticipantId, + participant_id: ParticipantId, ) -> Arc { - Arc::new(TestConnectivityInterface) + Arc::new(TestConnectivityInterface { + connected: !self + .transport + .is_blocked(self.my_participant_id, participant_id), + }) } fn send( @@ -936,6 +978,13 @@ pub mod testing { message: crate::primitives::MpcMessage, _connection_version: ConnectionVersion, ) -> anyhow::Result<()> { + anyhow::ensure!( + !self + .transport + .is_blocked(self.my_participant_id, recipient_id), + "no connection to {} (blocked in test partition)", + recipient_id + ); self.transport .senders .get(&recipient_id) @@ -971,12 +1020,19 @@ pub mod testing { pub fn new_test_transports( participants: Vec, + ) -> Vec<(Arc, Box)> { + new_test_transports_with_partition(participants, &[]) + } + + pub fn new_test_transports_with_partition( + participants: Vec, + blocked_pairs: &[(ParticipantId, ParticipantId)], ) -> Vec<(Arc, Box)> { let mut sender_by_participant_id = HashMap::new(); let mut senders = Vec::new(); let mut receivers = Vec::new(); for participant_id in &participants { - let (sender, receiver) = tokio::sync::mpsc::unbounded_channel(); + let (sender, receiver) = mpsc::unbounded_channel(); sender_by_participant_id.insert(*participant_id, sender.clone()); senders.push(sender); receivers.push(receiver); @@ -985,6 +1041,7 @@ pub mod testing { let transport = Arc::new(TestMeshTransport { participant_ids: participants.clone(), senders: sender_by_participant_id, + blocked_pairs: blocked_pairs.iter().copied().collect(), }); let mut transports = Vec::new(); @@ -1014,6 +1071,7 @@ pub mod testing { let transport = Arc::new(TestMeshTransport { participant_ids: participants.clone(), senders: HashMap::new(), + blocked_pairs: HashSet::new(), }); let transport_sender = Arc::new(TestMeshTransportSender { transport, @@ -1046,13 +1104,22 @@ pub mod testing { client_runner: F, ) -> anyhow::Result> where - F: Fn( - Arc, - tokio::sync::mpsc::UnboundedReceiver, - ) -> FR, - FR: std::future::Future> + Send + 'static, + F: Fn(Arc, mpsc::UnboundedReceiver) -> FR, + FR: Future> + Send + 'static, { - let transports = new_test_transports(participants.clone()); + run_test_clients_with_partition(participants, &[], client_runner).await + } + + pub async fn run_test_clients_with_partition( + participants: Vec, + blocked_pairs: &[(ParticipantId, ParticipantId)], + client_runner: F, + ) -> anyhow::Result> + where + F: Fn(Arc, mpsc::UnboundedReceiver) -> FR, + FR: Future> + Send + 'static, + { + let transports = new_test_transports_with_partition(participants.clone(), blocked_pairs); let join_handles = transports .into_iter() .enumerate() @@ -1069,8 +1136,7 @@ pub mod testing { futures::future::join_all(join_handles) .await .into_iter() - .collect::>() - .unwrap() + .collect::>()? } } @@ -1384,6 +1450,136 @@ mod tests { } } +#[cfg(test)] +mod participant_connection_wait_tests { + use super::conn::{ConnectionVersion, NodeConnectivityInterface}; + use super::{ChannelId, MeshNetworkTransportSender, NetworkTaskChannelSender}; + use crate::primitives::{IndexerHeightMessage, MpcMessage, MpcTaskId, ParticipantId, UniqueId}; + use crate::providers::EcdsaTaskId; + use crate::tracking::testing::start_root_task_with_periodic_dump; + use std::collections::HashMap; + use std::sync::Arc; + + /// A connectivity mock, returns immediately if `connected` is true, + /// and otherwise never resolves. + struct MockConnectivity { + connected: bool, + } + + #[async_trait::async_trait] + impl NodeConnectivityInterface for MockConnectivity { + fn connection_version(&self) -> ConnectionVersion { + ConnectionVersion::default() + } + fn was_connection_interrupted(&self, _version: ConnectionVersion) -> bool { + false + } + async fn wait_for_connection(&self, _version: ConnectionVersion) -> anyhow::Result<()> { + if self.connected { + Ok(()) + } else { + std::future::pending().await + } + } + fn is_bidirectionally_connected(&self) -> bool { + self.connected + } + } + + struct MockTransportSender { + my_participant_id: ParticipantId, + all_participant_ids: Vec, + unreachable_participant_id: ParticipantId, + } + + #[async_trait::async_trait] + impl MeshNetworkTransportSender for MockTransportSender { + fn my_participant_id(&self) -> ParticipantId { + self.my_participant_id + } + fn all_participant_ids(&self) -> Vec { + self.all_participant_ids.clone() + } + fn connectivity( + &self, + participant_id: ParticipantId, + ) -> Arc { + Arc::new(MockConnectivity { + connected: participant_id != self.unreachable_participant_id, + }) + } + fn send( + &self, + _recipient_id: ParticipantId, + _message: MpcMessage, + _connection_version: ConnectionVersion, + ) -> anyhow::Result<()> { + Ok(()) + } + fn send_indexer_height(&self, _height: IndexerHeightMessage) {} + async fn wait_for_ready( + &self, + _required_ready_count: usize, + _peers_to_consider: &[ParticipantId], + ) -> anyhow::Result<()> { + Ok(()) + } + } + #[tokio::test(start_paused = true)] + async fn rejects_quickly_when_a_participant_is_unreachable() { + let me = ParticipantId::from_raw(0); + let leader = ParticipantId::from_raw(1); + let unreachable = ParticipantId::from_raw(2); + let participants = vec![me, leader, unreachable]; + + let transport_sender = Arc::new(MockTransportSender { + my_participant_id: me, + all_participant_ids: participants.clone(), + unreachable_participant_id: unreachable, + }); + + let connection_versions: HashMap = participants + .iter() + .filter(|&&p| p != me) + .map(|&p| (p, transport_sender.connectivity(p).connection_version())) + .collect(); + + let sender = NetworkTaskChannelSender { + channel_id: ChannelId(UniqueId::new(me, 0, 0)), + task_id: MpcTaskId::EcdsaTaskId(EcdsaTaskId::ManyTriples { + start: UniqueId::new(me, 0, 0), + count: 1, + }), + leader, + my_participant_id: me, + participants, + connection_versions, + transport_sender, + }; + + let (start, result) = start_root_task_with_periodic_dump(async move { + let start = tokio::time::Instant::now(); + let result = sender.initialize_all_participants_connections().await; + (start, result) + }) + .await; + let err = result.unwrap_err(); + assert!( + err.to_string().contains(&format!( + "Not connected to participant {}", + &unreachable.to_string() + )), + "Not connected to participant {}", + err + ); + assert_eq!( + start.elapsed(), + mpc_node_config::PARTICIPANT_CONNECTION_WAIT_TIMEOUT, + "reject at PARTICIPANT_CONNECTION_WAIT_TIMEOUT, instead of computation timeout" + ); + } +} + #[cfg(test)] mod fault_handling_tests { use super::computation::MpcLeaderCentricComputation; @@ -1609,3 +1805,120 @@ mod fault_handling_tests { } } } + +#[cfg(test)] +mod partition_fault_handling_tests { + use super::computation::MpcLeaderCentricComputation; + use super::{MeshNetworkClient, NetworkTaskChannel}; + use crate::network::testing::run_test_clients_with_partition; + use crate::primitives::UniqueId; + use crate::providers::EcdsaTaskId; + use crate::tests::into_participant_ids; + use crate::tracking::testing::start_root_task_with_periodic_dump; + use std::sync::Arc; + use std::time::Duration; + use threshold_signatures::test_utils::generate_participants; + use tokio::sync::mpsc; + + const COMPUTATION_TIMEOUT: Duration = Duration::from_secs(10); + const FAST_FAILURE_BOUND: Duration = Duration::from_secs(5); + + struct NoOpLeader; + struct NoOpFollower; + + #[async_trait::async_trait] + impl MpcLeaderCentricComputation<()> for NoOpLeader { + async fn compute(self, channel: &mut NetworkTaskChannel) -> anyhow::Result<()> { + channel.receive().await?; + Ok(()) + } + fn leader_waits_for_success(&self) -> bool { + false + } + } + + #[async_trait::async_trait] + impl MpcLeaderCentricComputation<()> for NoOpFollower { + async fn compute(self, _channel: &mut NetworkTaskChannel) -> anyhow::Result<()> { + unreachable!("follower should reject before reaching compute()") + } + fn leader_waits_for_success(&self) -> bool { + false + } + } + + async fn run_partition_test_client( + client: Arc, + mut channel_receiver: mpsc::UnboundedReceiver, + ) -> anyhow::Result<()> { + let me = client.my_participant_id(); + let is_leader = me.raw() == 0; + let channel = if is_leader { + client.new_channel_for_task( + EcdsaTaskId::ManyTriples { + start: UniqueId::new(me, 0, 0), + count: 1, + }, + client.all_participant_ids(), + )? + } else { + channel_receiver.recv().await.unwrap() + }; + let start = tokio::time::Instant::now(); + let result = if is_leader { + NoOpLeader + .perform_leader_centric_computation(channel, COMPUTATION_TIMEOUT) + .await + } else { + NoOpFollower + .perform_leader_centric_computation(channel, COMPUTATION_TIMEOUT) + .await + }; + let elapsed = start.elapsed(); + assert!( + elapsed < FAST_FAILURE_BOUND, + "[{}] took {:?} to fail, expected well under the {:?} computation timeout", + me, + elapsed, + COMPUTATION_TIMEOUT, + ); + let err_string = result.unwrap_err().to_string(); + if is_leader { + assert!( + err_string.contains("Aborted by participant"), + "[{}] expected to receive failure, got: {}", + me, + err_string + ); + } else { + let other_follower = client + .all_participant_ids() + .into_iter() + .find(|&p| p != me && p.raw() != 0) + .unwrap(); + assert!( + err_string.contains("Not connected to participant") + && err_string.contains(&other_follower.to_string()), + "[{}] the unreachable peer {}, got: {}", + me, + other_follower, + err_string + ); + } + Ok(()) + } + + #[tokio::test(start_paused = true)] + async fn leader_learns_quickly_when_two_followers_cannot_reach_each_other() { + let participants = into_participant_ids(&generate_participants(3)); + let a = *participants.iter().find(|p| p.raw() == 1).unwrap(); + let b = *participants.iter().find(|p| p.raw() == 2).unwrap(); + start_root_task_with_periodic_dump(async move { + // Assertions are evaluated within the task + run_test_clients_with_partition(participants, &[(a, b)], run_partition_test_client) + .await + .unwrap(); + }) + .await; + } +} diff --git a/crates/node/src/network/computation.rs b/crates/node/src/network/computation.rs index 6f5c53353..14eb83297 100644 --- a/crates/node/src/network/computation.rs +++ b/crates/node/src/network/computation.rs @@ -33,8 +33,11 @@ pub trait MpcLeaderCentricComputation: Sized + 'static { // We'll wrap the following future in a timeout below. let fut = async move { - if !sender.is_leader() { - sender.initialize_all_participants_connections().await?; + if !sender.is_leader() + && let Err(err) = sender.initialize_all_participants_connections().await + { + sender.communicate_failure(&err); + return Err(err); } let result = self.compute(&mut channel).await; let result = match result {