Add preconfirmation gossip messages

.changes/added/2726.md

@@ -0,0 +1 @@
+Add a new gossip-sub message for transaction confirmations

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

@@ -205,6 +205,10 @@ pub struct P2PArgs {
     /// Number of threads to read from the tx pool.
     #[clap(long = "p2p-txpool-threads", default_value = "0", env)]
     pub tx_pool_threads: usize,
+
+    /// Subscribe to pre-confirmation gossip topic
+    #[clap(long = "subscribe-to-pre-confirmations", env)]
+    subscribe_to_pre_confirmations: bool,
 }
 
 #[derive(Debug, Clone, Args)]
@@ -349,6 +353,7 @@ impl P2PArgs {
             database_read_threads: self.database_read_threads,
             tx_pool_threads: self.tx_pool_threads,
             state: NotInitialized,
+            subscribe_to_pre_confirmations: self.subscribe_to_pre_confirmations,
         };
         Ok(Some(config))
     }

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

@@ -1,12 +1,16 @@
-use std::io;
-
-use fuel_core_types::fuel_tx::Transaction;
-
 use crate::gossipsub::messages::{
     GossipTopicTag,
     GossipsubBroadcastRequest,
     GossipsubMessage,
 };
+use fuel_core_types::{
+    fuel_tx::Transaction,
+    services::p2p::PreConfirmationMessage,
+};
+use std::{
+    io,
+    ops::Deref,
+};
 
 use super::{
     Decode,
@@ -22,16 +26,21 @@ pub struct GossipsubMessageHandler<Codec> {
 
 impl<Codec> GossipsubCodec for GossipsubMessageHandler<Codec>
 where
-    Codec:
-        Encode<Transaction, Error = io::Error> + Decode<Transaction, Error = io::Error>,
+    Codec: Encode<Transaction, Error = io::Error>
+        + Decode<Transaction, Error = io::Error>
+        + Encode<PreConfirmationMessage, Error = io::Error>
+        + Decode<PreConfirmationMessage, Error = io::Error>,
 {
     type RequestMessage = GossipsubBroadcastRequest;
     type ResponseMessage = GossipsubMessage;
 
     fn encode(&self, data: Self::RequestMessage) -> Result<Vec<u8>, io::Error> {
         match data {
             GossipsubBroadcastRequest::NewTx(tx) => {
-                Ok(self.codec.encode(&tx)?.into_bytes())
+                Ok(self.codec.encode(tx.deref())?.into_bytes())
+            }
+            GossipsubBroadcastRequest::TxPreConfirmations(msg) => {
+                Ok(self.codec.encode(msg.deref())?.into_bytes())
             }
         }
     }
@@ -45,6 +54,9 @@ where
             GossipTopicTag::NewTx => {
                 GossipsubMessage::NewTx(self.codec.decode(encoded_data)?)
             }
+            GossipTopicTag::TxPreConfirmations => {
+                GossipsubMessage::TxPreConfirmations(self.codec.decode(encoded_data)?)
+            }
         };
 
         Ok(decoded_response)

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

@@ -35,7 +35,10 @@ impl<T> Encode<T> for PostcardCodec
 where
     T: ?Sized + serde::Serialize,
 {
-    type Encoder<'a> = Cow<'a, [u8]> where T: 'a;
+    type Encoder<'a>
+        = Cow<'a, [u8]>
+    where
+        T: 'a;
     type Error = io::Error;
 
     fn encode<'a>(&self, value: &'a T) -> Result<Self::Encoder<'a>, Self::Error> {

crates/services/p2p/src/config.rs

@@ -146,6 +146,9 @@ pub struct Config<State = Initialized> {
     /// with the `NotInitialized` state. But it can be set into the `Initialized` state only with
     /// the `init` method.
     pub state: State,
+
+    /// If true, the node will subscribe to pre-confirmations topic
+    pub subscribe_to_pre_confirmations: bool,
 }
 
 /// The initialized state can be achieved only by the `init` function because `()` is private.
@@ -195,6 +198,7 @@ impl Config<NotInitialized> {
             database_read_threads: self.database_read_threads,
             tx_pool_threads: self.tx_pool_threads,
             state: Initialized(()),
+            subscribe_to_pre_confirmations: self.subscribe_to_pre_confirmations,
         })
     }
 }
@@ -249,6 +253,7 @@ impl Config<NotInitialized> {
             database_read_threads: 0,
             tx_pool_threads: 0,
             state: NotInitialized,
+            subscribe_to_pre_confirmations: false,
         }
     }
 }

crates/services/p2p/src/gossipsub/config.rs

