feat: add graceful shutdown

Cargo.toml

@@ -21,6 +21,7 @@ prometheus = "0.13.3"
 serde = { version = "1.0", features = ["derive"] }
 ethers = { version = "2.0", features = ["ws"] }
 tokio = { version = "1.28.0", features = ["rt-multi-thread", "rt", "macros"] }
+tokio-util = { version = "0.7.8" }
 thiserror = "1.0.40"
 tracing = "0.1.37"
 tracing-subscriber = "0.3.17"

compose.yaml

@@ -14,8 +14,8 @@ services:
       - URL=http://0.0.0.0:8545/
 
   fuel_node:
-    build: 
-      context: fuel_node  
+    build:
+      context: fuel_node
       args:
         fuel_core_version: "v${FUEL_CORE_VERSION:-0.18.3}"
     container_name: fuel-node

run_tests.sh

@@ -16,10 +16,10 @@ done
 cargo test --bin fuel-block-committer
 
 docker compose up -d
-trap 'docker compose down' EXIT
+trap 'docker compose down > /dev/null 2>&1' EXIT
 
 cargo test --test e2e -- --nocapture
 
 if [[ $show_logs = "true" ]]; then
-    docker compose logs block_committer
+    docker compose logs -f block_committer &
 fi

src/adapters/ethereum_adapter.rs

@@ -11,18 +11,31 @@ pub use websocket::EthereumWs;
 
 use crate::{adapters::fuel_adapter::FuelBlock, errors::Result};
 
-#[derive(Debug, Clone, Copy)]
-pub struct FuelBlockCommitedOnEth {
+#[derive(Clone, Copy)]
+pub struct FuelBlockCommittedOnEth {
     pub fuel_block_hash: [u8; 32],
     pub commit_height: U256,
 }
 
+impl std::fmt::Debug for FuelBlockCommittedOnEth {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let hash = self
+            .fuel_block_hash
+            .map(|byte| format!("{byte:02x?}"))
+            .join("");
+        f.debug_struct("FuelBlockCommittedOnEth")
+            .field("hash", &hash)
+            .field("commit_height", &self.commit_height)
+            .finish()
+    }
+}
+
 #[cfg_attr(test, mockall::automock)]
 #[async_trait]
 pub trait EventStreamer {
     async fn establish_stream<'a>(
         &'a self,
-    ) -> Result<Pin<Box<dyn Stream<Item = Result<FuelBlockCommitedOnEth>> + 'a + Send>>>;
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<FuelBlockCommittedOnEth>> + 'a + Send>>>;
 }
 
 #[cfg_attr(test, mockall::automock)]

src/adapters/ethereum_adapter/websocket/event_streamer.rs

@@ -9,7 +9,7 @@ use futures::{stream::TryStreamExt, Stream};
 
 use crate::{
     adapters::ethereum_adapter::{
-        websocket::adapter::CommitSubmittedFilter, EventStreamer, FuelBlockCommitedOnEth,
+        websocket::adapter::CommitSubmittedFilter, EventStreamer, FuelBlockCommittedOnEth,
     },
     errors::{Error, Result},
 };
