feat: keep smt memory

applications/minotari_node/src/bootstrap.rs

@@ -20,7 +20,12 @@
 //  WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 //  USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::{cmp, str::FromStr, sync::Arc, time::Duration};
+use std::{
+    cmp,
+    str::FromStr,
+    sync::{Arc, RwLock},
+    time::Duration,
+};
 
 use log::*;
 use minotari_app_utilities::{consts, identity_management, identity_management::load_from_json};
@@ -52,6 +57,7 @@ use tari_core::{
     mempool::{service::MempoolHandle, Mempool, MempoolServiceInitializer, MempoolSyncInitializer},
     proof_of_work::randomx_factory::RandomXFactory,
     transactions::CryptoFactories,
+    OutputSmt,
 };
 use tari_p2p::{
     auto_update::SoftwareUpdaterService,
@@ -81,6 +87,7 @@ pub struct BaseNodeBootstrapper<'a, B> {
     pub factories: CryptoFactories,
     pub randomx_factory: RandomXFactory,
     pub interrupt_signal: ShutdownSignal,
+    pub smt: Arc<RwLock<OutputSmt>>,
 }
 
 impl<B> BaseNodeBootstrapper<'_, B>

applications/minotari_node/src/builder.rs

@@ -20,7 +20,7 @@
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 
 use log::*;
 use tari_common::{
@@ -42,6 +42,7 @@ use tari_core::{
         transaction::TransactionFullValidator,
         DifficultyCalculator,
     },
+    OutputSmt,
 };
 use tari_p2p::{auto_update::SoftwareUpdaterHandle, services::liveness::LivenessHandle};
 use tari_service_framework::ServiceHandles;
