Send whole transaction pool upon subscription to gossip

CHANGELOG.md

@@ -6,6 +6,9 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [Unreleased]
 
+### Added
+- [2131](https://github.com/FuelLabs/fuel-core/pull/2131): Add flow in TxPool in order to ask to newly connected peers to share their transaction pool
+
 ## [Version 0.35.0]
 
 ### Added

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

@@ -66,6 +66,10 @@ pub struct P2PArgs {
     #[clap(long = "max-headers-per-request", default_value = "100", env)]
     pub max_headers_per_request: usize,
 
+    /// Max number of txs in a single txs request response
+    #[clap(long = "max-txs-per-request", default_value = "10000", env)]
+    pub max_txs_per_request: usize,
+
     /// Addresses of the bootstrap nodes
     /// They should contain PeerId within their `Multiaddr`
     #[clap(long = "bootstrap-nodes", value_delimiter = ',', env)]
@@ -304,6 +308,7 @@ impl P2PArgs {
             tcp_port: self.peering_port,
             max_block_size: self.max_block_size,
             max_headers_per_request: self.max_headers_per_request,
+            max_txs_per_request: self.max_txs_per_request,
             bootstrap_nodes: self.bootstrap_nodes,
             reserved_nodes: self.reserved_nodes,
             reserved_nodes_only_mode: self.reserved_nodes_only_mode,

crates/fuel-core/src/database.rs

@@ -76,7 +76,7 @@ use crate::state::{
 };
 #[cfg(feature = "rocksdb")]
 use std::path::Path;
-use fuel_core_gas_price_service::fuel_gas_price_updater::fuel_core_storage_adapter::storage::{ GasPriceMetadata};
+use fuel_core_gas_price_service::fuel_gas_price_updater::fuel_core_storage_adapter::storage::GasPriceMetadata;
 use crate::database::database_description::gas_price::GasPriceDatabase;
 
 // Storages implementation

crates/fuel-core/src/p2p_test_helpers.rs

@@ -21,6 +21,10 @@ use fuel_core_p2p::{
     codecs::postcard::PostcardCodec,
     network_service::FuelP2PService,
     p2p_service::FuelP2PEvent,
+    request_response::messages::{
+        RequestMessage,
+        ResponseMessage,
+    },
     service::to_message_acceptance,
 };
 use fuel_core_poa::{
@@ -153,16 +157,30 @@ impl Bootstrap {
                     }
                     event = bootstrap.next_event() => {
                         // The bootstrap node only forwards data without validating it.
-                        if let Some(FuelP2PEvent::GossipsubMessage {
-                            peer_id,
-                            message_id,
-                            ..
-                        }) = event {
-                            bootstrap.report_message_validation_result(
-                                &message_id,
+                        match event {
+                            Some(FuelP2PEvent::GossipsubMessage {
                                 peer_id,
-                                to_message_acceptance(&GossipsubMessageAcceptance::Accept)
-                            )
+                                message_id,
+                                ..
+                            }) => {
+                                bootstrap.report_message_validation_result(
+                                    &message_id,
+                                    peer_id,
+                                    to_message_acceptance(&GossipsubMessageAcceptance::Accept)
+                                )
+                            },
+                            Some(FuelP2PEvent::InboundRequestMessage {
+                                request_id,
+                                request_message
+                            }) => {
+                                if request_message == RequestMessage::AllTransactionsIds {
+                                    let _ = bootstrap.send_response_msg(
+                                        request_id,
+                                        ResponseMessage::AllTransactionsIds(Some(vec![])),
+                                    );
+                                }
+                            }
+                            _ => {}
                         }
                     }
                 }

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

@@ -1,8 +1,12 @@
-use super::BlockImporterAdapter;
+use super::{
+    BlockImporterAdapter,
+    TxPoolAdapter,
+};
 use crate::database::OnChainIterableKeyValueView;
 use fuel_core_p2p::ports::{
     BlockHeightImporter,
     P2pDb,
+    TxPool,
 };
 use fuel_core_services::stream::BoxStream;
 use fuel_core_storage::Result as StorageResult;
@@ -11,6 +15,7 @@ use fuel_core_types::{
         consensus::Genesis,
         SealedBlockHeader,
     },
+    fuel_tx::Transaction,
     fuel_types::BlockHeight,
     services::p2p::Transactions,
 };
@@ -49,3 +54,20 @@ impl BlockHeightImporter for BlockImporterAdapter {
         )
     }
 }
+
+impl TxPool for TxPoolAdapter {
+    fn get_all_tx_ids(&self) -> Vec<fuel_core_txpool::types::TxId> {
+        self.service.get_all_tx_ids()
+    }
+
+    fn get_full_txs(
+        &self,
+        tx_ids: Vec<fuel_core_txpool::types::TxId>,
+    ) -> Vec<Option<Transaction>> {
+        self.service
+            .find(tx_ids)
+            .into_iter()
+            .map(|tx_info| tx_info.map(|tx| tx.tx().as_ref().into()))
+            .collect()
+    }
+}

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

@@ -65,6 +65,7 @@ impl BlockImporter for BlockImporterAdapter {
 }
 
 #[cfg(feature = "p2p")]
+#[async_trait::async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -91,6 +92,21 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
         }
     }
 