@@ -28,7 +28,7 @@ pub struct EthEventStreamer {
 impl EventStreamer for EthEventStreamer {
     async fn establish_stream(
         &self,
-    ) -> Result<Pin<Box<dyn Stream<Item = Result<FuelBlockCommitedOnEth>> + '_ + Send>>> {
+    ) -> Result<Pin<Box<dyn Stream<Item = Result<FuelBlockCommittedOnEth>> + '_ + Send>>> {
         Ok(Box::pin(
             self.events
                 .subscribe()
@@ -37,7 +37,7 @@ impl EventStreamer for EthEventStreamer {
                 .map_ok(|event| {
                     let fuel_block_hash = event.block_hash;
                     let commit_height = event.commit_height;
-                    FuelBlockCommitedOnEth {
+                    FuelBlockCommittedOnEth {
                         fuel_block_hash,
                         commit_height,
                     }

src/adapters/fuel_adapter.rs

@@ -6,12 +6,22 @@ use serde::{Deserialize, Serialize};
 
 use crate::errors::Result;
 
-#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq)]
+#[derive(Serialize, Deserialize, Clone, Copy, PartialEq, Eq)]
 pub struct FuelBlock {
     pub hash: [u8; 32],
     pub height: u32,
 }
 
+impl std::fmt::Debug for FuelBlock {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        let hash = self.hash.map(|byte| format!("{byte:02x?}")).join("");
+        f.debug_struct("FuelBlock")
+            .field("hash", &hash)
+            .field("height", &self.height)
+            .finish()
+    }
+}
+
 #[cfg_attr(test, mockall::automock)]
 #[async_trait::async_trait]
 pub trait FuelAdapter: Send + Sync {

src/cli.rs

@@ -8,9 +8,9 @@ use crate::{
     errors::{Error, Result},
     setup::config::Config,
 };
-const ETHEREUM_RPC: &str = "http://127.0.0.1:8545/";
-const FUEL_GRAPHQL_ENDPOINT: &str = "https://127.0.0.1:4000";
-const STATE_CONTRACT_ADDRESS: Address = Address::zero();
+
+const ETHEREUM_RPC: &str = "ws://127.0.0.1:8545/";
+const FUEL_GRAPHQL_ENDPOINT: &str = "http://127.0.0.1:4000";
 const COMMIT_INTERVAL: u32 = 1;
 const HOST: &str = "127.0.0.1";
 const PORT: u16 = 8080;
@@ -34,27 +34,37 @@ struct Cli {
     ethereum_wallet_key: String,
 
     /// Ethereum RPC
-    #[arg(long,
-    env = "ETHEREUM_RPC",
-    default_value = ETHEREUM_RPC, value_name = "URL", help = "URL to a Ethereum RPC endpoint.")]
+    #[arg(
+        long,
+        env = "ETHEREUM_RPC",
+        default_value = ETHEREUM_RPC, value_name = "URL", help = "URL to a Ethereum RPC endpoint."
+    )]
     ethereum_rpc: Url,
 
     /// Fuel GraphQL endpoint
-    #[arg(long,
-    env = "FUEL_GRAPHQL_ENDPOINT",
-    default_value = FUEL_GRAPHQL_ENDPOINT, value_name = "URL", help = "URL to a fuel-core graphql endpoint.")]
+    #[arg(
+        long,
+        env = "FUEL_GRAPHQL_ENDPOINT",
+        default_value = FUEL_GRAPHQL_ENDPOINT, value_name = "URL", help = "URL to a fuel-core graphql endpoint."
+    )]
     fuel_graphql_endpoint: Url,
 
     /// State contract address
-    #[arg(long,
-    env = "STATE_CONTRACT_ADDRESS",
-    default_value_t = STATE_CONTRACT_ADDRESS, value_name = "BYTES20", help = "Ethereum address of the fuel chain state contract.")]
+    #[arg(
+        long,
+        env = "STATE_CONTRACT_ADDRESS",
+        value_name = "Address",
+        help = "Ethereum address of the fuel chain state contract."
+    )]
     state_contract_address: Address,
 
     /// Commit interval
-    #[arg(long,
-    env = "COMMIT_INTERVAL",
-    default_value_t = COMMIT_INTERVAL, value_name = "U32", help = "The number of fuel blocks between ethereum commits. If set to 1, then every block should be pushed to Ethereum.")]
+    #[arg(
+        long,
+        env = "COMMIT_INTERVAL",
+        default_value_t = COMMIT_INTERVAL, value_name = "U32", help = "The number of fuel \\
+        blocks between ethereum commits. If set to 1, then every block should be pushed to Ethereum."
+    )]
     commit_interval: u32,
 
     /// Host of the started API server
@@ -68,15 +78,19 @@ struct Cli {
     host: Ipv4Addr,
 
     /// Port of the started API server
-    #[arg(long,
-    env = "PORT",
-    default_value_t = PORT, value_name = "U16", help = "Port on which to start the API server.")]
+    #[arg(
+        long,
+        env = "PORT",
+        default_value_t = PORT, value_name = "U16", help = "Port on which to start the API server."
+    )]
     port: u16,
 
     /// Ethereum chain id