@@ -210,6 +211,7 @@ async fn build_node_context(
     let factories = CryptoFactories::default();
     let randomx_factory = RandomXFactory::new(app_config.base_node.max_randomx_vms);
     let difficulty_calculator = DifficultyCalculator::new(rules.clone(), randomx_factory.clone());
+    let smt = Arc::new(RwLock::new(OutputSmt::new()));
     let validators = Validators::new(
         BlockBodyFullValidator::new(rules.clone(), true),
         HeaderFullValidator::new(rules.clone(), difficulty_calculator.clone()),
@@ -226,6 +228,7 @@ async fn build_node_context(
         validators,
         app_config.base_node.storage,
         difficulty_calculator,
+        smt.clone(),
     )
     .map_err(|err| {
         if let ChainStorageError::DatabaseResyncRequired(reason) = err {
@@ -262,6 +265,7 @@ async fn build_node_context(
         factories: factories.clone(),
         randomx_factory,
         interrupt_signal: interrupt_signal.clone(),
+        smt,
     }
     .bootstrap()
     .await?;

applications/minotari_node/src/recovery.rs

@@ -25,7 +25,7 @@ use std::{
     env::temp_dir,
     fs,
     io::{self, Write},
-    sync::Arc,
+    sync::{Arc, RwLock},
 };
 
 use anyhow::anyhow;
@@ -53,6 +53,7 @@ use tari_core::{
         mocks::MockValidator,
         DifficultyCalculator,
     },
+    OutputSmt,
 };
 
 use crate::{BaseNodeConfig, DatabaseType};
@@ -97,6 +98,7 @@ pub async fn run_recovery(node_config: &BaseNodeConfig) -> Result<(), anyhow::Er
     let factories = CryptoFactories::default();
     let randomx_factory = RandomXFactory::new(node_config.max_randomx_vms);
     let difficulty_calculator = DifficultyCalculator::new(rules.clone(), randomx_factory);
+    let smt = Arc::new(RwLock::new(OutputSmt::new()));
     let validators = Validators::new(
         BlockBodyFullValidator::new(rules.clone(), true),
         HeaderFullValidator::new(rules.clone(), difficulty_calculator.clone()),
@@ -114,6 +116,7 @@ pub async fn run_recovery(node_config: &BaseNodeConfig) -> Result<(), anyhow::Er
         validators,
         node_config.storage,
         difficulty_calculator,
+        smt,
     )?;
     do_recovery(db.into(), temp_db).await?;
 
@@ -142,12 +145,14 @@ async fn do_recovery<D: BlockchainBackend + 'static>(
         MockValidator::new(true),
         MockValidator::new(true),
     );
+    let smt = Arc::new(RwLock::new(OutputSmt::new()));
     let source_database = BlockchainDatabase::new(
         source_backend,
         rules.clone(),
         validators,
         BlockchainDatabaseConfig::default(),
         DifficultyCalculator::new(rules, Default::default()),
+        smt,
     )?;
     let max_height = source_database
         .get_chain_metadata()

base_layer/core/src/base_node/sync/block_sync/synchronizer.rs

@@ -321,7 +321,8 @@ impl<'a, B: BlockchainBackend + 'static> BlockSynchronizer<'a, B> {
             let validator = self.block_validator.clone();
             let res = task::spawn_blocking(move || {
                 let txn = db.db_read_access()?;
-                validator.validate_body(&*txn, &task_block)
+                let smt = db.smt().clone();
+                validator.validate_body(&*txn, &task_block, smt)
             })
             .await?;
 
@@ -367,7 +368,7 @@ impl<'a, B: BlockchainBackend + 'static> BlockSynchronizer<'a, B> {
             self.db
                 .write_transaction()
                 .delete_orphan(header_hash)
-                .insert_tip_block_body(block.clone())
+                .insert_tip_block_body(block.clone(), self.db.inner().smt())
                 .set_best_block(
                     block.height(),
                     header_hash,

base_layer/core/src/base_node/sync/horizon_state_sync/synchronizer.rs

@@ -298,11 +298,14 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
         debug!(target: LOG_TARGET, "Synchronizing kernels");
         self.synchronize_kernels(sync_peer.clone(), client, to_header).await?;
         debug!(target: LOG_TARGET, "Synchronizing outputs");
+        let cloned_backup_smt = self.db.inner().smt_read_access()?.clone();
         match self.synchronize_outputs(sync_peer, client, to_header).await {
             Ok(_) => Ok(()),
             Err(err) => {
                 // We need to clean up the outputs
                 let _ = self.clean_up_failed_output_sync(to_header).await;
+                let mut smt = self.db.inner().smt_write_access()?;
+                *smt = cloned_backup_smt;
                 Err(err)
             },
         }
@@ -618,7 +621,7 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
         let mut utxo_counter = 0u64;
         let mut stxo_counter = 0u64;
         let timer = Instant::now();
-        let mut output_smt = db.fetch_tip_smt().await?;
+        let mut output_smt = (*db.inner().smt_write_access()?).clone();
         let mut last_sync_timer = Instant::now();
         let mut avg_latency = RollingAverageTime::new(20);
 
@@ -766,8 +769,8 @@ impl<'a, B: BlockchainBackend + 'static> HorizonStateSynchronization<'a, B> {
                 txn.commit().await?;
             }
         }
-        // This has a very low probability of failure
-        db.set_tip_smt(output_smt).await?;
+        let mut writing_lock_output_smt = db.inner().smt_write_access()?;
+        *writing_lock_output_smt = output_smt;
         debug!(
             target: LOG_TARGET,
             "Finished syncing TXOs: {} unspent and {} spent downloaded in {:.2?}",

base_layer/core/src/chain_storage/async_db.rs

@@ -19,7 +19,12 @@
 // SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
 // WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE
 // USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
-use std::{mem, ops::RangeBounds, sync::Arc, time::Instant};
+use std::{
+    mem,
+    ops::RangeBounds,
+    sync::{Arc, RwLock},
+    time::Instant,
+};
 
 use log::*;
 use primitive_types::U256;
@@ -222,10 +227,6 @@ impl<B: BlockchainBackend + 'static> AsyncBlockchainDb<B> {
 
     make_async_fn!(fetch_tip_header() -> ChainHeader, "fetch_tip_header");
 
-    make_async_fn!(fetch_tip_smt() -> OutputSmt, "fetch_tip_smt");
-
-    make_async_fn!(set_tip_smt(smt: OutputSmt) -> (), "set_tip_smt");
-
     make_async_fn!(insert_valid_headers(headers: Vec<ChainHeader>) -> (), "insert_valid_headers");
 
     //---------------------------------- Block --------------------------------------------//
@@ -393,8 +394,8 @@ impl<'a, B: BlockchainBackend + 'static> AsyncDbTransaction<'a, B> {
         self
     }
 
-    pub fn insert_tip_block_body(&mut self, block: Arc<ChainBlock>) -> &mut Self {
-        self.transaction.insert_tip_block_body(block);
+    pub fn insert_tip_block_body(&mut self, block: Arc<ChainBlock>, smt: Arc<RwLock<OutputSmt>>) -> &mut Self {
+        self.transaction.insert_tip_block_body(block, smt);
         self
     }
 

base_layer/core/src/chain_storage/blockchain_backend.rs

@@ -91,7 +91,7 @@ pub trait BlockchainBackend: Send + Sync {
     fn fetch_outputs_in_block_with_spend_state(
         &self,
         header_hash: &HashOutput,
-        spend_status_at_header: Option<HashOutput>,
+        spend_status_at_header: Option<&HashOutput>,
     ) -> Result<Vec<(TransactionOutput, bool)>, ChainStorageError>;
 
     /// Fetch a specific output. Returns the output
@@ -181,6 +181,6 @@ pub trait BlockchainBackend: Send + Sync {
         start_height: u64,
         end_height: u64,
     ) -> Result<Vec<TemplateRegistrationEntry>, ChainStorageError>;
-    /// Returns the tip utxo smt
-    fn fetch_tip_smt(&self) -> Result<OutputSmt, ChainStorageError>;
+    /// Calculates the tip utxo smt
+    fn calculate_tip_smt(&self) -> Result<OutputSmt, ChainStorageError>;
 }