[PLA-2087] Catch all handlers (#68)

.env

@@ -1,4 +1,4 @@
 PLATFORM_KEY=example
 KEY_PASS=example
 CONFIG_FILE=/opt/app/config.json
-RUST_LOG=wallet_lib=warn
+RUST_LOG=

.github/workflows/security-audit.yml

@@ -7,9 +7,7 @@ on:
 jobs:
   audit:
     runs-on: ubuntu-latest
-    permissions:
-      id-token: write
-      contents: read
+    permissions: write-all
 
     steps:
       - uses: actions/checkout@v4

Cargo.toml

@@ -1,8 +1,13 @@
 [package]
 name = "wallet-daemon"
-version = "2.0.2"
+version = "2.0.3"
 edition = "2021"
 
+[profile.release]
+debug = true
+debug-assertions = true
+overflow-checks = true
+
 [dependencies]
 tracing = "0.1.40"
 rpassword = "7.3.1"
@@ -31,3 +36,4 @@ sp-core = { version = "34.0.0", default-features = false, features = [
 bip39 = { version = "2.0.0", features = ["rand"] }
 autoincrement = "1.0.1"
 lru = "0.12.5"
+anyhow = "1.0.93"

src/main.rs

@@ -5,7 +5,6 @@ use std::sync::Arc;
 use std::time::Duration;
 use subxt::backend::rpc::reconnecting_rpc_client::{ExponentialBackoff, RpcClient};
 use subxt::{OnlineClient, PolkadotConfig};
-use tokio::signal;
 use wallet_daemon::config_loader::{load_config, load_wallet};
 use wallet_daemon::{
     set_multitenant, write_seed, DeriveWalletJob, SubscriptionJob, TransactionJob,
@@ -63,17 +62,19 @@ async fn main() -> Result<(), Box<dyn std::error::Error>> {
         platform_token.clone(),
     );
 
-    subscription_job.start();
-    transaction_poller.start();
-    transaction_processor.start();
-
     let (wallet_poller, wallet_processor) =
         DeriveWalletJob::create_job(keypair, platform_url, platform_token);
 
-    wallet_poller.start();
-    wallet_processor.start();
-
-    signal::ctrl_c().await.expect("Failed to listen for ctrl c");
+    tokio::select! {
+        _ = transaction_poller.start() => {}
+        _ =  transaction_processor.start() => {}
+        _ = wallet_poller.start() => {}
+        _ = wallet_processor.start() => {}
+        r = subscription_job.start() => {
+            let err = r.unwrap_err();
+            tracing::error!("Subscription job failed: {:?}", err);
+        }
+    }
 
     Ok(())
 }

src/subscription.rs

@@ -1,9 +1,10 @@
 use std::sync::{Arc, Mutex};
-use std::{panic, process};
+use std::{panic};
 use subxt::client::ClientRuntimeUpdater;
 use subxt::ext::subxt_core;
 use subxt::{OnlineClient, PolkadotConfig};
 use subxt_core::config::substrate;
+use tokio::task::JoinHandle;
 
 #[derive(Debug)]
 pub struct SubscriptionJob {
@@ -27,32 +28,17 @@ impl SubscriptionJob {
         SubscriptionJob::new(subscription)
     }
 
-    pub fn start(&self) {
-        let orig_hook = panic::take_hook();
-        panic::set_hook(Box::new(move |panic_info| {
-            orig_hook(panic_info);
-            process::exit(1);
-        }));
-
-        self.start_block();
-        self.start_runtime();
-    }
-
-    pub fn start_block(&self) {
+    pub fn start(&self) -> JoinHandle<()> {
         let block_sub = Arc::clone(&self.params);
-
-        tokio::spawn(async move {
-            block_sub.block_subscription().await;
-        });
-    }
-
-    pub fn start_runtime(&self) {
         let runtime_sub = Arc::clone(&self.params);
         let updater = runtime_sub.rpc.updater();
 
         tokio::spawn(async move {
-            runtime_sub.runtime_subscription(updater).await;
-        });
+            tokio::select! {
+                _ = block_sub.block_subscription() => {}
+                _ = runtime_sub.runtime_subscription(updater) => {}
+            }
+        })
     }
 
     pub fn get_params(&self) -> Arc<SubscriptionParams> {
@@ -99,6 +85,8 @@ impl SubscriptionParams {
                 },
             };
         }
+
+        tracing::error!("Runtime update stream ended unexpectedly");
     }
 
     async fn block_subscription(self: Arc<Self>) {
@@ -126,6 +114,8 @@ impl SubscriptionParams {
 
             *block_header = Some(block.header().clone());
         }
+
+        tracing::error!("Block subscription stream ended unexpectedly");
     }
 
     pub fn get_block_header(&self) -> substrate::SubstrateHeader<u32, substrate::BlakeTwo256> {

src/transaction.rs

@@ -107,10 +107,10 @@ impl TransactionJob {
         )
     }
 
-    pub fn start(self) {
+    pub fn start(self) -> JoinHandle<()> {
         tokio::spawn(async move {
             self.start_polling().await;
-        });
+        })
     }
 
     async fn start_polling(&self) {

src/wallet.rs

@@ -78,10 +78,10 @@ impl DeriveWalletJob {
         )
     }
 
-    pub fn start(self) {
+    pub fn start(self) -> JoinHandle<()> {
         tokio::spawn(async move {
             self.start_polling().await;
-        });
+        })
     }
 
     async fn start_polling(&self) {