Link and activate preconfirmations and add integrations tests

.changes/added/2848.md

@@ -0,0 +1 @@
+Link all components of preconfirmations and add E2E tests.

bin/fuel-core/Cargo.toml

@@ -28,6 +28,7 @@ dotenvy = { version = "0.15", optional = true }
 fuel-core = { workspace = true, features = ["wasm-executor"] }
 fuel-core-chain-config = { workspace = true }
 fuel-core-metrics = { workspace = true }
+fuel-core-poa = { workspace = true, optional = true }
 fuel-core-shared-sequencer = { workspace = true, optional = true }
 fuel-core-types = { workspace = true, features = ["std"] }
 hex = { workspace = true }
@@ -62,7 +63,7 @@ test-case = { workspace = true }
 default = ["env", "relayer", "rocksdb"]
 aws-kms = ["dep:aws-config", "dep:aws-sdk-kms", "fuel-core-types/aws-kms"]
 env = ["dep:dotenvy"]
-p2p = ["fuel-core/p2p", "const_format"]
+p2p = ["fuel-core/p2p", "const_format", "dep:fuel-core-poa"]
 shared-sequencer = [
   "dep:fuel-core-shared-sequencer",
   "fuel-core/shared-sequencer",

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

@@ -110,6 +110,8 @@ mod shared_sequencer;
 mod consensus;
 mod gas_price;
 mod graphql;
+#[cfg(feature = "p2p")]
+mod preconfirmation_signature_service;
 mod profiling;
 #[cfg(feature = "relayer")]
 mod relayer;
@@ -274,6 +276,11 @@ pub struct Command {
     #[cfg(feature = "p2p")]
     pub sync_args: p2p::SyncArgs,
 
+    #[cfg_attr(feature = "p2p", clap(flatten))]
+    #[cfg(feature = "p2p")]
+    pub pre_confirmation_signature_service_args:
+        preconfirmation_signature_service::PreconfirmationArgs,
+
     #[cfg_attr(feature = "shared-sequencer", clap(flatten))]
     #[cfg(feature = "shared-sequencer")]
     pub shared_sequencer_args: shared_sequencer::Args,
@@ -343,6 +350,8 @@ impl Command {
             p2p_args,
             #[cfg(feature = "p2p")]
             sync_args,
+            #[cfg(feature = "p2p")]
+            pre_confirmation_signature_service_args,
             #[cfg(feature = "shared-sequencer")]
             shared_sequencer_args,
             metrics,
@@ -406,6 +415,17 @@ impl Command {
             metrics.is_enabled(Module::P2P),
         )?;
 
+        #[cfg(feature = "p2p")]
+        let preconfirmation_signature_service_config =
+            fuel_core_poa::pre_confirmation_signature_service::config::Config {
+                key_rotation_interval: *pre_confirmation_signature_service_args
+                    .key_rotation_interval,
+                key_expiration_interval: *pre_confirmation_signature_service_args
+                    .key_expiration_interval,
+                echo_delegation_interval: *pre_confirmation_signature_service_args
+                    .echo_delegation_interval,
+            };
+
         let trigger: Trigger = poa_trigger.into();
 
         if trigger != Trigger::Never {
@@ -703,6 +723,8 @@ impl Command {
             p2p: p2p_cfg,
             #[cfg(feature = "p2p")]
             sync: sync_args.into(),
+            #[cfg(feature = "p2p")]
+            pre_confirmation_signature_service: preconfirmation_signature_service_config,
             #[cfg(feature = "shared-sequencer")]
             shared_sequencer: shared_sequencer_args.try_into()?,
             consensus_signer,

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

@@ -0,0 +1,24 @@
+#[derive(Debug, Clone, clap::Args)]
+pub struct PreconfirmationArgs {
+    /// The frequency at which we rotate the preconfirmation sub-key
+    #[clap(
+        long = "preconfirmation-key-rotation-frequency",
+        env,
+        default_value = "10m"
+    )]
+    pub key_rotation_interval: humantime::Duration,
+    /// The frequency at which the preconfirmation sub-key expires
+    #[clap(
+        long = "preconfirmation-key-expiration-frequency",
+        env,
+        default_value = "20m"
+    )]
+    pub key_expiration_interval: humantime::Duration,
+    /// The frequency at which the preconfirmation sub-key is echoed to the network
+    #[clap(
+        long = "preconfirmation-echo-delegation-frequency",
+        env,
+        default_value = "10s"
+    )]
+    pub echo_delegation_interval: humantime::Duration,
+}

bin/fuel-core/src/cli/snapshot.rs

@@ -290,7 +290,7 @@ mod tests {
             UtxoId,
         },
         fuel_types::ChainId,
