WIP: Pooled transactions exchange protocol

CHANGELOG.md

@@ -10,6 +10,7 @@ Description of the upcoming release here.
 
 ### Added
 
+- [#1228](https://github.com/FuelLabs/fuel-core/pull/1228): Add a new capability of sharing pooled transactions when two nodes connect.
 - [#1432](https://github.com/FuelLabs/fuel-core/pull/1432): Add a new `--api-request-timeout` argument to control TTL for GraphQL requests.
 - [#1419](https://github.com/FuelLabs/fuel-core/pull/1419): Add additional "sanity" benchmarks for arithmetic op code instructions.
 - [#1411](https://github.com/FuelLabs/fuel-core/pull/1411): Added WASM and `no_std` compatibility

bin/e2e-test-client/tests/integration_tests.rs

@@ -41,6 +41,7 @@ async fn works_in_multinode_local_env() {
     let mut rng = StdRng::seed_from_u64(line!() as u64);
     let secret = SecretKey::random(&mut rng);
     let pub_key = Input::owner(&secret.public_key());
+    let disable_block_production = false;
     let Nodes {
         mut producers,
         mut validators,
@@ -51,6 +52,7 @@ async fn works_in_multinode_local_env() {
             ProducerSetup::new(secret).with_txs(1).with_name("Alice"),
         )],
         [Some(ValidatorSetup::new(pub_key).with_name("Bob"))],
+        disable_block_production,
     )
     .await;
 

crates/fuel-core/src/p2p_test_helpers.rs

@@ -148,6 +148,7 @@ pub async fn make_nodes(
     bootstrap_setup: impl IntoIterator<Item = Option<BootstrapSetup>>,
     producers_setup: impl IntoIterator<Item = Option<ProducerSetup>>,
     validators_setup: impl IntoIterator<Item = Option<ValidatorSetup>>,
+    disable_block_production: bool,
 ) -> Nodes {
     let producers: Vec<_> = producers_setup.into_iter().collect();
 
@@ -227,6 +228,7 @@ pub async fn make_nodes(
                             .then_some(name)
                             .unwrap_or_else(|| format!("b:{i}")),
                         chain_config.clone(),
+                        disable_block_production,
                     );
                     if let Some(BootstrapSetup { pub_key, .. }) = boot {
                         match &mut node_config.chain_conf.consensus {
@@ -252,6 +254,7 @@ pub async fn make_nodes(
                 .then_some(name)
                 .unwrap_or_else(|| format!("p:{i}")),
             chain_config.clone(),
+            disable_block_production,
         );
 
         let mut test_txs = Vec::with_capacity(0);
@@ -284,6 +287,7 @@ pub async fn make_nodes(
                 .then_some(name)
                 .unwrap_or_else(|| format!("v:{i}")),
             chain_config.clone(),
+            disable_block_production,
         );
         node_config.block_production = Trigger::Never;
         node_config.p2p.as_mut().unwrap().bootstrap_nodes = boots.clone();
@@ -305,10 +309,17 @@ pub async fn make_nodes(
     }
 }
 
-pub fn make_config(name: String, chain_config: ChainConfig) -> Config {
+pub fn make_config(
+    name: String,
+    chain_config: ChainConfig,
+    disable_block_production: bool,
+) -> Config {
     let mut node_config = Config::local_node();
     node_config.chain_conf = chain_config;
     node_config.utxo_validation = true;
+    if disable_block_production {
+        node_config.block_production = Trigger::Never;
+    }
     node_config.name = name;
     node_config
 }

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

@@ -5,6 +5,7 @@ use crate::{
         P2PAdapter,
     },
 };
+
 use fuel_core_services::stream::BoxStream;
 use fuel_core_storage::{
     not_found,
@@ -37,6 +38,7 @@ use fuel_core_types::{
         p2p::{
             GossipsubMessageAcceptance,
             GossipsubMessageInfo,
+            PeerId,
             TransactionGossipData,
         },
     },
@@ -57,6 +59,7 @@ impl BlockImporter for BlockImporterAdapter {
 }
 
 #[cfg(feature = "p2p")]
+#[async_trait::async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -68,6 +71,20 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
         }
     }
 
+    async fn send_pooled_transactions(
+        &self,
+        peer_id: PeerId,
+        transactions: Vec<Transaction>,
+    ) -> anyhow::Result<()> {
+        if let Some(service) = &self.service {
+            service
+                .send_pooled_transactions_to_peer(peer_id.into(), transactions)
+                .await
+        } else {
+            Ok(())
+        }
+    }
+
     fn gossiped_transaction_events(&self) -> BoxStream<Self::GossipedTransaction> {
         use tokio_stream::{
             wrappers::BroadcastStream,
@@ -94,9 +111,40 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
             Ok(())
         }
     }
+
+    fn incoming_pooled_transactions(&self) -> BoxStream<Vec<Transaction>> {
+        use tokio_stream::{
+            wrappers::BroadcastStream,
+            StreamExt,
+        };
+        if let Some(service) = &self.service {
+            Box::pin(
+                BroadcastStream::new(service.subscribe_to_incoming_pooled_transactions())
+                    .filter_map(|result| result.ok()),
+            )
+        } else {
+            fuel_core_services::stream::IntoBoxStream::into_boxed(tokio_stream::pending())
+        }
+    }
+
+    fn new_connection(&self) -> BoxStream<PeerId> {
+        use tokio_stream::{
+            wrappers::BroadcastStream,
+            StreamExt,
+        };
+        if let Some(service) = &self.service {
+            Box::pin(
+                BroadcastStream::new(service.subscribe_to_connections())
+                    .filter_map(|result| result.ok()),
+            )
+        } else {
+            fuel_core_services::stream::IntoBoxStream::into_boxed(tokio_stream::pending())
+        }
+    }
 }
 
 #[cfg(not(feature = "p2p"))]
+#[async_trait::async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -118,6 +166,22 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     ) -> anyhow::Result<()> {
         Ok(())
     }
+
+    fn new_connection(&self) -> BoxStream<PeerId> {
+        Box::pin(fuel_core_services::stream::pending())
+    }
+
+    fn incoming_pooled_transactions(&self) -> BoxStream<Vec<Transaction>> {
+        Box::pin(fuel_core_services::stream::pending())
+    }
+
+    async fn send_pooled_transactions(
+        &self,
+        _peer_id: PeerId,
+        _transactions: Vec<Transaction>,
+    ) -> anyhow::Result<()> {
+        Ok(())
+    }
 }
 
 impl fuel_core_txpool::ports::TxPoolDb for Database {

crates/services/p2p/src/lib.rs

@@ -8,7 +8,7 @@ mod p2p_service;
 mod peer_manager;
 mod peer_report;
 pub mod ports;
-mod request_response;
+pub mod request_response;
 pub mod service;
 
 pub use gossipsub::config as gossipsub_config;

crates/services/p2p/src/p2p_service.rs

@@ -298,6 +298,20 @@ impl<Codec: NetworkCodec> FuelP2PService<Codec> {
         }
     }
 
+    /// Sends a one way message to a peer.
+    /// It leverages the `RequestMessage` type to send the message.
+    /// But, unlike `send_request_msg`, it does not expect a
+    /// response through a response channel.
+    pub fn send_msg(
+        &mut self,
+        peer_id: PeerId,
+        message_request: RequestMessage,
+    ) -> RequestId {
+        self.swarm
+            .behaviour_mut()
+            .send_request_msg(message_request, &peer_id)
+    }
+
     /// Sends RequestMessage to a peer
     /// If the peer is not defined it will pick one at random
     pub fn send_request_msg(
@@ -783,7 +797,23 @@ mod tests {
     #[tokio::test]
     #[instrument]
     async fn p2p_service_works() {
-        build_service_from_config(Config::default_initialized("p2p_service_works")).await;
+        let mut fuel_p2p_service =
+            build_service_from_config(Config::default_initialized("p2p_service_works"))
+                .await;
+
+        loop {
+            match fuel_p2p_service.swarm.select_next_some().await {
+                SwarmEvent::NewListenAddr { .. } => {
+                    // listener address registered, we are good to go
+                    break
+                }
+                SwarmEvent::Behaviour(_) => {}
+                other_event => {
+                    tracing::error!("Unexpected event: {:?}", other_event);
+                    panic!("Unexpected event {other_event:?}")
+                }
+            }
+        }
     }
 
     // Single sentry node connects to multiple reserved nodes and `max_peers_allowed` amount of non-reserved nodes.
@@ -1623,6 +1653,12 @@ mod tests {
                                             }
                                         });
                                     }
+                                    RequestMessage::PooledTransactions(_) => {
+                                        // This test case isn't applicable here because
+                                        // we send a one way message containing the pooled
+                                        // transactions. The node doesn't respond with anything
+                                        // so we can't test the response.
+                                    }
                                 }
                             }
                         }
