Integrate the new pre-confirmation signing task adapters (#2784)

## Linked Issues/PRs
<!-- List of related issues/PRs -->
Part of #2739

## Description
<!-- List of detailed changes -->

## 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?

---------

Co-authored-by: Green Baneling <[email protected]>
Co-authored-by: AurelienFT <[email protected]>

Author: 3 people

.changes/added/2784.md

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

Cargo.lock

@@ -19,6 +19,8 @@ axum = { workspace = true }
 clap = { workspace = true, features = ["derive"] }
 cosmrs = { version = "0.21", optional = true }
 derive_more = { version = "0.99" }
+ed25519 = { version = "2.2.3", features = ["serde"] }
+ed25519-dalek = { version = "2.1.1", features = ["serde"] }
 enum-iterator = { workspace = true }
 fuel-core-chain-config = { workspace = true, features = ["std"] }
 fuel-core-compression = { workspace = true }
@@ -49,7 +51,7 @@ mockall = { workspace = true, optional = true }
 num_cpus = { version = "1.16.0", optional = true }
 parking_lot = { workspace = true }
 paste = { workspace = true }
-postcard = { workspace = true, optional = true }
+postcard = { workspace = true }
 rand = { workspace = true }
 rocksdb = { version = "0.21", default-features = false, features = [
   "lz4",
@@ -102,7 +104,7 @@ smt = [
 p2p = ["dep:fuel-core-p2p", "dep:fuel-core-sync"]
 relayer = ["dep:fuel-core-relayer"]
 shared-sequencer = ["dep:fuel-core-shared-sequencer", "dep:cosmrs"]
-rocksdb = ["dep:rocksdb", "dep:tempfile", "dep:num_cpus", "dep:postcard"]
+rocksdb = ["dep:rocksdb", "dep:tempfile", "dep:num_cpus"]
 backup = ["rocksdb", "fuel-core-database/backup"]
 test-helpers = [
   "fuel-core-database/test-helpers",

crates/fuel-core/Cargo.toml

@@ -1,13 +1,5 @@
-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.rs

@@ -1,32 +1,41 @@
 use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::{
-    parent_signature::FuelParentSignature,
-    signing_key::DummyKey,
-    Preconfirmations,
+    key_generator::Ed25519Key,
+    parent_signature::FuelParentSigner,
 };
 use fuel_core_poa::pre_confirmation_signature_service::{
-    broadcast::Broadcast,
+    broadcast::{
+        Broadcast,
+        PublicKey,
+        Signature,
+    },
     error::Result as PoAResult,
-    Signed,
+    parent_signature::ParentSignature,
+};
+use fuel_core_types::services::p2p::{
+    DelegatePreConfirmationKey,
+    Preconfirmation,
 };
 
 /// 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;
+    type DelegateKey = Ed25519Key;
+    type ParentKey = FuelParentSigner;
+    type Preconfirmations = Vec<Preconfirmation>;
 
-    async fn broadcast_txs(
+    async fn broadcast_preconfirmations(
         &mut self,
-        _txs: Signed<Self::DelegateKey, Self::PreConfirmations>,
+        _: Self::Preconfirmations,
+        _: Signature<Self>,
     ) -> PoAResult<()> {
         todo!()
     }
 
     async fn broadcast_delegate_key(
         &mut self,
-        _delegate_key: Self::ParentSignature,
+        _: DelegatePreConfirmationKey<PublicKey<Self>>,
+        _: <Self::ParentKey as ParentSignature>::Signature,
     ) -> PoAResult<()> {
         todo!()
     }

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

@@ -1,17 +1,56 @@
-use crate::service::adapters::consensus_module::poa::pre_confirmation_signature::{
-    signing_key::DummyKey,
-};
+use ed25519::signature::Signer;
+use ed25519_dalek::SigningKey as DalekSigningKey;
 use fuel_core_poa::pre_confirmation_signature_service::{
-    error::Result as PoAResult,
+    error::{
+        Error as PoAError,
+        Result as PoAResult,
+    },
     key_generator::KeyGenerator,
+    signing_key::SigningKey,
+};
+use fuel_core_types::fuel_crypto::SecretKey;
+use rand::{
+    prelude::StdRng,
+    SeedableRng,
 };
+use serde::Serialize;
+use std::ops::Deref;
 
-pub struct DelegateKeyGenerator;
+pub struct Ed25519KeyGenerator;
 
-impl KeyGenerator for DelegateKeyGenerator {
-    type Key = DummyKey;
+impl KeyGenerator for Ed25519KeyGenerator {
+    type Key = Ed25519Key;
 
     async fn generate(&mut self) -> PoAResult<Self::Key> {
-        todo!()
+        let mut rng = StdRng::from_entropy();
+        let secret = SecretKey::random(&mut rng);
+
+        Ok(Ed25519Key {
+            signer: DalekSigningKey::from_bytes(secret.deref()),
+        })
+    }
+}
+
+#[derive(Clone)]
+pub struct Ed25519Key {
+    signer: DalekSigningKey,
+}
+
+impl SigningKey for Ed25519Key {
+    type Signature = ed25519::Signature;
+    type PublicKey = ed25519_dalek::VerifyingKey;
+
+    fn public_key(&self) -> Self::PublicKey {
+        self.signer.verifying_key()
+    }
+
+    fn sign<T>(&self, data: &T) -> PoAResult<Self::Signature>
+    where
+        T: Serialize,
+    {
+        let bytes = postcard::to_allocvec(data)
+            .map_err(|e| PoAError::Signature(format!("{e:?}")))?;
+        let signature = self.signer.sign(&bytes);
+        Ok(signature)
     }
 }

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

@@ -7,46 +7,28 @@ use fuel_core_poa::pre_confirmation_signature_service::{
 };
 use fuel_core_types::{
     fuel_crypto,
+    fuel_vm::Signature,
     signer::SignMode,
 };
+use serde::Serialize;
 
-pub struct FuelParentSigner<T> {
+pub struct FuelParentSigner {
     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>;
+impl ParentSignature for FuelParentSigner {
+    type Signature = Signature;
 
-    async fn sign(&self, data: T) -> PoAResult<Self::SignedData> {
-        let message = data.into();
+    async fn sign<T>(&self, data: &T) -> PoAResult<Self::Signature>
+    where
+        T: Serialize + Send + Sync,
+    {
+        let bytes = postcard::to_allocvec(data)
+            .map_err(|e| PoaError::ParentSignature(format!("{e:?}")))?;
+        let message = fuel_crypto::Message::new(bytes);
         let signature = self.mode.sign_message(message).await.map_err(|e| {
             PoaError::ParentSignature(format!("Failed to sign message: {}", e))
         })?;
-        Ok(signature.into())
+        Ok(signature)
     }
 }

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

@@ -1,58 +1,84 @@
-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,
-};
+use fuel_core_types::services::p2p::Preconfirmation;
+use tokio::sync::mpsc;
+
+pub struct PreconfirmationsReceiver {
+    capacity: usize,
+    receiver: mpsc::Receiver<Vec<Preconfirmation>>,
+}
+
+impl Default for PreconfirmationsReceiver {
+    fn default() -> Self {
+        let (_, receiver) = mpsc::channel(1);
+        Self::new(receiver)
+    }
+}
 
 // TODO(#2739): Remove when integrated
 // link: https://github.com/FuelLabs/fuel-core/issues/2739
 #[allow(dead_code)]
-pub struct MPSCTxReceiver<T> {
-    receiver: tokio::sync::mpsc::Receiver<T>,
+impl PreconfirmationsReceiver {
+    pub fn new(receiver: mpsc::Receiver<Vec<Preconfirmation>>) -> Self {
+        let capacity = receiver.capacity();
+        PreconfirmationsReceiver { capacity, receiver }
+    }
 }
 
-impl TxReceiver for MPSCTxReceiver<Vec<(TxId, PreconfirmationStatus)>> {
-    type Txs = Preconfirmations;
+impl TxReceiver for PreconfirmationsReceiver {
+    type Txs = Vec<Preconfirmation>;
 
     async fn receive(&mut self) -> PoAResult<Self::Txs> {
-        self.receiver.recv().await.ok_or(PoaError::TxReceiver(
-            "Failed to receive transaction, channel closed".to_string(),
-        ))
+        let mut buffer = Vec::new();
+        let received = self.receiver.recv_many(&mut buffer, self.capacity).await;
+
+        if received == 0 {
+            return Err(PoaError::TxReceiver(
+                "Failed to receive transaction, channel closed".to_string(),
+            ));
+        }
+
+        Ok(buffer.into_iter().flatten().collect::<Vec<_>>())
     }
 }
+
 #[cfg(test)]
 mod tests {
     #![allow(non_snake_case)]
+
     use super::*;
-    use fuel_core_types::fuel_types::BlockHeight;
+    use fuel_core_types::{
+        fuel_tx::TxId,
+        fuel_types::BlockHeight,
+        services::p2p::PreconfirmationStatus,
+    };
 
     #[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 {
+            Preconfirmation {
+                tx_id: TxId::default(),
+                status: PreconfirmationStatus::SuccessByBlockProducer {
                     block_height: BlockHeight::from(123),
                 },
-            ),
-            (
-                TxId::default(),
-                PreconfirmationStatus::SqueezedOutByBlockProducer {
+            },
+            Preconfirmation {
+                tx_id: TxId::default(),
+                status: PreconfirmationStatus::SqueezedOutByBlockProducer {
                     reason: "test".to_string(),
                 },
-            ),
+            },
         ];
 
-        let mut receiver = MPSCTxReceiver { receiver };
+        let (sender, receiver) = mpsc::channel(1);
+
+        let mut receiver = PreconfirmationsReceiver::new(receiver);
 
         // when
         sender.send(txs.clone()).await.unwrap();

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

@@ -11,7 +11,10 @@ use fuel_core_storage::transactional::{
 use fuel_core_types::blockchain::primitives::DaBlockHeight;
 
 impl Transactional for Database<Relayer> {
-    type Transaction<'a> = StorageTransaction<&'a mut Self> where Self: 'a;
+    type Transaction<'a>
+        = StorageTransaction<&'a mut Self>
+    where
+        Self: 'a;
 
     fn transaction(&mut self) -> Self::Transaction<'_> {
         self.into_transaction()