multi: refactor tokio runtime handling, listen for multiple signals (#21)

* multi: refactor tokio runtime handling, listen for multiple signals

The motivation for this is being able to pick up multiple different error sources that need to be propagated out from the main function. Prior to this commit, for example being unable to start the RPC server would not end in a fatal error.

* lib: fix clippy issues

* main: move egui function

Author: torkelrogstad

Cargo.lock

@@ -31,7 +31,7 @@ serde = "1.0.179"
 serde_json = "1.0.113"
 thiserror = "2.0.11"
 tiny-bip39 = "2.0.0"
-tokio = { version = "1.29.1", default-features = false }
+tokio = { version = "1.29.1", default-features = false, features = ["signal"] }
 tokio-util = "0.7.10"
 tonic = "0.12.3"
 tracing = "0.1.40"

Cargo.toml

@@ -11,7 +11,6 @@ anyhow = { workspace = true }
 bincode = { workspace = true }
 bitcoin = { workspace = true, features = ["serde"] }
 clap = { workspace = true, features = ["derive"] }
-ctrlc = "3.4.0"
 dirs = "5.0.1"
 eframe = "0.30.0"
 futures = { workspace = true }

app/Cargo.toml

@@ -1,9 +1,10 @@
 #![feature(let_chains)]
 
-use std::{path::Path, sync::mpsc};
+use std::path::Path;
 
 use clap::Parser as _;
 
+use tokio::{signal::ctrl_c, sync::oneshot};
 use tracing_subscriber::{
     filter as tracing_filter, fmt::format, layer::SubscriberExt, Layer,
 };
@@ -140,6 +141,26 @@ fn set_tracing_subscriber(
     Ok((line_buffer, rolling_log_guard))
 }
 
+fn run_egui_app(
+    config: &crate::cli::Config,
+    line_buffer: LineBuffer,
+    app: Result<crate::app::App, crate::app::Error>,
+) -> Result<(), eframe::Error> {
+    let native_options = eframe::NativeOptions::default();
+    eframe::run_native(
+        "Thunder",
+        native_options,
+        Box::new(move |cc| {
+            Ok(Box::new(gui::EguiApp::new(
+                app.ok(),
+                cc,
+                line_buffer,
+                config.rpc_addr,
+            )))
+        }),
+    )
+}
+
 fn main() -> anyhow::Result<()> {
     let cli = cli::Cli::parse();
     let config = cli.get_config()?;
@@ -148,55 +169,45 @@ fn main() -> anyhow::Result<()> {
         config.log_level,
         config.log_level_file,
     )?;
-    let app: Result<app::App, app::Error> = app::App::new(&config)
-        .inspect(|app| {
-            // spawn rpc server
-            app.runtime.spawn({
-                let app = app.clone();
-                async move {
-                    rpc_server::run_server(app, config.rpc_addr).await.unwrap()
+
+    let (app_tx, app_rx) = oneshot::channel::<anyhow::Error>();
+
+    let app = app::App::new(&config).inspect(|app| {
+        // spawn rpc server
+        app.runtime.spawn({
+            let app = app.clone();
+            async move {
+                tracing::info!("starting RPC server at `{}`", config.rpc_addr);
+                if let Err(err) =
+                    rpc_server::run_server(app, config.rpc_addr).await
+                {
+                    app_tx.send(err).expect("failed to send error to app");
                 }
-            });
-        })
-        .inspect_err(|err| {
-            tracing::error!("application error: {:?}", err);
+            }
         });
+    });
 
-    if config.headless {
-        tracing::info!("Running in headless mode");
-        drop(line_buffer);
-        let _app = app?;
-        // wait for ctrlc signal
-        let (tx, rx) = mpsc::channel();
-        ctrlc::set_handler(move || {
-            tx.send(()).unwrap();
-        })
-        .expect("Error setting Ctrl-C handler");
-        rx.recv().unwrap();
-        tracing::info!("Received Ctrl-C signal, exiting...");
-    } else {
-        let native_options = eframe::NativeOptions::default();
-        let app: Option<_> = app.map_or_else(
-            |err| {
-                let err = anyhow::Error::from(err);
-                tracing::error!("{err:#}");
-                None
-            },
-            Some,
-        );
-        eframe::run_native(
-            "Thunder",
-            native_options,
-            Box::new(move |cc| {
-                Ok(Box::new(gui::EguiApp::new(
-                    app,
-                    cc,
-                    line_buffer,
-                    config.rpc_addr,
-                )))
-            }),
-        )
-        .map_err(|err| anyhow::anyhow!("failed to launch egui app: {err}"))?
+    if !config.headless {
+        // For GUI mode we want the GUI to start, even if the app fails to start.
+        return run_egui_app(&config, line_buffer, app)
+            .map_err(|e| anyhow::anyhow!("failed to run egui app: {e:#}"));
     }
-    Ok(())
+
+    tracing::info!("Running in headless mode");
+    drop(line_buffer);
+
+    // If we're headless, we want to exit hard if the app fails to start.
+    let app = app?;
+
+    app.runtime.block_on(async move {
+        tokio::select! {
+            Ok(_) = ctrl_c() => {
+                tracing::info!("Shutting down due to process interruption");
+                Ok(())
+            }
+            Ok(err) = app_rx => {
+                Err(anyhow::anyhow!("received error from RPC server: {err:#} ({err:?})"))
+            }
+        }
+    })
 }

app/main.rs

@@ -259,7 +259,6 @@ pub async fn run_server(
     let server = Server::builder().build(rpc_addr).await?;
 
     let addr = server.local_addr()?;
-    tracing::info!("RPC server listening on {}", addr);
 
     let handle = server.start(RpcServerImpl { app }.into_rpc());
 

app/rpc_server.rs

@@ -105,9 +105,11 @@ pub fn verify_authorized_transaction(
     transaction: &AuthorizedTransaction,
 ) -> Result<(), Error> {
     let tx_bytes_canonical = borsh::to_vec(&transaction.transaction)?;
-    let messages: Vec<_> = std::iter::repeat(tx_bytes_canonical.as_slice())
-        .take(transaction.authorizations.len())
-        .collect();
+    let messages: Vec<_> = std::iter::repeat_n(
+        tx_bytes_canonical.as_slice(),
+        transaction.authorizations.len(),
+    )
+    .collect();
     let (verifying_keys, signatures): (Vec<VerifyingKey>, Vec<Signature>) =
         transaction
             .authorizations
@@ -137,7 +139,7 @@ pub fn verify_authorizations(body: &Body) -> Result<(), Error> {
         serialized_transactions.iter().map(Vec::as_slice);
     let messages = input_numbers.zip(serialized_transactions).flat_map(
         |(input_number, serialized_transaction)| {
-            std::iter::repeat(serialized_transaction).take(input_number)
+            std::iter::repeat_n(serialized_transaction, input_number)
         },
     );