@@ -1654,6 +1690,12 @@ mod tests {
                                 let transactions = vec![Transactions(txs)];
                                 let _ = node_b.send_response_msg(*request_id, OutboundResponse::Transactions(Some(Arc::new(transactions))));
                             }
+                            RequestMessage::PooledTransactions(_) => {
+                                // This test case isn't applicable here because
+                                // we send a one way message containing the pooled
+                                // transactions. The node doesn't respond with anything
+                                // so we can't test the response.
+                            }
                         }
                     }
 

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

@@ -8,6 +8,7 @@ use fuel_core_types::{
         SealedBlock,
         SealedBlockHeader,
     },
+    fuel_tx::Transaction,
     fuel_types::BlockHeight,
     services::p2p::Transactions,
 };
@@ -22,7 +23,7 @@ use tokio::sync::oneshot;
 pub(crate) const REQUEST_RESPONSE_PROTOCOL_ID: &[u8] = b"/fuel/req_res/0.0.1";
 
 /// Max Size in Bytes of the Request Message
-pub(crate) const MAX_REQUEST_SIZE: usize = core::mem::size_of::<RequestMessage>();
+pub const MAX_REQUEST_SIZE: usize = 32 * 1024;
 
 // Peer receives a `RequestMessage`.
 // It prepares a response in form of `OutboundResponse`
@@ -38,6 +39,7 @@ pub(crate) const MAX_REQUEST_SIZE: usize = core::mem::size_of::<RequestMessage>(
 pub enum RequestMessage {
     Block(BlockHeight),
     SealedHeaders(Range<u32>),
+    PooledTransactions(Vec<Transaction>),
     Transactions(Range<u32>),
 }