add implementations for pre conf task

.changes/added/2780.md

@@ -0,0 +1 @@
+Add implementations for the pre-confirmation signing task

crates/fuel-core/Cargo.toml

@@ -89,6 +89,8 @@ mockall = { workspace = true }
 proptest = { workspace = true }
 test-case = { workspace = true }
 test-strategy = { workspace = true }
+tokio-test = "0.4.4"
+tracing-subscriber = { workspace = true }
 
 [features]
 default = ["rocksdb"]

crates/fuel-core/src/lib.rs

@@ -3,6 +3,9 @@
 #![deny(unused_crate_dependencies)]
 #![deny(warnings)]
 
+#[cfg(test)]
+use tracing_subscriber as _;
+
 use crate::service::genesis::NotifyCancel;
 use tokio_util::sync::CancellationToken;
 

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

@@ -1,17 +1,5 @@
-use crate::{
-    database::{
-        database_description::relayer::Relayer,
-        Database,
-    },
-    fuel_core_graphql_api::ports::GasPriceEstimate,
-    service::{
-        sub_services::{
-            BlockProducerService,
-            TxPoolSharedState,
-        },
-        vm_pool::MemoryPool,
-    },
-};
+use std::sync::Arc;
+
 use fuel_core_consensus_module::{
     block_verifier::Verifier,
     RelayerConsensusConfig,
@@ -53,7 +41,21 @@ use fuel_core_types::{
 };
 //#[cfg(not(feature = "parallel-executor"))]
 use fuel_core_upgradable_executor::executor::Executor;
-use std::sync::Arc;
+
+use crate::{
+    database::{
+        database_description::relayer::Relayer,
+        Database,
+    },
+    fuel_core_graphql_api::ports::GasPriceEstimate,
+    service::{
+        sub_services::{
+            BlockProducerService,
+            TxPoolSharedState,
+        },
+        vm_pool::MemoryPool,
+    },
+};
 
 pub mod block_importer;
 pub mod chain_state_info_provider;
@@ -100,9 +102,10 @@ impl StaticGasPrice {
 mod universal_gas_price_provider_tests {
     #![allow(non_snake_case)]
 
-    use super::*;
     use proptest::proptest;
 
+    use super::*;
+
     fn _worst_case__correctly_calculates_value(
         gas_price: u64,
         starting_height: u32,

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

@@ -47,6 +47,8 @@ use tokio_stream::{
     StreamExt,
 };
 
+pub mod pre_confirmation_signature;
+
 impl PoAAdapter {
     pub fn new(shared_state: Option<SharedState>) -> Self {
         Self { shared_state }

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

@@ -0,0 +1,13 @@
+use fuel_core_types::{
+    fuel_tx::TxId,
+    services::p2p::PreconfirmationStatus,
+};
+
+pub mod broadcast;
+pub mod key_generator;
+pub mod parent_signature;
+pub mod signing_key;
+pub mod trigger;
+pub mod tx_receiver;
+
+pub type Preconfirmations = Vec<(TxId, PreconfirmationStatus)>;

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

@@ -0,0 +1,33 @@
+use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::{
+    parent_signature::FuelParentSignature,
+    signing_key::DummyKey,
+    Preconfirmations,
+};
+use fuel_core_poa::pre_confirmation_signature_service::{
+    broadcast::Broadcast,
+    error::Result as PoAResult,
+    Signed,
+};
+
+/// TODO: Implement `Broadcast` properly: <https://github.com/FuelLabs/fuel-core/issues/2783>
+pub struct P2PBroadcast;
+
+impl Broadcast for P2PBroadcast {
+    type PreConfirmations = Preconfirmations;
+    type ParentSignature = FuelParentSignature<DummyKey>;
+    type DelegateKey = DummyKey;
+
+    async fn broadcast_txs(
+        &mut self,
+        _txs: Signed<Self::DelegateKey, Self::PreConfirmations>,
+    ) -> PoAResult<()> {
+        todo!()
+    }
+
+    async fn broadcast_delegate_key(
+        &mut self,
+        _delegate_key: Self::ParentSignature,
+    ) -> PoAResult<()> {
+        todo!()
+    }
+}

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

@@ -0,0 +1,17 @@
+use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::{
+    signing_key::DummyKey,
+};
+use fuel_core_poa::pre_confirmation_signature_service::{
+    error::Result as PoAResult,
+    key_generator::KeyGenerator,
+};
+
+pub struct DelegateKeyGenerator;
+
+impl KeyGenerator for DelegateKeyGenerator {
+    type Key = DummyKey;
+
+    async fn generate(&mut self) -> PoAResult<Self::Key> {
+        todo!()
+    }
+}

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

@@ -0,0 +1,52 @@
+use fuel_core_poa::pre_confirmation_signature_service::{
+    error::{
+        Error as PoaError,
+        Result as PoAResult,
+    },
+    parent_signature::ParentSignature,
+};
+use fuel_core_types::{
+    fuel_crypto,
+    signer::SignMode,
+};
+
+pub struct FuelParentSigner<T> {
+    mode: SignMode,
+    _phantom: std::marker::PhantomData<T>,
+}
+
+pub struct FuelParentSignature<T> {
+    signature: fuel_core_types::fuel_vm::Signature,
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl<T> FuelParentSignature<T> {
+    pub fn signature(&self) -> fuel_core_types::fuel_vm::Signature {
+        self.signature
+    }
+}
+
+impl<T> From<fuel_core_types::fuel_vm::Signature> for FuelParentSignature<T> {
+    fn from(signature: fuel_core_types::fuel_vm::Signature) -> Self {
+        Self {
+            signature,
+            _phantom: std::marker::PhantomData,
+        }
+    }
+}
+
+impl<T> ParentSignature<T> for FuelParentSigner<T>
+where
+    T: Send + Sync,
+    T: Into<fuel_crypto::Message>,
+{
+    type SignedData = FuelParentSignature<T>;
+
+    async fn sign(&self, data: T) -> 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))
+        })?;
+        Ok(signature.into())
+    }
+}

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

@@ -0,0 +1,24 @@
+use fuel_core_poa::pre_confirmation_signature_service::signing_key::SigningKey;
+
+/// TODO: Decide what key to use for signing
+/// <https://github.com/FuelLabs/fuel-core/issues/2782>
+#[derive(Clone)]
+pub struct DummyKey;
+
+#[derive(Clone)]
+pub struct DummyKeySignature<T> {
+    _phantom: std::marker::PhantomData<T>,
+}
+
+impl SigningKey for DummyKey {
+    type Signature<T: Send + Clone> = DummyKeySignature<T>;
+
+    fn sign<T: Send + Clone>(
+        &self,
+        _data: T,
+    ) -> fuel_core_poa::pre_confirmation_signature_service::error::Result<
+        Self::Signature<T>,
+    > {
+        todo!()
+    }
+}

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

@@ -0,0 +1,71 @@
+use fuel_core_poa::pre_confirmation_signature_service::{
+    error::Result as PoAResult,
+    trigger::KeyRotationTrigger,
+};
+use std::time::Duration;
+
+#[allow(dead_code)]
+pub struct TimeBasedTrigger {
+    interval: tokio::time::Interval,
+}
+
+impl TimeBasedTrigger {
+    pub fn new(rotation_interval: Duration) -> Self {
+        Self {
+            interval: tokio::time::interval(rotation_interval),
+        }
+    }
+}
+
+impl KeyRotationTrigger for TimeBasedTrigger {
+    async fn next_rotation(&mut self) -> PoAResult<()> {
+        self.interval.tick().await;
+        Ok(())
+    }
+}
+#[cfg(test)]
+mod tests {
+    #![allow(non_snake_case)]
+    use super::*;
+
+    #[tokio::test]
+    async fn next_rotation__triggers_at_expected_time() {
+        tokio::time::pause();
+        // given
+        let rotation_interval = 10;
+        let rotation_interval_duration = Duration::from_secs(rotation_interval);
+
+        let mut trigger = TimeBasedTrigger::new(rotation_interval_duration);
+
+        // when
+        let mut fut = tokio_test::task::spawn(trigger.next_rotation());
+        tokio_test::assert_pending!(fut.poll());
+
+        // then
+        let advance_time = Duration::from_secs(rotation_interval + 1);
+        tokio::time::advance(advance_time).await;
+        tokio_test::assert_ready!(fut.poll()).expect("should trigger");
+    }
+
+    #[tokio::test]
+    async fn next_rotation__subsequent_triggers_are_same_interval() {
+        tokio::time::pause();
+        // given
+        let rotation_interval = 10;
+        let rotation_interval_duration = Duration::from_secs(rotation_interval);
+
+        let mut trigger = TimeBasedTrigger::new(rotation_interval_duration);
+
+        tokio::time::advance(Duration::from_secs(1)).await;
+        trigger.next_rotation().await.unwrap();
+
+        // when
+        let mut fut = tokio_test::task::spawn(trigger.next_rotation());
+        tokio_test::assert_pending!(fut.poll());
+
+        // then
+        let advance_time = Duration::from_secs(rotation_interval);
+        tokio::time::advance(advance_time).await;
+        tokio_test::assert_ready!(fut.poll()).expect("should trigger");
+    }
+}

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

@@ -0,0 +1,61 @@
+use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::Preconfirmations;
+use fuel_core_poa::pre_confirmation_signature_service::{
+    error::{
+        Error as PoaError,
+        Result as PoAResult,
+    },
+    tx_receiver::TxReceiver,
+};
+use fuel_core_types::{
+    fuel_tx::TxId,
+    services::p2p::PreconfirmationStatus,
+};
+
+pub struct MPSCTxReceiver<T> {
+    receiver: tokio::sync::mpsc::Receiver<T>,
+}
+
+impl TxReceiver for MPSCTxReceiver<Vec<(TxId, PreconfirmationStatus)>> {
+    type Txs = Preconfirmations;
+
+    async fn receive(&mut self) -> PoAResult<Self::Txs> {
+        self.receiver.recv().await.ok_or(PoaError::TxReceiver(
+            "Failed to receive transaction, channel closed".to_string(),
+        ))
+    }
+}
+#[cfg(test)]
+mod tests {
+    #![allow(non_snake_case)]
+    use super::*;
+    use fuel_core_types::fuel_types::BlockHeight;
+
+    #[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(),
+                PreconfirmationStatus::SuccessByBlockProducer {
+                    block_height: BlockHeight::from(123),
+                },
+            ),
+            (
+                TxId::default(),
+                PreconfirmationStatus::SqueezedOutByBlockProducer {
+                    reason: "test".to_string(),
+                },
+            ),
+        ];
+
+        let mut receiver = MPSCTxReceiver { receiver };
+
+        // when
+        sender.send(txs.clone()).await.unwrap();
+
+        // then
+        let received_txs = receiver.receive().await.unwrap();
+        assert_eq!(txs, received_txs);
+    }
+}