+    fn new_tx_subscription(&self) -> BoxStream<Vec<u8>> {
+        use tokio_stream::{
+            wrappers::BroadcastStream,
+            StreamExt,
+        };
+        if let Some(service) = &self.service {
+            Box::pin(
+                BroadcastStream::new(service.subscribe_new_tx_subscription())
+                    .filter_map(|result| result.ok()),
+            )
+        } else {
+            fuel_core_services::stream::IntoBoxStream::into_boxed(tokio_stream::pending())
+        }
+    }
+
     fn notify_gossip_transaction_validity(
         &self,
         message_info: GossipsubMessageInfo,
@@ -102,9 +118,48 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
             Ok(())
         }
     }
+
+    async fn request_tx_ids(
+        &self,
+        peer_id: Vec<u8>,
+    ) -> anyhow::Result<Vec<fuel_core_txpool::types::TxId>> {
+        if let Some(service) = &self.service {
+            match service.get_all_transactions_ids_from_peer(peer_id).await {
+                Ok(txs) => Ok(txs.unwrap_or_default()),
+                Err(e) => {
+                    tracing::error!("Error getting tx ids from peer: {:?}", e);
+                    Ok(vec![])
+                }
+            }
+        } else {
+            Ok(vec![])
+        }
+    }
+
+    async fn request_txs(
+        &self,
+        peer_id: Vec<u8>,
+        tx_ids: Vec<fuel_core_txpool::types::TxId>,
+    ) -> anyhow::Result<Vec<Option<Transaction>>> {
+        if let Some(service) = &self.service {
+            match service
+                .get_full_transactions_from_peer(peer_id, tx_ids)
+                .await
+            {
+                Ok(txs) => Ok(txs.unwrap_or_default()),
+                Err(e) => {
+                    tracing::error!("Error getting tx ids from peer: {:?}", e);
+                    Ok(vec![])
+                }
+            }
+        } else {
+            Ok(vec![])
+        }
+    }
 }
 
 #[cfg(not(feature = "p2p"))]
+#[async_trait::async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -126,6 +181,25 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     ) -> anyhow::Result<()> {
         Ok(())
     }