-        services::txpool::TransactionExecutionStatus,
+        services::transaction_status::TransactionExecutionStatus,
         tai64::Tai64,
     };
     use itertools::Itertools;

crates/fuel-core/Cargo.toml

@@ -114,6 +114,7 @@ test-helpers = [
   "fuel-core-services/test-helpers",
   "fuel-core-shared-sequencer?/test-helpers",
   "fuel-core-importer/test-helpers",
+  "fuel-core-poa/test-helpers",
   "dep:mockall",
 ]
 # features to enable in production, but increase build times

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

@@ -21,7 +21,7 @@ use fuel_core_types::{
         TxPointer,
     },
     fuel_types::Address,
-    services::txpool::TransactionExecutionStatus,
+    services::transaction_status::TransactionExecutionStatus,
 };
 
 #[cfg(feature = "test-helpers")]

crates/fuel-core/src/executor.rs

@@ -17,7 +17,10 @@ mod tests {
     };
 
     #[cfg(not(feature = "wasm-executor"))]
-    use fuel_core_types::services::preconfirmation::PreconfirmationStatus;
+    use fuel_core_types::services::preconfirmation::{
+        Preconfirmation,
+        PreconfirmationStatus,
+    };
 
     use crate as fuel_core;
     use fuel_core::database::Database;
