multi: refactor tokio runtime handling, listen for multiple signals

Cargo.toml

@@ -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"

app/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/main.rs

@@ -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/rpc_server.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());
 

lib/authorization.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)
         },
     );