refactor: handle repair requests and replies on separate networks and tokio tasks (#160)

Closes: #146

---------

Co-authored-by: Quentin Kniep <[email protected]>

Author: akhi3030

src/all2all/robust.rs

@@ -94,7 +94,8 @@ mod tests {
                 voting_pubkey: voting_sk.to_pk(),
                 all2all_address: localhost_ip_sockaddr(i.try_into().unwrap()),
                 disseminator_address: dontcare_sockaddr(),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }
 

src/all2all/trivial.rs

@@ -80,7 +80,8 @@ mod tests {
                 voting_pubkey: voting_sk.to_pk(),
                 all2all_address: localhost_ip_sockaddr(i.try_into().unwrap()),
                 disseminator_address: dontcare_sockaddr(),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }
 

src/bin/node.rs

@@ -120,13 +120,15 @@ fn create_node(config: ConfigFile) -> color_eyre::Result<Node> {
     let network = UdpNetwork::new(start_port + 1);
     let disseminator = Rotor::new(network, epoch_info.clone());
     let repair_network = UdpNetwork::new(start_port + 2);
-    let txs_receiver = UdpNetwork::new(start_port + 3);
+    let repair_request_network = UdpNetwork::new(start_port + 3);
+    let txs_receiver = UdpNetwork::new(start_port + 4);
     Ok(Alpenglow::new(
         config.identity_key,
         config.voting_key,
         all2all,
         disseminator,
         repair_network,
+        repair_request_network,
         epoch_info,
         txs_receiver,
     ))
@@ -161,7 +163,8 @@ async fn create_node_configs(
             voting_pubkey: voting_sks[id as usize].to_pk(),
             all2all_address: sockaddr,
             disseminator_address: SocketAddr::new(sockaddr.ip(), sockaddr.port() + 1),
-            repair_address: SocketAddr::new(sockaddr.ip(), sockaddr.port() + 2),
+            repair_request_address: SocketAddr::new(sockaddr.ip(), sockaddr.port() + 2),
+            repair_response_address: SocketAddr::new(sockaddr.ip(), sockaddr.port() + 3),
         });
     }
 

src/bin/performance_test.rs