@@ -3117,19 +3120,19 @@ mod tests {
     async fn execute_block__send_preconfirmations() {
         // Given
         struct MockPreconfirmationsSender {
-            sender: tokio::sync::mpsc::Sender<Vec<PreconfirmationStatus>>,
+            sender: tokio::sync::mpsc::Sender<Vec<Preconfirmation>>,
         }
 
         impl PreconfirmationSenderPort for MockPreconfirmationsSender {
             fn try_send(
                 &self,
-                preconfirmations: Vec<PreconfirmationStatus>,
-            ) -> Vec<PreconfirmationStatus> {
+                preconfirmations: Vec<Preconfirmation>,
+            ) -> Vec<Preconfirmation> {
                 preconfirmations
             }
 
             /// Send a batch of pre-confirmations, awaiting for the send to be successful.
-            async fn send(&self, preconfirmations: Vec<PreconfirmationStatus>) {
+            async fn send(&self, preconfirmations: Vec<Preconfirmation>) {
                 self.sender.send(preconfirmations).await.unwrap();
             }
         }
@@ -3173,7 +3176,7 @@ mod tests {
         let preconfirmations = receiver.recv().await.unwrap();
         assert_eq!(preconfirmations.len(), 1);
         assert!(matches!(
-            preconfirmations[0],
+            preconfirmations[0].status,
             PreconfirmationStatus::Success { .. }
         ));
         assert_eq!(res.skipped_transactions.len(), 0);

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

@@ -64,7 +64,7 @@ use fuel_core_types::{
     fuel_vm::BlobData,
     services::{
         graphql_api::ContractBalance,
-        txpool::TransactionExecutionStatus,
+        transaction_status::TransactionExecutionStatus,
     },
 };
 use futures::Stream;

crates/fuel-core/src/graphql_api/ports.rs

@@ -70,7 +70,7 @@ use fuel_core_types::{
         },
         graphql_api::ContractBalance,
         p2p::PeerInfo,
-        txpool::{
+        transaction_status::{
             self,
             TransactionStatus,
         },
@@ -90,7 +90,7 @@ pub trait OffChainDatabase: Send + Sync {
     fn tx_status(
         &self,
         tx_id: &TxId,
-    ) -> StorageResult<txpool::TransactionExecutionStatus>;
+    ) -> StorageResult<transaction_status::TransactionExecutionStatus>;
 
     fn balance(
         &self,
@@ -363,7 +363,7 @@ pub mod worker {
         fuel_types::BlockHeight,
         services::{
             block_importer::SharedImportResult,
-            txpool::{
+            transaction_status::{
                 self,
                 TransactionStatus,
             },
@@ -431,8 +431,8 @@ pub mod worker {
         fn update_tx_status(
             &mut self,
             id: &Bytes32,
-            status: txpool::TransactionExecutionStatus,
-        ) -> StorageResult<Option<txpool::TransactionExecutionStatus>>;
+            status: transaction_status::TransactionExecutionStatus,
+        ) -> StorageResult<Option<transaction_status::TransactionExecutionStatus>>;
 
         /// Update metadata about the total number of transactions on the chain.
         /// Returns the total count after the update.

crates/fuel-core/src/graphql_api/storage.rs

@@ -32,7 +32,7 @@ use fuel_core_types::{
         Bytes32,
     },
     fuel_types::BlockHeight,
-    services::txpool::TransactionExecutionStatus,
+    services::transaction_status::TransactionExecutionStatus,
 };
 use statistic::StatisticTable;
 

crates/fuel-core/src/graphql_api/storage/transactions.rs

@@ -23,7 +23,7 @@ use fuel_core_types::{
         Bytes32,
     },
     fuel_types::BlockHeight,
-    services::txpool::TransactionExecutionStatus,
+    services::transaction_status::TransactionExecutionStatus,
 };
 use std::{
     array::TryFromSliceError,

crates/fuel-core/src/graphql_api/worker_service/tests.rs

@@ -10,7 +10,7 @@ use fuel_core_storage::StorageAsRef;
 use fuel_core_types::{
     fuel_tx::Bytes32,
     fuel_types::BlockHeight,
-    services::txpool::TransactionStatus,
+    services::transaction_status::TransactionStatus,
 };
 use std::sync::Arc;
 

crates/fuel-core/src/p2p_test_helpers.rs

@@ -364,7 +364,6 @@ pub async fn make_nodes(
         );
 
         let mut test_txs = Vec::with_capacity(0);
-        node_config.block_production = Trigger::Instant;
 
         if let Some((
             ProducerSetup {
@@ -396,6 +395,7 @@ pub async fn make_nodes(
             }
 
             node_config.utxo_validation = utxo_validation;
+            node_config.txpool.utxo_validation = utxo_validation;
             update_signing_key(&mut node_config, Input::owner(&secret.public_key()));
 
             node_config.consensus_signer = SignMode::Key(Secret::new(secret.into()));
@@ -422,6 +422,7 @@ pub async fn make_nodes(
             overrides,
         );
         node_config.block_production = Trigger::Never;
+        node_config.consensus_signer = SignMode::Unavailable;
 
         if let Some(ValidatorSetup {
             pub_key,

crates/fuel-core/src/query/message.rs

@@ -31,7 +31,7 @@ use fuel_core_types::{
         MessageId,
         Nonce,
     },
-    services::txpool::TransactionExecutionStatus,
+    services::transaction_status::TransactionExecutionStatus,
 };
 use futures::{
     Stream,
@@ -354,7 +354,7 @@ mod tests {
             BlockHeight,
             Nonce,
         },
-        services::txpool::TransactionExecutionStatus,
+        services::transaction_status::TransactionExecutionStatus,
         tai64::Tai64,
     };
 

crates/fuel-core/src/query/subscriptions.rs

@@ -3,7 +3,7 @@ use fuel_core_storage::Result as StorageResult;
 use fuel_core_tx_status_manager::TxStatusMessage;
 use fuel_core_types::{
     fuel_types::Bytes32,
-    services::txpool::TransactionStatus,
+    services::transaction_status::TransactionStatus,
 };
 use futures::{
     stream::BoxStream,

crates/fuel-core/src/query/subscriptions/test.rs

@@ -24,7 +24,7 @@ use fuel_core_tx_status_manager::TxStatusMessage;
 use fuel_core_txpool::error::RemovedReason;
 use fuel_core_types::{
     fuel_types::Bytes32,
-    services::txpool::TransactionStatus,
+    services::transaction_status::TransactionStatus,
     tai64::Tai64,
 };
 use futures::StreamExt;

crates/fuel-core/src/query/tx.rs

@@ -14,7 +14,7 @@ use fuel_core_types::{
         TxPointer,
     },
     fuel_types::Address,
-    services::txpool::TransactionExecutionStatus,
+    services::transaction_status::TransactionExecutionStatus,
 };
 use futures::{
     Stream,

crates/fuel-core/src/schema/tx.rs

@@ -84,7 +84,7 @@ use fuel_core_types::{
         CheckPredicateParams,
         EstimatePredicates,
     },
-    services::txpool,
+    services::transaction_status,
 };
 use futures::{
     Stream,
@@ -712,7 +712,7 @@ async fn submit_and_await_status<'a>(
                 Err(anyhow::anyhow!("Failed to get transaction status").into())
             }
         })
-        .take(2))
+        .take(3))
 }
 
 struct StatusChangeState<'a> {
@@ -724,7 +724,7 @@ impl<'a> TxnStatusChangeState for StatusChangeState<'a> {
     async fn get_tx_status(
         &self,
         id: Bytes32,
-    ) -> StorageResult<Option<txpool::TransactionStatus>> {
+    ) -> StorageResult<Option<transaction_status::TransactionStatus>> {
         match self.query.tx_status(&id) {
             Ok(status) => Ok(Some(status.into())),
             Err(StorageError::NotFound(_, _)) => {