feat: make wallet db calls synchronous rather than async

applications/tari_app_utilities/src/initialization.rs

@@ -29,7 +29,7 @@ pub fn init_configuration(
     // Load and apply configuration file
     let cfg = bootstrap.load_configuration()?;
 
-    // Initialise the logger
+    // Initialise the logger (Comment out to enable tokio tracing via tokio-console)
     bootstrap.initialize_logging()?;
 
     log::info!(target: LOG_TARGET, "{} ({})", application_type, consts::APP_VERSION);

applications/tari_console_wallet/Cargo.toml

@@ -20,6 +20,12 @@ tari_shutdown = { path = "../../infrastructure/shutdown" }
 tari_key_manager = { path = "../../base_layer/key_manager" }
 tari_utilities = { git = "https://github.com/tari-project/tari_utilities.git", tag = "v0.3.1" }
 
+# Uncomment for tokio tracing via tokio-console (needs "tracing" featurs)
+#console-subscriber = "0.1.3"
+#tokio = { version = "1.14", features = ["signal", "tracing"] }
+# Uncomment for normal use (non tokio-console tracing)
+tokio = { version = "1.14", features = ["signal"] }
+
 sha2 = "0.9.5"
 digest = "0.9.0"
 chrono = { version = "0.4.19", default-features = false }
@@ -36,7 +42,6 @@ rpassword = "5.0"
 rustyline = "9.0"
 strum = "0.22"
 strum_macros = "0.22"
-tokio = { version = "1.11", features = ["signal"] }
 thiserror = "1.0.26"
 tonic = "0.6.2"
 tracing = "0.1.26"

applications/tari_console_wallet/src/automation/command_parser.rs

@@ -322,9 +322,9 @@ fn parse_make_it_rain(mut args: SplitWhitespace) -> Result<Vec<ParsedArgument>,
     // txs per second
     let txps = args.next().ok_or_else(|| ParseError::Empty("Txs/s".to_string()))?;
     let txps = txps.parse::<f64>().map_err(ParseError::Float)?;
-    if txps > 25.0 {
-        println!("Maximum transaction rate is 25/sec");
-        return Err(ParseError::Invalid("Maximum transaction rate is 25/sec".to_string()));
+    if txps > 250.0 {
+        println!("Maximum transaction rate is 250/sec");
+        return Err(ParseError::Invalid("Maximum transaction rate is 250/sec".to_string()));
     }
     parsed_args.push(ParsedArgument::Float(txps));
 

applications/tari_console_wallet/src/main.rs

@@ -75,6 +75,9 @@ pub mod wallet_modes;
 
 /// Application entry point
 fn main() {
+    // Uncomment to enable tokio tracing via tokio-console
+    // console_subscriber::init();
+
     match main_inner() {
         Ok(_) => process::exit(0),
         Err(err) => {

applications/tari_mining_node/src/stratum/error.rs

@@ -45,7 +45,7 @@ pub enum Error {
     #[error("Can't create TLS connector: {0}")]
     Tls(#[from] native_tls::Error),
     #[error("Can't establish TLS connection: {0}")]
-    Tcp(#[from] native_tls::HandshakeError<std::net::TcpStream>),
+    Tcp(#[from] Box<native_tls::HandshakeError<std::net::TcpStream>>),
     #[error("No connected stream")]
     NotConnected,
     #[error("Can't parse int: {0}")]
@@ -68,3 +68,9 @@ impl<T> From<std::sync::mpsc::SendError<T>> for Error {
         Error::General(format!("Failed to send to a channel: {:?}", error))
     }
 }
+
+impl From<native_tls::HandshakeError<std::net::TcpStream>> for Error {
+    fn from(error: native_tls::HandshakeError<std::net::TcpStream>) -> Self {
+        Error::General(format!("TLS handshake error: {:?}", error))
+    }
+}

applications/tari_mining_node/src/stratum/stream.rs

@@ -33,7 +33,7 @@ use crate::stratum::error::Error;
 
 pub(crate) enum Stream {
     Stream(BufStream<TcpStream>),
-    TlsStream(BufStream<TlsStream<TcpStream>>),
+    TlsStream(Box<BufStream<TlsStream<TcpStream>>>),
 }
 
 impl Stream {
@@ -46,7 +46,7 @@ impl Stream {
             let base_host = format!("{}.{}", split_url[split_url.len() - 2], split_url[split_url.len() - 1]);
             let mut stream = connector.connect(&base_host, conn)?;
             stream.get_mut().set_nonblocking(true)?;
-            Ok(Self::TlsStream(BufStream::new(stream)))
+            Ok(Self::TlsStream(Box::from(BufStream::new(stream))))
         } else {
             conn.set_nonblocking(true)?;
             Ok(Self::Stream(BufStream::new(conn)))

applications/tari_validator_node/src/main.rs

@@ -64,7 +64,9 @@ use crate::{
 const LOG_TARGET: &str = "tari::validator_node::app";
 
 fn main() {
+    // Uncomment to enable tokio tracing via tokio-console
     // console_subscriber::init();
+
     if let Err(err) = main_inner() {
         let exit_code = err.exit_code;
         eprintln!("{:?}", err);

base_layer/wallet/Cargo.toml

@@ -21,6 +21,12 @@ tari_storage = { version = "^0.30", path = "../../infrastructure/storage" }
 tari_common_sqlite = { path = "../../common_sqlite" }
 tari_utilities = { git = "https://github.com/tari-project/tari_utilities.git", tag = "v0.3.1" }
 
+# Uncomment for tokio tracing via tokio-console (needs "tracing" featurs)
+#console-subscriber = "0.1.3"
+#tokio = { version = "1.14", features = ["sync", "macros", "tracing"] }
+# Uncomment for normal use (non tokio-console tracing)
+tokio = { version = "1.14", features = ["sync", "macros"] }
+
 aes-gcm = "^0.8"
 async-trait = "0.1.50"
 argon2 = "0.2"
@@ -47,7 +53,6 @@ strum = "0.22"
 strum_macros = "0.22"
 tempfile = "3.1.0"
 thiserror = "1.0.26"
-tokio = { version = "1.11", features = ["sync", "macros"] }
 tower = "0.4"
 prost = "0.9"
 

base_layer/wallet/src/assets/asset_manager.rs

@@ -58,7 +58,6 @@ impl<T: OutputManagerBackend + 'static> AssetManager<T> {
         let outputs = self
             .output_database
             .fetch_with_features(OutputFlags::ASSET_REGISTRATION)
-            .await
             .map_err(|err| WalletError::OutputManagerError(err.into()))?;
 
         debug!(

base_layer/wallet/src/output_manager_service/error.rs

@@ -165,14 +165,14 @@ pub enum OutputManagerStorageError {
     DieselConnectionError(#[from] diesel::ConnectionError),
     #[error("Database migration error: `{0}`")]
     DatabaseMigrationError(String),
-    #[error("Blocking task spawn error: `{0}`")]
-    BlockingTaskSpawnError(String),
     #[error("Wallet db is already encrypted and cannot be encrypted until the previous encryption is removed")]
     AlreadyEncrypted,
     #[error("Byte array error: `{0}`")]
     ByteArrayError(#[from] ByteArrayError),
     #[error("Aead error: `{0}`")]
     AeadError(String),
+    #[error("Tried to insert a script that already exists in the database")]
+    DuplicateScript,
     #[error("Tari script error : {0}")]
     ScriptError(#[from] ScriptError),
     #[error("Binary not stored as valid hex:{0}")]

base_layer/wallet/src/output_manager_service/handle.rs

@@ -25,7 +25,7 @@ use std::{fmt, fmt::Formatter, sync::Arc};
 use aes_gcm::Aes256Gcm;
 use tari_common_types::{
     transaction::TxId,
-    types::{BlockHash, HashOutput, PrivateKey, PublicKey},
+    types::{HashOutput, PrivateKey, PublicKey},
 };
 use tari_core::{
     covenants::Covenant,
@@ -52,11 +52,8 @@ use tower::Service;
 
 use crate::output_manager_service::{
     error::OutputManagerError,
-    service::Balance,
-    storage::{
-        models::{KnownOneSidedPaymentScript, SpendingPriority},
-        OutputStatus,
-    },
+    service::{Balance, OutputStatusesByTxId},
+    storage::models::{KnownOneSidedPaymentScript, SpendingPriority},
 };
 
 /// API Request enum
@@ -247,21 +244,12 @@ pub enum OutputManagerResponse {
     RewoundOutputs(Vec<RecoveredOutput>),
     ScanOutputs(Vec<RecoveredOutput>),
     AddKnownOneSidedPaymentScript,
-    CreateOutputWithFeatures {
-        output: Box<UnblindedOutputBuilder>,
-    },
-    CreatePayToSelfWithOutputs {
-        transaction: Box<Transaction>,
-        tx_id: TxId,
-    },
+    CreateOutputWithFeatures { output: Box<UnblindedOutputBuilder> },
+    CreatePayToSelfWithOutputs { transaction: Box<Transaction>, tx_id: TxId },
     ReinstatedCancelledInboundTx,
     CoinbaseAbandonedSet,
     ClaimHtlcTransaction((TxId, MicroTari, MicroTari, Transaction)),
-    OutputStatusesByTxId {
-        statuses: Vec<OutputStatus>,
-        mined_height: Option<u64>,
-        block_hash: Option<BlockHash>,
-    },
+    OutputStatusesByTxId(OutputStatusesByTxId),
 }
 
 pub type OutputManagerEventSender = broadcast::Sender<Arc<OutputManagerEvent>>;
@@ -846,17 +834,13 @@ impl OutputManagerHandle {
     pub async fn get_output_statuses_by_tx_id(
         &mut self,
         tx_id: TxId,
-    ) -> Result<(Vec<OutputStatus>, Option<u64>, Option<BlockHash>), OutputManagerError> {
+    ) -> Result<OutputStatusesByTxId, OutputManagerError> {
         match self
             .handle
             .call(OutputManagerRequest::GetOutputStatusesByTxId(tx_id))
             .await??
         {
-            OutputManagerResponse::OutputStatusesByTxId {
-                statuses,
-                mined_height,
-                block_hash,
-            } => Ok((statuses, mined_height, block_hash)),
+            OutputManagerResponse::OutputStatusesByTxId(output_statuses_by_tx_id) => Ok(output_statuses_by_tx_id),
             _ => Err(OutputManagerError::UnexpectedApiResponse),
         }
     }

base_layer/wallet/src/output_manager_service/recovery/standard_outputs_recoverer.rs

@@ -142,7 +142,7 @@ where
             )?;
             let tx_id = TxId::new_random();
             let output_hex = db_output.commitment.to_hex();
-            if let Err(e) = self.db.add_unspent_output_with_tx_id(tx_id, db_output).await {
+            if let Err(e) = self.db.add_unspent_output_with_tx_id(tx_id, db_output) {
                 match e {
                     OutputManagerStorageError::DuplicateOutput => {
                         info!(