Refactored P2P Service to implement RunnableService trait and use `…

…ServiceRunner` (#872)

* Refactored P2P Service to implement `RunnableService` trait and use `ServiceRunner`.

* Apply PR comments

* update event matching and rename GetPeersIds to GetPeerIds

* Removed `async_trait` from `PeerToPeer`

* remove panic from service

* Moved `Database` from `fuel_core_txpool::SharedState` into `TxPool` because it is only consumer of it.

* add todo, fix error

Co-authored-by: Voxelot <brandonkite92@gmail.com>

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

@@ -3,8 +3,8 @@ use fuel_core::{
     p2p::{
         config::{
             convert_to_libp2p_keypair,
+            Config,
             NotInitialized,
-            P2PConfig,
         },
         gossipsub_config::default_gossipsub_builder,
         Multiaddr,
@@ -136,7 +136,7 @@ pub struct P2PArgs {
 }
 
 impl P2PArgs {
-    pub fn into_config(self, metrics: bool) -> anyhow::Result<P2PConfig<NotInitialized>> {
+    pub fn into_config(self, metrics: bool) -> anyhow::Result<Config<NotInitialized>> {
         let local_keypair = {
             match self.keypair {
                 Some(path) => {
@@ -179,7 +179,7 @@ impl P2PArgs {
             Some(Duration::from_secs(self.random_walk))
         };
 
-        Ok(P2PConfig {
+        Ok(Config {
             keypair: local_keypair,
             network_name: self.network,
             checksum: Default::default(),

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

@@ -1,20 +1,11 @@
 use crate::{
     database::Database,
-    service::{
-        modules::TxPoolService,
-        Config,
-    },
+    service::Config,
 };
-#[cfg(feature = "p2p")]
-use fuel_core_p2p::service::Service as P2PService;
-#[cfg(feature = "relayer")]
-use fuel_core_relayer::RelayerSynced;
+use fuel_core_txpool::service::SharedState as TxPoolSharedState;
 use fuel_core_types::blockchain::SealedBlock;
 use std::sync::Arc;
-use tokio::{
-    sync::broadcast::Sender,
-    task::JoinHandle,
-};
+use tokio::sync::broadcast::Sender;
 
 pub mod poa;
 pub mod producer;
@@ -30,11 +21,11 @@ pub struct BlockImportAdapter {
 }
 
 pub struct TxPoolAdapter {
-    service: TxPoolService,
+    service: TxPoolSharedState<P2PAdapter, Database>,
 }
 
 impl TxPoolAdapter {
-    pub fn new(service: TxPoolService) -> Self {
+    pub fn new(service: TxPoolSharedState<P2PAdapter, Database>) -> Self {
         Self { service }
     }
 }
@@ -47,7 +38,7 @@ pub struct ExecutorAdapter {
 pub struct MaybeRelayerAdapter {
     pub database: Database,
     #[cfg(feature = "relayer")]
-    pub relayer_synced: Option<RelayerSynced>,
+    pub relayer_synced: Option<fuel_core_relayer::RelayerSynced>,
 }
 
 pub struct BlockProducerAdapter {
@@ -57,7 +48,7 @@ pub struct BlockProducerAdapter {
 #[cfg(feature = "p2p")]
 #[derive(Clone)]
 pub struct P2PAdapter {
-    service: Arc<P2PService>,
+    service: fuel_core_p2p::service::SharedState,
 }
 
 #[cfg(not(feature = "p2p"))]
@@ -66,32 +57,14 @@ pub struct P2PAdapter;
 
 #[cfg(feature = "p2p")]
 impl P2PAdapter {
-    pub fn new(service: Arc<P2PService>) -> Self {
+    pub fn new(service: fuel_core_p2p::service::SharedState) -> Self {
         Self { service }
     }
-
-    pub async fn stop(&self) -> Option<JoinHandle<()>> {
-        self.service.stop().await
-    }
-
-    pub async fn start(&self) -> anyhow::Result<()> {
-        self.service.start().await
-    }
 }
 
 #[cfg(not(feature = "p2p"))]
 impl P2PAdapter {
     pub fn new() -> Self {
         Default::default()
     }
-
-    pub async fn stop(&self) -> Option<JoinHandle<()>> {
-        None
-    }
-
-    pub async fn start(&self) -> anyhow::Result<()> {
-        Ok(())
-    }
 }
-
-// TODO: Create generic `Service` type that support `start` and `stop`.

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

@@ -27,15 +27,15 @@ use fuel_core_types::{
 
 impl TransactionPool for TxPoolAdapter {
     fn pending_number(&self) -> usize {
-        self.service.shared.pending_number()
+        self.service.pending_number()
     }
 
     fn total_consumable_gas(&self) -> u64 {
-        self.service.shared.total_consumable_gas()
+        self.service.total_consumable_gas()
     }
 
     fn remove_txs(&self, ids: Vec<TxId>) -> Vec<ArcPoolTx> {
-        self.service.shared.remove_txs(ids)
+        self.service.remove_txs(ids)
     }
 
     fn transaction_status_events(&self) -> BoxStream<TxStatus> {
@@ -44,7 +44,7 @@ impl TransactionPool for TxPoolAdapter {
             StreamExt,
         };
         Box::pin(
-            BroadcastStream::new(self.service.shared.tx_status_subscribe())
+            BroadcastStream::new(self.service.tx_status_subscribe())
                 .filter_map(|result| result.ok()),
         )
     }

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

@@ -35,7 +35,7 @@ impl TxPool for TxPoolAdapter {
         _block_height: BlockHeight,
         max_gas: u64,
     ) -> Vec<ArcPoolTx> {
-        self.service.shared.select_transactions(max_gas)
+        self.service.select_transactions(max_gas)
     }
 }
 

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

@@ -2,7 +2,6 @@ use crate::service::adapters::{
     BlockImportAdapter,
     P2PAdapter,
 };
-use async_trait::async_trait;
 use fuel_core_services::stream::BoxStream;
 use fuel_core_txpool::ports::BlockImport;
 use fuel_core_types::{
@@ -35,7 +34,6 @@ impl BlockImport for BlockImportAdapter {
 }
 
 #[cfg(feature = "p2p")]
-#[async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -54,19 +52,17 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
         )
     }
 
-    async fn notify_gossip_transaction_validity(
+    fn notify_gossip_transaction_validity(
         &self,
         message: &Self::GossipedTransaction,
         validity: GossipsubMessageAcceptance,
-    ) {
+    ) -> anyhow::Result<()> {
         self.service
             .notify_gossip_transaction_validity(message, validity)
-            .await;
     }
 }
 
 #[cfg(not(feature = "p2p"))]
-#[async_trait]
 impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
     type GossipedTransaction = TransactionGossipData;
 
@@ -81,11 +77,11 @@ impl fuel_core_txpool::ports::PeerToPeer for P2PAdapter {
         Box::pin(fuel_core_services::stream::pending())
     }
 
-    async fn notify_gossip_transaction_validity(
+    fn notify_gossip_transaction_validity(
         &self,
         _message: &Self::GossipedTransaction,
         _validity: GossipsubMessageAcceptance,
-    ) {
-        // no-op
+    ) -> anyhow::Result<()> {
+        Ok(())
     }
 }

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

@@ -19,8 +19,8 @@ use strum_macros::{
 
 #[cfg(feature = "p2p")]
 use fuel_core_p2p::config::{
+    Config as P2PConfig,
     NotInitialized,
-    P2PConfig,
 };
 
 #[derive(Clone, Debug)]

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

@@ -27,7 +27,9 @@ pub type PoAService =
     fuel_core_poa::Service<Database, TxPoolAdapter, BlockProducerAdapter>;
 #[cfg(feature = "relayer")]
 pub type RelayerService = fuel_core_relayer::Service<Database>;
-pub type TxPoolService = fuel_core_txpool::Service<P2PAdapter>;
+#[cfg(feature = "p2p")]
+pub type P2PService = fuel_core_p2p::service::Service<Database>;
+pub type TxPoolService = fuel_core_txpool::Service<P2PAdapter, Database>;
 
 pub struct Modules {
     pub txpool: TxPoolService,
@@ -36,15 +38,15 @@ pub struct Modules {
     #[cfg(feature = "relayer")]
     pub relayer: Option<RelayerService>,
     #[cfg(feature = "p2p")]
-    pub network_service: P2PAdapter,
+    pub network_service: P2PService,
 }
 
 impl Modules {
     pub async fn stop(&self) {
         self.consensus_module.stop_and_await().await.unwrap();
         self.txpool.stop_and_await().await.unwrap();
         #[cfg(feature = "p2p")]
-        self.network_service.stop().await;
+        self.network_service.stop_and_await().await.unwrap();
     }
 }
 
@@ -71,16 +73,15 @@ pub async fn start_modules(
         let genesis = p2p_db.get_genesis()?;
         let p2p_config = config.p2p.clone().init(genesis)?;
 
-        Arc::new(fuel_core_p2p::service::Service::new(p2p_config, p2p_db))
+        fuel_core_p2p::service::new_service(p2p_config, p2p_db)
     };
 
     #[cfg(feature = "p2p")]
-    let p2p_adapter = P2PAdapter::new(network_service);
+    let p2p_adapter = P2PAdapter::new(network_service.shared.clone());
     #[cfg(not(feature = "p2p"))]
     let p2p_adapter = P2PAdapter::new();
 
     let p2p_adapter = p2p_adapter;
-    p2p_adapter.start().await?;
 
     let importer_adapter = BlockImportAdapter::new(block_import_tx);
 
@@ -89,7 +90,7 @@ pub async fn start_modules(
         database.clone(),
         TxStatusChange::new(100),
         importer_adapter.clone(),
-        p2p_adapter.clone(),
+        p2p_adapter,
     );
 
     // restrict the max number of concurrent dry runs to the number of CPUs
@@ -98,7 +99,7 @@ pub async fn start_modules(
     let block_producer = Arc::new(fuel_core_producer::Producer {
         config: config.block_producer.clone(),
         db: database.clone(),
-        txpool: Box::new(TxPoolAdapter::new(txpool_service.clone())),
+        txpool: Box::new(TxPoolAdapter::new(txpool_service.shared.clone())),
         executor: Arc::new(ExecutorAdapter {
             database: database.clone(),
             config: config.clone(),
@@ -122,7 +123,7 @@ pub async fn start_modules(
                 signing_key: config.consensus_key.clone(),
                 metrics: false,
             },
-            TxPoolAdapter::new(txpool_service.clone()),
+            TxPoolAdapter::new(txpool_service.shared.clone()),
             // TODO: Pass Importer
             importer_adapter.tx,
             BlockProducerAdapter {
@@ -138,6 +139,8 @@ pub async fn start_modules(
         relayer.start().expect("Should start relayer")
     }
     txpool_service.start()?;
+    #[cfg(feature = "p2p")]
+    network_service.start()?;
 
     Ok(Modules {
         txpool: txpool_service,
@@ -146,6 +149,6 @@ pub async fn start_modules(
         #[cfg(feature = "relayer")]
         relayer,
         #[cfg(feature = "p2p")]
-        network_service: p2p_adapter,
+        network_service,
     })
 }

crates/services/p2p/Cargo.toml

@@ -16,6 +16,7 @@ async-trait = "0.1"
 bincode = "1.3"
 fuel-core-chain-config = { path = "../../chain-config", version = "0.15.1" }
 fuel-core-metrics = { path = "../../metrics", version = "0.15.1" } # TODO make this a feature
+fuel-core-services = { path = "../../services", version = "0.15.1" }
 fuel-core-storage = { path = "../../storage", version = "0.15.1" }
 fuel-core-types = { path = "../../types", features = [
     "serde",

crates/services/p2p/src/behavior.rs

@@ -1,6 +1,6 @@
 use crate::{
     codecs::NetworkCodec,
-    config::P2PConfig,
+    config::Config,
     discovery::{
         DiscoveryBehaviour,
         DiscoveryConfig,
@@ -71,7 +71,7 @@ pub struct FuelBehaviour<Codec: NetworkCodec> {
 }
 
 impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
-    pub fn new(p2p_config: &P2PConfig, codec: Codec) -> Self {
+    pub fn new(p2p_config: &Config, codec: Codec) -> Self {
         let local_public_key = p2p_config.keypair.public();
         let local_peer_id = PeerId::from_public_key(&local_public_key);
 

crates/services/p2p/src/config.rs

@@ -85,7 +85,7 @@ impl From<[u8; 32]> for Checksum {
 }
 
 #[derive(Clone, Debug)]
-pub struct P2PConfig<State = Initialized> {
+pub struct Config<State = Initialized> {
     /// The keypair used for for handshake during communication with other p2p nodes.
     pub keypair: Keypair,
 
@@ -155,12 +155,12 @@ pub struct Initialized(());
 #[derive(Clone, Debug)]
 pub struct NotInitialized;
 
-impl P2PConfig<NotInitialized> {
+impl Config<NotInitialized> {
     /// Inits the `P2PConfig` with some lazily loaded data.
-    pub fn init(self, mut genesis: Genesis) -> anyhow::Result<P2PConfig<Initialized>> {
+    pub fn init(self, mut genesis: Genesis) -> anyhow::Result<Config<Initialized>> {
         use fuel_core_chain_config::GenesisCommitment;
 
-        Ok(P2PConfig {
+        Ok(Config {
             keypair: self.keypair,
             network_name: self.network_name,
             checksum: genesis.root()?.into(),
@@ -198,7 +198,7 @@ pub fn convert_to_libp2p_keypair(
     Ok(Keypair::Secp256k1(secret_key.into()))
 }
 
-impl P2PConfig<NotInitialized> {
+impl Config<NotInitialized> {
     pub fn default(network_name: &str) -> Self {
         let keypair = Keypair::generate_secp256k1();
 
@@ -235,9 +235,9 @@ impl P2PConfig<NotInitialized> {
 }
 
 #[cfg(any(feature = "test-helpers", test))]
-impl P2PConfig<Initialized> {
+impl Config<Initialized> {
     pub fn default_initialized(network_name: &str) -> Self {
-        P2PConfig::<NotInitialized>::default(network_name)
+        Config::<NotInitialized>::default(network_name)
             .init(Default::default())
             .expect("Expected correct initialization of config")
     }
@@ -247,7 +247,7 @@ impl P2PConfig<Initialized> {
 /// TCP/IP, Websocket
 /// Noise as encryption layer
 /// mplex or yamux for multiplexing
-pub(crate) fn build_transport(p2p_config: &P2PConfig) -> Boxed<(PeerId, StreamMuxerBox)> {
+pub(crate) fn build_transport(p2p_config: &Config) -> Boxed<(PeerId, StreamMuxerBox)> {
     let transport = {
         let generate_tcp_transport =
             || TokioTcpTransport::new(TcpConfig::new().port_reuse(true).nodelay(true));