Refactor Tokio runtime out of wallet container

Currently the wallet container in LibWallet holds the tokio runtime required by the wallet. This was done to enable the Wallet FFI library to function as the FFI wallet object needed to hold onto the runtime or else it would be dropped. However it is only needed for the FFI instance of the wallet and interferes with the standard way of managing the runtime in the other binary applications.

This PR removes the runtime from the wallet container in `tari_wallet` and creates a FFI specific TariWallet container which holds onto the runtime instance just for the FFI library.

applications/tari_console_wallet/src/main.rs

@@ -18,15 +18,14 @@ use tari_comms_dht::{DbConnectionUrl, DhtConfig};
 use tari_core::{consensus::Network as NetworkType, transactions::types::CryptoFactories};
 use tari_p2p::initialization::CommsConfig;
 use tari_wallet::{
-    contacts_service::storage::sqlite_db::ContactsServiceSqliteDatabase,
     error::WalletError,
-    output_manager_service::storage::sqlite_db::OutputManagerSqliteDatabase,
-    storage::{sqlite_db::WalletSqliteDatabase, sqlite_utilities::initialize_sqlite_database_backends},
-    transaction_service::{config::TransactionServiceConfig, storage::sqlite_db::TransactionServiceSqliteDatabase},
+    storage::sqlite_utilities::initialize_sqlite_database_backends,
+    transaction_service::config::TransactionServiceConfig,
     wallet::WalletConfig,
     Wallet,
+    WalletSqlite,
 };
-use tokio::{runtime::Runtime, sync::RwLock};
+use tokio::sync::RwLock;
 use tonic::transport::Server;
 use tui::backend::CrosstermBackend;
 
@@ -98,29 +97,28 @@ fn main_inner() -> Result<(), ExitCodes> {
         info!(target: LOG_TARGET, "Default configuration created. Done.");
         return Ok(());
     }
-    let runtime = tokio::runtime::Builder::new()
+    let mut runtime = tokio::runtime::Builder::new()
         .threaded_scheduler()
         .enable_all()
         .build()
         .unwrap();
 
-    let wallet = setup_wallet(&node_config, wallet_identity, runtime)?;
+    let wallet = runtime.block_on(setup_wallet(&node_config, wallet_identity))?;
     debug!(target: LOG_TARGET, "Starting app");
