Merge branch 'development' into core-sync-latency-based-peer-switch

* development:
  refactor: minor changes while investigating faucet (tari-project#3753)

base_layer/core/src/base_node/rpc/service.rs

@@ -35,7 +35,7 @@ use crate::{
         state_machine_service::states::StateInfo,
         StateMachineHandle,
     },
-    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, PrunedOutput, UtxoMinedInfo},
+    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, PrunedOutput},
     mempool::{service::MempoolHandle, TxStorageResponse},
     proto,
     proto::{
@@ -341,23 +341,37 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
 
     async fn utxo_query(&self, request: Request<UtxoQueryRequest>) -> Result<Response<UtxoQueryResponses>, RpcStatus> {
         let message = request.into_message();
+        if message.output_hashes.is_empty() {
+            return Err(RpcStatus::bad_request("Empty output hashes"));
+        }
+        const MAX_ALLOWED_QUERY_SIZE: usize = 512;
+        if message.output_hashes.len() > MAX_ALLOWED_QUERY_SIZE {
+            return Err(RpcStatus::bad_request(format!(
+                "Exceeded maximum allowed query hashes. Max: {}",
+                MAX_ALLOWED_QUERY_SIZE
+            )));
+        }
+
         let db = self.db();
-        let mut res = Vec::with_capacity(message.output_hashes.len());
-        for UtxoMinedInfo {
-            output,
-            mmr_position,
-            mined_height: height,
-            header_hash,
-        } in (db
+
+        debug!(
+            target: LOG_TARGET,
+            "Querying {} UTXO(s) for mined state",
+            message.output_hashes.len(),
+        );
+
+        let mined_info_resp = db
             .fetch_utxos_and_mined_info(message.output_hashes)
             .await
-            .map_err(RpcStatus::log_internal_error(LOG_TARGET))?)
-        .into_iter()
-        .flatten()
-        {
-            res.push((output, mmr_position, height, header_hash));
-        }
+            .map_err(RpcStatus::log_internal_error(LOG_TARGET))?;
 
+        let num_mined = mined_info_resp.iter().filter(|opt| opt.is_some()).count();
+        debug!(
+            target: LOG_TARGET,
+            "Found {} mined and {} unmined UTXO(s)",
+            num_mined,
+            mined_info_resp.len() - num_mined
+        );
         let metadata = self
             .db
             .get_chain_metadata()
@@ -367,20 +381,19 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
         Ok(Response::new(UtxoQueryResponses {
             height_of_longest_chain: metadata.height_of_longest_chain(),
             best_block: metadata.best_block().clone(),
-            responses: res
+            responses: mined_info_resp
                 .into_iter()
-                .map(
-                    |(output, mmr_position, mined_height, mined_in_block)| UtxoQueryResponse {
-                        mmr_position: mmr_position.into(),
-                        mined_height,
-                        mined_in_block,
-                        output_hash: output.hash(),
-                        output: match output {
-                            PrunedOutput::Pruned { .. } => None,
-                            PrunedOutput::NotPruned { output } => Some(output.into()),
-                        },
+                .flatten()
+                .map(|utxo| UtxoQueryResponse {
+                    mmr_position: utxo.mmr_position.into(),
+                    mined_height: utxo.mined_height,
+                    mined_in_block: utxo.header_hash,
+                    output_hash: utxo.output.hash(),
+                    output: match utxo.output {
+                        PrunedOutput::Pruned { .. } => None,
+                        PrunedOutput::NotPruned { output } => Some(output.into()),
                     },
-                )
+                })
                 .collect(),
         }))
     }

base_layer/service_framework/src/stack.rs

@@ -81,11 +81,9 @@ impl StackBuilder {
 
         // Run all the initializers concurrently and check each Result returning an error
         // on the first one that failed.
-        for result in future::join_all(init_futures).await {
-            result?;
-        }
+        future::try_join_all(init_futures).await?;
 
-        let _ = notifier.trigger();
+        notifier.trigger();
 
         Ok(context.into_inner())
     }

base_layer/wallet/src/wallet.rs

@@ -404,18 +404,19 @@ where
             .import_utxo(amount, source_public_key.clone(), message, Some(features.maturity))
             .await?;
 
+        let commitment_hex = unblinded_output
+            .as_transaction_input(&self.factories.commitment)?
+            .commitment()
+            .map_err(WalletError::TransactionError)?
+            .to_hex();
+
         self.output_manager_service
-            .add_unvalidated_output(tx_id, unblinded_output.clone(), None)
+            .add_unvalidated_output(tx_id, unblinded_output, None)
             .await?;
 
         info!(
             target: LOG_TARGET,
-            "UTXO (Commitment: {}) imported into wallet",
-            unblinded_output
-                .as_transaction_input(&self.factories.commitment)?
-                .commitment()
-                .map_err(WalletError::TransactionError)?
-                .to_hex()
+            "UTXO (Commitment: {}) imported into wallet", commitment_hex
         );
 
         Ok(tx_id)

base_layer/wallet/tests/wallet.rs

@@ -70,6 +70,7 @@ use tari_key_manager::{cipher_seed::CipherSeed, mnemonic::Mnemonic};
 use tari_p2p::{initialization::P2pConfig, transport::TransportType, Network, DEFAULT_DNS_NAME_SERVER};
 use tari_shutdown::{Shutdown, ShutdownSignal};
 use tari_test_utils::random;
+use tari_utilities::Hashable;
 use tari_wallet::{
     contacts_service::storage::{database::Contact, sqlite_db::ContactsServiceSqliteDatabase},
     error::{WalletError, WalletStorageError},
@@ -101,6 +102,7 @@ pub mod support;
 use support::{comms_and_services::get_next_memory_address, utils::make_input};
 use tari_common_types::transaction::TransactionStatus;
 use tari_core::covenants::Covenant;
+use tari_wallet::output_manager_service::storage::database::OutputManagerDatabase;
 
 fn create_peer(public_key: CommsPublicKey, net_address: Multiaddr) -> Peer {
     Peer::new(
@@ -685,8 +687,8 @@ fn test_store_and_forward_send_tx() {
 
 #[tokio::test]
 async fn test_import_utxo() {
-    let shutdown = Shutdown::new();
     let factories = CryptoFactories::default();
+    let shutdown = Shutdown::new();
     let alice_identity = NodeIdentity::random(
         &mut OsRng,
         "/ip4/127.0.0.1/tcp/24521".parse().unwrap(),
@@ -740,7 +742,7 @@ async fn test_import_utxo() {
         WalletDatabase::new(WalletSqliteDatabase::new(connection.clone(), None).unwrap()),
         TransactionServiceSqliteDatabase::new(connection.clone(), None),
         OutputManagerSqliteDatabase::new(connection.clone(), None),
-        ContactsServiceSqliteDatabase::new(connection),
+        ContactsServiceSqliteDatabase::new(connection.clone()),
         shutdown.to_signal(),
         CipherSeed::new(),
     )
@@ -754,6 +756,8 @@ async fn test_import_utxo() {
 
     let p = TestParams::new();
     let utxo = create_unblinded_output(script.clone(), features.clone(), p.clone(), 20000 * uT);
+    let output = utxo.as_transaction_output(&factories).unwrap();
+    let expected_output_hash = output.hash();
 
     let tx_id = alice_wallet
         .import_utxo(
@@ -787,6 +791,9 @@ async fn test_import_utxo() {
 
     assert_eq!(completed_tx.amount, 20000 * uT);
     assert_eq!(completed_tx.status, TransactionStatus::Imported);
+    let db = OutputManagerDatabase::new(OutputManagerSqliteDatabase::new(connection, None));
+    let outputs = db.fetch_outputs_by_tx_id(tx_id).await.unwrap();
+    assert!(outputs.iter().any(|o| { o.hash == expected_output_hash }));
 }
 
 #[test]