feat: batch transactions import

CHANGELOG.md

@@ -26,6 +26,7 @@ Description of the upcoming release here.
 
 ### Changed
 
+- [#1349](https://github.com/FuelLabs/fuel-core/pull/1349): Updated peer-to-peer transactions API to support multiple blocks in a single request, and updated block synchronization to request multiple blocks based on the configured range of headers.
 - [#1380](https://github.com/FuelLabs/fuel-core/pull/1380): Add preliminary, hard-coded config values for heartbeat peer reputation, removing `todo`.
 - [#1377](https://github.com/FuelLabs/fuel-core/pull/1377): Remove `DiscoveryEvent` and use `KademliaEvent` directly in `DiscoveryBehavior`.
 - [#1366](https://github.com/FuelLabs/fuel-core/pull/1366): Improve caching during docker builds in CI by replacing gha

Cargo.toml

@@ -82,6 +82,7 @@ anyhow = "1.0"
 async-trait = "0.1"
 cynic = { version = "2.2.1", features = ["http-reqwest"] }
 clap = "4.1"
+derive_more = { version = "0.99" }
 hyper = { version = "0.14.26" }
 rand = "0.8"
 parking_lot = "0.12"

crates/client/Cargo.toml

@@ -13,7 +13,7 @@ description = "Tx client and schema specification."
 [dependencies]
 anyhow = { workspace = true }
 cynic = { workspace = true }
-derive_more = { version = "0.99" }
+derive_more = { workspace = true }
 eventsource-client = { version = "0.10.2", optional = true }
 fuel-core-types = { workspace = true, features = ["serde"] }
 futures = { workspace = true, optional = true }

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

@@ -23,8 +23,8 @@ use fuel_core_types::{
         SealedBlock,
         SealedBlockHeader,
     },
-    fuel_tx::Transaction,
     fuel_types::BlockHeight,
+    services::p2p::Transactions,
 };
 use std::ops::Range;
 
@@ -127,12 +127,21 @@ impl Database {
         }
     }
 
-    pub fn get_transactions_on_block(
+    pub fn get_transactions_on_blocks(
         &self,
-        block_id: &BlockId,
-    ) -> StorageResult<Option<Vec<Transaction>>> {
-        Ok(self
-            .get_sealed_block_by_id(block_id)?
-            .map(|Sealed { entity: block, .. }| block.into_inner().1))
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Option<Vec<Transactions>>> {
+        let transactions = block_height_range
+            .into_iter()
+            .map(BlockHeight::from)
+            .map(|block_height| {
+                let transactions = self
+                    .get_sealed_block_by_height(&block_height)?
+                    .map(|Sealed { entity: block, .. }| block.into_inner().1)
+                    .map(Transactions);
+                Ok(transactions)
+            })
+            .collect::<StorageResult<_>>()?;
+        Ok(transactions)
     }
 }

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

@@ -8,12 +8,11 @@ use fuel_core_services::stream::BoxStream;
 use fuel_core_storage::Result as StorageResult;
 use fuel_core_types::{
     blockchain::{
-        primitives::BlockId,
         SealedBlock,
         SealedBlockHeader,
     },
-    fuel_tx::Transaction,
     fuel_types::BlockHeight,
+    services::p2p::Transactions,
 };
 use std::ops::Range;
 
@@ -41,9 +40,9 @@ impl P2pDb for Database {
 
     fn get_transactions(
         &self,
-        block_id: &BlockId,
-    ) -> StorageResult<Option<Vec<Transaction>>> {
-        self.get_transactions_on_block(block_id)
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Option<Vec<Transactions>>> {
+        self.get_transactions_on_blocks(block_height_range)
     }
 }
 

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

@@ -12,14 +12,10 @@ use fuel_core_sync::ports::{
 };
 use fuel_core_types::{
     blockchain::{
-        primitives::{
-            BlockId,
-            DaBlockHeight,
-        },
+        primitives::DaBlockHeight,
         SealedBlock,
         SealedBlockHeader,
     },
-    fuel_tx::Transaction,
     fuel_types::BlockHeight,
     services::p2p::{
         peer_reputation::{
@@ -28,6 +24,7 @@ use fuel_core_types::{
         },
         PeerId,
         SourcePeer,
+        Transactions,
     },
 };
 use std::ops::Range;
@@ -50,43 +47,41 @@ impl PeerToPeerPort for P2PAdapter {
 
     async fn get_sealed_block_headers(
         &self,
-        block_range_height: Range<u32>,
+        block_height_range: Range<u32>,
     ) -> anyhow::Result<SourcePeer<Option<Vec<SealedBlockHeader>>>> {
-        if let Some(service) = &self.service {
-            let (peer_id, headers) =
-                service.get_sealed_block_headers(block_range_height).await?;
-            let sourced_headers = SourcePeer {
-                peer_id: peer_id.into(),
-                data: headers,
-            };
-            Ok(sourced_headers)
+        let result = if let Some(service) = &self.service {
+            service.get_sealed_block_headers(block_height_range).await
         } else {
             Err(anyhow::anyhow!("No P2P service available"))
+        };
+        match result {
+            Ok((peer_id, headers)) => {
+                let peer_id: PeerId = peer_id.into();
+                let headers = peer_id.bind(headers);
+                Ok(headers)
+            }
+            Err(err) => Err(err),
         }
     }
 
     async fn get_transactions(
         &self,
-        block: SourcePeer<BlockId>,
-    ) -> anyhow::Result<Option<Vec<Transaction>>> {
+        range: SourcePeer<Range<u32>>,
+    ) -> anyhow::Result<Option<Vec<Transactions>>> {
         let SourcePeer {
             peer_id,
-            data: block,
-        } = block;
+            data: range,
+        } = range;
         if let Some(service) = &self.service {
             service
-                .get_transactions_from_peer(peer_id.into(), block)
+                .get_transactions_from_peer(peer_id.into(), range)
                 .await
         } else {
             Err(anyhow::anyhow!("No P2P service available"))
         }
     }
 
-    async fn report_peer(
-        &self,
-        peer: PeerId,
-        report: PeerReportReason,
-    ) -> anyhow::Result<()> {
+    fn report_peer(&self, peer: PeerId, report: PeerReportReason) -> anyhow::Result<()> {
         if let Some(service) = &self.service {
             let service_name = "Sync";
             let new_report = self.process_report(report);

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

@@ -274,13 +274,12 @@ impl ProtocolName for MessageExchangePostcardProtocol {
 
 #[cfg(test)]
 mod tests {
-    use fuel_core_types::blockchain::primitives::BlockId;
-
     use super::*;
 
     #[test]
     fn test_request_size_fits() {
-        let m = RequestMessage::Transactions(BlockId::default());
+        let arbitrary_range = 2..6;
+        let m = RequestMessage::Transactions(arbitrary_range);
         assert!(postcard::to_stdvec(&m).unwrap().len() <= MAX_REQUEST_SIZE);
     }
 }

crates/services/p2p/src/p2p_service.rs

@@ -696,15 +696,17 @@ mod tests {
                 BlockHeader,
                 PartialBlockHeader,
             },
-            primitives::BlockId,
             SealedBlock,
             SealedBlockHeader,
         },
         fuel_tx::{
             Transaction,
             TransactionBuilder,
         },
-        services::p2p::GossipsubMessageAcceptance,
+        services::p2p::{
+            GossipsubMessageAcceptance,
+            Transactions,
+        },
     };
     use futures::{
         future::join_all,
@@ -1557,7 +1559,7 @@ mod tests {
             tokio::select! {
                 message_sent = rx_test_end.recv() => {
                     // we received a signal to end the test
-                    assert!(message_sent.unwrap(), "Receuved incorrect or missing missing messsage");
+                    assert!(message_sent.unwrap(), "Received incorrect or missing message");
                     break;
                 }
                 node_a_event = node_a.next_event() => {
@@ -1604,7 +1606,7 @@ mod tests {
                                             }
                                         });
                                     }
-                                    RequestMessage::Transactions(_) => {
+                                    RequestMessage::Transactions(_range) => {
                                         let (tx_orchestrator, rx_orchestrator) = oneshot::channel();
                                         assert!(node_a.send_request_msg(None, request_msg.clone(), ResponseChannelItem::Transactions(tx_orchestrator)).is_ok());
                                         let tx_test_end = tx_test_end.clone();
@@ -1613,7 +1615,8 @@ mod tests {
                                             let response_message = rx_orchestrator.await;
 
                                             if let Ok(Some(transactions)) = response_message {
-                                                let _ = tx_test_end.send(transactions.len() == 5).await;
+                                                let check = transactions.len() == 1 && transactions[0].0.len() == 5;
+                                                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;
@@ -1647,7 +1650,8 @@ mod tests {
                                 let _ = node_b.send_response_msg(*request_id, OutboundResponse::SealedHeaders(Some(sealed_headers)));
                             }
                             RequestMessage::Transactions(_) => {
-                                let transactions = (0..5).map(|_| Transaction::default_test_tx()).collect();
+                                let txs = (0..5).map(|_| Transaction::default_test_tx()).collect();
+                                let transactions = vec![Transactions(txs)];
                                 let _ = node_b.send_response_msg(*request_id, OutboundResponse::Transactions(Some(Arc::new(transactions))));
                             }
                         }
@@ -1662,8 +1666,8 @@ mod tests {
     #[tokio::test]
     #[instrument]
     async fn request_response_works_with_transactions() {
-        request_response_works_with(RequestMessage::Transactions(BlockId::default()))
-            .await
+        let arbitrary_range = 2..6;
+        request_response_works_with(RequestMessage::Transactions(arbitrary_range)).await
     }
 
     #[tokio::test]
@@ -1675,7 +1679,8 @@ mod tests {
     #[tokio::test]
     #[instrument]
     async fn request_response_works_with_sealed_headers_range_inclusive() {
-        request_response_works_with(RequestMessage::SealedHeaders(2..6)).await
+        let arbitrary_range = 2..6;
+        request_response_works_with(RequestMessage::SealedHeaders(arbitrary_range)).await
     }
 
     #[tokio::test]

crates/services/p2p/src/ports.rs

@@ -2,12 +2,11 @@ use fuel_core_services::stream::BoxStream;
 use fuel_core_storage::Result as StorageResult;
 use fuel_core_types::{
     blockchain::{
-        primitives::BlockId,
         SealedBlock,
         SealedBlockHeader,
     },
-    fuel_tx::Transaction,
     fuel_types::BlockHeight,
+    services::p2p::Transactions,
 };
 use std::ops::Range;
 
@@ -29,8 +28,8 @@ pub trait P2pDb: Send + Sync {
 
     fn get_transactions(
         &self,
-        block_id: &BlockId,
-    ) -> StorageResult<Option<Vec<Transaction>>>;
+        block_height_range: Range<u32>,
+    ) -> StorageResult<Option<Vec<Transactions>>>;
 }
 
 pub trait BlockHeightImporter: Send + Sync {

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

@@ -5,22 +5,17 @@ use std::{
 
 use fuel_core_types::{
     blockchain::{
-        primitives::BlockId,
         SealedBlock,
         SealedBlockHeader,
     },
-    fuel_tx::Transaction,
     fuel_types::BlockHeight,
+    services::p2p::Transactions,
 };
 use libp2p::PeerId;
 use serde::{
     Deserialize,
     Serialize,
 };
-use serde_with::{
-    serde_as,
-    FromInto,
-};
 use thiserror::Error;
 use tokio::sync::oneshot;
 
@@ -34,33 +29,32 @@ pub(crate) const MAX_REQUEST_SIZE: usize = core::mem::size_of::<RequestMessage>(
 // This `OutboundResponse` gets prepared to be sent over the wire in `NetworkResponse` format.
 // The Peer that requested the message receives the response over the wire in `NetworkResponse` format.
 // It then unpacks it into `ResponseMessage`.
-// `ResponseChannelItem` is used to forward the data within `ResponseMessage` to the receving channel.
+// `ResponseChannelItem` is used to forward the data within `ResponseMessage` to the receiving channel.
 // Client Peer: `RequestMessage` (send request)
 // Server Peer: `RequestMessage` (receive request) -> `OutboundResponse` -> `NetworkResponse` (send response)
 // Client Peer: `NetworkResponse` (receive response) -> `ResponseMessage(data)` -> `ResponseChannelItem(channel, data)` (handle response)
 
-#[serde_as]
 #[derive(Serialize, Deserialize, Eq, PartialEq, Debug, Clone)]
 pub enum RequestMessage {
     Block(BlockHeight),
     SealedHeaders(Range<u32>),
-    Transactions(#[serde_as(as = "FromInto<[u8; 32]>")] BlockId),
+    Transactions(Range<u32>),
 }
 
 /// Final Response Message that p2p service sends to the Orchestrator
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum ResponseMessage {
     SealedBlock(Box<Option<SealedBlock>>),
     SealedHeaders(Option<Vec<SealedBlockHeader>>),
-    Transactions(Option<Vec<Transaction>>),
+    Transactions(Option<Vec<Transactions>>),
 }
 
 /// Holds oneshot channels for specific responses
 #[derive(Debug)]
 pub enum ResponseChannelItem {
     Block(oneshot::Sender<Option<SealedBlock>>),
     SealedHeaders(oneshot::Sender<(PeerId, Option<Vec<SealedBlockHeader>>)>),
-    Transactions(oneshot::Sender<Option<Vec<Transaction>>>),
+    Transactions(oneshot::Sender<Option<Vec<Transactions>>>),
 }
 
 /// Response that is sent over the wire
@@ -78,7 +72,7 @@ pub enum NetworkResponse {
 pub enum OutboundResponse {
     Block(Option<Arc<SealedBlock>>),
     SealedHeaders(Option<Vec<SealedBlockHeader>>),
-    Transactions(Option<Arc<Vec<Transaction>>>),
+    Transactions(Option<Arc<Vec<Transactions>>>),
 }
 
 #[derive(Debug, Error)]