-    #[arg(long,
+    #[arg(
+        long,
         env = "ETHEREUM_CHAIN",
-        default_value_t = CHAIN_NAME, value_name = "ChainName", help = "Chain id of the ethereum network.")]
+        default_value_t = CHAIN_NAME, value_name = "ChainName", help = "Chain id of the ethereum network."
+    )]
     ethereum_chain: Chain,
 
     #[arg(

src/errors.rs

@@ -1,5 +1,6 @@
 use actix_web::ResponseError;
 use ethers::signers::WalletError;
+use tokio::task::JoinError;
 use url::ParseError;
 
 #[derive(thiserror::Error, Debug)]
@@ -36,6 +37,18 @@ impl From<ParseError> for Error {
     }
 }
 
+impl From<std::io::Error> for Error {
+    fn from(error: std::io::Error) -> Self {
+        Self::Other(error.to_string())
+    }
+}
+
+impl From<JoinError> for Error {
+    fn from(error: JoinError) -> Self {
+        Self::Other(error.to_string())
+    }
+}
+
 impl ResponseError for Error {}
 
 pub type Result<T> = std::result::Result<T, Error>;

src/main.rs

@@ -12,45 +12,50 @@ use prometheus::Registry;
 use setup::{
     config::InternalConfig,
     helpers::{
-        create_eth_adapter, setup_logger, setup_storage, spawn_block_watcher,
+        create_eth_adapter, setup_logger, setup_storage, shut_down, spawn_block_watcher,
         spawn_eth_committer_and_listener, spawn_wallet_balance_tracker,
     },
 };
+use tokio_util::sync::CancellationToken;
 
 #[tokio::main]
 async fn main() -> Result<()> {
     setup_logger();
 
     let config = cli::parse()?;
     let internal_config = InternalConfig::default();
+    let cancel_token = CancellationToken::new();
 
     let storage = setup_storage(&config).await?;
 
     let metrics_registry = Registry::default();
 
-    let (rx_fuel_block, _block_watcher_handle, fuel_health_check) = spawn_block_watcher(
+    let (rx_fuel_block, block_watcher_handle, fuel_health_check) = spawn_block_watcher(
         &config,
         &internal_config,
         storage.clone(),
         &metrics_registry,
+        cancel_token.clone(),
     );
 
     let (ethereum_rpc, eth_health_check) =
         create_eth_adapter(&config, &internal_config, &metrics_registry).await?;
 
-    let _wallet_balance_tracker_handle = spawn_wallet_balance_tracker(
+    let wallet_balance_tracker_handle = spawn_wallet_balance_tracker(
         &config,
         &internal_config,
         &metrics_registry,
         ethereum_rpc.clone(),
+        cancel_token.clone(),
     )?;
 
-    let (_committer_handle, _listener_handle) = spawn_eth_committer_and_listener(
+    let (committer_handle, listener_handle) = spawn_eth_committer_and_listener(
         &internal_config,
         rx_fuel_block,
         ethereum_rpc,
         storage.clone(),
         &metrics_registry,
+        cancel_token.clone(),
     )?;
 
     launch_api_server(
@@ -62,5 +67,12 @@ async fn main() -> Result<()> {
     )
     .await?;
 
-    Ok(())
+    shut_down(
+        cancel_token,
+        block_watcher_handle,
+        wallet_balance_tracker_handle,
+        committer_handle,
+        listener_handle,
+    )
+    .await
 }

src/services/block_committer.rs

@@ -60,10 +60,12 @@ impl Runner for BlockCommitter {
             if let Err(error) = self.submit_block(fuel_block).await {
                 error!("{error}");
             } else {
-                info!("Submitted fuel block! ({fuel_block:?})",);
+                info!("submitted {fuel_block:?}!");
             }
         }
 
+        info!("Block Committer stopped");
+
         Ok(())
     }
 }