fix: stop race condition in output encumbrance (#4613)

Description
---
Add a mutex to stop a race condition in the output manager error. Its possible that if more than on transactions happens in near the same time, the transactions will select the same inputs. 

How Has This Been Tested?
---
Unit tests

base_layer/core/src/mempool/mempool_storage.rs

@@ -96,12 +96,12 @@ impl MempoolStorage {
                     self.unconfirmed_pool.insert(tx, Some(dependent_outputs), &weight)?;
                     Ok(TxStorageResponse::UnconfirmedPool)
                 } else {
-                    debug!(target: LOG_TARGET, "Validation failed due to unknown inputs");
+                    warn!(target: LOG_TARGET, "Validation failed due to unknown inputs");
                     Ok(TxStorageResponse::NotStoredOrphan)
                 }
             },
             Err(ValidationError::ContainsSTxO) => {
-                debug!(target: LOG_TARGET, "Validation failed due to already spent output");
+                warn!(target: LOG_TARGET, "Validation failed due to already spent input");
                 Ok(TxStorageResponse::NotStoredAlreadySpent)
             },
             Err(ValidationError::MaturityError) => {
@@ -112,6 +112,10 @@ impl MempoolStorage {
                 warn!(target: LOG_TARGET, "Validation failed due to consensus rule: {}", msg);
                 Ok(TxStorageResponse::NotStoredConsensus)
             },
+            Err(ValidationError::DuplicateKernelError(msg)) => {
+                warn!(target: LOG_TARGET, "Validation failed due to duplicate kernel: {}", msg);
+                Ok(TxStorageResponse::NotStoredConsensus)
+            },
             Err(e) => {
                 warn!(target: LOG_TARGET, "Validation failed due to error: {}", e);
                 Ok(TxStorageResponse::NotStored)

base_layer/core/src/validation/error.rs

@@ -113,6 +113,8 @@ pub enum ValidationError {
     },
     #[error("Consensus Error: {0}")]
     ConsensusError(String),
+    #[error("Duplicate kernel Error: {0}")]
+    DuplicateKernelError(String),
     #[error("Covenant failed to validate: {0}")]
     CovenantError(#[from] CovenantError),
     #[error("Invalid or unsupported blockchain version {version}")]

base_layer/core/src/validation/transaction_validators.rs

@@ -172,7 +172,7 @@ impl<B: BlockchainBackend> TxConsensusValidator<B> {
                     db_kernel.excess_sig.get_signature().to_hex(),
                 );
                 warn!(target: LOG_TARGET, "{}", msg);
-                return Err(ValidationError::ConsensusError(msg));
+                return Err(ValidationError::DuplicateKernelError(msg));
             };
         }
         Ok(())

base_layer/core/tests/mempool.rs

@@ -1188,6 +1188,7 @@ async fn consensus_validation_unique_excess_sig() {
     // trying to submit a transaction with an existing excess signature already in the chain is an error
     let tx = Arc::new(tx1);
     let response = mempool.insert(tx).await.unwrap();
+    dbg!(&response);
     assert!(matches!(response, TxStorageResponse::NotStoredConsensus));
 }
 

base_layer/wallet/src/output_manager_service/storage/sqlite_db/mod.rs

@@ -22,7 +22,7 @@
 
 use std::{
     convert::{TryFrom, TryInto},
-    sync::{Arc, RwLock},
+    sync::{Arc, Mutex, RwLock},
 };
 
 use chacha20poly1305::XChaCha20Poly1305;
@@ -68,13 +68,15 @@ const LOG_TARGET: &str = "wallet::output_manager_service::database::wallet";
 pub struct OutputManagerSqliteDatabase {
     database_connection: WalletDbConnection,
     cipher: Arc<RwLock<Option<XChaCha20Poly1305>>>,
+    encumber_lock: Arc<Mutex<()>>,
 }
 
 impl OutputManagerSqliteDatabase {
     pub fn new(database_connection: WalletDbConnection, cipher: Option<XChaCha20Poly1305>) -> Self {
         Self {
             database_connection,
             cipher: Arc::new(RwLock::new(cipher)),
+            encumber_lock: Arc::new(Mutex::new(())),
         }
     }
 
@@ -661,8 +663,15 @@ impl OutputManagerBackend for OutputManagerSqliteDatabase {
         let start = Instant::now();
         let conn = self.database_connection.get_pooled_connection()?;
         let acquire_lock = start.elapsed();
+        // We need to ensure that this whole encumber operation happens inside of a mutex to ensure thread safety as the
+        // transaction first check checks if it can encumber then encumbers them.
+        let _guard = self
+            .encumber_lock
+            .lock()
+            .map_err(|e| OutputManagerStorageError::UnexpectedResult(format!("Encumber lock poisoned: {}", e)))?;
 
         let mut outputs_to_be_spent = Vec::with_capacity(outputs_to_send.len());
+
         for i in outputs_to_send {
             let output = OutputSql::find_by_commitment_and_cancelled(i.commitment.as_bytes(), false, &conn)?;
             if output.status != (OutputStatus::Unspent as i32) {
@@ -714,6 +723,12 @@ impl OutputManagerBackend for OutputManagerSqliteDatabase {
         let conn = self.database_connection.get_pooled_connection()?;
         let acquire_lock = start.elapsed();
 
+        // We need to ensure that this whole encumber operation happens inside of a mutex to ensure thread safety as the
+        // transaction first check checks if it can encumber then encumbers them.
+        let _guard = self
+            .encumber_lock
+            .lock()
+            .map_err(|e| OutputManagerStorageError::UnexpectedResult(format!("Encumber lock poisoned: {}", e)))?;
         let outputs_to_be_received =
             OutputSql::find_by_tx_id_and_status(tx_id, OutputStatus::ShortTermEncumberedToBeReceived, &conn)?;
         for o in &outputs_to_be_received {