Download range of headers during syncing

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

@@ -24,6 +24,14 @@ impl P2pDb for Database {
         self.get_sealed_block_by_height(height)
     }
 
+    fn get_sealed_blocks_inclusive(
+        &self,
+        start_height: &BlockHeight,
+        end_height: &BlockHeight,
+    ) -> StorageResult<Vec<SealedBlock>> {
+        todo!()
+    }
+
     fn get_sealed_header(
         &self,
         height: &BlockHeight,

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

@@ -204,6 +204,10 @@ impl RequestResponseConverter for PostcardCodec {
 
                 Ok(ResponseMessage::SealedBlock(response))
             }
+            NetworkResponse::BlocksInclusive(block_bytes) => {
+                let response = self.deserialize(block_bytes)?;
+                Ok(ResponseMessage::SealedBlocksInclusive(response))
+            }
             NetworkResponse::Header(header_bytes) => {
                 let response = if let Some(header_bytes) = header_bytes {
                     Some(self.deserialize(header_bytes)?)
@@ -239,6 +243,10 @@ impl RequestResponseConverter for PostcardCodec {
 
                 Ok(NetworkResponse::Block(response))
             }
+            OutboundResponse::BlocksInclusive(sealed_blocks) => {
+                let bytes = self.serialize(sealed_blocks)?;
+                Ok(NetworkResponse::BlocksInclusive(bytes))
+            }
             OutboundResponse::SealedHeader(sealed_header) => {
                 let response = if let Some(sealed_header) = sealed_header {
                     Some(self.serialize(sealed_header.as_ref())?)

crates/services/p2p/src/p2p_service.rs

@@ -598,6 +598,17 @@ impl<Codec: NetworkCodec> FuelP2PService<Codec> {
                                     );
                                 }
                             }
+                            (
+                                Some(ResponseChannelItem::BlocksInclusive(channel)),
+                                Ok(ResponseMessage::SealedBlocksInclusive(blocks)),
+                            ) => {
+                                if channel.send(Some(blocks)).is_err() {
+                                    debug!(
+                                        "Failed to send through the channel for {:?}",
+                                        request_id
+                                    );
+                                }
+                            }
                             (
                                 Some(ResponseChannelItem::Transactions(channel)),
                                 Ok(ResponseMessage::Transactions(transactions)),
@@ -1514,6 +1525,31 @@ mod tests {
         }
     }
 
+    fn arbitrary_blocks() -> Vec<SealedBlock> {
+        let mut blocks = Vec::new();
+        for i in 2..=5 {
+            let block = Block::new(
+                PartialBlockHeader::default(),
+                (0..i).map(|_| Transaction::default_test_tx()).collect(),
+                &[],
+            );
+
+            let sealed_block = SealedBlock {
+                entity: block,
+                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: &SealedBlock, b: &SealedBlock) -> bool {
+        a.entity.header().application == b.entity.header().application
+            && a.entity.header().consensus == b.entity.header().consensus
+            && a.entity.transactions() == b.entity.transactions()
+    }
+
     async fn request_response_works_with(request_msg: RequestMessage) {
         let mut p2p_config = Config::default_initialized("request_response_works_with");
 
@@ -1524,15 +1560,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() => {
@@ -1560,6 +1596,24 @@ mod tests {
                                         });
 
                                     }
+                                    RequestMessage::BlocksInclusive(_, _) => {
+                                        let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
+                                        assert!(node_a.send_request_msg(None, request_msg, ResponseChannelItem::BlocksInclusive(tx_orchestrator)).is_ok());
+                                        let tx_test_end = tx_test_end.clone();
+
+                                        tokio::spawn(async move {
+                                            let response_message = rx_orchestrator.await;
+
+                                            if let Ok(Some(sealed_blocks)) = response_message {
+                                                let expected = arbitrary_blocks();
+                                                let check = sealed_blocks.iter().zip(expected.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::SealedHeader(_) => {
                                         let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
                                         assert!(node_a.send_request_msg(None, request_msg, ResponseChannelItem::SealedHeader(tx_orchestrator)).is_ok());
@@ -1613,6 +1667,11 @@ mod tests {
 
                                 let _ = node_b.send_response_msg(request_id, OutboundResponse::Block(Some(Arc::new(sealed_block))));
                             }
+                            RequestMessage::BlocksInclusive(_, _) => {
+                                let blocks = arbitrary_blocks();
+
+                                let _ = node_b.send_response_msg(request_id, OutboundResponse::BlocksInclusive(blocks));
+                            }
                             RequestMessage::SealedHeader(_) => {
                                 let header = Default::default();
 
@@ -1650,6 +1709,13 @@ mod tests {
         request_response_works_with(RequestMessage::Block(0.into())).await
     }
 
+    #[tokio::test]
+    #[instrument]
+    async fn request_response_works_with_blocks_inclusive() {
+        request_response_works_with(RequestMessage::BlocksInclusive(2.into(), 5.into()))
+            .await
+    }
+
     #[tokio::test]
     #[instrument]
     async fn request_response_works_with_sealed_header() {

crates/services/p2p/src/ports.rs

@@ -16,6 +16,12 @@ pub trait P2pDb: Send + Sync {
         height: &BlockHeight,
     ) -> StorageResult<Option<SealedBlock>>;
 
+    fn get_sealed_blocks_inclusive(
+        &self,
+        start_height: &BlockHeight,
+        end_height: &BlockHeight,
+    ) -> StorageResult<Vec<SealedBlock>>;
+
     fn get_sealed_header(
         &self,
         height: &BlockHeight,

crates/services/p2p/src/request_response/messages.rs

@@ -41,6 +41,7 @@ pub type ChannelItem<T> = oneshot::Sender<Option<T>>;
 #[derive(Serialize, Deserialize, Eq, PartialEq, Debug, Clone, Copy)]
 pub enum RequestMessage {
     Block(BlockHeight),
+    BlocksInclusive(BlockHeight, BlockHeight),
     SealedHeader(BlockHeight),
     Transactions(#[serde_as(as = "FromInto<[u8; 32]>")] BlockId),
 }
@@ -49,6 +50,7 @@ pub enum RequestMessage {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum ResponseMessage {
     SealedBlock(Option<SealedBlock>),
+    SealedBlocksInclusive(Vec<SealedBlock>),
     SealedHeader(Option<SealedBlockHeader>),
     Transactions(Option<Vec<Transaction>>),
 }
@@ -57,6 +59,7 @@ pub enum ResponseMessage {
 #[derive(Debug)]
 pub enum ResponseChannelItem {
     Block(ChannelItem<SealedBlock>),
+    BlocksInclusive(ChannelItem<Vec<SealedBlock>>),
     SealedHeader(ChannelItem<(PeerId, SealedBlockHeader)>),
     Transactions(ChannelItem<Vec<Transaction>>),
 }
@@ -66,6 +69,7 @@ pub enum ResponseChannelItem {
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum NetworkResponse {
     Block(Option<Vec<u8>>),
+    BlocksInclusive(Vec<u8>),
     Header(Option<Vec<u8>>),
     Transactions(Option<Vec<u8>>),
 }
@@ -75,6 +79,7 @@ pub enum NetworkResponse {
 #[derive(Debug, Clone)]
 pub enum OutboundResponse {
     Block(Option<Arc<SealedBlock>>),
+    BlocksInclusive(Vec<SealedBlock>),
     SealedHeader(Option<Arc<SealedBlockHeader>>),
     Transactions(Option<Arc<Vec<Transaction>>>),
 }

crates/services/p2p/src/service.rs

@@ -82,6 +82,11 @@ enum TaskRequest {
         height: BlockHeight,
         channel: oneshot::Sender<Option<SealedBlock>>,
     },
+    GetBlocksInclusive {
+        from_height: BlockHeight,
+        to_height: BlockHeight,
+        channel: oneshot::Sender<Option<Vec<SealedBlock>>>,
+    },
     GetSealedHeader {
         height: BlockHeight,
         channel: oneshot::Sender<Option<(PeerId, SealedBlockHeader)>>,
@@ -222,6 +227,12 @@ where
                         let peer = self.p2p_service.peer_manager().get_peer_id_with_height(&height);
                         let _ = self.p2p_service.send_request_msg(peer, request_msg, channel_item);
                     }
+                    Some(TaskRequest::GetBlocksInclusive { from_height, to_height, channel }) => {
+                        let request_msg = RequestMessage::BlocksInclusive(from_height, to_height);
+                        let channel_item = ResponseChannelItem::BlocksInclusive(channel);
+                        let peer = self.p2p_service.peer_manager().get_peer_id_with_height(&from_height);
+                        let _ = self.p2p_service.send_request_msg(peer, request_msg, channel_item);
+                    }
                     Some(TaskRequest::GetSealedHeader{ height, channel: response }) => {
                         let request_msg = RequestMessage::SealedHeader(height);
                         let channel_item = ResponseChannelItem::SealedHeader(response);
@@ -282,6 +293,11 @@ where
                                     .map(Arc::new);
                                 let _ = self.p2p_service.send_response_msg(request_id, OutboundResponse::Block(block_response));
                             }
+                            RequestMessage::BlocksInclusive(from_height, to_height) => {
+                                let blocks_response = self.db.get_sealed_blocks_inclusive(&from_height, &to_height)?;
+
+                                let _ = self.p2p_service.send_response_msg(request_id, OutboundResponse::BlocksInclusive(blocks_response));
+                            }
                             RequestMessage::Transactions(block_id) => {
                                 let transactions_response = self.db.get_transactions(&block_id)?
                                     .map(Arc::new);
@@ -366,6 +382,24 @@ impl SharedState {
         receiver.await.map_err(|e| anyhow!("{}", e))
     }
 
+    pub async fn get_blocks_inclusive(
+        &self,
+        from_height: BlockHeight,
+        to_height: BlockHeight,
+    ) -> anyhow::Result<Option<Vec<SealedBlock>>> {
+        let (sender, receiver) = oneshot::channel();
+
+        self.request_sender
+            .send(TaskRequest::GetBlocksInclusive {
+                from_height,
+                to_height,
+                channel: sender,
+            })
+            .await?;
+
+        receiver.await.map_err(|e| anyhow!("{}", e))
+    }
+
     pub async fn get_sealed_block_header(
         &self,
         height: BlockHeight,
@@ -556,6 +590,14 @@ pub mod tests {
             }))
         }
 
+        fn get_sealed_blocks_inclusive(
+            &self,
+            _start_height: &BlockHeight,
+            _end_height: &BlockHeight,
+        ) -> StorageResult<Vec<SealedBlock>> {
+            todo!()
+        }
+
         fn get_sealed_header(
             &self,
             _height: &BlockHeight,