Wait for remote peer connections in Core recovery header fetch

narwhal/network/src/p2p.rs

@@ -1,6 +1,7 @@
 // Copyright (c) Mysten Labs, Inc.
 // SPDX-License-Identifier: Apache-2.0
 
+use crate::anemo_ext::NetworkExt;
 use crate::traits::{PrimaryToPrimaryRpc, PrimaryToWorkerRpc, WorkerRpc};
 use crate::{
     traits::{Lucky, ReliableNetwork, UnreliableNetwork},
@@ -241,12 +242,12 @@ impl PrimaryToPrimaryRpc for anemo::Network {
     async fn get_latest_header(
         &self,
         peer: &NetworkPublicKey,
-        request: impl anemo::types::request::IntoRequest<LatestHeaderRequest> + Send,
+        request: LatestHeaderRequest,
     ) -> Result<LatestHeaderResponse> {
+        const LATEST_HEADER_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
+        let request = anemo::Request::new(request).with_timeout(LATEST_HEADER_REQUEST_TIMEOUT);
         let peer_id = PeerId(peer.0.to_bytes());
-        let peer = self
-            .peer(peer_id)
-            .ok_or_else(|| format_err!("Network has no connection with peer {peer_id}"))?;
+        let peer = self.waiting_peer(peer_id);
         let response = PrimaryToPrimaryClient::new(peer)
             .get_latest_header(request)
             .await

narwhal/network/src/traits.rs

@@ -111,7 +111,7 @@ pub trait PrimaryToPrimaryRpc {
     async fn get_latest_header(
         &self,
         peer: &NetworkPublicKey,
-        request: impl anemo::types::request::IntoRequest<LatestHeaderRequest> + Send,
+        request: LatestHeaderRequest,
     ) -> Result<LatestHeaderResponse>;
 }
 

narwhal/primary/src/core.rs

@@ -16,7 +16,6 @@ use fastcrypto::{hash::Hash as _, SignatureService};
 use futures::stream::FuturesUnordered;
 use futures::StreamExt;
 use network::{CancelOnDropHandler, P2pNetwork, PrimaryToPrimaryRpc, ReliableNetwork};
-use std::time::Duration;
 use std::{
     collections::{HashMap, HashSet},
     sync::Arc,
@@ -40,8 +39,6 @@ use types::{
 #[path = "tests/core_tests.rs"]
 pub mod core_tests;
 
-const LATEST_HEADER_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
-
 pub struct Core {
     /// The public key of this primary.
     name: PublicKey,
@@ -197,10 +194,11 @@ impl Core {
 
         for peer in peers.iter() {
             let network = self.network.network();
-            let request = anemo::Request::new(LatestHeaderRequest {})
-                .with_timeout(LATEST_HEADER_REQUEST_TIMEOUT);
-
-            header_futures.push(async move { network.get_latest_header(peer, request).await });
+            header_futures.push(async move {
+                network
+                    .get_latest_header(peer, LatestHeaderRequest {})
+                    .await
+            });
         }
 
         let mut latest_headers = Vec::new();

narwhal/primary/tests/integration_tests_validator_api.rs

@@ -31,7 +31,7 @@ use types::{
 };
 use worker::{metrics::initialise_metrics, Worker};
 
-#[tokio::test]
+#[tokio::test(flavor = "current_thread", start_paused = true)]
 async fn test_get_collections() {
     let parameters = Parameters {
         batch_size: 200, // Two transactions.
@@ -148,7 +148,7 @@ async fn test_get_collections() {
     );
 
     // Wait for tasks to start
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(15)).await;
 
     // Test gRPC server with client call
     let mut client = connect_to_validator_client(parameters.clone());
@@ -220,7 +220,7 @@ async fn test_get_collections() {
     );
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "current_thread", start_paused = true)]
 // #[cfg_attr(windows, ignore)]
 #[ignore]
 async fn test_remove_collections() {
@@ -324,7 +324,7 @@ async fn test_remove_collections() {
     );
 
     // Wait for tasks to start
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(15)).await;
 
     // Test gRPC server with client call
     let mut client = connect_to_validator_client(parameters.clone());
@@ -427,7 +427,7 @@ async fn test_remove_collections() {
     assert_eq!(Empty {}, actual_result);
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "current_thread", start_paused = true)]
 async fn test_read_causal_signed_certificates() {
     let fixture = CommitteeFixture::builder().randomize_ports(true).build();
     let committee = fixture.committee();
@@ -577,7 +577,7 @@ async fn test_read_causal_signed_certificates() {
     );
 
     // Wait for tasks to start
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(15)).await;
 
     // Test gRPC server with client call
     let mut client = connect_to_validator_client(primary_1_parameters.clone());
@@ -624,7 +624,7 @@ async fn test_read_causal_signed_certificates() {
     assert_eq!(1, response.into_inner().collection_ids.len());
 }
 
-#[tokio::test]
+#[tokio::test(flavor = "current_thread", start_paused = true)]
 async fn test_read_causal_unsigned_certificates() {
     telemetry_subscribers::init_for_testing();
 
@@ -781,7 +781,7 @@ async fn test_read_causal_unsigned_certificates() {
     );
 
     // Wait for tasks to start
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(15)).await;
 
     // Test gRPC server with client call
     let mut client = connect_to_validator_client(primary_1_parameters.clone());
@@ -849,6 +849,8 @@ async fn test_read_causal_unsigned_certificates() {
 /// * Primary 1 be able to fetch the payload for certificates 1 & 2
 #[tokio::test(flavor = "current_thread", start_paused = true)]
 async fn test_get_collections_with_missing_certificates() {
+    telemetry_subscribers::init_for_testing();
+
     // GIVEN keys for two primary nodes
     let fixture = CommitteeFixture::builder().randomize_ports(true).build();
     let committee = fixture.committee();
@@ -1019,7 +1021,7 @@ async fn test_get_collections_with_missing_certificates() {
     );
 
     // Wait for tasks to start
-    tokio::time::sleep(Duration::from_secs(1)).await;
+    tokio::time::sleep(Duration::from_secs(15)).await;
 
     // Test gRPC server with client call
     let mut client = connect_to_validator_client(parameters_1.clone());