Integrate the new pre-confirmation signing task adapters

.changes/added/2784.md

@@ -0,0 +1 @@
+Integrate the pre conf signature task into the main consensus task

crates/fuel-core/src/service/adapters/consensus_module/poa/pre_confirmation_signature/parent_signature.rs

@@ -6,9 +6,9 @@ use fuel_core_poa::pre_confirmation_signature_service::{
     parent_signature::ParentSignature,
 };
 use fuel_core_types::{
-    fuel_crypto,
     signer::SignMode,
 };
+use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::signing_key::DummyKey;
 
 pub struct FuelParentSigner<T> {
     mode: SignMode,
@@ -35,14 +35,10 @@ impl<T> From<fuel_core_types::fuel_vm::Signature> for FuelParentSignature<T> {
     }
 }
 
-impl<T> ParentSignature<T> for FuelParentSigner<T>
-where
-    T: Send + Sync,
-    T: Into<fuel_crypto::Message>,
-{
-    type SignedData = FuelParentSignature<T>;
+impl ParentSignature<DummyKey> for FuelParentSigner<DummyKey> {
+    type SignedData = FuelParentSignature<DummyKey>;
 
-    async fn sign(&self, data: T) -> PoAResult<Self::SignedData> {
+    async fn sign(&self, data: DummyKey) -> PoAResult<Self::SignedData> {
         let message = data.into();
         let signature = self.mode.sign_message(message).await.map_err(|e| {
             PoaError::ParentSignature(format!("Failed to sign message: {}", e))

crates/fuel-core/src/service/adapters/consensus_module/poa/pre_confirmation_signature/signing_key.rs

@@ -1,4 +1,5 @@
 use fuel_core_poa::pre_confirmation_signature_service::signing_key::SigningKey;
+use fuel_core_types::fuel_crypto;
 
 /// TODO: Decide what key to use for signing
 /// <https://github.com/FuelLabs/fuel-core/issues/2782>
@@ -22,3 +23,9 @@ impl SigningKey for DummyKey {
         todo!()
     }
 }
+
+impl From<DummyKey> for fuel_crypto::Message {
+    fn from(_value: DummyKey) -> Self {
+        todo!()
+    }
+}

crates/fuel-core/src/service/adapters/consensus_module/poa/pre_confirmation_signature/tx_receiver.rs

@@ -4,26 +4,67 @@ use fuel_core_poa::pre_confirmation_signature_service::{
         Error as PoaError,
         Result as PoAResult,
     },
-    tx_receiver::TxReceiver,
+    tx_receiver::{
+        TxReceiver,
+        TxSender,
+    },
 };
 use fuel_core_types::{
     fuel_tx::TxId,
     services::p2p::PreconfirmationStatus,
 };
 
 pub struct MPSCTxReceiver<T> {
+    sender: tokio::sync::mpsc::Sender<T>,
     receiver: tokio::sync::mpsc::Receiver<T>,
 }
 
+impl Default for MPSCTxReceiver<Vec<(TxId, PreconfirmationStatus)>> {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl<T> MPSCTxReceiver<T> {
+    pub fn new() -> Self {
+        let (sender, receiver) = tokio::sync::mpsc::channel(1);
+        MPSCTxReceiver { sender, receiver }
+    }
+}
+
+#[derive(Clone)]
+pub struct MPSCTxSender<T> {
+    sender: tokio::sync::mpsc::Sender<T>,
+}
+
 impl TxReceiver for MPSCTxReceiver<Vec<(TxId, PreconfirmationStatus)>> {
     type Txs = Preconfirmations;
+    type Sender = MPSCTxSender<Vec<(TxId, PreconfirmationStatus)>>;
 
     async fn receive(&mut self) -> PoAResult<Self::Txs> {
         self.receiver.recv().await.ok_or(PoaError::TxReceiver(
             "Failed to receive transaction, channel closed".to_string(),
         ))
     }
+
+    fn get_sender(&self) -> Self::Sender {
+        MPSCTxSender {
+            sender: self.sender.clone(),
+        }
+    }
 }
+
+impl TxSender for MPSCTxSender<Vec<(TxId, PreconfirmationStatus)>> {
+    type Txs = Preconfirmations;
+
+    async fn send(&mut self, txs: Self::Txs) -> PoAResult<()> {
+        self.sender
+            .send(txs)
+            .await
+            .map_err(|e| PoaError::TxReceiver(format!("{}", e)))
+    }
+}
+
 #[cfg(test)]
 mod tests {
     #![allow(non_snake_case)]
@@ -33,7 +74,6 @@ mod tests {
     #[tokio::test]
     async fn receive__gets_what_is_sent_through_channel() {
         // given
-        let (sender, receiver) = tokio::sync::mpsc::channel(1);
         let txs = vec![
             (
                 TxId::default(),
@@ -49,7 +89,8 @@ mod tests {
             ),
         ];
 
-        let mut receiver = MPSCTxReceiver { receiver };
+        let mut receiver = MPSCTxReceiver::new();
+        let mut sender = receiver.get_sender();
 
         // when
         sender.send(txs.clone()).await.unwrap();

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

@@ -22,7 +22,21 @@ use crate::{
     service::{
         adapters::{
             chain_state_info_provider,
-            consensus_module::poa::InDirectoryPredefinedBlocks,
+            consensus_module::poa::{
+                pre_confirmation_signature::{
+                    broadcast::P2PBroadcast,
+                    key_generator::DelegateKeyGenerator,
+                    parent_signature::{
+                        FuelParentSignature,
+                        FuelParentSigner,
+                    },
+                    signing_key::DummyKey,
+                    trigger::TimeBasedTrigger,
+                    tx_receiver::MPSCTxReceiver,
+                    Preconfirmations,
+                },
+                InDirectoryPredefinedBlocks,
+            },
             fuel_gas_price_provider::FuelGasPriceProvider,
             graphql_api::GraphQLBlockImporter,
             import_result_provider::ImportResultProvider,
@@ -52,7 +66,11 @@ use fuel_core_gas_price_service::v1::{
     metadata::V1AlgorithmConfig,
     uninitialized_task::new_gas_price_service_v1,
 };
-use fuel_core_poa::Trigger;
+use fuel_core_poa::{
+    pre_confirmation_signature_service::PreConfirmationSignatureTask,
+    Trigger,
+};
+use fuel_core_services::ServiceRunner;
 use fuel_core_storage::{
     self,
     transactional::AtomicView,
@@ -70,6 +88,12 @@ pub type PoAService = fuel_core_poa::Service<
     SignMode,
     InDirectoryPredefinedBlocks,
     SystemTime,
+    MPSCTxReceiver<Preconfirmations>,
+    P2PBroadcast,
+    FuelParentSignature<DummyKey>,
+    DelegateKeyGenerator,
+    DummyKey,
+    TimeBasedTrigger<SystemTime>,
 >;
 #[cfg(feature = "p2p")]
 pub type P2PService = fuel_core_p2p::service::Service<Database, TxPoolAdapter>;
@@ -87,6 +111,9 @@ pub type GraphQL = fuel_core_graphql_api::api_service::Service;
 // TODO: Add to consensus params https://github.com/FuelLabs/fuel-vm/issues/888
 pub const DEFAULT_GAS_PRICE_CHANGE_PERCENT: u16 = 10;
 
+#[allow(unreachable_code)]
+#[allow(unused_variables)]
+#[allow(clippy::diverging_sub_expression)]
 pub fn init_sub_services(
     config: &Config,
     database: CombinedDatabase,
@@ -246,6 +273,7 @@ pub fn init_sub_services(
     let tx_pool_adapter = TxPoolAdapter::new(txpool.shared.clone());
 
     #[cfg(feature = "p2p")]
+    #[allow(unused_mut)]
     let mut network = config.p2p.clone().zip(p2p_externals).map(
         |(p2p_config, (shared_state, request_receiver))| {
             fuel_core_p2p::service::new_service(
@@ -276,6 +304,7 @@ pub fn init_sub_services(
     let poa_config: fuel_core_poa::Config = config.into();
     let mut production_enabled = !matches!(poa_config.trigger, Trigger::Never);
 
+    #[allow(unused_assignments)]
     if !production_enabled && config.debug {
         production_enabled = true;
         tracing::info!("Enabled manual block production because of `debug` flag");
@@ -296,6 +325,19 @@ pub fn init_sub_services(
 
     let predefined_blocks =
         InDirectoryPredefinedBlocks::new(config.predefined_blocks_path.clone());
+
+    #[allow(clippy::type_complexity)]
+    let pre_confirmation_service: ServiceRunner<
+        PreConfirmationSignatureTask<
+            MPSCTxReceiver<Preconfirmations>,
+            P2PBroadcast,
+            FuelParentSigner<DummyKey>,
+            DelegateKeyGenerator,
+            DummyKey,
+            TimeBasedTrigger<SystemTime>,
+        >,
+    > = todo!();
+
     let poa = production_enabled.then(|| {
         fuel_core_poa::new_service(
             &last_block_header,
@@ -307,6 +349,7 @@ pub fn init_sub_services(
             signer,
             predefined_blocks,
             SystemTime,
+            pre_confirmation_service,
         )
     });
     let poa_adapter = PoAAdapter::new(poa.as_ref().map(|service| service.shared.clone()));

crates/services/consensus_module/poa/src/pre_confirmation_signature_service.rs

@@ -1,5 +1,6 @@
 use error::Result;
 use fuel_core_services::{
+    RunnableService,
     RunnableTask,
     StateWatcher,
     TaskNextAction,
@@ -43,6 +44,40 @@ pub struct PreConfirmationSignatureTask<
     key_rotation_trigger: KeyRotationTrigger,
 }
 
+#[async_trait::async_trait]
+impl<TxRcv, Brdcst, Parent, Gen, DelegateKey, Trigger> RunnableService
+    for PreConfirmationSignatureTask<TxRcv, Brdcst, Parent, Gen, DelegateKey, Trigger>
+where
+    TxRcv: TxReceiver,
+    <TxRcv as TxReceiver>::Sender: Clone,
+    Brdcst: Broadcast<
+        DelegateKey = DelegateKey,
+        ParentSignature = Parent::SignedData,
+        PreConfirmations: From<TxRcv::Txs>,
+    >,
+    Parent: ParentSignature<DelegateKey>,
+    Gen: KeyGenerator<Key = DelegateKey>,
+    DelegateKey: SigningKey,
+    Trigger: KeyRotationTrigger,
+{
+    const NAME: &'static str = "PreConfirmationSignatureTask";
+    type SharedData = TxRcv::Sender;
+    type Task = Self;
+    type TaskParams = ();
+
+    fn shared_data(&self) -> Self::SharedData {
+        self.tx_receiver.get_sender()
+    }
+
+    async fn into_task(
+        self,
+        _state_watcher: &StateWatcher,
+        _params: Self::TaskParams,
+    ) -> anyhow::Result<Self::Task> {
+        todo!()
+    }
+}
+
 impl<TxRcv, Brdcst, Parent, Gen, DelegateKey, Trigger>
     PreConfirmationSignatureTask<TxRcv, Brdcst, Parent, Gen, DelegateKey, Trigger>
 where