feat: add payref background task (#7280)

Description
---
Added background task to rebuild payref indexes based on a payref
rebuild status in the metadata database, rather than doing it as a
migration task at stsrtup.

Motivation and Context
---
Payref migration takes very long time to complete and is not essential
to base node operation.

How Has This Been Tested?
---
System-level testing [**TBD**]

What process can a PR reviewer use to test or verify this change?
---
- Code review
- System-level testing

<!-- Checklist -->
<!-- 1. Is the title of your PR in the form that would make nice release
notes? The title, excluding the conventional commit
tag, will be included exactly as is in the CHANGELOG, so please think
about it carefully. -->


Breaking Changes
---

- [x] None
- [ ] Requires data directory on base node to be deleted
- [ ] Requires hard fork
- [ ] Other - Please specify

<!-- Does this include a breaking change? If so, include this line as a
footer -->
<!-- BREAKING CHANGE: Description what the user should do, e.g. delete a
database, resync the chain -->


<!-- This is an auto-generated comment: release notes by coderabbit.ai
-->
## Summary by CodeRabbit

* **New Features**
* Added background rebuilding of PayRef indexes, improving index
availability and tracking rebuild progress.
* Introduced explicit status tracking for PayRef index rebuilds, with
user-facing error reporting if indexes are unavailable.
* **Bug Fixes**
* Enhanced error messages to provide more context when PayRef indexes
are not available.
* **Chores**
  * Improved logging and internal management of PayRef index rebuilds.
* Updated public exports to include PayRef rebuild status for easier
access in other modules.
<!-- end of auto-generated comment: release notes by coderabbit.ai -->

---------

Co-authored-by: SW van Heerden <swvheerden@gmail.com>

Author: hansieodendaal

base_layer/core/src/chain_storage/blockchain_backend.rs

@@ -6,7 +6,12 @@ use tari_common_types::{
     types::{BadBlock, CompressedCommitment, CompressedPublicKey, FixedHash, HashOutput, Signature},
 };
 
-use super::{lmdb_db::lmdb_tree_reader::OwnedLmdbTreeReader, MinedInfo, TemplateRegistrationEntry};
+use super::{
+    lmdb_db::lmdb_tree_reader::OwnedLmdbTreeReader,
+    MinedInfo,
+    PayrefRebuildStatus,
+    TemplateRegistrationEntry,
+};
 use crate::{
     blocks::{Block, BlockAccumulatedData, BlockHeader, BlockHeaderAccumulatedData, ChainBlock, ChainHeader},
     chain_storage::{
@@ -34,7 +39,7 @@ use crate::{
 /// us to keep the reading and writing API extremely simple. Extending the types of data that the backends can handle
 /// will entail adding to those enums, and the backends, while this trait can remain unchanged.
 #[allow(clippy::ptr_arg)]
-pub trait BlockchainBackend: Send + Sync {
+pub trait BlockchainBackend: Send + Sync + 'static {
     /// Commit the transaction given to the backend. If there is an error, the transaction must be rolled back, and
     /// the error condition returned. On success, every operation in the transaction will have been committed, and
     /// the function will return `Ok(())`.
@@ -138,6 +143,16 @@ pub trait BlockchainBackend: Send + Sync {
     fn fetch_tip_header(&self) -> Result<ChainHeader, ChainStorageError>;
     /// Returns the stored chain metadata.
     fn fetch_chain_metadata(&self) -> Result<ChainMetadata, ChainStorageError>;
+    /// Returns the stored payref rebuild status.
+    fn fetch_payref_rebuild_status(&self) -> Result<PayrefRebuildStatus, ChainStorageError>;
+    /// Builds the payref indexes for a given block height, with stats.
+    fn build_payref_indexes_for_height(
+        &self,
+        height: u64,
+        metadata_at_start: ChainMetadata,
+        initialize_stats: Option<u64>,
+        finalize: bool,
+    ) -> Result<PayrefRebuildStatus, ChainStorageError>;
     /// Returns the UTXO count
     fn utxo_count(&self) -> Result<usize, ChainStorageError>;
     /// Returns the kernel count

base_layer/core/src/chain_storage/blockchain_database.rs

@@ -59,7 +59,7 @@ use tari_hashing::TransactionHashDomain;
 use tari_mmr::pruned_hashset::PrunedHashSet;
 use tari_utilities::{epoch_time::EpochTime, hex::Hex, ByteArray};
 
-use super::{smt_hasher::SmtHasher, MinedInfo, TemplateRegistrationEntry};
+use super::{smt_hasher::SmtHasher, MinedInfo, PayrefRebuildStatus, TemplateRegistrationEntry};
 use crate::{
     block_output_mr_hash_from_pruned_mmr,
     blocks::{
@@ -352,6 +352,88 @@ where B: BlockchainBackend
             self.clear_all_reorgs()?;
         }
 
+        self.rebuild_payref_indexes_background_task()?;
+
+        Ok(())
+    }
+
+    /// This function will rebuild the payref indexes in the background if they are not already rebuilt.
+    pub fn rebuild_payref_indexes_background_task(&self) -> Result<(), ChainStorageError> {
+        let initial_status = {
+            let db = self.db_read_access()?;
+            db.fetch_payref_rebuild_status()?
+        };
+        if initial_status.is_rebuilt {
+            debug!(target: LOG_TARGET, "[PayRef] Payref indexes has already been rebuilt.");
+            return Ok(());
+        }
+
+        // If we had a previous start metadata, we will use that to continue the rebuild, otherwise we will use the
+        // current chain metadata to set a new target rebuild height. All new or re-orged blocks added to the database
+        // after this process started will have the correct payref indexes and therefor do not need to be processed.
+        let metadata_at_start = if let Some(metadata) = initial_status.metadata_at_start.clone() {
+            metadata
+        } else {
+            let db = self.db_read_access()?;
+            db.fetch_chain_metadata()?
+        };
+        let db_rw_lock = self.db.clone();
+
+        tokio::task::spawn(async move {
+            let start_height = initial_status.last_rebuild_height.unwrap_or_default();
+            let mut last_status = initial_status.clone();
+            debug!(
+                target: LOG_TARGET,
+                "[PayRef] Starting index rebuilding for heights {} to {}",
+                start_height, metadata_at_start.best_block_height()
+            );
+
+            let mut initialize_stats = Some(metadata_at_start.best_block_height());
+            for height in start_height..=metadata_at_start.best_block_height() {
+                let finalize = height == metadata_at_start.best_block_height();
+                let metadata = metadata_at_start.clone();
+                let db = db_rw_lock.clone();
+                // We use `spawn_blocking` with `.await` here to ensure that the async spawned task will be able to
+                // shut down when base node shutdown is triggered
+                let res = tokio::task::spawn_blocking(move || {
+                    process_payref_for_height(db, height, metadata, initialize_stats, finalize)
+                })
+                .await;
+                match res {
+                    Ok(Ok(current_status)) => {
+                        last_status = current_status;
+                    },
+                    Ok(Err(e)) => {
+                        error!(
+                            target: LOG_TARGET,
+                            "[PayRef] Index rebuilding failed. Initial status: {:?}. Last updated status: {:?} ({})",
+                            initial_status, last_status, e
+                        );
+                        break;
+                    },
+                    Err(e) => {
+                        error!(
+                            target: LOG_TARGET,
+                            "[PayRef] Index rebuilding failed. Initial status: {:?}. Last updated status: {:?} ({})",
+                            initial_status, last_status, e
+                        );
+                        break;
+                    },
+                }
+                if initialize_stats.is_some() {
+                    initialize_stats = None;
+                }
+                if finalize || last_status.is_rebuilt {
+                    debug!(
+                        target: LOG_TARGET,
+                        "[PayRef] Starting index rebuilding completed, Final status: {:?}",
+                        last_status
+                    );
+                    break;
+                }
+            }
+        });
+
         Ok(())
     }
 
@@ -2651,6 +2733,34 @@ fn convert_to_option_bounds<T: RangeBounds<u64>>(bounds: T) -> (Option<u64>, Opt
     (start, end)
 }
 
+/// Process a batch of outputs in one block for PayRef migration
+fn process_payref_for_height<B: BlockchainBackend>(
+    db: Arc<RwLock<B>>,
+    height: u64,
+    metadata_at_start: ChainMetadata,
+    initialize_stats: Option<u64>,
+    finalize: bool,
+) -> Result<PayrefRebuildStatus, ChainStorageError> {
+    debug!(target: LOG_TARGET, "[PayRef] Processing index rebuilding for height {}", height);
+
+    let write_txn = db
+        .write()
+        .map_err(|_e| ChainStorageError::AccessError("Write lock on blockchain backend failed".into()))?;
+
+    let status =
+        write_txn.build_payref_indexes_for_height(height, metadata_at_start.clone(), initialize_stats, finalize)?;
+
+    if finalize || status.is_rebuilt {
+        debug!(
+            target: LOG_TARGET,
+            "[PayRef] Finalized index rebuilding for heights {} to {}",
+            metadata_at_start.best_block_height(), height
+        );
+    }
+
+    Ok(status)
+}
+
 #[cfg(test)]
 mod test {
     use std::{collections::HashMap, sync};

base_layer/core/src/chain_storage/error.rs

@@ -66,6 +66,14 @@ pub enum ChainStorageError {
     InsertError { table: &'static str, error: String },
     #[error("An invalid query was attempted: {0}")]
     InvalidQuery(String),
+    #[error(
+        "PayRef index not available: current `{current_height}`, start `{start_height}`, target `{target_height}`"
+    )]
+    PayRefIndexNotAvailable {
+        current_height: u64,
+        start_height: u64,
+        target_height: u64,
+    },
     #[error("Invalid argument `{arg}` in `{func}`: {message}")]
     InvalidArguments {
         func: &'static str,
@@ -204,7 +212,8 @@ impl ChainStorageError {
             _err @ ChainStorageError::InvalidChainMetaData(_) |
             _err @ ChainStorageError::OutOfRange |
             _err @ ChainStorageError::MrHashError(_) |
-            _err @ ChainStorageError::JellyfishMerkleTreeError(_) => None,
+            _err @ ChainStorageError::JellyfishMerkleTreeError(_) |
+            _err @ ChainStorageError::PayRefIndexNotAvailable { .. } => None,
         }
     }
 }

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

@@ -65,13 +65,7 @@ where
 {
     let val_buf = serialize(val, None)?;
     match txn.access().put(db, key, &val_buf, put::NOOVERWRITE) {
-        Ok(_) => {
-            trace!(
-                target: LOG_TARGET, "Inserted {} bytes with key '{}' into '{}'",
-                val_buf.len(), to_hex(key.as_lmdb_bytes()), table_name
-            );
-            Ok(())
-        },
+        Ok(_) => Ok(()),
         err @ Err(lmdb_zero::Error::Code(lmdb_zero::error::KEYEXIST)) => {
             error!(
                 target: LOG_TARGET, "Could not insert {} bytes with key '{}' into '{}' ({:?})",