@@ -1,4 +1,7 @@
-use super::topics::NEW_TX_GOSSIP_TOPIC;
+use super::topics::{
+    NEW_TX_GOSSIP_TOPIC,
+    TX_PRECONFIRMATIONS_GOSSIP_TOPIC,
+};
 use crate::{
     config::{
         Config,
@@ -49,6 +52,8 @@ const MESH_SIZE: usize = 8;
 // The weight applied to the score for delivering new transactions.
 const NEW_TX_GOSSIP_WEIGHT: f64 = 0.05;
 
+const TX_PRECONFIRMATIONS_GOSSIP_WEIGHT: f64 = 0.05;
+
 // The threshold for a peer's score to be considered for greylisting.
 // If a peer's score falls below this value, they will be greylisted.
 // Greylisting is a lighter form of banning, where the peer's messages might be ignored or given lower priority,
@@ -222,7 +227,13 @@ fn initialize_gossipsub(gossipsub: &mut gossipsub::Behaviour, p2p_config: &Confi
         .with_peer_score(peer_score_params, peer_score_thresholds)
         .expect("gossipsub initialized with peer score");
 
-    let topics = vec![(NEW_TX_GOSSIP_TOPIC, NEW_TX_GOSSIP_WEIGHT)];
+    let mut topics = vec![(NEW_TX_GOSSIP_TOPIC, NEW_TX_GOSSIP_WEIGHT)];
+    if p2p_config.subscribe_to_pre_confirmations {
+        topics.push((
+            TX_PRECONFIRMATIONS_GOSSIP_TOPIC,
+            TX_PRECONFIRMATIONS_GOSSIP_WEIGHT,
+        ));
+    }
 
     // subscribe to gossipsub topics with the network name suffix
     for (topic, weight) in topics {

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

@@ -2,6 +2,7 @@ use std::sync::Arc;
 
 use fuel_core_types::fuel_tx::Transaction;
 
+use fuel_core_types::services::p2p::PreConfirmationMessage;
 use serde::{
     Deserialize,
     Serialize,
@@ -12,6 +13,7 @@ use serde::{
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
 pub enum GossipTopicTag {
     NewTx,
+    TxPreConfirmations,
 }
 
 /// Takes `Arc<T>` and wraps it in a matching GossipsubBroadcastRequest
@@ -20,9 +22,11 @@ pub enum GossipTopicTag {
 #[derive(Debug, Clone)]
 pub enum GossipsubBroadcastRequest {
     NewTx(Arc<Transaction>),
+    TxPreConfirmations(Arc<PreConfirmationMessage>),
 }
 
 #[derive(Serialize, Deserialize, Debug, Clone)]
 pub enum GossipsubMessage {
     NewTx(Transaction),
+    TxPreConfirmations(PreConfirmationMessage),
 }

crates/services/p2p/src/gossipsub/topics.rs

@@ -10,22 +10,26 @@ use super::messages::{
 };
 
 pub const NEW_TX_GOSSIP_TOPIC: &str = "new_tx";
+pub const TX_PRECONFIRMATIONS_GOSSIP_TOPIC: &str = "tx_preconfirmations";
 
 /// Holds used Gossipsub Topics
 /// Each field contains TopicHash of existing topics
 /// in order to avoid converting topics to TopicHash on each received message
 #[derive(Debug)]
 pub struct GossipsubTopics {
     new_tx_topic: TopicHash,
+    tx_confirmations_topic: TopicHash,
 }
 
 impl GossipsubTopics {
     pub fn new(network_name: &str) -> Self {
         let new_tx_topic: Sha256Topic =
             Topic::new(format!("{NEW_TX_GOSSIP_TOPIC}/{network_name}"));
-
+        let tx_confirmations_topic: Sha256Topic =
+            Topic::new(format!("{TX_PRECONFIRMATIONS_GOSSIP_TOPIC}/{network_name}"));
         Self {
             new_tx_topic: new_tx_topic.hash(),
+            tx_confirmations_topic: tx_confirmations_topic.hash(),
         }
     }
 
@@ -36,6 +40,9 @@ impl GossipsubTopics {
     ) -> Option<GossipTopicTag> {
         match incoming_topic {
             hash if hash == &self.new_tx_topic => Some(GossipTopicTag::NewTx),
+            hash if hash == &self.tx_confirmations_topic => {
+                Some(GossipTopicTag::TxPreConfirmations)
+            }
             _ => None,
         }
     }
@@ -48,6 +55,9 @@ impl GossipsubTopics {
     ) -> TopicHash {
         match outgoing_request {
             GossipsubBroadcastRequest::NewTx(_) => self.new_tx_topic.clone(),
+            GossipsubBroadcastRequest::TxPreConfirmations(_) => {
+                self.tx_confirmations_topic.clone()
+            }
         }
     }
 }