feat!: integration of new transaction and output validation

applications/tari_base_node/src/command_handler.rs

@@ -74,11 +74,6 @@ pub enum StatusOutput {
     Full,
 }
 
-pub enum StatusOutput {
-    Log,
-    Full,
-}
-
 pub struct CommandHandler {
     executor: runtime::Handle,
     config: Arc<GlobalConfig>,

applications/tari_base_node/src/main.rs

@@ -95,7 +95,7 @@ mod status_line;
 mod utils;
 
 use crate::command_handler::{CommandHandler, StatusOutput};
-use futures::{future::Fuse, pin_mut, FutureExt};
+use futures::{pin_mut, FutureExt};
 use log::*;
 use opentelemetry::{self, global, KeyValue};
 use parser::Parser;
@@ -119,7 +119,7 @@ use tari_shutdown::{Shutdown, ShutdownSignal};
 use tokio::{
     runtime,
     task,
-    time::{self, Delay},
+    time::{self},
 };
 use tonic::transport::Server;
 use tracing_subscriber::{layer::SubscriberExt, Registry};

applications/tari_console_wallet/src/init/mod.rs

@@ -47,13 +47,9 @@ use tari_shutdown::ShutdownSignal;
 use tari_wallet::{
     base_node_service::config::BaseNodeServiceConfig,
     error::{WalletError, WalletStorageError},
-    output_manager_service::{config::OutputManagerServiceConfig, TxoValidationType},
+    output_manager_service::config::OutputManagerServiceConfig,
     storage::{database::WalletDatabase, sqlite_utilities::initialize_sqlite_database_backends},
-    transaction_service::{
-        config::{TransactionRoutingMechanism, TransactionServiceConfig},
-        tasks::start_transaction_validation_and_broadcast_protocols::start_transaction_validation_and_broadcast_protocols,
-    },
-    types::ValidationRetryStrategy,
+    transaction_service::config::{TransactionRoutingMechanism, TransactionServiceConfig},
     Wallet,
     WalletConfig,
     WalletSqlite,
@@ -391,6 +387,7 @@ pub async fn init_wallet(
             prevent_fee_gt_amount: config.prevent_fee_gt_amount,
             event_channel_size: config.output_manager_event_channel_size,
             base_node_update_publisher_channel_size: config.base_node_update_publisher_channel_size,
+            num_confirmations_required: config.transaction_num_confirmations_required,
             ..Default::default()
         }),
         config.network.into(),
@@ -500,12 +497,7 @@ pub async fn start_wallet(
         if let Err(e) = wallet.transaction_service.restart_transaction_protocols().await {
             error!(target: LOG_TARGET, "Problem restarting transaction protocols: {}", e);
         }
-        if let Err(e) = start_transaction_validation_and_broadcast_protocols(
-            wallet.transaction_service.clone(),
-            ValidationRetryStrategy::UntilSuccess,
-        )
-        .await
-        {
+        if let Err(e) = wallet.transaction_service.validate_transactions().await {
             error!(
                 target: LOG_TARGET,
                 "Problem validating and restarting transaction protocols: {}", e
@@ -521,37 +513,12 @@ pub async fn start_wallet(
 async fn validate_txos(wallet: &mut WalletSqlite) -> Result<(), ExitCodes> {
     debug!(target: LOG_TARGET, "Starting TXO validations.");
 
-    // Unspent TXOs
-    wallet
-        .output_manager_service
-        .validate_txos(TxoValidationType::Unspent, ValidationRetryStrategy::UntilSuccess)
-        .await
-        .map_err(|e| {
-            error!(target: LOG_TARGET, "Error validating Unspent TXOs: {}", e);
-            ExitCodes::WalletError(e.to_string())
-        })?;
-
-    // Spent TXOs
-    wallet
-        .output_manager_service
-        .validate_txos(TxoValidationType::Spent, ValidationRetryStrategy::UntilSuccess)
-        .await
-        .map_err(|e| {
-            error!(target: LOG_TARGET, "Error validating Spent TXOs: {}", e);
-            ExitCodes::WalletError(e.to_string())
-        })?;
-
-    // Invalid TXOs
-    wallet
-        .output_manager_service
-        .validate_txos(TxoValidationType::Invalid, ValidationRetryStrategy::UntilSuccess)
-        .await
-        .map_err(|e| {
-            error!(target: LOG_TARGET, "Error validating Invalid TXOs: {}", e);
-            ExitCodes::WalletError(e.to_string())
-        })?;
+    wallet.output_manager_service.validate_txos().await.map_err(|e| {
+        error!(target: LOG_TARGET, "Error validating Unspent TXOs: {}", e);
+        ExitCodes::WalletError(e.to_string())
+    })?;
 
-    debug!(target: LOG_TARGET, "TXO validations completed.");
+    debug!(target: LOG_TARGET, "TXO validations started.");
 
     Ok(())
 }

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

@@ -218,7 +218,7 @@ impl TransactionsTab {
                 format!("{}", local_time.format("%Y-%m-%d %H:%M:%S")),
                 Style::default().fg(text_color),
             )));
-            let status = if t.cancelled && t.status == TransactionStatus::Coinbase {
+            let status = if (t.cancelled || !t.valid) && t.status == TransactionStatus::Coinbase {
                 "Abandoned".to_string()
             } else if t.cancelled {
                 "Cancelled".to_string()

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

@@ -52,12 +52,11 @@ use tari_wallet::{
     base_node_service::{handle::BaseNodeEventReceiver, service::BaseNodeState},
     connectivity_service::WalletConnectivityHandle,
     contacts_service::storage::database::Contact,
-    output_manager_service::{handle::OutputManagerEventReceiver, service::Balance, TxId, TxoValidationType},
+    output_manager_service::{handle::OutputManagerEventReceiver, service::Balance, TxId},
     transaction_service::{
         handle::TransactionEventReceiver,
         storage::models::{CompletedTransaction, TransactionStatus},
     },
-    types::ValidationRetryStrategy,
     WalletSqlite,
 };
 
@@ -324,7 +323,7 @@ impl AppState {
             self.cached_data
                 .completed_txs
                 .iter()
-                .filter(|tx| !(tx.cancelled && tx.status == TransactionStatus::Coinbase))
+                .filter(|tx| !((tx.cancelled || !tx.valid) && tx.status == TransactionStatus::Coinbase))
                 .collect()
         } else {
             self.cached_data.completed_txs.iter().collect()
@@ -794,33 +793,13 @@ impl AppStateInner {
         let mut output_manager_service = self.wallet.output_manager_service.clone();
 
         task::spawn(async move {
-            if let Err(e) = txn_service
-                .validate_transactions(ValidationRetryStrategy::UntilSuccess)
-                .await
-            {
+            if let Err(e) = txn_service.validate_transactions().await {
                 error!(target: LOG_TARGET, "Problem validating transactions: {}", e);
             }
 
-            if let Err(e) = output_manager_service
-                .validate_txos(TxoValidationType::Unspent, ValidationRetryStrategy::UntilSuccess)
-                .await
-            {
+            if let Err(e) = output_manager_service.validate_txos().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/ui/state/wallet_event_monitor.rs

@@ -176,7 +176,7 @@ impl WalletEventMonitor {
                         match result {
                             Ok(msg) => {
                                 trace!(target: LOG_TARGET, "Output Manager Service Callback Handler event {:?}", msg);
-                                if let OutputManagerEvent::TxoValidationSuccess(_,_) = &*msg {
+                                if let OutputManagerEvent::TxoValidationSuccess(_) = &*msg {
                                     self.trigger_balance_refresh().await;
                                 }
                             },

applications/tari_console_wallet/src/ui/widgets/list_state.rs

@@ -83,7 +83,7 @@ impl WindowedListState {
             let i = match self.selected {
                 Some(i) => {
                     if i >= self.num_items - 1 {
-                        i
+                        0
                     } else {
                         i + 1
                     }
@@ -101,7 +101,7 @@ impl WindowedListState {
             let i = match self.selected {
                 Some(i) => {
                     if i == 0 {
-                        i
+                        self.num_items - 1
                     } else {
                         i - 1
                     }

base_layer/core/src/base_node/proto/wallet_rpc.proto

@@ -67,13 +67,16 @@ message FetchUtxosResponse {
 message QueryDeletedRequest{
     repeated uint64 mmr_positions = 1;
     google.protobuf.BytesValue chain_must_include_header = 2;
+    bool include_deleted_block_data = 3;
 }
 
 message QueryDeletedResponse {
     repeated uint64 deleted_positions = 1;
     repeated uint64 not_deleted_positions = 2;
     bytes best_block = 3;
     uint64 height_of_longest_chain = 4;
+    repeated bytes blocks_deleted_in = 5;
+    repeated uint64 heights_deleted_at = 6;
 }
 
 message UtxoQueryRequest{

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

@@ -89,6 +89,12 @@ pub trait BaseNodeWalletService: Send + Sync + 'static {
         &self,
         request: Request<QueryDeletedRequest>,
     ) -> Result<Response<QueryDeletedResponse>, RpcStatus>;
+
+    #[rpc(method = 9)]
+    async fn get_header_by_height(
+        &self,
+        request: Request<u64>,
+    ) -> Result<Response<proto::core::BlockHeader>, RpcStatus>;
 }
 
 #[cfg(feature = "base_node")]

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

@@ -22,6 +22,7 @@
 use crate::{
     base_node::{rpc::BaseNodeWalletService, state_machine_service::states::StateInfo, StateMachineHandle},
     chain_storage::{async_db::AsyncBlockchainDb, BlockchainBackend, PrunedOutput, UtxoMinedInfo},
+    crypto::tari_utilities::Hashable,
     mempool::{service::MempoolHandle, TxStorageResponse},
     proto,
     proto::{
@@ -351,12 +352,6 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
             .await
             .map_err(RpcStatus::log_internal_error(LOG_TARGET))?;
 
-        // let deleted = self
-        //     .db
-        //     .fetch_complete_deleted_bitmap_at(metadata.best_block().clone())
-        //     .await
-        //     .map_err(RpcStatus::log_internal_error(LOG_TARGET))?
-        //     .into_bytes();
         Ok(Response::new(UtxoQueryResponses {
             height_of_longest_chain: metadata.height_of_longest_chain(),
             best_block: metadata.best_block().clone(),
@@ -428,11 +423,28 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
                 not_deleted_positions.push(position);
             }
         }
+
+        let mut blocks_deleted_in = Vec::new();
+        let mut heights_deleted_at = Vec::new();
+        if message.include_deleted_block_data {
+            let headers = self
+                .db
+                .fetch_headers_of_deleted_positions(deleted_positions.clone())
+                .await
+                .map_err(RpcStatus::log_internal_error(LOG_TARGET))?;
+            for header in headers.iter() {
+                heights_deleted_at.push(header.height);
+                blocks_deleted_in.push(header.hash());
+            }
+        }
+
         Ok(Response::new(QueryDeletedResponse {
             height_of_longest_chain: metadata.height_of_longest_chain(),
             best_block: metadata.best_block().clone(),
             deleted_positions,
             not_deleted_positions,
+            blocks_deleted_in,
+            heights_deleted_at,
         }))
     }
 
@@ -467,4 +479,19 @@ impl<B: BlockchainBackend + 'static> BaseNodeWalletService for BaseNodeWalletRpc
 
         Ok(Response::new(header.into()))
     }
+
+    async fn get_header_by_height(
+        &self,
+        request: Request<u64>,
+    ) -> Result<Response<proto::core::BlockHeader>, RpcStatus> {
+        let height = request.into_message();
+        let header = self
+            .db()
+            .fetch_header(height)
+            .await
+            .map_err(RpcStatus::log_internal_error(LOG_TARGET))?
+            .ok_or_else(|| RpcStatus::not_found(format!("Header not found at height {}", height)))?;
+
+        Ok(Response::new(header.into()))
+    }
 }

base_layer/core/src/chain_storage/async_db.rs

@@ -235,6 +235,8 @@ impl<B: BlockchainBackend + 'static> AsyncBlockchainDb<B> {
 
     make_async_fn!(fetch_complete_deleted_bitmap_at(hash: HashOutput) -> CompleteDeletedBitmap, "fetch_deleted_bitmap");
 
+    make_async_fn!(fetch_headers_of_deleted_positions(mmr_position: Vec<u64>) -> Vec<BlockHeader>, "fetch_headers_of_deleted_positions");
+
     make_async_fn!(get_stats() -> DbBasicStats, "get_stats");
 
     make_async_fn!(fetch_total_size_stats() -> DbTotalSizeStats, "fetch_total_size_stats");

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -923,6 +923,60 @@ where B: BlockchainBackend
         ))
     }
 
+    // TODO Update this method to make use of a (mmr_position, block_height) index in the lmdb backend instead of a
+    // linear search
+    pub fn fetch_headers_of_deleted_positions(
+        &self,
+        mut mmr_positions: Vec<u64>,
+    ) -> Result<Vec<BlockHeader>, ChainStorageError> {
+        let db = self.db_read_access()?;
+
+        if mmr_positions.is_empty() {
+            return Ok(Vec::new());
+        }
+
+        let chain_metadata = db.fetch_chain_metadata()?;
+        let mut height = chain_metadata.height_of_longest_chain();
+
+        mmr_positions.sort_unstable();
+
+        let mut headers = Vec::with_capacity(mmr_positions.len());
+
+        let mut target = mmr_positions.pop().expect("mmr_positions cannot be empty here");
+
+        loop {
+            if target > u32::MAX as u64 {
+                // TODO: in future, bitmap may support higher than u32
+                return Err(ChainStorageError::InvalidArguments {
+                    func: "fetch_header_of_deleted_position",
+                    arg: "mmr_positions",
+                    message: "mmr_positions should fit into u32".into(),
+                });
+            }
+            if db
+                .fetch_block_accumulated_data_by_height(height)
+                .or_not_found("BlockAccumulatedData", "height", height.to_string())?
+                .deleted()
+                .contains(target as u32)
+            {
+                headers.push(fetch_header(&(*db), height)?);
+                if let Some(pos) = mmr_positions.pop() {
+                    target = pos;
+                } else {
+                    break;
+                }
+            }
+
+            if height > 0 {
+                height -= 1;
+            } else {
+                break;
+            }
+        }
+
+        Ok(headers)
+    }
+
     pub fn get_stats(&self) -> Result<DbBasicStats, ChainStorageError> {
         let lock = self.db_read_access()?;
         lock.get_stats()