Add preconfirmation gossip messages (#2726)

## Linked Issues/PRs
<!-- List of related issues/PRs -->
Closes #2738

## Description
<!-- List of detailed changes -->
This is part of the greater tx confirmation work. This will enable the
block producer to gossip a batch of signed transactions around the
network for the senders of those transactions to receive.

~In an attempt to cleanup the P2P code, some of the tests were moved to
their own files (this makes reviewing slightly more difficult, so I've
marked the new tests with comments in those refactored test suites).~

TODO:

- [x] Finalize the `TxConfirmations` type
    - [x] What is the correct type for `signature`?
    - [x] What are all the statuses for the `txs`?
- [x] Default to having the Preconfirmation topic off to minimize
traffic, turn on with CLI arg
- [ ] ~Constrain peers for gossiping to minimize traffic~
#2740

## Checklist
- [ ] Breaking changes are clearly marked as such in the PR description
and changelog
- [ ] New behavior is reflected in tests
- [ ] [The specification](https://github.com/FuelLabs/fuel-specs/)
matches the implemented behavior (link update PR if changes are needed)

### Before requesting review
- [ ] I have reviewed the code myself
- [ ] I have created follow-up issues caused by this PR and linked them
here

### After merging, notify other teams

[Add or remove entries as needed]

- [ ] [Rust SDK](https://github.com/FuelLabs/fuels-rs/)
- [ ] [Sway compiler](https://github.com/FuelLabs/sway/)
- [ ] [Platform
documentation](https://github.com/FuelLabs/devrel-requests/issues/new?assignees=&labels=new+request&projects=&template=NEW-REQUEST.yml&title=%5BRequest%5D%3A+)
(for out-of-organization contributors, the person merging the PR will do
this)
- [ ] Someone else?

Author: MitchTurner

.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()
+            }
         }
     }
 }