Add preconfirmation signature verification to tx status manager (#2856)

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

## Description
<!-- List of detailed changes -->
Adds generic logic to the tx status manager task to accept delegate keys
and check signatures with them.

The logic of actually tracking the delegate keys and doing verification
is behind the `SignatureVerification` trait, which will be implemented
in a follow-up PR.


## Checklist
- [x] New behavior is reflected in tests

### Before requesting review
- [ ] I have reviewed the code myself

---------

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

Author: 3 people

.changes/added/2856.md

@@ -0,0 +1 @@
+Add generic logic for managing the signatures and delegate keys for pre-confirmations signatures

Cargo.lock

@@ -37,17 +37,17 @@ pub struct PoAV2 {
 
 impl PoAV2 {
     pub fn new(
-        genesis_signing_key: Address,
-        signing_key_overrides: BTreeMap<BlockHeight, Address>,
+        genesis_signing_key_address: Address,
+        signing_key_address_overrides: BTreeMap<BlockHeight, Address>,
     ) -> Self {
         PoAV2 {
-            genesis_signing_key,
-            signing_key_overrides,
+            genesis_signing_key: genesis_signing_key_address,
+            signing_key_overrides: signing_key_address_overrides,
         }
     }
 
-    /// Returns the signing key for the given block height.
-    pub fn signing_key_at(&self, height: BlockHeight) -> Address {
+    /// Returns the address for the signing key at block height.
+    pub fn address_for_height(&self, height: BlockHeight) -> Address {
         if self.signing_key_overrides.is_empty() {
             self.genesis_signing_key
         } else {
@@ -60,6 +60,15 @@ impl PoAV2 {
         }
     }
 
+    /// Returns the address of the latest signing key for the given block height.
+    pub fn latest_address(&self) -> Address {
+        self.signing_key_overrides
+            .last_key_value()
+            .map(|(_, key)| key)
+            .cloned()
+            .unwrap_or(self.genesis_signing_key)
+    }
+
     /// Returns overrides for all the signing keys.
     pub fn get_all_overrides(&self) -> &BTreeMap<BlockHeight, Address> {
         &self.signing_key_overrides
@@ -73,10 +82,12 @@ impl PoAV2 {
 
 #[cfg(test)]
 mod tests {
+    #![allow(non_snake_case)]
+
     use super::*;
 
     #[test]
-    fn signing_key_at_works() {
+    fn address_at_height__returns_expected_values() {
         // Given
         let genesis_signing_key = Address::from([1; 32]);
         let signing_key_after_10 = Address::from([2; 32]);
@@ -95,16 +106,16 @@ mod tests {
         };
 
         // When/Then
-        assert_eq!(poa.signing_key_at(0u32.into()), genesis_signing_key);
-        assert_eq!(poa.signing_key_at(9u32.into()), genesis_signing_key);
-        assert_eq!(poa.signing_key_at(10u32.into()), signing_key_after_10);
-        assert_eq!(poa.signing_key_at(19u32.into()), signing_key_after_10);
-        assert_eq!(poa.signing_key_at(20u32.into()), signing_key_after_20);
-        assert_eq!(poa.signing_key_at(29u32.into()), signing_key_after_20);
-        assert_eq!(poa.signing_key_at(30u32.into()), signing_key_after_30);
-        assert_eq!(poa.signing_key_at(40u32.into()), signing_key_after_30);
+        assert_eq!(poa.address_for_height(0u32.into()), genesis_signing_key);
+        assert_eq!(poa.address_for_height(9u32.into()), genesis_signing_key);
+        assert_eq!(poa.address_for_height(10u32.into()), signing_key_after_10);
+        assert_eq!(poa.address_for_height(19u32.into()), signing_key_after_10);
+        assert_eq!(poa.address_for_height(20u32.into()), signing_key_after_20);
+        assert_eq!(poa.address_for_height(29u32.into()), signing_key_after_20);
+        assert_eq!(poa.address_for_height(30u32.into()), signing_key_after_30);
+        assert_eq!(poa.address_for_height(40u32.into()), signing_key_after_30);
         assert_eq!(
-            poa.signing_key_at(4_000_000u32.into()),
+            poa.address_for_height(4_000_000u32.into()),
             signing_key_after_30
         );
     }

crates/chain-config/src/config/consensus.rs

@@ -1,7 +1,11 @@
-use fuel_core_services::stream::BoxStream;
-use fuel_core_tx_status_manager::ports::P2PPreConfirmationGossipData;
-
 use super::P2PAdapter;
+use fuel_core_chain_config::ConsensusConfig;
+use fuel_core_services::stream::BoxStream;
+use fuel_core_tx_status_manager::{
+    ports::P2PPreConfirmationGossipData,
+    service::ProtocolPublicKey,
+};
+use fuel_core_types::fuel_tx::Address;
 
 #[cfg(feature = "p2p")]
 impl fuel_core_tx_status_manager::ports::P2PSubscriptions for P2PAdapter {
@@ -32,3 +36,22 @@ impl fuel_core_tx_status_manager::ports::P2PSubscriptions for P2PAdapter {
         Box::pin(fuel_core_services::stream::pending())
     }
 }
+
+pub struct ConsensusConfigProtocolPublicKey {
+    inner: ConsensusConfig,
+}
+
+impl ConsensusConfigProtocolPublicKey {
+    pub fn new(inner: ConsensusConfig) -> Self {
+        Self { inner }
+    }
+}
+
+impl ProtocolPublicKey for ConsensusConfigProtocolPublicKey {
+    fn latest_address(&self) -> Address {
+        match &self.inner {
+            ConsensusConfig::PoA { signing_key } => *signing_key,
+            ConsensusConfig::PoAV2(poa_v2) => poa_v2.latest_address(),
+        }
+    }
+}

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

@@ -41,6 +41,15 @@ use crate::service::adapters::consensus_module::poa::pre_confirmation_signature:
     tx_receiver::PreconfirmationsReceiver,
 };
 
+use super::{
+    adapters::{
+        FuelBlockSigner,
+        P2PAdapter,
+        TxStatusManagerAdapter,
+    },
+    genesis::create_genesis_block,
+    DbType,
+};
 use crate::{
     combined_database::CombinedDatabase,
     database::Database,
@@ -58,6 +67,7 @@ use crate::{
             graphql_api::GraphQLBlockImporter,
             import_result_provider::ImportResultProvider,
             ready_signal::ReadySignal,
+            tx_status_manager::ConsensusConfigProtocolPublicKey,
             BlockImporterAdapter,
             BlockProducerAdapter,
             ChainStateInfoProvider,
@@ -76,16 +86,6 @@ use crate::{
     },
 };
 
-use super::{
-    adapters::{
-        FuelBlockSigner,
-        P2PAdapter,
-        TxStatusManagerAdapter,
-    },
-    genesis::create_genesis_block,
-    DbType,
-};
-
 pub type PoAService = fuel_core_poa::Service<
     TxPoolAdapter,
     BlockProducerAdapter,
@@ -257,9 +257,13 @@ pub fn init_sub_services(
         universal_gas_price_provider.clone(),
     );
 
+    let protocol_pubkey =
+        ConsensusConfigProtocolPublicKey::new(chain_config.consensus.clone());
+
     let tx_status_manager = fuel_core_tx_status_manager::new_service(
         p2p_adapter.clone(),
         config.tx_status_manager.clone(),
+        protocol_pubkey,
     );
     let tx_status_manager_adapter =
         TxStatusManagerAdapter::new(tx_status_manager.shared.clone());

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

@@ -31,7 +31,7 @@ pub fn verify_consensus(
         ConsensusConfig::PoAV2(poa) => {
             let id = header.id();
             let m = id.as_message();
-            let signing_key = poa.signing_key_at(*header.height());
+            let signing_key = poa.address_for_height(*header.height());
             consensus
                 .signature
                 .recover(m)

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

@@ -14,9 +14,10 @@ async-trait = { workspace = true }
 derive_more = { workspace = true }
 fuel-core-metrics = { workspace = true }
 fuel-core-services = { workspace = true }
-fuel-core-types = { workspace = true, features = ["std"] }
+fuel-core-types = { workspace = true, features = ["std", "serde"] }
 futures = { workspace = true }
 parking_lot = { workspace = true }
+postcard = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }
 tracing = { workspace = true }

crates/services/tx_status_manager/Cargo.toml

@@ -9,7 +9,7 @@ pub mod config;
 mod error;
 mod manager;
 pub mod ports;
-mod service;
+pub mod service;
 mod subscriptions;
 mod tx_status_stream;
 mod update_sender;