Unify the work of runnable services

Cargo.toml

@@ -10,6 +10,7 @@ members = [
   "crates/database",
   "crates/fuel-core",
   "crates/metrics",
+  "crates/services",
   "crates/services/consensus_module/bft",
   "crates/services/consensus_module/poa",
   "crates/services/executor",

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

@@ -157,7 +157,4 @@ pub enum BlockProduction {
         #[serde(flatten)]
         trigger: fuel_core_poa::Trigger,
     },
-    // TODO:
-    // RoundRobin,
-    // ProofOfStake,
 }

crates/fuel-core/Cargo.toml

@@ -20,7 +20,6 @@ axum = { version = "0.5" }
 bincode = "1.3"
 derive_more = { version = "0.99" }
 enum-iterator = "1.2"
-fuel-core-bft = { path = "../services/consensus_module/bft", version = "0.15.1" }
 fuel-core-chain-config = { path = "../chain-config", version = "0.15.1" }
 fuel-core-database = { path = "../database", version = "0.15.1" }
 fuel-core-executor = { path = "../services/executor", version = "0.15.1" }
@@ -30,6 +29,7 @@ fuel-core-p2p = { path = "../services/p2p", version = "0.15.1", optional = true
 fuel-core-poa = { path = "../services/consensus_module/poa", version = "0.15.1" }
 fuel-core-producer = { path = "../services/producer", version = "0.15.1" }
 fuel-core-relayer = { path = "../services/relayer", version = "0.15.1", optional = true }
+fuel-core-services = { path = "../services", version = "0.15.1" }
 fuel-core-storage = { path = "../storage", version = "0.15.1" }
 fuel-core-sync = { path = "../services/sync", version = "0.15.1" }
 fuel-core-txpool = { path = "../services/txpool", version = "0.15.1" }

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

@@ -45,7 +45,7 @@ pub struct MaybeRelayerAdapter {
     pub relayer_synced: Option<RelayerSynced>,
 }
 
-pub struct PoACoordinatorAdapter {
+pub struct BlockProducerAdapter {
     pub block_producer: Arc<fuel_core_producer::Producer<Database>>,
 }
 

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

@@ -1,7 +1,7 @@
 use crate::{
     database::Database,
     service::adapters::{
-        PoACoordinatorAdapter,
+        BlockProducerAdapter,
         TxPoolAdapter,
     },
 };
@@ -49,7 +49,7 @@ impl TransactionPool for TxPoolAdapter {
 }
 
 #[async_trait::async_trait]
-impl fuel_core_poa::ports::BlockProducer<Database> for PoACoordinatorAdapter {
+impl fuel_core_poa::ports::BlockProducer<Database> for BlockProducerAdapter {
     async fn produce_and_execute_block(
         &self,
         height: BlockHeight,

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

@@ -76,7 +76,7 @@ pub async fn start_server(
         .data(modules.block_importer.clone())
         .data(modules.block_producer.clone())
         .data(modules.sync.clone())
-        .data(modules.coordinator.clone());
+        .data(modules.consensus_module.clone());
     let schema = dap::init(schema, params).extension(Tracing).finish();
 
     let router = Router::new()

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

@@ -6,14 +6,15 @@ use crate::{
     service::{
         adapters::{
             BlockImportAdapter,
+            BlockProducerAdapter,
             ExecutorAdapter,
             MaybeRelayerAdapter,
-            PoACoordinatorAdapter,
             TxPoolAdapter,
         },
         Config,
     },
 };
+use fuel_core_services::Service as ServiceTrait;
 use fuel_core_txpool::{
     ports::TxPoolDb,
     service::TxStatusChange,
@@ -30,11 +31,13 @@ use tokio::{
     task::JoinHandle,
 };
 
+type POA = fuel_core_poa::Service<Database, TxPoolAdapter, BlockProducerAdapter>;
+
 pub struct Modules {
     pub txpool: Arc<fuel_core_txpool::Service>,
     pub block_importer: Arc<fuel_core_importer::Service>,
     pub block_producer: Arc<fuel_core_producer::Producer<Database>>,
-    pub coordinator: Arc<CoordinatorService>,
+    pub consensus_module: POA,
     pub sync: Arc<fuel_core_sync::Service>,
     #[cfg(feature = "relayer")]
     pub relayer: Option<fuel_core_relayer::RelayerHandle>,
@@ -44,8 +47,8 @@ pub struct Modules {
 
 impl Modules {
     pub async fn stop(&self) {
+        self.consensus_module.stop_and_await().await.unwrap();
         let stops: Vec<JoinHandle<()>> = vec![
-            self.coordinator.stop().await,
             #[cfg(feature = "p2p")]
             self.network_service.stop().await,
             self.txpool.stop().await,
@@ -60,19 +63,6 @@ impl Modules {
     }
 }
 
-pub enum CoordinatorService {
-    Poa(fuel_core_poa::Service),
-    Bft(fuel_core_bft::Service),
-}
-impl CoordinatorService {
-    pub async fn stop(&self) -> Option<tokio::task::JoinHandle<()>> {
-        match self {
-            CoordinatorService::Poa(s) => s.stop().await,
-            CoordinatorService::Bft(s) => s.stop().await,
-        }
-    }
-}
-
 pub async fn start_modules(
     config: &Config,
     database: &Database,
@@ -84,20 +74,6 @@ pub async fn start_modules(
         fuel_core_importer::Service::new(&config.block_importer, db).await?;
     let sync = fuel_core_sync::Service::new(&config.sync).await?;
 
-    let coordinator = match &config.chain_conf.block_production {
-        BlockProduction::ProofOfAuthority { trigger } => {
-            CoordinatorService::Poa(fuel_core_poa::Service::new(&fuel_core_poa::Config {
-                trigger: *trigger,
-                block_gas_limit: config.chain_conf.block_gas_limit,
-                signing_key: config.consensus_key.clone(),
-                metrics: false,
-            }))
-        } /* TODO: enable when bft config is ready to use
-           * CoordinatorConfig::Bft { config } => {
-           *     CoordinatorService::Bft(fuel_core_bft::Service::new(config, db).await?)
-           * } */
-    };
-
     #[cfg(feature = "relayer")]
     let relayer = if config.relayer.eth_client.is_some() {
         Some(fuel_core_relayer::RelayerHandle::start(
@@ -171,25 +147,26 @@ pub async fn start_modules(
 
     block_importer.start().await;
 
-    match &coordinator {
-        CoordinatorService::Poa(poa) => {
-            poa.start(
-                TxPoolAdapter {
-                    service: txpool_service.clone(),
-                    tx_status_rx: txpool_service.tx_status_subscribe(),
-                },
-                block_import_tx,
-                PoACoordinatorAdapter {
-                    block_producer: block_producer.clone(),
-                },
-                database.clone(),
-            )
-            .await;
-        }
-        CoordinatorService::Bft(bft) => {
-            bft.start().await;
-        }
-    }
+    let poa = match &config.chain_conf.block_production {
+        BlockProduction::ProofOfAuthority { trigger } => fuel_core_poa::new_service(
+            fuel_core_poa::Config {
+                trigger: *trigger,
+                block_gas_limit: config.chain_conf.block_gas_limit,
+                signing_key: config.consensus_key.clone(),
+                metrics: false,
+            },
+            TxPoolAdapter {
+                service: txpool_service.clone(),
+                tx_status_rx: txpool_service.tx_status_subscribe(),
+            },
+            block_import_tx,
+            BlockProducerAdapter {
+                block_producer: block_producer.clone(),
+            },
+            database.clone(),
+        ),
+    };
+    poa.start()?;
 
     sync.start(
         block_event_receiver,
@@ -203,7 +180,7 @@ pub async fn start_modules(
         txpool: txpool_service,
         block_importer: Arc::new(block_importer),
         block_producer,
-        coordinator: Arc::new(coordinator),
+        consensus_module: poa,
         sync: Arc::new(sync),
         #[cfg(feature = "relayer")]
         relayer,

crates/services/Cargo.toml

@@ -0,0 +1,16 @@
+[package]
+name = "fuel-core-services"
+version = "0.15.1"
+authors = ["Fuel Labs <contact@fuel.sh>"]
+edition = "2021"
+homepage = "https://fuel.network/"
+keywords = ["blockchain", "fuel", "consensus", "bft"]
+license = "BUSL-1.1"
+repository = "https://github.com/FuelLabs/fuel-core"
+description = "The common code for fuel core services."
+
+[dependencies]
+anyhow = "1.0"
+async-trait = "0.1"
+tokio = { version = "1.21", features = ["full"] }
+tracing = "0.1"

crates/services/consensus_module/poa/Cargo.toml

@@ -12,10 +12,10 @@ version = "0.15.1"
 [dependencies]
 anyhow = "1.0"
 async-trait = "0.1"
+fuel-core-services = { path = "../.." }
 fuel-core-storage = { path = "../../../storage", version = "0.15.1" }
 fuel-core-types = { path = "../../../types", version = "0.15.1" }
 humantime-serde = "1.1.1"
-parking_lot = "0.12"
 serde = { version = "1.0", features = ["derive"] }
 tokio = { version = "1.21", features = ["full"] }
 tracing = "0.1"

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

@@ -11,4 +11,7 @@ pub use config::{
     Config,
     Trigger,
 };
-pub use service::Service;
+pub use service::{
+    new_service,
+    Service,
+};

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

@@ -23,7 +23,7 @@ use fuel_core_types::{
 };
 
 #[async_trait::async_trait]
-pub trait TransactionPool {
+pub trait TransactionPool: Sync + Send {
     /// Returns the number of pending transactions in the `TxPool`.
     async fn pending_number(&self) -> anyhow::Result<usize>;