+
+    fn new_tx_subscription(&self) -> BoxStream<Vec<u8>> {
+        Box::pin(fuel_core_services::stream::pending())
+    }
+
+    async fn request_tx_ids(
+        &self,
+        _peer_id: Vec<u8>,
+    ) -> anyhow::Result<Vec<fuel_core_txpool::types::TxId>> {
+        Ok(vec![])
+    }
+
+    async fn request_txs(
+        &self,
+        _peer_id: Vec<u8>,
+        _tx_ids: Vec<fuel_core_txpool::types::TxId>,
+    ) -> anyhow::Result<Vec<Option<Transaction>>> {
+        Ok(vec![])
+    }
 }
 
 impl fuel_core_txpool::ports::TxPoolDb for OnChainIterableKeyValueView {

crates/fuel-core/src/service/sub_services.rs

@@ -73,7 +73,7 @@ pub type PoAService = fuel_core_poa::Service<
     InDirectoryPredefinedBlocks,
 >;
 #[cfg(feature = "p2p")]
-pub type P2PService = fuel_core_p2p::service::Service<Database>;
+pub type P2PService = fuel_core_p2p::service::Service<Database, TxPoolAdapter>;
 pub type TxPoolSharedState = fuel_core_txpool::service::SharedState<
     P2PAdapter,
     Database,
@@ -167,14 +167,10 @@ pub fn init_sub_services(
     };
 
     #[cfg(feature = "p2p")]
-    let mut network = config.p2p.clone().map(|p2p_config| {
-        fuel_core_p2p::service::new_service(
-            chain_id,
-            p2p_config,
-            database.on_chain().clone(),
-            importer_adapter.clone(),
-        )
-    });
+    let p2p_externals = config
+        .p2p
+        .clone()
+        .map(fuel_core_p2p::service::build_shared_state);
 
     #[cfg(feature = "p2p")]
     let p2p_adapter = {
@@ -190,7 +186,7 @@ pub fn init_sub_services(
             invalid_transactions: -100.,
         };
         P2PAdapter::new(
-            network.as_ref().map(|network| network.shared.clone()),
+            p2p_externals.as_ref().map(|ext| ext.0.clone()),
             peer_report_config,
         )
     };
@@ -227,6 +223,21 @@ pub fn init_sub_services(
     );
     let tx_pool_adapter = TxPoolAdapter::new(txpool.shared.clone());
 
+    #[cfg(feature = "p2p")]
+    let mut network = config.p2p.clone().zip(p2p_externals).map(
+        |(p2p_config, (shared_state, request_receiver))| {
+            fuel_core_p2p::service::new_service(
+                chain_id,
+                p2p_config,
+                shared_state,
+                request_receiver,
+                database.on_chain().clone(),
+                importer_adapter.clone(),
+                tx_pool_adapter.clone(),
+            )
+        },
+    );
+
     let block_producer = fuel_core_producer::Producer {
         config: config.block_producer.clone(),
         view_provider: database.on_chain().clone(),

crates/services/p2p/src/config.rs

@@ -49,6 +49,9 @@ pub const MAX_RESPONSE_SIZE: usize = 18 * 1024 * 1024;
 /// Maximum number of blocks per request.
 pub const MAX_HEADERS_PER_REQUEST: usize = 100;
 
+/// Maximum number of transactions ids asked per request.
+pub const MAX_TXS_PER_REQUEST: usize = 10000;
+
 #[derive(Clone, Debug)]
 pub struct Config<State = Initialized> {
     /// The keypair used for handshake during communication with other p2p nodes.
@@ -73,6 +76,9 @@ pub struct Config<State = Initialized> {
     pub max_block_size: usize,
     pub max_headers_per_request: usize,
 
+    // Maximum of txs id asked in a single request
+    pub max_txs_per_request: usize,
+
     // `DiscoveryBehaviour` related fields
     pub bootstrap_nodes: Vec<Multiaddr>,
     pub enable_mdns: bool,
@@ -151,6 +157,7 @@ impl Config<NotInitialized> {
             tcp_port: self.tcp_port,
             max_block_size: self.max_block_size,
             max_headers_per_request: self.max_headers_per_request,
+            max_txs_per_request: self.max_txs_per_request,
             bootstrap_nodes: self.bootstrap_nodes,
             enable_mdns: self.enable_mdns,
             max_peers_connected: self.max_peers_connected,
@@ -200,6 +207,7 @@ impl Config<NotInitialized> {
             tcp_port: 0,
             max_block_size: MAX_RESPONSE_SIZE,
             max_headers_per_request: MAX_HEADERS_PER_REQUEST,
+            max_txs_per_request: MAX_TXS_PER_REQUEST,
             bootstrap_nodes: vec![],
             enable_mdns: false,
             max_peers_connected: 50,