Merge branch 'development' into philip-saf-fixes

applications/tari_console_wallet/src/ui/components/base_node.rs

@@ -21,7 +21,7 @@
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
 use crate::ui::{components::Component, state::AppState};
-use tari_wallet::base_node_service::service::OnlineState;
+use tari_wallet::connectivity_service::OnlineStatus;
 use tui::{
     backend::Backend,
     layout::Rect,
@@ -45,17 +45,17 @@ impl<B: Backend> Component<B> for BaseNode {
         let base_node_state = app_state.get_base_node_state();
 
         let chain_info = match base_node_state.online {
-            OnlineState::Connecting => Spans::from(vec![
+            OnlineStatus::Connecting => Spans::from(vec![
                 Span::styled("Chain Tip:", Style::default().fg(Color::Magenta)),
                 Span::raw(" "),
                 Span::styled("Connecting...", Style::default().fg(Color::Reset)),
             ]),
-            OnlineState::Offline => Spans::from(vec![
+            OnlineStatus::Offline => Spans::from(vec![
                 Span::styled("Chain Tip:", Style::default().fg(Color::Magenta)),
                 Span::raw(" "),
                 Span::styled("Offline", Style::default().fg(Color::Red)),
             ]),
-            OnlineState::Online => {
+            OnlineStatus::Online => {
                 if let Some(metadata) = base_node_state.clone().chain_metadata {
                     let tip = metadata.height_of_longest_chain();
 

applications/tari_console_wallet/src/ui/state/app_state.rs

@@ -68,7 +68,10 @@ use tari_wallet::{
     util::emoji::EmojiId,
     WalletSqlite,
 };
-use tokio::sync::{watch, RwLock};
+use tokio::{
+    sync::{watch, RwLock},
+    task,
+};
 
 const LOG_TARGET: &str = "wallet::console_wallet::app_state";
 
@@ -665,15 +668,7 @@ impl AppStateInner {
             )
             .await?;
 
-        if let Err(e) = self
-            .wallet
-            .transaction_service
-            .validate_transactions(ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating transactions: {}", e);
-        }
-        self.validate_outputs().await;
+        self.spawn_transaction_revalidation_task();
 
         self.data.base_node_previous = self.data.base_node_selected.clone();
         self.data.base_node_selected = peer.clone();
@@ -701,15 +696,7 @@ impl AppStateInner {
             )
             .await?;
 
-        if let Err(e) = self
-            .wallet
-            .transaction_service
-            .validate_transactions(ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating transactions: {}", e);
-        }
-        self.validate_outputs().await;
+        self.spawn_transaction_revalidation_task();
 
         self.data.base_node_previous = self.data.base_node_selected.clone();
         self.data.base_node_selected = peer.clone();
@@ -751,15 +738,7 @@ impl AppStateInner {
             )
             .await?;
 
-        if let Err(e) = self
-            .wallet
-            .transaction_service
-            .validate_transactions(ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating transactions: {}", e);
-        }
-        self.validate_outputs().await;
+        self.spawn_transaction_revalidation_task();
 
         self.data.base_node_peer_custom = None;
         self.data.base_node_selected = previous;
@@ -778,33 +757,39 @@ impl AppStateInner {
         Ok(())
     }
 
-    pub async fn validate_outputs(&mut self) {
-        if let Err(e) = self
-            .wallet
-            .output_manager_service
-            .validate_txos(TxoValidationType::Unspent, ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating UTXOs: {}", e);
-        }
+    pub fn spawn_transaction_revalidation_task(&mut self) {
+        let mut txn_service = self.wallet.transaction_service.clone();
+        let mut output_manager_service = self.wallet.output_manager_service.clone();
 
-        if let Err(e) = self
-            .wallet
-            .output_manager_service
-            .validate_txos(TxoValidationType::Spent, ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating STXOs: {}", e);
-        }
+        task::spawn(async move {
+            if let Err(e) = txn_service
+                .validate_transactions(ValidationRetryStrategy::UntilSuccess)
+                .await
+            {
+                error!(target: LOG_TARGET, "Problem validating transactions: {}", e);
+            }
 
-        if let Err(e) = self
-            .wallet
-            .output_manager_service
-            .validate_txos(TxoValidationType::Invalid, ValidationRetryStrategy::UntilSuccess)
-            .await
-        {
-            error!(target: LOG_TARGET, "Problem validating Invalid TXOs: {}", e);
-        }
+            if let Err(e) = output_manager_service
+                .validate_txos(TxoValidationType::Unspent, ValidationRetryStrategy::UntilSuccess)
+                .await
+            {
+                error!(target: LOG_TARGET, "Problem validating UTXOs: {}", e);
+            }
+
+            if let Err(e) = output_manager_service
+                .validate_txos(TxoValidationType::Spent, ValidationRetryStrategy::UntilSuccess)
+                .await
+            {
+                error!(target: LOG_TARGET, "Problem validating STXOs: {}", e);
+            }
+
+            if let Err(e) = output_manager_service
+                .validate_txos(TxoValidationType::Invalid, ValidationRetryStrategy::UntilSuccess)
+                .await
+            {
+                error!(target: LOG_TARGET, "Problem validating Invalid TXOs: {}", e);
+            }
+        });
     }
 }
 

applications/tari_console_wallet/src/utils/crossterm_events.rs

@@ -70,7 +70,9 @@ impl CrosstermEvents {
                 ) {
                     Ok(true) => {
                         if let Ok(CEvent::Key(key)) = event::read() {
-                            tx.send(Event::Input(key)).unwrap();
+                            if let Err(e) = tx.send(Event::Input(key)) {
+                                warn!(target: LOG_TARGET, "Error sending Tick event on MPSC channel: {}", e);
+                            }
                         }
                     },
                     Ok(false) => {},

base_layer/core/src/base_node/comms_interface/inbound_handlers.rs

@@ -30,7 +30,7 @@ use crate::{
         OutboundNodeCommsInterface,
     },
     blocks::{block_header::BlockHeader, Block, NewBlock, NewBlockTemplate},
-    chain_storage::{async_db::AsyncBlockchainDb, BlockAddResult, BlockchainBackend, ChainBlock},
+    chain_storage::{async_db::AsyncBlockchainDb, BlockAddResult, BlockchainBackend, ChainBlock, PrunedOutput},
     consensus::{ConsensusConstants, ConsensusManager},
     mempool::{async_mempool, Mempool},
     proof_of_work::{Difficulty, PowAlgorithm},
@@ -224,12 +224,14 @@ where T: BlockchainBackend + 'static
             },
             NodeCommsRequest::FetchMatchingUtxos(utxo_hashes) => {
                 let mut res = Vec::with_capacity(utxo_hashes.len());
-                for (output, spent) in (self.blockchain_db.fetch_utxos(utxo_hashes).await?)
+                for (pruned_output, spent) in (self.blockchain_db.fetch_utxos(utxo_hashes).await?)
                     .into_iter()
                     .flatten()
                 {
-                    if !spent {
-                        res.push(output);
+                    if let PrunedOutput::NotPruned { output } = pruned_output {
+                        if !spent {
+                            res.push(output);
+                        }
                     }
                 }
                 Ok(NodeCommsResponse::TransactionOutputs(res))
@@ -240,7 +242,10 @@ where T: BlockchainBackend + 'static
                     .fetch_utxos(hashes)
                     .await?
                     .into_iter()
-                    .filter_map(|opt| opt.map(|(output, _)| output))
+                    .filter_map(|opt| match opt {
+                        Some((PrunedOutput::NotPruned { output }, _)) => Some(output),
+                        _ => None,
+                    })
                     .collect();
                 Ok(NodeCommsResponse::TransactionOutputs(res))
             },

base_layer/core/src/base_node/comms_interface/local_interface.rs

@@ -32,7 +32,7 @@ use crate::{
     chain_storage::HistoricalBlock,
     proof_of_work::PowAlgorithm,
     transactions::{
-        transaction::{TransactionKernel, TransactionOutput},
+        transaction::TransactionKernel,
         types::{Commitment, HashOutput, Signature},
     },
 };
@@ -43,7 +43,10 @@ use tokio::sync::broadcast;
 
 pub type BlockEventSender = broadcast::Sender<Arc<BlockEvent>>;
 pub type BlockEventReceiver = broadcast::Receiver<Arc<BlockEvent>>;
-use crate::base_node::comms_interface::comms_request::GetNewBlockTemplateRequest;
+use crate::{
+    base_node::comms_interface::comms_request::GetNewBlockTemplateRequest,
+    transactions::transaction::TransactionOutput,
+};
 
 /// The InboundNodeCommsInterface provides an interface to request information from the current local node by other
 /// internal services.

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

@@ -22,7 +22,7 @@
 
 use crate::{
     base_node::{rpc::BaseNodeWalletService, state_machine_service::states::StateInfo, StateMachineHandle},
-    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend},
+    chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, PrunedOutput},
     mempool::{service::MempoolHandle, TxStorageResponse},
     proto::{
         base_node::{
@@ -288,15 +288,17 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
 
         let db = self.db();
         let mut res = Vec::with_capacity(message.output_hashes.len());
-        for (output, spent) in (db
+        for (pruned_output, spent) in (db
             .fetch_utxos(message.output_hashes)
             .await
             .map_err(RpcStatus::log_internal_error(LOG_TARGET))?)
         .into_iter()
         .flatten()
         {
-            if !spent {
-                res.push(output);
+            if let PrunedOutput::NotPruned { output } = pruned_output {
+                if !spent {
+                    res.push(output);
+                }
             }
         }
 

base_layer/core/src/chain_storage/async_db.rs

@@ -139,9 +139,9 @@ impl<B: BlockchainBackend + 'static> AsyncBlockchainDb<B> {
     make_async_fn!(fetch_horizon_data() -> Option<HorizonData>, "fetch_horizon_data");
 
     //---------------------------------- TXO --------------------------------------------//
-    make_async_fn!(fetch_utxo(hash: HashOutput) -> Option<TransactionOutput>, "fetch_utxo");
+    make_async_fn!(fetch_utxo(hash: HashOutput) -> Option<PrunedOutput>, "fetch_utxo");
 
-    make_async_fn!(fetch_utxos(hashes: Vec<HashOutput>) -> Vec<Option<(TransactionOutput, bool)>>, "fetch_utxos");
+    make_async_fn!(fetch_utxos(hashes: Vec<HashOutput>) -> Vec<Option<(PrunedOutput, bool)>>, "fetch_utxos");
 
     make_async_fn!(fetch_utxos_by_mmr_position(start: u64, end: u64, deleted: Arc<Bitmap>) -> (Vec<PrunedOutput>, Bitmap), "fetch_utxos_by_mmr_position");
 

base_layer/core/src/chain_storage/blockchain_backend.rs

@@ -15,7 +15,7 @@ use crate::{
         MmrTree,
     },
     transactions::{
-        transaction::{TransactionInput, TransactionKernel, TransactionOutput},
+        transaction::{TransactionInput, TransactionKernel},
         types::{HashOutput, Signature},
     },
 };
@@ -103,10 +103,7 @@ pub trait BlockchainBackend: Send + Sync {
     ) -> Result<(Vec<PrunedOutput>, Bitmap), ChainStorageError>;
 
     /// Fetch a specific output. Returns the output and the leaf index in the output MMR
-    fn fetch_output(
-        &self,
-        output_hash: &HashOutput,
-    ) -> Result<Option<(TransactionOutput, u32, u64)>, ChainStorageError>;
+    fn fetch_output(&self, output_hash: &HashOutput) -> Result<Option<(PrunedOutput, u32, u64)>, ChainStorageError>;
 
     /// Fetch all outputs in a block
     fn fetch_outputs_in_block(&self, header_hash: &HashOutput) -> Result<Vec<PrunedOutput>, ChainStorageError>;

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -47,7 +47,7 @@ use crate::{
     proof_of_work::{monero_rx::MoneroPowData, PowAlgorithm, TargetDifficultyWindow},
     tari_utilities::epoch_time::EpochTime,
     transactions::{
-        transaction::{TransactionKernel, TransactionOutput},
+        transaction::TransactionKernel,
         types::{Commitment, HashDigest, HashOutput, Signature},
     },
     validation::{DifficultyCalculator, HeaderValidation, OrphanValidation, PostOrphanBodyValidation, ValidationError},
@@ -284,18 +284,15 @@ where B: BlockchainBackend
     }
 
     // Fetch the utxo
-    pub fn fetch_utxo(&self, hash: HashOutput) -> Result<Option<TransactionOutput>, ChainStorageError> {
+    pub fn fetch_utxo(&self, hash: HashOutput) -> Result<Option<PrunedOutput>, ChainStorageError> {
         let db = self.db_read_access()?;
         Ok(db.fetch_output(&hash)?.map(|(out, _index, _)| out))
     }
 
     /// Return a list of matching utxos, with each being `None` if not found. If found, the transaction
     /// output, and a boolean indicating if the UTXO was spent as of the block hash specified or the tip if not
     /// specified.
-    pub fn fetch_utxos(
-        &self,
-        hashes: Vec<HashOutput>,
-    ) -> Result<Vec<Option<(TransactionOutput, bool)>>, ChainStorageError> {
+    pub fn fetch_utxos(&self, hashes: Vec<HashOutput>) -> Result<Vec<Option<(PrunedOutput, bool)>>, ChainStorageError> {
         let db = self.db_read_access()?;
         let deleted = db.fetch_deleted_bitmap()?;
 

base_layer/core/src/chain_storage/lmdb_db/lmdb_db.rs

@@ -1616,10 +1616,7 @@ impl BlockchainBackend for LMDBDatabase {
         Ok((result, difference_bitmap))
     }
 
-    fn fetch_output(
-        &self,
-        output_hash: &HashOutput,
-    ) -> Result<Option<(TransactionOutput, u32, u64)>, ChainStorageError> {
+    fn fetch_output(&self, output_hash: &HashOutput) -> Result<Option<(PrunedOutput, u32, u64)>, ChainStorageError> {
         debug!(target: LOG_TARGET, "Fetch output: {}", output_hash.to_hex());
         let txn = self.read_transaction()?;
         if let Some((index, key)) =
@@ -1632,20 +1629,33 @@ impl BlockchainBackend for LMDBDatabase {
                 index,
                 key
             );
-            if let Some(output) = lmdb_get::<_, TransactionOutputRowData>(&txn, &self.utxos_db, key.as_str())? {
-                if output.output.is_none() {
-                    error!(
-                        target: LOG_TARGET,
-                        "Tried to fetch pruned output: {} ({}, {})",
-                        output_hash.to_hex(),
-                        index,
-                        key
-                    );
-                    unimplemented!("Output has been pruned");
-                }
-                Ok(Some((output.output.unwrap(), output.mmr_position, output.mined_height)))
-            } else {
-                Ok(None)
+            match lmdb_get::<_, TransactionOutputRowData>(&txn, &self.utxos_db, key.as_str())? {
+                Some(TransactionOutputRowData {
+                    output: Some(o),
+                    mmr_position,
+                    mined_height,
+                    ..
+                }) => Ok(Some((
+                    PrunedOutput::NotPruned { output: o },
+                    mmr_position,
+                    mined_height,
+                ))),
+                Some(TransactionOutputRowData {
+                    output: None,
+                    mmr_position,
+                    mined_height,
+                    hash,
+                    witness_hash,
+                    ..
+                }) => Ok(Some((
+                    PrunedOutput::Pruned {
+                        output_hash: hash,
+                        witness_hash,
+                    },
+                    mmr_position,
+                    mined_height,
+                ))),
+                _ => Ok(None),
             }
         } else {
             debug!(

base_layer/core/src/chain_storage/pruned_output.rs

@@ -22,6 +22,7 @@
 use crate::transactions::{transaction::TransactionOutput, types::HashOutput};
 
 #[allow(clippy::large_enum_variant)]
+#[derive(Debug, PartialEq)]
 pub enum PrunedOutput {
     Pruned {
         output_hash: HashOutput,