-    let handle = wallet.runtime.handle().clone();
 
     let node_identity = wallet.comms.node_identity().as_ref().clone();
     let wallet = Arc::new(RwLock::new(wallet));
     let grpc = crate::grpc::WalletGrpcServer::new(wallet.clone());
 
     if !bootstrap.daemon_mode {
-        handle.spawn(run_grpc(grpc, node_config.grpc_address));
+        runtime.spawn(run_grpc(grpc, node_config.grpc_address));
         let app = App::<CrosstermBackend<Stdout>>::new(
             "Tari Console Wallet".into(),
             &node_identity,
             wallet.clone(),
             node_config.network,
         );
-        handle.block_on(ui::run(app))?;
+        runtime.block_on(ui::run(app))?;
         println!("The wallet is shutting down.");
         info!(
             target: LOG_TARGET,
@@ -130,7 +128,7 @@ fn main_inner() -> Result<(), ExitCodes> {
         Ok(())
     } else {
         println!("Starting grpc server");
-        handle
+        runtime
             .block_on(run_grpc(grpc, node_config.grpc_address))
             .map_err(|err| ExitCodes::GrpcError(err))?;
         println!("Shutting down");
@@ -139,20 +137,7 @@ fn main_inner() -> Result<(), ExitCodes> {
 }
 
 /// Setup the app environment and state for use by the UI
-fn setup_wallet(
-    config: &GlobalConfig,
-    node_identity: Arc<NodeIdentity>,
-    runtime: Runtime,
-) -> Result<
-    Wallet<
-        WalletSqliteDatabase,
-        TransactionServiceSqliteDatabase,
-        OutputManagerSqliteDatabase,
-        ContactsServiceSqliteDatabase,
-    >,
-    ExitCodes,
->
-{
+async fn setup_wallet(config: &GlobalConfig, node_identity: Arc<NodeIdentity>) -> Result<WalletSqlite, ExitCodes> {
     create_wallet_folder(
         &config
             .wallet_db_file
@@ -211,12 +196,12 @@ fn setup_wallet(
 
     let mut wallet = Wallet::new(
         wallet_config,
-        runtime,
         wallet_backend,
         transaction_backend.clone(),
         output_manager_backend,
         contacts_backend,
     )
+    .await
     .map_err(|e| {
         if let WalletError::CommsInitializationError(ce) = e {
             ExitCodes::WalletError(format!("Error initializing Comms: {}", ce.to_friendly_string()))
@@ -239,6 +224,7 @@ fn setup_wallet(
                     .expect("The seed peers should have an address")
                     .to_string(),
             )
+            .await
             .map_err(|e| ExitCodes::WalletError(format!("Error setting wallet base node peer. {}", e)))?;
     }
 

base_layer/core/tests/wallet.rs

@@ -165,27 +165,26 @@ fn wallet_base_node_integration_test() {
         }),
         Network::Rincewind,
     );
-    let alice_runtime = create_runtime();
-    let mut alice_wallet = Wallet::new(
-        alice_wallet_config,
-        alice_runtime,
-        WalletMemoryDatabase::new(),
-        TransactionMemoryDatabase::new(),
-        OutputManagerMemoryDatabase::new(),
-        ContactsServiceMemoryDatabase::new(),
-    )
-    .unwrap();
+    let mut runtime = create_runtime();
+    let mut alice_wallet = runtime
+        .block_on(Wallet::new(
+            alice_wallet_config,
+            WalletMemoryDatabase::new(),
+            TransactionMemoryDatabase::new(),
+            OutputManagerMemoryDatabase::new(),
+            ContactsServiceMemoryDatabase::new(),
+        ))
+        .unwrap();
     let mut alice_event_stream = alice_wallet.transaction_service.get_event_stream_fused();
 
-    alice_wallet
-        .set_base_node_peer(
+    runtime
+        .block_on(alice_wallet.set_base_node_peer(
             (*base_node_identity.public_key()).clone(),
             base_node_identity.public_address().clone().to_string(),
-        )
+        ))
         .unwrap();
 
-    alice_wallet
-        .runtime
+    runtime
         .block_on(alice_wallet.comms.peer_manager().add_peer(create_peer(
             bob_node_identity.public_key().clone(),
             bob_node_identity.public_address(),
@@ -209,18 +208,18 @@ fn wallet_base_node_integration_test() {
         user_agent: "tari/test-wallet".to_string(),
     };
     let bob_wallet_config = WalletConfig::new(bob_comms_config, factories.clone(), None, Network::Rincewind);
-    let bob_runtime = create_runtime();
-    let mut bob_wallet = Wallet::new(
-        bob_wallet_config,
-        bob_runtime,
-        WalletMemoryDatabase::new(),
-        TransactionMemoryDatabase::new(),
-        OutputManagerMemoryDatabase::new(),
-        ContactsServiceMemoryDatabase::new(),
-    )
-    .unwrap();
-    bob_wallet
-        .runtime
+
+    let bob_wallet = runtime
+        .block_on(Wallet::new(
+            bob_wallet_config,
+            WalletMemoryDatabase::new(),
+            TransactionMemoryDatabase::new(),
+            OutputManagerMemoryDatabase::new(),
+            ContactsServiceMemoryDatabase::new(),
+        ))
+        .unwrap();
+
+    runtime
         .block_on(bob_wallet.comms.peer_manager().add_peer(create_peer(
             alice_node_identity.public_key().clone(),
             alice_node_identity.public_address(),
@@ -230,13 +229,10 @@ fn wallet_base_node_integration_test() {
     log::info!("Finished Starting Wallets");
 
     // Transaction
-    let mut runtime = create_runtime();
-    alice_wallet
-        .runtime
+    runtime
         .block_on(alice_wallet.output_manager_service.add_output(utxo0))
         .unwrap();
-    alice_wallet
-        .runtime
+    runtime
         .block_on(
             alice_wallet
                 .comms
@@ -246,8 +242,7 @@ fn wallet_base_node_integration_test() {
         .unwrap();
 
     let value = MicroTari::from(1000);
-    alice_wallet
-        .runtime
+    runtime
         .block_on(alice_wallet.transaction_service.send_transaction(
             bob_node_identity.public_key().clone(),
             value,
@@ -353,8 +348,8 @@ fn wallet_base_node_integration_test() {
         assert!(mined, "Transaction has not been mined before timeout");
     });
 
-    alice_wallet.shutdown();
-    bob_wallet.shutdown();
+    runtime.block_on(alice_wallet.shutdown());
+    runtime.block_on(bob_wallet.shutdown());
     let _ = shutdown.trigger();
     runtime.block_on(base_node.comms.shutdown());
 }

base_layer/wallet/src/lib.rs

@@ -16,8 +16,6 @@ pub mod wallet;
 #[cfg(feature = "test_harness")]
 pub mod testnet_utils;
 
-pub use wallet::Wallet;
-
 #[macro_use]
 extern crate diesel;
 #[macro_use]
@@ -27,3 +25,18 @@ extern crate lazy_static;
 
 pub mod schema;
 // pub mod text_message_service;
+
+pub use wallet::Wallet;
+
+use crate::{
+    contacts_service::storage::sqlite_db::ContactsServiceSqliteDatabase,
+    output_manager_service::storage::sqlite_db::OutputManagerSqliteDatabase,
+    storage::sqlite_db::WalletSqliteDatabase,
+    transaction_service::storage::sqlite_db::TransactionServiceSqliteDatabase,
+};
+pub type WalletSqlite = Wallet<
+    WalletSqliteDatabase,
+    TransactionServiceSqliteDatabase,
+    OutputManagerSqliteDatabase,
+    ContactsServiceSqliteDatabase,
+>;