Download range of headers during syncing

CHANGELOG.md

@@ -15,8 +15,7 @@ Description of the upcoming release here.
 
 ### Changed
 
-- Something changed here 1
-- Something changed here 2
+- [#1270](https://github.com/FuelLabs/fuel-core/pull/1270): Modify the way block headers are retrieved from peers to be done in batches.
 
 #### Breaking
 - [#1262](https://github.com/FuelLabs/fuel-core/pull/1262): The `ConsensusParameters` aggregates all configuration data related to the consensus. It contains many fields that are segregated by the usage. The API of some functions was affected to use lesser types instead the whole `ConsensusParameters`. It is a huge breaking change requiring repetitively monotonically updating all places that use the `ConsensusParameters`. But during updating, consider that maybe you can use lesser types. Usage of them may simplify signatures of methods and make them more user-friendly and transparent.

bin/fuel-core/src/cli/run/p2p.rs

@@ -170,12 +170,15 @@ pub struct P2PArgs {
 
 #[derive(Debug, Clone, Args)]
 pub struct SyncArgs {
-    /// The maximum number of get header requests to make in a single batch.
-    #[clap(long = "sync-max-get-header", default_value = "10", env)]
-    pub max_get_header_requests: usize,
     /// The maximum number of get transaction requests to make in a single batch.
     #[clap(long = "sync-max-get-txns", default_value = "10", env)]
     pub max_get_txns_requests: usize,
+    /// The maximum number of headers to request in a single batch.
+    #[clap(long = "sync-header-batch-size", default_value = "10", env)]
+    pub header_batch_size: u32,
+    /// The maximum number of header batch requests to have active at one time.
+    #[clap(long = "sync-max-header-batch-requests", default_value = "10", env)]
+    pub max_header_batch_requests: usize,
 }
 
 #[derive(Clone, Debug)]
@@ -206,8 +209,9 @@ impl KeypairArg {
 impl From<SyncArgs> for fuel_core::sync::Config {
     fn from(value: SyncArgs) -> Self {
         Self {
-            max_get_header_requests: value.max_get_header_requests,
             max_get_txns_requests: value.max_get_txns_requests,
+            header_batch_size: value.header_batch_size,
+            max_header_batch_requests: value.max_header_batch_requests,
         }
     }
 }

crates/fuel-core/src/database/block.rs

@@ -42,7 +42,6 @@ use fuel_core_types::{
 use itertools::Itertools;
 use std::{
     borrow::{
-        Borrow,
         BorrowMut,
         Cow,
     },
@@ -271,9 +270,8 @@ impl Database {
             .get(commit_block_height)?
             .ok_or(not_found!(FuelBlockMerkleMetadata))?;
 
-        let storage = self.borrow();
         let tree: MerkleTree<FuelBlockMerkleData, _> =
-            MerkleTree::load(storage, commit_merkle_metadata.version)
+            MerkleTree::load(self, commit_merkle_metadata.version)
                 .map_err(|err| StorageError::Other(err.into()))?;
 
         let proof_index = message_merkle_metadata.version - 1;

crates/fuel-core/src/database/sealed_block.rs

@@ -26,6 +26,7 @@ use fuel_core_types::{
     fuel_tx::Transaction,
     fuel_types::BlockHeight,
 };
+use std::ops::Range;
 
 impl DatabaseColumn for SealedBlockConsensus {
     fn column() -> Column {
@@ -93,6 +94,20 @@ impl Database {
         self.get_sealed_block_header(&block_id)
     }
 
+    pub fn get_sealed_block_headers(
+        &self,
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Vec<SealedBlockHeader>> {
+        let headers = block_height_range
+            .map(BlockHeight::from)
+            .map(|height| self.get_sealed_block_header_by_height(&height))
+            .collect::<StorageResult<Vec<_>>>()?
+            .into_iter()
+            .flatten()
+            .collect();
+        Ok(headers)
+    }
+
     pub fn get_sealed_block_header(
         &self,
         block_id: &BlockId,

crates/fuel-core/src/service/adapters/p2p.rs

@@ -15,6 +15,7 @@ use fuel_core_types::{
     fuel_tx::Transaction,
     fuel_types::BlockHeight,
 };
+use std::ops::Range;
 
 impl P2pDb for Database {
     fn get_sealed_block(
@@ -31,6 +32,13 @@ impl P2pDb for Database {
         self.get_sealed_block_header_by_height(height)
     }
 
+    fn get_sealed_headers_range(
+        &self,
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Vec<SealedBlockHeader>> {
+        self.get_sealed_block_headers(block_height_range)
+    }
+
     fn get_transactions(
         &self,
         block_id: &BlockId,

crates/fuel-core/src/service/adapters/sync.rs

@@ -20,8 +20,12 @@ use fuel_core_types::{
     },
     fuel_tx::Transaction,
     fuel_types::BlockHeight,
-    services::p2p::SourcePeer,
+    services::p2p::{
+        PeerId,
+        SourcePeer,
+    },
 };
+use std::ops::Range;
 
 #[async_trait::async_trait]
 impl PeerToPeerPort for P2PAdapter {
@@ -56,6 +60,30 @@ impl PeerToPeerPort for P2PAdapter {
         }
     }
 
+    async fn get_sealed_block_headers(
+        &self,
+        block_range_height: Range<u32>,
+    ) -> anyhow::Result<Vec<SourcePeer<SealedBlockHeader>>> {
+        if let Some(service) = &self.service {
+            Ok(service
+                .get_sealed_block_headers(block_range_height)
+                .await?
+                .map(|(peer_id, headers)| {
+                    let peer_id: PeerId = peer_id.into();
+                    headers
+                        .into_iter()
+                        .map(|header| SourcePeer {
+                            peer_id: peer_id.clone(),
+                            data: header,
+                        })
+                        .collect()
+                })
+                .unwrap_or(Vec::new()))
+        } else {
+            Ok(Vec::new())
+        }
+    }
+
     async fn get_transactions(
         &self,
         block: SourcePeer<BlockId>,

crates/services/p2p/src/codecs/postcard.rs

@@ -222,6 +222,10 @@ impl RequestResponseConverter for PostcardCodec {
 
                 Ok(ResponseMessage::Transactions(response))
             }
+            NetworkResponse::Headers(response) => {
+                let deserialized = self.deserialize(response)?;
+                Ok(ResponseMessage::SealedHeaders(deserialized))
+            }
         }
     }
 
@@ -257,6 +261,10 @@ impl RequestResponseConverter for PostcardCodec {
 
                 Ok(NetworkResponse::Transactions(response))
             }
+            OutboundResponse::SealedHeadersRangeInclusive(headers) => {
+                let serialized = self.serialize(headers)?;
+                Ok(NetworkResponse::Headers(serialized))
+            }
         }
     }
 }

crates/services/p2p/src/p2p_service.rs

@@ -620,6 +620,17 @@ impl<Codec: NetworkCodec> FuelP2PService<Codec> {
                                     );
                                 }
                             }
+                            (
+                                Some(ResponseChannelItem::SealedHeaders(channel)),
+                                Ok(ResponseMessage::SealedHeaders(headers)),
+                            ) => {
+                                if channel.send(Some((peer, headers))).is_err() {
+                                    debug!(
+                                        "Failed to send through the channel for {:?}",
+                                        request_id
+                                    );
+                                }
+                            }
 
                             (Some(_), Err(e)) => {
                                 debug!("Failed to convert IntermediateResponse into a ResponseMessage {:?} with {:?}", response, e);
@@ -692,7 +703,10 @@ mod tests {
                 Consensus,
                 ConsensusVote,
             },
-            header::PartialBlockHeader,
+            header::{
+                BlockHeader,
+                PartialBlockHeader,
+            },
             primitives::BlockId,
             SealedBlock,
             SealedBlockHeader,
@@ -1514,6 +1528,27 @@ mod tests {
         }
     }
 
+    fn arbitrary_headers() -> Vec<SealedBlockHeader> {
+        let mut blocks = Vec::new();
+        for i in 2..=5 {
+            let mut header: BlockHeader = Default::default();
+            header.consensus.height = i.into();
+
+            let sealed_block = SealedBlockHeader {
+                entity: header,
+                consensus: Consensus::PoA(PoAConsensus::new(Default::default())),
+            };
+            blocks.push(sealed_block);
+        }
+        blocks
+    }
+
+    // Metadata gets skipped during serialization, so this is the fuzzy way to compare blocks
+    fn eq_except_metadata(a: &SealedBlockHeader, b: &SealedBlockHeader) -> bool {
+        a.entity.application == b.entity.application
+            && a.entity.consensus == b.entity.consensus
+    }
+
     async fn request_response_works_with(request_msg: RequestMessage) {
         let mut p2p_config = Config::default_initialized("request_response_works_with");
 
@@ -1524,15 +1559,15 @@ mod tests {
         p2p_config.bootstrap_nodes = node_a.multiaddrs();
         let mut node_b = build_service_from_config(p2p_config.clone()).await;
 
-        let (tx_test_end, mut rx_test_end) = mpsc::channel(1);
+        let (tx_test_end, mut rx_test_end) = mpsc::channel::<bool>(1);
 
         let mut request_sent = false;
 
         loop {
             tokio::select! {
                 message_sent = rx_test_end.recv() => {
                     // we received a signal to end the test
-                    assert_eq!(message_sent, Some(true), "Receuved incorrect or missing missing messsage");
+                    assert!(message_sent.unwrap(), "Receuved incorrect or missing missing messsage");
                     break;
                 }
                 node_a_event = node_a.next_event() => {
@@ -1542,10 +1577,10 @@ mod tests {
                             if !peer_addresses.is_empty() && !request_sent {
                                 request_sent = true;
 
-                                match request_msg {
+                                match &request_msg {
                                     RequestMessage::Block(_) => {
                                         let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
-                                        assert!(node_a.send_request_msg(None, request_msg, ResponseChannelItem::Block(tx_orchestrator)).is_ok());
+                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseChannelItem::Block(tx_orchestrator)).is_ok());
                                         let tx_test_end = tx_test_end.clone();
 
                                         tokio::spawn(async move {
@@ -1562,7 +1597,7 @@ mod tests {
                                     }
                                     RequestMessage::SealedHeader(_) => {
                                         let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
-                                        assert!(node_a.send_request_msg(None, request_msg, ResponseChannelItem::SealedHeader(tx_orchestrator)).is_ok());
+                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseChannelItem::SealedHeader(tx_orchestrator)).is_ok());
                                         let tx_test_end = tx_test_end.clone();
 
                                         tokio::spawn(async move {
@@ -1576,9 +1611,28 @@ mod tests {
                                             }
                                         });
                                     }
+                                    RequestMessage::SealedHeaders(_) => {
+                                        let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
+                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseChannelItem::SealedHeaders(tx_orchestrator)).is_ok());
+                                        let tx_test_end = tx_test_end.clone();
+
+                                        tokio::spawn(async move {
+                                            let response_message = rx_orchestrator.await;
+
+                                            let expected = arbitrary_headers();
+
+                                            if let Ok(Some((_, sealed_headers))) = response_message {
+                                                let check = expected.iter().zip(sealed_headers.iter()).all(|(a, b)| eq_except_metadata(a, b));
+                                                let _ = tx_test_end.send(check).await;
+                                            } else {
+                                                tracing::error!("Orchestrator failed to receive a message: {:?}", response_message);
+                                                let _ = tx_test_end.send(false).await;
+                                            }
+                                        });
+                                    }
                                     RequestMessage::Transactions(_) => {
                                         let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
-                                        assert!(node_a.send_request_msg(None, request_msg, ResponseChannelItem::Transactions(tx_orchestrator)).is_ok());
+                                        assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseChannelItem::Transactions(tx_orchestrator)).is_ok());
                                         let tx_test_end = tx_test_end.clone();
 
                                         tokio::spawn(async move {
@@ -1601,7 +1655,7 @@ mod tests {
                 },
                 node_b_event = node_b.next_event() => {
                     // 2. Node B receives the RequestMessage from Node A initiated by the NetworkOrchestrator
-                    if let Some(FuelP2PEvent::RequestMessage{ request_id, request_message: received_request_message }) = node_b_event {
+                    if let Some(FuelP2PEvent::RequestMessage{ request_id, request_message: received_request_message }) = &node_b_event {
                         match received_request_message {
                             RequestMessage::Block(_) => {
                                 let block = Block::new(PartialBlockHeader::default(), (0..5).map(|_| Transaction::default_test_tx()).collect(), &[]);
@@ -1611,7 +1665,7 @@ mod tests {
                                     consensus: Consensus::PoA(PoAConsensus::new(Default::default())),
                                 };
 
-                                let _ = node_b.send_response_msg(request_id, OutboundResponse::Block(Some(Arc::new(sealed_block))));
+                                let _ = node_b.send_response_msg(*request_id, OutboundResponse::Block(Some(Arc::new(sealed_block))));
                             }
                             RequestMessage::SealedHeader(_) => {
                                 let header = Default::default();
@@ -1621,11 +1675,16 @@ mod tests {
                                     consensus: Consensus::PoA(PoAConsensus::new(Default::default())),
                                 };
 
-                                let _ = node_b.send_response_msg(request_id, OutboundResponse::SealedHeader(Some(Arc::new(sealed_header))));
+                                let _ = node_b.send_response_msg(*request_id, OutboundResponse::SealedHeader(Some(Arc::new(sealed_header))));
+                            }
+                            RequestMessage::SealedHeaders(_) => {
+                                let sealed_headers: Vec<_> = arbitrary_headers();
+
+                                let _ = node_b.send_response_msg(*request_id, OutboundResponse::SealedHeadersRangeInclusive(sealed_headers));
                             }
                             RequestMessage::Transactions(_) => {
                                 let transactions = (0..5).map(|_| Transaction::default_test_tx()).collect();
-                                let _ = node_b.send_response_msg(request_id, OutboundResponse::Transactions(Some(Arc::new(transactions))));
+                                let _ = node_b.send_response_msg(*request_id, OutboundResponse::Transactions(Some(Arc::new(transactions))));
                             }
                         }
 
@@ -1650,6 +1709,12 @@ mod tests {
         request_response_works_with(RequestMessage::Block(0.into())).await
     }
 
+    #[tokio::test]
+    #[instrument]
+    async fn request_response_works_with_sealed_headers_range_inclusive() {
+        request_response_works_with(RequestMessage::SealedHeaders(0..0)).await
+    }
+
     #[tokio::test]
     #[instrument]
     async fn request_response_works_with_sealed_header() {

crates/services/p2p/src/ports.rs

@@ -9,6 +9,7 @@ use fuel_core_types::{
     fuel_tx::Transaction,
     fuel_types::BlockHeight,
 };
+use std::ops::Range;
 
 pub trait P2pDb: Send + Sync {
     fn get_sealed_block(
@@ -21,6 +22,11 @@ pub trait P2pDb: Send + Sync {
         height: &BlockHeight,
     ) -> StorageResult<Option<SealedBlockHeader>>;
 
+    fn get_sealed_headers_range(
+        &self,
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Vec<SealedBlockHeader>>;
+
     fn get_transactions(
         &self,
         block_id: &BlockId,