Merge pull request #67 from interlay/greg/fix/throw-svc-error

fix: throw error on service start if not shutdown

faucet/src/error.rs

@@ -1,6 +1,6 @@
 use chrono::ParseError;
 use jsonrpc_http_server::jsonrpc_core::Error as JsonRpcError;
-use kv::Error as KVError;
+use kv::Error as KvError;
 use parity_scale_codec::Error as CodecError;
 use runtime::Error as RuntimeError;
 use std::net::AddrParseError;
@@ -16,8 +16,8 @@ pub enum Error {
     JsonRpcError(#[from] JsonRpcError),
     #[error("AddrParseError: {0}")]
     AddrParseError(#[from] AddrParseError),
-    #[error("KV store error: {0}")]
-    KVError(#[from] KVError),
+    #[error("Kv store error: {0}")]
+    KvError(#[from] KvError),
     #[error("Error parsing datetime string: {0}")]
     DatetimeParsingError(#[from] ParseError),
     #[error("Too many faucet requests")]

faucet/src/system.rs

@@ -21,16 +21,21 @@ pub struct FaucetService {
 
 #[async_trait]
 impl Service<FaucetServiceConfig, PolkaBtcProvider> for FaucetService {
-    async fn start(
+    fn new_service(
         btc_parachain: PolkaBtcProvider,
         config: FaucetServiceConfig,
         handle: tokio::runtime::Handle,
         shutdown: ShutdownReceiver,
-    ) -> Result<(), RuntimeError> {
+    ) -> Self {
         FaucetService::new(btc_parachain, config, handle, shutdown)
-            .run_service()
-            .await
-            .map_err(|_| RuntimeError::ChannelClosed)
+    }
+
+    async fn start(&self) -> Result<(), RuntimeError> {
+        match self.run_service().await {
+            Ok(_) => Ok(()),
+            Err(Error::RuntimeError(err)) => Err(err),
+            Err(err) => Err(RuntimeError::Other(err.to_string())),
+        }
     }
 }
 

runtime/src/conn.rs

@@ -17,15 +17,16 @@ pub type ShutdownReceiver = tokio::sync::watch::Receiver<Option<()>>;
 
 #[async_trait]
 pub trait Provider {
-    async fn connect<T>(rpc_client: T, signer: PolkaBtcSigner) -> Result<Self, Error>
+    async fn new_provider<T>(rpc_client: T, signer: PolkaBtcSigner) -> Result<Self, Error>
     where
         Self: Sized,
         T: Into<RpcClient> + Send;
 }
 
 #[async_trait]
 pub trait Service<C, P: Provider> {
-    async fn start(provider: P, config: C, handle: Handle, shutdown: ShutdownReceiver) -> Result<(), Error>;
+    fn new_service(provider: P, config: C, handle: Handle, shutdown: ShutdownReceiver) -> Self;
+    async fn start(&self) -> Result<(), Error>;
 }
 
 pub(crate) async fn new_websocket_client(
@@ -148,26 +149,28 @@ impl<C: Clone + Send + 'static, P: Provider + Send, S: Service<C, P>> Manager<C,
             let (shutdown_sender, shutdown_receiver) = tokio::sync::watch::channel(None);
             let wait_for_shutdown = is_connected(ws_client.clone());
 
+            let provider = P::new_provider(ws_client, signer).await?;
+            let service = S::new_service(provider, config, handle, shutdown_receiver);
+
             // run service and shutdown listener to terminate child
             // processes if the websocket client disconnects
-            let _ = tokio::join! {
+            match tokio::try_join! {
                 async move {
                     // TODO: propogate shutdown signal from children
                     wait_for_shutdown.await;
                     // signal shutdown to child processes
                     let _ = shutdown_sender.broadcast(Some(()));
-                },
-                async move {
-                    let provider = P::connect(ws_client, signer).await?;
-                    let _ = S::start(provider, config, handle, shutdown_receiver).await;
                     Ok::<(), Error>(())
-                }
-            };
+                },
+                service.start()
+            } {
+                Ok(_) => (),
+                Err(err) => return Err(err),
+            }
 
             info!("Disconnected");
-
             match self.manager_config.restart_policy {
-                RestartPolicy::Never => return Err(Error::ChannelClosed),
+                RestartPolicy::Never => return Err(Error::ClientShutdown),
                 RestartPolicy::Always => continue,
             };
         }

runtime/src/error.rs

@@ -30,9 +30,9 @@ pub enum Error {
     VaultCommittedTheft,
     #[error("Channel closed unexpectedly")]
     ChannelClosed,
+    #[error("Client has shutdown unexpectedly")]
+    ClientShutdown,
 
-    #[error("Callback error: {0}")]
-    CallbackError(Box<dyn std::error::Error + Send + Sync>),
     #[error("Failed to load credentials from file: {0}")]
     KeyLoadingFailure(#[from] KeyLoadingError),
     #[error("Error serializing: {0}")]
@@ -54,6 +54,10 @@ pub enum Error {
     TimeElapsed(#[from] Elapsed),
     #[error("UrlParseError: {0}")]
     UrlParseError(#[from] UrlParseError),
+
+    /// Other error
+    #[error("Other: {0}")]
+    Other(String),
 }
 
 #[derive(Error, Debug)]

runtime/src/rpc.rs

@@ -201,7 +201,7 @@ impl PolkaBtcProvider {
 
 #[async_trait]
 impl Provider for PolkaBtcProvider {
-    async fn connect<T>(rpc_client: T, signer: PolkaBtcSigner) -> Result<Self, Error>
+    async fn new_provider<T>(rpc_client: T, signer: PolkaBtcSigner) -> Result<Self, Error>
     where
         Self: Sized,
         T: Into<RpcClient> + Send,

staked-relayer/src/error.rs

@@ -4,19 +4,21 @@ use bitcoin::{BitcoinError as BitcoinCoreError, Error as BitcoinError};
 use jsonrpc_core_client::RpcError;
 use parity_scale_codec::Error as CodecError;
 use runtime::{substrate_subxt::Error as XtError, Error as RuntimeError};
-use std::{net::AddrParseError, time::Duration};
+use std::time::Duration;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
 pub enum Error {
-    #[error("Internal error")]
-    InternalError,
     #[error("Could not verify that the oracle is offline")]
     CheckOracleOffline,
     #[error("Suggested status update does not contain block hash")]
     EventNoBlockHash,
     #[error("Error fetching transaction")]
     TransactionFetchingError,
+    #[error("Mathematical operation caused an overflow")]
+    ArithmeticOverflow,
+    #[error("Mathematical operation caused an underflow")]
+    ArithmeticUnderflow,
 
     #[error("RuntimeError: {0}")]
     RuntimeError(#[from] RuntimeError),
@@ -26,8 +28,6 @@ pub enum Error {
     SubXtError(#[from] XtError),
     #[error("CoreError: {0}")]
     CoreError(#[from] CoreError<RelayError>),
-    #[error("AddrParseError: {0}")]
-    AddrParseError(#[from] AddrParseError),
     #[error("CodecError: {0}")]
     CodecError(#[from] CodecError),
     #[error("BitcoinError: {0}")]
@@ -36,8 +36,6 @@ pub enum Error {
     BitcoinCoreError(#[from] BitcoinCoreError),
     #[error("RPC error: {0}")]
     RpcError(#[from] RpcError),
-    #[error("Mathematical operation error")]
-    MathError,
 }
 
 /// Gets the default retrying policy

staked-relayer/src/faucet.rs

@@ -50,9 +50,9 @@ pub async fn fund_and_register(provider: &PolkaBtcProvider, faucet_url: &String)
     let user_allowance_in_dot: u128 = get_faucet_allowance(connection.clone(), "user_allowance").await?;
     let registration_stake = user_allowance_in_dot
         .checked_mul(PLANCK_PER_DOT)
-        .ok_or(Error::MathError)?
+        .ok_or(Error::ArithmeticOverflow)?
         .checked_sub(TX_FEES)
-        .ok_or(Error::MathError)?;
+        .ok_or(Error::ArithmeticUnderflow)?;
     provider.register_staked_relayer(registration_stake).await?;
 
     // Receive staked relayer allowance from faucet

staked-relayer/src/system.rs

@@ -34,16 +34,21 @@ pub struct RelayerService {
 
 #[async_trait]
 impl Service<RelayerServiceConfig, PolkaBtcProvider> for RelayerService {
-    async fn start(
+    fn new_service(
         btc_parachain: PolkaBtcProvider,
         config: RelayerServiceConfig,
         handle: tokio::runtime::Handle,
         shutdown: ShutdownReceiver,
-    ) -> Result<(), RuntimeError> {
+    ) -> Self {
         RelayerService::new(btc_parachain, config, handle, shutdown)
-            .run_service()
-            .await
-            .map_err(|_| RuntimeError::ChannelClosed)
+    }
+
+    async fn start(&self) -> Result<(), RuntimeError> {
+        match self.run_service().await {
+            Ok(_) => Ok(()),
+            Err(Error::RuntimeError(err)) => Err(err),
+            Err(err) => Err(RuntimeError::Other(err.to_string())),
+        }
     }
 }
 

vault/src/error.rs

@@ -3,13 +3,10 @@ use hex::FromHexError;
 use jsonrpc_core_client::RpcError;
 use parity_scale_codec::Error as CodecError;
 use runtime::{substrate_subxt::Error as XtError, Error as RuntimeError};
-use std::net::AddrParseError;
 use thiserror::Error;
 
 #[derive(Error, Debug)]
 pub enum Error {
-    #[error("Internal error")]
-    InternalError,
     #[error("Insufficient funds available")]
     InsufficientFunds,
     #[error("Value below dust amount")]
@@ -30,10 +27,6 @@ pub enum Error {
     ArithmeticOverflow,
     #[error("Mathematical operation caused an underflow")]
     ArithmeticUnderflow,
-    #[error("Mathematical operation error")]
-    MathError,
-    #[error("Vault has uncompleted redeem requests")]
-    UncompletedRedeemRequests,
 
     #[error("RPC error: {0}")]
     RpcError(#[from] RpcError),
@@ -47,6 +40,4 @@ pub enum Error {
     SubXtError(#[from] XtError),
     #[error("CodecError: {0}")]
     CodecError(#[from] CodecError),
-    #[error("AddrParseError: {0}")]
-    AddrParseError(#[from] AddrParseError),
 }

vault/src/system.rs

@@ -37,16 +37,21 @@ pub struct VaultService {
 
 #[async_trait]
 impl Service<VaultServiceConfig, PolkaBtcProvider> for VaultService {
-    async fn start(
+    fn new_service(
         btc_parachain: PolkaBtcProvider,
         config: VaultServiceConfig,
         handle: tokio::runtime::Handle,
         shutdown: ShutdownReceiver,
-    ) -> Result<(), RuntimeError> {
+    ) -> Self {
         VaultService::new(btc_parachain, config, handle, shutdown)
-            .run_service()
-            .await
-            .map_err(|_| RuntimeError::ChannelClosed)
+    }
+
+    async fn start(&self) -> Result<(), RuntimeError> {
+        match self.run_service().await {
+            Ok(_) => Ok(()),
+            Err(Error::RuntimeError(err)) => Err(err),
+            Err(err) => Err(RuntimeError::Other(err.to_string())),
+        }
     }
 }