Don't capture signals within lib (#991)

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

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

@@ -257,7 +257,14 @@ pub async fn exec(command: Command) -> anyhow::Result<()> {
     // initialize the server
     let server = FuelService::new_node(config).await?;
     // pause the main task while service is running
-    server.await_stop().await?;
+    tokio::select! {
+        result = server.await_stop() => {
+            result?;
+        }
+        _ = shutdown_signal() => {}
+    }
+
+    server.stop_and_await().await?;
 
     Ok(())
 }
@@ -281,3 +288,32 @@ fn load_consensus_key(
         Ok(None)
     }
 }
+
+async fn shutdown_signal() -> anyhow::Result<()> {
+    #[cfg(unix)]
+    {
+        let mut sigterm =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
+
+        let mut sigint =
+            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
+        loop {
+            tokio::select! {
+                _ = sigterm.recv() => {
+                    tracing::info!("sigterm received");
+                    break;
+                }
+                _ = sigint.recv() => {
+                    tracing::log::info!("sigint received");
+                    break;
+                }
+            }
+        }
+    }
+    #[cfg(not(unix))]
+    {
+        tokio::signal::ctrl_c().await?;
+        tracing::log::info!("CTRL+C received");
+    }
+    Ok(())
+}

crates/fuel-core/src/service.rs

@@ -9,10 +9,7 @@ use fuel_core_services::{
     State,
     StateWatcher,
 };
-use std::{
-    net::SocketAddr,
-    panic,
-};
+use std::net::SocketAddr;
 use tracing::log::warn;
 
 pub use config::{
@@ -221,7 +218,6 @@ impl RunnableTask for Task {
         for service in &self.services {
             stop_signals.push(service.await_stop())
         }
-        stop_signals.push(Box::pin(shutdown_signal()));
         stop_signals.push(Box::pin(watcher.while_started()));
 
         let (result, _, _) = futures::future::select_all(stop_signals).await;
@@ -247,35 +243,6 @@ impl RunnableTask for Task {
     }
 }
 
-async fn shutdown_signal() -> anyhow::Result<State> {
-    #[cfg(unix)]
-    {
-        let mut sigterm =
-            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?;
-
-        let mut sigint =
-            tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?;
-        loop {
-            tokio::select! {
-                _ = sigterm.recv() => {
-                    tracing::info!("sigterm received");
-                    break;
-                }
-                _ = sigint.recv() => {
-                    tracing::log::info!("sigint received");
-                    break;
-                }
-            }
-        }
-    }
-    #[cfg(not(unix))]
-    {
-        tokio::signal::ctrl_c().await?;
-        tracing::log::info!("CTRL+C received");
-    }
-    Ok(State::Stopped)
-}
-
 #[cfg(test)]
 mod tests {
     use crate::service::{

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

@@ -126,15 +126,17 @@ pub fn init_sub_services(
     let poa_adapter = PoAAdapter::new(poa.shared.clone());
 
     #[cfg(feature = "p2p")]
-    let sync = {
-        fuel_core_sync::service::new_service(
-            last_height,
-            p2p_adapter,
-            importer_adapter.clone(),
-            verifier,
-            config.sync,
-        )?
-    };
+    let sync = (!production_enabled)
+        .then(|| {
+            fuel_core_sync::service::new_service(
+                last_height,
+                p2p_adapter,
+                importer_adapter.clone(),
+                verifier,
+                config.sync,
+            )
+        })
+        .transpose()?;
 
     // TODO: Figure out on how to move it into `fuel-core-graphql-api`.
     let schema = dap::init(
@@ -194,7 +196,7 @@ pub fn init_sub_services(
     #[cfg(feature = "p2p")]
     {
         services.push(Box::new(network));
-        if !production_enabled {
+        if let Some(sync) = sync {
             services.push(Box::new(sync));
         }
     }

crates/services/src/service.rs

@@ -218,7 +218,7 @@ where
     }
 }
 
-#[tracing::instrument(skip_all)]
+#[tracing::instrument(skip_all, fields(service = S::NAME))]
 /// Initialize the background loop as a spawned task.
 fn initialize_loop<S>(service: S) -> Shared<watch::Sender<State>>
 where
@@ -230,7 +230,9 @@ where
     // Spawned as a task to check if the service is already running and to capture any panics.
     tokio::task::spawn(
         async move {
+            tracing::debug!("running");
             let join_handler = run(service, stop_sender.clone());
+            tracing::debug!("awaiting run");
             let result = join_handler.await;
 
             let stopped_state = if let Err(e) = result {
@@ -239,11 +241,15 @@ where
                 State::Stopped
             };
 
+            tracing::debug!("shutting down {:?}", stopped_state);
+
             let _ = stop_sender.send_if_modified(|state| {
                 if !state.stopped() {
                     *state = stopped_state;
+                    tracing::debug!("Wasn't stopped, so sent stop.");
                     true
                 } else {
+                    tracing::debug!("Was already stopped.");
                     false
                 }
             });
@@ -294,8 +300,10 @@ where
                 match task.run(&mut state).await {
                     Ok(should_continue) => {
                         if !should_continue {
+                            tracing::debug!("stopping");
                             return
                         }
+                        tracing::debug!("run loop");
                     }
                     Err(e) => {
                         let e: &dyn std::error::Error = &*e;

crates/services/src/state.rs

@@ -74,13 +74,15 @@ impl StateWatcher {
 }
 
 impl StateWatcher {
+    #[tracing::instrument(level = "debug", skip(self), err, ret)]
     /// Infinity loop while the state is `State::Started`. Returns the next received state.
     pub async fn while_started(&mut self) -> anyhow::Result<State> {
         loop {
             let state = self.borrow().clone();
             if !state.started() {
                 return Ok(state)
             }
+            tracing::debug!("Service is started, waiting for the next state...");
 
             self.changed().await?;
         }

tests/tests/messages.rs

@@ -70,7 +70,7 @@ async fn can_submit_genesis_message() {
     assert!(
         matches!(status, TransactionStatus::Success { .. }),
         "expected success, received {status:?}",
-    )
+    );
 }
 
 #[tokio::test]