@@ -41,8 +41,8 @@ async fn create_test_nodes(count: u64) -> Vec<TestNode> {
     let core = Arc::new(SimulatedNetworkCore::default().with_packet_loss(0.0));
     let mut networks = VecDeque::new();
     let mut udp_networks = VecDeque::new();
-    for i in 0..4 * count {
-        if i % 4 == 2 || i % 4 == 3 {
+    for i in 0..5 * count {
+        if i % 5 == 2 || i % 5 == 3 || i % 5 == 4 {
             udp_networks.push_back(UdpNetwork::new_with_any_port());
         } else {
             networks.push_back(core.join_unlimited(i).await);
@@ -51,19 +51,19 @@ async fn create_test_nodes(count: u64) -> Vec<TestNode> {
     for a in 0..count {
         for b in 0..count {
             if a < 6 && b < 6 {
-                core.set_latency(4 * a, 4 * b, Duration::from_millis(20))
+                core.set_latency(5 * a, 5 * b, Duration::from_millis(20))
                     .await;
-                core.set_latency(4 * a + 1, 4 * b + 1, Duration::from_millis(20))
+                core.set_latency(5 * a + 1, 5 * b + 1, Duration::from_millis(20))
                     .await;
             } else if (6..10).contains(&a) && (6..10).contains(&b) {
-                core.set_latency(4 * a, 4 * b, Duration::from_millis(60))
+                core.set_latency(5 * a, 5 * b, Duration::from_millis(60))
                     .await;
-                core.set_latency(4 * a + 1, 4 * b + 1, Duration::from_millis(60))
+                core.set_latency(5 * a + 1, 5 * b + 1, Duration::from_millis(60))
                     .await;
             } else {
-                core.set_latency(4 * a, 4 * b, Duration::from_millis(100))
+                core.set_latency(5 * a, 5 * b, Duration::from_millis(100))
                     .await;
-                core.set_latency(4 * a + 1, 4 * b + 1, Duration::from_millis(100))
+                core.set_latency(5 * a + 1, 5 * b + 1, Duration::from_millis(100))
                     .await;
             }
         }
@@ -77,17 +77,19 @@ async fn create_test_nodes(count: u64) -> Vec<TestNode> {
     for id in 0..count {
         sks.push(SecretKey::new(&mut rng));
         voting_sks.push(aggsig::SecretKey::new(&mut rng));
-        let all2all_address = localhost_ip_sockaddr((4 * id).try_into().unwrap());
-        let disseminator_address = localhost_ip_sockaddr((4 * id + 1).try_into().unwrap());
-        let repair_address = localhost_ip_sockaddr(udp_networks[id as usize].port());
+        let all2all_address = localhost_ip_sockaddr((5 * id).try_into().unwrap());
+        let disseminator_address = localhost_ip_sockaddr((5 * id + 1).try_into().unwrap());
+        let repair_request_address = localhost_ip_sockaddr(udp_networks[id as usize].port());
+        let repair_response_address = localhost_ip_sockaddr(udp_networks[id as usize].port());
         validators.push(ValidatorInfo {
             id,
             stake: 1,
             pubkey: sks[id as usize].to_pk(),
             voting_pubkey: voting_sks[id as usize].to_pk(),
             all2all_address,
             disseminator_address,
-            repair_address,
+            repair_request_address,
+            repair_response_address,
         });
     }
 
@@ -99,13 +101,15 @@ async fn create_test_nodes(count: u64) -> Vec<TestNode> {
             let all2all = TrivialAll2All::new(validators.clone(), networks.pop_front().unwrap());
             let disseminator = Rotor::new(networks.pop_front().unwrap(), epoch_info.clone());
             let repair_network = udp_networks.pop_front().unwrap();
+            let repair_request_network = udp_networks.pop_front().unwrap();
             let txs_receiver = udp_networks.pop_front().unwrap();
             Alpenglow::new(
                 sks[v.id as usize].clone(),
                 voting_sks[v.id as usize].clone(),
                 all2all,
                 disseminator,
                 repair_network,
+                repair_request_network,
                 epoch_info,
                 txs_receiver,
             )

src/consensus.rs

@@ -46,7 +46,7 @@ use self::votor::Votor;
 use crate::consensus::block_producer::BlockProducer;
 use crate::crypto::{aggsig, signature};
 use crate::network::{Network, NetworkMessage, NetworkSendError};
-use crate::repair::Repair;
+use crate::repair::{Repair, RepairRequestHandler};
 use crate::shredder::Shred;
 use crate::{All2All, Disseminator, Slot, ValidatorInfo};
 
@@ -93,13 +93,18 @@ where
     T: Network + Sync + Send + 'static,
 {
     /// Creates a new Alpenglow consensus node.
+    ///
+    /// `repair_network` - Network from which the node sends [`RepairRequest`] messages and receives [`RepairResponse`] messages.
+    /// `repair_request_network` - Network where the node receives [`RepairRequest`] messages and sends [`RepairResponse`] messages.
     #[must_use]
+    #[allow(clippy::too_many_arguments)]
     pub fn new<R: Network + Sync + Send + 'static>(
         secret_key: signature::SecretKey,
         voting_secret_key: aggsig::SecretKey,
         all2all: A,
         disseminator: D,
         repair_network: R,
+        repair_request_network: R,
         epoch_info: Arc<EpochInfo>,
         txs_receiver: T,
     ) -> Self {
@@ -119,6 +124,14 @@ where
         ));
         let pool = Arc::new(RwLock::new(pool));
 
+        let repair_request_handler = RepairRequestHandler::new(
+            epoch_info.clone(),
+            blockstore.clone(),
+            repair_request_network,
+        );
+        let _repair_request_handler =
+            tokio::spawn(async move { repair_request_handler.run().await });
+
         let mut repair = Repair::new(
             Arc::clone(&blockstore),
             Arc::clone(&pool),

src/consensus/blockstore.rs

@@ -347,7 +347,8 @@ mod tests {
             voting_pubkey: voting_sk.to_pk(),
             all2all_address: dontcare_sockaddr(),
             disseminator_address: dontcare_sockaddr(),
-            repair_address: dontcare_sockaddr(),
+            repair_request_address: dontcare_sockaddr(),
+            repair_response_address: dontcare_sockaddr(),
         };
         let validators = vec![info];
         let epoch_info = EpochInfo::new(0, validators);

src/consensus/cert.rs

@@ -577,7 +577,8 @@ mod tests {
                 voting_pubkey: voting_sks.last().unwrap().to_pk(),
                 all2all_address: dontcare_sockaddr(),
                 disseminator_address: dontcare_sockaddr(),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }
 

src/disseminator/rotor.rs

@@ -164,7 +164,8 @@ mod tests {
                 voting_pubkey: voting_sks[i as usize].to_pk(),
                 all2all_address: dontcare_sockaddr(),
                 disseminator_address: localhost_ip_sockaddr(base_port + i as u16),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }
 

src/disseminator/rotor/sampling_strategy.rs

@@ -672,7 +672,8 @@ mod tests {
                 voting_pubkey: voting_sk.to_pk(),
                 all2all_address: dontcare_sockaddr(),
                 disseminator_address: dontcare_sockaddr(),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }
         validators

src/disseminator/trivial.rs

@@ -82,7 +82,8 @@ mod tests {
                 voting_pubkey: voting_sks[i as usize].to_pk(),
                 all2all_address: dontcare_sockaddr(),
                 disseminator_address: localhost_ip_sockaddr(base_port + i as u16),
-                repair_address: dontcare_sockaddr(),
+                repair_request_address: dontcare_sockaddr(),
+                repair_response_address: dontcare_sockaddr(),
             });
         }