Freesig/sync e2e testing

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

@@ -244,6 +244,7 @@ impl Command {
             #[cfg(feature = "p2p")]
             sync: sync_args.into(),
             consensus_key,
+            name: String::default(),
         })
     }
 }

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

@@ -168,11 +168,26 @@ impl FromStr for ChainConfig {
 
 impl GenesisCommitment for ChainConfig {
     fn root(&self) -> anyhow::Result<MerkleRoot> {
-        // TODO: Hash settlement configuration, consensus block production
+        // # Dev-note: If `ChainConfig` got a new field, maybe we need to hash it too.
+        // Avoid using the `..` in the code below. Use `_` instead if you don't need to hash
+        // the field. Explicit fields help to prevent a bug of missing fields in the hash.
+        let ChainConfig {
+            chain_name,
+            block_gas_limit,
+            // Skip the `initial_state` bec
+            initial_state: _,
+            transaction_parameters,
+            gas_costs,
+            consensus,
+        } = self;
+
+        // TODO: Hash settlement configuration when it will be available.
         let config_hash = *Hasher::default()
-            .chain(self.block_gas_limit.to_be_bytes())
-            .chain(self.transaction_parameters.root()?)
-            .chain(self.chain_name.as_bytes())
+            .chain(chain_name.as_bytes())
+            .chain(block_gas_limit.to_be_bytes())
+            .chain(transaction_parameters.root()?)
+            .chain(gas_costs.root()?)
+            .chain(consensus.root()?)
             .finalize();
 
         Ok(config_hash)
@@ -188,3 +203,23 @@ impl GenesisCommitment for ConsensusParameters {
         Ok(params_hash.into())
     }
 }
+
+impl GenesisCommitment for GasCosts {
+    fn root(&self) -> anyhow::Result<MerkleRoot> {
+        // TODO: Define hash algorithm for `GasCosts`
+        let bytes = postcard::to_stdvec(&self)?;
+        let hash = Hasher::default().chain(bytes).finalize();
+
+        Ok(hash.into())
+    }
+}
+
+impl GenesisCommitment for ConsensusConfig {
+    fn root(&self) -> anyhow::Result<MerkleRoot> {
+        // TODO: Define hash algorithm for `ConsensusConfig`
+        let bytes = postcard::to_stdvec(&self)?;
+        let hash = Hasher::default().chain(bytes).finalize();
+
+        Ok(hash.into())
+    }
+}

crates/fuel-core/Cargo.toml

@@ -76,5 +76,6 @@ metrics = ["dep:fuel-core-metrics"]
 p2p = ["dep:fuel-core-p2p", "dep:fuel-core-sync"]
 relayer = ["dep:fuel-core-relayer"]
 rocksdb = ["dep:rocksdb", "dep:tempfile"]
+test-helpers = []
 # features to enable in production, but increase build times
 rocksdb-production = ["rocksdb?/jemalloc"]

crates/fuel-core/src/service.rs

@@ -26,7 +26,7 @@ use self::adapters::BlockImporterAdapter;
 
 pub mod adapters;
 pub mod config;
-pub(crate) mod genesis;
+pub mod genesis;
 pub mod metrics;
 pub mod sub_services;
 
@@ -44,6 +44,9 @@ pub struct SharedState {
     pub graph_ql: crate::fuel_core_graphql_api::service::SharedState,
     /// Subscribe to new block production.
     pub block_importer: BlockImporterAdapter,
+    #[cfg(feature = "test-helpers")]
+    /// The config of the service.
+    pub config: Config,
 }
 
 pub struct FuelService {
@@ -61,7 +64,7 @@ pub struct FuelService {
 
 impl FuelService {
     /// Creates a `FuelService` instance from service config
-    #[tracing::instrument(skip_all)]
+    #[tracing::instrument(skip_all, fields(name = %config.name))]
     pub fn new(database: Database, mut config: Config) -> anyhow::Result<Self> {
         Self::make_config_consistent(&mut config);
         let task = Task::new(database, config)?;
@@ -212,6 +215,7 @@ impl RunnableService for Task {
 
 #[async_trait::async_trait]
 impl RunnableTask for Task {
+    #[tracing::instrument(skip_all)]
     async fn run(&mut self, watcher: &mut StateWatcher) -> anyhow::Result<bool> {
         let mut stop_signals = vec![];
         for service in &self.services {
@@ -326,8 +330,8 @@ mod tests {
         // }
         #[cfg(feature = "p2p")]
         {
-            // p2p + sync
-            expected_services += 2;
+            // p2p
+            expected_services += 1;
         }
 
         // # Dev-note: Update the `expected_services` when we add/remove a new/old service.

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

@@ -62,6 +62,17 @@ impl PeerToPeerPort for P2PAdapter {
 
 #[async_trait::async_trait]
 impl BlockImporterPort for BlockImporterAdapter {
+    fn committed_height_stream(&self) -> BoxStream<BlockHeight> {
+        use futures::StreamExt;
+        fuel_core_services::stream::IntoBoxStream::into_boxed(
+            tokio_stream::wrappers::BroadcastStream::new(self.block_importer.subscribe())
+                .filter_map(|r| {
+                    futures::future::ready(
+                        r.ok().map(|r| *r.sealed_block.entity.header().height()),
+                    )
+                }),
+        )
+    }
     async fn execute_and_commit(&self, block: SealedBlock) -> anyhow::Result<()> {
         self.execute_and_commit(block).await
     }

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

@@ -50,6 +50,7 @@ pub struct Config {
     #[cfg(feature = "p2p")]
     pub sync: fuel_core_sync::Config,
     pub consensus_key: Option<Secret<SecretKeyWrapper>>,
+    pub name: String,
 }
 
 impl Config {
@@ -84,6 +85,7 @@ impl Config {
                 max_get_txns_requests: 10,
             },
             consensus_key: Some(Secret::new(default_consensus_dev_key().into())),
+            name: String::default(),
         }
     }
 }

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

@@ -66,7 +66,7 @@ use fuel_core_types::{
 use itertools::Itertools;
 
 /// Loads state from the chain config into database
-pub(crate) fn maybe_initialize_state(
+pub fn maybe_initialize_state(
     config: &Config,
     database: &Database,
 ) -> anyhow::Result<()> {

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

@@ -23,6 +23,7 @@ use crate::{
         SubServices,
     },
 };
+use fuel_core_poa::Trigger;
 use std::sync::Arc;
 use tokio::sync::{
     Mutex,
@@ -112,9 +113,12 @@ pub fn init_sub_services(
     };
     let producer_adapter = BlockProducerAdapter::new(block_producer);
 
+    let poa_config: fuel_core_poa::Config = config.try_into()?;
+    let production_enabled =
+        !matches!(poa_config.trigger, Trigger::Never) || config.manual_blocks_enabled;
     let poa = fuel_core_poa::new_service(
         last_height,
-        config.try_into()?,
+        poa_config,
         tx_pool_adapter,
         producer_adapter.clone(),
         importer_adapter.clone(),
@@ -166,17 +170,22 @@ pub fn init_sub_services(
         relayer: relayer_service.as_ref().map(|r| r.shared.clone()),
         graph_ql: graph_ql.shared.clone(),
         block_importer: importer_adapter,
+        #[cfg(feature = "test-helpers")]
+        config: config.clone(),
     };
 
     #[allow(unused_mut)]
     // `FuelService` starts and shutdowns all sub-services in the `services` order
     let mut services: SubServices = vec![
         // GraphQL should be shutdown first, so let's start it first.
         Box::new(graph_ql),
-        Box::new(poa),
         Box::new(txpool),
     ];
 
+    if production_enabled {
+        services.push(Box::new(poa));
+    }
+
     #[cfg(feature = "relayer")]
     if let Some(relayer) = relayer_service {
         services.push(Box::new(relayer));
@@ -185,7 +194,9 @@ pub fn init_sub_services(
     #[cfg(feature = "p2p")]
     {
         services.push(Box::new(network));
-        services.push(Box::new(sync));
+        if !production_enabled {
+            services.push(Box::new(sync));
+        }
     }
 
     Ok((services, shared))

crates/services/importer/src/importer.rs

@@ -155,6 +155,15 @@ where
         self._commit_result(result)
     }
 
+    #[tracing::instrument(
+        skip_all,
+        fields(
+            block_id = %result.result().sealed_block.entity.id(),
+            height = **result.result().sealed_block.entity.header().height(),
+            tx_status = ?result.result().tx_status,
+        ),
+        err
+    )]
     fn _commit_result<ExecutorDatabase>(
         &self,
         result: UncommittedResult<StorageTransaction<ExecutorDatabase>>,
@@ -221,6 +230,7 @@ where
 
         db_tx.commit()?;
 
+        tracing::info!("Committed block");
         let _ = self.broadcast.send(Arc::new(result));
         Ok(())
     }

crates/services/p2p/src/config.rs

@@ -206,7 +206,7 @@ impl Config<NotInitialized> {
             max_peers_connected: 50,
             max_connections_per_peer: 3,
             allow_private_addresses: true,
-            random_walk: Some(Duration::from_secs(5)),
+            random_walk: Some(Duration::from_millis(500)),
             connection_idle_timeout: Some(Duration::from_secs(120)),
             reserved_nodes: vec![],
             reserved_nodes_only_mode: false,

crates/services/p2p/src/discovery.rs

@@ -257,10 +257,6 @@ impl NetworkBehaviour for DiscoveryBehaviour {
             }
         }
 
-        if let Some(next_event) = self.pending_events.pop_front() {
-            return Poll::Ready(NetworkBehaviourAction::GenerateEvent(next_event))
-        }
-
         Poll::Pending
     }
 

crates/services/p2p/src/heartbeat/handler.rs

@@ -62,6 +62,12 @@ impl HeartbeatConfig {
     }
 }
 
+impl Default for HeartbeatConfig {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
 #[derive(Debug)]
 pub enum HeartbeatFailure {
     Timeout,

crates/services/p2p/src/lib.rs

@@ -1,5 +1,5 @@
 mod behavior;
-mod codecs;
+pub mod codecs;
 pub mod config;
 mod discovery;
 mod gossipsub;
@@ -13,9 +13,15 @@ pub mod service;
 pub use gossipsub::config as gossipsub_config;
 
 pub use libp2p::{
+    multiaddr::Protocol,
     Multiaddr,
     PeerId,
 };
 
+#[cfg(feature = "test-helpers")]
+pub mod network_service {
+    pub use crate::p2p_service::*;
+}
+
 #[cfg(test)]
 fuel_core_trace::enable_tracing!();

crates/services/p2p/src/p2p_service.rs

@@ -212,6 +212,11 @@ impl<Codec: NetworkCodec> FuelP2PService<Codec> {
         Ok(())
     }
 
+    #[cfg(feature = "test-helpers")]
+    pub fn listeners(&self) -> impl Iterator<Item = &Multiaddr> {
+        self.swarm.listeners()
+    }
+
     pub fn get_peers_ids(&self) -> impl Iterator<Item = &PeerId> {
         self.swarm.behaviour().get_peers_ids()
     }